# WP3 训练循环与进化引擎 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.
> **前置依赖：WP1（模板已迁移，进化引擎可运行）+ WP4（cache_salt 能力已就绪）必须先完成。**

**Goal:** 修复训练循环与进化引擎的 9 处缺陷，使诊断拿到真实轨迹、早停按 epoch 语义、微型题型不崩、崩溃可幂等续跑、进化 patch 不破坏冻结区、降级信号不驱动错误进化、跨 epoch 评估真实重采样。

**Architecture:** 诊断经 `StepsJsonRunLog` 从 steps_json 重建轨迹（算法 #7 恢复）；早停计数单位 step→epoch；可训练性预检在 gate 建立前剔除微型题型；`_run_step` 幂等（先 DELETE 再写）；patch 冻结区检查整个 target 跨度（算法 #8 加固）；降级/未判定题按 lapse 保守分流；训练推理用 run_id 作 cache_salt（run_id 已含 epoch，天然跨 epoch 重采样）。

**Tech Stack:** Python 3.11、asyncio、SQLite、pytest。

**设计源**：`research-wiki/designs/2026-07-16-preflight-fixes-design.md §6-7`

---

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

| 用途 | 位置 |
|------|------|
| 训练主流程 | `app/harness/runner.py:789-859` `train`；`_setup_train_run:865-879`（预检插入点 L874 前）；`_run_step:1001-1026`（run_id L1011、诊断 L1019、无 DELETE） |
| 早停 | `app/harness/runner.py:291-316` `_should_early_stop`（L315 `+= steps_this_epoch`）；`_TrainState:95-121`（`steps_since_best_improved` L118）；`_maybe_promote_best:1587`（置 0） |
| 诊断调用 | `app/harness/runner.py:2163-2196` `_run_diagnosis`（`RunLogImpl` L2173 未包 StepsJsonRunLog）；DiagnosisResult `degraded_count` 未被引用 |
| gate 刷新 | `app/harness/runner.py:1833-1879` `_refresh_gate_ladder`（save L1878 → set observed L1879）；checkpoint 落盘晚在 train L837 |
| holdout | `app/harness/runner.py:1881-1921` `_holdout_four_way`；`_pick_mixed_best:1923-1963`；`_eval_version_on_pool:2148-2161` |
| 推理落库 | `app/harness/inference.py:363-447` `_run_single_question`（prediction L422 未归一、insert L446 try 外）；`_to_text_field:147-162`；`run_inference:473` |
| Agent Loop | `core/agent/loop.py:103` `run`（session_id L110）；`_call_llm` chat 调用 `:329`（`self._llm.chat(messages, session_id=session_id)`） |
| batching | `app/harness/batching.py:186-202` `_classify_unit`（L200 缺 correctness→None） |
| checkpoint | `app/harness/checkpoint.py:37-44` `_STRUCTURAL_KEYS`；`serialize_state:76-106`；`write_checkpoint:212-260`（原子写） |
| 诊断分流 | `core/evolution/diagnose.py:1485-1519` `_build_skill_case_packs`（lapse 分流 L1492）；`_process_question:2139-2206`（except L2194 cause_category 留 None） |
| traces 适配 | `app/harness/baseline_run_log.py:13` `StepsJsonRunLog`；`steps_json_traces.py:13` |
| patch | `core/evolution/patch.py:285-287` `_in_ranges`；`_do_insert_after:309-326`（L320）；`_do_replace_delete:329-347`（L343）；markers L11-18；`validate_skill` in `evolve.py:296` |

## 核心算法保真校验

触及算法 #7（诊断瀑布，Task 1 恢复轨迹）、#8（patch 引擎，Task 5 冻结区加固）、#10（Agent Loop，Task 9 透传 cache_salt）、#5/#12（信息阶梯/训练编排，Task 7 checkpoint 时序）。**均为恢复/加固/透传，不改算法逻辑**：Task 1 让诊断拿到本就该有的轨迹；Task 5 把"只查起点"补成"查整跨度"（保护方向不变）；Task 9 只加透传参数；Task 7 只调 checkpoint 落盘时机。每个相关 Task 设保真检查点。

---

## Task 1: traces 适配（诊断拿到真实轨迹，算法 #7）

**Files:**
- Modify: `app/harness/runner.py:2163-2196`（_run_diagnosis）
- Test: `tests/unit/test_runner_diag_tree_inject.py` 或 `test_harness_runner.py`

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

