Video-Tree-TRM5/research-wiki/plans/2026-07-09-question-gen-synth.md

# 赛题生成工具实现计划

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

**Goal:** 实现基于视频树的题目合成工具（generate + calibrate），含推理依赖 factory 提取。

**Architecture:** `app/question_gen/synthesizer.py` 承载核心业务逻辑（节点采样、prompt 构造、去重），`app/harness/factory.py` 提取推理依赖组装（TreeEnvironment + SearchToolDispatcher + PromptManager），`tools/generate_questions.py` 作为 CLI 壳编排并发和 I/O。

**Tech Stack:** Python 3.11, asyncio, GovernedVLMClient, EmbeddingProvider, scipy.stats.fisher_exact, loguru

**核心算法保真校验：** 本计划不涉及核心算法迁移（13 项均已在先前 PR 完成），保真校验不适用。

---

## 文件结构总览

| 动作 | 文件 | 职责 |
|------|------|------|
| 修改 | `app/harness/config.py:24` | 前置修复 `_VIDEO_MME_TASK_TYPE_COUNT` 11→12 |
| 新建 | `app/question_gen/synthesizer.py` | 出题核心逻辑 |
| 新建 | `app/harness/factory.py` | 推理依赖组装 |
| 新建 | `tools/generate_questions.py` | CLI 壳 |
| 新建 | `tests/unit/test_synthesizer.py` | synthesizer 单测 |
| 新建 | `tests/unit/test_factory.py` | factory 单测 |
| 新建 | `tests/unit/test_generate_questions.py` | CLI 集成测试 |
| 修改 | `app/question_gen/__init__.py` | 追加 synthesizer re-export |

---

### Task 0: 前置修复 _VIDEO_MME_TASK_TYPE_COUNT

**Files:**
- Modify: `app/harness/config.py:24`
- Test: `tests/unit/test_harness_config.py`

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

在 `tests/unit/test_harness_config.py` 中追加：

```python
def test_video_mme_task_type_count_is_12():
    """Video-MME 实际有 12 种题型，常量必须与之一致。"""
    from app.harness.config import _VIDEO_MME_TASK_TYPE_COUNT
    assert _VIDEO_MME_TASK_TYPE_COUNT == 12
```

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

```bash
conda activate Video-Tree-TRM & pytest tests/unit/test_harness_config.py::test_video_mme_task_type_count_is_12 -v
```

预期：FAIL，`assert 11 == 12`

- [ ] **Step 3: 修改常量**

`app/harness/config.py:24`：`_VIDEO_MME_TASK_TYPE_COUNT = 11` → `_VIDEO_MME_TASK_TYPE_COUNT = 12`

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

```bash
conda activate Video-Tree-TRM & pytest tests/unit/test_harness_config.py -v
```

预期：全部 PASS

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

```bash
git add app/harness/config.py tests/unit/test_harness_config.py
git commit -m "fix(config): _VIDEO_MME_TASK_TYPE_COUNT 11→12，Video-MME 实际有 12 种题型"
```

---

### Task 1: synthesizer.py — AnchorContext + 题型映射常量

**Files:**
- Create: `app/question_gen/synthesizer.py`
- Create: `tests/unit/test_synthesizer.py`

- [ ] **Step 1: 写失败测试 — 题型映射完整性**

```python
# tests/unit/test_synthesizer.py
from app.question_gen.synthesizer import TASK_TYPE_LEVEL_MAP, AnchorContext

ALL_12_TYPES = [
    "Object Recognition", "Attribute Perception", "OCR Problems",
    "Spatial Reasoning", "Spatial Perception",
    "Action Recognition", "Action Reasoning", "Counting Problem",
    "Temporal Perception",
    "Temporal Reasoning", "Information Synopsis",
    "Object Reasoning",
]

def test_task_type_level_map_covers_all_12_types():
    """映射表必须覆盖全部 12 种 Video-MME 题型。"""
    assert set(TASK_TYPE_LEVEL_MAP.keys()) == set(ALL_12_TYPES)

def test_anchor_context_frozen():
    """AnchorContext 是不可变的。"""
    ctx = AnchorContext(
        node_id="L3_001",
        card_text="test",
        frame_paths=["a.jpg"],
        subtitle="",
        distractor_texts=["other node"],
    )
    assert ctx.node_id == "L3_001"
```

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

```bash
conda activate Video-Tree-TRM & pytest tests/unit/test_synthesizer.py -v
```

预期：ImportError

- [ ] **Step 3: 实现 AnchorContext + TASK_TYPE_LEVEL_MAP**

创建 `app/question_gen/synthesizer.py`：

