Video-Tree-TRM5/research-wiki/plans/2026-07-09-main-inference-entry.md

# main.py 推理入口 + 初始 Prompt 集 实现计划

> **For agentic workers:** REQUIRED SUB-SKILL: Use subagent-driven-development to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.

**Goal:** 实现 `main.py` CLI 入口（infer 模式），准备初始 prompt/skill 集，完成 900 道题推理基线。

**Architecture:** Clean Architecture Composition Root 模式。main.py 构建全套 adapters 和 InferenceDepsRouter，注入 Runner。Router 按 video_id 懒加载 per-video InferenceDeps，在 prompt_builder 调用时注册 question_id→video_id 映射，dispatch 通过 context["session_id"] 路由。

**Tech Stack:** Python 3.11, asyncio, argparse, pydantic-settings, loguru

**设计文档:** `research-wiki/designs/2026-07-09-main-inference-entry-design.md`

---

## File Structure

| 操作 | 文件 | 职责 |
|------|------|------|
| **Task 0: subtitle 迁入 Card** | | |
| Modify | `app/tree/index.py:73-194` | L3Card/L2Card 加 subtitle 字段；L3Node 去 subtitle 字段 |
| Modify | `app/tree/index.py:267-390` | to_dict/from_dict 序列化适配 |
| Modify | `app/tree/subtitle.py:239-308` | Voronoi 写入 Card.subtitle 而非 Node.subtitle |
| Modify | `app/tree/environment.py:448-489` | _node_full_text/_node_anchored_text 简化 |
| Modify | `app/tree/environment.py:367-381` | get_subtitle 读 Card.subtitle |
| Modify | `app/tree/video_builder.py:475-486` | Phase 7 保留但改写 Card.subtitle |
| Modify | `app/tree/verify.py` | `l3.subtitle` → `l3.card.subtitle` |
| Modify | `app/question_gen/synthesizer.py` | `l3.subtitle` / `l2.subtitle` → `card.subtitle` |
| Create | `tools/migrate_subtitle_to_card.py` | 迁移脚本：300 棵树 subtitle 从 Node 移入 Card |
| Modify | 9 个测试文件 | Card 构造加 subtitle 参数 |
| **Task 1-8: 原有任务** | | |
| Create | `app/harness/deps_router.py` | 按 video_id 懒加载 InferenceDeps 并路由 dispatch/prompt_builder |
| Modify | `app/ports.py` (文件末尾追加) | 新增 4 个 Protocol |
| Modify | `app/harness/runner.py:454-469` | __init__ 增加 2 个 factory 参数 + fail-fast 校验 |
| Modify | `app/harness/runner.py:2064-2082` | _make_* 方法优先用注入值 |
| Create | `main.py` | Composition Root |
| Move | `store/prompts/*.md` → `store/prompts/v1/` | 版本化目录重组 |
| Create | `store/skills/v1/` (13 files) | 从 TRM4 v1 精简 + 注入 TRM5 card 字段 |
| Modify | `config/default.yaml:29,31` | concurrency=24, max_steps=40 |
| Test | `tests/unit/test_deps_router.py` | Router 单元测试 |
| Test | `tests/unit/test_ports_factory.py` | Protocol 结构测试 |

---

### Task 0: subtitle 迁入 Card + 建树管线修正 + 300 棵树迁移

**Files:**
- Modify: `app/tree/index.py` (L3Card, L2Card, L3Node, to_dict, from_dict)
- Modify: `app/tree/subtitle.py:239-308` (assign_subtitles_voronoi)
- Modify: `app/tree/environment.py:367-381,448-489` (_node_full_text, get_subtitle)
- Modify: `app/tree/video_builder.py:475-486,902-996` (builder pipeline)
- Create: `tools/migrate_subtitle_to_card.py`
- Modify: 9 个测试文件 (Card 构造适配)

**设计决策:**
- L3Card/L2Card 加 `subtitle: str = ""` 字段（放在字段列表末尾，`_collect_card_strings` 自动收集）
- L1Card 不加 subtitle（用户确认）
- L3Node 移除 `subtitle: str | None` 字段（数据迁入 Card）
- Card 保持 `frozen=True`，建树时在 Card 创建前计算好 subtitle 传入构造函数；迁移脚本用 `dataclasses.replace()` 创建新 Card
- 建树管线中 L2/L3 直接在 Card 构造时注入字幕，移除 Phase 7 Voronoi 后处理
- L2 subtitle = `extract_subtitle_for_range(srt_entries, l2_time_range)` 在 `_build_l2_video_async` 中计算
- L3 subtitle 仍用 Voronoi 逻辑（精确分配帧级字幕），但写入 Card 而非 Node

- [ ] **Step 1: 修改 Card dataclass**

在 `app/tree/index.py` 中：

**L3Card** (第 73-92 行) — 末尾加 `subtitle: str = ""`：

```python
@dataclass(frozen=True)
class L3Card:
    """L3 帧级语义卡片（不可变）。"""

    frame_summary: str
    visible_entities: list[str]
    ongoing_actions: list[str]
    visible_text: list[str]
    spatial_layout: str
    visual_attributes: dict[str, Any]
    subtitle: str = ""
```

**L2Card** (第 96-117 行) — 末尾加 `subtitle: str = ""`：

```python
@dataclass(frozen=True)
class L2Card:
    """L2 事件级语义卡片（不可变）。"""

    event_description: str
    entities: list[str]
    actions: list[str]
    action_subjects: list[str]
    visible_text: list[str]
    spatial_relations: str
    state_changes: str | None
    subtitle: str = ""
```

**L3Node** (第 176-194 行) — 移除 `subtitle` 字段：

```python
@dataclass
class L3Node:
    """L3 帧级语义节点（叶层）。"""

    id: str
    card: L3Card
    embedding: np.ndarray | None = None
    timestamp: float | None = None
    frame_path: str | None = None
    # subtitle 已迁入 L3Card，此处不再保留
```

- [ ] **Step 2: 修改序列化/反序列化**

**to_dict** — L3 `l3_to_dict` (第 267-284 行): subtitle 从 card 输出，移除节点级 subtitle：

```python
def l3_to_dict(n: L3Node) -> dict[str, Any]:
    d: dict[str, Any] = {
        "id": n.id,
        "card": {
            "frame_summary": n.card.frame_summary,
            "visible_entities": n.card.visible_entities,
            "ongoing_actions": n.card.ongoing_actions,
            "visible_text": n.card.visible_text,
            "spatial_layout": n.card.spatial_layout,
            "visual_attributes": n.card.visual_attributes,
            "subtitle": n.card.subtitle,
        },
        "timestamp": n.timestamp,
        "frame_path": n.frame_path,
    }
    if include_embedding:
        d["embedding"] = _embed_to_str(n.embedding)
    return d
```

**to_dict** — L2 `l2_to_dict` (第 286-303 行): card 字典加 subtitle：

```python
def l2_to_dict(n: L2Node) -> dict[str, Any]:
    d: dict[str, Any] = {
        "id": n.id,
        "card": {
            "event_description": n.card.event_description,
            "entities": n.card.entities,
            "actions": n.card.actions,
            "action_subjects": n.card.action_subjects,
            "visible_text": n.card.visible_text,
            "spatial_relations": n.card.spatial_relations,
            "state_changes": n.card.state_changes,
            "subtitle": n.card.subtitle,
        },
        "time_range": list(n.time_range) if n.time_range else None,
        "children": [l3_to_dict(c) for c in n.children],
    }
    if include_embedding:
        d["embedding"] = _embed_to_str(n.embedding)
    return d
```

**from_dict** — L3 反序列化 (第 337-354 行): subtitle 从 card 读取，兼容旧格式从节点级读取：

