Video-Tree-TRM5/research-wiki/plans/2026-07-16-preflight-wp2-split-wiring.md

# WP2 切分与接线 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:** 让冻结的 video-split 切分能正确进入训练 workspace（seed 携带 pools.json + global 一致性校验），并把切分质量三处优化（val_ratio 0.4、tier 感知 diag/val 分配、val 功效修复）与冻结产物覆盖保护落地。

**Architecture:** 切分产物由 `video_split_cli` 冻结到 `workspaces/video-split/`；本 WP 让 seed 携带该产物、训练 fresh 时拷入 workspace 并校验一致性。tier 感知在 `_split_trainval_by_video_group` 内实现——错题视频组按 T2(defect) 含量升序进 val（保留 T2 高的组在 diag），并把 `val_wrong_min` 前置到切分内做功效修复（不足则从 diag 换出低 T2 错题组补 val，耗尽 fail-loud）。

**Tech Stack:** Python 3.11、pytest、SQLite、frozen dataclass、shutil、原子写（tmp+os.replace）。

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

---

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

| 用途 | 位置 |
|------|------|
| trainval→diag/val 切分 | `app/harness/pools.py:126-199` `split_by_video_assignment`；`:289-338` `_split_trainval_by_video_group`；`:118` `InsufficientValSignal` |
| global 加载（无校验） | `app/harness/pools.py:713-813` `build_or_load_pools`（L813 `return load_pools` 前无 global 校验）；`:587-620` `load_pools` |
| 冻结编排 | `app/harness/build_split.py:104-219` `build_split`（signal_rows 含 tier L154；split_by_video_assignment 调用 L184-191；save_pools L192）；`:46-71` `SplitBuildConfig`（无 val_wrong_min） |
| CLI 构造 | `app/harness/video_split_cli.py:559-577` `SplitBuildConfig(...)`；`:580` `check_mcnemar_power`；`:751-773` `build_arg_parser`（无 --force） |
| seed | `app/harness/store.py:184-232` `init_seed`（拷 skills/prompts/baseline.db，不拷 pools）；`:269-307` `extract_run_db`（不去重） |
| workspace | `app/harness/workspace.py:156-200` `init_workspace_from_seed`（copy2 baseline.db→harness.db L197，不拷 pools） |
| manifest | `app/harness/split_manifest.py:19-59` `write_manifest`（pools_sha256 L54） |
| 配置 | `config/video_split.yaml`（val_ratio L15=0.3、val_wrong_min L14=20） |
| 测试 | `tests/unit/test_pools_video_atomic.py`、`test_split_selection.py`、`test_harness_pools.py`、`test_harness_store.py`、`test_harness_workspace.py` |

## 核心算法保真校验

触及算法 #5（信息阶梯）的**上游输入**：本 WP 只改"哪些视频进 diag/val"，pools 内仍是逐 unit 列表，`gate_ladder` 消费的 unit+correctness 结构不变。**保真检查点（Task 3 Step 6）**：确认 `_split_trainval_by_video_group` 返回后 diagnosis/validation 仍是逐题 `GeneratedQuestion` 列表、视频组原子性（同 video 全部题同池）不被 tier 排序破坏。不改算法 #6/#9。

---

## Task 1: val_ratio 0.3 → 0.4

**Files:**
- Modify: `config/video_split.yaml:15`

- [ ] **Step 1: 改配置**

`config/video_split.yaml` 第 15 行：
```yaml
  val_ratio: 0.3               # validation 占 trainval 视频组总数的比例
```
改为：
```yaml
  val_ratio: 0.4               # validation 占 trainval 视频组总数的比例（0.3→0.4 提升整包终审功效，WP2）
```

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

```bash
git add config/video_split.yaml
git commit -m "chore: bump video_split val_ratio 0.3->0.4 for terminal-eval power"
```

---

## Task 2: extract_run_db 每题去重（902→900 canonical）

**Files:**
- Modify: `app/harness/store.py:269-307`
- Test: `tests/unit/test_harness_store.py`（`TestExtractRunDb`）

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