```python
"""赛题合成核心逻辑 — 节点采样、prompt 构造、去重。

纯函数为主，异步编排仅 generate_one。
通过 DI 接收 VLMProvider / EmbeddingProvider，不 import adapters/。
"""

from __future__ import annotations

from dataclasses import dataclass
from typing import Any


@dataclass(frozen=True)
class AnchorContext:
    """锚节点上下文——生成单道题所需的全部素材。

    属性:
        node_id: 锚节点 ID。
        card_text: 锚节点 card 序列化文本。
        frame_paths: 帧图片路径列表。
        subtitle: 对应字幕（可空）。
        distractor_texts: 同视频其他节点摘要（供 VLM 生成干扰项）。
    """

    node_id: str
    card_text: str
    frame_paths: list[str]
    subtitle: str
    distractor_texts: list[str]


@dataclass(frozen=True)
class TaskTypeSpec:
    """题型的生成规格。

    属性:
        level: 锚定层级（"L3" / "L2" / "L1" / "L1-L2"）。
        needs_frames: 是否必须提供帧图。
        frame_count: 帧数范围描述（如 "1", "2-3", "0-1"）。
        context_fields: 需要提取的 card 字段元组。
    """

    level: str
    needs_frames: bool
    frame_count: str
    context_fields: tuple[str, ...]


TASK_TYPE_LEVEL_MAP: dict[str, TaskTypeSpec] = {
    "Object Recognition": TaskTypeSpec("L3", True, "1", ("frame_summary",)),
    "Attribute Perception": TaskTypeSpec("L3", True, "1", ("frame_summary",)),
    "OCR Problems": TaskTypeSpec("L3", True, "1", ("frame_summary",)),
    "Spatial Reasoning": TaskTypeSpec("L3", True, "1", ("frame_summary", "spatial_layout")),
    "Spatial Perception": TaskTypeSpec("L3", True, "1", ("frame_summary",)),
    "Action Recognition": TaskTypeSpec("L2", True, "2-3", ("event_description",)),
    "Action Reasoning": TaskTypeSpec("L2", True, "2-3", ("event_description",)),
    "Counting Problem": TaskTypeSpec("L2", True, "2-3", ("event_description",)),
    "Temporal Perception": TaskTypeSpec("L2", False, "0-1", ("event_description", "time_range")),
    "Temporal Reasoning": TaskTypeSpec("L1", True, "per-L2", ("scene_summary",)),
    "Information Synopsis": TaskTypeSpec("L1", True, "per-L2", ("scene_summary",)),
    "Object Reasoning": TaskTypeSpec("L1-L2", True, "per-L2", ("event_description",)),
}
```

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

```bash
conda activate Video-Tree-TRM & pytest tests/unit/test_synthesizer.py -v
```

预期：PASS

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

```bash
git add app/question_gen/synthesizer.py tests/unit/test_synthesizer.py
git commit -m "feat(question_gen): AnchorContext + 12 题型-层级映射常量"
```

---

### Task 2: synthesizer.py — sample_anchor

**Files:**
- Modify: `app/question_gen/synthesizer.py`
- Modify: `tests/unit/test_synthesizer.py`

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

```python
import json
import random
from pathlib import Path

from app.tree.index import TreeIndex
from app.question_gen.synthesizer import sample_anchor


def _load_test_tree() -> tuple[TreeIndex, str]:
    """加载真实测试树（store/videos/ 下第一棵）。"""
    videos_dir = Path("store/videos")
    first_vid = sorted(videos_dir.iterdir())[0]
    tree = TreeIndex.load_json(str(first_vid / "tree.json"))
    return tree, first_vid.name


class TestSampleAnchor:
    def test_l3_type_returns_single_frame(self):
        tree, vid = _load_test_tree()
        ctx = sample_anchor(tree, "Object Recognition", set(), random.Random(42))
        assert len(ctx.frame_paths) == 1
        assert ctx.node_id.startswith("L")
        assert len(ctx.distractor_texts) > 0

    def test_l2_type_returns_multiple_frames(self):
        tree, vid = _load_test_tree()
        ctx = sample_anchor(tree, "Action Reasoning", set(), random.Random(42))
        assert 2 <= len(ctx.frame_paths) <= 3

    def test_temporal_perception_zero_or_one_frame(self):
        """Temporal Perception 帧数 0-1，且 card_text 含 time_range。"""
        tree, vid = _load_test_tree()
        ctx = sample_anchor(tree, "Temporal Perception", set(), random.Random(42))
        assert len(ctx.frame_paths) <= 1
        assert "time_range" in ctx.card_text.lower() or "time" in ctx.card_text.lower()

    def test_information_synopsis_uses_all_l2(self):
        """Information Synopsis 必须包含全部 L2 card（非采样子集）。"""
        tree, vid = _load_test_tree()
        ctx = sample_anchor(tree, "Information Synopsis", set(), random.Random(42))
        total_l2 = sum(len(r.children) for r in tree.roots)
        # card_text 中应包含全部 L2 的事件描述
        assert len(ctx.frame_paths) >= min(total_l2, 1)

    def test_l1_type_l2_nodes_in_time_order(self):
        """L1 题型的 L2 子节点应按时间顺序组织。"""
        tree, vid = _load_test_tree()
        ctx = sample_anchor(tree, "Temporal Reasoning", set(), random.Random(42))
        assert len(ctx.frame_paths) >= 1
        assert len(ctx.card_text) > 20

    def test_used_node_ids_excluded(self):
        tree, vid = _load_test_tree()
        rng = random.Random(42)
        ctx1 = sample_anchor(tree, "Object Recognition", set(), rng)
        ctx2 = sample_anchor(tree, "Object Recognition", {ctx1.node_id}, random.Random(43))
        assert ctx2.node_id != ctx1.node_id

    def test_insufficient_nodes_raises(self):
        tree, vid = _load_test_tree()
        all_l3_ids = set()
        for root in tree.roots:
            for l2 in root.children:
                for l3 in l2.children:
                    all_l3_ids.add(l3.id)
        import pytest
        with pytest.raises(ValueError, match="锚节点不足"):
            sample_anchor(tree, "Object Recognition", all_l3_ids, random.Random(42))
```

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

```bash
conda activate Video-Tree-TRM & pytest tests/unit/test_synthesizer.py::TestSampleAnchor -v
```