在 `tests/unit/test_runner_diag_tree_inject.py` 追加（构造只写 steps_json 不写 traces 表的 run，断言诊断能拿到轨迹）：
```python
@pytest.mark.asyncio
async def test_diagnosis_reads_traces_from_steps_json(...):
    """traces 表为空但 predictions.steps_json 有轨迹时，诊断仍拿到非空 traces。"""
    # 依现有 runner 测试 fixture 造一个 run：predictions 有 steps_json，traces 表空；
    # 调 _run_diagnosis 后断言 diagnose 收到的 traces 非空（可 patch run_diagnosis 捕获入参）
    ...
```
> 依 `test_runner_diag_tree_inject.py` 现有 fixture；实现前读对齐 runner 构造与 patch 点。

- [ ] **Step 2: 运行确认失败**

Run: `conda run -n Video-Tree-TRM python -m pytest tests/unit/test_runner_diag_tree_inject.py -k reads_traces_from_steps_json -v`
Expected: FAIL（当前 RunLogImpl 直读空 traces 表）。

- [ ] **Step 3: 包 StepsJsonRunLog**

`app/harness/runner.py` `_run_diagnosis`，把传给 `run_diagnosis` 的 `run_log`（当前 `RunLogImpl(...)`，L2173 附近）包一层：
```python
    from app.harness.baseline_run_log import StepsJsonRunLog
    from app.harness.log import RunLogImpl

    run_log = StepsJsonRunLog(RunLogImpl(str(self._paths.db_path)))
```
（`StepsJsonRunLog.get_traces` 在底层 traces 空时从 predictions.steps_json 经 `steps_json_to_trace_rows` 重建；`get_predictions` 透传。）

- [ ] **Step 4: 保真检查点 + 测试通过**

Run: `conda run -n Video-Tree-TRM python -m pytest tests/unit/test_runner_diag_tree_inject.py tests/unit/test_baseline_run_log.py -q`
Expected: 全 PASS。保真：确认诊断瀑布拿到的是逐 step `{tool_name,tool_args,tool_output,thought}` 行（对齐 TRM4 诊断输入）。

- [ ] **Step 5: 提交**

```bash
git add app/harness/runner.py tests/unit/test_runner_diag_tree_inject.py
git commit -m "fix: wrap diagnosis run_log with StepsJsonRunLog (restore algo #7 traces)"
```

---

## Task 2: prediction 归一化 + 落库加固

**Files:**
- Modify: `app/harness/inference.py:417-447`
- Test: `tests/unit/test_harness_inference.py`

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

在 `tests/unit/test_harness_inference.py` 追加（LLM 提交非标量 answer 不崩 gather）：
```python
@pytest.mark.asyncio
async def test_nonscalar_prediction_does_not_crash(...):
    """submit_answer 返回 {'answer': ['B']} 等非标量时归一化落库，不抛 sqlite 绑定异常。"""
    # 依现有 inference 测试 fixture，让 AgentLoop 返回 result={'answer': ['B']}；
    # run_inference 应正常完成、predictions 行 prediction 为字符串（如 '["B"]'），不崩
    ...
```
> 依 `test_harness_inference.py` 现有 fake loop/dispatch fixture；实现前读对齐。

- [ ] **Step 2: 运行确认失败**

Run: `conda run -n Video-Tree-TRM python -m pytest tests/unit/test_harness_inference.py -k nonscalar_prediction -v`
Expected: FAIL（sqlite `InterfaceError: Error binding parameter`）。

- [ ] **Step 3: 归一化 prediction + insert 加固**

`app/harness/inference.py`：新增归一化 helper（None 保留、str 原样、其余 `_to_text_field`）：
```python
def _normalize_prediction(answer: object) -> str | None:
    """归一化 prediction：None 保留（INFRA 空预测语义），str 原样，其余 JSON 序列化。"""
    if answer is None or isinstance(answer, str):
        return answer
    return _to_text_field(answer)
```
L422 `"prediction": result_dict.get("answer"),` 改为 `"prediction": _normalize_prediction(result_dict.get("answer")),`。
L446 的 `await asyncio.to_thread(log.insert, "predictions", record)` 包 try，绑定异常降级为最小 error 行不击穿 gather：
```python
    try:
        await asyncio.to_thread(log.insert, "predictions", record)
    except (sqlite3.InterfaceError, sqlite3.ProgrammingError):
        logger.exception("[{}] QA {} 落库绑定异常，降级为 error 行", qa.video_id, qa.question_id)
        record["prediction"] = None
        record["stop_reason"] = "error"
        await asyncio.to_thread(
            log.insert, "predictions",
            {k: v for k, v in record.items() if isinstance(v, (str, int, float, type(None)))},
        )
```
确认 `import sqlite3` 在文件顶部（无则加）。