在 `tests/unit/test_harness_store.py` 的 `TestExtractRunDb` 类追加：
```python
    def test_dedupe_per_question_keeps_first_row(self, tmp_path):
        """dedupe_per_question=True 时每 question_id 只保留 rowid 最小的首行。"""
        import sqlite3

        src = tmp_path / "src.db"
        conn = sqlite3.connect(src)
        conn.execute(
            "CREATE TABLE _runs (run_id TEXT PRIMARY KEY, started_at TEXT)"
        )
        conn.execute("INSERT INTO _runs VALUES ('r1', 't0')")
        conn.execute(
            "CREATE TABLE predictions (run_id TEXT, question_id TEXT, prediction TEXT)"
        )
        # 743-1 三行（模拟 error/budget/finished），首行 prediction=NULL
        conn.executemany(
            "INSERT INTO predictions VALUES (?,?,?)",
            [
                ("r1", "743-1", None),
                ("r1", "743-1", None),
                ("r1", "743-1", "C"),
                ("r1", "q2", "A"),
            ],
        )
        conn.commit()
        conn.close()

        dst = tmp_path / "dst.db"
        from app.harness.store import extract_run_db

        extract_run_db(src, dst, "r1", dedupe_per_question=True)

        out = sqlite3.connect(dst)
        rows = out.execute(
            "SELECT question_id, prediction FROM predictions ORDER BY question_id"
        ).fetchall()
        out.close()
        assert rows == [("743-1", None), ("q2", "A")], f"未按 rowid 首行去重: {rows}"
```

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

Run: `conda run -n Video-Tree-TRM python -m pytest tests/unit/test_harness_store.py::TestExtractRunDb::test_dedupe_per_question_keeps_first_row -v`
Expected: FAIL（`extract_run_db() got an unexpected keyword argument 'dedupe_per_question'`）。

- [ ] **Step 3: 实现去重**

`app/harness/store.py` `extract_run_db` 签名改为：
```python
def extract_run_db(
    src_db: Path, dst_db: Path, run_id: str, *, dedupe_per_question: bool = False
) -> None:
```
docstring 补一句参数说明：
```
        dedupe_per_question: True 时 predictions 表每 question_id 仅保留 rowid 最小
            的首行（对齐 canonical「每 question_id 取第一行 ORDER BY rowid」口径，
            902→900）。_runs 表不受影响。
```
把 predictions 分支的取行 SQL（L297-299）改为按 `dedupe_per_question` 分派：
```python
            if table == "predictions" and dedupe_per_question:
                rows = src.execute(
                    f"SELECT {col_sql} FROM {table} WHERE run_id=? "
                    "AND rowid IN (SELECT MIN(rowid) FROM predictions "
                    "WHERE run_id=? GROUP BY question_id)",
                    (run_id, run_id),
                ).fetchall()
            else:
                rows = src.execute(
                    f"SELECT {col_sql} FROM {table} WHERE run_id=?", (run_id,)
                ).fetchall()
```
> NULL question_id 说明：predictions 的 question_id 是题标识、语义上非空（canonical 900 题均有 id），`GROUP BY question_id` 的 NULL 折叠风险不适用。若源库异常出现 NULL question_id，`MIN(rowid) GROUP BY` 会把它们折叠成一行——本任务不为该异常兜底（预测数据契约保证非空），保持 fail-visible。

- [ ] **Step 4: 运行确认通过 + 回归**

Run: `conda run -n Video-Tree-TRM python -m pytest tests/unit/test_harness_store.py -q`
Expected: 全 PASS（默认 `dedupe_per_question=False` 保持既有行为，旧测试不受影响）。

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

```bash
git add app/harness/store.py tests/unit/test_harness_store.py
git commit -m "feat: add dedupe_per_question to extract_run_db (canonical 902->900)"
```

---

## Task 3: tier 感知 + val 功效修复的 diag/val 分配

**Files:**
- Modify: `app/harness/pools.py:126-199,289-338`
- Modify: `app/harness/build_split.py:46-71,181-192`
- Modify: `app/harness/video_split_cli.py:565-573`
- Test: `tests/unit/test_pools_video_atomic.py`

- [ ] **Step 1: 写失败测试（tier 优先 + 功效修复）**

