Video-Tree-TRM5/research-wiki/plans/2026-07-15-question-gen-v3-phase1-contract.md

question-gen v3 — Phase 1 契约地基 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: 建立 QuestionUnit(single|pair) 契约并让它贯穿 harness 采样/分批/聚合/续跑/gate 全部拆-pair 入口，使后续阶段产出的 AR pair 题能被评测/训练正确消费，且 11 非 AR single 题行为字节级不变。

Architecture: 引入 QuestionUnit 判别联合（core 领域实体）+ question_units.py helper（build/flatten/validate/unit-correctness）。改造 pools/loader/batching/inference/runner/gate_ladder 使 pair 同池、同 batch 整锁、配对聚合（双向 AND）、同池切分；correctness 分"逐题 predictions（溯源）/ unit correctness（进化消费）"两口径；pair 原子成对落盘；gate_ladder 加 schema_version 迁移；非 AR rng 独立 namespace。在现有 app/question_gen/ 内原地改，不建 _v3 目录。

Tech Stack: Python 3.11 / pytest / SQLite（run_store）/ asyncio。全程 conda run -n Video-Tree-TRM。

依据：设计 research-wiki/designs/2026-07-15-question-gen-v3-construction-paradigm-design.md §5/§8/§12；日志 schema research-wiki/schemas/v3-question-gen-logging.md。本阶段不产出题目，只建契约，交付物 = 全套 pair-契约回归测试绿 + 非 AR byte-identical 绿。

---

文件结构映射

文件	职责	改动
core/types.py	领域实体	加 QuestionUnit + GeneratedQuestion 4 字段
app/harness/question_units.py	unit helper（新建）	build/flatten/validate_units/unit_correctness
app/harness/pools.py	三池切分	build_pools/_sample_excluding/_split_one_category 以 unit 为原子 + _q_to_dict/_dict_to_q 序列化 pair 字段
app/question_gen/loader.py	采样/加载	stratified_sample 按 unit + load_benchmark 读回 pair 字段
app/harness/batching.py	分批	build_batches unit 整锁 + _select_mixed_by_task_type 按 unit correctness 分桶 + 非 AR 独立 rng
app/harness/inference.py	聚合	run_inference pair-level 双向 AND + unit 粒度 total/correct
app/harness/runner.py	训练主环	correctness 消费点改 unit 视图 + checkpoint 存 unit_id 序列
app/harness/gate_ladder.py	信息阶梯	entry 迁 unit_id + schema_version 迁移 + BaselineCache unit 键
core/evolution/validate.py	e-process 统计	pair_block/compute_accuracy 按 unit
app/harness/validate.py	gate 块实际执行（Codex C-2）	baseline/candidate block、baseline_cache.get/put、n_used 按 unit
app/question_gen/run_store.py	落库（唯一 run_store，非 app/harness/）	加 facts/unit_verdict/collapse_metrics/quarantine/resume_state 表
app/question_gen/pair_atomic_writer.py	pair 原子写 helper（新建）	pending buffer + os.replace + 孤儿剔除（wiring 挪 Phase 2）

---

Task 1: QuestionUnit 领域实体 + GeneratedQuestion 字段扩展

Files:

• Modify: core/types.py

• Test: tests/unit/test_question_unit.py

• Step 1: 写失败测试

# tests/unit/test_question_unit.py
from core.types import GeneratedQuestion, QuestionUnit

def _q(qid, role="single", pair_id=None, flip_axis=None):
    return GeneratedQuestion(
        question_id=qid, video_id="v", task_type="Action Recognition",
        question="?", options=("A. a", "B. b", "C. c", "D. d"), answer="A",
        source_nodes=("n1",), difficulty="hard",
        unit_id=pair_id or qid, pair_id=pair_id,
        question_role=role, flip_axis=flip_axis,
    )