预期：ImportError（sample_anchor 不存在）

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

在 `app/question_gen/synthesizer.py` 中追加 `sample_anchor` 函数，核心逻辑：

1. 根据 `TASK_TYPE_LEVEL_MAP[task_type].level` 确定采样层级
2. L3 题型：从所有 L3 节点中随机选一个（排除 used_node_ids），取单帧 + card
3. L2 题型（Action Recognition / Action Reasoning / Counting Problem）：随机选一个 L2 节点，均匀采样 2-3 子帧，card 取 event_description
4. **Temporal Perception 特例**：随机选一个 L2 节点，取 0-1 帧（有子帧取 1 帧，无则 0），card_text 必须包含 event_description + time_range
5. L1 题型：取根节点 card（scene_summary）。**Information Synopsis 使用全部 L2 card，Temporal Reasoning 选 ≥3 个**。L2 子节点按 time_range 升序排列，每个 L2 取 1 张代表帧
6. L1-L2 题型：随机选 2-3 个 L2 节点（按 time_range 排序），每个取 1 张代表帧
7. distractor_texts：收集同树中**其他**同层级节点的摘要文本
8. 候选不足时 `raise ValueError("锚节点不足: ...")`

详细实现需参考 `app/tree/index.py` 中 L1Node/L2Node/L3Node 的字段结构（L3Card.frame_summary, L2Card.event_description, L1Card.scene_summary）和 frame_path 位置。

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

```bash
conda activate Video-Tree-TRM & pytest tests/unit/test_synthesizer.py::TestSampleAnchor -v
```

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

```bash
git add app/question_gen/synthesizer.py tests/unit/test_synthesizer.py
git commit -m "feat(question_gen): sample_anchor — 按题型层级采样锚节点"
```

---

### Task 3: synthesizer.py — build_generation_prompt + parse_vlm_response

**Files:**
- Modify: `app/question_gen/synthesizer.py`
- Modify: `tests/unit/test_synthesizer.py`

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

```python
from core.types import GeneratedQuestion
from app.question_gen.synthesizer import (
    build_generation_prompt,
    parse_vlm_response,
    AnchorContext,
)


class TestBuildGenerationPrompt:
    def test_messages_structure(self):
        anchor = AnchorContext(
            node_id="L3_001",
            card_text="A person typing on a laptop",
            frame_paths=["store/videos/test/frames/L1_000_L2_000_L3_000.jpg"],
            subtitle="Hello world",
            distractor_texts=["Another person walking in park"],
        )
        exemplars = [
            GeneratedQuestion(
                question_id="ex-1", video_id="v1", task_type="Object Recognition",
                question="What object?", options=("A. Cat", "B. Dog", "C. Bird", "D. Fish"),
                answer="A", source_nodes=(), difficulty="medium",
            ),
        ]
        messages, image_paths = build_generation_prompt(
            "Object Recognition", anchor, exemplars,
        )
        assert messages[0]["role"] == "system"
        assert "Object Recognition" in messages[0]["content"]
        assert any("What object?" in str(m) for m in messages)
        assert image_paths == anchor.frame_paths

    def test_distractor_in_user_message(self):
        anchor = AnchorContext(
            node_id="L2_003",
            card_text="Event card text",
            frame_paths=["a.jpg", "b.jpg"],
            subtitle="",
            distractor_texts=["Distractor node summary"],
        )
        messages, _ = build_generation_prompt("Action Reasoning", anchor, [])
        user_msg = [m for m in messages if m["role"] == "user"][0]
        assert "Distractor node summary" in user_msg["content"]


class TestParseVlmResponse:
    def test_valid_json(self):
        raw = '{"question": "What?", "options": ["A. X", "B. Y", "C. Z", "D. W"], "answer": "A"}'
        result = parse_vlm_response(raw, "vid1", "Object Recognition", 1)
        assert result["question"] == "What?"
        assert result["answer"] == "A"
        assert len(result["options"]) == 4

    def test_invalid_json_raises(self):
        import pytest
        with pytest.raises(ValueError, match="VLM 返回"):
            parse_vlm_response("not json", "vid1", "Object Recognition", 1)

    def test_missing_fields_raises(self):
        import pytest
        raw = '{"question": "What?"}'
        with pytest.raises(ValueError):
            parse_vlm_response(raw, "vid1", "Object Recognition", 1)

    def test_options_must_be_four(self):
        import pytest
        raw = '{"question": "Q?", "options": ["A. X", "B. Y"], "answer": "A"}'
        with pytest.raises(ValueError, match="4"):
            parse_vlm_response(raw, "vid1", "Object Recognition", 1)

    def test_answer_must_be_abcd(self):
        import pytest
        raw = '{"question": "Q?", "options": ["A. X", "B. Y", "C. Z", "D. W"], "answer": "E"}'
        with pytest.raises(ValueError, match="A.*D"):
            parse_vlm_response(raw, "vid1", "Object Recognition", 1)
```

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

```bash
conda activate Video-Tree-TRM & pytest tests/unit/test_synthesizer.py::TestBuildGenerationPrompt -v
conda activate Video-Tree-TRM & pytest tests/unit/test_synthesizer.py::TestParseVlmResponse -v
```

预期：ImportError

- [ ] **Step 3: 实现 build_generation_prompt + parse_vlm_response**