- [ ] **Step 4: 测试通过 + 回归**

Run: `conda run -n Video-Tree-TRM python -m pytest tests/unit/test_harness_inference.py -q`
Expected: 全 PASS。

- [ ] **Step 5: 提交**

```bash
git add app/harness/inference.py tests/unit/test_harness_inference.py
git commit -m "fix: normalize non-scalar prediction; harden predictions insert"
```

---

## Task 3: early_stop 改 epoch 计数

**Files:**
- Modify: `app/harness/runner.py:291-316,118,1587,849-857`
- Modify: `app/harness/checkpoint.py`（若字段入 state）
- Test: `tests/unit/test_harness_runner.py`

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

在 `tests/unit/test_harness_runner.py` 追加：
```python
def test_early_stop_counts_epochs_not_steps(tmp_path):
    """patience=2 表示连续 2 个 epoch 无 best 刷新才停（不是步数）。"""
    from app.harness.runner import _should_early_stop, _TrainState
    # 造 state + workspace，best 停在 epoch 1；
    # epoch 2 无刷新 → epochs_since_best_improved=1 → 不停；
    # epoch 3 无刷新 → =2 → 停
    ...
```
> 依现有 `_TrainState`/`read_best` fixture；实现前读对齐 workspace best 写入。

- [ ] **Step 2: 运行确认失败**

Run: `conda run -n Video-Tree-TRM python -m pytest tests/unit/test_harness_runner.py -k early_stop_counts_epochs -v`
Expected: FAIL（当前累加 steps_this_epoch）。

- [ ] **Step 3: 字段改名 + 计数改 epoch**

全局把 `steps_since_best_improved` 改名 `epochs_since_best_improved`（`grep -rn steps_since_best_improved app/`：`_TrainState:118`、`_should_early_stop:313,315`、`_maybe_promote_best:1587`，以及 checkpoint serialize/deserialize 若含此字段）。
`_should_early_stop`（L315）`state.steps_since_best_improved += steps_this_epoch` 改为 `state.epochs_since_best_improved += 1`；签名删除 `steps_this_epoch` 参数（改为 `_should_early_stop(workspace_dir, epoch, state, patience)`），train 调用点（L849-857）同步去掉 `len(batches)` 实参。docstring 改为"epoch 粒度"。

- [ ] **Step 4: checkpoint 兼容**

若 `epochs_since_best_improved` 入 checkpoint state（`grep -n steps_since_best_improved app/harness/checkpoint.py`），同步改名。本轮为 fresh 训练无旧 checkpoint，无迁移负担。

- [ ] **Step 5: 测试通过 + 回归**

Run: `conda run -n Video-Tree-TRM python -m pytest tests/unit/test_harness_runner.py tests/unit/test_harness_checkpoint.py -q`
Expected: 全 PASS。

- [ ] **Step 6: 提交**

```bash
git add app/harness/runner.py app/harness/checkpoint.py tests/unit/test_harness_runner.py
git commit -m "fix: early_stop patience counts epochs not steps"
```

---

## Task 4: 可训练性预检

**Files:**
- Modify: `app/harness/runner.py`（`train` 入口过滤 + `_setup_train_run` 接收 filtered task_types）
- Modify: `app/harness/config.py`（RunConfig 加 `trainable_min_units` + 正整数校验）
- Modify: `app/harness/checkpoint.py:37`（`trainable_min_units` 入 `_STRUCTURAL_KEYS` 指纹）
- Test: `tests/unit/test_harness_runner.py`、`tests/unit/test_harness_checkpoint.py`

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

```python
def test_untrainable_types_filtered_before_gate():
    """val<eval_min_per_class 或 非test单元<trainable_min_units 的题型从 diag/val/task_types 剔除。"""
    from app.harness.runner import _filter_untrainable_types
    # 构造 pools：题型 A（val=5, units=40）可训；B（val=0, units=2）不可训
    # 调 _filter_untrainable_types(pools, task_types=[A,B], eval_min_per_class=2, trainable_min_units=8)
    # 断言返回 pools 不含 B、task_types 不含 B
    ...
```

- [ ] **Step 2: 运行确认失败**

Run: `conda run -n Video-Tree-TRM python -m pytest tests/unit/test_harness_runner.py -k untrainable_types_filtered -v`
Expected: FAIL（`_filter_untrainable_types` 不存在）。

- [ ] **Step 3: 实现纯函数预检**