def test_single_defaults_backward_compatible():
    q = GeneratedQuestion(question_id="q", video_id="v", task_type="X",
        question="?", options=("A. a","B. b","C. c","D. d"), answer="A",
        source_nodes=("n1",), difficulty="hard")
    assert q.question_role == "single"
    assert q.pair_id is None and q.flip_axis is None and q.unit_id == "q"

def test_pair_unit_carries_two_questions():
    p = _q("q_o", "pair_original", "pid", "before_after")
    m = _q("q_m", "pair_mirror", "pid", "before_after")
    u = QuestionUnit.from_pair(p, m)
    assert u.kind == "pair" and u.size == 2 and u.unit_id == "pid"
    assert {qq.question_id for qq in u.questions} == {"q_o", "q_m"}

• Step 2: 跑测试确认失败 — Run: conda run -n Video-Tree-TRM pytest tests/unit/test_question_unit.py -v — Expected: FAIL（QuestionUnit 未定义 / 字段缺失）

• Step 3: 实现

# core/types.py —— GeneratedQuestion 增字段（默认值保非 AR 不变）
# 在 GeneratedQuestion dataclass 追加：
    unit_id: str = ""                 # 默认在 __post_init__ 回填为 question_id
    pair_id: str | None = None
    question_role: str = "single"     # single | pair_original | pair_mirror
    flip_axis: str | None = None
# __post_init__ 中：if not self.unit_id: object.__setattr__(self, "unit_id", self.pair_id or self.question_id)

@dataclass(frozen=True)
class QuestionUnit:
    kind: str                          # "single" | "pair"
    unit_id: str
    task_type: str
    questions: tuple[GeneratedQuestion, ...]
    unit_hash: str = ""                 # P/Q payload 合成 hash，断点续跑失效检测（T11 用）
    # 注：设计 §8 的 collapse_metric 是验收产物，不进 Phase 1 契约实体，Phase 3/5 落 collapse_metrics 表
    @property
    def size(self) -> int: return len(self.questions)
    @classmethod
    def from_single(cls, q):
        return cls("single", q.unit_id, q.task_type, (q,))
    @classmethod
    def from_pair(cls, original, mirror):
        assert original.pair_id and original.pair_id == mirror.pair_id
        assert original.video_id == mirror.video_id and original.task_type == mirror.task_type
        assert original.flip_axis == mirror.flip_axis
        return cls("pair", original.pair_id, original.task_type, (original, mirror))

• Step 4: 跑测试确认通过 — Run: 同上 — Expected: PASS
• Step 5: 提交 — git add core/types.py tests/unit/test_question_unit.py && git commit（用 commit skill）

---

Task 2: question_units.py helper（组装/展开/校验/unit correctness）

Files:

• Create: app/harness/question_units.py

• Test: tests/unit/test_question_units_helper.py

• Step 1: 写失败测试

# tests/unit/test_question_units_helper.py
from app.harness.question_units import build_units, flatten_units, validate_units, unit_correctness
from core.types import GeneratedQuestion

def _q(qid, role="single", pid=None):
    return GeneratedQuestion(question_id=qid, video_id="v", task_type="AR",
        question="?", options=("A. a","B. b","C. c","D. d"), answer="A",
        source_nodes=("n",), difficulty="hard", pair_id=pid, question_role=role,
        unit_id=pid or qid, flip_axis="ax" if pid else None)

def test_build_units_groups_pair_and_keeps_single():
    qs = [_q("s1"), _q("po","pair_original","p"), _q("pm","pair_mirror","p")]
    units = build_units(qs)
    kinds = sorted(u.kind for u in units)
    assert kinds == ["pair", "single"]

def test_validate_units_rejects_orphan_pair():
    import pytest
    with pytest.raises(ValueError):
        validate_units(build_units([_q("po","pair_original","p")]))  # 只 1 条

def test_flatten_roundtrip():
    qs = [_q("po","pair_original","p"), _q("pm","pair_mirror","p")]
    assert {q.question_id for q in flatten_units(build_units(qs))} == {"po","pm"}

