fix: dedup three-bucket projection + honest C3 error-handling note

2026-07-15 12:29:58 -04:00
parent aa10485b9f
commit 84b0b30213
2 changed files with 159 additions and 51 deletions
@@ -9,11 +9,14 @@
    4. 逐行 store.upsert 落盘，单行单事务 → 崩溃最多丢正在写的一行。

 错误处理诚实标注（不谎称全传播）:
-    - run_diagnosis 内部对 judge/C3 判别异常是 `except Exception`→warning→默认
-      lapse（core/evolution/diagnose.py:2186-2192），非全传播；judge 语义歧义
-      按现有保护性 lapse 处理，本编排原样接受其判定，不二次兜底。
-    - 网络/API 层失败经 GovernedLLMClient 重试栈后仍失败会从 run_diagnosis
-      向上抛出，本编排不捕获、不掩盖，直接冒泡给调用方。
+    - run_diagnosis 的 C1/C2 阶段（指标计算、错误归因）网络/API 失败经
+      GovernedLLMClient 重试栈后仍失败会向上抛出，本编排不捕获、不掩盖，
+      直接冒泡给调用方。
+    - 但 C3 阶段（defect/lapse judge）的调用整体包在 `except Exception` 内
+      （core/evolution/diagnose.py:2186），故 C3 judge 的**全部异常（含网络/API
+      失败）都被吞并→warning→默认归为 lapse**，不会向上抛；judge 语义歧义同样
+      按此保护性 fallback 处理。本编排原样接受该判定，不二次兜底、也不谎称
+      C3 阶段网络失败会传播。
 """

 from __future__ import annotations
@@ -166,13 +169,69 @@ def _project_and_persist(
        {tier: 行数} 计数字典（T2/T1/T0/uncertain），供上层日志与 manifest。

    关键实现:
-        逐行 upsert（单行单事务），中途崩溃最多丢正在写的一行；三类产物互斥，
-        同一 question_id 不会在两类中重复出现（run_diagnosis 保证）。
+        逐行 upsert（单行单事务），中途崩溃最多丢正在写的一行。三桶**非互斥**：
+        同一 degraded 错题可能同时出现在 error_attributions（judge 解析失败仍建
+        attribution）里，故按 **degraded > infra > attribution** 优先级去重——先落
+        degraded/infra，再在 attribution 循环跳过已落题，保证**每题恰写一行、
+        counts 恰计一次**（否则同 PK 覆盖会导致 counts 双计且分层错乱）。
    """
    counts = {"T2": 0, "T1": 0, "T0": 0, "uncertain": 0}
+    # 优先级去重：degraded > infra > attribution。先记录高优先集合，
+    # attribution 循环遇到已落题即跳过，确保每题唯一落库。
+    persisted: set[str] = set()

-    # error_attributions：defect→T2 / lapse→T1 / 其它→uncertain（由 score_signal 判定）
+    # degraded_question_ids（最高优先）：judge 解析失败降级 → uncertain，信号不可信排除出 T2
+    for qid in result.degraded_question_ids:
+        q = questions[qid]
+        store.upsert(
+            DiagnosisSignalRow(
+                question_id=qid,
+                video_id=q.video_id,
+                baseline_run_id=baseline_run_id,
+                diag_fingerprint=diag_fingerprint,
+                task_type=q.task_type,
+                error_type=None,
+                cause_category=None,
+                tier="uncertain",
+                evolution_target=None,
+                degraded=True,
+                infra=False,
+                session_id=None,
+            )
+        )
+        persisted.add(qid)
+        counts["uncertain"] += 1
+
+    # infra_question_ids：基础设施失败护栏排除 → T0，不参与训练主体
+    # （防御性跳过已落 degraded 题，虽 infra 通常已在诊断前过滤不重叠）
+    for qid in result.infra_question_ids:
+        if qid in persisted:
+            continue
+        q = questions[qid]
+        store.upsert(
+            DiagnosisSignalRow(
+                question_id=qid,
+                video_id=q.video_id,
+                baseline_run_id=baseline_run_id,
+                diag_fingerprint=diag_fingerprint,
+                task_type=q.task_type,
+                error_type=None,
+                cause_category=None,
+                tier="T0",
+                evolution_target=None,
+                degraded=False,
+                infra=True,
+                session_id=None,
+            )
+        )
+        persisted.add(qid)
+        counts["T0"] += 1
+
+    # error_attributions（最低优先）：defect→T2 / lapse→T1 / 其它→uncertain（由 score_signal 判定）
+    # 跳过已作为 degraded/infra 落库的题，避免同 PK 覆盖与 counts 双计。
    for ea in result.error_attributions:
+        if ea.question_id in persisted:
+            continue
        q = questions[ea.question_id]
        tier = score_signal(cause_category=ea.cause_category, infra=False, degraded=False).tier
        # error_type 是 ErrorAttribution 必填字段（永远已知），确定性派生进化目标。
@@ -193,48 +252,7 @@ def _project_and_persist(
                session_id=None,
            )
        )
+        persisted.add(ea.question_id)
        counts[tier] = counts.get(tier, 0) + 1