`build_generation_prompt(task_type, anchor, exemplars) -> (messages, image_paths)`：
- system message：角色设定 + 题型 + exemplar 示例 + 约束（基于节点内容、干扰项来自其他节点）
- user message：锚节点 card_text + subtitle + distractor_texts
- image_paths：直接取 anchor.frame_paths

`parse_vlm_response(raw, video_id, task_type, seq) -> dict`：
- 尝试 `json.loads(raw)`，失败时尝试从 markdown code block 提取 JSON
- 校验必需字段 question / options / answer 存在
- 返回 `{"question_id": f"gen-{video_id}-{seq:03d}", "question": ..., "options": [...], "answer": ...}`
- 缺字段或解析失败 → `raise ValueError("VLM 返回...")`

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

```bash
conda activate Video-Tree-TRM & pytest tests/unit/test_synthesizer.py -v
```

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

```bash
git add app/question_gen/synthesizer.py tests/unit/test_synthesizer.py
git commit -m "feat(question_gen): build_generation_prompt + parse_vlm_response"
```

---

### Task 4: synthesizer.py — is_duplicate + generate_one

**Files:**
- Modify: `app/question_gen/synthesizer.py`
- Modify: `tests/unit/test_synthesizer.py`

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

```python
import numpy as np
from unittest.mock import AsyncMock, MagicMock
from app.question_gen.synthesizer import is_duplicate, generate_one


class TestIsDuplicate:
    @staticmethod
    def _fake_embed(texts):
        """确定性 + L2 归一化的 fake embedding。"""
        if isinstance(texts, str):
            texts = [texts]
        vecs = []
        for t in texts:
            rs = np.random.RandomState(hash(t) % 2**31)
            v = rs.randn(4).astype(np.float32)
            v /= np.linalg.norm(v)
            vecs.append(v)
        return np.array(vecs, dtype=np.float32)

    def test_empty_pool_never_duplicate(self):
        pool = np.zeros((0, 4), dtype=np.float32)
        assert is_duplicate("anything", pool, self._fake_embed, 0.85) is False

    def test_identical_text_is_duplicate(self):
        text = "What is happening in the video?"
        emb = self._fake_embed(text)
        pool = emb.copy()
        assert is_duplicate(text, pool, self._fake_embed, 0.85) is True

    def test_different_text_not_duplicate(self):
        pool_texts = ["aaa", "bbb", "ccc", "ddd", "eee"]
        pool = self._fake_embed(pool_texts)
        assert is_duplicate("completely unique text xyz", pool, self._fake_embed, 0.99) is False


class TestGenerateOne:
    @staticmethod
    async def test_success_path():
        """mock VLM 返回合法 JSON，应成功生成。"""
        vlm = AsyncMock()
        vlm.chat_with_images.return_value = MagicMock(
            content='{"question":"Q?","options":["A. 1","B. 2","C. 3","D. 4"],"answer":"A"}'
        )
        embed_fn = lambda t: np.zeros((1, 4) if isinstance(t, str) else (len(t), 4), dtype=np.float32)

        tree, vid = _load_test_tree()
        result = await generate_one(
            vlm=vlm, embed_fn=embed_fn, tree=tree, video_id=vid,
            task_type="Object Recognition", seq=1,
            exemplars=[], pool_embeddings=np.zeros((0, 4), dtype=np.float32),
            used_node_ids=set(), max_retries=3, similarity_threshold=0.85,
            rng=random.Random(42), session_id="test",
        )
        assert result is not None
        assert result.question_id == f"gen-{vid}-001"
        assert result.task_type == "Object Recognition"

    @staticmethod
    async def test_all_retries_exhausted_returns_none():
        """VLM 始终返回无效 JSON，耗尽重试后返回 None。"""
        vlm = AsyncMock()
        vlm.chat_with_images.return_value = MagicMock(content="invalid")

        embed_fn = lambda t: np.zeros((1, 4), dtype=np.float32)
        tree, vid = _load_test_tree()

        result = await generate_one(
            vlm=vlm, embed_fn=embed_fn, tree=tree, video_id=vid,
            task_type="Object Recognition", seq=1,
            exemplars=[], pool_embeddings=np.zeros((0, 4), dtype=np.float32),
            used_node_ids=set(), max_retries=2, similarity_threshold=0.85,
            rng=random.Random(42), session_id="test",
        )
        assert result is None
```

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

```bash
conda activate Video-Tree-TRM & pytest tests/unit/test_synthesizer.py::TestIsDuplicate -v
conda activate Video-Tree-TRM & pytest tests/unit/test_synthesizer.py::TestGenerateOne -v
```

- [ ] **Step 3: 实现 is_duplicate + generate_one**

`is_duplicate(question_text, pool_embeddings, embed_fn, threshold) -> bool`：
- `embed_fn(question_text)` → `[1, D]`，squeeze 为 `[D]`
- `pool_embeddings @ query` 余弦相似度（pool 和 query 都已 L2 归一化）
- `max(similarities) >= threshold` → True

`generate_one(vlm, embed_fn, tree, video_id, task_type, seq, *, ...)` → `GeneratedQuestion | None`：
- 循环最多 `max_retries` 次：
  1. `sample_anchor(tree, task_type, used_node_ids, rng)` → anchor
  2. `build_generation_prompt(task_type, anchor, exemplars)` → messages, images
  3. `await vlm.chat_with_images(messages, images, session_id=session_id)` → response
  4. `parse_vlm_response(response.content, video_id, task_type, seq)` → parsed_dict（含四选一 schema 校验）
  5. 用 anchor.node_id 补齐 `source_nodes`，`difficulty="medium"`
  6. 构造并返回 `GeneratedQuestion`（**不在此处做去重**——去重在调用方的单线程汇总点原子执行）
