docs: revise Phase B plan per Codex review (C1-C3,I1-I6,M1-M3)

research-wiki/plans/2026-07-14-adversarial-question-gen-phaseB-plan.md

@@ -17,7 +17,7 @@
 - **环境**：每条 Python/pytest/ruff 命令前缀 `conda run -n Video-Tree-TRM`。示例：`conda run -n Video-Tree-TRM pytest tests/unit/test_x.py -v`。
 - **路径隔离铁律**：Phase B 只读 `accepted_questions.json`，只对 `filter_task_types` 内题型跑 agent 门。**不改** Phase A 的 accepted 语义、`on_accept`、`record_item`/`update_gates`、`load_progress`。每个改到公共文件（`run_store.py`/`pipeline_v2.py`/`strategy*.py`）的 Task 末尾须证明 11 非 AR 题型与现状字节级不变（默认参数/默认字段）。
 - **风格**：中文 docstring；禁止 `print`、禁止裸 `except`（捕获具体异常类型）；radon 无函数低于 C 级（复杂函数须拆分）。
-- **提交**：每个 Task 末尾 commit，走 `commit` skill 消息规范（英文、imperative、`<type>: <desc>`，**禁止任何 AI 署名**）。
+- **提交**：每个 Task 末尾 commit，用常规 git commit 消息（英文、imperative、`<type>: <desc>`，**禁止任何 AI 署名**）。
 - **保真**：Phase B **不迁移** `research-wiki/ARCHITECTURE.md §6` 的 12 项核心算法（建树 4 + 训练 8）。见文末保真校验。
 
 ---
@@ -182,6 +182,12 @@ def test_agent_config_change_invalidates(tmp_path):
     store = _store(tmp_path)
     store.record_verdict(**_row(stage="cheat"))
     store.invalidate_stale_config("v1_Action Recognition_0001", "cfg2")
+    # 旧 config 行必须被真正删除（不能只靠 cfg2 查空——no-op 也满足那个弱断言）
+    assert store.completed_stages("v1_Action Recognition_0001", "h1", "cfg1") == set()
+    cfg1_rows = store._conn.execute(
+        "SELECT COUNT(*) FROM adversarial_verdicts WHERE agent_config='cfg1'"
+    ).fetchone()[0]
+    assert cfg1_rows == 0
     assert store.completed_stages("v1_Action Recognition_0001", "h1", "cfg2") == set()
     store.close()
 
@@ -208,13 +214,25 @@ def test_cheat_accuracy_aggregation(tmp_path):
     store.close()
 
 
-def test_passed_question_ids(tmp_path):
+def test_final_passed_question_ids_survives_both_gates(tmp_path):
     store = _store(tmp_path)
+    # q1 太简单被作弊门剔除；q2 过两门；q3 被翻转门剔除（filtered_no_flip）
     store.record_verdict(**_row(question_id="q1", question_hash="a", stage="cheat",
                                 verdict="filtered_too_easy"))
     store.record_verdict(**_row(question_id="q2", question_hash="b", stage="cheat",
                                 verdict="passed"))
-    assert store.passed_question_ids() == {"q2"}
+    store.record_verdict(**_row(question_id="q3", question_hash="c", stage="cheat",
+                                verdict="passed"))
+    store.record_verdict(**_row(question_id="q3", question_hash="c", stage="flip_mirror",
+                                verdict="filtered_no_flip"))
+    passed = store.final_passed_question_ids(
+        {"q1": "a", "q2": "b", "q3": "c"}, "cfg1"
+    )
+    assert passed == {"q2"}  # 仅 q2：cheat=passed 且无 filtered_no_flip
+    # stale hash 不泄漏（当前 hash 不匹配旧行）
+    assert store.final_passed_question_ids({"q2": "stale"}, "cfg1") == set()
+    # stale config 不泄漏
+    assert store.final_passed_question_ids({"q2": "b"}, "cfgX") == set()
     store.close()
 ```
 
@@ -352,12 +370,42 @@ _DDL_VERDICTS_INDEXES = [
         ).fetchone()
         return float(row[0]) if row and row[0] is not None else 0.0
 
-    def passed_question_ids(self) -> set[str]:
-        """所有 verdict=passed 的 question_id 集合（final JSON 全量重建用）。"""
+    def final_passed_question_ids(
+        self, hash_by_qid: dict[str, str], agent_config: str
+    ) -> set[str]:
+        """在当前 hash+config 下通过两门的 question_id 集合（final JSON 全量重建用）。
+
+        终判规则（防 stale 泄漏）：仅当该题在 **当前 question_hash + 当前
+        agent_config** 下同时满足——存在 stage='cheat' 且 verdict='passed'
+        （agent 答错=不太简单），且不存在任何 stage 的 verdict='filtered_no_flip'
+        （未被翻转门剔除）——才计入 final-passed。stale hash / stale config 的旧行
+        因不匹配传入的 (qid, hash, config) 天然被排除，绝不泄漏进最终题库。
+
+        参数
+        ----
+        hash_by_qid : dict[str, str]
+            question_id → 当前 question_hash 映射（来自本轮 all_questions）。
+        agent_config : str
+            当前 agent 配置指纹。
+
+        返回
+        ----
+        终判 passed 的 question_id 集合。
+        """
+        passed: set[str] = set()
+        for qid, qhash in hash_by_qid.items():
             rows = self._conn.execute(
-            "SELECT DISTINCT question_id FROM adversarial_verdicts WHERE verdict='passed'"
+                "SELECT stage, verdict FROM adversarial_verdicts "
+                "WHERE question_id=? AND question_hash=? AND agent_config=?",
+                (qid, qhash, agent_config),
             ).fetchall()
-        return {r[0] for r in rows}
+            if not rows:
+                continue
+            cheat_passed = any(stage == "cheat" and verdict == "passed" for stage, verdict in rows)
+            no_flip = any(verdict == "filtered_no_flip" for _, verdict in rows)
+            if cheat_passed and not no_flip:
+                passed.add(qid)
+        return passed
 ```
 
 > 注：形参名 `round` 遮蔽内建，但与设计列名一致、仅 kwargs 传入无实际风险；若 radon/ruff 报 A002，改列语义名 `round_no` 并在 SQL 保持列名 `round`。
