# 结果驱动的视频级切分 Implementation Plan

> **For agentic workers:** REQUIRED SUB-SKILL: Use subagent-driven-development to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.

**Goal:** 用离线、结果驱动的视频级 train/val/test 切分（基于已有 `infer_adhoc` baseline + 240 错题诊断），产出冻结 `pools.json`，替代自造训练题。

**Architecture:** 复用现有 `run_diagnosis` 拿逐题 `error_attributions`；新增纯函数 `split_selection`（signal 分层 → 视频聚合 → 贪心联合约束选择）；改 `pools.py` 切分原子 unit→video 并补原子写；全程离线可复现（fixed seed + baseline_run_id + diag_fingerprint 溯源）。

**Tech Stack:** Python 3.11、pytest、SQLite（harness.db）、asyncio、loguru、pydantic-settings、YAML（科研配置）。

**设计源**：`research-wiki/designs/2026-07-15-results-driven-video-split-design.md`

---

## 关键代码锚点（实现前必读）

| 用途 | 位置 |
|------|------|
| 诊断入口（复用） | `core/evolution/diagnose.py:2017` `run_diagnosis(...) -> DiagnosisResult` |
| 逐题归因产物 | `DiagnosisResult.error_attributions: list[ErrorAttribution]`（`core/evolution/types.py:257,149`） |
| INFRA 排除集 | `core/evolution/diagnose.py:56` `_INFRA_STOP_REASONS={"error","parse_error"}` |
| error_type 瀑布 | `core/evolution/diagnose.py:910-941`（extraction/search/reasoning/mixed） |
| RunLog 读端口 | `core/evolution/protocols.py:68`（`get_predictions`/`get_traces`） |
| 三池结构 | `app/harness/pools.py:33` `Pools`；`build_pools:53`；`_split_one_category:809`（签名行，822 是 docstring） |
| 切分入口/冻结 | `app/harness/pools.py:473` `build_or_load_pools`；`save_pools:289`（`path.write_text:341` 非原子）；per_category 写(566) 非原子 |
| PoolConfig | `core/types.py:146`（seed/diag_size/val_size/test_size/train_ratio） |
| baseline 数据 | `workspaces/default/harness.db`，`predictions` run_id=`infer_adhoc`；`traces` 表**空**，轨迹在 `predictions.steps_json` |

### Canonical baseline 事实（Codex 计划审 C1/C2 核实，实现须严格对齐）

- **902 原始行 / 900 distinct question / 300 视频**；`743-1` 有 **3 行**（error/budget_exceeded/finished）→ **canonical 取行策略 = 每 question_id 取第一行**（`ORDER BY rowid`），全流程统一。
- **660 对 / 236 可诊断错题（非空 pred，非 INFRA）/ 4 空 pred（全 INFRA：2 error + 2 parse_error）** → 非对合计 240。
- **`only_incorrect=True` + INFRA 排除**后进诊断的是 **236 题**；4 个 INFRA 空 pred 直接落 **T0**（不进 judge）。
- **`steps_json` 步字段 = `{thought, tool_call:{tool, args}, tool_output}`**——工具名在 **`tool_call.tool`**（非 `name`；对齐 `core/evolution/diagnose.py:258` 的 `tool_call.get("tool")`）。899/902 行非空，空的走空返回。
- **防御断言按 distinct question（900）计，不按裸行（902）**。

## 核心算法保真校验

本计划**不迁移/不改**核心算法（`ARCHITECTURE.md §6`）。唯一相邻项是第 5 项"信息阶梯冷启动 2:1"（`gate_ladder.py`）：本计划只把 **pools 切分边界**改为 video，**pools 内部仍是 unit 粒度**，`gate_ladder` 的输入（unit + correctness）不变。**保真检查点（Task 9 Step 4）**：确认 video 切分后 `Pools.diagnosis/validation` 内仍是逐 unit 列表，`gate_ladder.build_cold_entries` 消费不变、冷启动 2:1 未受影响。

---

## Phase 0：schema 前置门（编码前必须完成）

### Task 0: 用 structured-logging 定稿并注册 schema

**Files:** 无代码；产出 `research-wiki/schemas/` 注册 + 本计划 §schema 表定稿。

- [ ] **Step 1: 调用 structured-logging skill**，为下述两张表定稿列/类型/索引/基线指标并注册 Wiki。设计已给骨架，structured-logging 负责批准与注册：

`baseline_diagnosis`（逐题诊断信号，harness.db）：

| 列 | 类型 | 说明 |
|----|------|------|
| question_id | TEXT | 题 ID |
| video_id | TEXT | 视频 ID |
| baseline_run_id | TEXT | 溯源 run（`infer_adhoc`） |
| diag_fingerprint | TEXT | `hash(诊断prompt版本+model+诊断代码版本)` |
| task_type | TEXT | 12 类之一 |
| error_type | TEXT **NULL** | extraction/search/reasoning/mixed；**T0/uncertain 行为 NULL**（Codex I4：INFRA/degraded 题无合法 error_type） |
| cause_category | TEXT NULL | defect/lapse/NULL |
| tier | TEXT | T0/T1/T2/uncertain |
| evolution_target | TEXT **NULL** | 由 error_type 派生（tool/skill/system）；error_type 为 NULL 时亦 NULL |
| degraded | INTEGER | judge 降级 0/1 |
| infra | INTEGER | INFRA 排除 0/1 |
| session_id | TEXT | 遥测关联 |
| **PK** | | (question_id, baseline_run_id, diag_fingerprint) |