def test_unit_correctness_bidirectional_and():
    qs = [_q("po","pair_original","p"), _q("pm","pair_mirror","p")]
    u = build_units(qs)[0]
    assert unit_correctness(u, {"po": True, "pm": True}) is True
    assert unit_correctness(u, {"po": True, "pm": False}) is False  # AND

• Step 2: 跑确认失败 — conda run -n Video-Tree-TRM pytest tests/unit/test_question_units_helper.py -v — Expected: FAIL

• Step 3: 实现

# app/harness/question_units.py
"""QuestionUnit 组装/展开/校验/单元正确性——pair 契约唯一入口。"""
from collections import defaultdict
from core.types import GeneratedQuestion, QuestionUnit

def build_units(questions: list[GeneratedQuestion]) -> list[QuestionUnit]:
    by_pair: dict[str, list[GeneratedQuestion]] = defaultdict(list)
    singles: list[QuestionUnit] = []
    for q in questions:
        if q.pair_id:
            by_pair[q.pair_id].append(q)
        else:
            singles.append(QuestionUnit.from_single(q))
    pairs = []
    for pid, qs in by_pair.items():
        if len(qs) != 2:
            raise ValueError(f"pair {pid} 数量={len(qs)}≠2（孤儿）")
        o = next(q for q in qs if q.question_role == "pair_original")
        m = next(q for q in qs if q.question_role == "pair_mirror")
        pairs.append(QuestionUnit.from_pair(o, m))
    return singles + pairs

def flatten_units(units: list[QuestionUnit]) -> list[GeneratedQuestion]:
    return [q for u in units for q in u.questions]

def validate_units(units: list[QuestionUnit]) -> list[QuestionUnit]:
    for u in units:
        if u.kind == "pair" and u.size != 2:
            raise ValueError(f"unit {u.unit_id} pair 不成对")
    return units

def unit_correctness(unit: QuestionUnit, per_q: dict[str, bool]) -> bool:
    """AR pair = P AND Q；single = 单题。缺任一条 → KeyError（防静默）。"""
    return all(per_q[q.question_id] for q in unit.questions)

• Step 4: 跑确认通过 — Expected: PASS
• Step 5: 提交

---

Task 3: pools.py 三池切分以 unit 为原子（pair 同池）

Files:

• Modify: app/harness/pools.py（build_pools / _sample_excluding / _split_one_category）

• Test: tests/unit/test_pools_pair_atomic.py

• Step 1: 写失败测试（pair 两题必同池，不被 progressive exclusion 劈开）

# tests/unit/test_pools_pair_atomic.py
from app.harness.pools import build_pools
from core.types import GeneratedQuestion

def _pair(pid):
    base = dict(video_id="v", task_type="AR", question="?",
        options=("A. a","B. b","C. c","D. d"), answer="A",
        source_nodes=("n",), difficulty="hard", pair_id=pid, unit_id=pid, flip_axis="ax")
    return [GeneratedQuestion(question_id=f"{pid}_o", question_role="pair_original", **base),
            GeneratedQuestion(question_id=f"{pid}_m", question_role="pair_mirror", **base)]

def test_pair_never_split_across_pools():
    # 真实签名（app/harness/pools.py:51）：build_pools(questions, correctness, diag_cfg, val_cfg, test_cfg, baseline_run_id)
    qs = [q for pid in [f"p{i}" for i in range(12)] for q in _pair(pid)]
    cfg = {"ratio": 0.34}  # 用真实 PoolConfig/口径填三档；此处示意
    pools = build_pools(qs, correctness={}, diag_cfg=cfg, val_cfg=cfg, test_cfg=cfg, baseline_run_id="b")
    loc = {}
    for name, pool in pools.items():
        for q in pool:
            loc.setdefault(q.pair_id, set()).add(name)
    assert all(len(s) == 1 for s in loc.values()), "pair 被劈到多个池"

