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 = ""：

@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 = ""：

@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 字段：

@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：

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：

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 读取，兼容旧格式从节点级读取：

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：

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 均支持：

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 需单独标签/锚标处理）：

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：

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：

_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：

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 行：

# 原: l3.subtitle = subtitle_text if subtitle_text else None
# 改:
l3.card = dataclasses.replace(l3.card, subtitle=subtitle_text or "")

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

# 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 中注入字幕（避免双写）。仅更新注释：

# 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（若存在）：

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: 迁移脚本

# 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

验证：

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

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 文件

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 目录

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

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 共用，插入在 ## 搜索步骤 之前）：

## 视频树字段索引

| 层级 | 字段 | 适用场景 |
|------|------|---------|
| 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

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 结构测试

# 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 行之后）追加：

@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

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 末尾追加：

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 校验：

    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 行）：

    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 行）：

    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

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 单元测试

# 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

# 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

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 中修改：

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

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

"""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

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 不存在的错误，需要先手动初始化：

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: 最终提交

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 不涉及核心算法迁移。