From 1d222d9f18485443ab367a474bf61154cdc65bc3 Mon Sep 17 00:00:00 2001
From: iomgaa <iomgaaycz@gmail.com>
Date: Tue, 14 Jul 2026 16:58:32 -0400
Subject: [PATCH] feat: wire real agent runner, backfill assembly and
 adversarial-filter CLI

---
 app/question_gen/adversarial_filter.py        | 283 +++++++++++++++++-
 .../test_adversarial_filter_e2e.py            | 277 +++++++++++++++++
 tools/generate_questions.py                   | 175 +++++++++++
 3 files changed, 734 insertions(+), 1 deletion(-)
 create mode 100644 tests/integration/test_adversarial_filter_e2e.py

diff --git a/app/question_gen/adversarial_filter.py b/app/question_gen/adversarial_filter.py
index a8e76b4..a7739d6 100644
--- a/app/question_gen/adversarial_filter.py
+++ b/app/question_gen/adversarial_filter.py
@@ -9,6 +9,7 @@ shortcut：作弊门（agent 秒杀=太简单，剔除）+ 翻转门（agent 答
 
 from __future__ import annotations
 
+import dataclasses
 import enum
 import hashlib
 import json
@@ -22,11 +23,16 @@ from loguru import logger
 from core.types import GeneratedQuestion
 
 if TYPE_CHECKING:
+    from collections.abc import Callable
+
+    import numpy as np
+
     from app.question_gen.adversarial_config import AdversarialFilterConfig
+    from app.question_gen.pipeline_v2 import PipelineConfig
     from app.question_gen.run_store import QuestionGenStore
     from app.question_gen.sampler_v2 import MaterialContext
     from app.tree.index import TreeIndex
-    from core.protocols import VLMProvider
+    from core.protocols import LLMProvider, VLMProvider
 
 _VALID_LETTERS = ("A", "B", "C", "D")
 