实现者注：先 Read app/harness/pools.py:51 确认 diag_cfg/val_cfg/test_cfg 的真实类型（PoolConfig dataclass 还是 dict），测试用真实构造。红因必须是"pair 被拆"而非 TypeError——签名不对会假红。

• Step 2: 跑确认失败（现按 question_id 互斥会劈开）— Expected: FAIL（断言 pair 被拆，非 TypeError）

• Step 3: 实现 — build_pools 内先 units = build_units(questions)；三次 _sample_excluding 的互斥集合与采样对象都改 unit_id；_split_one_category 输入改 unit；build_incremental（pools.py:822-860）的 categories qid 返回也改 unit 口径；最后 flatten_units 展开。exclude 集合用 unit.unit_id。保留原 test→val→diag 顺序与比例（按 unit 计数）。

• Step 4: 跑确认通过 — Expected: PASS

• Step 5: 提交

---

Task 4: loader.stratified_sample 按 unit + load_benchmark 读回 pair 字段

Files:

• Modify: app/question_gen/loader.py（stratified_sample / load_benchmark）

• Test: tests/unit/test_loader_unit_sampling.py

• Step 1: 写失败测试：(a) pair 采样同进同出不拆；(b) 比例按 unit 计数；(c) min_per_class 补足路径 _backfill_per_class（loader.py:154-172）不拆 pair；(d) load_benchmark 读回 pair_id/question_role/flip_axis（.get 兼容旧 JSON）。

• Step 2: 跑确认失败

• Step 3: 实现 — stratified_sample 先 build_units，按 unit 分层/去重/补足/rng.sample（用 Task 5 的 _rng_ns）、_backfill_per_class candidates 按 unit 枚举，返回前 flatten_units；load_benchmark 反序列化补 pair_id=d.get("pair_id") 等 4 字段。

• Step 4: 跑确认通过

• Step 5: 提交

---

Task 5: batching unit 整锁 + 按 unit correctness 分桶 + 非 AR 独立 rng

Files:

• Modify: app/harness/batching.py（build_batches / _select_mixed_by_task_type / _distribute_large_classes）

• Test: tests/unit/test_batching_pair_lock.py

• Step 1: 写失败测试：(a) 同 pair_id 两题落同一 batch（像小类整组不拆）；(b) pair 按 unit correctness（双向 AND） 落 correct/error 桶，不因 P 对 Q 错被劈；(c) 非 AR byte-identical：AR pair 折叠不改变非 AR 的 rng.sample/shuffle 抽样序列（黄金对照）。

• Step 2: 跑确认失败

• Step 3: 实现 — 分批输入改 unit 列表；FFD 容量按 unit.size；_select_mixed_by_task_type 用 unit_correctness 分桶、以 unit 为分发粒度；AR 与非 AR 各用独立稳定派生的 rng（禁用 Python 内置 hash()——受 hash randomization 影响跨进程不可复现）：

import hashlib
def _rng_ns(seed: int, ns: str):
    import random
    h = int.from_bytes(hashlib.sha256(f"{ns}:{seed}".encode()).digest()[:8], "big")
    return random.Random(h)
# 非 AR 用 _rng_ns(seed, "nonAR")，AR pair 用 _rng_ns(seed, "AR")，二者 draw 流互不干扰

把 namespace 派生集中到此 helper（loader/pools/batching 共用），使 unit 折叠不干扰非 AR draw 流。flatten 前保证 pair 同 batch。

• Step 4: 跑确认通过
• Step 5: 提交

---

Task 6: inference pair-level 双向 AND 聚合

Files:

• Modify: app/harness/inference.py（run_inference 聚合段）

• Test: tests/unit/test_inference_pair_aggregate.py

• Step 1: 写失败测试：逐题推理照常写 predictions；pair 按 pair_id 收齐两条合成 1 条 unit record，pair 正确 = P对 AND Q对；InferenceResult.total/correct/per_task_type 按 unit 粒度（total=single 数+pair 数）；孤儿 pair（收不齐）→ 告警并剔除不计入 total（不静默）。