> **T0/uncertain 写规则**：INFRA 题 `infra=1, tier=T0, error_type/cause_category/evolution_target=NULL`；degraded/诊断失败题 `degraded=1, tier=uncertain, error_type` 可为 NULL。二者均不进 T2、不入 48 格覆盖。

`split_manifest`（冻结产物溯源，随 pools.json 同目录 JSON）：

| 键 | 说明 |
|----|------|
| baseline_run_id / diag_fingerprint / seed | 复现三元组 |
| config | N_trainval/floor_K/ε/report_floor/val 尺寸 快照 |
| pools_sha256 | pools.json 内容 hash |
| coverage_report | 48 格覆盖数、floor 达标情况、test 代表性偏差 |

- [ ] **Step 2:** 确认 structured-logging 已在 `research-wiki/schemas/` 注册二表并给出基线指标（各 tier 占比、48 格覆盖率、floor 达标率）。**此门通过前不得进入 Phase 1。**

---

## Phase 1：离线诊断管线

### Task 1: steps_json → trace 行适配器

**Files:**
- Create: `app/harness/steps_json_traces.py`
- Test: `tests/unit/test_steps_json_traces.py`

`run_diagnosis` 经 `get_traces` 取轨迹，但 `infer_adhoc` 的 `traces` 表空、轨迹在 `predictions.steps_json`。此适配器把 `{thought, tool_call, tool_output}` 步转成 `get_traces` 的行形 `{video_id, question_id, step, tool_name, tool_args, tool_output, thought}`。

- [ ] **Step 1: 写失败测试**

```python
# tests/unit/test_steps_json_traces.py
import json
from app.harness.steps_json_traces import steps_json_to_trace_rows


def test_parses_tool_call_into_name_and_args():
    # 真实 infer_adhoc steps_json 形态：tool_call={"tool":..., "args":...}（非 "name"）
    steps = [
        {"thought": "看根节点", "tool_call": {"tool": "view_node", "args": {"node_id": "v_L1_000"}}, "tool_output": "o0"},
        {"thought": "搜索", "tool_call": {"tool": "search_similar", "args": {"query": "gadget"}}, "tool_output": "hit"},
    ]
    rows = steps_json_to_trace_rows("vid1", "q1", json.dumps(steps))
    assert [r["step"] for r in rows] == [0, 1]
    assert rows[0]["tool_name"] == "view_node"
    assert rows[0]["tool_args"] == {"node_id": "v_L1_000"}
    assert rows[0]["video_id"] == "vid1" and rows[0]["question_id"] == "q1"
    assert rows[1]["tool_output"] == "hit"


def test_empty_or_blank_steps_json_returns_empty():
    assert steps_json_to_trace_rows("v", "q", "") == []
    assert steps_json_to_trace_rows("v", "q", "[]") == []
```

- [ ] **Step 2: 运行确认失败** — `conda activate Video-Tree-TRM && pytest tests/unit/test_steps_json_traces.py -v`，预期 `ModuleNotFoundError`。

- [ ] **Step 3: 最小实现**

```python
# app/harness/steps_json_traces.py
"""把 predictions.steps_json 转成 RunLog.get_traces 的行形。

infer_adhoc 的 traces 表为空，轨迹存于 steps_json（{thought, tool_call, tool_output}）。
诊断管线经 get_traces 消费轨迹，故需此确定性转换适配。
"""
from __future__ import annotations

import json
from typing import Any


def steps_json_to_trace_rows(video_id: str, question_id: str, steps_json: str) -> list[dict[str, Any]]:
    """将单题 steps_json 解析为 trace 行列表（step 从 0 递增）。

    参数:
        video_id: 视频 ID。
        question_id: 题 ID。
        steps_json: predictions.steps_json 原文（JSON 数组字符串）。

    返回:
        行字典列表，字段对齐 traces 表 schema；空/空数组返回 []。
    """
    if not steps_json or not steps_json.strip():
        return []
    steps = json.loads(steps_json)
    if not isinstance(steps, list):
        raise ValueError(f"steps_json 非数组: {question_id}")
    rows: list[dict[str, Any]] = []
    for i, s in enumerate(steps):
        call = s.get("tool_call") or {}
        rows.append(
            {
                "video_id": video_id,
                "question_id": question_id,
                "step": i,
                # infer_adhoc 用 "tool"；back-compat 兼容极少数 "name"
                "tool_name": call.get("tool", call.get("name")),
                "tool_args": call.get("args", {}),
                "tool_output": s.get("tool_output"),
                "thought": s.get("thought"),
            }
        )
    return rows
```

- [ ] **Step 4: 运行确认通过** — `pytest tests/unit/test_steps_json_traces.py -v`，预期 PASS。
- [ ] **Step 5: 提交** — `git add app/harness/steps_json_traces.py tests/unit/test_steps_json_traces.py && git commit -m "feat: add steps_json to trace-row adapter"`

