@@ -0,0 +1,694 @@
# 结果驱动的视频级切分 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（
**设计源 ** ： `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` （ ）
| 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` （ ）
| 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 行 ** （ **canonical 取行策略 = 每 question_id 取第一行 ** （ `ORDER BY rowid` ），全流程统一。
- **660 对 / 236 可诊断错题（非空 pred，非 INFRA） ： ）
- **`only_incorrect=True` + INFRA 排除**后进诊断的是 **236 题 ** ； **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（ ）
## 核心算法保真校验
本计划**不迁移/不改**核心算法（`ARCHITECTURE.md §6` ）。唯一相邻项是第 5 项"信息阶梯冷启动 2:1"（ `gate_ladder.py` ）：本计划只把 **pools 切分边界**改为 video， ** ， `gate_ladder` 的输入（unit + correctness）不变。**保真检查点（Task 9 Step 4） `Pools.diagnosis/validation` 内仍是逐 unit 列表，`gate_ladder.build_cold_entries` 消费不变、冷启动 2:1 未受影响。
---
## Phase 0：
### 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； （ ： ）
| 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 写规则**： ；
`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"；
" 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； """
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` （
- 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` （ ）
编排：算 `diag_fingerprint` → 查已完成集（续跑）→ 对剩余错题调 `run_diagnosis` （经 `StepsJsonRunLog` ）→ 投影 `error_attributions` + INFRA + degraded 为 `DiagnosisSignalRow` （ `score_signal` ，此处先落 error_type/cause_category，
- [ ] **Step 1: 写失败集成测试 ** （
``` 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` （ `score_signal` ， `infra_question_ids` （ ） `degraded_question_ids` （
> **错误处理实况（Codex I3） ： （ （
- [ ] **Step 4: 运行确认通过 ** 。
- [ ] **Step 5: 提交 ** — `git commit -m "feat: add offline baseline diagnosis orchestration"`
---
## Phase 2：
### 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` ；
- [ ] **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` （ `tier: str` ）：优先级 INFRA→T0； ； `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` ；
> **Codex 计划审 C4**：
- [ ] **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` ；
- [ ] **Step 1: 写失败测试 ** （输入为 Task 7 的 `VideoRecord` ；
``` 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：
### 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（ `_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` （ `VideoRecord` 模型（Task 7）；I1 val_wrong_min 接回切分（Task 12）、I2 `derive_reportable_types` （ `diag_fingerprint` 函数+测试；M1 锚点 809、M2 保真点 Task 9、M3 seed 只控预洗牌。
## 里程碑
| 里程碑 | 完成 Task |
|--------|----------|
| M1 诊断信号就绪（236 可诊断错题分层 + 4 INFRA-null 落 T0）
| 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，