-    # infra_question_ids：基础设施失败护栏排除 → T0，不参与训练主体
-    for qid in result.infra_question_ids:
-        q = questions[qid]
-        store.upsert(
-            DiagnosisSignalRow(
-                question_id=qid,
-                video_id=q.video_id,
-                baseline_run_id=baseline_run_id,
-                diag_fingerprint=diag_fingerprint,
-                task_type=q.task_type,
-                error_type=None,
-                cause_category=None,
-                tier="T0",
-                evolution_target=None,
-                degraded=False,
-                infra=True,
-                session_id=None,
-            )
-        )
-        counts["T0"] += 1
-
-    # degraded_question_ids：judge 解析失败降级 → uncertain，信号不可信排除出 T2
-    for qid in result.degraded_question_ids:
-        q = questions[qid]
-        store.upsert(
-            DiagnosisSignalRow(
-                question_id=qid,
-                video_id=q.video_id,
-                baseline_run_id=baseline_run_id,
-                diag_fingerprint=diag_fingerprint,
-                task_type=q.task_type,
-                error_type=None,
-                cause_category=None,
-                tier="uncertain",
-                evolution_target=None,
-                degraded=True,
-                infra=False,
-                session_id=None,
-            )
-        )
-        counts["uncertain"] += 1
-
    return counts
@@ -1,8 +1,11 @@
 """离线诊断编排集成测试（LLM 类：MD 产出到 tests/outputs/）。

 用 fake run_diagnosis + fake deps 覆盖编排契约，不实际调 LLM/VLM：
-    1. 续跑幂等 —— 已落盘题跳过，第二次无剩余则 run_diagnosis 收到空列表。
+    1. 续跑幂等 —— 已落盘题跳过；第二次 remaining 为空则直接早返回、不调用
+       run_diagnosis。
    2. 投影正确 —— defect→T2、lapse→T1，evolution_target 由 error_type 派生。
+    3. 三桶去重 —— 同题同时在 degraded 与 attributions 时按 degraded>infra>
+       attribution 优先级只落 1 行、tier=uncertain。

 测试结束把编排过程（remaining、各 tier 计数、投影样例）写入
 tests/outputs/test_baseline_diagnosis/<test>_<固定 ts>.md（CLAUDE.md §4.6）。
@@ -262,3 +265,90 @@ async def test_infra_and_degraded_projection(tmp_path, monkeypatch):
    )
    assert md_path.exists()
    store.close()
+
+
+@pytest.mark.asyncio
+async def test_degraded_overrides_attribution(tmp_path, monkeypatch):
+    """三桶去重：同题同时在 degraded 与 attributions 时只落 1 行且 tier=uncertain。
+
+    judge 解析失败会生成 degraded metrics，若仍是错题还会建 attribution，
+    故同一 question_id 可同时出现在两桶。按 degraded>infra>attribution 优先级，
+    该题必须只落 1 行、tier=uncertain（degraded 置位），attribution 行被跳过。
+    """
+    calls: list[tuple[str, ...]] = []
+
+    async def fake_run_diagnosis(
+        run_id,
+        questions,
+        tree_data,
+        llm,
+        run_log,
+        skill_store,
+        prompts,
+        *,
+        concurrency,
+        question_ids=None,
+        task_types=None,
+        only_incorrect=False,
+    ):
+        calls.append(tuple(question_ids or []))
+        from core.evolution.types import DiagnosisResult, ErrorAttribution
+
+        # q5 同时出现在 attributions（judge 解析失败仍建归因）与 degraded_question_ids。
+        return DiagnosisResult(
+            run_id=run_id,
+            error_attributions=[ErrorAttribution("q5", "reasoning_failure", None, "defect")],
+            infra_question_ids=[],
+            degraded_question_ids=["q5"],
+        )
+
+    monkeypatch.setattr("app.harness.baseline_diagnosis.run_diagnosis", fake_run_diagnosis)
+    deps = DiagnosisDeps(
+        run_log=_FakeRunLog(),
+        llm=_FakeLLM(),
+        skill_store=_FakeSkillStore(),
+        prompts=object(),
+        tree_data={},
+        concurrency=2,
+    )
+    store = SqliteDiagnosisSignalStore(str(tmp_path / "h.db"))
+    q_by_id = {"q5": _mk_q("q5")}
+
+    await run_baseline_diagnosis(
+        baseline_run_id="infer_adhoc",
+        diag_fingerprint="fp",
+        wrong_ids=["q5"],
+        questions=q_by_id,
+        store=store,
+        deps=deps,
+    )
+
+    all_rows = store.load("infer_adhoc", "fp")
+    assert len(all_rows) == 1  # 只落 1 行（无同 PK 双写）
+    row = all_rows[0]
+    assert row.question_id == "q5"
+    assert row.tier == "uncertain"  # degraded 优先级最高
+    assert row.degraded is True
+    assert row.error_type is None  # 走 degraded 投影而非 attribution
+    assert row.evolution_target is None
+
+    md_path = _write_md(
+        "test_degraded_overrides_attribution",
+        [
+            "# 离线诊断编排：三桶优先级去重",
+            "",
+            "## 任务描述",
+            "q5 同时出现在 error_attributions（defect）与 degraded_question_ids，"
+            "验证按 degraded>infra>attribution 优先级只落 1 行。",
+            "",
+            "## 落库结果",
+            "| question_id | 落库行数 | tier | degraded | error_type |",
+            "|---|---|---|---|---|",
+            f"| q5 | {len(all_rows)} | {row.tier} | {row.degraded} | {row.error_type} |",
+            "",
+            "## 结论",
+            "degraded 优先级最高，attribution 行被跳过 → 每题恰写一行、counts 恰计一次。",
+        ],
+    )
+    assert md_path.exists()
+    store.close()