• Step 2: 跑确认失败

• Step 3: 实现 — 聚合前 build_units；per-question prediction 仍逐题落 predictions 表；unit 层用 unit_correctness 计 correct；孤儿按设计 §8（聚合入口）+ §12（原子性/读回校验）剔除+告警。

• Step 4: 跑确认通过

• Step 5: 提交

---

Task 7: correctness 三对象 API + runner 消费点改 unit 视图

Files:

• Modify: app/harness/runner.py（rollout 回写 163-190 / accept 合并 1117-1141 / val 回写 1442-1454 / quadrant / momentum 1561 / probation 1385-1428）、core/evolution/validate.py（pair_block:12 / compute_accuracy:72）、app/harness/validate.py（C-2 关键遗漏：gate 块实际执行路径）

• Test: tests/unit/test_correctness_unit_view.py、tests/unit/test_gate_block_unit.py

• Step 1: 写失败测试：(a) 进化引擎（gate/quadrant/momentum/probation/pair_block/compute_accuracy）消费 unit correctness（AR=双向 AND、非 AR=单题）；逐题 predictions 仅溯源；混格下 e-process delta / 准确率分母按 unit 计、不被 P/Q 单题计分污染。(b) app/harness/validate.py::validate_skill_local 的 gate 块按 unit 跑——baseline_cache.get/put 键含 unit_id（validate.py:282-301）、块 qids 换 unit 折叠（:553-565）、n_used 按 unit 累加（非 len(chunk) 逐题）。

• Step 2: 跑确认失败

• Step 3: 实现 — 新增 unit_correctness_view(units, per_q) -> dict[unit_id,bool]；上述 6+ runner 调用点从 correctness[qid] 改消费 unit view；core/evolution/validate.py::pair_block 按 unit 折叠比对基线/候选臂、compute_accuracy 分母改 unit 数；app/harness/validate.py 的 baseline/candidate block、evidence rows、n_used、baseline_cache.get/put 全部改 unit 口径，predictions 仍逐题溯源（否则 gate 保真定义不进实际运行路径）；quadrant/momentum 键改 unit_id。

• Step 4: 跑确认通过

• Step 5: 提交（commit message 标注核心算法保真#5 相关）

---

Task 8: gate_ladder 迁 unit_id + schema_version 迁移

Files:

• Modify: app/harness/gate_ladder.py（LadderEntry:43 / build_cold_entries:57 / ladder_for:135 / update_probs:167 / GatePools.save/load:181/200 / BaselineCache:277）

• 依赖: app/harness/validate.py 的 baseline_cache.get/put(..., q.question_id) 调用侧（T7 已改 unit 键；本 Task 保证 gate_ladder 侧 key schema 与之对齐）

• Test: tests/unit/test_gate_ladder_unit_migration.py

• Step 1: 写失败测试：(a) LadderEntry 按 unit_id；(b) 冷启动"错优先 2:1"unit 错=P 或 Q 任一错；(c) update_probs 观测先折叠成 unit 再匹配（防按 qid 匹配失效致 EMA 停摆）；(d) GatePools.save/load 带 schema_version；存量无版本 json 加载→明确报错或走迁移（不静默混用）；(e) BaselineCache 键含 unit_id。

• Step 2: 跑确认失败

• Step 3: 实现 — 见设计 §17 gate-unit 迁移 ADR 七项；p_hat 初值 Beta 先验按 unit 定义；probe_quota 按 unit 抽；反泄漏（run_id 含 _gate_ 过滤）不变。写一次性迁移函数 migrate_gate_pools_v1_to_unit()。

• Step 4: 跑确认通过

• Step 5: 提交（标注核心算法保真#5，需逐行比对参考 Video-Tree-TRM4/core/harness/gate_ladder.py）

---