- 全部重试失败（parse 异常）→ return None

**并发去重安全**：`generate_one` 只负责生成候选题。调用方（tools/ CLI）在收到候选后，在单线程汇总点（async for + await）原子执行：① `is_duplicate` 检查当前题型的 embedding 池 → ② 通过则添加 embedding + 写 JSON + 更新 progress → ③ 不通过则丢弃并重试。embedding 池按 `dict[str, np.ndarray]`（key=task_type）维护，确保只在同题型内去重。

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

```bash
conda activate Video-Tree-TRM & pytest tests/unit/test_synthesizer.py -v
```

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

```bash
git add app/question_gen/synthesizer.py tests/unit/test_synthesizer.py
git commit -m "feat(question_gen): is_duplicate + generate_one — 去重与单题生成编排"
```

---

### Task 5: factory.py — build_inference_deps

**Files:**
- Create: `app/harness/factory.py`
- Create: `tests/unit/test_factory.py`

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

```python
# tests/unit/test_factory.py
import random
from pathlib import Path
from unittest.mock import MagicMock, AsyncMock

import numpy as np
import pytest

from app.harness.factory import build_inference_deps, InferenceDeps


class TestBuildInferenceDeps:
    def test_returns_inference_deps(self, tmp_path):
        """用 fake adapters 验证返回类型和字段非 None。"""
        # 准备一棵最小树
        import json
        vid_dir = tmp_path / "videos" / "test_vid"
        vid_dir.mkdir(parents=True)
        frames_dir = vid_dir / "frames"
        frames_dir.mkdir()
        minimal_tree = {
            "metadata": {"source_path": "test", "modality": "video"},
            "roots": [{
                "id": "L1_000", "card": {
                    "scene_summary": "s", "main_setting": "s",
                    "key_entities": [], "main_actions": [],
                    "topic_keywords": [], "visible_text": [],
                    "temporal_flow": "s",
                }, "time_range": [0, 10], "children": [],
            }],
        }
        (vid_dir / "tree.json").write_text(json.dumps(minimal_tree))

        # 准备 prompts
        prompts_dir = tmp_path / "prompts"
        prompts_dir.mkdir()
        (prompts_dir / "system.md").write_text("You are a search agent.")

        fake_llm = AsyncMock()
        fake_vlm = AsyncMock()
        fake_embed = MagicMock()
        fake_embed.dim = 4
        fake_embed.embed = lambda t: np.zeros((1, 4), dtype=np.float32)

        deps = build_inference_deps(
            store_dir=tmp_path,
            video_id="test_vid",
            prompts_dir=prompts_dir,
            skills_dir=None,
            skill_mode="none",
            embed_provider=fake_embed,
            llm=fake_llm,
            vlm=fake_vlm,
            ocr=None,
            verify_vision=False,
            anchor=False,
            assemble_mode="ids",
        )
        assert isinstance(deps, InferenceDeps)
        assert deps.llm is fake_llm
        assert callable(deps.tool_dispatch_fn)
        assert callable(deps.prompt_builder)

        # 验证 prompt_builder 真正可调用（wiring 正确）
        from core.types import GeneratedQuestion
        fake_q = GeneratedQuestion(
            question_id="q1", video_id="test_vid", task_type="Object Recognition",
            question="What?", options=("A. X", "B. Y", "C. Z", "D. W"),
            answer="A", source_nodes=(), difficulty="medium",
        )
        system, user = deps.prompt_builder(fake_q)
        assert isinstance(system, str) and len(system) > 0
        assert isinstance(user, str) and "What?" in user

    def test_missing_tree_raises(self, tmp_path):
        prompts_dir = tmp_path / "prompts"
        prompts_dir.mkdir()
        (prompts_dir / "system.md").write_text("x")
        vid_dir = tmp_path / "videos" / "nonexist"
        vid_dir.mkdir(parents=True)

        with pytest.raises(FileNotFoundError):
            build_inference_deps(
                store_dir=tmp_path, video_id="nonexist",
                prompts_dir=prompts_dir, skills_dir=None, skill_mode="none",
                embed_provider=MagicMock(), llm=AsyncMock(), vlm=AsyncMock(),
                ocr=None, verify_vision=False, anchor=False, assemble_mode="ids",
            )
```

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

```bash
conda activate Video-Tree-TRM & pytest tests/unit/test_factory.py -v
```

预期：ImportError

- [ ] **Step 3: 实现 factory.py**

创建 `app/harness/factory.py`：

