From 53989078a023acef7c1dcfbc8b8d1c00687279a3 Mon Sep 17 00:00:00 2001
From: iomgaa <iomgaaycz@gmail.com>
Date: Wed, 15 Jul 2026 12:22:07 -0400
Subject: [PATCH] feat: add offline baseline diagnosis orchestration

---
 app/harness/baseline_diagnosis.py            | 240 +++++++++++++++++
 tests/integration/test_baseline_diagnosis.py | 264 +++++++++++++++++++
 2 files changed, 504 insertions(+)
 create mode 100644 app/harness/baseline_diagnosis.py
 create mode 100644 tests/integration/test_baseline_diagnosis.py

diff --git a/app/harness/baseline_diagnosis.py b/app/harness/baseline_diagnosis.py
new file mode 100644
index 0000000..5f17993
--- /dev/null
+++ b/app/harness/baseline_diagnosis.py
@@ -0,0 +1,240 @@
+"""离线诊断编排：把 baseline run 的错题诊断投影为逐题信号行并断点续跑落库。
+
+"结果驱动视频级切分"离线管线的诊断步。给定一批可诊断错题：
+    1. 算 remaining（跳过 store 已完成题）实现续跑幂等；
+    2. 对剩余错题调 core.evolution.diagnose.run_diagnosis（经 StepsJsonRunLog
+       包装内层 RunLog，兼容 traces 未落表的历史 run）；
+    3. 把 error_attributions / infra / degraded 三类产物确定性投影为
+       DiagnosisSignalRow（tier 由 split_selection.score_signal 判定）；
+    4. 逐行 store.upsert 落盘，单行单事务 → 崩溃最多丢正在写的一行。
+
+错误处理诚实标注（不谎称全传播）:
+    - run_diagnosis 内部对 judge/C3 判别异常是 `except Exception`→warning→默认
+      lapse（core/evolution/diagnose.py:2186-2192），非全传播；judge 语义歧义
+      按现有保护性 lapse 处理，本编排原样接受其判定，不二次兜底。
+    - 网络/API 层失败经 GovernedLLMClient 重试栈后仍失败会从 run_diagnosis
+      向上抛出，本编排不捕获、不掩盖，直接冒泡给调用方。
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import TYPE_CHECKING, Any
+
+from loguru import logger
+
+from app.harness.baseline_run_log import StepsJsonRunLog
+from app.harness.split_selection import evolution_target_of, score_signal
+from core.evolution.diagnose import run_diagnosis
+from core.evolution.types import DiagnosisSignalRow
+
+if TYPE_CHECKING:
+    from core.evolution.protocols import DiagnosisSignalStore
+    from core.evolution.types import DiagnosisResult
+    from core.types import GeneratedQuestion
+
+
+@dataclass(frozen=True)
+class DiagnosisDeps:
+    """离线诊断编排的依赖束（一次编排的全部外部端口 + 运行参数）。
+
+    frozen 保证一次编排内依赖不可变；LLM/RunLog/SkillStore/prompts 走 Protocol
+    注入，便于测试替换成假实现。
+
+    属性:
+        run_log: 内层 RunLog 实现（提供 get_predictions/get_traces），
+            编排内部再用 StepsJsonRunLog 包装以兼容 traces 未落表的 run。
+        llm: LLM 调用端口（治理后的 GovernedLLMClient）。
+        skill_store: 技能文件读取端口。
+        prompts: 诊断模板束（DiagnosePrompts）。
+        tree_data: 树结构字典（多视频 {video_id: tree} 或单棵树），透传给 run_diagnosis。
+        concurrency: 诊断并发上限。
+    """
+
+    run_log: Any
+    llm: Any
+    skill_store: Any
+    prompts: Any
+    tree_data: dict[str, Any]
+    concurrency: int
+
+
+async def run_baseline_diagnosis(
+    *,
+    baseline_run_id: str,
+    diag_fingerprint: str,
+    wrong_ids: list[str],
+    questions: dict[str, GeneratedQuestion],
+    store: DiagnosisSignalStore,
+    deps: DiagnosisDeps,
+) -> None:
+    """对 baseline run 的错题跑离线诊断并把信号逐行落库（断点续跑幂等）。
+
+    参数:
+        baseline_run_id: baseline run 标识（如 "infer_adhoc"），信号行主键之一。
+        diag_fingerprint: 诊断口径指纹，隔离不同诊断配置的信号，主键之一。
+        wrong_ids: 本次待诊断的可诊断错题 question_id 列表（保序）。
+        questions: question_id → GeneratedQuestion 映射，需覆盖 wrong_ids 全部题
+            及 run_diagnosis 返回的所有 infra/degraded 题（用于取 video_id/task_type）。
+        store: 诊断信号存储端口，逐行 upsert 落盘并提供 done_question_ids 续跑查询。
+        deps: 外部依赖束（见 DiagnosisDeps）。
+
+    返回:
+        None。副作用为把逐题 DiagnosisSignalRow 写入 store。
+
+    关键实现:
+        - remaining = wrong_ids 去除 store 已完成题；空则直接 return（续跑幂等，
+          重复调用零副作用）。
+        - run_diagnosis 只诊断 remaining，避免重复 LLM 调用浪费。
+        - 三类产物投影互斥落库：error_attributions（defect/lapse）、infra_question_ids
+          （T0）、degraded_question_ids（uncertain）。
+    """
+    # Phase 1: 算 remaining（续跑幂等）
+    done = store.done_question_ids(baseline_run_id, diag_fingerprint)
+    remaining = [qid for qid in wrong_ids if qid not in done]
+    if not remaining:
+        logger.info(
+            "离线诊断续跑：baseline={} fingerprint={} 无剩余错题（已完成 {} 题），跳过。",
+            baseline_run_id,
+            diag_fingerprint,
+            len(done),
+        )
+        return
+
+    logger.info(
+        "离线诊断开始：baseline={} fingerprint={} 剩余 {}/{} 题待诊断。",
+        baseline_run_id,
+        diag_fingerprint,
+        len(remaining),
+        len(wrong_ids),
+    )
+
+    # Phase 2: 对剩余错题跑诊断（StepsJsonRunLog 兼容 traces 未落表的历史 run）
+    result = await run_diagnosis(
+        baseline_run_id,
+        [questions[qid] for qid in remaining],
+        deps.tree_data,
+        deps.llm,
+        StepsJsonRunLog(deps.run_log),
+        deps.skill_store,
+        deps.prompts,
+        concurrency=deps.concurrency,
+        question_ids=list(remaining),
+        only_incorrect=True,
+    )
+
+    # Phase 3: 投影落库
+    counts = _project_and_persist(
+        result=result,
+        baseline_run_id=baseline_run_id,
+        diag_fingerprint=diag_fingerprint,
+        questions=questions,
+        store=store,
+    )
+
+    logger.info(
+        "离线诊断落库完成：baseline={} fingerprint={} "
+        "T2={} T1={} T0(infra)={} uncertain(degraded)={} 共 {} 行。",
+        baseline_run_id,
+        diag_fingerprint,
+        counts["T2"],
+        counts["T1"],
+        counts["T0"],
+        counts["uncertain"],
+        sum(counts.values()),
+    )
+
+
+def _project_and_persist(
+    *,
+    result: DiagnosisResult,
+    baseline_run_id: str,
+    diag_fingerprint: str,
+    questions: dict[str, GeneratedQuestion],
+    store: DiagnosisSignalStore,
+) -> dict[str, int]:
+    """把 DiagnosisResult 三类产物投影为信号行并逐行 upsert，返回各 tier 计数。
+
+    参数:
+        result: run_diagnosis 的返回，含 error_attributions/infra/degraded 三类产物。
+        baseline_run_id: 信号行主键之一。
+        diag_fingerprint: 信号行主键之一。
+        questions: question_id → GeneratedQuestion，用于取 video_id/task_type。
+        store: 诊断信号存储端口。
+
+    返回:
+        {tier: 行数} 计数字典（T2/T1/T0/uncertain），供上层日志与 manifest。
+
+    关键实现:
+        逐行 upsert（单行单事务），中途崩溃最多丢正在写的一行；三类产物互斥，
+        同一 question_id 不会在两类中重复出现（run_diagnosis 保证）。
+    """
+    counts = {"T2": 0, "T1": 0, "T0": 0, "uncertain": 0}
+
+    # error_attributions：defect→T2 / lapse→T1 / 其它→uncertain（由 score_signal 判定）
+    for ea in result.error_attributions:
+        q = questions[ea.question_id]
+        tier = score_signal(cause_category=ea.cause_category, infra=False, degraded=False).tier
+        # error_type 是 ErrorAttribution 必填字段（永远已知），确定性派生进化目标。
+        evolution_target = evolution_target_of(ea.error_type)
+        store.upsert(
+            DiagnosisSignalRow(
+                question_id=ea.question_id,
+                video_id=q.video_id,
+                baseline_run_id=baseline_run_id,
+                diag_fingerprint=diag_fingerprint,
+                task_type=q.task_type,
+                error_type=ea.error_type,
+                cause_category=ea.cause_category,
+                tier=tier,
+                evolution_target=evolution_target,
+                degraded=False,
+                infra=False,
+                session_id=None,
+            )
+        )
+        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
diff --git a/tests/integration/test_baseline_diagnosis.py b/tests/integration/test_baseline_diagnosis.py
new file mode 100644
index 0000000..c2f4d5e
--- /dev/null
+++ b/tests/integration/test_baseline_diagnosis.py
@@ -0,0 +1,264 @@
+"""离线诊断编排集成测试（LLM 类：MD 产出到 tests/outputs/）。
+
+用 fake run_diagnosis + fake deps 覆盖编排契约，不实际调 LLM/VLM：
+    1. 续跑幂等 —— 已落盘题跳过，第二次无剩余则 run_diagnosis 收到空列表。
+    2. 投影正确 —— defect→T2、lapse→T1，evolution_target 由 error_type 派生。
+
+测试结束把编排过程（remaining、各 tier 计数、投影样例）写入
+tests/outputs/test_baseline_diagnosis/<test>_<固定 ts>.md（CLAUDE.md §4.6）。
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from adapters.baseline_diagnosis_store import SqliteDiagnosisSignalStore
+from app.harness.baseline_diagnosis import DiagnosisDeps, run_baseline_diagnosis
+
+# 固定时间戳：库代码不用 datetime.now，测试传入固定值保证 MD 可复现。
+_FIXED_TS = "20260715_000000"
+_OUTPUT_DIR = Path(__file__).resolve().parents[1] / "outputs" / "test_baseline_diagnosis"
+
+
+class _FakeRunLog:
+    """内层 RunLog 假实现：predictions/traces 均返回空，编排不真正诊断。"""
+
+    async def get_predictions(self, run_id, *, question_ids=None):
+        return []
+
+    async def get_traces(self, run_id, *, question_ids=None):
+        return []
+
+
+class _FakeLLM: ...
+
+
+class _FakeSkillStore: ...
+
+
+def _mk_q(qid):
+    """构造最小可用 GeneratedQuestion（补齐必填 source_nodes/difficulty）。"""
+    from core.types import GeneratedQuestion
+
+    return GeneratedQuestion(
+        question_id=qid,
+        video_id="v",
+        task_type="Counting Problem",
+        question="",
+        options=("A", "B", "C", "D"),
+        answer="A",
+        source_nodes=(),
+        difficulty="easy",
+    )
+
+
+def _deps(monkeypatch, calls):
+    """构造 DiagnosisDeps 并 monkeypatch run_diagnosis 记录每次 question_ids。"""
+
+    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
+
+        # 仅对本次传入的题产出归因，续跑时空列表 → 无归因。
+        attributions = []
+        if "q1" in (question_ids or []):
+            attributions.append(ErrorAttribution("q1", "search_failure", None, "defect"))
+        if "q2" in (question_ids or []):
+            attributions.append(ErrorAttribution("q2", "mixed", None, "lapse"))
+        return DiagnosisResult(
+            run_id=run_id,
+            error_attributions=attributions,
+            infra_question_ids=[],
+            degraded_question_ids=[],
+        )
+
+    monkeypatch.setattr("app.harness.baseline_diagnosis.run_diagnosis", fake_run_diagnosis)
+    return DiagnosisDeps(
+        run_log=_FakeRunLog(),
+        llm=_FakeLLM(),
+        skill_store=_FakeSkillStore(),
+        prompts=object(),
+        tree_data={},
+        concurrency=2,
+    )
+
+
+def _write_md(test_name: str, lines: list[str]) -> Path:
+    """把编排过程写入 tests/outputs 下的 MD（人类可读结构化）。"""
+    _OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
+    path = _OUTPUT_DIR / f"{test_name}_{_FIXED_TS}.md"
+    path.write_text("\n".join(lines), encoding="utf-8")
+    return path
+
+
+@pytest.mark.asyncio
+async def test_resume_skips_done(tmp_path, monkeypatch):
+    """续跑幂等 + 投影正确：首次全诊断落库，第二次无剩余；tier 投影符合分层。"""
+    calls: list[tuple[str, ...]] = []
+    deps = _deps(monkeypatch, calls)
+    store = SqliteDiagnosisSignalStore(str(tmp_path / "h.db"))
+    q_by_id = {"q1": _mk_q("q1"), "q2": _mk_q("q2")}
+
+    # 第一次：两题都是剩余，全部诊断落库。
+    await run_baseline_diagnosis(
+        baseline_run_id="infer_adhoc",
+        diag_fingerprint="fp",
+        wrong_ids=["q1", "q2"],
+        questions=q_by_id,
+        store=store,
+        deps=deps,
+    )
+    assert store.done_question_ids("infer_adhoc", "fp") == {"q1", "q2"}
+    assert calls[0] == ("q1", "q2")
+
+    # 投影正确性：q1 defect→T2、q2 lapse→T1，evolution_target 由 error_type 派生。
+    rows = {r.question_id: r for r in store.load("infer_adhoc", "fp")}
+    assert rows["q1"].tier == "T2"
+    assert rows["q1"].cause_category == "defect"
+    assert rows["q1"].evolution_target == "skill"  # search_failure → skill
+    assert rows["q1"].error_type == "search_failure"
+    assert rows["q1"].infra is False
+    assert rows["q1"].degraded is False
+    assert rows["q2"].tier == "T1"
+    assert rows["q2"].cause_category == "lapse"
+    assert rows["q2"].evolution_target == "system"  # mixed → system
+
+    # 第二次：两题已完成 → remaining 为空。编排按规约直接 return（续跑幂等，
+    # 不浪费 LLM 诊断调用），故不会向 run_diagnosis 新增调用。
+    n_calls_before = len(calls)
+    await run_baseline_diagnosis(
+        baseline_run_id="infer_adhoc",
+        diag_fingerprint="fp",
+        wrong_ids=["q1", "q2"],
+        questions=q_by_id,
+        store=store,
+        deps=deps,
+    )
+    assert len(calls) == n_calls_before  # 第二次无剩余 → 未触发诊断
+
+    md_path = _write_md(
+        "test_resume_skips_done",
+        [
+            "# 离线诊断编排：续跑幂等 + 投影正确",
+            "",
+            "## 任务描述",
+            "对 infer_adhoc 的错题跑离线诊断，投影为逐题信号行并落库；验证续跑幂等。",
+            "",
+            "## run_diagnosis 每次收到的 question_ids",
+            f"- 第 1 次: {calls[0]}",
+            "- 第 2 次: 未触发（remaining 为空，编排直接 return）",
+            "",
+            "## 落库信号投影样例",
+            "| question_id | tier | cause_category | error_type | evolution_target |",
+            "|---|---|---|---|---|",
+            f"| q1 | {rows['q1'].tier} | {rows['q1'].cause_category} | "
+            f"{rows['q1'].error_type} | {rows['q1'].evolution_target} |",
+            f"| q2 | {rows['q2'].tier} | {rows['q2'].cause_category} | "
+            f"{rows['q2'].error_type} | {rows['q2'].evolution_target} |",
+            "",
+            "## tier 计数",
+            "- T2: 1（defect，可训练核心）",
+            "- T1: 1（lapse，低信号）",
+            "",
+            "## 结论",
+            "首次两题全落库，第二次 remaining 为空 → 续跑幂等成立；tier/evolution_target 投影正确。",
+        ],
+    )
+    assert md_path.exists()
+    store.close()
+
+
+@pytest.mark.asyncio
+async def test_infra_and_degraded_projection(tmp_path, monkeypatch):
+    """INFRA→T0、degraded→uncertain 的投影：对应字段置位、error_type/target 为 None。"""
+    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
+
+        return DiagnosisResult(
+            run_id=run_id,
+            error_attributions=[],
+            infra_question_ids=["q3"],
+            degraded_question_ids=["q4"],
+        )
+
+    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 = {"q3": _mk_q("q3"), "q4": _mk_q("q4")}
+
+    await run_baseline_diagnosis(
+        baseline_run_id="infer_adhoc",
+        diag_fingerprint="fp",
+        wrong_ids=["q3", "q4"],
+        questions=q_by_id,
+        store=store,
+        deps=deps,
+    )
+
+    rows = {r.question_id: r for r in store.load("infer_adhoc", "fp")}
+    assert rows["q3"].tier == "T0"
+    assert rows["q3"].infra is True
+    assert rows["q3"].error_type is None
+    assert rows["q3"].evolution_target is None
+    assert rows["q4"].tier == "uncertain"
+    assert rows["q4"].degraded is True
+    assert rows["q4"].error_type is None
+    assert store.done_question_ids("infer_adhoc", "fp") == {"q3", "q4"}
+
+    md_path = _write_md(
+        "test_infra_and_degraded_projection",
+        [
+            "# 离线诊断编排：INFRA / degraded 投影",
+            "",
+            "## 落库信号投影样例",
+            "| question_id | tier | infra | degraded | error_type | evolution_target |",
+            "|---|---|---|---|---|---|",
+            f"| q3 | {rows['q3'].tier} | {rows['q3'].infra} | {rows['q3'].degraded} | "
+            f"{rows['q3'].error_type} | {rows['q3'].evolution_target} |",
+            f"| q4 | {rows['q4'].tier} | {rows['q4'].infra} | {rows['q4'].degraded} | "
+            f"{rows['q4'].error_type} | {rows['q4'].evolution_target} |",
+            "",
+            "## 结论",
+            "INFRA→T0（infra 置位）、degraded→uncertain（degraded 置位），error_type/target 均 None。",
+        ],
+    )
+    assert md_path.exists()
+    store.close()