feat(question_gen): add v2 pipeline with retry loop and heavy check

- PipelineConfig: YAML-driven configuration with family_ratios, retry,
  concurrency, dedup threshold, and heavy sampling rate
- _assign_slots: deterministic round-robin slot assignment across videos
  with per-family weighted random selection
- _process_one_slot: full retry loop (generate → postprocess → gates →
  dedup) with reject-reason feedback to VLM on retry
- _heavy_check_one: blind LLM agent trial-answer for difficulty_steps
- run_pipeline_v2: orchestration with semaphore-bounded concurrency,
  progress/resume support, and store integration
- is_duplicate: cosine similarity dedup against embedding pool

Tests: 11 integration tests covering slot assignment, retry behavior,
max-retries exhaustion, full pipeline flow, progress resume, heavy
sampling, and store record completeness.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

app/question_gen/pipeline_v2.py

@@ -0,0 +1,735 @@
+"""v2 出题管线编排 — 重出循环 + 重量抽检 + 并发控制。
+
+完整编排流程:
+1. 基于视频列表和任务类型分配 slot
+2. 每个 slot 执行 generate → postprocess → gates → 重出循环
+3. 去重检测（embedding cosine similarity）
+4. 接受题目的随机子集做重量抽检（盲 Agent 试答）
+5. 记录全部结果到 QuestionGenStore
+
+典型调用::
+
+    result = await run_pipeline_v2(
+        video_ids=["v1", "v2"],
+        trees={"v1": tree1, "v2": tree2},
+        vlm=vlm_client,
+        llm=llm_client,
+        embed_fn=embed_fn,
+        store=store,
+        config=config,
+        task_types=TASK_TYPES,
+    )
+"""
+
+from __future__ import annotations
+
+import asyncio
+import random
+import uuid
+from dataclasses import dataclass
+from pathlib import Path
+from typing import TYPE_CHECKING
+
+import numpy as np
+import yaml
+from loguru import logger
+
+from app.question_gen.families import QuestionFamilySpec, get_family_for_slot
+from app.question_gen.gates import GateReport, GateResult, GateVerdict, run_gates
+from app.question_gen.generator_v2 import CandidateQuestion, generate_one_v2
+from app.question_gen.postprocess import run_postprocess
+from app.question_gen.sampler_v2 import sample_material_v2
+from core.types import GeneratedQuestion
+
+if TYPE_CHECKING:
+    from collections.abc import Callable
+
+    from app.question_gen.run_store import QuestionGenStore
+    from app.tree.index import TreeIndex
+    from core.protocols import LLMProvider, VLMProvider
+
+
+# ---------------------------------------------------------------------------
+# 数据类型
+# ---------------------------------------------------------------------------
+
+
+@dataclass(frozen=True)
+class SlotAssignment:
+    """出题 slot 分配 — 一个 slot 对应一道待生成的题目。
+
+    属性:
+        slot_id: slot 唯一标识。
+        video_id: 分配到的视频 ID。
+        task_type: 任务类型。
+        family: 分配的问题家族规格。
+        seq: slot 序号（同 task_type 内从 1 开始）。
+    """
+
+    slot_id: str
+    video_id: str
+    task_type: str
+    family: QuestionFamilySpec
+    seq: int
+
+
+@dataclass
+class PipelineResult:
+    """管线运行结果。
+
+    属性:
+        accepted: 最终通过的题目列表。
+        rejected_count: 被拒绝（含重出耗尽）的题目数。
+        heavy_sampled: 重量抽检结果列表 — (question_id, difficulty_steps)。
+    """
+
+    accepted: list[GeneratedQuestion]
+    rejected_count: int
+    heavy_sampled: list[tuple[str, int]]
+
+
+@dataclass(frozen=True)
+class PipelineConfig:
+    """管线配置。
+
+    属性:
+        family_ratios: 家族名到权重的映射。
+        per_type: 每种 task_type 生成的题目数。
+        retry_limit: 单 slot 最大重出次数。
+        heavy_sample_rate: 重量抽检采样比例 [0.0, 1.0]。
+        dedup_threshold: 去重余弦相似度阈值。
+        concurrency: 并发 slot 数上限。
+        seed: 随机种子。
+        output_dir: 输出目录。
+        gate_models: 门控模型配置字典。
+        heavy_agent_model: 重量抽检使用的模型名。
+    """
+
+    family_ratios: dict[str, float]
+    per_type: int
+    retry_limit: int
+    heavy_sample_rate: float
+    dedup_threshold: float
+    concurrency: int
+    seed: int
+    output_dir: Path
+    gate_models: dict[str, str]
+    heavy_agent_model: str
+
+
+# ---------------------------------------------------------------------------
+# 配置加载
+# ---------------------------------------------------------------------------
+
+
+def load_pipeline_config(yaml_path: Path) -> PipelineConfig:
+    """从 YAML 文件加载管线配置。
+
+    读取 question_gen_v2 区段，映射为 PipelineConfig dataclass。
+
+    参数:
+        yaml_path: YAML 配置文件路径。
+
+    返回:
+        PipelineConfig 实例。
+
+    异常:
+        FileNotFoundError: 文件不存在。
+        KeyError: 缺少 question_gen_v2 区段。
+    """
+    if not yaml_path.exists():
+        msg = f"配置文件不存在: {yaml_path}"
+        raise FileNotFoundError(msg)
+
+    with yaml_path.open(encoding="utf-8") as f:
+        raw = yaml.safe_load(f)
+
+    if "question_gen_v2" not in raw:
+        msg = f"配置文件缺少 'question_gen_v2' 区段: {yaml_path}"
+        raise KeyError(msg)
+
+    section = raw["question_gen_v2"]
+
+    # 家族名统一为大写
+    raw_ratios = section["family_ratios"]
+    family_ratios = {k.upper(): float(v) for k, v in raw_ratios.items()}
+
+    return PipelineConfig(
+        family_ratios=family_ratios,
+        per_type=int(section["per_type"]),
+        retry_limit=int(section["retry_limit"]),
+        heavy_sample_rate=float(section["heavy_sample_rate"]),
+        dedup_threshold=float(section["dedup_threshold"]),
+        concurrency=int(section["concurrency"]),
+        seed=int(section["seed"]),
+        output_dir=Path(section["output_dir"]),
+        gate_models=section["gate"],
+        heavy_agent_model=str(section["heavy_agent_model"]),
+    )
+
+
+# ---------------------------------------------------------------------------
+# Slot 分配
+# ---------------------------------------------------------------------------
+
+
+def _assign_slots(
+    video_ids: list[str],
+    task_types: list[str],
+    per_type: int,
+    family_ratios: dict[str, float],
+    rng: random.Random,
+) -> list[SlotAssignment]:
+    """将出题目标分配为具体 slot 列表。
+
+    总 slot 数 = len(task_types) * per_type。
+    在视频间 round-robin 分配，每个 slot 通过 get_family_for_slot 决定家族。
+
+    参数:
+        video_ids: 视频 ID 列表。
+        task_types: 任务类型列表。
+        per_type: 每种 task_type 的目标题数。
+        family_ratios: 家族权重映射。
+        rng: 可控随机数生成器。
+
+    返回:
+        SlotAssignment 列表。
+    """
+    slots: list[SlotAssignment] = []
+    global_seq = 0
+
+    for task_type in task_types:
+        for i in range(per_type):
+            video_id = video_ids[i % len(video_ids)]
+            family = get_family_for_slot(task_type, family_ratios, rng)
+            global_seq += 1
+            slot_id = f"{task_type}_{global_seq:04d}"
+            slots.append(
+                SlotAssignment(
+                    slot_id=slot_id,
+                    video_id=video_id,
+                    task_type=task_type,
+                    family=family,
+                    seq=global_seq,
+                )
+            )
+
+    return slots
+
+
+# ---------------------------------------------------------------------------
+# 去重检测
+# ---------------------------------------------------------------------------
+
+
+def _is_duplicate(
+    question_text: str,
+    embed_pool: list[np.ndarray],
+    embed_fn: Callable[[str], np.ndarray],
+    threshold: float,
+) -> bool:
+    """检测题目文本是否与已接受题库重复。
+
+    计算题目 embedding 与 embed_pool 中所有向量的余弦相似度，
+    若最大相似度超过阈值则视为重复。
+
+    参数:
+        question_text: 待检测的题目文本。
+        embed_pool: 已接受题目的 embedding 向量列表。
+        embed_fn: 文本到向量的映射函数。
+        threshold: 余弦相似度阈值。
+
+    返回:
+        True 表示重复。
+    """
+    if not embed_pool:
+        return False
+
+    query_vec = embed_fn(question_text)
+    query_norm = np.linalg.norm(query_vec)
+    if query_norm == 0:
+        return False
+
+    for pool_vec in embed_pool:
+        pool_norm = np.linalg.norm(pool_vec)
+        if pool_norm == 0:
+            continue
+        similarity = float(np.dot(query_vec, pool_vec) / (query_norm * pool_norm))
+        if similarity > threshold:
+            return True
+
+    return False
+
+
+# ---------------------------------------------------------------------------
+# CandidateQuestion → GeneratedQuestion 转换
+# ---------------------------------------------------------------------------
+
+
+def _to_generated_question(candidate: CandidateQuestion) -> GeneratedQuestion:
+    """将 CandidateQuestion 转换为 GeneratedQuestion。
+
+    参数:
+        candidate: 门控通过的候选题目。
+
+    返回:
+        GeneratedQuestion 实例（difficulty_steps 初始为 None）。
+    """
+    return GeneratedQuestion(
+        question_id=candidate.question_id,
+        video_id=candidate.video_id,
+        task_type=candidate.task_type,
+        question=candidate.question,
+        options=candidate.options,
+        answer=candidate.answer,
+        source_nodes=candidate.source_nodes,
+        difficulty=candidate.difficulty,
+        skill_target=candidate.skill_target,
+        difficulty_steps=None,
+    )
+
+
+# ---------------------------------------------------------------------------
+# 单 Slot 处理（重出循环）
+# ---------------------------------------------------------------------------
+
+
+async def _process_one_slot(
+    slot: SlotAssignment,
+    tree: TreeIndex,
+    vlm: VLMProvider,
+    llm: LLMProvider,
+    embed_fn: Callable[[str], np.ndarray],
+    embed_pool: list[np.ndarray],
+    store: QuestionGenStore,
+    config: PipelineConfig,
+    used_node_ids: set[str],
+    rng: random.Random,
+    sem: asyncio.Semaphore,
+    *,
+    session_id: str,
+    run_id: str,
+) -> GeneratedQuestion | None:
+    """处理单个 slot 的完整重出循环。
+
+    流程:
+    1. 采样素材
+    2. 调用 VLM 生成候选题
+    3. 记录到 store
+    4. 后处理（洗牌、检测）
+    5. verbatim 检查
+    6. 四门质量检查
+    7. 去重检测
+    8. 通过 → 接受；失败 → 重出（最多 retry_limit 次）
+
+    参数:
+        slot: slot 分配信息。
+        tree: 视频树索引。
+        vlm: VLM 调用端口。
+        llm: LLM 调用端口。
+        embed_fn: embedding 函数。
+        embed_pool: 已接受题目的 embedding 池。
+        store: 日志记录器。
+        config: 管线配置。
+        used_node_ids: 已用节点 ID 集合。
+        rng: 随机数生成器。
+        sem: 并发信号量。
+        session_id: 会话 ID。
+        run_id: 批次 ID。
+
+    返回:
+        GeneratedQuestion（通过全部检查）或 None（重出耗尽）。
+    """
+    async with sem:
+        prev_reason: str | None = None
+
+        for attempt in range(1, config.retry_limit + 1):
+            # Phase 1: 采样素材
+            try:
+                material = sample_material_v2(
+                    tree=tree,
+                    family_spec=slot.family,
+                    task_type=slot.task_type,
+                    used_node_ids=used_node_ids,
+                    rng=rng,
+                )
+            except (RuntimeError, KeyError) as e:
+                logger.warning(
+                    "slot {} 采样失败 (attempt {}/{}): {}",
+                    slot.slot_id,
+                    attempt,
+                    config.retry_limit,
+                    e,
+                )
+                continue
+
+            # Phase 2: 生成候选题
+            try:
+                candidate = await generate_one_v2(
+                    vlm=vlm,
+                    tree=tree,
+                    material=material,
+                    family_spec=slot.family,
+                    task_type=slot.task_type,
+                    seq=slot.seq,
+                    video_id=slot.video_id,
+                    reject_reason=prev_reason,
+                    session_id=session_id,
+                )
+            except (ValueError, FileNotFoundError) as e:
+                logger.warning(
+                    "slot {} 生成失败 (attempt {}/{}): {}",
+                    slot.slot_id,
+                    attempt,
+                    config.retry_limit,
+                    e,
+                )
+                continue
+
+            # Phase 3: 记录到 store
+            item_id = f"{slot.slot_id}_att{attempt}_{uuid.uuid4().hex[:8]}"
+            store.record_item(
+                item_id=item_id,
+                run_id=run_id,
+                slot_id=slot.slot_id,
+                video_id=slot.video_id,
+                family=slot.family.name,
+                task_type=slot.task_type,
+                skill_target=slot.family.skill_target,
+                attempt=attempt,
+                question_text=candidate.question,
+            )
+
+            # Phase 4: 后处理
+            source_texts = list(material.subtitle_sentences)
+            pp = run_postprocess(
+                question_text=candidate.question,
+                options=candidate.options,
+                answer=candidate.answer,
+                source_texts=source_texts,
+                rng=rng,
+            )
+
+            # Phase 5: verbatim 短路检查
+            if pp.verbatim_ratio > 0.5:
+                logger.info(
+                    "slot {} verbatim 过高 ({:.3f}), attempt {}/{}",
+                    slot.slot_id,
+                    pp.verbatim_ratio,
+                    attempt,
+                    config.retry_limit,
+                )
+                prev_reason = f"verbatim_ratio={pp.verbatim_ratio:.3f} exceeds 0.5"
+                # 仍需记录门控结果
+                skip_result = GateResult(
+                    verdict=GateVerdict.SKIP,
+                    reason="skipped due to verbatim",
+                    raw_response="",
+                )
+                verbatim_report = GateReport(
+                    key_verify=GateResult(
+                        verdict=GateVerdict.FAIL,
+                        reason=prev_reason,
+                        raw_response="",
+                    ),
+                    blind_answer=skip_result,
+                    multi_true=skip_result,
+                    leak_test=skip_result,
+                )
+                store.update_gates(item_id, verbatim_report)
+                continue
+
+            # Phase 6: 四门质量检查
+            report = await run_gates(
+                candidate=candidate,
+                tree=tree,
+                llm=llm,
+                family_spec=slot.family,
+                postprocess=pp,
+                session_id=session_id,
+            )
+            store.update_gates(item_id, report)
+
+            if not report.passed:
+                prev_reason = report.reject_reason
+                logger.info(
+                    "slot {} 门控失败 (attempt {}/{}): {}",
+                    slot.slot_id,
+                    attempt,
+                    config.retry_limit,
+                    prev_reason,
+                )
+                continue
+
+            # Phase 7: 去重检测
+            if _is_duplicate(candidate.question, embed_pool, embed_fn, config.dedup_threshold):
+                prev_reason = "duplicate detected by embedding similarity"
+                logger.info(
+                    "slot {} 重复题被拒绝 (attempt {}/{})",
+                    slot.slot_id,
+                    attempt,
+                    config.retry_limit,
+                )
+                continue
+
+            # Phase 8: 通过全部检查 → 接受
+            result = _to_generated_question(candidate)
+            # 将题目 embedding 加入池
+            embed_pool.append(embed_fn(candidate.question))
+            # 标记使用的节点
+            used_node_ids.update(candidate.source_nodes)
+
+            logger.info(
+                "slot {} 接受: question_id={}, attempt={}/{}",
+                slot.slot_id,
+                result.question_id,
+                attempt,
+                config.retry_limit,
+            )
+            return result
+
+        # 重出耗尽
+        logger.warning("slot {} 重出耗尽 ({} 次)", slot.slot_id, config.retry_limit)
+        return None
+
+
+# ---------------------------------------------------------------------------
+# 重量抽检
+# ---------------------------------------------------------------------------
+
+
+async def _heavy_check_one(
+    question: GeneratedQuestion,
+    tree: TreeIndex,
+    llm: LLMProvider,
+    *,
+    session_id: str,
+) -> int:
+    """盲 Agent 试答 — 计算推理步数作为难度指标。
+
+    让 LLM 在不访问树的情况下尝试回答问题，
+    统计其推理步骤数作为 difficulty_steps。
+
+    参数:
+        question: 待检测的题目。
+        tree: 视频树索引（本函数不使用，保留接口一致性）。
+        llm: LLM 调用端口。
+        session_id: 会话 ID。
+
+    返回:
+        推理步数（int）。解析失败时返回默认值 1。
+    """
+    prompt = (
+        "You are a blind test agent. Answer this multiple-choice question "
+        "WITHOUT any video context. Think step by step.\n\n"
+        f"Question: {question.question}\n"
+        f"Options:\n" + "\n".join(question.options) + "\n\n"
+        "Respond with ONLY a JSON object:\n"
+        '{"steps": [{"thought": "..."}], "answer": "A|B|C|D"}'
+    )
+
+    response = await llm.chat(
+        [{"role": "user", "content": prompt}],
+        session_id=session_id,
+    )
+
+    # 解析步骤数
+    try:
+        import json
+
+        content = response.content.strip()
+        if "```" in content:
+            parts = content.split("```")
+            for part in parts:
+                stripped = part.strip()
+                if stripped.startswith("json"):
+                    stripped = stripped[4:].strip()
+                if stripped.startswith("{"):
+                    content = stripped
+                    break
+
+        data = json.loads(content)
+        steps = data.get("steps", [])
+        return max(len(steps), 1)
+    except (json.JSONDecodeError, TypeError, AttributeError):
+        logger.debug("heavy_check 响应解析失败，返回默认步数 1")
+        return 1
+
+
+# ---------------------------------------------------------------------------
+# 管线主入口
+# ---------------------------------------------------------------------------
+
+
+async def run_pipeline_v2(
+    video_ids: list[str],
+    trees: dict[str, TreeIndex],
+    vlm: VLMProvider,
+    llm: LLMProvider,
+    embed_fn: Callable[[str], np.ndarray],
+    store: QuestionGenStore,
+    config: PipelineConfig,
+    *,
+    task_types: list[str] | None = None,
+    progress: dict[str, str] | None = None,
+) -> PipelineResult:
+    """v2 出题管线主入口 — 编排全部 slot 的生成、检查与抽检。
+
+    流程:
+    1. 分配 slot
+    2. 跳过 progress 中已完成的 slot
+    3. 信号量限流并发处理每个 slot
+    4. 对接受题目按 heavy_sample_rate 随机抽检
+    5. 更新 store 统计
+    6. 返回 PipelineResult
+
+    参数:
+        video_ids: 视频 ID 列表。
+        trees: video_id → TreeIndex 映射。
+        vlm: VLM 调用端口。
+        llm: LLM 调用端口。
+        embed_fn: embedding 函数。
+        store: 日志记录器。
+        config: 管线配置。
+        task_types: 任务类型列表（默认使用 12 类标准集）。
+        progress: 已完成 slot 映射 {slot_id → "accepted"|"rejected"}。
+
+    返回:
+        PipelineResult 实例。
+    """
+    # 默认 12 类标准任务类型
+    if task_types is None:
+        task_types = [
+            "Action Recognition",
+            "Action Reasoning",
+            "Action Prediction",
+            "Action Sequence",
+            "Object Recognition",
+            "Object Reasoning",
+            "Object Interaction",
+            "Scene Understanding",
+            "Event Reasoning",
+            "Causal Reasoning",
+            "Temporal Reasoning",
+            "Spatial Reasoning",
+        ]
+
+    progress = progress or {}
+    rng = random.Random(config.seed)
+
+    # Phase 1: 分配 slot
+    slots = _assign_slots(video_ids, task_types, config.per_type, config.family_ratios, rng)
+    logger.info(
+        "管线启动: {} slots, {} 视频, {} 任务类型", len(slots), len(video_ids), len(task_types)
+    )
+
+    # Phase 2: 创建 run 记录
+    run_id = uuid.uuid4().hex
+    import subprocess
+
+    try:
+        git_sha = subprocess.check_output(
+            ["git", "rev-parse", "--short", "HEAD"],
+            text=True,
+            timeout=5,
+        ).strip()
+    except (subprocess.SubprocessError, FileNotFoundError):
+        git_sha = "unknown"
+
+    store.record_run_start(run_id, git_sha, str(config))
+
+    # Phase 3: 过滤已完成 slot
+    pending_slots = [s for s in slots if s.slot_id not in progress]
+    logger.info(
+        "待处理 slots: {} / {} (已跳过 {})",
+        len(pending_slots),
+        len(slots),
+        len(slots) - len(pending_slots),
+    )
+
+    # Phase 4: 并发处理
+    sem = asyncio.Semaphore(config.concurrency)
+    embed_pool: list[np.ndarray] = []
+    used_node_ids: set[str] = set()
+    session_id = f"pipeline_v2_{run_id[:8]}"
+
+    async def _process_wrapper(slot: SlotAssignment) -> GeneratedQuestion | None:
+        tree = trees.get(slot.video_id)
+        if tree is None:
+            logger.warning("slot {} 对应视频 {} 的树不存在，跳过", slot.slot_id, slot.video_id)
+            return None
+        return await _process_one_slot(
+            slot=slot,
+            tree=tree,
+            vlm=vlm,
+            llm=llm,
+            embed_fn=embed_fn,
+            embed_pool=embed_pool,
+            store=store,
+            config=config,
+            used_node_ids=used_node_ids,
+            rng=rng,
+            sem=sem,
+            session_id=session_id,
+            run_id=run_id,
+        )
+
+    results = await asyncio.gather(*[_process_wrapper(s) for s in pending_slots])
+
+    # Phase 5: 统计结果
+    accepted: list[GeneratedQuestion] = [r for r in results if r is not None]
+    rejected_count = len(pending_slots) - len(accepted)
+
+    logger.info(
+        "管线生成完成: accepted={}, rejected={}",
+        len(accepted),
+        rejected_count,
+    )
+
+    # Phase 6: 重量抽检
+    heavy_sampled: list[tuple[str, int]] = []
+    if config.heavy_sample_rate > 0 and accepted:
+        sample_count = max(1, int(len(accepted) * config.heavy_sample_rate))
+        sample_count = min(sample_count, len(accepted))
+        heavy_rng = random.Random(config.seed + 1)
+        sampled_questions = heavy_rng.sample(accepted, sample_count)
+
+        logger.info("重量抽检: {} / {} 题", len(sampled_questions), len(accepted))
+
+        heavy_tasks = []
+        for q in sampled_questions:
+            tree = trees.get(q.video_id)
+            if tree is None:
+                continue
+            heavy_tasks.append(_heavy_check_one(q, tree, llm, session_id=session_id))
+
+        heavy_results = await asyncio.gather(*heavy_tasks)
+
+        for q, steps in zip(sampled_questions, heavy_results, strict=True):
+            heavy_sampled.append((q.question_id, steps))
+            # 更新 store
+            # 找到对应的 item_id（最后一次 attempt 的记录）
+            cursor = store._conn.execute(
+                "SELECT item_id FROM question_gen_items WHERE slot_id LIKE ? AND final_status='accepted' LIMIT 1",
+                (f"%{q.question_id.split('_')[-1]}%",),
+            )
+            row = cursor.fetchone()
+            if row:
+                store.update_difficulty(row[0], steps)
+
+    # Phase 7: 更新 run 统计
+    from app.question_gen.run_store import RunStats
+
+    stats = RunStats(
+        total_slots=len(slots),
+        accepted=len(accepted),
+        rejected=rejected_count,
+        heavy_sampled=len(heavy_sampled),
+    )
+    store.record_run_end(run_id, "completed", stats)
+
+    return PipelineResult(
+        accepted=accepted,
+        rejected_count=rejected_count,
+        heavy_sampled=heavy_sampled,
+    )

