Files

T

iomgaa cfa510a7e5 docs: add results-driven video-split implementation plan

2026-07-15 11:38:32 -04:00

37 KiB

Raw Permalink Blame History

结果驱动的视频级切分 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: 写失败测试

# 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: 最小实现

# 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: 写失败测试

# 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: 最小实现

# 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: 写失败测试

# 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 追加：

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 题）

# 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: 写失败测试

# 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: 实现

# 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: 写失败测试

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 题结构）

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）

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: 写失败测试

# 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: 写失败测试

# 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 诊断行）

# 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 时抛错。

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 放松旋钮

37 KiB Raw Permalink Blame History Unescape Escape

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

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

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

核心算法保真校验

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

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

Phase 1：离线诊断管线

Task 1: steps_json → trace 行适配器

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

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

Task 4: 离线诊断编排

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

Task 5: evolution_target 派生 + cell 定义

Task 6: score_signal 分层

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

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

Phase 3：pools 视频原子 + 冻结

Task 9: pools 切分原子 unit→video

Task 10: 原子冻结 + manifest

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

Phase 4：配置与标定

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

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

里程碑

验收标准

风险

37 KiB

Raw Permalink Blame History