@@ -757,3 +763,278 @@ async def run_adversarial_rounds(
             all_questions[q.question_id] = q
         pending = new_qs  # 只对新补的题重新过滤（已判题走续跑）
     logger.info("对抗过滤结束: final={} 题（target={}）", passed_now, target)
+
+
+# ---------------------------------------------------------------------------
+# Task 10: 真实 agent 装配 — _RealAgentRunner（run_inference + RunLogImpl 回读）
+# ---------------------------------------------------------------------------
+
+
+class _RealAgentRunner:
+    """AgentRunner 实现 — 复用 run_inference + RunLogImpl 读回预测。
+
+    每次 predict 开一次 HarnessLog（按 run_id 幂等 upsert），跑完整 agent 推理把
+    prediction 落 predictions 表，再用只读 RunLogImpl 按 (run_id, question_ids) 读回。
+    prediction 是 submit_answer 的答案字母（A/B/C/D）；agent 报错/未提交时该行
+    prediction 为 None，据此保守处理（不误判）。
+
+    参数:
+        llm: 推理 LLMProvider（共享注入）。
+        tool_dispatch_fn: InferenceDepsRouter.create_dispatch() 返回的调度闭包。
+        prompt_builder: InferenceDepsRouter.create_prompt_builder() 返回的构建闭包。
+        db_path: HarnessLog / RunLogImpl 的 sqlite 路径。
+        concurrency: run_inference 并发数（asyncio.Semaphore 容量）。
+        skill_mode: skill 模式（"auto"/"manual"/"none"）——同时是指纹来源，须与
+            router 的 skill_mode 一致，保证续跑/失效口径稳定。
+        model: 推理模型名——指纹来源之一。
+    """
+
+    def __init__(
+        self,
+        *,
+        llm: LLMProvider,
+        tool_dispatch_fn: Callable[..., object],
+        prompt_builder: Callable[[GeneratedQuestion], tuple[str, str]],
+        db_path: str,
+        concurrency: int,
+        skill_mode: str,
+        model: str,
+    ) -> None:
+        self._llm = llm
+        self._dispatch = tool_dispatch_fn
+        self._builder = prompt_builder
+        self._db_path = db_path
+        self._concurrency = concurrency
+        self.skill_mode = skill_mode
+        self.model = model
+
+    async def predict(
+        self,
+        questions: list[GeneratedQuestion],
+        *,
+        max_steps: int,
+        run_id: str,
+    ) -> dict[str, str | None]:
+        """跑完整 agent，回读 predictions 表，返回 question_id → 预测字母（无预测 None）。
+
+        参数:
+            questions: 待推理题目列表。
+            max_steps: AgentLoop 单题最大步数。
+            run_id: 本次推理 run 标识（predictions 表按此过滤回读）。
+
+        返回:
+            question_id → 预测答案字母（缺失/未提交为 None）。
+        """
+        from app.harness.inference import run_inference
+        from app.harness.log import HarnessLog, RunLogImpl
+
+        with HarnessLog(self._db_path, run_id) as log:
+            await run_inference(
+                questions,
+                llm=self._llm,
+                tool_dispatch_fn=self._dispatch,
+                prompt_builder=self._builder,
+                log=log,
+                run_id=run_id,
+                concurrency=self._concurrency,
+                max_steps=max_steps,
+                skill_mode=self.skill_mode,
+            )
+        rows = await RunLogImpl(self._db_path).get_predictions(
+            run_id, question_ids=[q.question_id for q in questions]
+        )
+        return {r["question_id"]: r["prediction"] for r in rows}
+
+
+# ---------------------------------------------------------------------------
+# Task 10: 真实 backfill 装配 — 缺额驱动 run_pipeline_v2
+# ---------------------------------------------------------------------------
+
+
+def _max_existing_seq(existing: dict[str, GeneratedQuestion]) -> int:
+    """从已有题 question_id（格式 "{video_id}_{task_type}_{seq:04d}"）取最大 seq。
+
+    只解析末段为纯数字的 id；无可解析 id → 0（新 run 从 seq_offset+1 起编，防撞。）
+    """
+    max_seq = 0
+    for qid in existing:
+        tail = qid.rsplit("_", 1)[-1]
+        if tail.isdigit():
+            max_seq = max(max_seq, int(tail))
+    return max_seq
+
+
+def build_backfill(
+    *,
+    trees: dict[str, TreeIndex],
+    vlm: VLMProvider,
+    llm: LLMProvider,
+    embed_fn: Callable[[str], np.ndarray],
+    store: QuestionGenStore,
+    pipeline_config: PipelineConfig,
+    filter_task_types: tuple[str, ...],
+    session_id: str,
+) -> BackfillFn:
+    """组装真实 backfill 回调 — 缺额驱动 run_pipeline_v2，返回新增 AR 题。
+
+    闭包捕获 run_pipeline_v2 全部依赖。每轮按缺额 deficit 用 dataclasses.replace 生成
+    per_type=deficit 的新 config（PipelineConfig frozen，绝不原地改），并传：
+      - initial_used_node_ids = 已有题 source_nodes 并集（避开已用节点）；
+      - initial_embed_pool = 已有题 embedding（跨 run 去重）；
+      - seq_offset = 已有题最大 seq（新题续编，防撞 question_id）。
+    仅补 filter_task_types（当前锁定为 AR 单类），返回 PipelineResult.accepted。
+
+    参数:
+        trees: video_id → 三层树索引（生成素材来源）。
+        vlm: VLM 端口。
+        llm: LLM 端口（门控用）。
+        embed_fn: 文本嵌入函数。
+        store: 出题持久化。
+        pipeline_config: ar30 原配置（除 per_type 外全部继承）。
+        filter_task_types: 补生成的题型（仅这些）。
+        session_id: 遥测会话 ID（透传给管线子调用）。
+
+    返回:
+        BackfillFn 闭包。
+    """
+    from app.question_gen.pipeline_v2 import run_pipeline_v2
+
+    video_ids = list(trees.keys())
+
+    async def _backfill(
+        deficit: int,
+        round_no: int,
+        existing: dict[str, GeneratedQuestion],
+    ) -> list[GeneratedQuestion]:
+        """按缺额补生成 deficit 道 AR 题。"""
+        run_cfg = dataclasses.replace(pipeline_config, per_type=deficit)
+        used_node_ids: set[str] = set()
+        for q in existing.values():
+            used_node_ids.update(q.source_nodes)
+        embed_pool = [embed_fn(q.question).flatten() for q in existing.values()]
+        seq_offset = _max_existing_seq(existing)
+        logger.info(
+            "backfill round={}: deficit={}, seq_offset={}, used_nodes={}",
+            round_no,
+            deficit,
+            seq_offset,
+            len(used_node_ids),
+        )
+        result = await run_pipeline_v2(
+            video_ids=video_ids,
+            trees=trees,
+            vlm=vlm,
+            llm=llm,
+            embed_fn=embed_fn,
+            store=store,
+            config=run_cfg,
+            task_types=list(filter_task_types),
+            initial_used_node_ids=used_node_ids,
+            initial_embed_pool=embed_pool,
+            seq_offset=seq_offset,
+        )
+        return result.accepted
+
+    return _backfill
+
+
+# ---------------------------------------------------------------------------
+# Task 10: 顶层入口 — run_adversarial_filter
+# ---------------------------------------------------------------------------
+
+
+def _load_filter_questions(
+    accepted_path: Path, filter_task_types: tuple[str, ...]
+) -> list[GeneratedQuestion]:
+    """从 accepted_questions.json 读题并过滤到 filter_task_types（只读，不改 Phase A）。
+
+    accepted_questions.json 是 Phase A 产物（JSON 列表）；仅 filter_task_types 的题
+    进 agent 门，其余原样留在 Phase A 文件中（本层不触碰）。
+
+    参数:
+        accepted_path: accepted_questions.json 路径。
+        filter_task_types: 需过滤的题型集合。
+
+    返回:
+        过滤后的题目列表（保持文件内顺序）。
+
+    异常:
+        FileNotFoundError: 文件不存在。
+        ValueError: JSON 结构非列表。
+    """
+    if not accepted_path.exists():
+        raise FileNotFoundError(f"accepted_questions.json 不存在: {accepted_path}")
+    raw = json.loads(accepted_path.read_text(encoding="utf-8"))
+    if not isinstance(raw, list):
+        raise ValueError(f"accepted_questions.json 顶层结构应为列表: {accepted_path}")
+    types = set(filter_task_types)
+    questions: list[GeneratedQuestion] = []
+    for item in raw:
+        if item.get("task_type") not in types:
+            continue
+        questions.append(
+            GeneratedQuestion(
+                question_id=item["question_id"],
+                video_id=item["video_id"],
+                task_type=item["task_type"],
+                question=item["question"],
+                options=tuple(item["options"]),
+                answer=item["answer"],
+                source_nodes=tuple(item.get("source_nodes", ())),
+                difficulty=item.get("difficulty", "medium"),
+                family=item.get("family"),
+                skill_target=item.get("skill_target"),
+                difficulty_steps=item.get("difficulty_steps"),
+                sub_pattern=item.get("sub_pattern"),
+            )
+        )
+    return questions
+
+
+async def run_adversarial_filter(
+    *,
+    accepted_path: Path,
+    final_path: Path,
+    agent: AgentRunner,
+    vlm: VLMProvider,
+    trees: dict[str, TreeIndex],
+    store: QuestionGenStore,
+    filter_config: AdversarialFilterConfig,
+    backfill: BackfillFn,
+    session_id: str,
+) -> None:
+    """Phase B 顶层入口：读 accepted_questions.json 过滤 AR，跑两门 + 补生成迭代。
+
+    路径隔离：只处理 filter_task_types 的题，accepted_questions.json 只读，Phase A
+    状态机不受影响。target 锁定为首轮过滤出的 AR 题数（缺额 = target - passed）。
+
+    参数:
+        accepted_path: Phase A 产物 accepted_questions.json 路径（只读）。
+        final_path: 最终题库 JSON 路径（每轮全量原子重写）。
+        agent: 完整 agent 试答端口（Task 10 的 _RealAgentRunner，单测可 mock）。
+        vlm: 镜像题生成 VLM 端口。
+        trees: video_id → 三层树索引（重建镜像素材 / 补生成用）。
+        store: verdict 持久化。
+        filter_config: 后置对抗过滤配置（题型 / 轮次 / max_steps / 难度阈值）。
+        backfill: 补生成回调（CLI 用 build_backfill 组装真实实现，单测可 mock）。
+        session_id: 遥测会话 ID（派生各门 run_id）。
+    """
+    initial_questions = _load_filter_questions(accepted_path, filter_config.filter_task_types)
+    target = len(initial_questions)
+    logger.info(
+        "对抗过滤启动: 过滤题型={}, 首轮 AR 题数={}",
+        filter_config.filter_task_types,
+        target,
+    )
+    await run_adversarial_rounds(
+        initial_questions,
+        agent=agent,
+        vlm=vlm,
+        store=store,
+        trees=trees,
+        config=filter_config,
+        final_path=final_path,
+        target=target,
+        backfill=backfill,
+        session_id=session_id,
+    )
diff --git a/tests/integration/test_adversarial_filter_e2e.py b/tests/integration/test_adversarial_filter_e2e.py
new file mode 100644
index 0000000..93ad4ca
--- /dev/null
+++ b/tests/integration/test_adversarial_filter_e2e.py
@@ -0,0 +1,277 @@
+"""Task 10：run_adversarial_filter 顶层入口 + _RealAgentRunner 真实装配 e2e。
+
+覆盖两条路径：
+- run_adversarial_filter 编排（mock agent + mock backfill）：过滤 filter_task_types、
+  非 AR 题不进 agent 门、final 仅含 passed、断点续跑不重跑已判题。
+- _RealAgentRunner 真实装配 smoke（I6）：predict 经 run_inference 落 predictions 表再
+  读回（LLM mock，路径真穿过 HarnessLog / RunLogImpl，非假 runner 短路）。
+"""
+
+from __future__ import annotations
+
+import json
+from typing import TYPE_CHECKING, Any
+from unittest.mock import AsyncMock, MagicMock
+
+import numpy as np
+import pytest
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+from app.harness.deps_router import InferenceDepsRouter
+from app.question_gen.adversarial_config import AdversarialFilterConfig
+from app.question_gen.adversarial_filter import _RealAgentRunner, run_adversarial_filter
+from app.question_gen.run_store import QuestionGenStore
+from core.types import GeneratedQuestion, LLMResponse
+
+
+def _ar_q(qid: str, video_id: str = "v1") -> dict:
+    """构造一条 AR 题 JSON 记录；sub_pattern 缺省 → 不支持 flip，翻转门直接放行。"""
+    return {
+        "question_id": qid,
+        "video_id": video_id,
+        "task_type": "Action Recognition",
+        "question": f"{qid} 之前做了什么？",
+        "options": ["A. 蒸", "B. 炒", "C. 煮", "D. 炸"],
+        "answer": "A",
+        "source_nodes": ["n1"],
+        "difficulty": "hard",
+    }
+
+
+def _non_ar_q(qid: str, video_id: str = "v1") -> dict:
+    """构造一条非 AR 题 JSON 记录（不应进 agent 门）。"""
+    return {
+        "question_id": qid,
+        "video_id": video_id,
+        "task_type": "Object Recognition",
+        "question": f"{qid} 里的物体是什么？",
+        "options": ["A. 锅", "B. 碗", "C. 盘", "D. 勺"],
+        "answer": "A",
+        "source_nodes": ["n2"],
+        "difficulty": "hard",
+    }
+
+
+class _FakeAgent:
+    """完整 agent 试答桩：对每题返回固定预测，记录被调用的 question_id。"""
+
+    def __init__(self, pred: str = "B", model: str = "m1", skill_mode: str = "auto") -> None:
+        self._pred = pred
+        self.model = model
+        self.skill_mode = skill_mode
+        self.calls: list[str] = []
+
+    async def predict(
+        self, questions: list[GeneratedQuestion], *, max_steps: int, run_id: str
+    ) -> dict[str, str]:
+        self.calls.extend(q.question_id for q in questions)
+        return {q.question_id: self._pred for q in questions}
+
+
+class _NoBackfill:
+    """补生成回调桩：记录调用次数，永远返回空（首轮即达标时不应被调用）。"""
+
+    def __init__(self) -> None:
+        self.calls = 0
+
+    async def __call__(
+        self, deficit: int, round_no: int, existing: dict[str, GeneratedQuestion]
+    ) -> list[GeneratedQuestion]:
+        self.calls += 1
+        return []
+
+
+def _write_accepted(path: Path, records: list[dict]) -> None:
+    """写 accepted_questions.json（Phase A 产物形态：JSON 列表）。"""
+    path.write_text(json.dumps(records, ensure_ascii=False, indent=2), encoding="utf-8")
+
+
+@pytest.mark.asyncio
+async def test_filter_only_processes_ar_and_writes_passed(tmp_path: Path) -> None:
+    """非 AR 题不进 agent 门；final 仅含 passed AR 题。"""
+    accepted = tmp_path / "accepted_questions.json"
+    _write_accepted(accepted, [_ar_q("q1"), _ar_q("q2"), _non_ar_q("obj1")])
+    final_path = tmp_path / "accepted_questions_final.json"
+    store = QuestionGenStore(str(tmp_path / "q.db"))
+    agent = _FakeAgent(pred="B")  # 答错 → 过作弊门；sub_pattern=None → 过翻转门
+    backfill = _NoBackfill()
+
+    await run_adversarial_filter(
+        accepted_path=accepted,
+        final_path=final_path,
+        agent=agent,
+        vlm=object(),
+        trees={},
+        store=store,
+        filter_config=AdversarialFilterConfig(adversarial_max_rounds=3),
+        backfill=backfill,
+        session_id="s",
+    )
+
+    # 只有 AR 题进 agent 门
+    assert set(agent.calls) == {"q1", "q2"}
+    # 非 AR 题不出现在 verdicts 表
+    rows = store._conn.execute(
+        "SELECT question_id FROM adversarial_verdicts WHERE question_id=?", ("obj1",)
+    ).fetchall()
+    assert rows == []
+    # final 仅含 passed AR 题
+    data = json.loads(final_path.read_text(encoding="utf-8"))
+    assert {d["question_id"] for d in data} == {"q1", "q2"}
+    assert backfill.calls == 0  # 首轮即达标（target=2, passed=2）
+    store.close()
+
+
+@pytest.mark.asyncio
+async def test_filter_resume_does_not_rerun_judged(tmp_path: Path) -> None:
+    """断点续跑：第二次调用不重跑已判题（agent 调用计数不变）。"""
+    accepted = tmp_path / "accepted_questions.json"
+    _write_accepted(accepted, [_ar_q("q1"), _ar_q("q2")])
+    final_path = tmp_path / "accepted_questions_final.json"
+    store = QuestionGenStore(str(tmp_path / "q.db"))
+    agent = _FakeAgent(pred="B")
+    backfill = _NoBackfill()
+
+    async def _run() -> None:
+        await run_adversarial_filter(
+            accepted_path=accepted,
+            final_path=final_path,
+            agent=agent,
+            vlm=object(),
+            trees={},
+            store=store,
+            filter_config=AdversarialFilterConfig(adversarial_max_rounds=3),
+            backfill=backfill,
+            session_id="s",
+        )
+
+    await _run()
+    first_calls = list(agent.calls)
+    await _run()
+    assert agent.calls == first_calls  # 第二次未新增 agent 调用
+    store.close()
+
+
+# ---------------------------------------------------------------------------
+# I6：_RealAgentRunner 真实装配 smoke（LLM mock，路径真穿过 predictions 表）
+# ---------------------------------------------------------------------------
+
+
+class _MockLLM:
+    """最小 LLM 桩：一步即产出 submit_answer，让真实 AgentLoop 稳定收敛。"""
+
+    def __init__(self, answer: str) -> None:
+        self._answer = answer
+
+    async def chat(
+        self,
+        messages: list[dict[str, Any]],
+        *,
+        session_id: str | None = None,
+        parent_call_id: str | None = None,
+    ) -> LLMResponse:
+        content = json.dumps(
+            {"action": {"tool": "submit_answer", "args": {"answer": self._answer}}}
+        )
+        return LLMResponse(
+            content=content,
+            thinking="",
+            model="mock",
+            provider="mock",
+            prompt_tokens=1,
+            completion_tokens=1,
+            latency_ms=1,
+            ttft_ms=None,
+            max_inter_token_ms=None,
+            cache_hit=False,
+            call_id="mock-call",
+        )
+
+
+def _build_real_router(tmp_path: Path) -> InferenceDepsRouter:
+    """构建真实 InferenceDepsRouter（仅 embed/vlm 打桩，router/deps 装配全真实）。"""
+    vid = "smoke_vid"
+    vid_dir = tmp_path / "videos" / vid
+    (vid_dir / "frames").mkdir(parents=True)
+    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_dir = tmp_path / "prompts"
+    prompts_dir.mkdir()
+    (prompts_dir / "system.md").write_text("You are a search agent.")
+
+    fake_embed = MagicMock()
+    fake_embed.dim = 4
+    fake_embed.embed = lambda t: np.zeros((1, 4), dtype=np.float32)
+
+    return InferenceDepsRouter(
+        store_dir=tmp_path,
+        embed_provider=fake_embed,
+        llm=_MockLLM(answer="B"),
+        vlm=AsyncMock(),
+        ocr=None,
+        default_prompts_dir=prompts_dir,
+        default_skills_dir=None,
+        skill_mode="none",
+        verify_vision=False,
+        anchor=False,
+        assemble_mode="ids",
+    )
+
+
+def _smoke_q(qid: str) -> GeneratedQuestion:
+    """构造 smoke 题（video_id 对应真实 router 的树 fixture）。"""
+    return GeneratedQuestion(
+        question_id=qid,
+        video_id="smoke_vid",
+        task_type="Action Recognition",
+        question="他之前做了什么？",
+        options=("A. 蒸", "B. 炒", "C. 煮", "D. 炸"),
+        answer="A",
+        source_nodes=("L1_000",),
+        difficulty="hard",
+    )
+
+
+@pytest.mark.asyncio
+async def test_real_agent_runner_predict_roundtrips_predictions(tmp_path: Path) -> None:
+    """真实装配 smoke：predict 经 run_inference 落 predictions 表再读回（LLM mock）。"""
+    router = _build_real_router(tmp_path)
+    runner = _RealAgentRunner(
+        llm=_MockLLM(answer="B"),
+        tool_dispatch_fn=router.create_dispatch(),
+        prompt_builder=router.create_prompt_builder(),
+        db_path=str(tmp_path / "harness.db"),
+        concurrency=1,
+        skill_mode="none",
+        model="mock",
+    )
+    preds = await runner.predict([_smoke_q("smoke")], max_steps=2, run_id="smoke_r0")
+    assert preds["smoke"] == "B"  # 真的从 predictions 表读回，非 mock 直返
+
+    # 断言确实写进了 predictions 表（穿过 HarnessLog / RunLogImpl）
+    from app.harness.log import RunLogImpl
+
+    rows = await RunLogImpl(str(tmp_path / "harness.db")).get_predictions(
+        "smoke_r0", question_ids=["smoke"]
+    )
+    assert rows and rows[0]["prediction"] == "B"
diff --git a/tools/generate_questions.py b/tools/generate_questions.py
index acfdb7d..8796a23 100644
--- a/tools/generate_questions.py
+++ b/tools/generate_questions.py
@@ -1173,6 +1173,176 @@ async def _run_generate_v2(args: argparse.Namespace) -> None:
         logger.warning("超过 50% 的 slot 被拒绝，建议检查 VLM/门控配置")
 
 
+def _load_trees_abs(videos_dir: Path, video_ids: list[str]) -> dict:
+    """加载视频树并将相对帧路径解析为绝对路径（复用 generate-v2 逻辑）。
+
+    参数:
+        videos_dir: store/videos 目录。
+        video_ids: 待加载的 video_id 列表。
+
+    返回:
+        video_id → TreeIndex 映射（加载失败的视频被跳过）。
+    """
+    from app.tree.index import TreeIndex
+
+    trees: dict = {}
+    for vid in video_ids:
+        tree_path = videos_dir / vid / "tree.json"
+        try:
+            tree = TreeIndex.load_json(str(tree_path))
+        except (OSError, ValueError, KeyError) as exc:
+            logger.warning("加载树 {} 失败，跳过: {}", tree_path, exc)
+            continue
+        video_dir = videos_dir / vid
+        for l1 in tree.roots:
+            for l2 in l1.children:
+                for l3 in l2.children:
+                    if l3.frame_path and not Path(l3.frame_path).is_absolute():
+                        l3.frame_path = str(video_dir / l3.frame_path)
+        trees[vid] = tree
+    return trees
+
+
+def _add_adversarial_filter_parser(subparsers: argparse._SubParsersAction) -> None:
+    """注册 adversarial-filter 子命令（Phase B 后置对抗过滤 CLI 入口）。
+
+    参数:
+        subparsers: argparse 子命令注册器。
+    """
+    p = subparsers.add_parser(
+        "adversarial-filter",
+        help="Phase B 后置对抗过滤（作弊门 + 翻转门 + 缺额补生成）",
+    )
+    p.add_argument("--config", type=Path, required=True, help="YAML 配置（含 question_gen_v2 / adversarial_filter / embed 段）")
+    p.add_argument("--store-dir", type=Path, required=True, help="store 根目录（含 videos/ prompts/ skills/）")
+    p.add_argument("--accepted-path", type=Path, required=True, help="Phase A 产物 accepted_questions.json 路径（只读）")
+    p.add_argument("--final-path", type=Path, default=None, help="最终题库输出路径（默认 accepted 同目录 accepted_questions_final.json）")
+    p.add_argument("--db-path", type=Path, default=Path("logs/question_gen.db"), help="QuestionGenStore SQLite 路径")
+    p.add_argument("--harness-db", type=Path, default=Path("logs/adversarial_harness.db"), help="agent 推理 HarnessLog SQLite 路径")
+    p.add_argument("--prompts-version", type=str, default="v1", help="推理 prompt 版本目录名（store/prompts/<version>）")
+    p.add_argument("--skills-version", type=str, default="v1", help="推理 skill 版本目录名（store/skills/<version>）")
+    p.add_argument("--skill-mode", type=str, choices=["auto", "manual", "none"], default="auto", help="skill 模式")
+    p.add_argument("--concurrency", type=int, default=4, help="agent 推理并发数")
+    p.add_argument("--session-id", type=str, default="adversarial", help="遥测会话 ID（派生各门 run_id）")
+
+
+async def _run_adversarial_filter(args: argparse.Namespace) -> None:
+    """adversarial-filter 子命令主流程。
+
+    装配 adapters（同 main._build_adapters）、InferenceDepsRouter（同 main.py 参数）、
+    QuestionGenStore、视频树（帧路径绝对化）、真实 _RealAgentRunner 与真实 backfill，
+    调 run_adversarial_filter 跑两门 + 缺额补生成迭代。
+
+    参数:
+        args: CLI 参数（config, store_dir, accepted_path, final_path, db_path,
+              harness_db, prompts_version, skills_version, skill_mode, concurrency,
+              session_id）。
+    """
+    import yaml
+
+    from app.harness.deps_router import InferenceDepsRouter
+    from app.question_gen.adversarial_config import load_adversarial_config
+    from app.question_gen.adversarial_filter import (
+        _RealAgentRunner,
+        build_backfill,
+        run_adversarial_filter,
+    )
+    from app.question_gen.pipeline_v2 import load_pipeline_config
+    from app.question_gen.run_store import QuestionGenStore
+    from main import InfraSettings, _build_adapters
+
+    config_path = args.config.resolve()
+    store_dir = args.store_dir.resolve()
+
+    # Phase 1: 加载配置（对抗过滤 + 出题管线 + embed 段）
+    with config_path.open(encoding="utf-8") as f:
+        raw_yaml = yaml.safe_load(f) or {}
+    embed_cfg = raw_yaml.get("embed", {})
+    filter_config = load_adversarial_config(config_path)
+    pipeline_config = load_pipeline_config(config_path)
+
+    # Phase 2: 装配 adapters
+    settings = InfraSettings()
+    adapters = _build_adapters(settings, embed_cfg)
+
+    # Phase 3: 加载视频树（帧路径绝对化）
+    videos_dir = store_dir / "videos"
+    if not videos_dir.exists():
+        logger.error("视频目录不存在: {}", videos_dir)
+        sys.exit(1)
+    video_ids = sorted(
+        d.name for d in videos_dir.iterdir() if d.is_dir() and (d / "tree.json").exists()
+    )
+    trees = _load_trees_abs(videos_dir, video_ids)
+    if not trees:
+        logger.error("所有视频树加载失败，无法继续")
+        sys.exit(1)
+    logger.info("成功加载 {} / {} 棵视频树", len(trees), len(video_ids))
+
+    # Phase 4: InferenceDepsRouter（同 main.py 参数）
+    router = InferenceDepsRouter(
+        store_dir=store_dir,
+        embed_provider=adapters.embed,
+        llm=adapters.llm,
+        vlm=adapters.vlm,
+        ocr=adapters.ocr,
+        default_prompts_dir=store_dir / "prompts" / args.prompts_version,
+        default_skills_dir=store_dir / "skills" / args.skills_version,
+        skill_mode=args.skill_mode,
+        verify_vision=True,
+        anchor=True,
+        assemble_mode="ids_expand",
+    )
+
+    # Phase 5: QuestionGenStore + 真实 agent + 真实 backfill
+    db_path = args.db_path.resolve()
+    db_path.parent.mkdir(parents=True, exist_ok=True)
+    store = QuestionGenStore(str(db_path))
+
+    harness_db = args.harness_db.resolve()
+    harness_db.parent.mkdir(parents=True, exist_ok=True)
+    agent = _RealAgentRunner(
+        llm=adapters.llm,
+        tool_dispatch_fn=router.create_dispatch(),
+        prompt_builder=router.create_prompt_builder(),
+        db_path=str(harness_db),
+        concurrency=args.concurrency,
+        skill_mode=args.skill_mode,
+        model=settings.search_llm_model,
+    )
+    backfill = build_backfill(
+        trees=trees,
+        vlm=adapters.vlm,
+        llm=adapters.llm,
+        embed_fn=adapters.embed.embed,
+        store=store,
+        pipeline_config=pipeline_config,
+        filter_task_types=filter_config.filter_task_types,
+        session_id=args.session_id,
+    )
+
+    # Phase 6: 运行对抗过滤
+    accepted_path = args.accepted_path.resolve()
+    final_path = (
+        args.final_path.resolve()
+        if args.final_path is not None
+        else accepted_path.parent / "accepted_questions_final.json"
+    )
+    await run_adversarial_filter(
+        accepted_path=accepted_path,
+        final_path=final_path,
+        agent=agent,
+        vlm=adapters.vlm,
+        trees=trees,
+        store=store,
+        filter_config=filter_config,
+        backfill=backfill,
+        session_id=args.session_id,
+    )
+    store.close()
+    logger.info("对抗过滤完成，final 已写入: {}", final_path)
+
+
 def _parse_args() -> argparse.Namespace:
     """解析命令行参数。"""
     parser = argparse.ArgumentParser(description="赛题生成工具：generate + calibrate + generate-v2")
@@ -1181,6 +1351,9 @@ def _parse_args() -> argparse.Namespace:
     # generate-v2 子命令
     _add_generate_v2_parser(subparsers)
 
+    # adversarial-filter 子命令（Phase B 后置对抗过滤）
+    _add_adversarial_filter_parser(subparsers)
+
     # generate 子命令
     gen_parser = subparsers.add_parser("generate", help="生成新题目（v1 传统模式）")
     gen_parser.add_argument(
@@ -1280,6 +1453,8 @@ def main() -> None:
         asyncio.run(_run_generate(args))
     elif args.command == "generate-v2":
         asyncio.run(_run_generate_v2(args))
+    elif args.command == "adversarial-filter":
+        asyncio.run(_run_adversarial_filter(args))
     elif args.command == "calibrate":
         _run_calibrate(args)