```python
"""推理依赖组装 — 给定 store + config 构建可工作的推理依赖集。

factory 只做组装，不持有状态。adapter 实例由调用方创建并传入。
消费者：tools/generate_questions.py（校准）、未来 main.py、Runner。
"""

from __future__ import annotations

from dataclasses import dataclass
from functools import partial
from pathlib import Path
from typing import TYPE_CHECKING, Any

from app.search.prompt import PromptManager
from app.search.skills import SkillRegistry, discover_skills
from app.search.tools import SearchToolDispatcher
from app.tree.environment import TreeEnvironment
from app.tree.index import TreeIndex

if TYPE_CHECKING:
    from collections.abc import Callable

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


@dataclass(frozen=True)
class InferenceDeps:
    """跑一次推理所需的全套依赖（不含 HarnessLog）。

    属性:
        llm: LLM 端口实例。
        tool_dispatch_fn: SearchToolDispatcher.dispatch 的绑定方法。
        prompt_builder: (GeneratedQuestion) -> (system_prompt, user_prompt)。
    """

    llm: LLMProvider
    tool_dispatch_fn: Callable[..., Any]
    prompt_builder: Callable[[GeneratedQuestion], tuple[str, str]]


def build_inference_deps(
    *,
    store_dir: Path,
    video_id: str,
    prompts_dir: Path,
    skills_dir: Path | None,
    skill_mode: str,
    embed_provider: EmbeddingProvider,
    llm: LLMProvider,
    vlm: VLMProvider,
    ocr: OCRProvider | None,
    verify_vision: bool,
    anchor: bool,
    assemble_mode: str,
) -> InferenceDeps:
    """组装单个视频的推理依赖。

    参数:
        store_dir: store 根目录（含 videos/{video_id}/tree.json）。
        video_id: 视频标识。
        prompts_dir: prompt 文件目录（含 system.md）。
        skills_dir: skill 文件目录（None 不启用）。
        skill_mode: "auto" / "manual" / "none"。
        embed_provider: 文本嵌入端口。
        llm: LLM 端口。
        vlm: VLM 端口。
        ocr: OCR 端口（None 不启用）。
        verify_vision: observe_frame 是否执行验证轮。
        anchor: view_node 是否启用行号锚模式。
        assemble_mode: 锚模式装配形态。

    返回:
        InferenceDeps 实例。

    异常:
        FileNotFoundError: tree.json 不存在。
    """
    # Phase 1: 加载树
    tree_path = store_dir / "videos" / video_id / "tree.json"
    if not tree_path.exists():
        raise FileNotFoundError(f"树文件不存在: {tree_path}")
    tree = TreeIndex.load_json(str(tree_path))

    frames_dir = store_dir / "videos" / video_id / "frames"
    env = TreeEnvironment(tree, frames_dir if frames_dir.exists() else None)

    # Phase 2: Skills
    skills: SkillRegistry | None = None
    always_skills_text = ""
    task_skill_map: dict[str, str] = {}
    catalog_text = ""
    if skills_dir and skills_dir.exists():
        always_skills_text, task_skill_map, catalog_text, skills = discover_skills(skills_dir)

    # Phase 3: SearchToolDispatcher
    dispatcher = SearchToolDispatcher(
        env=env,
        tool_llm=llm,
        vlm=vlm,
        ocr=ocr,
        prompts_dir=prompts_dir,
        skills=skills,
        embed_fn=embed_provider.embed,
        verify_vision=verify_vision,
        anchor=anchor,
        assemble_mode=assemble_mode,
    )

    # Phase 4: PromptManager → prompt_builder 偏函数
    pm = PromptManager(prompts_dir)
    l1_ids = [r.id for r in tree.roots]

    def _prompt_builder(
        qa: GeneratedQuestion,
        _pm: PromptManager = pm,
        _skill_mode: str = skill_mode,
        _always: str = always_skills_text,
        _tsm: dict = task_skill_map,
        _cat: str = catalog_text,
        _l1_ids: list = l1_ids,
    ) -> tuple[str, str]:
        system = _pm.build_inference_prompt(
            _skill_mode, qa.task_type, _always, _tsm, _cat,
        )
        user = _pm.format_user_prompt(
            qa.question, list(qa.options), _l1_ids, qa.task_type,
        )
        return system, user

    return InferenceDeps(
        llm=llm,
        tool_dispatch_fn=dispatcher.dispatch,
        prompt_builder=_prompt_builder,
    )
```

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

```bash
conda activate Video-Tree-TRM & pytest tests/unit/test_factory.py -v
```

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

```bash
git add app/harness/factory.py tests/unit/test_factory.py
git commit -m "feat(harness): factory.py — 推理依赖组装，可复用于 calibrate + main.py"
```

---

### Task 6: tools/generate_questions.py — generate 子命令

**Files:**
- Create: `tools/generate_questions.py`
- Modify: `tests/unit/test_generate_questions.py`（新建）

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