在 `tests/unit/test_pools_video_atomic.py` 追加：
```python
def test_tier_aware_keeps_high_t2_in_diag():
    """错题视频组按 T2 含量升序进 val：T2 高的组保留在 diagnosis。"""
    from app.harness.pools import split_by_video_assignment
    from core.types import GeneratedQuestion

    def _q(qid, vid):
        return GeneratedQuestion(
            question_id=qid, video_id=vid, task_type="X", question="q",
            options=["A", "B"], answer="A", source_nodes=[], difficulty="easy",
        )

    # 4 个错题视频（每视频 1 题），T2 数分别 2/1/0/0
    questions = [_q(f"{v}-1", v) for v in ("vA", "vB", "vC", "vD")]
    assignment = {v: "trainval" for v in ("vA", "vB", "vC", "vD")}
    correctness = {f"{v}-1": False for v in ("vA", "vB", "vC", "vD")}
    wrong_tier = {"vA": 2, "vB": 1, "vC": 0, "vD": 0}

    pools = split_by_video_assignment(
        questions, assignment, correctness, val_ratio=0.5, seed=7,
        wrong_tier_by_video=wrong_tier,
    )
    diag_vids = {q.video_id for q in pools.diagnosis}
    # T2 最高的 vA 必留 diag；T2=0 的组优先进 val
    assert "vA" in diag_vids
    assert "vB" in diag_vids


def test_val_wrong_min_repair_pulls_from_diag():
    """val 错题不足 val_wrong_min 时从 diag 换入低 T2 错题组补足。"""
    from app.harness.pools import split_by_video_assignment
    from core.types import GeneratedQuestion

    def _q(qid, vid, correct):
        return GeneratedQuestion(
            question_id=qid, video_id=vid, task_type="X", question="q",
            options=["A", "B"], answer="A", source_nodes=[], difficulty="easy",
        )

    # 8 错题视频 + 2 正确视频；val_ratio 小使初分 val 错题不足，触发修复
    vids_wrong = [f"w{i}" for i in range(8)]
    vids_correct = ["c0", "c1"]
    questions = [_q(f"{v}-1", v, False) for v in vids_wrong] + [
        _q(f"{v}-1", v, True) for v in vids_correct
    ]
    assignment = {v: "trainval" for v in vids_wrong + vids_correct}
    correctness = {f"{v}-1": False for v in vids_wrong}
    correctness.update({f"{v}-1": True for v in vids_correct})
    wrong_tier = {v: i for i, v in enumerate(vids_wrong)}  # 递增 T2

    pools = split_by_video_assignment(
        questions, assignment, correctness, val_ratio=0.1, seed=7,
        wrong_tier_by_video=wrong_tier, val_wrong_min=4,
    )
    val_wrong = sum(1 for q in pools.validation if not correctness[q.question_id])
    assert val_wrong >= 4, f"功效修复后 val 错题 {val_wrong} < 4"
```

> 注：`GeneratedQuestion` 的真实字段以 `app/question_gen/types.py` 为准；若构造签名不符，读该文件对齐必填字段（勿臆造）。

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

Run: `conda run -n Video-Tree-TRM python -m pytest tests/unit/test_pools_video_atomic.py -k "tier_aware or val_wrong_min_repair" -v`
Expected: FAIL（`unexpected keyword argument 'wrong_tier_by_video'`）。

- [ ] **Step 3: 改 `_split_trainval_by_video_group` 加 tier 感知 + 功效修复**

`app/harness/pools.py` 函数签名改为：
```python
def _split_trainval_by_video_group(
    trainval_qs: list[GeneratedQuestion],
    correctness: dict[str, bool],
    val_ratio: float,
    rng: random.Random,
    wrong_tier_by_video: dict[str, int] | None = None,
    val_wrong_min: int = 0,
) -> tuple[list[GeneratedQuestion], list[GeneratedQuestion]]:
```
把分层块（L329-334 的 else 分支）改为 tier 感知：`wrong_vids` 按 T2 含量升序（T2 少的优先进 val），保留 T2 高的组在 diag；`wrong_tier_by_video=None` 时退化为原 shuffle：
```python
    else:
        val_correct = math.floor(n_correct * n_val / n_total)
        val_wrong = n_val - val_correct
        rng.shuffle(correct_vids)
        if wrong_tier_by_video is None:
            rng.shuffle(wrong_vids)
        else:
            # T2 少的错题组优先进 val（保留 T2 高的组在 diag），确定性排序
            wrong_vids.sort(key=lambda v: (wrong_tier_by_video.get(v, 0), v))
        val_vids = set(correct_vids[:val_correct] + wrong_vids[:val_wrong])
```
在 `val_vids` 确定后、返回前，加**功效修复**（从 diag 的错题组按 T2 升序补入 val 直到满足 val_wrong_min）：
```python
    if val_wrong_min > 0:
        val_wrong_now = sum(
            1 for v in val_vids for q in groups[v] if not correctness[q.question_id]
        )
        # diag 侧仍在的错题组，按 T2 升序（低价值优先移交 val）
        diag_wrong_pool = sorted(
            (v for v in wrong_vids if v not in val_vids),
            key=lambda v: ((wrong_tier_by_video or {}).get(v, 0), v),
        )
        for v in diag_wrong_pool:
            if val_wrong_now >= val_wrong_min:
                break
            val_vids.add(v)
            val_wrong_now += sum(1 for q in groups[v] if not correctness[q.question_id])
        if val_wrong_now < val_wrong_min:
            raise InsufficientValSignal(
                f"trainval 错题不足以让 val 达到 val_wrong_min={val_wrong_min}"
                f"（修复后仅 {val_wrong_now}），请放大 val_ratio 或调整 trainval 归属。"
            )
```
（`InsufficientValSignal` 已在 pools.py:118 定义，无需新增；需确认函数内可见 `math`/`defaultdict`，文件顶部已 import。）