```python
l3_card = L3Card(
    frame_summary=l3d["card"]["frame_summary"],
    visible_entities=l3d["card"]["visible_entities"],
    ongoing_actions=l3d["card"]["ongoing_actions"],
    visible_text=l3d["card"]["visible_text"],
    spatial_layout=l3d["card"]["spatial_layout"],
    visual_attributes=l3d["card"]["visual_attributes"],
    subtitle=l3d["card"].get("subtitle", "") or l3d.get("subtitle", "") or "",
)
l3_nodes.append(
    L3Node(
        id=l3d["id"],
        card=l3_card,
        embedding=_embed_from_str(l3d.get("embedding")),
        timestamp=l3d.get("timestamp"),
        frame_path=l3d.get("frame_path"),
    )
)
```

**from_dict** — L2 反序列化 (第 355-363 行): 加 subtitle：

```python
l2_card = L2Card(
    event_description=l2d["card"]["event_description"],
    entities=l2d["card"]["entities"],
    actions=l2d["card"]["actions"],
    action_subjects=l2d["card"]["action_subjects"],
    visible_text=l2d["card"]["visible_text"],
    spatial_relations=l2d["card"]["spatial_relations"],
    state_changes=l2d["card"]["state_changes"],
    subtitle=l2d["card"].get("subtitle", ""),
)
```

- [ ] **Step 3: 修改 environment.py**

**`get_subtitle`** (第 367-381 行) — 读 Card.subtitle，L1/L2/L3 均支持：

```python
def get_subtitle(self, node_id: str) -> str:
    """返回节点字幕文本。L2/L3 从 Card 读取，L1 返回空串。"""
    node = self._id_to_node.get(node_id)
    if node is None:
        return ""
    if isinstance(node, (L2Node, L3Node)):
        return node.card.subtitle or ""
    return ""
```

**`_collect_card_strings`** (第 67-78 行) — 新增 `skip_fields` 参数，排除 subtitle（subtitle 需单独标签/锚标处理）：

```python
def _collect_card_strings(node: AnyNode, *, skip_fields: frozenset[str] = frozenset()) -> list[str]:
    """从节点 card 中递归收集所有非空字符串字段，可排除指定字段。"""
    result: list[str] = []
    _collect_from_obj(node.card, result, skip_fields=skip_fields)
    return result
```

`_collect_from_obj` (第 81-101 行) — 在 dataclass 分支中跳过 `skip_fields`：

```python
def _collect_from_obj(obj: object, out: list[str], *, skip_fields: frozenset[str] = frozenset()) -> None:
    if isinstance(obj, str):
        stripped = obj.strip()
        if stripped:
            out.append(stripped)
    elif isinstance(obj, dict):
        for v in obj.values():
            _collect_from_obj(v, out, skip_fields=skip_fields)
    elif isinstance(obj, (list, tuple)):
        for item in obj:
            _collect_from_obj(item, out, skip_fields=skip_fields)
    elif hasattr(obj, "__dataclass_fields__"):
        for field_name in obj.__dataclass_fields__:
            if field_name in skip_fields:
                continue
            _collect_from_obj(getattr(obj, field_name), out, skip_fields=skip_fields)
```

**`_node_full_text`** (第 448-461 行) — 保持"字幕:"标签，改为读 `card.subtitle`：

```python
_SUBTITLE_SKIP = frozenset({"subtitle"})

def _node_full_text(self, node: AnyNode) -> str:
    """获取节点完整文本（card 非 subtitle 字段 + 带标签的字幕）。"""
    card_strings = _collect_card_strings(node, skip_fields=_SUBTITLE_SKIP)
    text = "\n".join(card_strings)
    if isinstance(node, (L2Node, L3Node)) and node.card.subtitle:
        text += f"\n字幕: {node.card.subtitle}"
    return text
```

**`_node_anchored_text`** (第 463-489 行) — 保持 [cN]/[sN] 双锚语义，改为读 `card.subtitle`：

```python
def _node_anchored_text(self, node: AnyNode) -> str:
    """获取带行号锚的节点文本。card 字段 [cN]，字幕 [sN]。"""
    card_strings = _collect_card_strings(node, skip_fields=_SUBTITLE_SKIP)
    card_lines: list[str] = []
    for s in card_strings:
        card_lines.extend(ln for ln in s.splitlines() if ln.strip())

    sub_lines: list[str] = []
    if isinstance(node, (L2Node, L3Node)) and node.card.subtitle:
        sub_lines = [ln for ln in node.card.subtitle.splitlines() if ln.strip()]

    anchored: list[str] = []
    for i, line in enumerate(card_lines, 1):
        anchored.append(f"[c{i}] {line}")
    for i, line in enumerate(sub_lines, 1):
        anchored.append(f"[s{i}] {line}")
    return "\n".join(anchored)
```

- [ ] **Step 4: 修改 subtitle.py — Voronoi 写入 Card**

`assign_subtitles_voronoi` (第 239-308 行) — 用 `dataclasses.replace()` 写入 Card.subtitle：

在函数顶部加 `import dataclasses`，然后将第 302 行：
```python
# 原: l3.subtitle = subtitle_text if subtitle_text else None
# 改:
l3.card = dataclasses.replace(l3.card, subtitle=subtitle_text or "")
```

同时在函数末尾为每个 L2 聚合子节点字幕：

```python
# L2 字幕聚合（在 L3 分配完成后）
for l1 in index.roots:
    for l2 in l1.children:
        l2_sub_parts = [c.card.subtitle for c in l2.children if c.card.subtitle]
        if l2_sub_parts:
            l2.card = dataclasses.replace(l2.card, subtitle="\n".join(l2_sub_parts))
```

- [ ] **Step 5: 修改 video_builder.py — 建树管线**

Phase 7 保留 `assign_subtitles_voronoi`（它现在写 Card.subtitle + L2 聚合），不在 `_build_l2_video_async` 中注入字幕（避免双写）。仅更新注释：

```python
# Phase 7: 字幕注入 Card（L3 Voronoi 分配 + L2 聚合）
if srt_entries:
    assign_subtitles_voronoi(index, srt_entries)
    logger.info("字幕已注入 Card: L3 Voronoi + L2 聚合", n_entries=len(srt_entries))
```

`_build_l2_video_async` 和 `_build_l3_video_async` 不做字幕注入改动。字幕注入统一由 Phase 7 `assign_subtitles_voronoi` 负责。

- [ ] **Step 5.5: 修改 verify.py 和 synthesizer.py 中的 subtitle 访问**

搜索并替换所有 `l3.subtitle` → `l3.card.subtitle`，`l2.subtitle` → `l2.card.subtitle`（若存在）：

```bash
grep -rn "\.subtitle" app/tree/verify.py app/question_gen/synthesizer.py
```

对每处命中做替换，例如：
- `app/tree/verify.py`: `node.subtitle` → `node.card.subtitle`（仅 L3/L2 节点）
- `app/question_gen/synthesizer.py`: `l3.subtitle` → `l3.card.subtitle`

- [ ] **Step 6: 迁移脚本**