```python
# tests/unit/test_generate_questions.py
import json
import sys
from pathlib import Path
from unittest.mock import AsyncMock, MagicMock, patch

import numpy as np
import pytest


class TestGenerateSmoke:
    def test_generate_writes_json(self, tmp_path):
        """mock VLM + 1 棵真实树 + per_type=1，验证 JSON 输出格式。"""
        import shutil
        from unittest.mock import AsyncMock, MagicMock, patch

        # 复制一棵真实树到 tmp
        src = Path("store/videos") / sorted(Path("store/videos").iterdir())[0].name
        dst = tmp_path / "videos" / src.name
        shutil.copytree(src, dst)

        # 准备 benchmark（至少 1 道题做 exemplar）
        bench_dir = tmp_path / "questions" / "benchmarks"
        bench_dir.mkdir(parents=True)
        bench_file = bench_dir / f"{src.name}.json"
        bench_file.write_text(json.dumps([{
            "question_id": "ex-1", "task_type": "Object Recognition",
            "question": "What?", "options": ["A. X", "B. Y", "C. Z", "D. W"],
            "answer": "A",
        }]))

        output_dir = tmp_path / "output"
        output_dir.mkdir()

        # import CLI 模块的内部函数
        sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
        from tools.generate_questions import _load_or_init_progress, _select_exemplars

        # 验证 progress 初始化
        progress = _load_or_init_progress(output_dir)
        assert progress["completed"] == {}

        # 验证 exemplar 选择
        from app.question_gen.loader import load_benchmark
        bench_qs = load_benchmark(bench_dir)
        exemplars = _select_exemplars(bench_qs, "Object Recognition", 3, random.Random(42))
        assert len(exemplars) >= 1
        assert all(e.task_type == "Object Recognition" for e in exemplars)


class TestProgressResume:
    def test_skips_completed_and_rebuilds_pool(self, tmp_path):
        """progress.json 中已完成的题应被跳过，embedding 池应从已有 JSON 恢复。"""
        output_dir = tmp_path / "output"
        output_dir.mkdir()

        # 写一个已完成的 JSON
        (output_dir / "test_vid.json").write_text(json.dumps([{
            "question_id": "gen-test_vid-001", "task_type": "Object Recognition",
            "question": "Existing question?",
            "options": ["A. X", "B. Y", "C. Z", "D. W"], "answer": "A",
            "source_nodes": ["L3_001"], "difficulty": "medium",
        }]))

        progress = {
            "completed": {"Object Recognition": ["gen-test_vid-001"]},
            "output_dir": str(output_dir),
        }
        (output_dir / "progress.json").write_text(json.dumps(progress))

        from tools.generate_questions import _load_or_init_progress
        loaded = _load_or_init_progress(output_dir)
        assert "gen-test_vid-001" in loaded["completed"]["Object Recognition"]
```

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

```bash
conda activate Video-Tree-TRM & pytest tests/unit/test_generate_questions.py -v
```

- [ ] **Step 3: 实现 tools/generate_questions.py — generate 子命令**

创建 `tools/generate_questions.py`，核心结构：

```python
#!/usr/bin/env python3
"""赛题生成工具：generate + calibrate。

用法:
    conda activate Video-Tree-TRM
    python tools/generate_questions.py generate --store-dir store ...
    python tools/generate_questions.py calibrate --generated-dir ... --benchmark-dir ...

app/core/adapters 不 import 此脚本。
"""

from __future__ import annotations

import argparse
import asyncio
import json
import sys
from pathlib import Path

PROJECT_ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(PROJECT_ROOT))

from dotenv import load_dotenv
from loguru import logger

load_dotenv(PROJECT_ROOT / ".env")

# generate 子命令：
# 1. 加载 video_id 列表 + benchmark exemplars + 初始化 embedding 池
# 2. 断点续跑：读 progress.json + 恢复已生成题 embedding + used_node_ids
# 3. 实例化 GovernedVLMClient + EmbeddingProvider（从 .env 读配置）
# 4. 对 12 题型 × per_type，asyncio.Semaphore 并发调 generate_one
# 5. 单线程汇总：检查去重 → 加入 pool → 写 JSON → 更新 progress
# 6. 全部完成删除 progress.json
```

实现要点：
- `_load_or_init_progress(output_dir)` / `_save_progress(output_dir, progress)` 断点续跑
- `_rebuild_embedding_pool(output_dir, embed_fn, benchmark_questions)` 续跑时恢复 embedding
- `_build_vlm_client()` / `_build_embed_provider()` 从 .env 实例化 adapters
- `_select_exemplars(benchmark, task_type, n, rng)` 跨视频采样 few-shot
- `async def _run_generate(args)` 主流程
- 并发模型：Semaphore 限流 VLM 调用，但去重+写入在主协程中顺序执行

- [ ] **Step 4: 运行测试 + lint**

```bash
conda activate Video-Tree-TRM & pytest tests/unit/test_generate_questions.py -v
conda activate Video-Tree-TRM & ruff check tools/generate_questions.py --fix
```

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

```bash
git add tools/generate_questions.py tests/unit/test_generate_questions.py
git commit -m "feat(tools): generate_questions.py generate 子命令 — VLM 出题 + 去重 + 断点续跑"
```

---

### Task 7: tools/generate_questions.py — calibrate 子命令

**Files:**
- Modify: `tools/generate_questions.py`
- Modify: `tests/unit/test_generate_questions.py`

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

```python
from scipy.stats import fisher_exact


class TestCalibrateJudgment:
    def test_pass_when_delta_small(self):
        """差异小于 tolerance → PASS。"""
        from tools.generate_questions import _judge_task_type
        verdict = _judge_task_type(
            bench_correct=60, bench_total=100,
            gen_correct=12, gen_total=20,
            tolerance=0.10, alpha=0.05,
        )
        assert verdict == "PASS"

    def test_fail_when_delta_large_and_significant(self):
        """差异大且统计显著 → FAIL。"""
        from tools.generate_questions import _judge_task_type
        verdict = _judge_task_type(
            bench_correct=144, bench_total=240,
            gen_correct=6, gen_total=20,
            tolerance=0.10, alpha=0.05,
        )
        assert verdict == "FAIL"

    def test_warn_when_delta_large_but_not_significant(self):
        """差异大但样本不足（p > alpha）→ WARN。"""
        from tools.generate_questions import _judge_task_type
        verdict = _judge_task_type(
            bench_correct=2, bench_total=3,
            gen_correct=8, gen_total=20,
            tolerance=0.10, alpha=0.05,
        )
        assert verdict == "WARN"


class TestCalibrateIntegration:
    def test_baseline_params_must_be_paired(self):
        """--baseline-db 和 --baseline-run-id 必须成对出现。"""
        from tools.generate_questions import _validate_calibrate_args
        import pytest
        with pytest.raises(ValueError, match="成对"):
            _validate_calibrate_args(baseline_db="some.db", baseline_run_id=None)

    def test_has_fail_returns_exit_code_1(self):
        """存在 FAIL 判定时，_calibrate_exit_code 返回 1。"""
        from tools.generate_questions import _calibrate_exit_code
        verdicts = {"Object Recognition": "PASS", "Action Reasoning": "FAIL"}
        assert _calibrate_exit_code(verdicts) == 1

    def test_all_pass_or_warn_returns_exit_code_0(self):
        from tools.generate_questions import _calibrate_exit_code
        verdicts = {"Object Recognition": "PASS", "Spatial Perception": "WARN"}
        assert _calibrate_exit_code(verdicts) == 0
```

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