`app/harness/runner.py` 新增模块级纯函数：
```python
def _filter_untrainable_types(
    pools: Pools,
    task_types: list[str] | None,
    eval_min_per_class: int,
    trainable_min_units: int,
) -> tuple[Pools, list[str] | None]:
    """剔除不可训练题型（val<eval_min_per_class 或 非test单元<trainable_min_units）。

    非test单元数 = 该题型 diag+val 题数（single 题 unit==题；等于 gate 阶梯该类候选数）。
    test 池不过滤（继续报告全题型准确率）。返回过滤后 (pools, task_types)。
    """
    from collections import Counter

    diag_by_type = Counter(q.task_type for q in pools.diagnosis)
    val_by_type = Counter(q.task_type for q in pools.validation)
    keep: set[str] = set()
    dropped: list[tuple[str, str]] = []
    for tt in set(diag_by_type) | set(val_by_type):
        n_val = val_by_type.get(tt, 0)
        n_units = diag_by_type.get(tt, 0) + n_val
        if n_val < eval_min_per_class:
            dropped.append((tt, f"val={n_val}<{eval_min_per_class}"))
        elif n_units < trainable_min_units:
            dropped.append((tt, f"units={n_units}<{trainable_min_units}"))
        else:
            keep.add(tt)
    for tt, why in sorted(dropped):
        logger.warning("可训练性预检剔除题型 {}（{}）", tt, why)
    new_pools = replace(
        pools,
        diagnosis=[q for q in pools.diagnosis if q.task_type in keep],
        validation=[q for q in pools.validation if q.task_type in keep],
    )
    new_types = [t for t in task_types if t in keep] if task_types is not None else sorted(keep)
    return new_pools, new_types
```
（确认 `from dataclasses import replace` 已 import；`Pools` 是否 frozen dataclass 支持 `replace`——若非，按其构造方式重建。）

**过滤结果必须回传主循环（Codex Critical）**：`RunConfig` 是 `@dataclass(frozen=True)`，**不能** `self._config.task_types = ...`（会 FrozenInstanceError），且过滤后的 pools 必须被 `train()` 后续的 batch/step/slow-update/final-eval 全部使用。实现方式：
- 在 `train(pools)` **入口第一步**（`_setup_train_run` 调用之前）过滤：
```python
        pools, filtered_task_types = _filter_untrainable_types(
            pools, self._config.task_types,
            self._config.eval_min_per_class, self._config.trainable_min_units,
        )
```
- 把 filtered `pools` 传给 `_setup_train_run(pools)` 与 train() 后续所有消费点（build_batches / slow_update / final_eval 均用这个 filtered pools，不再触碰原始 pools）。
- 把 `filtered_task_types` 传给 gate 建立（`_setup_train_run`/`_init_gate_pools` 用它而非 `self._config.task_types`）——新增参数透传，不改 frozen config。

`app/harness/config.py`：`RunConfig` 加字段 `trainable_min_units: int`（无默认，显式配置；train yaml 提供）；在配置校验函数（如 `validate_config`，config.py:277 附近）加 `trainable_min_units >= 1` 断言（<1 报错）。
`app/harness/checkpoint.py:37` `_STRUCTURAL_KEYS` 加入 `"trainable_min_units"`——该值改变会改变 pools 过滤结果与训练轨迹，必须纳入 checkpoint 结构指纹，resume 时变化即拒绝复用旧 checkpoint。

- [ ] **Step 4: 测试通过 + 回归**

Run: `conda run -n Video-Tree-TRM python -m pytest tests/unit/test_harness_runner.py -q`
Expected: 全 PASS。

- [ ] **Step 5: 提交**

```bash
git add app/harness/runner.py app/harness/config.py tests/unit/test_harness_runner.py
git commit -m "feat: pre-flight filter of untrainable task types before gate"
```

---

## Task 5: patch 冻结区跨度 + 注入 + marker 校验（算法 #8）

**Files:**
- Modify: `core/evolution/patch.py:285-347`
- Modify: `core/evolution/evolve.py:296`（validate_skill）
- Test: `tests/unit/test_patch.py`

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