- [ ] **Step 4: `split_by_video_assignment` 透传新参数**

`app/harness/pools.py` `split_by_video_assignment` 签名加 `wrong_tier_by_video: dict[str, int] | None = None`（放在 `val_wrong_min` 之后），并把 `_split_trainval_by_video_group` 调用（L175-177）改为：
```python
    diagnosis, validation = _split_trainval_by_video_group(
        trainval_qs, correctness, val_ratio, random.Random(seed),
        wrong_tier_by_video=wrong_tier_by_video,
        val_wrong_min=val_wrong_min,
    )
```
删除原 L179-186 的独立 `val_wrong_min` 事后校验块（功效已在 `_split_trainval_by_video_group` 内保证，避免重复校验语义）。docstring 的 `val_wrong_min` 说明改为"切分时保证（不足则从 diag 换入低 T2 错题组补足，耗尽 fail-loud）"。

- [ ] **Step 5: build_split 计算并传入 tier + val_wrong_min**

`app/harness/build_split.py`：`SplitBuildConfig` 加字段 `val_wrong_min: int`（放 `split_seed` 之后，docstring 补"validation 池最少错题数，切分时保证功效"）。build_split Phase 3（L182-191）改为：
```python
    questions = load_benchmark(questions_dir)
    correctness = {pred["question_id"]: pred["correct"] for pred in preds}
    tier_by_q = {row["question_id"]: row["tier"] for row in signal_rows}
    wrong_tier_by_video: dict[str, int] = defaultdict(int)
    for pred in preds:
        if not pred["correct"] and tier_by_q.get(pred["question_id"]) == "T2":
            wrong_tier_by_video[pred["video_id"]] += 1
    pools = split_by_video_assignment(
        questions,
        assignment,
        correctness,
        config.val_ratio,
        config.split_seed,
        baseline_run_id=baseline_run_id,
        val_wrong_min=config.val_wrong_min,
        wrong_tier_by_video=dict(wrong_tier_by_video),
    )
```
更新 build_split docstring 的"契约（Task 11...）"段：删除"有意保持 val_wrong_min-agnostic"表述，改为"val_wrong_min 前置到切分内保证功效；CLI 的 check_mcnemar_power 作冗余最终确认"。确认 `defaultdict` 已 import（`from collections import Counter, defaultdict`）。

- [ ] **Step 6: CLI 传 val_wrong_min + 保真检查**

`app/harness/video_split_cli.py` 的 `SplitBuildConfig(...)`（L565-573）加一行 `val_wrong_min=config.val_wrong_min,`。
保真检查点：确认 `pools.diagnosis`/`pools.validation` 仍是逐题 `GeneratedQuestion` 列表、同 video 全部题同池（`test_pools_video_atomic.py::test_video_group_atomic_in_trainval_split` 覆盖）。

- [ ] **Step 7: 运行测试确认通过 + 回归**

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

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

```bash
git add app/harness/pools.py app/harness/build_split.py app/harness/video_split_cli.py tests/unit/test_pools_video_atomic.py
git commit -m "feat: tier-aware diag/val split with val-power repair (design 5.1)"
```

---

## Task 4: 冻结产物覆盖保护 + --force

**Files:**
- Modify: `app/harness/build_split.py:104-192`
- Modify: `app/harness/video_split_cli.py:751-773`（build_arg_parser）+ run_pipeline 传参
- Test: `tests/unit/test_video_split_cli.py`

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

