docs: add question-gen v3 construction-paradigm design and phase1 plan

Complete v3 planning: construction-first paradigm (frame-perception grounded fact extraction + 4-family independent judges + 6-layer verification + QuestionUnit contract), adversarial audit, paradigm-shift finding, real-data spike validation, logging schema, and phase1 contract implementation plan.
2026-07-15 05:41:10 -04:00
parent d9f7dee2df
commit 0fe1c96393
9 changed files with 1125 additions and 5 deletions
@@ -0,0 +1,461 @@
+# 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: 写失败测试**
+
+```python
+# 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: 实现**
+
+```python
+# 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: 写失败测试**
+
+```python
+# 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: 实现**
+
+```python
+# 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 劈开）
+
+```python
+# 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 影响跨进程不可复现**）：
+
+```python
+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: 写守卫测试**（防旧逐题路径复活/残留）
+
+```python
+# 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 同构。