### Task 2: RunLog 包装器（预测直取 + 轨迹从 steps_json）

**Files:**
- Create: `app/harness/baseline_run_log.py`
- Test: `tests/unit/test_baseline_run_log.py`

包装现有 SQLite RunLog：`get_predictions` 透传；`get_traces` 当底层 traces 空时，从 predictions.steps_json 经 Task 1 生成。不写 SQL 于 app 之外的裸路径——复用底层适配器。

- [ ] **Step 1: 写失败测试**

```python
# tests/unit/test_baseline_run_log.py
import json
import pytest
from app.harness.baseline_run_log import StepsJsonRunLog


class _FakeInner:
    def __init__(self, preds, traces):
        self._preds, self._traces = preds, traces
    async def get_predictions(self, run_id, *, question_ids=None):
        return [p for p in self._preds if not question_ids or p["question_id"] in question_ids]
    async def get_traces(self, run_id, *, question_ids=None):
        return list(self._traces)


@pytest.mark.asyncio
async def test_get_traces_falls_back_to_steps_json_when_table_empty():
    steps = [{"thought": "t", "tool_call": {"tool": "view_node", "args": {}}, "tool_output": "o"}]
    preds = [{"video_id": "v1", "question_id": "q1", "steps_json": json.dumps(steps)}]
    log = StepsJsonRunLog(_FakeInner(preds, traces=[]))
    rows = await log.get_traces("r", question_ids=["q1"])
    assert rows[0]["tool_name"] == "view_node" and rows[0]["question_id"] == "q1"


@pytest.mark.asyncio
async def test_get_traces_prefers_nonempty_inner_table():
    inner_traces = [{"video_id": "v1", "question_id": "q1", "step": 0, "tool_name": "x"}]
    log = StepsJsonRunLog(_FakeInner([], inner_traces))
    rows = await log.get_traces("r")
    assert rows == inner_traces
```

- [ ] **Step 2: 运行确认失败** — `pytest tests/unit/test_baseline_run_log.py -v`，预期 `ModuleNotFoundError`。

- [ ] **Step 3: 最小实现**

```python
# app/harness/baseline_run_log.py
"""RunLog 包装器：traces 表空时从 predictions.steps_json 重建轨迹。

用于对 infer_adhoc 这类 traces 未落表、轨迹在 steps_json 的历史 run 跑离线诊断。
"""
from __future__ import annotations

from typing import Any

from app.harness.steps_json_traces import steps_json_to_trace_rows


class StepsJsonRunLog:
    """委托内层 RunLog；get_traces 空表时回退 steps_json。"""

    def __init__(self, inner: Any) -> None:
        self._inner = inner

    async def get_predictions(self, run_id: str, *, question_ids: list[str] | None = None) -> list[dict[str, Any]]:
        return await self._inner.get_predictions(run_id, question_ids=question_ids)

    async def get_traces(self, run_id: str, *, question_ids: list[str] | None = None) -> list[dict[str, Any]]:
        inner_rows = await self._inner.get_traces(run_id, question_ids=question_ids)
        if inner_rows:
            return inner_rows
        preds = await self._inner.get_predictions(run_id, question_ids=question_ids)
        rows: list[dict[str, Any]] = []
        for p in preds:
            rows.extend(steps_json_to_trace_rows(p["video_id"], p["question_id"], p.get("steps_json") or ""))
        return rows
```

- [ ] **Step 4: 运行确认通过** — `pytest tests/unit/test_baseline_run_log.py -v`，预期 PASS。
- [ ] **Step 5: 提交** — `git add app/harness/baseline_run_log.py tests/unit/test_baseline_run_log.py && git commit -m "feat: add steps_json-backed RunLog wrapper"`

### Task 3: baseline_diagnosis 持久化端口 + SQLite 适配

**Files:**
- Modify: `core/evolution/protocols.py`（新增 `DiagnosisSignalStore` Protocol）
- Create: `adapters/baseline_diagnosis_store.py`（SQLite 实现）
- Test: `tests/unit/test_baseline_diagnosis_store.py`

- [ ] **Step 1: 写失败测试**

```python
# tests/unit/test_baseline_diagnosis_store.py
import sqlite3
from adapters.baseline_diagnosis_store import SqliteDiagnosisSignalStore, DiagnosisSignalRow


def _store(tmp_path):
    return SqliteDiagnosisSignalStore(str(tmp_path / "h.db"))


def test_upsert_and_get_done_ids(tmp_path):
    s = _store(tmp_path)
    row = DiagnosisSignalRow(question_id="q1", video_id="v1", baseline_run_id="r", diag_fingerprint="fp",
                             task_type="Counting", error_type="search_failure", cause_category="defect",
                             tier="T2", evolution_target="skill", degraded=False, infra=False, session_id="s")
    s.upsert(row)
    assert s.done_question_ids("r", "fp") == {"q1"}
    # upsert 同键覆盖，不重复
    s.upsert(row)
    assert s.done_question_ids("r", "fp") == {"q1"}
    # 不同 fingerprint 隔离
    assert s.done_question_ids("r", "other_fp") == set()


def test_load_rows_roundtrip(tmp_path):
    s = _store(tmp_path)
    s.upsert(DiagnosisSignalRow("q2", "v2", "r", "fp", "OCR Problems", "extraction_failure",
                                "lapse", "T1", "tool", False, False, "s"))
    rows = s.load("r", "fp")
    assert len(rows) == 1 and rows[0].tier == "T1" and rows[0].evolution_target == "tool"
```