在 `tests/unit/test_video_split_cli.py` 追加（用最小 build_split 覆盖场景，或直接测保护函数）：
```python
def test_build_split_refuses_overwrite_without_force(tmp_path):
    """已存在指纹不同的 pools.json 时，force=False 必须报错不覆盖。"""
    from app.harness.build_split import _guard_frozen_products

    out_path = tmp_path / "pools.json"
    out_path.write_text('{"split_mode":"global"}', encoding="utf-8")
    manifest_path = tmp_path / "split_manifest.json"

    with pytest.raises(FileExistsError, match="已存在冻结产物"):
        _guard_frozen_products(out_path, manifest_path, force=False)


def test_build_split_force_backs_up_old(tmp_path):
    """force=True 时旧产物被备份为 .bak.* 再允许覆盖。"""
    from app.harness.build_split import _guard_frozen_products

    out_path = tmp_path / "pools.json"
    out_path.write_text('{"old":1}', encoding="utf-8")
    manifest_path = tmp_path / "split_manifest.json"
    manifest_path.write_text('{"pools_sha256":"deadbeef00000000"}', encoding="utf-8")

    _guard_frozen_products(out_path, manifest_path, force=True)
    baks = list(tmp_path.glob("pools.json.bak.*"))
    assert len(baks) == 1, f"未备份旧产物: {list(tmp_path.iterdir())}"
```

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

Run: `conda run -n Video-Tree-TRM python -m pytest tests/unit/test_video_split_cli.py -k "refuses_overwrite or force_backs_up" -v`
Expected: FAIL（`cannot import name '_guard_frozen_products'`）。

- [ ] **Step 3: 实现覆盖保护函数**

`app/harness/build_split.py` 顶部 import 区确认有 `import shutil`（无则加）。新增函数（放 build_split 之前）：
```python
def _guard_frozen_products(out_path: Path, manifest_path: Path, *, force: bool) -> None:
    """冻结前的覆盖保护：产物已存在时按 force 决定报错或备份。

    参数:
        out_path: 目标 pools.json 路径。
        manifest_path: 目标 split_manifest.json 路径。
        force: False 时已存在即 FileExistsError；True 时把旧产物重命名为
            .bak.<旧 pools_sha256 前 8 位或 timestamp-less 序号> 再放行。

    异常:
        FileExistsError: force=False 且产物已存在（防静默覆盖冻结锚点）。
    """
    if not out_path.exists() and not manifest_path.exists():
        return
    if not force:
        raise FileExistsError(
            f"已存在冻结产物 {out_path}（或其 manifest）。重跑切分会覆盖训练依赖的"
            "冻结锚点——确认要替换请加 --force（旧产物将备份为 .bak.*）。"
        )
    # 备份后缀取旧 manifest 的 pools_sha256 前 8 位，无则用 'prev'
    suffix = "prev"
    if manifest_path.exists():
        try:
            old = json.loads(manifest_path.read_text(encoding="utf-8"))
            suffix = str(old.get("pools_sha256", "prev"))[:8] or "prev"
        except (json.JSONDecodeError, OSError):
            suffix = "prev"
    for p in (out_path, manifest_path):
        if p.exists():
            p.rename(p.with_name(f"{p.name}.bak.{suffix}"))
```
确认 build_split.py 已 import `json`（无则加 `import json`）。在 `build_split` 签名加参数 `force: bool = False`（放 `generated_at` 之后），并在 Phase 3 `save_pools` 之前（L192 前）调用 `_guard_frozen_products(out_path, manifest_path, force=force)`。

- [ ] **Step 4: CLI 暴露 --force 并透传**

`app/harness/video_split_cli.py` `build_arg_parser`（L751-773）追加：
```python
    parser.add_argument(
        "--force",
        action="store_true",
        help="覆盖已存在的冻结 pools.json/manifest（旧产物备份为 .bak.*）",
    )
```
`run_pipeline` 签名加 `force: bool = False` 参数，build_split 调用（L559-577）加 `force=force,`；`main()` 里把 `args.force` 透传给 `run_pipeline`。

- [ ] **Step 5: 运行测试确认通过 + CLI 回归**

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

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

```bash
git add app/harness/build_split.py app/harness/video_split_cli.py tests/unit/test_video_split_cli.py
git commit -m "feat: guard frozen split products against silent overwrite (--force)"
```