在 `tests/unit/test_patch.py` 追加：
```python
def test_replace_spanning_into_protected_is_skipped():
    """target 起点在正文、末端伸入冻结区的 replace 被跳过（不破坏 marker）。"""
    from core.evolution.patch import apply_patch_with_report, APPENDIX_START, APPENDIX_END

    body = "正文最后一段。"
    appendix = f"{APPENDIX_START}\n## 执行提醒\n- 规则A\n{APPENDIX_END}"
    content = body + "\n\n" + appendix
    # target 从正文末尾跨入 APPENDIX_START
    target = "正文最后一段。\n\n" + APPENDIX_START
    edits = [{"op": "delete", "target": target, "content": ""}]
    new_content, report = apply_patch_with_report(content, edits, protected_spans=[appendix])
    assert APPENDIX_START in new_content and APPENDIX_END in new_content  # marker 未被破坏


def test_edit_payload_with_marker_literal_rejected():
    """edit payload/target 含 marker 字面量 → 拒绝该 edit。"""
    from core.evolution.patch import apply_patch_with_report, APPENDIX_START
    edits = [{"op": "append", "target": "", "content": f"注入 {APPENDIX_START} 破坏"}]
    _, report = apply_patch_with_report("正文", edits, protected_spans=[])
    assert any("marker" in str(s).lower() or "reject" in str(s).lower() for s in report)
```
> marker 常量名以 `core/evolution/patch.py:11-18` 为准（`APPENDIX_START` 等），实现前读对齐 import 名。

- [ ] **Step 2: 运行确认失败**

Run: `conda run -n Video-Tree-TRM python -m pytest tests/unit/test_patch.py -k "spanning_into_protected or marker_literal" -v`
Expected: FAIL（当前只查起点 pos，跨入未拦；无注入检查）。

- [ ] **Step 3: 跨度检查 + 注入检查**

`core/evolution/patch.py` 新增跨度 helper：
```python
def _span_overlaps_ranges(pos: int, length: int, ranges: list[tuple[int, int]]) -> bool:
    """判断 [pos, pos+length) 是否与任一冻结区间相交（不止起点）。"""
    end = pos + length
    return any(start < end and pos < r_end for start, r_end in ranges)
```
`_do_insert_after` L320 `if _in_ranges(pos, ranges):` 改为 `if _span_overlaps_ranges(pos, len(target), ranges):`。
`_do_replace_delete` L343 `if _in_ranges(pos, ranges):` 改为 `if _span_overlaps_ranges(pos, len(target), ranges):`。
在 `apply_patch_with_report`（L387）应用每个 edit 前加注入检查：payload/target 含任一 marker 字面量（`APPENDIX_START/END`、`MOMENTUM_START/END`）→ 跳过该 edit 并记 `skipped_marker_injection`。

- [ ] **Step 4: validate_skill 加 marker 完整性校验**

`core/evolution/evolve.py:296` `validate_skill`：在现有 frontmatter/长度/代码块校验后加——统计 evolved 中 `APPENDIX_START/END`、`MOMENTUM_START/END` 出现次数，要求成对（START 数==END 数）、各至多一对、START 在 END 前；违反则返回校验失败（该候选整体 reject）。

- [ ] **Step 5: 保真检查点 + 测试通过**

Run: `conda run -n Video-Tree-TRM python -m pytest tests/unit/test_patch.py tests/unit/test_evolve.py -q`
Expected: 全 PASS。保真：确认"保护跨度"方向未变（仍是保护 appendix/momentum 不被误改），只是从"查起点"补成"查整跨度"+ 注入/完整性双防线。

- [ ] **Step 6: 提交**

```bash
git add core/evolution/patch.py core/evolution/evolve.py tests/unit/test_patch.py
git commit -m "fix: patch checks full target span + marker injection/integrity (algo #8)"
```

---

## Task 6: 诊断降级分流 + 占比中止

**Files:**
- Modify: `core/evolution/diagnose.py:1485-1497`
- Modify: `app/harness/runner.py:1019`（诊断后 degraded 占比检查）
- Test: `tests/unit/test_diagnose.py`、`tests/unit/test_harness_runner.py`

- [ ] **Step 1: 写失败测试（分流）**

在 `tests/unit/test_diagnose.py` 追加：
```python
def test_none_cause_and_degraded_route_to_lapse():
    """cause_category=None（判别失败）与 degraded 题按 lapse 处置，不进 defect 正文路径。"""
    from core.evolution.diagnose import _build_skill_case_packs
    # 构造 metrics_group：一题 attr.cause_category=None（非 degraded）、一题 qm.degraded=True；
    # 断言二者都不出现在 failure_cases（wrong_by_error），只可能进 lapse_notes
    ...
```
> 依 `test_diagnose.py` 现有 QuestionMetrics/ErrorAttribution 构造（`:753` 附近）；实现前读对齐字段。

- [ ] **Step 2: 运行确认失败**

Run: `conda run -n Video-Tree-TRM python -m pytest tests/unit/test_diagnose.py -k none_cause_and_degraded_route -v`
Expected: FAIL（当前 None → wrong_by_error 走 defect 正文）。

- [ ] **Step 3: 改分流逻辑**