```python
# tools/migrate_subtitle_to_card.py
"""迁移 300 棵树：L3 subtitle 从 Node 级移入 Card 级，L2 聚合子节点字幕。"""

import dataclasses
import json
import sys
from pathlib import Path

from loguru import logger


def migrate_tree(tree_path: Path) -> bool:
    """迁移单棵树，返回是否有变更。幂等：已迁移的树不会被修改。"""
    data = json.loads(tree_path.read_text(encoding="utf-8"))
    changed = False

    for l1 in data.get("roots", []):
        for l2 in l1.get("children", []):
            l2_sub_parts: list[str] = []

            for l3 in l2.get("children", []):
                # L3: node 级 subtitle 迁入 card（幂等：已有 card subtitle 则跳过）
                card_sub = l3.get("card", {}).get("subtitle", "")
                node_sub = l3.get("subtitle")  # None = 从未有字幕

                if node_sub is not None and "subtitle" in l3:
                    # 旧格式：有 node 级 subtitle
                    final_sub = card_sub or (node_sub if node_sub else "")
                    l3["card"]["subtitle"] = final_sub
                    del l3["subtitle"]
                    changed = True
                elif "subtitle" not in l3.get("card", {}):
                    # 无任何 subtitle 数据
                    l3.setdefault("card", {})["subtitle"] = ""
                    changed = True

                # 收集 L3 字幕用于 L2 聚合
                effective_sub = l3.get("card", {}).get("subtitle", "")
                if effective_sub:
                    l2_sub_parts.append(effective_sub)

            # L2: 聚合 L3 字幕（幂等：已有 card subtitle 则跳过）
            l2_card = l2.setdefault("card", {})
            if "subtitle" not in l2_card:
                l2_card["subtitle"] = "\n".join(l2_sub_parts) if l2_sub_parts else ""
                changed = True

    if changed:
        tree_path.write_text(
            json.dumps(data, ensure_ascii=False, indent=2),
            encoding="utf-8",
        )
    return changed


def main() -> None:
    """迁移 store/videos/ 下所有 tree.json。"""
    videos_dir = Path(sys.argv[1]) if len(sys.argv) > 1 else Path("store/videos")
    tree_files = sorted(videos_dir.glob("*/tree.json"))
    logger.info("发现 {} 棵树待迁移", len(tree_files))

    migrated = 0
    for tf in tree_files:
        if migrate_tree(tf):
            migrated += 1
    logger.info("迁移完成: {}/{} 棵树有变更", migrated, len(tree_files))


if __name__ == "__main__":
    main()
```

Run: `conda run -n Video-Tree-TRM python tools/migrate_subtitle_to_card.py store/videos`

验证：
```bash
python3 -c "
import json; d=json.loads(open('store/videos/068rdc75mHM/tree.json').read())
l1=d['roots'][0]; l2=l1['children'][0]; l3=l2['children'][0]
print('L3 card subtitle:', bool(l3['card'].get('subtitle')))
print('L3 node subtitle:', 'subtitle' in l3 and l3.get('subtitle') is not None)
print('L2 card subtitle:', bool(l2['card'].get('subtitle')))
"
```
Expected: L3 card subtitle: True, L3 node subtitle: False, L2 card subtitle: True

- [ ] **Step 7: 更新测试文件**

需要更新的 9 个文件中所有 `L3Card(...)` 和 `L2Card(...)` 构造调用。由于 `subtitle` 有默认值 `""`，大多数测试不需要改动（默认空串即可）。但 `L3Node(...)` 构造中如果传了 `subtitle=` 参数需要移除。

需要检查并修改的具体文件：

```
tests/integration/test_tree_build_e2e.py
tests/unit/test_repair_detector.py
tests/unit/test_repair_regenerator.py
tests/unit/test_search_tools.py
tests/unit/test_subtitle.py
tests/unit/test_tree_environment.py
tests/unit/test_tree_index.py
tests/unit/test_verify.py
tests/unit/test_video_builder.py
```

对每个文件：
1. `L3Node(..., subtitle=xxx)` → 移除 `subtitle=` 参数，改为在 L3Card 构造时传入 `subtitle=xxx`
2. `L3Card(a, b, c, d, e, f)` 位置参数调用 → 不影响（subtitle 有默认值）
3. 确认 `L2Card(...)` 构造不受影响（subtitle 有默认值）

- [ ] **Step 8: 运行全量测试**

Run: `conda run -n Video-Tree-TRM pytest tests/ -x -q`
Expected: ALL PASSED

- [ ] **Step 9: Commit**

```bash
git add app/tree/index.py app/tree/subtitle.py app/tree/environment.py app/tree/video_builder.py tools/migrate_subtitle_to_card.py tests/
git commit -m "refactor(tree): subtitle 迁入 L3Card/L2Card + 建树管线修正 + 300 棵树迁移

- L3Card/L2Card 新增 subtitle: str 字段
- L3Node 移除 subtitle 字段（数据迁入 Card）
- assign_subtitles_voronoi 改写 Card.subtitle
- _node_full_text 简化（Card 已含 subtitle）
- 建树管线 L2 直接注入字幕到 Card
- 迁移脚本处理 300 棵现有树"
```

---

### Task 1: store/ 目录重组

**Files:**
- Move: `store/prompts/*.md` → `store/prompts/v1/*.md`
- Create: `store/skills/v1/` (empty, 后续 Task 填充)

- [ ] **Step 1: 创建 v1 子目录并移动 prompt 文件**

```bash
mkdir -p store/prompts/v1
git mv store/prompts/system.md store/prompts/v1/
git mv store/prompts/observe_frame_extract.md store/prompts/v1/
git mv store/prompts/observe_frame_verify.md store/prompts/v1/
git mv store/prompts/search_similar_extract.md store/prompts/v1/
git mv store/prompts/search_similar_verify.md store/prompts/v1/
git mv store/prompts/view_node_extract.md store/prompts/v1/
git mv store/prompts/view_node_verify.md store/prompts/v1/
git mv store/prompts/view_node_children_extract.md store/prompts/v1/
git mv store/prompts/view_node_children_verify.md store/prompts/v1/
```

- [ ] **Step 2: 创建 skills/v1 目录**

```bash
mkdir -p store/skills/v1
```

- [ ] **Step 3: 修正 system.md — L1 去掉 subtitle 描述**

编辑 `store/prompts/v1/system.md`，在 L1 字段表中删除 `| subtitle | 完整字幕（较长） |` 行。在信任层级段落末尾将"三个层级都包含 visible_text 和 subtitle 字段"改为"三个层级都包含 visible_text 字段，L2 和 L3 额外包含 subtitle 字段"。

- [ ] **Step 4: 验证目录结构**

Run: `ls store/prompts/v1/ && ls store/skills/v1/`
Expected: 9 个 .md 文件在 prompts/v1/ 下，skills/v1/ 为空目录。

Run: `grep subtitle store/prompts/v1/system.md`
Expected: 仅 L2 和 L3 部分出现 subtitle。

- [ ] **Step 4: Commit**

```bash
git add store/prompts/ store/skills/
git commit -m "refactor(store): prompts 版本化目录重组 + skills/v1 骨架"
```

---

### Task 2: store/skills/v1 — 13 个精简 + 注入 skill

**Files:**
- Create: `store/skills/v1/default-strategy.md`
- Create: `store/skills/v1/action-reasoning.md`
- Create: `store/skills/v1/action-recognition.md`
- Create: `store/skills/v1/attribute-perception.md`
- Create: `store/skills/v1/counting-problem.md`
- Create: `store/skills/v1/information-synopsis.md`
- Create: `store/skills/v1/object-reasoning.md`
- Create: `store/skills/v1/object-recognition.md`
- Create: `store/skills/v1/ocr-problems.md`
- Create: `store/skills/v1/spatial-perception.md`
- Create: `store/skills/v1/spatial-reasoning.md`
- Create: `store/skills/v1/temporal-perception.md`
- Create: `store/skills/v1/temporal-reasoning.md`

**变换规则**（每个 skill 文件统一适用）：

| 操作 | 对象 | 说明 |
|------|------|------|
| 保留 | YAML frontmatter | name, description, task_type 原样保留 |
| 保留 | `## 适用场景` 节 | 原样保留 |
| 保留 | `## 搜索步骤` 的 Step 标题 | 如 `### Step 1: 事件定位` |
| 精简 | Step 正文 | 保留第一句话意图描述，移除数据驱动统计（如 "75% 正确率"）、精确转换条件、详细操作指令 |
| 保留 | `## 输出格式` 节 | JSON schema (reflect/plan/action) 原样保留 |
| 精简 | `## 自检信号` 节 | 最多保留 1 条最通用的自检信号 |
| 精简 | `## 常见陷阱` 节 | 最多保留 2 条最通用的陷阱警告，移除特定失败模式 |
| **新增** | `## 视频树字段索引` 节 | 插入 card 字段索引表（见下方） |

**card 字段索引表**（全部 13 个 skill 共用，插入在 `## 搜索步骤` 之前）：