---

## Task 5: seed 携带 pools.json + 训练拷入

**Files:**
- Modify: `app/harness/store.py:184-232`（init_seed）
- Modify: `app/harness/workspace.py:156-200`（init_workspace_from_seed）
- Test: `tests/unit/test_harness_store.py`、`tests/unit/test_harness_workspace.py`

- [ ] **Step 1: 写失败测试（seed 携带）**

在 `tests/unit/test_harness_store.py::TestInitSeed` 追加：
```python
    def test_init_seed_carries_pools(self, tmp_path):
        """提供 pools_json/split_manifest 时拷入 seed 目录。"""
        from app.harness.store import init_seed

        store = tmp_path / "store"
        skills = tmp_path / "sk"; skills.mkdir(); (skills / "s.md").write_text("x")
        prompts = tmp_path / "pr"; prompts.mkdir(); (prompts / "p.md").write_text("y")
        db = tmp_path / "b.db"; db.write_text("db")
        pools = tmp_path / "pools.json"; pools.write_text('{"split_mode":"global"}')
        manifest = tmp_path / "split_manifest.json"; manifest.write_text('{"pools_sha256":"a"}')

        seed_dir = init_seed(
            store, "s1", skills, prompts, db, "infer_adhoc", None, "d",
            pools_json=pools, split_manifest=manifest,
        )
        assert (seed_dir / "pools.json").exists()
        assert (seed_dir / "split_manifest.json").exists()
```

在 `tests/unit/test_harness_workspace.py` 追加：
```python
def test_init_workspace_from_seed_carries_pools(store_dir, workspace_dir):
    """seed 目录含 pools.json 时拷入 workspace。"""
    import shutil
    from app.harness.store import init_seed
    from app.harness.workspace import init_workspace_from_seed
    # 复用现有 fixture 构造 seed 的方式；此处补 pools.json 到 seed 后初始化 workspace
    # （具体 fixture 依 test_harness_workspace.py 现有 helper，读文件对齐）
    ...
```
> 该 workspace 测试需依 `test_harness_workspace.py` 现有 fixture（`store_dir`/`workspace_dir` 及既有 seed 构造 helper）填充；实现前读该文件 `test_init_workspace_from_seed`（L151）复用其 seed 搭建，再在 seed 目录写 `pools.json` 后断言 workspace 内出现 `pools.json`。

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

Run: `conda run -n Video-Tree-TRM python -m pytest tests/unit/test_harness_store.py::TestInitSeed::test_init_seed_carries_pools -v`
Expected: FAIL（`unexpected keyword argument 'pools_json'`）。

- [ ] **Step 3: init_seed 加可选携带参数**

`app/harness/store.py` `init_seed` 签名加：
```python
def init_seed(
    store_dir: Path,
    name: str,
    skills_dir: Path,
    prompts_dir: Path,
    baseline_db: Path,
    baseline_run_id: str,
    parent: str | None,
    description: str,
    *,
    pools_json: Path | None = None,
    split_manifest: Path | None = None,
) -> Path:
```
在 `copy2(baseline_db, ...)`（L218）之后加：
```python
    if pools_json is not None:
        shutil.copy2(pools_json, seed_dir / "pools.json")
    if split_manifest is not None:
        shutil.copy2(split_manifest, seed_dir / "split_manifest.json")
```
docstring 补两参说明。

- [ ] **Step 4: init_workspace_from_seed 拷入 pools**

`app/harness/workspace.py` `init_workspace_from_seed` 在 `shutil.copy2(seed_dir / "baseline.db", workspace_dir / "harness.db")`（L197）之后加：
```python
    seed_pools = seed_dir / "pools.json"
    if seed_pools.exists():
        shutil.copy2(seed_pools, workspace_dir / "pools.json")
        seed_manifest = seed_dir / "split_manifest.json"
        if seed_manifest.exists():
            shutil.copy2(seed_manifest, workspace_dir / "split_manifest.json")
```

- [ ] **Step 5: 运行测试确认通过 + 回归**

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

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

```bash
git add app/harness/store.py app/harness/workspace.py tests/unit/test_harness_store.py tests/unit/test_harness_workspace.py
git commit -m "feat: seed carries frozen pools.json into training workspace"
```

---

## Task 6: build_or_load_pools global 一致性校验

**Files:**
- Modify: `app/harness/pools.py:746-813`
- Test: `tests/unit/test_harness_pools.py`

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