Task 9: pools.json 冻结/解冻序列化 pair 字段

Files:

• Modify: app/harness/pools.py（_q_to_dict / _dict_to_q / save_pools.categories）

• Test: tests/unit/test_pools_serialization.py

• Step 1: 写失败测试：pair 冻结进 pools.json 再 load_pools，pair_id/question_role/flip_axis/unit_id 不丢；categories 块以 unit 记 train/val；旧 JSON（无字段）.get 兼容不崩。

• Step 2: 跑确认失败

• Step 3: 实现 — _q_to_dict 写出 4 字段；_dict_to_q .get(..., 默认) 读回并回填 unit_id。

• Step 4: 跑确认通过

• Step 5: 提交

---

Task 10: checkpoint/resume + 非 AR byte-identical 黄金测试

Files:

• Modify: app/harness/runner.py（epoch_batches / _batch_from_ids）

• Test: tests/integration/test_checkpoint_pair.py、tests/unit/test_non_ar_byte_identical.py

• Step 1: 写失败测试：(a) checkpoint 存 unit_id 序列，_batch_from_ids 恢复时展开完整 unit（断点续跑后 pair 不拆）；(b) 纯非 AR 题库过 pools/batching/inference 的抽样与分批结果与"引入 QuestionUnit 前"逐字节一致（黄金文件）；(c) runner.py:1561 momentum 采样（random.Random(epoch).sample(candidates)，Codex I-3）：混格下 AR pair 折叠会改 candidates 长度/顺序→非 AR momentum 样本漂移。要么 momentum candidates 用独立 rng namespace（Task 5 helper）+ 加混格 momentum golden，要么在计划显式记录"Phase 1 不保证 momentum 混格 byte-identical"作为设计偏差。二选一，不留隐患。

• Step 2: 跑确认失败

• Step 3: 实现 — checkpoint 序列化 unit_id；恢复用 build_units + 展开；非 AR 走 size=1 unit 且独立 rng namespace（Task 5）保证黄金一致。

• Step 4: 跑确认通过

• Step 5: 提交

---

Task 11: pair 原子成对落盘 helper（可复用纯件；wiring 挪 Phase 2）

背景（Codex C-3）：app/question_gen/run_store.py 是 SQLite 日志类，无 on_accept/accepted-JSON 钩子；真实"accepted 题库文件写入"在生成侧（旧 pipeline_v2 的 on_accept 回调 / adversarial_filter.write_final_bank，均 Phase 2 重建）。故 Phase 1 只建可复用、可独立测试的 pair 原子写 helper，实际接到新 pipeline.on_accept 的 wiring 在 Phase 2 做（新 pipeline 落地时）。

Files:

• Create: app/question_gen/pair_atomic_writer.py（纯 helper）

• Test: tests/unit/test_pair_atomic_write.py

• Step 1: 审计定位真实 accepted 写入点 — grep -rn "on_accept\|write_final_bank\|accepted_questions" app/question_gen/ tools/ 记录当前 accepted 文件写入函数（供 Phase 2 wiring 参考），写入计划注释。

• Step 2: 写失败测试 — 针对 pair_atomic_writer：pending buffer 按 pair_id 收齐 original+mirror 才一次性 emit unit；write_accepted(path, units) 全量 tmp + os.replace；模拟"只落 P 未落 Q"→读回 validate_units 剔孤儿；single 恒直接成 unit；unit_hash 不一致→拒。

• Step 3: 跑确认失败

• Step 4: 实现 — PairPendingBuffer.add(q)（按 pair_id 收齐才 emit）+ write_accepted(path, units)（tmp+os.replace）+ read_accepted(path)（validate_units 剔孤儿）。纯函数，不依赖 pipeline。

• Step 5: 跑确认通过 + 提交（计划注释标注：on_accept wiring 见 Phase 2）

---

Task 12: run_store v3 表（facts/unit_verdict/collapse_metrics/quarantine/resume_state）