`core/evolution/diagnose.py` `_build_skill_case_packs` 的分流循环（L1488-1497）改为"仅明确 defect 且非 degraded 才进正文路径"：
```python
        for qm in metrics_group:
            if qm.correct:
                continue
            attr = attribution_map.get(qm.question_id)
            is_defect = (
                attr is not None
                and attr.cause_category == "defect"
                and not qm.degraded
            )
            if not is_defect:
                # lapse / None（判别失败）/ degraded → 保守，不驱动正文进化
                if attr is not None and attr.lapse_note and attr.lapse_note.strip():
                    lapse_notes.append(attr.lapse_note)
                continue
            wrong_by_error[attr.error_type].append(qm)
```

- [ ] **Step 4: 写失败测试（占比中止）+ 实现**

`tests/unit/test_harness_runner.py` 追加：诊断结果 degraded_count/总题数 > 0.5 时 `_run_step` 后应 raise（疑似基础设施故障）。
`app/harness/runner.py` `_run_step`（L1019 拿到 `diagnosis` 后）加：
```python
        n_wrong = sum(1 for q in batch if not state.correctness.get(q.question_id, True))
        if n_wrong > 0 and diagnosis.degraded_count / n_wrong > 0.5:
            raise RuntimeError(
                f"本 step 诊断降级占比 {diagnosis.degraded_count}/{n_wrong} > 50%，"
                "疑似 judge 基础设施故障，中止训练（不以降级信号驱动进化）。"
            )
```
（确认 `DiagnosisResult.degraded_count` 字段可用，定义在 `core/evolution/types.py:299`。）

- [ ] **Step 5: 测试通过 + 回归**

Run: `conda run -n Video-Tree-TRM python -m pytest tests/unit/test_diagnose.py tests/unit/test_harness_runner.py -q`
Expected: 全 PASS。

- [ ] **Step 6: 提交**

```bash
git add core/evolution/diagnose.py app/harness/runner.py tests/unit/test_diagnose.py tests/unit/test_harness_runner.py
git commit -m "fix: route None/degraded diagnoses to lapse; abort on high degrade rate"
```

---

## Task 7: step 幂等（DELETE）+ gate_epoch_observed 立即落盘

**Files:**
- Modify: `app/harness/runner.py:1001-1026`（_run_step 开头 DELETE）
- Modify: `app/harness/runner.py:1378-1498,1833-1879`（checkpoint 提到 gate save 之后）
- Test: `tests/unit/test_harness_runner.py`

- [ ] **Step 1: 写失败测试（幂等）**

```python
@pytest.mark.asyncio
async def test_run_step_deletes_stale_rows_before_rerun(...):
    """同 run_id 重跑前先清 predictions/traces，避免重复行双计。"""
    # 预置该 step run_id 的旧 predictions 行；调 _run_step；断言旧行被清、只剩本次
    ...
```

- [ ] **Step 2: 运行确认失败**

Run: `conda run -n Video-Tree-TRM python -m pytest tests/unit/test_harness_runner.py -k deletes_stale_rows -v`
Expected: FAIL（当前 append，无 DELETE）。

- [ ] **Step 3: _run_step 开头 DELETE**

`app/harness/runner.py` `_run_step`，在 rollout（L1012）之前加（**用 `IF EXISTS` 避免 fresh workspace 首跑时 predictions/traces 表尚未由 `run_inference._ensure_tables` 创建导致 `OperationalError: no such table`**）：
```python
        with HarnessLog(str(self._paths.db_path), run_id, register_run=False) as log:
            log.execute("DELETE FROM predictions WHERE run_id=?", (run_id,))
            log.execute("DELETE FROM traces WHERE run_id=?", (run_id,))
```
> SQLite `DELETE FROM <t>` 对不存在的表会抛 `no such table`。两种消解方式择一（实现前读 log.py 确认）：① rollout 由 `run_inference` 先建表——把 DELETE 移到**首次 rollout 之后、诊断之前**并只在 resume 重跑（step 已有旧行）时执行；② 或 DELETE 前先 `CREATE TABLE IF NOT EXISTS`（复用 inference 的 PREDICTIONS_SCHEMA/TRACES_SCHEMA），保证幂等无害。推荐 ②（无害且简单）。`register_run=False` 来自 WP4；若 HarnessLog 无 `execute` 便捷方法，用其现有连接接口。

- [ ] **Step 4: checkpoint 提到 gate save 之后（消除双计窗口）**