在 `tests/unit/test_harness_pools.py::TestBuildOrLoadPoolsFrozen` 追加：
```python
    def test_global_frozen_rejects_baseline_mismatch(self, tmp_path, ...):
        """global 冻结 pools 的 baseline_run_id 与 seed 不符时 fail-loud。"""
        # 依现有 fixture 造 workspace + 冻结 pools.json（split_mode=global,
        # baseline_run_id="other"），seed.json baseline_run_id="infer_adhoc"
        # 调 build_or_load_pools 应 raise ValueError(match="baseline_run_id")
        ...
```
> 依 `TestBuildOrLoadPoolsFrozen`（L246）现有 fixture 复用其 workspace/seed 搭建；实现前读该类对齐 RunConfig/strategy 构造，勿臆造。

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

Run: `conda run -n Video-Tree-TRM python -m pytest tests/unit/test_harness_pools.py::TestBuildOrLoadPoolsFrozen -v`
Expected: 新用例 FAIL（当前 global 分支无校验，误加载不报错）。

- [ ] **Step 3: 加 global 一致性校验**

`app/harness/pools.py` `build_or_load_pools`，在 global 加载分支（`if pools_path.exists():` 块内、`per_category` 校验的 `else` 侧，即 L813 `return load_pools(pools_path)` 之前）加：
```python
        else:  # global：校验 baseline_run_id 与（若有）manifest 内容指纹
            frozen_baseline = raw.get("baseline_run_id")
            if frozen_baseline != baseline_run_id:
                raise ValueError(
                    f"冻结 pools.json 的 baseline_run_id={frozen_baseline!r} 与 seed "
                    f"的 {baseline_run_id!r} 不一致，拒绝静默加载错配切分。"
                )
            manifest_path = config.workspace_dir / "split_manifest.json"
            if manifest_path.exists():
                import hashlib

                manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
                actual_sha = hashlib.sha256(
                    pools_path.read_text(encoding="utf-8").encode("utf-8")
                ).hexdigest()
                if manifest.get("pools_sha256") != actual_sha:
                    raise ValueError(
                        "pools.json 内容指纹与 split_manifest.pools_sha256 不符，"
                        "冻结产物疑被篡改，拒绝加载。"
                    )
```
（确认该 `else` 与 L751 `if frozen_split_mode == "per_category":` 配对；若现有结构非 if/else 而是 if 后直接 return，则把校验插在 `return load_pools(pools_path)` 前并用 `if frozen_split_mode != "per_category":` 守卫。实现前读 L746-813 对齐控制流。）

- [ ] **Step 4: 运行测试确认通过 + 回归**

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

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

```bash
git add app/harness/pools.py tests/unit/test_harness_pools.py
git commit -m "fix: validate global frozen pools baseline_run_id + sha256 on load"
```

---

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

- [ ] val_ratio=0.4 已改；tier 感知 + 功效修复在同一函数、退化路径（`wrong_tier_by_video=None`）保持旧行为。
- [ ] val_wrong_min 从 CLI→SplitBuildConfig→build_split→split_by_video_assignment→_split_trainval_by_video_group 全链路贯通；旧的 pools.py 事后校验块已删（不重复）。
- [ ] extract_run_db 去重默认关闭，不破坏既有调用。
- [ ] seed 携带 + workspace 拷入 + global 一致性校验三者闭环：冻结产物有唯一路径进训练且被校验。
- [ ] 覆盖保护默认 force=False，离线 CLI 重跑需显式 --force。

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

本计划触及算法 #5 的上游输入（哪些视频进 diag/val），**不改** gate_ladder 的 unit+correctness 消费结构；Task 3 Step 6 已设保真检查点确认逐 unit 列表与视频组原子性。不涉及算法 #4/#6/#8/#9 逻辑。

## 验收标准

1. `pytest tests/unit/test_pools_video_atomic.py tests/unit/test_harness_pools.py tests/unit/test_harness_store.py tests/unit/test_harness_workspace.py tests/unit/test_split_selection.py tests/unit/test_video_split_cli.py` 全绿。
2. tier 感知：T2 高的错题视频组留 diag，T2 低的优先进 val。
3. seed 携带 pools.json → init_workspace_from_seed 拷入 → build_or_load_pools 校验 baseline_run_id + sha256。
4. 冻结产物 force=False 时拒绝覆盖。