@@ -876,7 +924,7 @@ git commit -m "feat: add adversarial filter core decision helpers"
 4. `with HarnessLog(str(db_path), run_id) as log:` → `await run_inference(questions=..., llm=llm, tool_dispatch_fn=..., prompt_builder=..., log=log, run_id=run_id, concurrency=..., max_steps=<adversarial_agent_max_steps>, skill_mode=<mode>)`。
 5. 读预测：`await RunLogImpl(str(db_path)).get_predictions(run_id, question_ids=[...])` → list[dict]，每行含 `question_id`/`prediction`/`answer`。
 
-Phase B 不重复造装配：由 Task 11 的顶层入口注入一个 `AgentRunner` Protocol（下）。作弊门只依赖该 Protocol，便于 mock 单测。
+Phase B 不重复造装配：由 Task 10 的顶层入口注入一个 `AgentRunner` Protocol（下）。作弊门只依赖该 Protocol，便于 mock 单测。
 
 **Files:**
 - Modify: `app/question_gen/adversarial_filter.py`（`AgentRunner` Protocol + `run_cheater_gate`）
@@ -900,9 +948,10 @@ from app.question_gen.run_store import QuestionGenStore
 class _FakeAgent:
     """按 question_id → 预测字母返回的 mock AgentRunner。"""
 
-    def __init__(self, preds: dict[str, str], model: str = "m1"):
+    def __init__(self, preds: dict[str, str], model: str = "m1", skill_mode: str = "auto"):
         self._preds = preds
         self.model = model
+        self.skill_mode = skill_mode
         self.calls: list[str] = []
 
     async def predict(self, questions, *, max_steps, run_id):
@@ -952,6 +1001,27 @@ async def test_cheater_gate_resume_skips_completed(tmp_path):
                            config=cfg, round_no=0, run_id="r1")
     assert agent.calls == first  # 第二次不重跑（已有 cheat verdict）
     store.close()
+
+
+@pytest.mark.asyncio
+async def test_cheater_gate_resume_mixed_keeps_all_survivors(tmp_path):
+    """混合续跑：部分题已有 cheat verdict、部分未判——已完成的存活者不得被丢。"""
+    store = QuestionGenStore(str(tmp_path / "q.db"))
+    cfg = AdversarialFilterConfig()
+    # 第一轮：先只判 done_hard（答错=存活），落表
+    agent1 = _FakeAgent({"done_hard": "B"})  # 答错(正解 A)
+    await run_cheater_gate([_q("done_hard")], agent=agent1, store=store,
+                           config=cfg, round_no=0, run_id="r0")
+    # 第二轮：done_hard 已判 + 新题 new_hard 未判混在一起
+    agent2 = _FakeAgent({"new_hard": "C"})  # 新题答错(正解 A)=存活
+    survivors = await run_cheater_gate(
+        [_q("done_hard"), _q("new_hard")], agent=agent2, store=store,
+        config=cfg, round_no=0, run_id="r1",
+    )
+    ids = {q.question_id for q in survivors}
+    assert ids == {"done_hard", "new_hard"}  # 已完成存活者 done_hard 未被丢
+    assert agent2.calls == ["new_hard"]  # 只对未判题跑 agent
+    store.close()
 ```
 
 - [ ] **Step 2: 跑测试确认失败**
@@ -967,10 +1037,11 @@ Expected: FAIL
 class AgentRunner(Protocol):
     """完整 inference agent 试答端口 — Phase B 只依赖此接口（便于 mock）。
 
-    实现见 Task 11 的 _RealAgentRunner（复用 run_inference + RunLogImpl）。
+    实现见 Task 10 的 _RealAgentRunner（复用 run_inference + RunLogImpl）。
     """
 
     model: str
+    skill_mode: str  # 从首次定义即入 Protocol，保证 Task 6/8/10 指纹口径一致
 
     async def predict(
         self,
@@ -1006,25 +1077,30 @@ async def run_cheater_gate(
         run_id: agent 推理 run 标识。
 
     返回:
-        agent 答错的题（进翻转门）；答错题的 cheat 预测字母暂存于返回题的
-        question_id → 预测，由调用方（翻转门）复用，见 run_flip_gate。
+        agent 答错的题（进翻转门）——含"已完成续跑恢复的存活者"与"本轮新判答错者"
+        两部分合并。答错题的 cheat 预测字母已落 adversarial_verdicts 表
+        （stage='cheat'），翻转门经 `_read_cheat_prediction` 从表读取复用（不重跑）。
     """
     cfg_fp = agent_config_fingerprint(
-        skill_mode="", max_steps=config.adversarial_agent_max_steps, model=agent.model
+        skill_mode=agent.skill_mode,
+        max_steps=config.adversarial_agent_max_steps,
+        model=agent.model,
     )
     todo: list[GeneratedQuestion] = []
+    completed: list[GeneratedQuestion] = []
     for q in questions:
         h = question_hash(q)
         store.invalidate_stale_config(q.question_id, cfg_fp)
         if "cheat" in store.completed_stages(q.question_id, h, cfg_fp):
-            continue
+            completed.append(q)
+        else:
             todo.append(q)
 
-    survivors: list[GeneratedQuestion] = []
-    if not todo:
-        # 从已有 verdict 恢复 survivors（cheat 记 passed 且非 filtered_too_easy）
-        return _recover_survivors(questions, store, cfg_fp)
+    # C1: 无条件先从已完成题恢复存活者（agent 答错），再对未判题跑 agent 追加。
+    # 两者都流向翻转门——绝不因 todo 非空而丢掉已完成的存活者（混合续跑正确性）。
+    survivors: list[GeneratedQuestion] = _recover_survivors(completed, store, cfg_fp)
 
+    if todo:
         preds = await agent.predict(
             todo, max_steps=config.adversarial_agent_max_steps, run_id=run_id
         )
@@ -1040,8 +1116,8 @@ async def run_cheater_gate(
             if not correct:
                 survivors.append(q)
     logger.info(
-        "作弊门: {} 题 → 剔除太简单 {}，存活 {}",
-        len(todo), len(todo) - len(survivors), len(survivors),
+        "作弊门: {} 题（续跑复用 {}，新判 {}）→ 存活 {}",
+        len(questions), len(completed), len(todo), len(survivors),
     )
     return survivors
 ```