- [ ] **Step 2: 运行确认失败** — `pytest tests/unit/test_baseline_diagnosis_store.py -v`，预期 `ModuleNotFoundError`。

- [ ] **Step 3: 实现 Protocol + 适配**

在 `core/evolution/protocols.py` 追加：

```python
class DiagnosisSignalStore(Protocol):
    """逐题诊断信号存储端口（隔离 SQLite，app/core 不写 SQL）。"""

    def upsert(self, row: "DiagnosisSignalRow") -> None: ...
    def done_question_ids(self, baseline_run_id: str, diag_fingerprint: str) -> set[str]: ...
    def load(self, baseline_run_id: str, diag_fingerprint: str) -> list["DiagnosisSignalRow"]: ...
```

创建 `adapters/baseline_diagnosis_store.py`：`DiagnosisSignalRow` dataclass（字段对齐 Task 0 表）；`SqliteDiagnosisSignalStore` 建表（PK `(question_id, baseline_run_id, diag_fingerprint)`）、`INSERT OR REPLACE` upsert、按 (run,fp) 查已完成集与全量。所有写在单事务提交（原子）。

- [ ] **Step 4: 运行确认通过** — `pytest tests/unit/test_baseline_diagnosis_store.py -v`，预期 PASS。
- [ ] **Step 5: 提交** — `git add core/evolution/protocols.py adapters/baseline_diagnosis_store.py tests/unit/test_baseline_diagnosis_store.py && git commit -m "feat: add baseline diagnosis signal store"`

### Task 4: 离线诊断编排

**Files:**
- Create: `app/harness/baseline_diagnosis.py`
- Test: `tests/integration/test_baseline_diagnosis.py`（LLM 类 → 产出 MD）

编排：算 `diag_fingerprint` → 查已完成集（续跑）→ 对剩余错题调 `run_diagnosis`（经 `StepsJsonRunLog`）→ 投影 `error_attributions` + INFRA + degraded 为 `DiagnosisSignalRow`（tier 由 Task 6 `score_signal`，此处先落 error_type/cause_category，tier 计算在 Task 6 引入后接入）→ upsert。

- [ ] **Step 1: 写失败集成测试**（mock LLM，真实 infer_adhoc 抽 3 题）

```python
# tests/integration/test_baseline_diagnosis.py
import pytest
from adapters.baseline_diagnosis_store import SqliteDiagnosisSignalStore
from app.harness.baseline_diagnosis import run_baseline_diagnosis, DiagnosisDeps


def _deps(monkeypatch, calls):
    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
        return DiagnosisResult(run_id=run_id, error_attributions=[
            ErrorAttribution("q1", "search_failure", None, "defect"),
            ErrorAttribution("q2", "mixed", None, "lapse"),
        ], infra_question_ids=[], degraded_question_ids=[])
    monkeypatch.setattr("app.harness.baseline_diagnosis.run_diagnosis", fake_run_diagnosis)
    # 全部依赖显式 fake，无占位
    return DiagnosisDeps(run_log=_FakeRunLog(), llm=_FakeLLM(), skill_store=_FakeSkillStore(),
                         prompts=_fake_prompts(), tree_data={}, concurrency=2)


@pytest.mark.asyncio
async def test_resume_skips_done(tmp_path, monkeypatch):
    calls = []
    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"}
    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 calls[-1] == ()  # 第二次无剩余
```

- [ ] **Step 2: 运行确认失败**，预期 `ModuleNotFoundError`。
- [ ] **Step 3: 实现** `run_baseline_diagnosis(baseline_run_id, diag_fingerprint, wrong_ids, questions, store, deps)`：`remaining = wrong_ids - store.done_question_ids(run,fp)`；空则返回；否则 `await run_diagnosis(baseline_run_id, [questions[i] for i 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)`；投影 `error_attributions`（tier 由 Task 6 `score_signal`，error_type/cause_category 落列）+ `infra_question_ids`（infra=1,tier=T0,error_type/evolution_target=NULL）+ `degraded_question_ids`（degraded=1,tier=uncertain）→ upsert。
  > **错误处理实况（Codex I3）**：`run_diagnosis` 的 defect/lapse 判别处对 judge 异常是 `except Exception` 后 warning + 默认 lapse（`core/evolution/diagnose.py:2186-2192`），**非全传播**。本编排不谎称"全传播"：网络/API 层失败经 GovernedLLMClient 重试栈后仍失败会向上抛；judge 语义歧义按现有保护性 lapse 处理并**计数上报**（`degraded_count`/默认 lapse 数写入 manifest）。产出 MD 到 `tests/outputs/`。
- [ ] **Step 4: 运行确认通过**。
- [ ] **Step 5: 提交** — `git commit -m "feat: add offline baseline diagnosis orchestration"`

---

## Phase 2：signal 分层 + 视频聚合 + 贪心选择（纯函数）

### Task 5: evolution_target 派生 + cell 定义