目标：`_refresh_gate_ladder` 内 `gate_pools.save`（L1878）+ `gate_epoch_observed=True`（L1879）之后，**立即** `write_checkpoint`（phase="epoch_done"），不等到 `train` L837。实现：把 `_slow_update_cycle` Phase 10（调 `_refresh_gate_ladder` L1496-1498）之后的 checkpoint 落盘从 `train`（L837）移入 `_slow_update_cycle` 末尾，或让 `_refresh_gate_ladder` 接收 checkpoint 所需上下文（epoch/progress/batches）并在 save 后落盘。实现前读 `write_checkpoint` 签名（checkpoint.py:212）与 `train` L833-848 对齐参数，确保 gate_pools.json 与 checkpoint 的 `gate_epoch_observed` 同一时刻一致。

- [ ] **Step 5: 保真检查点 + 测试通过**

Run: `conda run -n Video-Tree-TRM python -m pytest tests/unit/test_harness_runner.py tests/unit/test_harness_checkpoint.py -q`
Expected: 全 PASS。保真（算法 #5/#12）：确认 γ-EMA 更新（`update_probs`）仍每 epoch 一次、checkpoint 落盘不改变慢更新十步序的语义顺序。

- [ ] **Step 6: 提交**

```bash
git add app/harness/runner.py tests/unit/test_harness_runner.py
git commit -m "fix: idempotent _run_step (DELETE stale) + checkpoint after gate save"
```

---

## Task 8: holdout 四向去重

**Files:**
- Modify: `app/harness/runner.py:1881-1963`（_holdout_four_way / _pick_mixed_best）
- Test: `tests/unit/test_harness_runner.py`

> **说明**：目标是每 epoch 的四向 test 评估从"4×600 全跑"降为"仅 final 必跑 + best_hard 未评过才跑 + baseline 从基线预测推导 + best_mixed 引用赢家"。去重做在 harness 逻辑层（配合 WP4 epoch 盐，同版本不重采样）。

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

```python
@pytest.mark.asyncio
async def test_holdout_dedup_skips_reevaluated_versions(...):
    """baseline 不跑推理（从基线预测推导）；best_hard==final 时不重复评估。"""
    # 统计 _eval_version_on_pool 被调次数：baseline=0，best_hard==final 时该向复用不重跑
    ...
```
> 依现有 runner holdout fixture；实现前读 `_holdout_four_way`/`write_holdout_eval` 对齐。

- [ ] **Step 2: 运行确认失败**

Run: `conda run -n Video-Tree-TRM python -m pytest tests/unit/test_harness_runner.py -k holdout_dedup -v`
Expected: FAIL（当前四向各跑一次）。

- [ ] **Step 3: 实现去重（进程内备忘录，不改 holdout_eval schema）**

> **schema 约束（Codex Critical）**：当前 `holdout_eval` 表（observation.py:78）不存 skills_version/prompts_version/pointer，无法按版本反查做跨-run hydrate。本 task **不扩展该 schema**（避免结构性风险），改用 **train() 进程内备忘录** `dict[(skills_v,prompts_v), float]` 去重。代价：resume 后备忘录清空、已评版本会重评一次——resume 是异常路径、重评 600 题成本可接受，换取零 schema 变更风险。完整跨-run hydrate 记 future work。

`app/harness/runner.py` `_holdout_four_way`（在 `_TrainState` 加一个 `holdout_memo: dict[tuple[str,str], float] = field(default_factory=dict)` 字段）：
- **baseline 向**：不调 `_eval_version_on_pool`，改从基线 predictions（`baseline_run_id`）读 test 题对错算 acc（test 题在 infer_adhoc 已全推理过）；结果存 memo，epoch>1 直接复用（0 推理）。
- **final 向**：真评 600，算完存 `memo[(final_sv,final_pv)]`。
- **best_hard 向**：若 `(best_sv,best_pv)` 已在 `memo`（== final 或往轮已评）则引用，否则真评并存 memo。
- **best_mixed 向**：`_pick_mixed_best` 选出的赢家必是 best_hard 或 final 之一，其 test acc 已在 memo，直接引用写 holdout_eval，0 推理。

实现前完整读 `_holdout_four_way`（~1885）、`write_holdout_eval`、`_eval_version_on_pool` 对齐；`write_holdout_eval` 调用保持不变（仍逐向写观测行，只是 acc 来源改为 memo 复用/基线推导）。

- [ ] **Step 4: 测试通过 + 回归**

Run: `conda run -n Video-Tree-TRM python -m pytest tests/unit/test_harness_runner.py -q`
Expected: 全 PASS。

- [ ] **Step 5: 提交**

```bash
git add app/harness/runner.py tests/unit/test_harness_runner.py
git commit -m "perf: dedup holdout four-way eval (baseline derive, best_hard memo)"
```

---

## Task 9: 训练推理注入 epoch 盐（cache_salt=run_id，算法 #10 透传）