@@ -1076,7 +1152,15 @@ def _recover_survivors(
 Run: `conda run -n Video-Tree-TRM pytest tests/unit/test_adversarial_cheater_gate.py -v`
 Expected: PASS
 
-- [ ] **Step 5: 提交**
+- [ ] **Step 5: 验证 agent 预测形态为字母（强制，锁定"已确认的实现决策"第 2 条）**
+
+在接线 `_RealAgentRunner`（Task 10）之前不必等待——此处用最小真实链路验证 `prediction` 字段形态：跑一次真实 agent（LLM 可 mock，但须真正落一条 `predictions` 行），抽查该行 `prediction` 字段：
+- 若为**字母**（"A"/"B"/…）：`canonical_answer_text` 现有实现即可，继续。
+- 若为**选项全文**（非字母）：给 `canonical_answer_text` 补一个"按选项文本反查字母/直接按文本匹配选项"的回退分支后再继续（保证 `judge_flip` 的 canonical 比较仍成立）。
+
+此步骤是"已确认的实现决策"第 2 条（agent prediction=字母 + flip_skipped 防御回退）的落地验证锚点，两处互为交叉引用。
+
+- [ ] **Step 6: 提交**
 
 ```bash
 git add app/question_gen/adversarial_filter.py tests/unit/test_adversarial_cheater_gate.py
@@ -1193,6 +1277,16 @@ async def test_mirror_null_returns_none():
     assert mirror is None
 
 
+@pytest.mark.asyncio
+async def test_mirror_malformed_response_returns_none():
+    # 畸形 VLM 响应（连 json_repair 都救不回）不得抛异常中断本轮，须返 None
+    vlm = _FakeVLM("对不起，我无法完成这个请求。")
+    mirror = await generate_mirror_question(
+        _q(), flip_axis="before/after", vlm=vlm, material=_FakeMaterial(), session_id="s",
+    )
+    assert mirror is None
+
+
 class _FakeMaterial:
     subtitle_sentences = ["先炒后蒸"]
     frame_paths = ["/f1.jpg"]
@@ -1246,7 +1340,11 @@ def _parse_mirror(raw: str) -> dict | None:
             if s.startswith("{"):
                 content = s
                 break
+    try:
         data = json.loads(repair_json(content, return_objects=False))
+    except (json.JSONDecodeError, TypeError, ValueError):
+        # 畸形 VLM 响应绝不中断本轮：解析失败 → None（上游按 flip_skipped 处理，设计 §4.2）
+        return None
     if not isinstance(data, dict):
         return None
     mirror = data.get("mirror")
@@ -1337,27 +1435,168 @@ git commit -m "feat: add mirror question generation with canonical distinctness
 
 - [ ] **Step 1: 写失败测试（mock agent + mock VLM）**
 
-新建 `tests/unit/test_adversarial_flip_gate.py`：覆盖四种路径（不支持 flip→passed；P/Q 答案不同→passed；相同→filtered_no_flip；镜像生成 None→flip_skipped）。构造复用 Task 6/7 的 `_FakeAgent`/`_FakeVLM`；`store` 预置 P 的 cheat 预测（`record_verdict stage="cheat"`）。断言 `adversarial_verdicts` 中该题终判 verdict 与 `pair_id`（flip 分支）非空、镜像 `stage="flip_mirror"` 有独立行。示例断言骨架：
+新建 `tests/unit/test_adversarial_flip_gate.py`：**四条路径各写一个具体测试并做表/集合断言**——(a) 答案翻转→passed；(b) 答案相同→filtered_no_flip；(c) 无效/镜像失败→flip_skipped（保留，不误杀）；(d) 镜像题**不进**最终题库 + 不支持 flip 的子模式直接 passed。每例断言 `stage`、`pair_id`、**cheat 预测复用（C2：agent 不在原题 P 上被重跑）**、agent 调用计数、final-kept 集合。用 `monkeypatch` 打桩 `_rebuild_material`，把建树素材隔离掉，专测门逻辑：
 
 ```python
-@pytest.mark.asyncio
-async def test_flip_gate_different_answer_passed(tmp_path):
-    store = QuestionGenStore(str(tmp_path / "q.db"))
-    q = _q("hard", sub="temporal_reasoning_failure")
-    # 预置 P 的 cheat 预测 = "A"(蒸)
-    store.record_verdict(question_id="hard", question_hash=question_hash(q), stage="cheat",
-                         round=0, agent_prediction="A", agent_correct=False,
-                         verdict="passed", pair_id=None, agent_config=_fp())
-    agent = _FakeAgent({"hard_mirror": "A"})  # 镜像正解洗牌后 A=炒 → canonical 与 P(蒸)不同
-    vlm = _FakeVLM('{"mirror": {"question": "X 之后？", '
-                   '"options": ["A. 炒", "B. 蒸", "C. 煮", "D. 炸"], "answer": "A"}}')
-    passed = await run_flip_gate([q], agent=agent, vlm=vlm, store=store,
-                                 trees={"v1": _FakeTree()}, config=AdversarialFilterConfig(),
-                                 round_no=0, run_id="r0", session_id="s")
-    assert {x.question_id for x in passed} == {"hard"}
-```
+"""翻转门四路径：passed / filtered_no_flip / flip_skipped / 镜像不入库。"""
 
-（其余三例类比：镜像 agent 选到 canonical=蒸 → filtered_no_flip；VLM 返回 `{"mirror": null}` → flip_skipped 但仍 passed 保留，因退回只经作弊门；不支持 flip 的子模式 → 直接 passed 不跑 VLM/agent。测试须断言这些语义。）
+import json
+
+import pytest
+from core.types import GeneratedQuestion, LLMResponse
+
+from app.question_gen.adversarial_config import AdversarialFilterConfig
+from app.question_gen.adversarial_filter import (
+    agent_config_fingerprint,
+    question_hash,
+    run_flip_gate,
+    write_final_bank,
+)
+from app.question_gen.run_store import QuestionGenStore
+
+
+class _FakeAgent:
+    """复用 Task 6 语义；带 skill_mode 属性（AgentRunner Protocol 要求）。"""
+
+    def __init__(self, preds, model="m1", skill_mode="auto"):
+        self._preds = preds
+        self.model = model
+        self.skill_mode = skill_mode
+        self.calls: list[str] = []
+
+    async def predict(self, questions, *, max_steps, run_id):
+        self.calls.extend(q.question_id for q in questions)
+        return {q.question_id: self._preds.get(q.question_id) for q in questions}
+
+
+class _FakeVLM:
+    def __init__(self, content):
+        self._content = content
+
+    async def chat_with_images(self, messages, images, *, session_id=None, parent_call_id=None):
+        return LLMResponse(
+            content=self._content, thinking="", model="fake", provider="fake",
+            prompt_tokens=0, completion_tokens=0, latency_ms=0,
+            ttft_ms=None, max_inter_token_ms=None, cache_hit=False, call_id="c",
+        )
+
+
+class _FakeMaterial:
+    subtitle_sentences = ["先炒后蒸"]
+    frame_paths = ["/f1.jpg"]
+
+
+def _q(qid, sub="temporal_reasoning_failure"):
+    return GeneratedQuestion(
+        question_id=qid, video_id="v1", task_type="Action Recognition",
+        question="X 之前做了什么？", options=("A. 蒸", "B. 炒", "C. 煮", "D. 炸"),
+        answer="A", source_nodes=("n1",), difficulty="hard", sub_pattern=sub,
+    )
+
+
+def _fp():
+    return agent_config_fingerprint(skill_mode="auto", max_steps=40, model="m1")
+
+
+@pytest.fixture(autouse=True)
+def _stub_material(monkeypatch):
+    """隔离建树素材重建，直接给镜像生成喂假素材。"""
+    monkeypatch.setattr(
+        "app.question_gen.adversarial_filter._rebuild_material",
+        lambda tree, source_nodes: _FakeMaterial(),
+    )
+
+
+def _preset_cheat(store, q, pred="A"):
+    """预置作弊门 P 预测行（翻转门须复用它，不重跑 agent）。"""
+    store.record_verdict(question_id=q.question_id, question_hash=question_hash(q),
+                         stage="cheat", round=0, agent_prediction=pred, agent_correct=False,
+                         verdict="passed", pair_id=None, agent_config=_fp())
+
+
+@pytest.mark.asyncio
+async def test_flip_gate_answer_flips_passed(tmp_path):
+    store = QuestionGenStore(str(tmp_path / "q.db"))
+    q = _q("hard")
+    _preset_cheat(store, q, pred="A")  # P canonical="蒸"
+    agent = _FakeAgent({"hard_mirror": "A"})  # 镜像洗牌后 A=炒 → canonical≠蒸
+    vlm = _FakeVLM(json.dumps({"mirror": {"question": "X 之后？",
+        "options": ["A. 炒", "B. 蒸", "C. 煮", "D. 炸"], "answer": "A"}}, ensure_ascii=False))
+    kept = await run_flip_gate([q], agent=agent, vlm=vlm, store=store,
+                               trees={"v1": object()}, config=AdversarialFilterConfig(),
+                               round_no=0, run_id="r0", session_id="s")
+    assert {x.question_id for x in kept} == {"hard"}
+    assert agent.calls == ["hard_mirror"]  # C2: 原题 P 未被重跑，只跑镜像
+    cheat = store._conn.execute(
+        "SELECT verdict FROM adversarial_verdicts WHERE question_id='hard' AND stage='cheat'"
+    ).fetchone()[0]
+    assert cheat == "passed"
+    mrow = store._conn.execute(
+        "SELECT verdict, pair_id FROM adversarial_verdicts WHERE stage='flip_mirror'"
+    ).fetchone()
+    assert mrow[0] == "passed" and mrow[1]  # 镜像独立行 + pair_id 非空
+    store.close()
+
+
+@pytest.mark.asyncio
+async def test_flip_gate_same_answer_filtered(tmp_path):
+    store = QuestionGenStore(str(tmp_path / "q.db"))
+    q = _q("stick")
+    _preset_cheat(store, q, pred="A")  # P canonical="蒸"
+    agent = _FakeAgent({"stick_mirror": "A"})  # 镜像 A=蒸 → canonical 与 P 相同
+    vlm = _FakeVLM(json.dumps({"mirror": {"question": "X 之后？",
+        "options": ["A. 蒸", "B. 炒", "C. 煮", "D. 炸"], "answer": "B"}}, ensure_ascii=False))
+    kept = await run_flip_gate([q], agent=agent, vlm=vlm, store=store,
+                               trees={"v1": object()}, config=AdversarialFilterConfig(),
+                               round_no=0, run_id="r0", session_id="s")
+    assert kept == []  # 未随问题翻转 → 剔除
+    cheat = store._conn.execute(
+        "SELECT verdict FROM adversarial_verdicts WHERE question_id='stick' AND stage='cheat'"
+    ).fetchone()[0]
+    assert cheat == "filtered_no_flip"  # cheat 行被改写 → final 不含它
+    store.close()
+
+
+@pytest.mark.asyncio
+async def test_flip_gate_invalid_mirror_skipped_but_kept(tmp_path):
+    store = QuestionGenStore(str(tmp_path / "q.db"))
+    q = _q("murky")
+    _preset_cheat(store, q, pred="A")
+    agent = _FakeAgent({})  # 镜像造不出 → agent 不该被调用
+    vlm = _FakeVLM('{"mirror": null}')
+    kept = await run_flip_gate([q], agent=agent, vlm=vlm, store=store,
+                               trees={"v1": object()}, config=AdversarialFilterConfig(),
+                               round_no=0, run_id="r0", session_id="s")
+    assert {x.question_id for x in kept} == {"murky"}  # 退回只经作弊门，保留不误杀
+    assert agent.calls == []  # 镜像 None → 未跑 agent
+    cheat = store._conn.execute(
+        "SELECT verdict FROM adversarial_verdicts WHERE question_id='murky' AND stage='cheat'"
+    ).fetchone()[0]
+    assert cheat == "passed"
+    store.close()
+
+
+@pytest.mark.asyncio
+async def test_flip_gate_mirror_excluded_and_unsupported_passes(tmp_path):
+    store = QuestionGenStore(str(tmp_path / "q.db"))
+    q = _q("hard")                                        # 支持 flip
+    npq = _q("plain", sub="premature_evidence_anchoring")  # 不支持 flip
+    _preset_cheat(store, q, pred="A")
+    _preset_cheat(store, npq, pred="B")
+    agent = _FakeAgent({"hard_mirror": "A"})
+    vlm = _FakeVLM(json.dumps({"mirror": {"question": "X 之后？",
+        "options": ["A. 炒", "B. 蒸", "C. 煮", "D. 炸"], "answer": "A"}}, ensure_ascii=False))
+    kept = await run_flip_gate([q, npq], agent=agent, vlm=vlm, store=store,
+                               trees={"v1": object()}, config=AdversarialFilterConfig(),
+                               round_no=0, run_id="r0", session_id="s")
+    assert {x.question_id for x in kept} == {"hard", "plain"}  # 不支持 flip 直接 passed
+    assert agent.calls == ["hard_mirror"]  # 不支持 flip 的题不跑 agent/VLM
+    out = tmp_path / "final.json"
+    write_final_bank(out, store, {"hard": q, "plain": npq}, _fp())
+    ids = [d["question_id"] for d in json.loads(out.read_text(encoding="utf-8"))]
+    assert "hard_mirror" not in ids and set(ids) == {"hard", "plain"}  # 镜像不入题库
+    store.close()
+```
 
 - [ ] **Step 2: 跑测试确认失败**
 
@@ -1392,7 +1631,9 @@ async def run_flip_gate(
     from app.question_gen.strategy_action_recognition import _AR_PATTERN_BY_NAME
 
     cfg_fp = agent_config_fingerprint(
-        skill_mode="", max_steps=config.adversarial_agent_max_steps, model=agent.model
+        skill_mode=agent.skill_mode,
+        max_steps=config.adversarial_agent_max_steps,
+        model=agent.model,
     )
     kept: list[GeneratedQuestion] = []
     for q in survivors:
@@ -1401,8 +1642,8 @@ async def run_flip_gate(
             kept.append(q)  # cheat 已记 passed，无需改写
             continue
         decision, mirror_pred = await _judge_one_flip(
-            q, sp.flip_axis, agent=agent, vlm=vlm, trees=trees,
-            config=config, run_id=run_id, session_id=session_id,
+            q, sp.flip_axis, agent=agent, vlm=vlm, trees=trees, store=store,
+            cfg_fp=cfg_fp, config=config, run_id=run_id, session_id=session_id,
         )
         pair_id = f"{q.question_id}::{round_no}"
         _persist_flip(store, q, decision, mirror_pred, round_no, cfg_fp, pair_id)
@@ -1422,11 +1663,18 @@ async def _judge_one_flip(
     agent: AgentRunner,
     vlm: VLMProvider,
     trees: dict[str, TreeIndex],
+    store: QuestionGenStore,
+    cfg_fp: str,
     config: AdversarialFilterConfig,
     run_id: str,
     session_id: str,
 ) -> tuple[FlipDecision, str | None]:
-    """跑单题翻转判定，返回 (decision, 镜像预测字母)。"""
+    """跑单题翻转判定，返回 (decision, 镜像预测字母)。
+
+    原题 P 预测**只从 adversarial_verdicts 表读作弊门落的行**（不重跑 agent），
+    故 `store` 与 `cfg_fp` 必传（C2：按 (question_id, question_hash, stage='cheat',
+    agent_config) 定位那条预测）。
+    """
     tree = trees.get(q.video_id)
     if tree is None or flip_axis is None:
         return FlipDecision.FLIP_SKIPPED, None
@@ -1440,7 +1688,7 @@ async def _judge_one_flip(
         [mirror], max_steps=config.adversarial_agent_max_steps, run_id=f"{run_id}_mirror"
     )
     q_pred = preds.get(mirror.question_id)
-    p_pred = _read_cheat_prediction(q)  # 复用作弊门 P 预测
+    p_pred = _read_cheat_prediction(store, q, cfg_fp)  # 复用作弊门 P 预测（不重跑）
     p_text = canonical_answer_text(q, p_pred)
     q_text = canonical_answer_text(mirror, q_pred)
     return judge_flip(p_text=p_text, q_text=q_text), q_pred
@@ -1449,12 +1697,16 @@ async def _judge_one_flip(
 `_read_cheat_prediction` 从表读 P 的 cheat 预测；`_persist_flip` 写 flip_original（复用 P 预测的原题终判 verdict）+ flip_mirror（镜像预测）两条 stage 行，并把原题 cheat 行的 verdict 依 decision 改写（passed 保持 passed；filtered_no_flip 改判剔除；flip_skipped 保持 passed）。这两个辅助各 <15 行，直接读/写 `store._conn` 或调 `store.record_verdict`。实现时确保：
 
 ```python
-def _read_cheat_prediction(q: GeneratedQuestion) -> str | None:
+def _read_cheat_prediction(
+    store: QuestionGenStore, q: GeneratedQuestion, cfg_fp: str
+) -> str | None:
     ...  # SELECT agent_prediction FROM adversarial_verdicts
-        # WHERE question_id=? AND question_hash=? AND stage='cheat'
+        # WHERE question_id=? AND question_hash=question_hash(q)
+        #   AND stage='cheat' AND agent_config=cfg_fp
+        # 只读表、绝不重跑 agent（C2：P 预测来自作弊门落库结果）
 ```
 
-`_persist_flip` 用 `store.record_verdict` 写 stage="flip_mirror"（agent_prediction=mirror_pred, verdict=decision.value, pair_id）与 stage="flip_original"（verdict=decision.value, pair_id）。**同时**：若 decision 为 FILTERED_NO_FLIP，改写 cheat 行 verdict→`filtered_no_flip`（保证 `passed_question_ids` 不含它）；passed/flip_skipped 时 cheat 行保持 `passed`。
+`_persist_flip` 用 `store.record_verdict` 写 stage="flip_mirror"（agent_prediction=mirror_pred, verdict=decision.value, pair_id）与 stage="flip_original"（verdict=decision.value, pair_id）。**同时**：若 decision 为 FILTERED_NO_FLIP，改写 cheat 行 verdict→`filtered_no_flip`（保证 `final_passed_question_ids` 不含它——终判规则也独立排除任何 `filtered_no_flip` 行，双保险）；passed/flip_skipped 时 cheat 行保持 `passed`。
 
 - [ ] **Step 4: 跑测试确认通过**
 
@@ -1481,29 +1733,43 @@ git commit -m "feat: add pairwise flip gate reusing P prediction and mirror agen
 - [ ] **Step 1: 写失败测试**
 
 新建 `tests/unit/test_adversarial_iteration.py`，覆盖：
-- `write_final_bank`：全量重写（tmp+os.replace）、内容仅含 `passed` 题、可从空 verdicts 表重建为 `[]`。
+- `write_final_bank`：全量重写（tmp+os.replace）、内容仅含**当前 hash+config 下过两门**的题、可从空 verdicts 表重建为 `[]`、stale-config 旧 passed 行被排除（C3）。
 - 缺额计算：`deficit = target - passed`；deficit≤0 或 round≥max → 停止（用假的 backfill 回调计数验证调用次数）。
 - 难度报告：agent 正确率 > 阈值 → `caplog` 捕获 warning。
 
 ```python
 def test_write_final_bank_only_passed(tmp_path):
     store = QuestionGenStore(str(tmp_path / "q.db"))
-    store.record_verdict(question_id="q1", question_hash="a", stage="cheat", round=0,
-                         agent_prediction="B", agent_correct=False, verdict="passed",
-                         pair_id=None, agent_config="c")
-    store.record_verdict(question_id="q2", question_hash="b", stage="cheat", round=0,
-                         agent_prediction="A", agent_correct=True,
+    q1, q2 = _q("q1"), _q("q2")
+    # question_hash 必须与 write_final_bank 内部按 all_questions 计算的一致，否则被当 stale 排除
+    store.record_verdict(question_id="q1", question_hash=question_hash(q1), stage="cheat",
+                         round=0, agent_prediction="B", agent_correct=False,
+                         verdict="passed", pair_id=None, agent_config="c")
+    store.record_verdict(question_id="q2", question_hash=question_hash(q2), stage="cheat",
+                         round=0, agent_prediction="A", agent_correct=True,
                          verdict="filtered_too_easy", pair_id=None, agent_config="c")
-    all_qs = {"q1": _q("q1"), "q2": _q("q2")}
+    all_qs = {"q1": q1, "q2": q2}
     out = tmp_path / "accepted_questions_final.json"
-    write_final_bank(out, store, all_qs)
+    write_final_bank(out, store, all_qs, "c")  # 显式传当前 agent_config
     data = json.loads(out.read_text(encoding="utf-8"))
     assert [d["question_id"] for d in data] == ["q1"]
 
 
+def test_write_final_bank_excludes_stale_config(tmp_path):
+    store = QuestionGenStore(str(tmp_path / "q.db"))
+    q1 = _q("q1")
+    # 旧 config 下的 passed 行不得泄漏进 final（C3）
+    store.record_verdict(question_id="q1", question_hash=question_hash(q1), stage="cheat",
+                         round=0, agent_prediction="B", agent_correct=False,
+                         verdict="passed", pair_id=None, agent_config="OLD")
+    out = tmp_path / "accepted_questions_final.json"
+    write_final_bank(out, store, {"q1": q1}, "NEW")
+    assert json.loads(out.read_text(encoding="utf-8")) == []
+
+
 def test_difficulty_warns_above_threshold(tmp_path, caplog):
     store = QuestionGenStore(str(tmp_path / "q.db"))
-    for i in range(4):  # 3 对 1 错 = 0.75... 设 3 对 => 0.75；用 4 对 => 1.0 > 0.85
+    for i in range(4):  # 4 题全对 = 正确率 1.0 > 阈值 0.85 → 必触发告警
         store.record_verdict(question_id=f"q{i}", question_hash=str(i), stage="cheat",
                              round=0, agent_prediction="A", agent_correct=True,
                              verdict="filtered_too_easy", pair_id=None, agent_config="c")
@@ -1526,24 +1792,27 @@ def write_final_bank(
     final_path: Path,
     store: QuestionGenStore,
     all_questions: dict[str, GeneratedQuestion],
+    agent_config: str,
 ) -> int:
     """全量重写 accepted_questions_final.json（tmp+os.replace 原子）。
 
-    内容 = store 中所有 verdict=passed 的题（可随时从 verdicts 表重建）。
+    内容 = 在**当前 question_hash + 当前 agent_config** 下通过两门（cheat=passed 且
+    无 filtered_no_flip）的题。stale-config / stale-hash 的旧 passed 行绝不泄漏（C3）。
 
     参数:
         final_path: 输出路径。
         store: verdict 来源。
-        all_questions: question_id → GeneratedQuestion（重建 payload）。
+        all_questions: question_id → GeneratedQuestion（同时提供当前 hash 与 payload）。
+        agent_config: 当前 agent 配置指纹（终判过滤维度）。
 
     返回:
         写入的题数。
     """
-    passed_ids = store.passed_question_ids()
+    hash_by_qid = {qid: question_hash(q) for qid, q in all_questions.items()}
+    passed_ids = store.final_passed_question_ids(hash_by_qid, agent_config)
     entries = [
         _question_to_final_entry(all_questions[qid])
         for qid in sorted(passed_ids)
-        if qid in all_questions
     ]
     final_path.parent.mkdir(parents=True, exist_ok=True)
     tmp = final_path.with_suffix(".tmp")
@@ -1577,7 +1846,7 @@ def _report_difficulty(store: QuestionGenStore, *, round_no: int, threshold: flo
     return acc
 ```
 
-`run_adversarial_rounds` 编排迭代（用 Protocol 化的 backfill 回调，便于测；真实实现由 Task 11 注入）：
+`run_adversarial_rounds` 编排迭代（用 Protocol 化的 backfill 回调，便于测；真实实现由 Task 10 注入）：
 
 ```python
 async def run_adversarial_rounds(
@@ -1599,10 +1868,16 @@ async def run_adversarial_rounds(
         initial_questions: 首轮 AR 题（来自 accepted_questions.json 过滤）。
         target: 目标 passed 题数（缺额 = target - passed）。
         backfill: 补生成回调 (deficit, round, used_node_ids, embed_pool, seq_offset)
-                  -> 新增题列表；由 Task 11 用 run_pipeline_v2 实现，测试可 mock。
+                  -> 新增题列表；由 Task 10 用 run_pipeline_v2 实现，测试可 mock。
     """
+    cfg_fp = agent_config_fingerprint(
+        skill_mode=agent.skill_mode,
+        max_steps=config.adversarial_agent_max_steps,
+        model=agent.model,
+    )
     all_questions: dict[str, GeneratedQuestion] = {q.question_id: q for q in initial_questions}
     pending = list(initial_questions)
+    passed_now = 0
     for round_no in range(config.adversarial_max_rounds):
         survivors = await run_cheater_gate(
             pending, agent=agent, store=store, config=config,
@@ -1613,7 +1888,7 @@ async def run_adversarial_rounds(
             config=config, round_no=round_no, run_id=f"{session_id}_flip_{round_no}",
             session_id=session_id,
         )
-        passed_now = write_final_bank(final_path, store, all_questions)
+        passed_now = write_final_bank(final_path, store, all_questions, cfg_fp)
         _report_difficulty(
             store, round_no=round_no, threshold=config.difficulty_warn_threshold
         )
@@ -1624,7 +1899,7 @@ async def run_adversarial_rounds(
         for q in new_qs:
             all_questions[q.question_id] = q
         pending = new_qs  # 只对新补的题重新过滤
-    logger.info("对抗过滤结束: final={} 题", len(store.passed_question_ids()))
+    logger.info("对抗过滤结束: final={} 题", passed_now)
 ```
 
 补 `BackfillFn` Protocol：
@@ -1722,22 +1997,54 @@ class _RealAgentRunner:
         return {r["question_id"]: r["prediction"] for r in rows}
 ```
 
-> 指纹用 `agent_config_fingerprint(skill_mode=self._skill_mode, max_steps=..., model=self.model)`——注意 Task 6/8 现用 `skill_mode=""` 占位。**统一**：把 `run_cheater_gate`/`run_flip_gate` 的指纹计算改为接收 agent 暴露的 `skill_mode`（给 `AgentRunner` Protocol 加 `skill_mode: str` 属性，`_FakeAgent` 补一个默认值）。实现本 Task 时一并修正 Task 6/8 的 `skill_mode=""` 为 `agent.skill_mode`，并更新那两个测试的 `_FakeAgent`（加 `skill_mode="auto"`）。
+> 指纹用 `agent_config_fingerprint(skill_mode=self._skill_mode, max_steps=..., model=self.model)`。`skill_mode` 自 Task 6 起即为 `AgentRunner` Protocol 的属性（`run_cheater_gate`/`run_flip_gate`/`run_adversarial_rounds` 全用 `agent.skill_mode` 计算指纹），故 Task 6/8/10 指纹口径天然一致，本 Task **无需回改**前序任务——`_RealAgentRunner` 只要如实暴露 `self.skill_mode` 即可。
 
-- [ ] **Step 4: 实现 `run_adversarial_filter` 入口**
+- [ ] **Step 4: 真实装配 smoke 测试（I6：走真实 predict 链路，非全 mock）**
 
-组装真实 `backfill`（闭包捕获 `run_pipeline_v2` 所需依赖：trees/vlm/llm/embed_fn/store/pipeline_config；每轮算 `seq_offset`=已用最大 seq、传 `initial_used_node_ids`=已用 source_nodes 并集、`initial_embed_pool`=已接受题 embedding），读 `accepted_questions.json` 过滤 `filter_task_types`，`target`=首轮 AR 题数（见待确认项），调 `run_adversarial_rounds`。函数签名接收已装配好的 `agent`/`vlm`/`trees`/`store`/两个 config/路径，保持可测。
+前述 Step 1 的 e2e 用 mock `AgentRunner`，证明门编排但**不覆盖真实装配接线**。追加一个最小 smoke，验证 `_RealAgentRunner.predict → HarnessLog → run_inference → get_predictions` 这条真实链路能跑通、预测确实经 `predictions` 表落库再读回（**LLM 可 mock**，但路径必须真穿过 `_RealAgentRunner.predict` 与 predictions 表，不得再用假 runner 短路）。加到 `tests/integration/test_adversarial_filter_e2e.py`：
 
-- [ ] **Step 5: 加 CLI 子命令**
+```python
+@pytest.mark.asyncio
+async def test_real_agent_runner_predict_roundtrips_predictions(tmp_path):
+    """真实装配 smoke：predict 经 run_inference 落 predictions 表再读回（LLM mock）。"""
+    from app.question_gen.adversarial_filter import _RealAgentRunner
+
+    llm = _MockLLM(answer="B")  # 最小 mock：让 agent 一步产出 {"answer": "B"}
+    router = _build_real_router(tmp_path)  # 复用 main._build_adapters + InferenceDepsRouter（真实）
+    runner = _RealAgentRunner(
+        llm=llm, tool_dispatch_fn=router.create_dispatch(),
+        prompt_builder=router.create_prompt_builder(),
+        db_path=str(tmp_path / "harness.db"), concurrency=1,
+        skill_mode="auto", model="mock",
+    )
+    preds = await runner.predict([_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"
+```
+
+> `_MockLLM`/`_build_real_router` 是本测试的最小真实装配辅助（router 用真实 `InferenceDepsRouter`，仅 LLM 打桩）。若真实 agent 一步无法稳定产出答案，允许把 `max_steps` 调到能收敛的最小值；关键是**路径真实**，不是断言具体答案的稳定性。
+
+- [ ] **Step 5: 实现 `run_adversarial_filter` 入口**
+
+组装真实 `backfill`（闭包捕获 `run_pipeline_v2` 所需依赖：trees/vlm/llm/embed_fn/store/pipeline_config；每轮算 `seq_offset`=已用最大 seq、传 `initial_used_node_ids`=已用 source_nodes 并集、`initial_embed_pool`=已接受题 embedding），读 `accepted_questions.json` 过滤 `filter_task_types`，`target`=首轮 AR 题数（见"已确认的实现决策"第 1 条），调 `run_adversarial_rounds`。函数签名接收已装配好的 `agent`/`vlm`/`trees`/`store`/两个 config/路径，保持可测。
+
+> **缺额驱动 per_type（I1）**：`PipelineConfig` 是 frozen dataclass，backfill 闭包**不得**原地改字段，须 `import dataclasses` 后用 `run_cfg = dataclasses.replace(pipeline_config, per_type=deficit)` 生成一份新 config 再传给 `run_pipeline_v2`（其余字段继承 ar30 原配置）。补生成返回后**断言** `assert len(new_qs) == deficit`（`filter_task_types` 仅 AR 时补的即 `deficit` 道 AR 题）——数量对不上即 backfill 契约被破坏，直接报错而非静默继续。
+
+- [ ] **Step 6: 加 CLI 子命令**
 
 `tools/generate_questions.py` 加 `adversarial-filter` 子命令：装配 adapters（`main._build_adapters` 同款：`InfraSettings()`+YAML embed 段）、`InferenceDepsRouter`（同 `main.py` 参数）、`QuestionGenStore`、加载 trees（复用 Phase 6 逻辑，含帧路径绝对化），`_RealAgentRunner`，调 `run_adversarial_filter`。
 
-- [ ] **Step 6: 跑测试确认通过**
+- [ ] **Step 7: 跑测试确认通过**
 
 Run: `conda run -n Video-Tree-TRM pytest tests/integration/test_adversarial_filter_e2e.py -v`
 Expected: PASS
 
-- [ ] **Step 7: 提交**
+- [ ] **Step 8: 提交**
 
 ```bash
 git add app/question_gen/adversarial_filter.py tools/generate_questions.py tests/integration/test_adversarial_filter_e2e.py
@@ -1799,7 +2106,7 @@ git commit -m "docs: register Phase B plan in research wiki"
 
 **Placeholder 扫描：** 每个 code Step 均为可直接落地的真实代码（DDL、方法体、prompt 全文、prompt 解析、判定分支）。仅 Task 8 的 `_read_cheat_prediction`/`_persist_flip` 与 Task 10 的 `run_adversarial_filter`/CLI 给出精确契约与 SQL 语义而非逐字节代码（因 <15 行且依赖前序 Task 的已定型接口）——非占位符，是有明确输入输出的收尾实现。
 
-**类型一致性（跨 Task）：** `GeneratedQuestion.sub_pattern`（Phase A 已落）贯穿 Task 1/5/7/9；`AgentRunner` Protocol（`model`/`skill_mode`/`predict`）在 Task 6 定义、Task 8/10 复用（Task 10 Step 3 统一 `skill_mode` 指纹）；`FlipDecision` 枚举 Task 5 定义、Task 8 消费；`AdversarialFilterConfig` Task 4 定义、Task 6/8/9/10 消费；`question_hash`/`agent_config_fingerprint` Task 5 定义、Task 6/8 消费；verdict 四枚举值 (`passed`/`filtered_too_easy`/`filtered_no_flip`/`flip_skipped`) 表约束（Task 2）与写入点（Task 6/8）一致。
+**类型一致性（跨 Task）：** `GeneratedQuestion.sub_pattern`（Phase A 已落）贯穿 Task 1/5/7/9；`AgentRunner` Protocol（`model`/`skill_mode`/`predict`）在 Task 6 首次定义即含 `skill_mode` 属性、Task 8/10 复用（指纹口径自 Task 6 起一致，无需后置统一）；`FlipDecision` 枚举 Task 5 定义、Task 8 消费；`AdversarialFilterConfig` Task 4 定义、Task 6/8/9/10 消费；`question_hash`/`agent_config_fingerprint` Task 5 定义、Task 6/8 消费；verdict 四枚举值 (`passed`/`filtered_too_easy`/`filtered_no_flip`/`flip_skipped`) 表约束（Task 2）与写入点（Task 6/8）一致。
 
 **核心算法保真（N/A）：** Phase B 全部改动局限于 question_gen 后置过滤层（新模块 + 新表 + 3 个可选 pipeline 参数 + SubPattern 2 字段），**不涉及** `research-wiki/ARCHITECTURE.md §6` 的 12 项核心算法（建树 4 + 训练 8）。作弊门/翻转门复用既有 `run_inference`（AgentLoop 完整树搜索）**未改其内部**。**保真校验不适用。**