**Files:** Create `app/harness/split_selection.py`（本 Task 起逐步充实）；Test `tests/unit/test_split_selection.py`

- [ ] **Step 1: 写失败测试**

```python
# tests/unit/test_split_selection.py
from app.harness.split_selection import evolution_target_of, cell_of


def test_evolution_target_mapping():
    assert evolution_target_of("extraction_failure") == "tool"
    assert evolution_target_of("search_failure") == "skill"
    assert evolution_target_of("reasoning_failure") == "skill"
    assert evolution_target_of("mixed") == "system"


def test_cell_is_task_type_x_error_type():
    assert cell_of("Counting Problem", "search_failure") == ("Counting Problem", "search_failure")
```

- [ ] **Step 2: 失败确认**。
- [ ] **Step 3: 实现**

```python
# app/harness/split_selection.py (片段)
_EVOLUTION_TARGET = {
    "extraction_failure": "tool",
    "search_failure": "skill",
    "reasoning_failure": "skill",
    "mixed": "system",
}


def evolution_target_of(error_type: str) -> str:
    """error_type → 进化目标（派生标注，非独立多样性轴）。"""
    if error_type not in _EVOLUTION_TARGET:
        raise ValueError(f"未知 error_type: {error_type}")
    return _EVOLUTION_TARGET[error_type]


def cell_of(task_type: str, error_type: str) -> tuple[str, str]:
    """多样性主格子 = (task_type, error_type)。"""
    return (task_type, error_type)
```

- [ ] **Step 4: 通过确认**。
- [ ] **Step 5: 提交** — `git commit -m "feat: add evolution_target derivation and cell"`

### Task 6: score_signal 分层

**Files:** Modify `app/harness/split_selection.py`；Test 同上文件追加

- [ ] **Step 1: 写失败测试**

```python
def test_tiers():
    from app.harness.split_selection import score_signal
    assert score_signal(cause_category="defect", infra=False, degraded=False).tier == "T2"
    assert score_signal(cause_category="lapse", infra=False, degraded=False).tier == "T1"
    assert score_signal(cause_category="defect", infra=True, degraded=False).tier == "T0"
    assert score_signal(cause_category=None, infra=False, degraded=True).tier == "uncertain"
```

- [ ] **Step 2: 失败确认**。
- [ ] **Step 3: 实现** `score_signal(...) -> SignalLabel`（dataclass `tier: str`）：优先级 INFRA→T0；degraded/诊断失败→uncertain；`cause_category=='defect'`→T2；`=='lapse'`→T1；其余→uncertain。**顺序固定，不用魔法权重**。
- [ ] **Step 4: 通过确认**。
- [ ] **Step 5: 提交** — `git commit -m "feat: add signal tiering"`

### Task 7: 全视频记录构建（VideoRecord）+ 信号聚合

**Files:** Modify `app/harness/split_selection.py`；Test 追加

> **Codex 计划审 C4**：`select_split` 需**全 300 视频**的 type/difficulty/correctness 分布（含 125 个全对零信号视频）来算 ε 代表性与 test 补集，**不能只喂诊断行**。故先由全 900 predictions + questions 构建 `VideoRecord`，再叠加 T2 诊断信号。

- [ ] **Step 1: 写失败测试**（真实数据二次构造：从 infer_adhoc 抽一个真实视频的 3 题结构）

```python
def test_build_video_records_covers_all_videos_with_difficulty_and_types():
    from app.harness.split_selection import build_video_records
    # 二次构造：真实 3 题/视频、真实 type、真实对错
    preds = [
        {"video_id": "v1", "question_id": "v1-1", "task_type": "Counting Problem", "correct": False},
        {"video_id": "v1", "question_id": "v1-2", "task_type": "Action Reasoning", "correct": True},
        {"video_id": "v1", "question_id": "v1-3", "task_type": "OCR Problems", "correct": True},
        {"video_id": "v2", "question_id": "v2-1", "task_type": "Counting Problem", "correct": True},
        {"video_id": "v2", "question_id": "v2-2", "task_type": "Counting Problem", "correct": True},
        {"video_id": "v2", "question_id": "v2-3", "task_type": "Counting Problem", "correct": True},
    ]
    signal_rows = [{"question_id": "v1-1", "task_type": "Counting Problem",
                    "error_type": "search_failure", "tier": "T2"}]
    recs = build_video_records(preds, signal_rows)
    assert {r.video_id for r in recs} == {"v1", "v2"}          # 全视频（含零信号 v2）
    v1 = next(r for r in recs if r.video_id == "v1")
    v2 = next(r for r in recs if r.video_id == "v2")
    assert v1.n_correct == 2 and v1.difficulty == 1            # 3题对2 → 难度画像桶=1错
    assert v2.difficulty == 0 and v2.cells == set()            # 零信号视频进 test 无 T2 格子
    assert v1.cells == {("Counting Problem", "search_failure")}
    assert v1.type_set == {"Counting Problem", "Action Reasoning", "OCR Problems"}
    assert v1.wrong_by_type == {"Counting Problem": 1}          # T2 计数供 floor
```