**Files:**
- Modify: `core/agent/loop.py:103-141,271-329`
- Modify: `app/harness/inference.py:408-415,473`
- Test: `tests/unit/test_agent_loop`（或现有 loop 测试）、`tests/unit/test_harness_inference.py`

> **原理**：训练/val/test/holdout 推理的 run_id 已含 `_e{epoch}`（如 `{base}_e{epoch}_s{step}`、`{run_id}_holdout_{kind}_e{epoch}`），用 run_id 作 cache_salt 即天然跨 epoch 重采样、同 epoch 续跑仍命中。judge/evolve 不经此路径（默认 salt=None）。gate 基线臂走 BaselineCache 不受影响；候选臂 messages 含 skill 版本天然区分。

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

```python
@pytest.mark.asyncio
async def test_agent_loop_forwards_cache_salt():
    """AgentLoop.run(cache_salt=...) 透传到 llm.chat。"""
    from core.agent.loop import AgentLoop
    # fake llm 记录 chat 收到的 cache_salt kwarg；loop.run(..., cache_salt='run:e2')
    # 断言 fake_llm.chat 收到 cache_salt='run:e2'
    ...
```

- [ ] **Step 2: 运行确认失败**

Run: `conda run -n Video-Tree-TRM python -m pytest -k agent_loop_forwards_cache_salt -v`
Expected: FAIL（`run()` 无 cache_salt 参数）。

- [ ] **Step 3: AgentLoop 透传 cache_salt**

`core/agent/loop.py`：`run`（L103）、`_step`/`_call_llm`（L271,317）签名加 `cache_salt: str | None = None`（keyword，随 session_id 透传）；L329 `self._llm.chat(messages, session_id=session_id)` 改为 `self._llm.chat(messages, session_id=session_id, cache_salt=cache_salt)`。

- [ ] **Step 4: inference 用 run_id 作 salt**

`app/harness/inference.py` `_run_single_question`：`loop.run(...)`（L409）加 `cache_salt=run_id`（run_id 从 run_inference 透传到每题；`_run_single_question` 已有 run_id 上下文——若无则从 run_inference 参数透传）。确认 run_inference→_run_single_question 的 run_id 传递链完整。

- [ ] **Step 5: 保真检查点 + 测试通过**

Run: `conda run -n Video-Tree-TRM python -m pytest tests/unit/test_harness_inference.py tests/integration/test_agent_governed_e2e.py -q`
Expected: 全 PASS。保真（算法 #10）：确认只加透传参数，Thinking+JSON/json_repair/pluggy hook 逻辑不变。

- [ ] **Step 6: 提交**

```bash
git add core/agent/loop.py app/harness/inference.py tests/unit/
git commit -m "feat: inject run_id as cache_salt for per-epoch resampling (algo #10)"
```

---

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

- [ ] traces 适配后诊断拿到真实轨迹（算法 #7 恢复）。
- [ ] prediction 归一化 None 保留、非标量序列化；insert 绑定异常不击穿 gather。
- [ ] early_stop 字段全局改名一致、计数改 epoch。
- [ ] 预检剔除不可训练题型（test 池不动）。
- [ ] patch 查整跨度 + 注入 + marker 完整性三防线（算法 #8 保护方向不变）。
- [ ] None/degraded 按 lapse 保守分流；降级占比>50% 中止。
- [ ] _run_step 幂等；gate_pools 与 checkpoint 的 gate_epoch_observed 同刻一致。
- [ ] holdout 去重后 baseline 0 推理、best_hard 备忘录 resume 可 hydrate。
- [ ] cache_salt=run_id 贯穿 AgentLoop，run_id 含 epoch 保证跨 epoch 重采样。

## 核心算法保真校验结论

触及算法 #5/#7/#8/#10/#12，均为恢复（#7 轨迹）/加固（#8 跨度）/透传（#10 salt）/时序（#5/#12 checkpoint），各 Task 已设保真检查点，不改算法核心逻辑。Task 5/7/9 需在实现时对照 TRM4 参考确认无行为漂移。

## 验收标准

1. `pytest tests/unit/test_harness_runner.py tests/unit/test_harness_inference.py tests/unit/test_diagnose.py tests/unit/test_patch.py tests/unit/test_harness_checkpoint.py tests/unit/test_baseline_run_log.py tests/unit/test_runner_diag_tree_inject.py` 全绿。
2. 诊断拿到非空轨迹；非标量 prediction 不崩；early_stop 按 epoch。
3. 微型题型被预检剔除；patch 不破坏 marker；降级题不驱动进化。
4. _run_step 幂等；cache_salt=run_id 贯穿。