```markdown
## 视频树字段索引

| 层级 | 字段 | 适用场景 |
|------|------|---------|
| L1 | scene_summary | 整体概况 |
| L1 | key_entities | 查找人物/物体 |
| L1 | main_actions | 主要动作 |
| L1 | temporal_flow | 时间线概览 |
| L1 | topic_keywords | 主题定位 |
| L2 | event_description | 事件因果 |
| L2 | entities / actions | 实体和动作细节 |
| L2 | state_changes | 状态转变 |
| L2 | spatial_relations | 空间关系变化 |
| L3 | frame_summary | 精确视觉证据 |
| L3 | visible_entities | 具体物体确认 |
| L3 | ongoing_actions | 正在发生的动作 |
| L3 | spatial_layout | 精确空间位置 |
| L3 | visual_attributes | 光照、色调、机位 |
| L2/L3 | subtitle | 字幕转写（L1 无此字段） |
| 全层 | visible_text | 画面文字（OCR） |
```

**TRM4 v1 源文件路径:** `/home/iomgaa/Projects/Video-Tree-TRM4/store/skills/v1/`

- [ ] **Step 1: 逐个读取 TRM4 v1 skill，按变换规则精简 + 注入，写入 store/skills/v1/**

对每个 skill 文件执行：
1. 读取 `/home/iomgaa/Projects/Video-Tree-TRM4/store/skills/v1/<name>.md`
2. 应用上述变换规则
3. 写入 `store/skills/v1/<name>.md`

以 `default-strategy.md` 为首个示例（最重要的通用策略）。

- [ ] **Step 2: 验证 13 个文件完整性**

Run: `ls store/skills/v1/ | wc -l`
Expected: 13

Run: `head -5 store/skills/v1/default-strategy.md`
Expected: YAML frontmatter with `task_type: _default`

Run: `grep "视频树字段索引" store/skills/v1/*.md | wc -l`
Expected: 13（每个文件都有字段索引表）

- [ ] **Step 3: Commit**

```bash
git add store/skills/v1/
git commit -m "feat(store): skills/v1 初始集 — TRM4 精简 + TRM5 card 字段注入"
```

---

### Task 3: app/ports.py — 新增 4 个 Protocol

**Files:**
- Modify: `app/ports.py:76` (文件末尾追加)
- Test: `tests/unit/test_ports_factory.py`

- [ ] **Step 1: 编写 Protocol 结构测试**

```python
# tests/unit/test_ports_factory.py
"""ToolDispatchFactory / PromptBuilderFactory Protocol 结构验证。"""

from __future__ import annotations

from pathlib import Path
from typing import Any

import pytest

from app.ports import (
    PromptBuilderFactory,
    PromptBuilderFn,
    ToolDispatchFactory,
    ToolDispatchFn,
)


class TestToolDispatchFnProtocol:
    """ToolDispatchFn 签名检查。"""

    def test_conforming_callable_passes_isinstance(self) -> None:
        async def dispatch(
            tool_name: str, args: dict[str, Any], *, context: dict[str, Any]
        ) -> str:
            return ""

        assert isinstance(dispatch, ToolDispatchFn)

    def test_wrong_return_type_noted(self) -> None:
        """仅验证签名存在；runtime_checkable 不检查返回类型。"""
        async def bad(
            tool_name: str, args: dict[str, Any], *, context: dict[str, Any]
        ) -> int:
            return 0

        # runtime_checkable 仅检查方法存在，不验证类型注解
        assert isinstance(bad, ToolDispatchFn)


class TestToolDispatchFactoryProtocol:
    """ToolDispatchFactory 签名检查。"""

    def test_conforming_class_passes(self) -> None:
        class Factory:
            def __call__(self, *, skills_dir: Path | None = None) -> Any:
                return None

        assert isinstance(Factory(), ToolDispatchFactory)


class TestPromptBuilderFnProtocol:
    """PromptBuilderFn 签名检查。"""

    def test_conforming_callable_passes(self) -> None:
        from core.types import GeneratedQuestion

        def builder(qa: GeneratedQuestion) -> tuple[str, str]:
            return ("", "")

        assert isinstance(builder, PromptBuilderFn)


class TestPromptBuilderFactoryProtocol:
    """PromptBuilderFactory 签名检查。"""

    def test_conforming_class_passes(self) -> None:
        class Factory:
            def __call__(
                self, *, skills_dir: Path | None = None, prompts_dir: Path | None = None
            ) -> Any:
                return None

        assert isinstance(Factory(), PromptBuilderFactory)
```

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

Run: `conda run -n Video-Tree-TRM pytest tests/unit/test_ports_factory.py -v`
Expected: ImportError（ToolDispatchFn 等尚未定义）

- [ ] **Step 3: 在 app/ports.py 末尾追加 4 个 Protocol**

在 `app/ports.py` 文件末尾（第 77 行之后）追加：

```python


@runtime_checkable
class ToolDispatchFn(Protocol):
    """工具调度函数签名。

    参数:
        tool_name: 工具名称。
        args: 工具参数字典。
        context: 上下文字典（包含 session_id）。

    返回:
        工具执行结果文本。
    """

    async def __call__(
        self, tool_name: str, args: dict[str, Any], *, context: dict[str, Any]
    ) -> str: ...


@runtime_checkable
class ToolDispatchFactory(Protocol):
    """per-version 工具调度工厂。

    根据可选的 skills_dir 覆盖构建工具调度函数。

    参数:
        skills_dir: 可选的 skills 版本目录覆盖。

    返回:
        ToolDispatchFn 实例。
    """

    def __call__(self, *, skills_dir: Path | None = None) -> ToolDispatchFn: ...


@runtime_checkable
class PromptBuilderFn(Protocol):
    """Prompt 构建函数签名。

    参数:
        qa: 生成的题目实例。

    返回:
        (system_prompt, user_prompt) 二元组。
    """

    def __call__(self, qa: GeneratedQuestion) -> tuple[str, str]: ...


@runtime_checkable
class PromptBuilderFactory(Protocol):
    """per-version prompt 构建工厂。

    参数:
        skills_dir: 可选的 skills 版本目录覆盖。
        prompts_dir: 可选的 prompts 版本目录覆盖。

    返回:
        PromptBuilderFn 实例。
    """

    def __call__(
        self,
        *,
        skills_dir: Path | None = None,
        prompts_dir: Path | None = None,
    ) -> PromptBuilderFn: ...
```

同时在文件顶部 `if TYPE_CHECKING:` 块中确保 `Any` 已导入（已有 `from typing import ... Protocol`，需追加 `Any`）。

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

Run: `conda run -n Video-Tree-TRM pytest tests/unit/test_ports_factory.py -v`
Expected: 5 tests PASSED

- [ ] **Step 5: Commit**

```bash
git add app/ports.py tests/unit/test_ports_factory.py
git commit -m "feat(ports): 新增 ToolDispatchFactory/PromptBuilderFactory Protocol"
```

---

### Task 4: app/harness/runner.py — factory 注入

**Files:**
- Modify: `app/harness/runner.py:440-469` (__init__)
- Modify: `app/harness/runner.py:2064-2082` (_make_* 方法)
- Test: `tests/unit/test_harness_runner.py` (追加)

- [ ] **Step 1: 编写 factory 注入测试**

在 `tests/unit/test_harness_runner.py` 末尾追加：

```python
class TestRunnerFactoryInjection:
    """Runner factory 注入 fail-fast 校验。"""

    def test_infer_mode_missing_factory_raises(self, tmp_path: Path) -> None:
        """mode=infer 时缺少 factory 参数 → 立即 ValueError。"""
        from unittest.mock import AsyncMock
        from app.harness.config import RunConfig

        config = RunConfig(
            workspace_dir=tmp_path,
            store_dir=tmp_path,
            mode="infer",
            concurrency=1,
            max_steps=5,
            skill_mode="none",
            n_samples=0,
            questions="benchmarks/Video-MME",
            skills_version="v1",
            prompts_version="v1",
            epochs=1,
            diag_size=10,
            diag_correct_ratio=0.5,
            val_size=24,
            val_correct_ratio=0.5,
            edit_budget_start=5,
            edit_budget_end=2,
            batch_size=5,
            min_class_per_batch=2,
            eval_min_per_class=2,
            early_stop_patience=3,
            test_size=10,
            use_slow_momentum=False,
            gate_e_confirm=20.0,
            gate_e_provisional=3.0,
            gate_w_net_min=2,
            gate_delta_min=0.02,
            gate_lambda_dir=-0.642,
            gate_e_rollback=10.0,
            gate_block=8,
            gate_n_max=40,
            gate_p_low=0.05,
            gate_p_high=0.95,
            gate_probe_quota=0.2,
            gate_gamma_decay=0.9,
            gate_cooldown_steps=2,
            gate_guard_err=0.10,
            skill_update_mode="patch",
            appendix_consolidate_threshold=6,
        )

        with pytest.raises(ValueError, match="tool_dispatch_factory"):
            Runner(
                config,
                llm=AsyncMock(),
                evolve_llm=AsyncMock(),
                vlm=AsyncMock(),
                telemetry=AsyncMock(),
            )

    def test_diagnose_mode_allows_none_factory(self, tmp_path: Path) -> None:
        """mode=diagnose 不需要 factory（不走推理路径）→ 不报错。"""
        from unittest.mock import AsyncMock
        from app.harness.config import RunConfig

        ws = tmp_path / "ws"
        ws.mkdir()
        (ws / "manifest.json").write_text('{"name":"ws","created_at":"","store":"../store","current":{"videos":"v","questions":"q","skills":"s","prompts":"p"},"history":[]}')

        config = RunConfig(
            workspace_dir=ws,
            store_dir=tmp_path,
            mode="diagnose",
            run_id="test_run",
            concurrency=1,
            max_steps=5,
            skill_mode="none",
            n_samples=0,
            questions="benchmarks/Video-MME",
            skills_version="v1",
            prompts_version="v1",
            epochs=1,
            diag_size=10,
            diag_correct_ratio=0.5,
            val_size=24,
            val_correct_ratio=0.5,
            edit_budget_start=5,
            edit_budget_end=2,
            batch_size=5,
            min_class_per_batch=2,
            eval_min_per_class=2,
            early_stop_patience=3,
            test_size=10,
            use_slow_momentum=False,
            gate_e_confirm=20.0,
            gate_e_provisional=3.0,
            gate_w_net_min=2,
            gate_delta_min=0.02,
            gate_lambda_dir=-0.642,
            gate_e_rollback=10.0,
            gate_block=8,
            gate_n_max=40,
            gate_p_low=0.05,
            gate_p_high=0.95,
            gate_probe_quota=0.2,
            gate_gamma_decay=0.9,
            gate_cooldown_steps=2,
            gate_guard_err=0.10,
            skill_update_mode="patch",
            appendix_consolidate_threshold=6,
        )

        # 不应抛异常
        runner = Runner(
            config,
            llm=AsyncMock(),
            evolve_llm=AsyncMock(),
            vlm=AsyncMock(),
            telemetry=AsyncMock(),
        )
        assert runner is not None
```

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

Run: `conda run -n Video-Tree-TRM pytest tests/unit/test_harness_runner.py::TestRunnerFactoryInjection -v`
Expected: FAIL（Runner.__init__ 不接受 factory 参数 / 不做校验）

- [ ] **Step 3: 修改 Runner.__init__ 和 _make_* 方法**

在 `app/harness/runner.py` 中：

**3a.** 修改 `Runner.__init__`（第 454-469 行），增加 2 个可选参数 + fail-fast 校验：

```python
    def __init__(
        self,
        config: RunConfig,
        *,
        llm: LLMProvider,
        evolve_llm: LLMProvider,
        vlm: VLMProvider,
        telemetry: TelemetryRecorder,
        tool_dispatch_factory: Any | None = None,
        prompt_builder_factory: Any | None = None,
    ) -> None:
        # fail-fast 校验必须在 _ensure_workspace 之前，避免 workspace 报错掩盖 factory 缺失
        if config.mode in {"infer", "eval", "train"}:
            if tool_dispatch_factory is None or prompt_builder_factory is None:
                raise ValueError(
                    f"mode={config.mode!r} 需要 tool_dispatch_factory 和 "
                    f"prompt_builder_factory（不可为 None）"
                )
        self._config = config
        self._llm = llm
        self._evolve_llm = evolve_llm
        self._vlm = vlm
        self._telemetry = telemetry
        self._tool_dispatch_factory = tool_dispatch_factory
        self._prompt_builder_factory = prompt_builder_factory
        self._ensure_workspace()
        self._paths: ResolvedPaths = resolve_paths(config.workspace_dir)
```

**3b.** 修改 `_make_tool_dispatch_fn`（第 2064-2072 行）：

```python
    def _make_tool_dispatch_fn(self, *, skills_dir: Path | None = None):
        """构造工具调度函数。优先用注入的 factory，fallback 为显式报错。"""
        if self._tool_dispatch_factory is not None:
            return self._tool_dispatch_factory(skills_dir=skills_dir)

        async def _noop_dispatch(tool_name: str, args: dict, *, context: dict) -> str:
            raise NotImplementedError(
                f"工具 {tool_name} 调度未配置（需由 main.py 注入 tool_dispatch_fn）"
            )

        return _noop_dispatch
```

**3c.** 修改 `_make_prompt_builder`（第 2074-2082 行）：

```python
    def _make_prompt_builder(
        self, *, skills_dir: Path | None = None, prompts_dir: Path | None = None
    ):
        """构造 prompt 构建函数。优先用注入的 factory，fallback 为显式报错。"""
        if self._prompt_builder_factory is not None:
            return self._prompt_builder_factory(
                skills_dir=skills_dir, prompts_dir=prompts_dir
            )

        def _noop_builder(qa: GeneratedQuestion) -> tuple[str, str]:
            raise NotImplementedError("prompt_builder 未配置（需由 main.py 注入）")

        return _noop_builder
```

**3d.** 更新 docstring（第 440-452 行）增加两个新参数说明。

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

Run: `conda run -n Video-Tree-TRM pytest tests/unit/test_harness_runner.py -v`
Expected: ALL tests PASSED（含新增的 2 个 + 原有 34 个）

- [ ] **Step 5: Commit**

```bash
git add app/harness/runner.py tests/unit/test_harness_runner.py
git commit -m "feat(runner): 注入 tool_dispatch_factory/prompt_builder_factory + fail-fast"
```

---

### Task 5: app/harness/deps_router.py — per-video 路由器

**Files:**
- Create: `app/harness/deps_router.py`
- Test: `tests/unit/test_deps_router.py`

- [ ] **Step 1: 编写 Router 单元测试**

```python
# tests/unit/test_deps_router.py
"""InferenceDepsRouter 单元测试。"""

from __future__ import annotations

from pathlib import Path
from typing import Any
from unittest.mock import AsyncMock, MagicMock, patch

import pytest

from app.harness.deps_router import InferenceDepsRouter


@pytest.fixture()
def mock_deps():
    """构造 mock InferenceDeps。"""
    deps = MagicMock()
    deps.prompt_builder = lambda qa: (f"system_{qa.video_id}", f"user_{qa.question}")
    deps.tool_dispatch_fn = AsyncMock(return_value="tool_result")
    return deps


@pytest.fixture()
def router(mock_deps):
    """构造带 mock build_inference_deps 的 Router。"""
    r = InferenceDepsRouter(
        store_dir=Path("store"),
        embed_provider=MagicMock(),
        llm=MagicMock(),
        vlm=MagicMock(),
        ocr=None,
        default_prompts_dir=Path("store/prompts/v1"),
        default_skills_dir=Path("store/skills/v1"),
        skill_mode="auto",
        verify_vision=True,
        anchor=True,
        assemble_mode="default",
    )
    # 替换 _build_deps 为 mock
    r._build_deps = MagicMock(return_value=mock_deps)
    return r


class TestPromptBuilder:
    """prompt_builder 注册映射并返回 prompt。"""

    def test_registers_qid_to_vid_mapping(self, router, mock_deps) -> None:
        from core.types import GeneratedQuestion

        qa = GeneratedQuestion(
            question_id="q1", video_id="vid1", task_type="Action Reasoning",
            question="test?", options=("A. a", "B. b", "C. c", "D. d"),
            answer="A", source_nodes=(), difficulty="medium",
        )
        builder = router.create_prompt_builder()
        builder(qa)
        assert router._qid_to_vid["q1"] == "vid1"

    def test_returns_prompt_from_deps(self, router, mock_deps) -> None:
        from core.types import GeneratedQuestion

        qa = GeneratedQuestion(
            question_id="q1", video_id="vid1", task_type="Action Reasoning",
            question="test?", options=("A. a", "B. b", "C. c", "D. d"),
            answer="A", source_nodes=(), difficulty="medium",
        )
        builder = router.create_prompt_builder()
        system, user = builder(qa)
        assert "vid1" in system


class TestDispatch:
    """dispatch 通过 session_id 路由到正确视频。"""

    @pytest.mark.asyncio()
    async def test_routes_by_session_id(self, router, mock_deps) -> None:
        from core.types import GeneratedQuestion

        qa = GeneratedQuestion(
            question_id="q1", video_id="vid1", task_type="Action Reasoning",
            question="test?", options=("A. a", "B. b", "C. c", "D. d"),
            answer="A", source_nodes=(), difficulty="medium",
        )
        # 先注册映射
        builder = router.create_prompt_builder()
        builder(qa)

        dispatch = router.create_dispatch()
        result = await dispatch("view_node", {"node_id": "L1_000"}, context={"session_id": "q1"})
        assert result == "tool_result"
        mock_deps.tool_dispatch_fn.assert_called_once()

    @pytest.mark.asyncio()
    async def test_unknown_session_id_raises(self, router) -> None:
        dispatch = router.create_dispatch()
        with pytest.raises(KeyError, match="未注册"):
            await dispatch("view_node", {}, context={"session_id": "unknown"})

    @pytest.mark.asyncio()
    async def test_missing_session_id_raises(self, router) -> None:
        dispatch = router.create_dispatch()
        with pytest.raises(KeyError, match="未注册"):
            await dispatch("view_node", {}, context={})


class TestDepsCache:
    """同一 video_id 复用缓存。"""

    def test_same_video_reuses_deps(self, router, mock_deps) -> None:
        from core.types import GeneratedQuestion

        qa1 = GeneratedQuestion(
            question_id="q1", video_id="vid1", task_type="Action Reasoning",
            question="test1?", options=("A. a", "B. b", "C. c", "D. d"),
            answer="A",
        )
        qa2 = GeneratedQuestion(
            question_id="q2", video_id="vid1", task_type="Action Reasoning",
            question="test2?", options=("A. a", "B. b", "C. c", "D. d"),
            answer="B",
        )
        builder = router.create_prompt_builder()
        builder(qa1)
        builder(qa2)
        # 同一 video_id 只调用一次 _build_deps
        assert router._build_deps.call_count == 1


class TestClearCache:
    """clear_cache 清空缓存和映射。"""

    def test_clears_deps_and_mapping(self, router, mock_deps) -> None:
        from core.types import GeneratedQuestion

        qa = GeneratedQuestion(
            question_id="q1", video_id="vid1", task_type="Action Reasoning",
            question="test?", options=("A. a", "B. b", "C. c", "D. d"),
            answer="A", source_nodes=(), difficulty="medium",
        )
        builder = router.create_prompt_builder()
        builder(qa)
        assert len(router._qid_to_vid) == 1

        router.clear_cache()
        assert len(router._qid_to_vid) == 0
        assert len(router._deps_cache) == 0
```

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

Run: `conda run -n Video-Tree-TRM pytest tests/unit/test_deps_router.py -v`
Expected: ModuleNotFoundError（deps_router 尚未创建）

- [ ] **Step 3: 实现 InferenceDepsRouter**

```python
# app/harness/deps_router.py
"""按 video_id 懒加载 InferenceDeps 并路由 dispatch/prompt_builder。

将 Runner 的全局统一 dispatch/prompt_builder 接口路由到 per-video 的
InferenceDeps。时序保证：prompt_builder(qa) 先于 dispatch 被调用，
在 prompt_builder 中注册 question_id → video_id 映射。
"""

from __future__ import annotations

from typing import TYPE_CHECKING, Any

from loguru import logger

from app.harness.factory import InferenceDeps, build_inference_deps

if TYPE_CHECKING:
    from pathlib import Path

    from app.ports import EmbeddingProvider, OCRProvider
    from core.protocols import LLMProvider, VLMProvider
    from core.types import GeneratedQuestion


class InferenceDepsRouter:
    """按 video_id 懒加载 InferenceDeps 并路由工具调度和 prompt 构建。

    参数:
        store_dir: store 根目录。
        embed_provider: 嵌入端口。
        llm: LLM 端口。
        vlm: VLM 端口。
        ocr: OCR 端口（None 不启用）。
        default_prompts_dir: 默认 prompts 版本目录。
        default_skills_dir: 默认 skills 版本目录。
        skill_mode: skill 加载模式。
        verify_vision: observe_frame 是否验证。
        anchor: view_node 是否锚模式。
        assemble_mode: 锚模式装配形态。
    """

    def __init__(
        self,
        *,
        store_dir: Path,
        embed_provider: EmbeddingProvider,
        llm: LLMProvider,
        vlm: VLMProvider,
        ocr: OCRProvider | None,
        default_prompts_dir: Path,
        default_skills_dir: Path,
        skill_mode: str,
        verify_vision: bool,
        anchor: bool,
        assemble_mode: str,
    ) -> None:
        self._store_dir = store_dir
        self._embed = embed_provider
        self._llm = llm
        self._vlm = vlm
        self._ocr = ocr
        self._default_prompts_dir = default_prompts_dir
        self._default_skills_dir = default_skills_dir
        self._skill_mode = skill_mode
        self._verify_vision = verify_vision
        self._anchor = anchor
        self._assemble_mode = assemble_mode
        self._deps_cache: dict[tuple[str, str, str], InferenceDeps] = {}
        self._qid_to_vid: dict[str, str] = {}

    def create_dispatch(
        self, *, skills_dir: Path | None = None
    ) -> Any:
        """创建工具调度函数，通过 context["session_id"] 路由到 per-video dispatcher。

        参数:
            skills_dir: 可选的 skills 版本目录覆盖。

        返回:
            async (tool_name, args, *, context) -> str。
        """
        effective_skills = skills_dir or self._default_skills_dir

        async def _dispatch(
            tool_name: str, args: dict[str, Any], *, context: dict[str, Any]
        ) -> str:
            session_id = context.get("session_id")
            if not session_id or session_id not in self._qid_to_vid:
                raise KeyError(
                    f"未注册的 session_id={session_id!r}，"
                    f"已注册 {len(self._qid_to_vid)} 条映射"
                )
            video_id = self._qid_to_vid[session_id]
            deps = self._ensure_deps(video_id, effective_skills, self._default_prompts_dir)
            return await deps.tool_dispatch_fn(tool_name, args, context=context)

        return _dispatch

    def create_prompt_builder(
        self,
        *,
        skills_dir: Path | None = None,
        prompts_dir: Path | None = None,
    ) -> Any:
        """创建 prompt 构建函数，在调用时注册 question_id→video_id 映射。

        参数:
            skills_dir: 可选的 skills 版本目录覆盖。
            prompts_dir: 可选的 prompts 版本目录覆盖。

        返回:
            (GeneratedQuestion) -> (system_prompt, user_prompt)。
        """
        effective_skills = skills_dir or self._default_skills_dir
        effective_prompts = prompts_dir or self._default_prompts_dir

        def _builder(qa: GeneratedQuestion) -> tuple[str, str]:
            self._qid_to_vid[qa.question_id] = qa.video_id
            deps = self._ensure_deps(qa.video_id, effective_skills, effective_prompts)
            return deps.prompt_builder(qa)

        return _builder

    def _ensure_deps(
        self, video_id: str, skills_dir: Path, prompts_dir: Path
    ) -> InferenceDeps:
        """懒加载 per-video InferenceDeps，按 (video_id, skills_dir, prompts_dir) 缓存。"""
        key = (video_id, str(skills_dir), str(prompts_dir))
        if key not in self._deps_cache:
            self._deps_cache[key] = self._build_deps(video_id, skills_dir, prompts_dir)
            logger.debug("InferenceDeps 已缓存: video_id={}", video_id)
        return self._deps_cache[key]

    def _build_deps(
        self, video_id: str, skills_dir: Path, prompts_dir: Path
    ) -> InferenceDeps:
        """调用 factory.build_inference_deps 构建 per-video 依赖。"""
        return build_inference_deps(
            store_dir=self._store_dir,
            video_id=video_id,
            prompts_dir=prompts_dir,
            skills_dir=skills_dir,
            skill_mode=self._skill_mode,
            embed_provider=self._embed,
            llm=self._llm,
            vlm=self._vlm,
            ocr=self._ocr,
            verify_vision=self._verify_vision,
            anchor=self._anchor,
            assemble_mode=self._assemble_mode,
        )

    def clear_cache(self) -> None:
        """清空 deps 缓存和 question_id 映射。"""
        self._deps_cache.clear()
        self._qid_to_vid.clear()
```

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

Run: `conda run -n Video-Tree-TRM pytest tests/unit/test_deps_router.py -v`
Expected: 7 tests PASSED

- [ ] **Step 5: Commit**

```bash
git add app/harness/deps_router.py tests/unit/test_deps_router.py
git commit -m "feat(harness): InferenceDepsRouter per-video 路由器"
```

---

### Task 6: 配置变更

**Files:**
- Modify: `config/default.yaml:29,31`
- Modify: `.env` (LLM_CIRCUIT_BREAKER_THRESHOLD)

- [ ] **Step 1: 更新 default.yaml**

在 `config/default.yaml` 中修改：

```yaml
harness:
  concurrency: 24              # was 12
  max_steps: 40                # was 15
```

- [ ] **Step 2: 更新 .env 和 .env.example**

在 `.env` 中修改（本地生效，不提交）：
```
LLM_CIRCUIT_BREAKER_THRESHOLD=48
```

在 `.env.example` 中同步（提交到 Git）：
```
LLM_CIRCUIT_BREAKER_THRESHOLD=48  # 实际阈值 = max(此值, concurrency*2)
```

- [ ] **Step 3: 验证配置加载**

Run: `conda run -n Video-Tree-TRM python -c "from app.harness.config import load_config; from pathlib import Path; c = load_config(Path('config/default.yaml')); print(f'concurrency={c.concurrency}, max_steps={c.max_steps}')"`
Expected: `concurrency=24, max_steps=40`

- [ ] **Step 4: Commit**

```bash
git add config/default.yaml .env.example
git commit -m "config: concurrency=24, max_steps=40, breaker_threshold=48"
```

---

### Task 7: main.py — Composition Root

**Files:**
- Create: `main.py`

- [ ] **Step 1: 实现 main.py**

```python
"""CLI 入口 — Composition Root：构建适配器，注入 Runner，调度执行。

三层配置合并（YAML > .env > CLI）由 load_config 完成。
适配器参数通过 InfraSettings(BaseSettings) 从 .env 加载。
"""

from __future__ import annotations

import argparse
import asyncio
import os
from pathlib import Path
from typing import NamedTuple

from dotenv import load_dotenv
from loguru import logger
from pydantic_settings import BaseSettings, SettingsConfigDict


class InfraSettings(BaseSettings):
    """工程配置（少变/敏感），从 .env 加载。"""

    model_config = SettingsConfigDict(env_file=".env", extra="ignore")

    search_llm_model: str = ""
    search_llm_base_url: str = ""
    search_llm_api_key: str = ""

    vl_llm_model: str = ""
    vl_llm_base_url: str = ""
    vl_llm_api_key: str = ""

    evolve_llm_model: str = ""
    evolve_llm_base_url: str = ""
    evolve_llm_api_key: str = ""

    embed_api_key: str = ""
    embed_api_url: str = ""

    monkey_ocr_urls: str = ""
    redis_url: str = ""
    redis_cache_ttl: int = 86400

    llm_timeout: float = 120.0
    llm_max_retries: int = 3
    llm_retry_base_delay: float = 2.0
    llm_retry_max_delay: float = 30.0
    llm_circuit_breaker_threshold: int = 48
    llm_circuit_breaker_cooldown: float = 60.0
    llm_ttft_timeout: float = 30.0
    llm_inter_token_timeout: float = 15.0


class _Adapters(NamedTuple):
    """全套适配器实例。"""

    llm: object
    evolve_llm: object
    vlm: object
    telemetry: object
    embed: object
    ocr: object


def _build_adapters(settings: InfraSettings, embed_cfg: dict) -> _Adapters:
    """从 InfraSettings 构建全套适配器。

    参数:
        settings: 工程配置。
        embed_cfg: 嵌入配置字典（来自 YAML tree.embed 或 harness 段）。

    返回:
        _Adapters 命名元组。
    """
    from adapters.breaker import CircuitBreaker
    from adapters.embedding import LocalEmbeddingProvider, RemoteEmbeddingProvider
    from adapters.llm import GovernedLLMClient
    from adapters.telemetry import SQLiteTelemetryRecorder
    from adapters.vlm import GovernedVLMClient

    breaker = CircuitBreaker(
        fail_threshold=max(settings.llm_circuit_breaker_threshold, 1),
        cooldown_s=settings.llm_circuit_breaker_cooldown,
    )

    cache = None
    if settings.redis_url:
        try:
            from adapters.redis_cache import RedisResponseCache

            cache = RedisResponseCache(
                redis_url=settings.redis_url, ttl=settings.redis_cache_ttl
            )
        except Exception:
            logger.warning("Redis 缓存不可用，降级为无缓存模式")

    telemetry_db = Path("logs/telemetry.db")
    telemetry_db.parent.mkdir(parents=True, exist_ok=True)
    telemetry = SQLiteTelemetryRecorder(telemetry_db)

    def _make_llm(model: str, base_url: str, api_key: str, *, thinking: bool) -> GovernedLLMClient:
        return GovernedLLMClient(
            model=model,
            base_url=base_url,
            api_key=api_key,
            provider=model.split("-")[0] if model else "unknown",
            thinking=thinking,
            breaker=breaker,
            cache=cache,
            telemetry=telemetry,
            timeout_s=settings.llm_timeout,
            ttft_timeout_s=settings.llm_ttft_timeout,
            inter_token_timeout_s=settings.llm_inter_token_timeout,
            max_retries=settings.llm_max_retries,
            retry_base_delay_s=settings.llm_retry_base_delay,
            retry_max_delay_s=settings.llm_retry_max_delay,
        )

    llm = _make_llm(
        settings.search_llm_model,
        settings.search_llm_base_url,
        settings.search_llm_api_key,
        thinking=True,
    )

    evolve_llm = llm  # 设计要求本次传同一实例

    vl_llm = _make_llm(
        settings.vl_llm_model,
        settings.vl_llm_base_url,
        settings.vl_llm_api_key,
        thinking=False,
    )
    vlm = GovernedVLMClient(governed_llm=vl_llm)

    backend = embed_cfg.get("backend", "local")
    if backend == "local":
        embed = LocalEmbeddingProvider(
            model_name=embed_cfg.get("model_name", "BAAI/bge-base-zh-v1.5"),
            embed_dim=embed_cfg.get("embed_dim", 768),
            device=embed_cfg.get("device", "cpu"),
        )
    else:
        from adapters.embedding import RemoteEmbeddingProvider

        embed = RemoteEmbeddingProvider(
            model_name=embed_cfg.get("model_name", ""),
            embed_dim=embed_cfg.get("embed_dim", 768),
            api_key=settings.embed_api_key,
            api_url=settings.embed_api_url,
        )

    ocr = None
    if settings.monkey_ocr_urls:
        from adapters.ocr import MonkeyOCRClient

        urls = [u.strip() for u in settings.monkey_ocr_urls.split(",") if u.strip()]
        if urls:
            ocr = MonkeyOCRClient(urls=urls)

    return _Adapters(
        llm=llm,
        evolve_llm=evolve_llm,
        vlm=vlm,
        telemetry=telemetry,
        embed=embed,
        ocr=ocr,
    )


def _build_parser() -> argparse.ArgumentParser:
    """构建 CLI 参数解析器。所有参数 default=None，未传入时使用 YAML 默认值。"""
    parser = argparse.ArgumentParser(description="Video-Tree-TRM5 实验运行器")
    parser.add_argument(
        "--config", type=Path, default=Path("config/default.yaml"),
        help="YAML 配置文件路径",
    )
    parser.add_argument("--workspace-dir", type=Path, dest="workspace_dir")
    parser.add_argument("--store-dir", type=Path, dest="store_dir")
    parser.add_argument("--mode", choices=["infer", "train", "diagnose", "evolve", "eval", "promote"])
    parser.add_argument("--run-id", type=str, dest="run_id")
    parser.add_argument("--concurrency", type=int)
    parser.add_argument("--max-steps", type=int, dest="max_steps")
    parser.add_argument("--skill-mode", choices=["auto", "manual", "none"], dest="skill_mode")
    parser.add_argument("--n-samples", type=int, dest="n_samples")
    parser.add_argument("--questions", type=str)
    parser.add_argument("--skills-version", type=str, dest="skills_version")
    parser.add_argument("--prompts-version", type=str, dest="prompts_version")
    parser.add_argument("--task-types", nargs="+", dest="task_types")
    parser.add_argument("--resume", action="store_true", dest="resume")
    parser.add_argument("--fresh", action="store_true", dest="fresh")
    parser.add_argument("--seed", type=str, dest="seed")
    parser.add_argument("--epochs", type=int)
    return parser


def _log_result(result: object) -> None:
    """输出推理结果摘要。"""
    logger.info("=" * 60)
    logger.info("运行 ID: {}", result.run_id)
    logger.info("总体准确率: {:.2%} ({}/{})", result.accuracy, result.correct, result.total)
    logger.info("平均步数: {:.1f}", result.steps_mean)
    logger.info(
        "Token 用量: prompt={}, completion={}",
        result.token_usage["prompt_tokens"],
        result.token_usage["completion_tokens"],
    )
    if result.per_task_type:
        logger.info("--- 按任务类型 ---")
        for task_type, stats in sorted(result.per_task_type.items()):
            logger.info(
                "  {}: {:.2%} ({}/{})",
                task_type, stats["accuracy"], stats["correct"], stats["total"],
            )
    logger.info(
        "停止原因: {}",
        ", ".join(f"{k}={v}" for k, v in result.stop_reason_counts.items()),
    )
    logger.info("=" * 60)


def main() -> None:
    """入口函数。"""
    load_dotenv()

    parser = _build_parser()
    args = parser.parse_args()

    import yaml

    with open(args.config, encoding="utf-8") as f:
        raw_yaml = yaml.safe_load(f)

    from app.harness.config import load_config

    cli_overrides = {k: v for k, v in vars(args).items() if k != "config" and k != "task_types"}
    config = load_config(args.config, cli_overrides)

    logger.info("配置加载完成: mode={}, workspace={}", config.mode, config.workspace_dir)

    settings = InfraSettings()
    embed_cfg = raw_yaml.get("embed", {})
    adapters = _build_adapters(settings, embed_cfg)

    from app.harness.deps_router import InferenceDepsRouter
    from app.harness.runner import Runner
    from app.harness.workspace import resolve_paths

    router = InferenceDepsRouter(
        store_dir=Path(config.store_dir),
        embed_provider=adapters.embed,
        llm=adapters.llm,
        vlm=adapters.vlm,
        ocr=adapters.ocr,
        default_prompts_dir=Path(config.store_dir) / "prompts" / config.prompts_version,
        default_skills_dir=Path(config.store_dir) / "skills" / config.skills_version,
        skill_mode=config.skill_mode,
        verify_vision=True,
        anchor=True,
        assemble_mode="default",
    )

    runner = Runner(
        config,
        llm=adapters.llm,
        evolve_llm=adapters.evolve_llm,
        vlm=adapters.vlm,
        telemetry=adapters.telemetry,
        tool_dispatch_factory=router.create_dispatch,
        prompt_builder_factory=router.create_prompt_builder,
    )

    if config.mode == "infer":
        task_types = getattr(args, "task_types", None)
        result = asyncio.run(runner.infer(task_types=task_types))
        _log_result(result)
    else:
        raise SystemExit(f"模式 {config.mode!r} 尚未实现")


if __name__ == "__main__":
    main()
```

- [ ] **Step 2: 验证 main.py 可导入**

Run: `conda run -n Video-Tree-TRM python -c "import main; print('OK')"`
Expected: `OK`

- [ ] **Step 3: 验证 --help 输出**

Run: `conda run -n Video-Tree-TRM python main.py --help`
Expected: 显示参数帮助文本

- [ ] **Step 4: Commit**

```bash
git add main.py
git commit -m "feat: main.py Composition Root（仅 infer 模式）"
```

---

### Task 8: 冒烟测试

**Files:**
- No new files, end-to-end validation

- [ ] **Step 1: 验证 workspace 初始化**

需要先初始化 workspace（将 store 的 v1 资源拷贝到 workspace）。

Run: `conda run -n Video-Tree-TRM python main.py --mode infer --n-samples 1 --concurrency 1 --max-steps 3`

检查：
- 是否成功初始化 workspace
- 是否加载了题目
- 是否创建了 InferenceDeps
- LLM 调用是否经过 GovernedLLMClient

如果出现 workspace 不存在的错误，需要先手动初始化：

```python
from pathlib import Path
from app.harness.workspace import init_workspace
init_workspace(
    Path("workspaces/default"),
    Path("store"),
    "benchmarks/Video-MME",
    "v1",
    "v1",
)
```

- [ ] **Step 2: 检查结果输出**

Expected: 看到日志输出包含：
- `配置加载完成: mode=infer`
- `InferenceDeps 已缓存: video_id=...`
- `推理完成: accuracy=...`

- [ ] **Step 3: 运行全量测试确认无回归**

Run: `conda run -n Video-Tree-TRM pytest tests/unit/ -v --tb=short`
Expected: ALL PASSED

- [ ] **Step 4: 最终提交**

```bash
git add -A
git commit -m "test: 冒烟测试通过，900 题推理管线就绪"
```

---

## 核心算法保真校验

Task 0 涉及 Voronoi 字幕分配（算法清单中非独立项，属于建树模块的后处理步骤）。改动范围：

| 算法 | 改动性质 | 保真判定 |
|------|---------|---------|
| Voronoi 字幕分配 (`subtitle.py`) | 写入目标从 `L3Node.subtitle` 改为 `L3Card.subtitle` + 新增 L2 聚合 | 核心逻辑（中点计算、范围扩展）不变，仅写入目标变更。**保真**。 |
| L2 建树 (`video_builder.py`) | Card 构造后注入 subtitle | 新增逻辑（原来不存在 L2 subtitle），不涉及算法简化。**不适用**。 |

其余 Task 1-8 不涉及核心算法迁移。