```bash
conda activate Video-Tree-TRM & pytest tests/unit/test_generate_questions.py::TestCalibrateJudgment -v
```

- [ ] **Step 3: 实现 calibrate 子命令**

在 `tools/generate_questions.py` 中追加：

`_judge_task_type(bench_correct, bench_total, gen_correct, gen_total, tolerance, alpha) -> str`：
- `delta = abs(gen_correct/gen_total - bench_correct/bench_total)`
- `delta <= tolerance` → "PASS"
- Fisher exact test p-value：`table = [[bench_correct, bench_total-bench_correct], [gen_correct, gen_total-gen_correct]]`
- `p < alpha and delta > tolerance` → "FAIL"
- else → "WARN"

`async def _run_calibrate(args)` 主流程：
1. `load_benchmark` 加载两组题
2. benchmark 基线：有 `--baseline-db` 则从 DB 读，否则按 video_id 分组 → `build_inference_deps` → `run_inference`
3. 生成题同理按 video_id 分组 → 分组推理
4. 汇总 per_task_type accuracy → `_judge_task_type` 逐题型判定
5. 输出对比表
6. 有 FAIL → `sys.exit(1)`

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

```bash
conda activate Video-Tree-TRM & pytest tests/unit/test_generate_questions.py -v
```

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

```bash
git add tools/generate_questions.py tests/unit/test_generate_questions.py
git commit -m "feat(tools): generate_questions.py calibrate 子命令 — Fisher exact test 校准"
```

---

### Task 8: __init__.py 更新 + lint + 全量测试

**Files:**
- Modify: `app/question_gen/__init__.py`

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

```python
# 在 tests/unit/test_synthesizer.py 中追加
def test_public_api_importable():
    """synthesizer 的公共 API 必须可从 app.question_gen 直接 import。"""
    from app.question_gen import generate_one, AnchorContext, TASK_TYPE_LEVEL_MAP, sample_anchor
    assert callable(generate_one)
    assert callable(sample_anchor)
```

运行确认失败（当前 __init__.py 不 export 这些）：
```bash
conda activate Video-Tree-TRM & pytest tests/unit/test_synthesizer.py::test_public_api_importable -v
```

- [ ] **Step 2: 更新 __init__.py re-export**

```python
"""出题模块 — benchmark 加载、分层采样与赛题合成。"""

from app.question_gen.loader import load_benchmark, stratified_sample
from app.question_gen.synthesizer import (
    TASK_TYPE_LEVEL_MAP,
    AnchorContext,
    generate_one,
    sample_anchor,
)

__all__ = [
    "load_benchmark",
    "stratified_sample",
    "TASK_TYPE_LEVEL_MAP",
    "AnchorContext",
    "generate_one",
    "sample_anchor",
]
```

- [ ] **Step 2: 全量 lint**

```bash
conda activate Video-Tree-TRM & ruff check app/question_gen/ app/harness/factory.py tools/generate_questions.py --fix
conda activate Video-Tree-TRM & ruff format app/question_gen/ app/harness/factory.py tools/generate_questions.py
```

- [ ] **Step 3: 全量测试**

```bash
conda activate Video-Tree-TRM & pytest tests/ -v --tb=short
```

预期：全部 PASS

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

```bash
git add app/question_gen/__init__.py
git commit -m "refactor(question_gen): __init__.py 追加 synthesizer re-export"
```

---

## Self-Review 核对

**范围说明**：设计 §4.4 要求 `Runner._make_tool_dispatch_fn` / `_make_prompt_builder` 委托 factory，本计划不包含该改造——Runner 改造随 `main.py` 一起实施更合理。factory.py 已就绪可复用。

| 设计文档章节 | 对应 Task |
|-------------|-----------|
| §2 模块结构 | Task 1-5 (synthesizer) + Task 5 (factory) + Task 6-7 (CLI) |
| §3.1 题型映射 | Task 1 |
| §3.2 AnchorContext | Task 1 |
| §3.3 函数签名 | Task 2 (sample_anchor) + Task 3 (prompt/parse) + Task 4 (dedup/generate) |
| §3.4 exemplar 选择 | Task 6 (_select_exemplars) |
| §3.6 去重 + 并发安全 | Task 4 (is_duplicate) + Task 6 (单线程汇总) |
| §4 factory.py | Task 5 |
| §5 CLI 设计 | Task 6 (generate) + Task 7 (calibrate) |
| §6 Fisher 校准 | Task 7 (_judge_task_type) |
| §7 断点续跑 | Task 6 (progress + embedding 恢复) |
| §8 输出格式 | Task 6 (JSON 写入) |
| §9 前置修复 | Task 0 |
| §10 测试策略 | Task 1-7 各含测试 |