Files:

• Modify: app/question_gen/run_store.py（建表 + insert 方法）

• Test: tests/unit/test_run_store_v3_tables.py

• Step 1: 写失败测试：store.insert_fact/insert_unit_verdict/insert_collapse_metrics/quarantine/upsert_resume_state 落库+读回；quarantine 内容指纹去重（同指纹 upsert 不重复）；ts 由外部传入（禁进程内 now，保幂等/可复现）。

• Step 2: 跑确认失败

• Step 3: 实现 — 按 research-wiki/schemas/v3-question-gen-logging.md 5 张表建表 + 索引 + insert 方法；对齐 CLAUDE.md §4.8。ts 边界（Codex I-4）明确：v3 新表（facts/unit_verdict/collapse_metrics/quarantine/resume_state）ts 一律外部传入、禁进程内 now（保幂等可复现）；v2 deprecated 表（question_gen_runs/items/adversarial_verdicts）暂保留现有 now() 行为不动（Phase 2/4 删）——不因本 Task 误改旧表。

• Step 4: 跑确认通过

• Step 5: 提交

---

Task 13: Phase 1 集成回归（pair 契约全链路 + 混格评分）

Files:

• Test: tests/integration/test_v3_contract_e2e.py

• Step 1: 写集成测试：构造混格题库（若干 AR pair + 若干非 AR single），跑 build_pools→build_batches→run_inference 全链路，断言：pair 全程不拆、同批、双向 AND 聚合正确、unit 粒度 total/correct 正确、孤儿被剔、非 AR byte-identical。

• Step 2: 跑确认失败

• Step 3: 补齐前序 Task 遗漏

• Step 4: 跑全套 — conda run -n Video-Tree-TRM pytest tests/ -q 全绿 + 覆盖率≥80%

• Step 5: 提交

---

Task 14: 旧代码处理与死代码清除（Phase 1 收尾）

Files:

• Modify: app/harness/pools.py / batching.py / inference.py / runner.py / gate_ladder.py / core/evolution/validate.py（Task 3-10 改过的文件）、app/question_gen/run_store.py
• Test: tests/unit/test_no_dead_perquestion_paths.py

背景：Task 3-10 按 CLAUDE.md §4.2"直接改原文件"做原地替换（非新增并行路径），本 Task 确保替换后不留孤儿死代码，并明确 v2 落库表的过渡去向。不删 v2 生成模块（那在 Phase 2，替代品落地后才删）。

• Step 1: 审计 unit 迁移后的孤儿函数 — 对 Task 3-10 改过的文件，用 conda run -n Video-Tree-TRM ruff check --select F811,F401 app/harness/ core/evolution/ + grep -rn "def _batch_from_ids\|def <被替换的逐题helper>" app/harness/ 逐一确认：被 unit 版替换掉的旧逐题函数/helper（如仅旧 _batch_from_ids 逐题重建、旧逐题 correctness helper）是否仍被引用。列出无引用者。

• Step 2: 写守卫测试（防旧逐题路径复活/残留）

# tests/unit/test_no_dead_perquestion_paths.py
import inspect
from app.harness import batching, pools, inference

def test_no_parallel_perquestion_split_helpers():
    """契约迁 unit 后，不得残留会拆 pair 的旧逐题分批/切分/gate 块路径。"""
    from app.harness import validate as hvalidate  # gate 块真实路径
    src = inspect.getsource(batching) + inspect.getsource(pools) + inspect.getsource(hvalidate)
    # 旧逐题标志（按实际被替换的函数名调整）：断言已被 unit 版取代、无并行残留
    assert "correctness.get(qid)" not in src, "batching 仍有逐题分桶残留"
    # gate 块残留探测（Codex：validate.py 是逐题 gate 核心残留点）
    assert "baseline_cache.get(" not in src or "unit_id" in src, "validate baseline_cache 仍按 qid"
    assert "n_used += len(chunk)" not in src, "validate n_used 仍逐题累加（应按 unit）"
    assert src.count("build_units") >= 1 or "unit_id" in src, "pools/batching/validate 未走 unit 化"