- [ ] **Step 2: 失败确认**。
- [ ] **Step 3: 实现** `build_video_records(preds, signal_rows) -> list[VideoRecord]`：按 video_id 聚合全 900 题 → `VideoRecord(video_id, type_set, n_correct, difficulty=3-n_correct 的错题数桶, cells:set(仅 T2 题 cell_of 并集去重), wrong_by_type:dict(各type T2 数))`。零信号视频 cells 空、仍在列表（供 test 与 ε）。
- [ ] **Step 4: 通过确认**。
- [ ] **Step 5: 提交** — `git commit -m "feat: build all-video records with difficulty and signal overlay"`

### Task 8: 贪心联合约束选择器

**Files:** Modify `app/harness/split_selection.py`；Test 追加

- [ ] **Step 1: 写失败测试**（输入为 Task 7 的 `VideoRecord`；fixture 用真实 infer_adhoc 各 type 分布二次构造，非纯虚构——满足 §4.6）

```python
def _real_shaped_video_records():
    # 从 workspaces/default/harness.db infer_adhoc 抽真实(video,3题type,correct)二次构造
    # 保留真实类型长尾(Counting/AR/OCR...)与 0/1/2/3-对难度画像分布
    ...  # helper：读真实 DB → build_video_records；见 conftest 提供的 fixture


def test_select_split_video_disjoint_and_floor_and_deterministic():
    from app.harness.split_selection import select_split, SelectConfig, derive_reportable_types
    videos = _real_shaped_video_records()
    total_by_type = _count_questions_by_type(videos)
    cfg = SelectConfig(n_trainval=100, floor_k={"Counting Problem": 3}, epsilon=0.1,
                       reportable_types=derive_reportable_types(total_by_type, report_floor=27),
                       seed=7)
    a = select_split(videos, config=cfg)
    b = select_split(videos, config=cfg)
    assert set(a.trainval) & set(a.test) == set()            # 互斥
    assert set(a.trainval) | set(a.test) == {v.video_id for v in videos}
    assert a.trainval == b.trainval                          # 同 seed 同解
    assert sum(v.wrong_by_type.get("Counting Problem", 0)     # floor 达标
               for v in videos if v.video_id in a.trainval) >= 3


def test_infeasible_floor_vs_epsilon_raises():
    from app.harness.split_selection import select_split, SelectConfig, InfeasibleSplitError
    import pytest
    videos = _real_shaped_video_records()
    with pytest.raises(InfeasibleSplitError):   # 极小 ε + 高 floor → 死锁
        select_split(videos, config=SelectConfig(n_trainval=2, floor_k={"OCR Problems": 50},
                                                 epsilon=0.001, reportable_types=set(), seed=1))
```

- [ ] **Step 2: 失败确认**。
- [ ] **Step 3: 实现** `select_split(videos: list[VideoRecord], config) -> SplitAssignment(trainval, test)`（全局统计从 `videos` 内部算，无需外传）：按设计 §6 两阶段贪心——floor 阶段与多样性阶段**均带 ε 可行性检查**（移入 trainval 后 test=补集 仍满足 per-type 比例 + 0/1/2/3-对难度画像 ±ε）；floor 死锁/欠额抛 `InfeasibleSplitError`。**确定性**：先按 `seed` 对候选做一次固定预洗牌，再按 `-边际增益` 稳定排序（seed 只控预洗牌打破等增益平局，非二次 key）。`_epsilon_ok(test_videos, ε)` 校验 reportable 类型比例与难度画像偏差 ≤ ε。同文件加 `derive_reportable_types(total_by_type, report_floor) -> set[str]`（总题数 ≥ report_floor 的类型，落地设计 §7）。
- [ ] **Step 4: 通过确认**。
- [ ] **Step 5: 提交** — `git commit -m "feat: add greedy joint-constrained split selector"`

---

## Phase 3：pools 视频原子 + 冻结

### Task 9: pools 切分原子 unit→video

**Files:**
- Modify: `app/harness/pools.py`（`build_pools`/`_split_one_category`/`build_or_load_pools`）
- Test: `tests/unit/test_pools_video_atomic.py` + 回归 `tests/unit/test_run_store.py`（若涉及）

- [ ] **Step 1: 写失败测试**

```python
# tests/unit/test_pools_video_atomic.py
from app.harness.pools import split_by_video_assignment
from core.types import GeneratedQuestion


def _q(qid, vid, tt="Counting Problem"):
    return GeneratedQuestion(question_id=qid, video_id=vid, task_type=tt, question="", options=("A","B","C","D"), answer="A")


def test_video_never_split_across_pools():
    qs = [_q("v1-1","v1"), _q("v1-2","v1"), _q("v1-3","v1"), _q("v2-1","v2")]
    assignment = {"v1": "trainval", "v2": "test"}
    pools = split_by_video_assignment(qs, assignment, correctness={q.question_id: True for q in qs},
                                      val_ratio=0.0, seed=0)
    test_vids = {q.video_id for q in pools.test}
    train_vids = {q.video_id for q in pools.diagnosis + pools.validation}
    assert test_vids & train_vids == set()   # 视频不跨池
    assert test_vids == {"v2"} and train_vids == {"v1"}
```