tests/integration/test_pipeline_v2.py

@@ -0,0 +1,650 @@
+"""v2 出题管线集成测试 — 覆盖 slot 分配、重出循环、重量抽检与完整编排。
+
+测试策略：使用受控 mock VLM/LLM 返回，验证管线逻辑正确性。
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import random
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+import numpy as np
+import pytest
+
+from app.question_gen.families import ALL_FAMILIES
+from app.question_gen.pipeline_v2 import (
+    PipelineConfig,
+    PipelineResult,
+    SlotAssignment,
+    _assign_slots,
+    _process_one_slot,
+    run_pipeline_v2,
+)
+from app.tree.index import (
+    IndexMeta,
+    L1Card,
+    L1Node,
+    L2Card,
+    L2Node,
+    L3Card,
+    L3Node,
+    TreeIndex,
+)
+from core.types import LLMResponse
+
+# ---------------------------------------------------------------------------
+# 测试用 fixtures
+# ---------------------------------------------------------------------------
+
+
+def _make_llm_response(content: str) -> LLMResponse:
+    """构造标准 LLMResponse。"""
+    return LLMResponse(
+        content=content,
+        thinking="",
+        model="test-model",
+        provider="test",
+        prompt_tokens=10,
+        completion_tokens=20,
+        latency_ms=100,
+        ttft_ms=50.0,
+        max_inter_token_ms=10.0,
+        cache_hit=False,
+        call_id="call-001",
+    )
+
+
+def _make_gate_pass_response() -> str:
+    """门控全部通过的 JSON 响应。"""
+    return json.dumps({"verdict": "pass", "reason": "looks good"})
+
+
+def _make_gate_fail_response(reason: str = "quality issue") -> str:
+    """门控失败的 JSON 响应。"""
+    return json.dumps({"verdict": "fail", "reason": reason})
+
+
+def _make_candidate_json(
+    question: str = "What happened next?",
+    answer: str = "A",
+) -> str:
+    """构造 VLM 返回的候选题 JSON。"""
+    return json.dumps(
+        {
+            "question": question,
+            "options": [
+                "A. The cat jumped",
+                "B. The dog ran",
+                "C. Nothing happened",
+                "D. It rained",
+            ],
+            "answer": answer,
+            "difficulty": "medium",
+        }
+    )
+
+
+def _make_tree() -> TreeIndex:
+    """构造最小合法三层树。"""
+    l3_nodes = [
+        L3Node(
+            id=f"vid_L1_000_L2_000_L3_{i:03d}",
+            card=L3Card(
+                frame_summary=f"Frame {i} shows activity",
+                visible_entities=["person", "object"],
+                ongoing_actions=["walking"],
+                visible_text=[],
+                spatial_layout="center",
+                visual_attributes={},
+                subtitle=f"Subtitle sentence {i} with some unique content here",
+            ),
+            timestamp=float(i * 2),
+            frame_path=f"frames/L1_000_L2_000_L3_{i:03d}.jpg",
+        )
+        for i in range(5)
+    ]
+    l2 = L2Node(
+        id="vid_L1_000_L2_000",
+        card=L2Card(
+            event_description="A person walks through the park",
+            entities=["person", "park"],
+            actions=["walking"],
+            action_subjects=["person"],
+            visible_text=[],
+            spatial_relations="person in center of park",
+            state_changes=None,
+            subtitle="Person walking in park doing activities",
+        ),
+        children=l3_nodes,
+    )
+    l2_b = L2Node(
+        id="vid_L1_000_L2_001",
+        card=L2Card(
+            event_description="A dog runs across the field",
+            entities=["dog", "field"],
+            actions=["running"],
+            action_subjects=["dog"],
+            visible_text=[],
+            spatial_relations="dog in the field",
+            state_changes=None,
+            subtitle="Dog running across the field",
+        ),
+        children=[
+            L3Node(
+                id=f"vid_L1_000_L2_001_L3_{i:03d}",
+                card=L3Card(
+                    frame_summary=f"Dog frame {i}",
+                    visible_entities=["dog"],
+                    ongoing_actions=["running"],
+                    visible_text=[],
+                    spatial_layout="wide",
+                    visual_attributes={},
+                    subtitle=f"Dog subtitle {i}",
+                ),
+                timestamp=float(10 + i * 2),
+                frame_path=f"frames/L1_000_L2_001_L3_{i:03d}.jpg",
+            )
+            for i in range(4)
+        ],
+    )
+    l1 = L1Node(
+        id="vid_L1_000",
+        card=L1Card(
+            scene_summary="Outdoor activities in a park",
+            main_setting="outdoor park",
+            key_entities=["person", "dog"],
+            main_actions=["walking", "running"],
+            topic_keywords=["outdoor", "activity"],
+            visible_text=[],
+            temporal_flow="sequential activities",
+        ),
+        children=[l2, l2_b],
+    )
+    meta = IndexMeta(source_path="test_video.mp4", modality="video")
+    return TreeIndex(metadata=meta, roots=[l1])
+
+
+class MockVLM:
+    """受控 VLM mock — 每次调用返回候选题 JSON。"""
+
+    def __init__(self, responses: list[str] | None = None) -> None:
+        self._responses = responses or [_make_candidate_json()]
+        self._call_count = 0
+
+    async def chat_with_images(
+        self,
+        messages: list[dict[str, Any]],
+        images: list[str | Path],
+        *,
+        session_id: str | None = None,
+        parent_call_id: str | None = None,
+    ) -> LLMResponse:
+        idx = min(self._call_count, len(self._responses) - 1)
+        self._call_count += 1
+        return _make_llm_response(self._responses[idx])
+
+
+class MockLLM:
+    """受控 LLM mock — 支持配置门控 pass/fail 序列。"""
+
+    def __init__(self, responses: list[str] | None = None) -> None:
+        self._responses = responses or [_make_gate_pass_response()]
+        self._call_count = 0
+
+    async def chat(
+        self,
+        messages: list[dict[str, Any]],
+        *,
+        session_id: str | None = None,
+        parent_call_id: str | None = None,
+    ) -> LLMResponse:
+        idx = min(self._call_count, len(self._responses) - 1)
+        self._call_count += 1
+        return _make_llm_response(self._responses[idx])
+
+
+def _mock_embed_fn(text: str) -> np.ndarray:
+    """确定性 embedding：基于文本 hash 生成向量。"""
+    rng = np.random.default_rng(hash(text) % (2**32))
+    vec = rng.standard_normal(64).astype(np.float32)
+    return vec / np.linalg.norm(vec)
+
+
+@pytest.fixture
+def tree() -> TreeIndex:
+    return _make_tree()
+
+
+@pytest.fixture
+def default_config(tmp_path: Path) -> PipelineConfig:
+    return PipelineConfig(
+        family_ratios={
+            "RETRIEVAL": 0.30,
+            "REASONING": 0.25,
+            "ENUMERATION": 0.20,
+            "VISUAL": 0.15,
+            "SPATIAL": 0.10,
+        },
+        per_type=2,
+        retry_limit=3,
+        heavy_sample_rate=0.15,
+        dedup_threshold=0.85,
+        concurrency=2,
+        seed=42,
+        output_dir=tmp_path / "output",
+        gate_models={
+            "blind_answer_model": "gpt-4.1-mini",
+            "leak_test_model": "gpt-4.1-mini",
+            "key_verify_model": "gpt-4.1-mini",
+            "multi_true_model": "gpt-4.1-mini",
+        },
+        heavy_agent_model="gpt-4.1-mini",
+    )
+
+
+@pytest.fixture
+def store(tmp_path: Path):
+    from app.question_gen.run_store import QuestionGenStore
+
+    db_path = tmp_path / "test_qgen.db"
+    s = QuestionGenStore(db_path=db_path)
+    yield s
+    s.close()
+
+
+# ---------------------------------------------------------------------------
+# TestSlotAssignment
+# ---------------------------------------------------------------------------
+
+
+class TestSlotAssignment:
+    """_assign_slots 单元测试。"""
+
+    def test_per_type_count(self):
+        """验证生成的 slot 总数 = len(task_types) * per_type。"""
+        video_ids = ["vid_001", "vid_002"]
+        task_types = ["Action Recognition", "Object Recognition", "Causal Reasoning"]
+        per_type = 4
+        family_ratios = {"RETRIEVAL": 0.5, "VISUAL": 0.5}
+        rng = random.Random(42)
+
+        slots = _assign_slots(video_ids, task_types, per_type, family_ratios, rng)
+
+        assert len(slots) == len(task_types) * per_type
+
+    def test_family_distribution(self):
+        """验证家族分配来自 get_family_for_slot（合法 family 与 task_type 匹配）。"""
+        video_ids = ["vid_001", "vid_002", "vid_003"]
+        task_types = ["Action Recognition"]
+        per_type = 20
+        family_ratios = {"RETRIEVAL": 0.5, "VISUAL": 0.5}
+        rng = random.Random(42)
+
+        slots = _assign_slots(video_ids, task_types, per_type, family_ratios, rng)
+
+        for slot in slots:
+            assert slot.task_type == "Action Recognition"
+            # Family 必须是 task_type 合法的家族之一
+            family = slot.family
+            assert slot.task_type in family.legal_task_types
+
+    def test_round_robin_across_videos(self):
+        """验证 slot 在视频间轮转分配。"""
+        video_ids = ["vid_A", "vid_B"]
+        task_types = ["Action Recognition"]
+        per_type = 4
+        family_ratios = {"RETRIEVAL": 1.0}
+        rng = random.Random(42)
+
+        slots = _assign_slots(video_ids, task_types, per_type, family_ratios, rng)
+
+        video_assignments = [s.video_id for s in slots]
+        # 应该轮转分配
+        assert video_assignments.count("vid_A") == 2
+        assert video_assignments.count("vid_B") == 2
+
+    def test_deterministic_with_seed(self):
+        """相同 seed 产出相同 slot 序列。"""
+        video_ids = ["vid_001", "vid_002"]
+        task_types = ["Action Recognition", "Causal Reasoning"]
+        per_type = 3
+        family_ratios = {"RETRIEVAL": 0.5, "REASONING": 0.5}
+
+        rng1 = random.Random(99)
+        slots1 = _assign_slots(video_ids, task_types, per_type, family_ratios, rng1)
+
+        rng2 = random.Random(99)
+        slots2 = _assign_slots(video_ids, task_types, per_type, family_ratios, rng2)
+
+        for s1, s2 in zip(slots1, slots2, strict=True):
+            assert s1.slot_id == s2.slot_id
+            assert s1.video_id == s2.video_id
+            assert s1.family.name == s2.family.name
+
+
+# ---------------------------------------------------------------------------
+# TestProcessOneSlot
+# ---------------------------------------------------------------------------
+
+
+class TestProcessOneSlot:
+    """_process_one_slot 集成测试。"""
+
+    @pytest.mark.asyncio
+    async def test_happy_path(self, tree, default_config, store, tmp_path):
+        """首次生成即通过门控 → 返回 GeneratedQuestion。"""
+        vlm = MockVLM()
+        llm = MockLLM([_make_gate_pass_response()] * 4)
+        sem = asyncio.Semaphore(2)
+        used_node_ids: set[str] = set()
+        rng = random.Random(42)
+
+        family = ALL_FAMILIES[0]  # RETRIEVAL
+        slot = SlotAssignment(
+            slot_id="slot_001",
+            video_id="vid_L1_000",
+            task_type="Action Recognition",
+            family=family,
+            seq=1,
+        )
+
+        run_id = "test-run-001"
+        store.record_run_start(run_id, "abc123", "{}")
+
+        result = await _process_one_slot(
+            slot=slot,
+            tree=tree,
+            vlm=vlm,
+            llm=llm,
+            embed_fn=_mock_embed_fn,
+            embed_pool=[],
+            store=store,
+            config=default_config,
+            used_node_ids=used_node_ids,
+            rng=rng,
+            sem=sem,
+            session_id="sess-001",
+            run_id=run_id,
+        )
+
+        assert result is not None
+        assert result.question_id
+        assert result.skill_target == family.skill_target
+
+    @pytest.mark.asyncio
+    async def test_retry_on_fail(self, tree, default_config, store, tmp_path):
+        """第一次门控失败，第二次通过 → 重出成功。"""
+        vlm = MockVLM(
+            [
+                _make_candidate_json("First question?"),
+                _make_candidate_json("Second better question?"),
+            ]
+        )
+        # 第一轮 4 个门有一个 fail，第二轮 4 个门全 pass
+        llm_responses = [
+            _make_gate_fail_response("answer not grounded"),
+            _make_gate_pass_response(),
+            _make_gate_pass_response(),
+            _make_gate_pass_response(),
+            # 第二轮
+            _make_gate_pass_response(),
+            _make_gate_pass_response(),
+            _make_gate_pass_response(),
+            _make_gate_pass_response(),
+        ]
+        llm = MockLLM(llm_responses)
+        sem = asyncio.Semaphore(2)
+        used_node_ids: set[str] = set()
+        rng = random.Random(42)
+
+        family = ALL_FAMILIES[0]  # RETRIEVAL
+        slot = SlotAssignment(
+            slot_id="slot_002",
+            video_id="vid_L1_000",
+            task_type="Action Recognition",
+            family=family,
+            seq=2,
+        )
+
+        run_id = "test-run-002"
+        store.record_run_start(run_id, "abc123", "{}")
+
+        result = await _process_one_slot(
+            slot=slot,
+            tree=tree,
+            vlm=vlm,
+            llm=llm,
+            embed_fn=_mock_embed_fn,
+            embed_pool=[],
+            store=store,
+            config=default_config,
+            used_node_ids=used_node_ids,
+            rng=rng,
+            sem=sem,
+            session_id="sess-002",
+            run_id=run_id,
+        )
+
+        assert result is not None
+        # 确认第二个问题被接受
+        assert "Second" in result.question or result.question_id is not None
+
+    @pytest.mark.asyncio
+    async def test_max_retries_none(self, tree, default_config, store, tmp_path):
+        """所有重试均失败 → 返回 None。"""
+        vlm = MockVLM([_make_candidate_json()] * 5)
+        # 所有门控均失败
+        llm = MockLLM([_make_gate_fail_response("always fails")] * 20)
+        sem = asyncio.Semaphore(2)
+        used_node_ids: set[str] = set()
+        rng = random.Random(42)
+
+        family = ALL_FAMILIES[0]  # RETRIEVAL
+        slot = SlotAssignment(
+            slot_id="slot_003",
+            video_id="vid_L1_000",
+            task_type="Action Recognition",
+            family=family,
+            seq=3,
+        )
+
+        run_id = "test-run-003"
+        store.record_run_start(run_id, "abc123", "{}")
+
+        result = await _process_one_slot(
+            slot=slot,
+            tree=tree,
+            vlm=vlm,
+            llm=llm,
+            embed_fn=_mock_embed_fn,
+            embed_pool=[],
+            store=store,
+            config=default_config,
+            used_node_ids=used_node_ids,
+            rng=rng,
+            sem=sem,
+            session_id="sess-003",
+            run_id=run_id,
+        )
+
+        assert result is None
+
+
+# ---------------------------------------------------------------------------
+# TestPipelineV2
+# ---------------------------------------------------------------------------
+
+
+class TestPipelineV2:
+    """run_pipeline_v2 完整流程测试。"""
+
+    @pytest.mark.asyncio
+    async def test_full_flow(self, tree, default_config, store, tmp_path):
+        """完整管线运行，产出 PipelineResult。"""
+        vlm = MockVLM([_make_candidate_json(f"Question {i}?") for i in range(50)])
+        llm = MockLLM([_make_gate_pass_response()] * 200)
+
+        config = PipelineConfig(
+            family_ratios=default_config.family_ratios,
+            per_type=2,
+            retry_limit=2,
+            heavy_sample_rate=0.5,  # 高比例便于测试
+            dedup_threshold=0.85,
+            concurrency=2,
+            seed=42,
+            output_dir=tmp_path / "out",
+            gate_models=default_config.gate_models,
+            heavy_agent_model="gpt-4.1-mini",
+        )
+
+        # 只用一个 task_type 确保树能满足采样
+        result = await run_pipeline_v2(
+            video_ids=["vid_L1_000"],
+            trees={"vid_L1_000": tree},
+            vlm=vlm,
+            llm=llm,
+            embed_fn=_mock_embed_fn,
+            store=store,
+            config=config,
+            task_types=["Action Recognition"],
+        )
+
+        assert isinstance(result, PipelineResult)
+        assert result.rejected_count >= 0
+        # 至少一题被接受（VLM 和 LLM 全部正常返回）
+        assert len(result.accepted) > 0
+
+    @pytest.mark.asyncio
+    async def test_progress_resume(self, tree, default_config, store, tmp_path):
+        """传入 progress dict → 已处理 slot 被跳过。"""
+        llm = MockLLM([_make_gate_pass_response()] * 100)
+
+        config = PipelineConfig(
+            family_ratios=default_config.family_ratios,
+            per_type=2,
+            retry_limit=2,
+            heavy_sample_rate=0.0,  # 不做 heavy check
+            dedup_threshold=0.85,
+            concurrency=2,
+            seed=42,
+            output_dir=tmp_path / "out",
+            gate_models=default_config.gate_models,
+            heavy_agent_model="gpt-4.1-mini",
+        )
+
+        # 用相同 seed 计算 slot_ids（_assign_slots 是确定性的）
+        rng_preview = random.Random(config.seed)
+        preview_slots = _assign_slots(
+            ["vid_L1_000"],
+            ["Action Recognition"],
+            config.per_type,
+            config.family_ratios,
+            rng_preview,
+        )
+        # 构造 progress 标记所有 slot 已完成
+        progress = {s.slot_id: "accepted" for s in preview_slots}
+
+        # 用 progress 跑管线 → 所有 slot 被跳过
+        vlm2 = MockVLM([_make_candidate_json()] * 20)
+
+        result2 = await run_pipeline_v2(
+            video_ids=["vid_L1_000"],
+            trees={"vid_L1_000": tree},
+            vlm=vlm2,
+            llm=llm,
+            embed_fn=_mock_embed_fn,
+            store=store,
+            config=config,
+            task_types=["Action Recognition"],
+            progress=progress,
+        )
+
+        # 所有 slot 在 progress 中 → VLM 零调用
+        assert vlm2._call_count == 0
+        assert len(result2.accepted) == 0
+
+    @pytest.mark.asyncio
+    async def test_heavy_check_samples(self, tree, default_config, store, tmp_path):
+        """heavy_sample_rate > 0 时有题被抽检。"""
+        # LLM 响应：前面是 gate pass，后面增加 heavy check 的步骤响应
+        heavy_response = json.dumps(
+            {
+                "steps": [
+                    {"thought": "step 1"},
+                    {"thought": "step 2"},
+                    {"thought": "step 3"},
+                ],
+                "answer": "A",
+            }
+        )
+        llm = MockLLM([_make_gate_pass_response()] * 100 + [heavy_response] * 20)
+        vlm = MockVLM([_make_candidate_json(f"Q{i}?") for i in range(20)])
+
+        config = PipelineConfig(
+            family_ratios=default_config.family_ratios,
+            per_type=2,
+            retry_limit=2,
+            heavy_sample_rate=1.0,  # 100% 抽检
+            dedup_threshold=0.85,
+            concurrency=2,
+            seed=42,
+            output_dir=tmp_path / "out",
+            gate_models=default_config.gate_models,
+            heavy_agent_model="gpt-4.1-mini",
+        )
+
+        result = await run_pipeline_v2(
+            video_ids=["vid_L1_000"],
+            trees={"vid_L1_000": tree},
+            vlm=vlm,
+            llm=llm,
+            embed_fn=_mock_embed_fn,
+            store=store,
+            config=config,
+            task_types=["Action Recognition"],
+        )
+
+        # 100% 抽检 → heavy_sampled 等于 accepted 数
+        if result.accepted:
+            assert len(result.heavy_sampled) == len(result.accepted)
+
+    @pytest.mark.asyncio
+    async def test_store_records_all(self, tree, default_config, store, tmp_path):
+        """验证 store 中记录了每道题的生成与门控结果。"""
+        vlm = MockVLM([_make_candidate_json()] * 10)
+        llm = MockLLM([_make_gate_pass_response()] * 50)
+
+        config = PipelineConfig(
+            family_ratios=default_config.family_ratios,
+            per_type=2,
+            retry_limit=2,
+            heavy_sample_rate=0.0,
+            dedup_threshold=0.85,
+            concurrency=1,
+            seed=42,
+            output_dir=tmp_path / "out",
+            gate_models=default_config.gate_models,
+            heavy_agent_model="gpt-4.1-mini",
+        )
+
+        result = await run_pipeline_v2(
+            video_ids=["vid_L1_000"],
+            trees={"vid_L1_000": tree},
+            vlm=vlm,
+            llm=llm,
+            embed_fn=_mock_embed_fn,
+            store=store,
+            config=config,
+            task_types=["Action Recognition"],
+        )
+
+        # 查询 store 中的 items
+        cursor = store._conn.execute("SELECT COUNT(*) FROM question_gen_items")
+        item_count = cursor.fetchone()[0]
+        # 至少有 accepted 数量的记录
+        assert item_count >= len(result.accepted)