• Step 3: 删除孤儿死代码 + 标注 v2 表过渡 — 删掉 Step 1 确认无引用的旧逐题函数；run_store.py 里 v2 表（question_gen_runs/question_gen_items/adversarial_verdicts）保留但加 deprecation 注释（# DEPRECATED(v3): v2 生成落库，Phase 2 生成替换后删；过渡期与 v3 表并存）——不在 Phase 1 删（生成逻辑还没换）。

• Step 4: 跑测试 + lint — Run: conda run -n Video-Tree-TRM pytest tests/unit/test_no_dead_perquestion_paths.py -v && ruff check app/harness/ core/evolution/ — Expected: PASS + 无 F811/F401

• Step 5: 提交

---

Self-Review 检查

• Spec 覆盖：设计 §8 的 ≥13 入口逐一对应——pools 三池(T3)/序列化(T9)、loader(T4)、batching 整锁+分桶+rng(T5,T10)、inference 聚合(T6)、correctness+validate(T7)、gate_ladder+迁移(T8)、checkpoint+BaselineCache(T10,T8)、load_benchmark(T4)、pair 原子写(T11)、run_store 表(T12)。§12 非功能性（原子/续跑/幂等）→ T10/T11/T12。日志 schema → T12。全覆盖。
• 类型一致：QuestionUnit/build_units/unit_correctness/flatten_units/validate_units 在 T1/T2 定义，T3-T13 一致引用。
• 无占位：各 Task 有真实测试+实现骨架+命令。机械 Task（T4/T9）复用 T2/T3 已给模式。

核心算法保真校验

本阶段触及核心算法#5 信息阶梯（gate_ladder 迁 unit，Task 7/8）：须逐行比对参考 /home/iomgaa/Projects/Video-Tree-TRM4/core/harness/gate_ladder.py（冷启动 2:1/gamma-EMA/Beta 先验/反泄漏），按 unit 重定义不简化，Task 8 已设"保真校验"检查点。其余 12 项不涉及。

后续阶段（各自成计划）+ 旧代码删除清单（显式，防隐式漂）

Phase 2 帧感知抽取+构造器 / Phase 3 六层验证栈 / Phase 4 对抗前移+产量 / Phase 5 验收面板+混格全链路——进入时各写独立 plan。

旧代码删除必须落成对应阶段的显式任务（不得只"替换"而留死壳）：

废弃模块	处置	落哪阶段（显式删除任务）
pipeline_v2.py	被 pipeline（帧感知主编排）原地替换	Phase 2
generator_v2.py	被 constructor+twin_builder（构造）取代	Phase 2
synthesizer.py	v1 遗留生成，直接删	Phase 2
distractor_selector.py	打分 selector 废（GroundAttack 软肋），删	Phase 2
gates.py	四门（AFLite 单信号死路）废；blind_answer 思路迁 §7 坍缩度量后删	Phase 3（坍缩度量落地后）
adversarial_filter.py	拆解：孪生构造→Phase 2、活求解器探针/verdicts 续跑→Phase 4；机制迁完删壳	Phase 4
store/prompts/question_gen/ar_distractor_*、gate_*	打分/后置门 prompt 废；ar_mirror_question/gate_blind_answer 借鉴重写后删旧	Phase 2/3
config/question_gen_ar30.yaml 的 candidate_pool_size/selector_delta_*	selector 参数废	Phase 2
v2 run_store 表 question_gen_runs/items/adversarial_verdicts	Phase 1 加 deprecation 注释保留；生成替换后删	Phase 2（items/runs）/ Phase 4（verdicts）

每阶段计划的最后一个 Task 必须是"旧代码删除 + 死代码清除 + lint 无 F811/F401"，与本 Phase 的 Task 14 同构。