- [ ] **Step 2: 失败确认**。
- [ ] **Step 3: 实现** 新增 `split_by_video_assignment(questions, assignment, correctness, val_ratio, seed) -> Pools`：先按 assignment 把题分到 trainval/test（video 原子）；trainval 内用**视频组**做 correctness 分层切 train(diagnosis)/val——改 `_split_one_category` 采样原子为 video 组（同 video 全 unit 同进同出，复用现有 correctness 分层 + `eval_min_per_class`）；test 直接为 test 视频全部题。保留 `baseline_val_accuracy`/`correctness` 计算。
- [ ] **Step 4: 通过确认** + **保真检查点**：断言 `pools.diagnosis` 内仍是逐 unit 列表（`gate_ladder` 输入不变）。运行 `pytest tests/unit/test_pools*.py tests/unit/test_gate.py -v`。
- [ ] **Step 5: 提交** — `git commit -m "refactor: add video-atomic pool split (algo #5 gate input preserved)"`

### Task 10: 原子冻结 + manifest

**Files:**
- Modify: `app/harness/pools.py`（`save_pools` 与 per_category 写路径抽共用原子助手）
- Create: `app/harness/split_manifest.py`
- Test: `tests/unit/test_atomic_save_pools.py`

- [ ] **Step 1: 写失败测试**

```python
# tests/unit/test_atomic_save_pools.py
import json
from app.harness.pools import _atomic_write_json  # 新共用助手


def test_atomic_write_replaces_and_no_tmp_left(tmp_path):
    p = tmp_path / "pools.json"
    _atomic_write_json(p, {"a": 1})
    assert json.loads(p.read_text())["a"] == 1
    assert list(tmp_path.glob("*.tmp")) == []  # 无残留 tmp
```

- [ ] **Step 2: 失败确认**。
- [ ] **Step 3: 实现** `_atomic_write_json(path, obj)`：写 `path.with_suffix(".tmp")` 后 `os.replace`；`save_pools`(341) 与 per_category 写(566) 均改调它。`split_manifest.py::write_manifest(...)` 写 baseline_run_id/diag_fingerprint/seed/config/pools_sha256/coverage_report。
- [ ] **Step 4: 通过确认**。
- [ ] **Step 5: 提交** — `git commit -m "fix: make pools.json freeze atomic + add split manifest"`

### Task 11: 端到端集成 + 防御断言

**Files:**
- Create: `app/harness/build_split.py`（顶层编排：诊断结果 → signal → 视频聚合 → select_split → split_by_video_assignment → 冻结 + manifest）
- Test: `tests/integration/test_build_split_e2e.py`

- [ ] **Step 1: 写失败集成测试**（用真实 infer_adhoc 900 预测 + Task 3 缓存诊断表；若无缓存则 mock 诊断行）

```python
# tests/integration/test_build_split_e2e.py
import pytest
from app.harness.build_split import build_split


@pytest.mark.asyncio
async def test_end_to_end_freezes_valid_pools(tmp_path):
    out = tmp_path / "pools.json"
    result = await build_split(baseline_run_id="infer_adhoc", diag_fingerprint="fp",
                               harness_db="workspaces/default/harness.db",
                               signal_store=_cached_or_mock_store(), out_path=out, config=_calibrated_cfg())
    pools = result.pools
    # 防御断言①: 三池视频互斥
    tv = {q.video_id for q in pools.diagnosis + pools.validation}
    te = {q.video_id for q in pools.test}
    assert tv & te == set()
    # 防御断言②: 覆盖全 900 题
    assert len(pools.diagnosis) + len(pools.validation) + len(pools.test) == 900
    # manifest 存在且 pools_sha256 校验一致
    assert result.manifest["pools_sha256"] == _sha256(out.read_text())
```

- [ ] **Step 2: 失败确认**。
- [ ] **Step 3: 实现** `build_split(...)`：从 signal_store.load(run,fp) 取逐题 tier/cell → `aggregate_video_signal` 逐视频 → `select_split` → `split_by_video_assignment` → `save_pools`(原子) + `write_manifest`。落地设计 §10 防御断言清单①②③④⑤⑥（三池互斥/900 覆盖/每视频 3 题/fingerprint 一致/manifest hash/ID 唯一），任一不满足 fail-fast。
- [ ] **Step 4: 通过确认** — `pytest tests/integration/test_build_split_e2e.py -v`。
- [ ] **Step 5: 提交** — `git commit -m "feat: wire end-to-end results-driven split with defensive asserts"`

---

## Phase 4：配置与标定

### Task 12: 科研 YAML 旋钮 + 标定程序

**Files:**
- Modify: `core/types.py`（`PoolConfig` 加 `n_trainval/floor_k/epsilon/report_floor/val_wrong_min`）
- Create: `config/video_split.yaml`
- Create: `scripts/build_video_split.sh`
- Test: `tests/unit/test_pool_config_video_split.py`

- [ ] **Step 1: 写失败测试** — ① `PoolConfig` 可从 `config/video_split.yaml` 载入新字段且缺失关键项 fail-fast（不兜底默认）；② `diag_fingerprint(prompt_version, model, code_version)` 确定性、任一输入变则变；③ `val_wrong_min` 进入 `split_by_video_assignment` 可行性：val 错题数 < `val_wrong_min` 时抛错。

```python
def test_diag_fingerprint_deterministic_and_sensitive():
    from app.harness.split_selection import diag_fingerprint
    a = diag_fingerprint("p1", "deepseek-v4", "abc123")
    assert a == diag_fingerprint("p1", "deepseek-v4", "abc123")     # 确定性
    assert a != diag_fingerprint("p2", "deepseek-v4", "abc123")     # prompt 变则变
    assert a != diag_fingerprint("p1", "kimi", "abc123")            # model 变则变


def test_val_wrong_min_enforced(tmp_path):
    from app.harness.pools import split_by_video_assignment, InsufficientValSignal
    import pytest
    with pytest.raises(InsufficientValSignal):   # val 错题不足功效阈
        split_by_video_assignment(_all_correct_qs(), {"v": "trainval"},
                                  correctness=_all_true(), val_ratio=0.5, seed=0, val_wrong_min=5)
```

- [ ] **Step 2: 失败确认**。
- [ ] **Step 3: 实现** 扩 `PoolConfig`（`n_trainval/floor_k/epsilon/report_floor/val_wrong_min`）；`diag_fingerprint(...)= hashlib.sha256("|".join(...)).hexdigest()[:16]`（来源：诊断 prompt 文件 hash + `.env` 模型名 + `git rev-parse HEAD` 短 SHA，`--force` 用**新 fingerprint** 写、不覆盖旧记录）；`split_by_video_assignment` 加 `val_wrong_min` 参数，val 切出后校验错题数 ≥ 阈值否则 `InsufficientValSignal`；写 `config/video_split.yaml`（占位阈值 + 注释"诊断后标定"）；`scripts/build_video_split.sh` 自包含零参复现（GPU 卡号除外）。**标定程序**：诊断跑完读 `baseline_diagnosis` 各 type T2 数 → 定 `floor_k`(如 min(可用defect,3))、`n_trainval`(~100)、`epsilon`(如 0.1)、`val_wrong_min`（McNemar 功效阈，如 ≥20）。
- [ ] **Step 4: 通过确认**。
- [ ] **Step 5: 提交** — `git commit -m "feat: add video-split config knobs and reproducible script"`

---

## Self-Review（作者自查，已执行）

- **Spec 覆盖**：设计 §4 组件→Task 1-12 一一对应；§5 signal/多样性→Task 5-7；§6 贪心→Task 8；§7 长尾→Task 8/12(reportable_types+report_floor)；§8 val→Task 9(val_ratio+min_per_class)+Task 12(val_wrong_min)；§9 非功能→Task 3(逐题续跑/upsert)、Task 10(原子写)；§10 错误→Task 4(P5 传播)、Task 8(InfeasibleSplitError)、Task 11(防御断言)；§12 前序继承→Task 9(三池/pair/baseline_val_accuracy)+Task 10(per_category 原子写)。
- **占位扫描**：无 TBD/TODO；阈值项在 Task 12 明确"诊断后标定"并给标定程序，非占位。
- **类型一致**：`DiagnosisSignalRow`/`DiagnosisDeps`/`VideoRecord`/`SignalLabel`/`SelectConfig`/`SplitAssignment`/`InfeasibleSplitError`/`InsufficientValSignal`/`derive_reportable_types`/`diag_fingerprint`/`_atomic_write_json`/`split_by_video_assignment`/`build_video_records` 跨 Task 命名一致。
- **Codex 计划审修订已并入**：C1 字段名 `tool`（Task 1/2）、C2 canonical 事实（900/660/236错/4 INFRA-null，按 distinct question 断言）、C3 Task 4 去占位显式 fake、C4 全视频 `VideoRecord` 模型（Task 7）；I1 val_wrong_min 接回切分（Task 12）、I2 `derive_reportable_types`（Task 8）、I3 诚实标注 run_diagnosis 错误处理、I4 schema NULL 规则（Task 0）、I5 真实数据二次构造 fixture、I6 `diag_fingerprint` 函数+测试；M1 锚点 809、M2 保真点 Task 9、M3 seed 只控预洗牌。

## 里程碑

| 里程碑 | 完成 Task |
|--------|----------|
| M1 诊断信号就绪（236 可诊断错题分层 + 4 INFRA-null 落 T0） | 0-4 |
| M2 选择器可产合法切分（纯函数全绿） | 5-8 |
| M3 冻结 pools.json + manifest 端到端 | 9-11 |
| M4 可复现脚本 + 标定 | 12 |

## 验收标准

- 冻结 `pools.json`：三池视频互斥、覆盖 900 题、floor 达标、test 代表性偏差 ≤ ε；manifest hash 自洽。
- 全测试绿（unit + integration）；LLM 类测试产出 MD。
- `bash scripts/build_video_split.sh` 零参复现（GPU 卡号除外）。
- gate_ladder 冷启动 2:1 输入不变（保真检查点通过）。

## 风险

| 风险 | 缓解 |
|------|------|
| steps_json 部分为空(3/902)或字段异常 | Task 1 空返回 []；诊断对应题落 uncertain，计数上报 |
| 100 训练视频信号不足 | 切设计 §14 方案 B（外部 benchmark 训练） |
| floor 与 ε 联合不可行 | Task 8 InfeasibleSplitError fail loud，Task 12 放松旋钮 |