iomgaa/Video-Tree-TRM5
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
abc9c097d2695365864fe09f83f7985f54608c6f
Video-Tree-TRM5/research-wiki/plans/2026-07-14-adversarial-question-gen-phaseB-plan.md
T

74 KiB
Raw Blame History

Adversarial Question-Gen Phase B Implementation Plan

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

Goal: 在 Phase A grounded 单题产物 accepted_questions.json 之上，加一层独立后置过滤：用完整 inference agent 跑作弊者门（agent 秒杀=太简单，剔除）与配对翻转门（agent 答案必须随问题翻转，否则揪出偏好蒙答），产出 accepted_questions_final.json。Phase A 状态机零改动。

Architecture: Phase B 是 additive 后置层，新模块 app/question_gen/adversarial_filter.py。路径隔离靠 filter 层配置 filter_task_types（默认 [Action Recognition]）——只有该配置内的题型走 agent 门；11 个非 AR 题型与 Phase A 的 on_accept/record_item/update_gates/load_progress 状态机完全不触及。过滤进度存独立 adversarial_verdicts 表，与 Phase A final_status 正交。补生成通过给 run_pipeline_v2 新增三个可选参数（不传=现状）实现，不改 11 题型行为。

Tech Stack: Python 3.11、asyncio、run_inference（完整 AgentLoop 树搜索）、InferenceDepsRouter、HarnessLog/RunLogImpl、VLMProvider（chat_with_images）、sqlite3（幂等 ALTER TABLE）、json_repair、pytest。全部命令在 conda 环境 Video-Tree-TRM 内执行。

设计来源（权威）： research-wiki/designs/2026-07-14-adversarial-question-gen-phaseB-design.md（读全）。

前置约定（所有任务通用）

  • 环境：每条 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 署名）。
  • 保真：Phase B 不迁移 research-wiki/ARCHITECTURE.md §6 的 12 项核心算法（建树 4 + 训练 8）。见文末保真校验。

Task 1: SubPattern 加 supports_flip/flip_axis + 声明 2 个 AR 子模式（纯数据）

Phase B 按题的 sub_pattern 查其 SubPattern 的 supports_flip/flip_axis 决定是否走翻转门。默认值保证 11 非 AR + 4 个非 flip 的 AR 子模式不受影响。

Files:

  • Modify: app/question_gen/strategy.py（SubPattern 加两字段）

  • Modify: app/question_gen/strategy_action_recognition.py（_TEMPORAL_REASONING_FAILURE、_CROSS_SEGMENT_ENTITY_TRACKING 设 flip）

  • Test: tests/unit/test_sub_pattern_flip.py（新建）

  • Step 1: 写失败测试

新建 tests/unit/test_sub_pattern_flip.py：

"""SubPattern.supports_flip/flip_axis 默认值 + AR 两个子模式的翻转声明。"""

from app.question_gen.strategy import SubPattern
from app.question_gen.strategy_action_recognition import AR_SUB_PATTERNS

_FLIP_EXPECTED = {
    "temporal_reasoning_failure": "before/after",
    "cross_segment_entity_tracking": "first/last",
}


def test_sub_pattern_defaults_no_flip():
    sp = SubPattern(
        name="x", weight=1.0, sampling_level_override=None,
        constraint_override=None, instruction="i",
    )
    assert sp.supports_flip is False
    assert sp.flip_axis is None


def test_ar_flip_declarations():
    by_name = {sp.name: sp for sp in AR_SUB_PATTERNS}
    for name, axis in _FLIP_EXPECTED.items():
        assert by_name[name].supports_flip is True, name
        assert by_name[name].flip_axis == axis, name


def test_other_ar_sub_patterns_keep_defaults():
    for sp in AR_SUB_PATTERNS:
        if sp.name in _FLIP_EXPECTED:
            continue
        assert sp.supports_flip is False, sp.name
        assert sp.flip_axis is None, sp.name
  • Step 2: 跑测试确认失败

Run: conda run -n Video-Tree-TRM pytest tests/unit/test_sub_pattern_flip.py -v Expected: FAIL（SubPattern 无 supports_flip）

  • Step 3: 改 SubPattern 数据类

app/question_gen/strategy.py，SubPattern 末尾追加两字段（保持 frozen，带默认值）：

    positive_examples: list[dict] = field(default_factory=list)
    negative_examples: list[dict] = field(default_factory=list)
    distractor_rules: str = ""
    supports_flip: bool = False
    flip_axis: str | None = None

docstring 属性列表补两行：supports_flip: 是否支持配对翻转门（Phase B 用，默认 False）。 / flip_axis: 翻转轴（"before/after" | "first/last"），None 表示不翻转。

  • Step 4: 声明 2 个 AR 子模式的翻转轴

app/question_gen/strategy_action_recognition.py：_TEMPORAL_REASONING_FAILURE = SubPattern(...) 的构造末尾（distractor_rules=(...) 之后）加：

    supports_flip=True,
    flip_axis="before/after",

_CROSS_SEGMENT_ENTITY_TRACKING = SubPattern(...) 的构造末尾加：

    supports_flip=True,
    flip_axis="first/last",

其余 4 个 AR 子模式（_PREMATURE_EVIDENCE_ANCHORING/_SEMANTIC_RIGIDITY/_FINE_GRAINED_VISUAL_ACTION/_EVIDENCE_GAP_CONFABULATION）不动（用默认）。

  • Step 5: 跑测试确认通过

Run: conda run -n Video-Tree-TRM pytest tests/unit/test_sub_pattern_flip.py -v Expected: PASS

  • Step 6: 回归 AR 策略既有测试（默认字段不破坏 11 题型）

Run: conda run -n Video-Tree-TRM pytest tests/unit/ -k "action or strategy or families or sub_pattern" -v Expected: PASS

  • Step 7: 提交
git add app/question_gen/strategy.py app/question_gen/strategy_action_recognition.py tests/unit/test_sub_pattern_flip.py
git commit -m "feat: declare supports_flip/flip_axis on flippable AR sub-patterns"

Task 2: adversarial_verdicts 表 + Store 方法（run_store.py，幂等/续跑/聚合）

新表存 agent 门的每次试答结果，支持按 (question_id, question_hash, stage) 续跑、按 agent_config 变化作废、聚合 agent 正确率。仿 sub_pattern/selector_scores 的幂等 DDL 风格。

Files:

  • Modify: app/question_gen/run_store.py（新增 _DDL_VERDICTS + 索引 + 4 个方法）

  • Modify: research-wiki/schemas/question-gen-items.md（登记新表；若无该 schema 则新建 research-wiki/schemas/adversarial-verdicts.md）

  • Test: tests/unit/test_adversarial_verdicts_store.py（新建）

  • Step 1: 写失败测试

新建 tests/unit/test_adversarial_verdicts_store.py：

"""adversarial_verdicts 表：写入 / 续跑查询 / agent_config 作废 / 正确率聚合。"""

from app.question_gen.run_store import QuestionGenStore


def _store(tmp_path):
    return QuestionGenStore(str(tmp_path / "q.db"))


def _row(**kw):
    base = dict(
        question_id="v1_Action Recognition_0001", round=0, stage="cheat",
        question_hash="h1", agent_prediction="B", agent_correct=False,
        verdict="passed", pair_id=None, agent_config="cfg1",
    )
    base.update(kw)
    return base


def test_table_created(tmp_path):
    store = _store(tmp_path)
    cols = {r[1] for r in store._conn.execute("PRAGMA table_info(adversarial_verdicts)")}
    assert {"question_id", "round", "stage", "question_hash", "agent_prediction",
            "agent_correct", "verdict", "pair_id", "agent_config"} <= cols
    store.close()


def test_record_and_resume_lookup(tmp_path):
    store = _store(tmp_path)
    store.record_verdict(**_row(stage="cheat"))
    done = store.completed_stages("v1_Action Recognition_0001", "h1", "cfg1")
    assert done == {"cheat"}
    # 不同 hash 视为未完成
    assert store.completed_stages("v1_Action Recognition_0001", "h2", "cfg1") == set()
    store.close()


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")
    assert store.completed_stages("v1_Action Recognition_0001", "h1", "cfg2") == set()
    store.close()


def test_upsert_same_key_overwrites(tmp_path):
    store = _store(tmp_path)
    store.record_verdict(**_row(agent_prediction="A"))
    store.record_verdict(**_row(agent_prediction="C"))
    rows = store._conn.execute(
        "SELECT agent_prediction FROM adversarial_verdicts "
        "WHERE question_id=? AND question_hash=? AND stage=?",
        ("v1_Action Recognition_0001", "h1", "cheat"),
    ).fetchall()
    assert len(rows) == 1 and rows[0][0] == "C"
    store.close()


def test_cheat_accuracy_aggregation(tmp_path):
    store = _store(tmp_path)
    store.record_verdict(**_row(question_id="q1", question_hash="a", agent_correct=True))
    store.record_verdict(**_row(question_id="q2", question_hash="b", agent_correct=False))
    store.record_verdict(**_row(question_id="q3", question_hash="c", agent_correct=True))
    assert store.cheat_agent_accuracy(round_no=0) == 2 / 3
    store.close()


def test_passed_question_ids(tmp_path):
    store = _store(tmp_path)
    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.close()
  • Step 2: 跑测试确认失败

Run: conda run -n Video-Tree-TRM pytest tests/unit/test_adversarial_verdicts_store.py -v Expected: FAIL（表与方法均不存在）

  • Step 3: 加 DDL 常量 + 索引

app/question_gen/run_store.py，在 _DDL_INDEXES 之后追加：

_DDL_VERDICTS = """
CREATE TABLE IF NOT EXISTS adversarial_verdicts (
    question_id      TEXT NOT NULL,
    question_hash    TEXT NOT NULL,
    stage            TEXT NOT NULL,
    round            INTEGER NOT NULL,
    agent_prediction TEXT,
    agent_correct    INTEGER,
    verdict          TEXT NOT NULL,
    pair_id          TEXT,
    agent_config     TEXT NOT NULL,
    created_at       TEXT NOT NULL DEFAULT (datetime('now')),
    PRIMARY KEY (question_id, question_hash, stage)
);
"""

_DDL_VERDICTS_INDEXES = [
    "CREATE INDEX IF NOT EXISTS idx_av_qid ON adversarial_verdicts(question_id);",
    "CREATE INDEX IF NOT EXISTS idx_av_verdict ON adversarial_verdicts(verdict);",
    "CREATE INDEX IF NOT EXISTS idx_av_round ON adversarial_verdicts(round);",
]
  • Step 4: 在 _init_schema 幂等建表

_init_schema 内，for idx_sql in _DDL_INDEXES: 循环之后、self._conn.commit() 之前插入：

        self._conn.execute(_DDL_VERDICTS)
        for idx_sql in _DDL_VERDICTS_INDEXES:
            self._conn.execute(idx_sql)

（CREATE TABLE IF NOT EXISTS 天然幂等，无需 ALTER。）

  • Step 5: 加 4 个方法

在 update_difficulty 之后追加：

    def record_verdict(
        self,
        *,
        question_id: str,
        question_hash: str,
        stage: str,
        round: int,
        agent_prediction: str | None,
        agent_correct: bool | None,
        verdict: str,
        pair_id: str | None,
        agent_config: str,
    ) -> None:
        """写入一条 agent 门判定（同 (question_id, question_hash, stage) upsert）。

        Parameters
        ----------
        question_id, question_hash, stage : str
            续跑主键三元组。
        round : int
            过滤轮次。
        agent_prediction : str | None
            agent 预测答案字母。
        agent_correct : bool | None
            作弊门是否答对（翻转门 stage 可为 None）。
        verdict : str
            passed | filtered_too_easy | filtered_no_flip | flip_skipped。
        pair_id : str | None
            关联原题与镜像题。
        agent_config : str
            agent 配置指纹（skill_mode/max_steps/model）。
        """
        self._conn.execute(
            """
            INSERT INTO adversarial_verdicts
                (question_id, question_hash, stage, round, agent_prediction,
                 agent_correct, verdict, pair_id, agent_config)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
            ON CONFLICT(question_id, question_hash, stage) DO UPDATE SET
                round=excluded.round,
                agent_prediction=excluded.agent_prediction,
                agent_correct=excluded.agent_correct,
                verdict=excluded.verdict,
                pair_id=excluded.pair_id,
                agent_config=excluded.agent_config,
                created_at=datetime('now')
            """,
            (
                question_id, question_hash, stage, round, agent_prediction,
                None if agent_correct is None else int(agent_correct),
                verdict, pair_id, agent_config,
            ),
        )
        self._conn.commit()

    def completed_stages(
        self, question_id: str, question_hash: str, agent_config: str
    ) -> set[str]:
        """返回该题在当前 hash+config 下已完成的 stage 集合（续跑用）。"""
        rows = self._conn.execute(
            "SELECT stage FROM adversarial_verdicts "
            "WHERE question_id=? AND question_hash=? AND agent_config=?",
            (question_id, question_hash, agent_config),
        ).fetchall()
        return {r[0] for r in rows}

    def invalidate_stale_config(self, question_id: str, agent_config: str) -> None:
        """agent_config 变化时，删除该题所有非当前 config 的旧 verdict。"""
        self._conn.execute(
            "DELETE FROM adversarial_verdicts "
            "WHERE question_id=? AND agent_config!=?",
            (question_id, agent_config),
        )
        self._conn.commit()

    def cheat_agent_accuracy(self, round_no: int) -> float:
        """某轮作弊门 agent 正确率（agent_correct 聚合），无数据返 0.0。"""
        row = self._conn.execute(
            "SELECT AVG(agent_correct) FROM adversarial_verdicts "
            "WHERE stage='cheat' AND round=?",
            (round_no,),
        ).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 全量重建用）。"""
        rows = self._conn.execute(
            "SELECT DISTINCT question_id FROM adversarial_verdicts WHERE verdict='passed'"
        ).fetchall()
        return {r[0] for r in rows}

注：形参名 round 遮蔽内建，但与设计列名一致、仅 kwargs 传入无实际风险；若 radon/ruff 报 A002，改列语义名 round_no 并在 SQL 保持列名 round。

  • Step 6: 跑测试确认通过

Run: conda run -n Video-Tree-TRM pytest tests/unit/test_adversarial_verdicts_store.py -v Expected: PASS

  • Step 7: 登记 schema 文档

新建/追加 research-wiki/schemas/adversarial-verdicts.md：登记表名、9 列语义（同设计 §4.1 表）、主键 (question_id, question_hash, stage)、续跑与 agent_config 作废语义、verdict 四枚举值。风格与既有 question-gen-items.md 一致。

  • Step 8: 回归 run_store 既有测试

Run: conda run -n Video-Tree-TRM pytest tests/unit/ -k "run_store" -v Expected: PASS（新表不影响既有 question_gen_items 行为）

  • Step 9: 提交
git add app/question_gen/run_store.py research-wiki/schemas/adversarial-verdicts.md tests/unit/test_adversarial_verdicts_store.py
git commit -m "feat: add adversarial_verdicts table with resume and aggregation"

Task 3: run_pipeline_v2 补生成三参数（可选，默认=现状）

补生成需继承已用节点/已接受题 embedding、续编 seq 防撞 ID。新增三个可选参数，不传时行为与现状字节级一致。

Files:

  • Modify: app/question_gen/pipeline_v2.py（_assign_slots 加 seq_offset；run_pipeline_v2 加 3 参数并织入）

  • Test: tests/unit/test_pipeline_v2_resume_params.py（新建）

  • Step 1: 写失败测试

新建 tests/unit/test_pipeline_v2_resume_params.py：

"""补生成参数：_assign_slots seq_offset 续编 + run_pipeline_v2 默认签名兼容。"""

import inspect

from app.question_gen.pipeline_v2 import _assign_slots, run_pipeline_v2


def test_assign_slots_seq_offset_continues_numbering():
    slots = _assign_slots(["v1"], ["Action Recognition"], 2, seq_offset=10)
    assert [s.seq for s in slots] == [11, 12]
    assert slots[0].slot_id == "Action Recognition_0011"


def test_assign_slots_default_offset_unchanged():
    slots = _assign_slots(["v1"], ["Action Recognition"], 2)
    assert [s.seq for s in slots] == [1, 2]
    assert slots[0].slot_id == "Action Recognition_0001"


def test_run_pipeline_v2_new_optional_params_default_none():
    sig = inspect.signature(run_pipeline_v2)
    for name in ("initial_used_node_ids", "initial_embed_pool", "seq_offset"):
        assert name in sig.parameters, name
        assert sig.parameters[name].kind == inspect.Parameter.KEYWORD_ONLY
    assert sig.parameters["initial_used_node_ids"].default is None
    assert sig.parameters["initial_embed_pool"].default is None
    assert sig.parameters["seq_offset"].default == 0
  • Step 2: 跑测试确认失败

Run: conda run -n Video-Tree-TRM pytest tests/unit/test_pipeline_v2_resume_params.py -v Expected: FAIL

  • Step 3: _assign_slots 加 seq_offset

app/question_gen/pipeline_v2.py，改签名与循环：

def _assign_slots(
    video_ids: list[str],
    task_types: list[str],
    per_type: int,
    seq_offset: int = 0,
) -> list[SlotAssignment]:
    """将出题目标分配为具体 slot 列表。

    ...（docstring 补一行）
    参数:
        seq_offset: 全局序号起始偏移（补生成续编，默认 0）。
    """
    slots: list[SlotAssignment] = []
    global_seq = seq_offset

    for task_type in task_types:
        for i in range(per_type):
            video_id = video_ids[i % len(video_ids)]
            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,
                    seq=global_seq,
                )
            )
    return slots
  • Step 4: run_pipeline_v2 加 3 参数并织入

签名（on_accept 之后）追加：

    on_accept: Callable[[GeneratedQuestion], None] | None = None,
    initial_used_node_ids: set[str] | None = None,
    initial_embed_pool: list[np.ndarray] | None = None,
    seq_offset: int = 0,
) -> PipelineResult:

docstring 参数区补三行说明（补生成继承已用节点/embedding、续编 seq）。

Phase 1 建 slot 处（约 941 行）：

    slots = _assign_slots(video_ids, task_types, config.per_type, seq_offset=seq_offset)

Phase 3 初始化处（约 954 行）改为继承传入值（默认空，不传=现状）：

    embed_pool: list[np.ndarray] = list(initial_embed_pool) if initial_embed_pool else []
    used_node_ids: set[str] = set(initial_used_node_ids) if initial_used_node_ids else set()

用 list(...)/set(...) 复制，避免补生成 run 就地改动调用方传入的容器。

  • Step 5: 跑测试确认通过

Run: conda run -n Video-Tree-TRM pytest tests/unit/test_pipeline_v2_resume_params.py -v Expected: PASS

  • Step 6: 回归出题管线集成测试（默认参数=现状）

Run: conda run -n Video-Tree-TRM pytest tests/integration/test_pipeline_v2.py -v Expected: PASS（未传新参 → 空初始化 + seq_offset=0 = 原行为）

  • Step 7: 提交
git add app/question_gen/pipeline_v2.py tests/unit/test_pipeline_v2_resume_params.py
git commit -m "feat: add optional resume params to run_pipeline_v2 for backfill"

Task 4: AdversarialFilterConfig dataclass + YAML 加载（filter 层配置）

filter 层配置（非 strategy 属性）：filter_task_types/adversarial_max_rounds/adversarial_agent_max_steps/difficulty_warn_threshold。仿 PipelineConfig/load_pipeline_config。

Files:

  • Create: app/question_gen/adversarial_config.py

  • Modify: config/question_gen_ar30.yaml（补 adversarial_filter 区段）

  • Test: tests/unit/test_adversarial_config.py（新建）

  • Step 1: 写失败测试

新建 tests/unit/test_adversarial_config.py：

"""AdversarialFilterConfig 默认值 + YAML 加载。"""

from app.question_gen.adversarial_config import (
    AdversarialFilterConfig,
    load_adversarial_config,
)


def test_defaults():
    cfg = AdversarialFilterConfig()
    assert cfg.filter_task_types == ("Action Recognition",)
    assert cfg.adversarial_max_rounds == 5
    assert cfg.adversarial_agent_max_steps == 40
    assert cfg.difficulty_warn_threshold == 0.85


def test_load_from_yaml(tmp_path):
    p = tmp_path / "c.yaml"
    p.write_text(
        "adversarial_filter:\n"
        "  filter_task_types: [Action Recognition, Object Recognition]\n"
        "  adversarial_max_rounds: 3\n"
        "  adversarial_agent_max_steps: 20\n"
        "  difficulty_warn_threshold: 0.7\n",
        encoding="utf-8",
    )
    cfg = load_adversarial_config(p)
    assert cfg.filter_task_types == ("Action Recognition", "Object Recognition")
    assert cfg.adversarial_max_rounds == 3
    assert cfg.adversarial_agent_max_steps == 20
    assert cfg.difficulty_warn_threshold == 0.7


def test_load_missing_section_uses_defaults(tmp_path):
    p = tmp_path / "c.yaml"
    p.write_text("question_gen_v2:\n  per_type: 3\n", encoding="utf-8")
    cfg = load_adversarial_config(p)
    assert cfg == AdversarialFilterConfig()
  • Step 2: 跑测试确认失败

Run: conda run -n Video-Tree-TRM pytest tests/unit/test_adversarial_config.py -v Expected: FAIL（模块不存在）

  • Step 3: 建模块

新建 app/question_gen/adversarial_config.py：

"""Phase B 对抗过滤层配置 — filter 层配置（非 strategy 属性）。

设计: research-wiki/designs/2026-07-14-adversarial-question-gen-phaseB-design.md §8
"""

from __future__ import annotations

from dataclasses import dataclass, field
from pathlib import Path

import yaml


@dataclass(frozen=True)
class AdversarialFilterConfig:
    """后置对抗过滤配置。

    属性:
        filter_task_types: 被过滤的题型（仅这些走 agent 门），默认仅 AR。
        adversarial_max_rounds: 补生成迭代上限。
        adversarial_agent_max_steps: agent 试答步数上限。
        difficulty_warn_threshold: 批次 agent 正确率告警阈值。
    """

    filter_task_types: tuple[str, ...] = ("Action Recognition",)
    adversarial_max_rounds: int = 5
    adversarial_agent_max_steps: int = 40
    difficulty_warn_threshold: float = 0.85


def load_adversarial_config(config_path: Path) -> AdversarialFilterConfig:
    """从 YAML 的 adversarial_filter 区段加载配置，缺段/缺键用默认值。

    参数:
        config_path: YAML 配置文件路径。

    返回:
        AdversarialFilterConfig 实例。
    """
    with open(config_path, encoding="utf-8") as f:
        raw = yaml.safe_load(f) or {}
    section = raw.get("adversarial_filter", {}) or {}
    default = AdversarialFilterConfig()
    types = section.get("filter_task_types")
    return AdversarialFilterConfig(
        filter_task_types=tuple(types) if types else default.filter_task_types,
        adversarial_max_rounds=int(
            section.get("adversarial_max_rounds", default.adversarial_max_rounds)
        ),
        adversarial_agent_max_steps=int(
            section.get("adversarial_agent_max_steps", default.adversarial_agent_max_steps)
        ),
        difficulty_warn_threshold=float(
            section.get("difficulty_warn_threshold", default.difficulty_warn_threshold)
        ),
    )

field 导入若未用则删除（ruff）。此处未用可去掉。

  • Step 4: 跑测试确认通过

Run: conda run -n Video-Tree-TRM pytest tests/unit/test_adversarial_config.py -v Expected: PASS

  • Step 5: 补 YAML 区段

config/question_gen_ar30.yaml 追加顶层区段：

adversarial_filter:
  filter_task_types: [Action Recognition]
  adversarial_max_rounds: 5
  adversarial_agent_max_steps: 40
  difficulty_warn_threshold: 0.85
  • Step 6: 提交
git add app/question_gen/adversarial_config.py config/question_gen_ar30.yaml tests/unit/test_adversarial_config.py
git commit -m "feat: add AdversarialFilterConfig for phase B filter layer"

Task 5: adversarial_filter.py 纯判定核心（hash / 指纹 / canonical / verdict）

先落地无 I/O 的纯逻辑：question_hash、agent_config 指纹、canonical 选项比较、翻转判定。这是消除判定噪声的核心（设计 §4.2 工程化细则），单测最密集。

Files:

  • Create: app/question_gen/adversarial_filter.py（骨架 + 纯函数）

  • Test: tests/unit/test_adversarial_filter_core.py（新建）

  • Step 1: 写失败测试

新建 tests/unit/test_adversarial_filter_core.py：

"""adversarial_filter 纯判定：hash / 指纹 / canonical / 翻转判定。"""

from core.types import GeneratedQuestion

from app.question_gen.adversarial_filter import (
    FlipDecision,
    agent_config_fingerprint,
    canonical_answer_text,
    judge_flip,
    question_hash,
)


def _q(qid="q1", options=("A. 蒸", "B. 炒", "C. 煮", "D. 炸"), answer="A"):
    return GeneratedQuestion(
        question_id=qid, video_id="v1", task_type="Action Recognition",
        question="?", options=options, answer=answer,
        source_nodes=("n1",), difficulty="hard",
        sub_pattern="temporal_reasoning_failure",
    )


def test_question_hash_stable_and_payload_sensitive():
    h1 = question_hash(_q())
    h2 = question_hash(_q())
    assert h1 == h2
    h3 = question_hash(_q(answer="B"))  # answer 变 → hash 变
    assert h1 != h3
    h4 = question_hash(_q(options=("A. 蒸", "B. 炒", "C. 煮", "D. 烤")))  # option 变 → 变
    assert h1 != h4


def test_agent_config_fingerprint_changes_with_inputs():
    a = agent_config_fingerprint(skill_mode="auto", max_steps=40, model="m1")
    b = agent_config_fingerprint(skill_mode="auto", max_steps=41, model="m1")
    c = agent_config_fingerprint(skill_mode="manual", max_steps=40, model="m1")
    assert a != b and a != c


def test_canonical_answer_text_maps_letter_to_option_text():
    assert canonical_answer_text(_q(), "C") == "煮"
    assert canonical_answer_text(_q(), "c") == "煮"


def test_canonical_answer_text_invalid_returns_none():
    assert canonical_answer_text(_q(), "Z") is None
    assert canonical_answer_text(_q(), "") is None
    assert canonical_answer_text(_q(), None) is None


def test_judge_flip_different_answers_passed():
    # P 选"蒸"，Q（镜像）选"炒"→ 语义不同 → passed
    d = judge_flip(p_text="蒸", q_text="炒")
    assert d is FlipDecision.PASSED


def test_judge_flip_same_answer_filtered():
    d = judge_flip(p_text="蒸", q_text="蒸")
    assert d is FlipDecision.FILTERED_NO_FLIP


def test_judge_flip_invalid_answer_skipped():
    assert judge_flip(p_text=None, q_text="炒") is FlipDecision.FLIP_SKIPPED
    assert judge_flip(p_text="蒸", q_text=None) is FlipDecision.FLIP_SKIPPED
  • Step 2: 跑测试确认失败

Run: conda run -n Video-Tree-TRM pytest tests/unit/test_adversarial_filter_core.py -v Expected: FAIL（模块不存在）

  • Step 3: 建模块骨架 + 纯函数

新建 app/question_gen/adversarial_filter.py：

"""Phase B 独立后置对抗过滤层 — 作弊者门 + 配对翻转门。

在 Phase A 产物 accepted_questions.json 之上，用完整 inference agent 揪残余
shortcut：作弊门（agent 秒杀=太简单，剔除）+ 翻转门（agent 答案须随问题翻转）。
不改 Phase A 状态机；过滤进度存独立 adversarial_verdicts 表。

设计: research-wiki/designs/2026-07-14-adversarial-question-gen-phaseB-design.md
"""

from __future__ import annotations

import enum
import hashlib
import json
from typing import TYPE_CHECKING

if TYPE_CHECKING:
    from core.types import GeneratedQuestion


class FlipDecision(enum.Enum):
    """翻转门判定结果。"""

    PASSED = "passed"
    FILTERED_NO_FLIP = "filtered_no_flip"
    FLIP_SKIPPED = "flip_skipped"


def question_hash(question: GeneratedQuestion) -> str:
    """题 payload（question+options+answer）的稳定 hash，防 JSON 变动误用旧 verdict。

    参数:
        question: 题目。

    返回:
        16 位十六进制摘要。
    """
    payload = json.dumps(
        {
            "question": question.question,
            "options": list(question.options),
            "answer": question.answer,
        },
        ensure_ascii=False,
        sort_keys=True,
    )
    return hashlib.sha256(payload.encode("utf-8")).hexdigest()[:16]


def agent_config_fingerprint(*, skill_mode: str, max_steps: int, model: str) -> str:
    """agent 配置指纹（skill_mode/max_steps/model），变化则该题 verdict 作废。"""
    raw = f"{skill_mode}|{max_steps}|{model}"
    return hashlib.sha256(raw.encode("utf-8")).hexdigest()[:16]


def canonical_answer_text(question: GeneratedQuestion, letter: str | None) -> str | None:
    """把 agent 预测的选项字母映射为选项规范化文本；非法/越界返回 None。

    镜像题选项会重洗牌，字母无语义，必须按选项文本比较。

    参数:
        question: 题目（提供 options）。
        letter: agent 预测字母（大小写不敏感），None/空/越界视为无效。

    返回:
        去掉 "X. " 前缀的选项文本；无效时 None。
    """
    if not letter or not isinstance(letter, str):
        return None
    idx = ord(letter.strip().upper()) - ord("A")
    if not 0 <= idx < len(question.options):
        return None
    opt = question.options[idx]
    prefix = f"{letter.strip().upper()}. "
    return opt[len(prefix):] if opt.startswith(prefix) else opt


def judge_flip(*, p_text: str | None, q_text: str | None) -> FlipDecision:
    """按 canonical 文本判翻转：任一无效→skipped；不同→passed；相同→filtered。

    参数:
        p_text: 原题 P 的 agent 所选 canonical 文本。
        q_text: 镜像题 Q 的 agent 所选 canonical 文本。

    返回:
        FlipDecision。
    """
    if p_text is None or q_text is None:
        return FlipDecision.FLIP_SKIPPED
    if p_text.strip() != q_text.strip():
        return FlipDecision.PASSED
    return FlipDecision.FILTERED_NO_FLIP
  • Step 4: 跑测试确认通过

Run: conda run -n Video-Tree-TRM pytest tests/unit/test_adversarial_filter_core.py -v Expected: PASS

  • Step 5: 提交
git add app/question_gen/adversarial_filter.py tests/unit/test_adversarial_filter_core.py
git commit -m "feat: add adversarial filter core decision helpers"

Task 6: 作弊者门 — 复用真实 inference agent 试答

对每道 AR 题跑完整 agent（标准设定=完整题）→ 从 predictions 表读预测 → agent_correct = (prediction == answer)。答对=filtered_too_easy；答错=进翻转门。核心是复用 ar30 的推理装配。

装配来源（实现者零上下文，照此拼）

main.py 与 runner.infer 的组装方式（已验证）：

  1. main._build_adapters(settings, embed_cfg) → llm / vlm / embed / ocr（InfraSettings() 读 .env，embed_cfg 来自 YAML embed 段）。
  2. InferenceDepsRouter(store_dir=<store>, embed_provider=embed, llm=llm, vlm=vlm, ocr=ocr, default_prompts_dir=store/prompts/<prompts_version>, default_skills_dir=store/skills/<skills_version>, skill_mode=<mode>, verify_vision=True, anchor=True, assemble_mode="ids_expand")。
  3. tool_dispatch_fn = router.create_dispatch()；prompt_builder = router.create_prompt_builder()。
  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 单测。

Files:

  • Modify: app/question_gen/adversarial_filter.py（AgentRunner Protocol + run_cheater_gate）

  • Test: tests/unit/test_adversarial_cheater_gate.py（新建）

  • Step 1: 写失败测试（mock agent）

新建 tests/unit/test_adversarial_cheater_gate.py：

"""作弊门：agent 答对=filtered_too_easy 并落表；答错=cheat verdict=passed 待翻转。"""

import pytest
from core.types import GeneratedQuestion

from app.question_gen.adversarial_config import AdversarialFilterConfig
from app.question_gen.adversarial_filter import run_cheater_gate
from app.question_gen.run_store import QuestionGenStore


class _FakeAgent:
    """按 question_id → 预测字母返回的 mock AgentRunner。"""

    def __init__(self, preds: dict[str, str], model: str = "m1"):
        self._preds = preds
        self.model = model
        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}


def _q(qid, answer="A"):
    return GeneratedQuestion(
        question_id=qid, video_id="v1", task_type="Action Recognition",
        question="?", options=("A. 蒸", "B. 炒", "C. 煮", "D. 炸"), answer=answer,
        source_nodes=("n1",), difficulty="hard", sub_pattern="temporal_reasoning_failure",
    )


@pytest.mark.asyncio
async def test_cheater_gate_filters_too_easy_and_keeps_hard(tmp_path):
    store = QuestionGenStore(str(tmp_path / "q.db"))
    agent = _FakeAgent({"easy": "A", "hard": "B"})  # easy 答对(A), hard 答错
    cfg = AdversarialFilterConfig()
    survivors = await run_cheater_gate(
        [_q("easy"), _q("hard")], agent=agent, store=store,
        config=cfg, round_no=0, run_id="r0",
    )
    ids = {q.question_id for q in survivors}
    assert ids == {"hard"}  # 只有答错的进翻转门
    verdicts = {
        r[0]: r[1] for r in store._conn.execute(
            "SELECT question_id, verdict FROM adversarial_verdicts WHERE stage='cheat'"
        )
    }
    assert verdicts["easy"] == "filtered_too_easy"
    # hard 在 cheat 阶段先记 passed（待翻转门可能改写；不支持翻转的题即终判 passed）
    assert verdicts["hard"] == "passed"
    store.close()


@pytest.mark.asyncio
async def test_cheater_gate_resume_skips_completed(tmp_path):
    store = QuestionGenStore(str(tmp_path / "q.db"))
    agent = _FakeAgent({"hard": "B"})
    cfg = AdversarialFilterConfig()
    await run_cheater_gate([_q("hard")], agent=agent, store=store,
                           config=cfg, round_no=0, run_id="r0")
    first = list(agent.calls)
    await run_cheater_gate([_q("hard")], agent=agent, store=store,
                           config=cfg, round_no=0, run_id="r1")
    assert agent.calls == first  # 第二次不重跑（已有 cheat verdict）
    store.close()
  • Step 2: 跑测试确认失败

Run: conda run -n Video-Tree-TRM pytest tests/unit/test_adversarial_cheater_gate.py -v Expected: FAIL

  • Step 3: 加 AgentRunner Protocol + run_cheater_gate

app/question_gen/adversarial_filter.py 追加。顶部导入区补 from typing import Protocol（在 TYPE_CHECKING 外）与 from loguru import logger：

class AgentRunner(Protocol):
    """完整 inference agent 试答端口 — Phase B 只依赖此接口（便于 mock）。

    实现见 Task 11 的 _RealAgentRunner（复用 run_inference + RunLogImpl）。
    """

    model: str

    async def predict(
        self,
        questions: list[GeneratedQuestion],
        *,
        max_steps: int,
        run_id: str,
    ) -> dict[str, str | None]:
        """跑完整 agent，返回 question_id → 预测答案字母（无预测为 None）。"""
        ...


async def run_cheater_gate(
    questions: list[GeneratedQuestion],
    *,
    agent: AgentRunner,
    store: QuestionGenStore,
    config: AdversarialFilterConfig,
    round_no: int,
    run_id: str,
) -> list[GeneratedQuestion]:
    """作弊门：完整 agent 试答；答对→filtered_too_easy，答错→cheat passed 待翻转。

    续跑：已在当前 hash+config 有 cheat verdict 的题跳过重跑。agent_config 变
    化时先作废该题旧 verdict。预测立即落表（崩溃不丢）。

    参数:
        questions: 待判定的 AR 题列表。
        agent: 完整 agent 试答端口。
        store: verdict 持久化。
        config: 过滤配置（提供 max_steps）。
        round_no: 当前轮次。
        run_id: agent 推理 run 标识。

    返回:
        agent 答错的题（进翻转门）；答错题的 cheat 预测字母暂存于返回题的
        question_id → 预测，由调用方（翻转门）复用，见 run_flip_gate。
    """
    cfg_fp = agent_config_fingerprint(
        skill_mode="", max_steps=config.adversarial_agent_max_steps, model=agent.model
    )
    todo: 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
        todo.append(q)

    survivors: list[GeneratedQuestion] = []
    if not todo:
        # 从已有 verdict 恢复 survivors（cheat 记 passed 且非 filtered_too_easy）
        return _recover_survivors(questions, store, cfg_fp)

    preds = await agent.predict(
        todo, max_steps=config.adversarial_agent_max_steps, run_id=run_id
    )
    for q in todo:
        pred = preds.get(q.question_id)
        correct = pred is not None and pred.strip().upper() == q.answer.strip().upper()
        verdict = "filtered_too_easy" if correct else "passed"
        store.record_verdict(
            question_id=q.question_id, question_hash=question_hash(q), stage="cheat",
            round=round_no, agent_prediction=pred, agent_correct=correct,
            verdict=verdict, pair_id=None, agent_config=cfg_fp,
        )
        if not correct:
            survivors.append(q)
    logger.info(
        "作弊门: {} 题 → 剔除太简单 {}，存活 {}",
        len(todo), len(todo) - len(survivors), len(survivors),
    )
    return survivors

补 _recover_survivors（续跑恢复）：

def _recover_survivors(
    questions: list[GeneratedQuestion],
    store: QuestionGenStore,
    cfg_fp: str,
) -> list[GeneratedQuestion]:
    """从已落 cheat verdict 恢复"agent 答错"的题（续跑，不重跑 agent）。"""
    survivors: list[GeneratedQuestion] = []
    for q in questions:
        rows = store._conn.execute(
            "SELECT agent_correct FROM adversarial_verdicts "
            "WHERE question_id=? AND question_hash=? AND stage='cheat' AND agent_config=?",
            (q.question_id, question_hash(q), cfg_fp),
        ).fetchall()
        if rows and rows[0][0] == 0:
            survivors.append(q)
    return survivors

在 adversarial_filter.py 顶部补 import：from app.question_gen.adversarial_config import AdversarialFilterConfig、from app.question_gen.run_store import QuestionGenStore（这两个模块不反向依赖 adversarial_filter，无环）。

关于 agent 预测的复用（翻转门需要）：作弊门已把答错题的 agent_prediction 落表（stage=cheat）。翻转门原题 P 的预测从表里读（SELECT agent_prediction WHERE stage='cheat'），不重跑——满足设计 §4.2 第 3 点。

  • Step 4: 跑测试确认通过

Run: conda run -n Video-Tree-TRM pytest tests/unit/test_adversarial_cheater_gate.py -v Expected: PASS

  • Step 5: 提交
git add app/question_gen/adversarial_filter.py tests/unit/test_adversarial_cheater_gate.py
git commit -m "feat: add cheater gate reusing full inference agent"

Task 7: 镜像题生成（重建素材 + VLM 生成 + canonical 正解校验）

对 supports_flip 的存活题：从 source_nodes 重建 MaterialContext（Phase B 持树），VLM 按翻转 flip_axis 生成镜像题；生成后校验 canonical_correct(P) != canonical_correct(Q)，否则 flip_skipped。镜像题不进最终题库。

Files:

  • Create: store/prompts/question_gen/ar_mirror_question.md

  • Modify: app/question_gen/adversarial_filter.py（_rebuild_material + generate_mirror_question）

  • Test: tests/unit/test_adversarial_mirror.py（新建）

  • Step 1: 建镜像生成 prompt

新建 store/prompts/question_gen/ar_mirror_question.md：

You generate a MIRROR (axis-flipped) version of a video Action Recognition
multiple-choice question, using the SAME video material.

## Given
- The original question, its four options, and the correct answer.
- The flip axis (e.g. "before/after" or "first/last").
- Subtitle context and video frames.

## Rules
- Flip ONLY the given axis: turn "before X" into "after X", "first" into
  "last", etc. Everything else (subject, granularity, style) stays identical.
- The mirror question MUST have a genuinely DIFFERENT correct answer than the
  original — it asks about the opposite side of the same axis.
- Reuse the SAME candidate option texts where possible, re-shuffled; the letter
  of the correct option WILL differ from the original.
- If the axis cannot be flipped into a well-formed question with a distinct
  correct answer (e.g. list-style or "cannot determine" answers), output
  {"mirror": null}.

## Output
Respond with ONLY a JSON object:
```json
{"mirror": {"question": "...", "options": ["A. ...", "B. ...", "C. ...", "D. ..."], "answer": "C"}}

Or {"mirror": null} if no valid mirror exists.


- [ ] **Step 2: 写失败测试（mock VLM）**

新建 `tests/unit/test_adversarial_mirror.py`：

```python
"""镜像生成：成功造出正解相反的镜像；正解相同/生成 null → 返回 None。"""

import pytest
from core.types import GeneratedQuestion, LLMResponse

from app.question_gen.adversarial_filter import generate_mirror_question


class _FakeVLM:
    def __init__(self, content: str):
        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",
        )


def _q():
    return GeneratedQuestion(
        question_id="q1", video_id="v1", task_type="Action Recognition",
        question="X 之前做了什么？", options=("A. 蒸", "B. 炒", "C. 煮", "D. 炸"),
        answer="A", source_nodes=("n1",), difficulty="hard",
        sub_pattern="temporal_reasoning_failure",
    )


@pytest.mark.asyncio
async def test_mirror_distinct_correct_ok():
    vlm = _FakeVLM('{"mirror": {"question": "X 之后做了什么？", '
                   '"options": ["A. 炒", "B. 蒸", "C. 煮", "D. 炸"], "answer": "A"}}')
    mirror = await generate_mirror_question(
        _q(), flip_axis="before/after", vlm=vlm, material=_FakeMaterial(), session_id="s",
    )
    assert mirror is not None
    # 原正解 canonical="蒸"，镜像正解 canonical="炒" → 相异，有效
    assert mirror.answer == "A"
    assert mirror.options[0] == "A. 炒"


@pytest.mark.asyncio
async def test_mirror_same_correct_rejected():
    # 镜像正解 canonical 仍是"蒸" → 造不出有效对 → None
    vlm = _FakeVLM('{"mirror": {"question": "X 之后？", '
                   '"options": ["A. 蒸", "B. 炒", "C. 煮", "D. 炸"], "answer": "A"}}')
    mirror = await generate_mirror_question(
        _q(), flip_axis="before/after", vlm=vlm, material=_FakeMaterial(), session_id="s",
    )
    assert mirror is None


@pytest.mark.asyncio
async def test_mirror_null_returns_none():
    vlm = _FakeVLM('{"mirror": null}')
    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"]
  • Step 3: 跑测试确认失败

Run: conda run -n Video-Tree-TRM pytest tests/unit/test_adversarial_mirror.py -v Expected: FAIL

  • Step 4: 实现 _rebuild_material + generate_mirror_question

app/question_gen/adversarial_filter.py 追加。顶部补 import：from pathlib import Path、from json_repair import repair_json；TYPE_CHECKING 区补 from app.tree.index import TreeIndex、from core.protocols import VLMProvider、from app.question_gen.sampler_v2 import MaterialContext。

_PROMPTS_DIR = Path(__file__).resolve().parent.parent.parent / "store" / "prompts" / "question_gen"


def _rebuild_material(tree: TreeIndex, source_nodes: tuple[str, ...]) -> MaterialContext:
    """从 source_nodes 重建镜像生成所需素材（字幕 + 帧）。

    复用 sampler_v2 的采集辅助；anchor/cross_l2_texts 镜像生成不需要，置空。
    """
    from app.question_gen.sampler_v2 import (
        _collect_frame_paths,
        _collect_subtitle_sentences,
    )
    from app.question_gen.sampler_v2 import MaterialContext as _MC

    subtitles = _collect_subtitle_sentences(tree, source_nodes)
    frames: list[str] = []
    for nid in source_nodes:
        frames.extend(_collect_frame_paths(tree, nid))
    return _MC(
        anchor=None,  # 镜像 prompt 不用 anchor
        source_nodes=source_nodes,
        subtitle_sentences=subtitles,
        frame_paths=frames,
        cross_l2_texts=[],
    )


def _parse_mirror(raw: str) -> dict | None:
    """解析 VLM 镜像响应；{"mirror": null} 或解析失败 → None。"""
    content = raw.strip()
    if "```" in content:
        for part in content.split("```"):
            s = part.strip()
            if s.startswith("json"):
                s = s[4:].strip()
            if s.startswith("{"):
                content = s
                break
    data = json.loads(repair_json(content, return_objects=False))
    if not isinstance(data, dict):
        return None
    mirror = data.get("mirror")
    return mirror if isinstance(mirror, dict) else None


async def generate_mirror_question(
    question: GeneratedQuestion,
    *,
    flip_axis: str,
    vlm: VLMProvider,
    material: MaterialContext,
    session_id: str,
) -> GeneratedQuestion | None:
    """VLM 生成翻转 flip_axis 的镜像题；正解 canonical 与原题相同则返 None。

    参数:
        question: 原题。
        flip_axis: 翻转轴（"before/after" | "first/last"）。
        vlm: VLM 端口。
        material: 重建素材（frame_paths / subtitles）。
        session_id: 遥测会话 ID。

    返回:
        镜像 GeneratedQuestion（question_id 加 "_mirror" 后缀，不进题库）；
        无法造出有效对（null / 正解相同 / 解析失败）返回 None。
    """
    system = (_PROMPTS_DIR / "ar_mirror_question.md").read_text(encoding="utf-8")
    subs = "\n".join(f"  - {s}" for s in material.subtitle_sentences)
    user = (
        f"## Original Question\n{question.question}\n"
        f"## Options\n" + "\n".join(question.options) + "\n"
        f"## Correct Answer\n{question.answer}\n"
        f"## Flip Axis\n{flip_axis}\n"
        f"## Subtitles\n{subs}\n"
    )
    messages = [{"role": "system", "content": system}, {"role": "user", "content": user}]
    resp = await vlm.chat_with_images(
        messages, list(material.frame_paths), session_id=session_id
    )
    mirror = _parse_mirror(resp.content)
    if mirror is None:
        return None
    try:
        options = tuple(str(o) for o in mirror["options"])
        answer = str(mirror["answer"]).strip().upper()
        m_question = str(mirror["question"])
    except (KeyError, TypeError):
        return None
    mirror_q = GeneratedQuestion(
        question_id=f"{question.question_id}_mirror",
        video_id=question.video_id, task_type=question.task_type,
        question=m_question, options=options, answer=answer,
        source_nodes=question.source_nodes, difficulty=question.difficulty,
        sub_pattern=question.sub_pattern,
    )
    # 镜像正解字面校验：canonical(P) 必须 != canonical(Q)
    p_text = canonical_answer_text(question, question.answer)
    q_text = canonical_answer_text(mirror_q, answer)
    if p_text is None or q_text is None or p_text.strip() == q_text.strip():
        return None
    return mirror_q

GeneratedQuestion 构造参数须与 core/types.py 字段一致（Phase A 已加 sub_pattern）。若该类要求 family/skill_target 等有默认值即可省略；实现时以实际 dataclass 默认值为准。

  • Step 5: 跑测试确认通过

Run: conda run -n Video-Tree-TRM pytest tests/unit/test_adversarial_mirror.py -v Expected: PASS

  • Step 6: 提交
git add store/prompts/question_gen/ar_mirror_question.md app/question_gen/adversarial_filter.py tests/unit/test_adversarial_mirror.py
git commit -m "feat: add mirror question generation with canonical distinctness check"

Task 8: 配对翻转门 — 复用 P 预测 + agent 跑镜像 Q + 判定

存活的"agent 答错"题：不支持 flip 的终判 passed；支持 flip 的重建素材→生成镜像→agent 跑镜像→按 canonical 是否翻转判 passed/filtered_no_flip；任一无效/生成失败→flip_skipped（退回只经作弊门，不误杀）。

Files:

  • Modify: app/question_gen/adversarial_filter.py（run_flip_gate）

  • Test: tests/unit/test_adversarial_flip_gate.py（新建）

  • 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" 有独立行。示例断言骨架：

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

（其余三例类比：镜像 agent 选到 canonical=蒸 → filtered_no_flip；VLM 返回 {"mirror": null} → flip_skipped 但仍 passed 保留，因退回只经作弊门；不支持 flip 的子模式 → 直接 passed 不跑 VLM/agent。测试须断言这些语义。）

  • Step 2: 跑测试确认失败

Run: conda run -n Video-Tree-TRM pytest tests/unit/test_adversarial_flip_gate.py -v Expected: FAIL

  • Step 3: 实现 run_flip_gate

app/question_gen/adversarial_filter.py 追加。查 SubPattern 的 flip 声明用 strategy_action_recognition._AR_PATTERN_BY_NAME：

async def run_flip_gate(
    survivors: list[GeneratedQuestion],
    *,
    agent: AgentRunner,
    vlm: VLMProvider,
    store: QuestionGenStore,
    trees: dict[str, TreeIndex],
    config: AdversarialFilterConfig,
    round_no: int,
    run_id: str,
    session_id: str,
) -> list[GeneratedQuestion]:
    """翻转门：不支持 flip 的终判 passed；支持的按 canonical 翻转判定。

    P 预测复用作弊门落表结果（不重跑）；仅新跑镜像 Q。任一无效/镜像失败→
    flip_skipped（保留题，只经作弊门）。镜像题不进题库。

    返回:
        终判 verdict∈{passed, flip_skipped} 的题（filtered_no_flip 被剔除）。
    """
    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
    )
    kept: list[GeneratedQuestion] = []
    for q in survivors:
        sp = _AR_PATTERN_BY_NAME.get(q.sub_pattern or "")
        if sp is None or not sp.supports_flip:
            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,
        )
        pair_id = f"{q.question_id}::{round_no}"
        _persist_flip(store, q, decision, mirror_pred, round_no, cfg_fp, pair_id)
        if decision is not FlipDecision.FILTERED_NO_FLIP:
            kept.append(q)  # passed 或 flip_skipped 都保留
    logger.info("翻转门: {} 存活 → 保留 {}", len(survivors), len(kept))
    return kept

补两个辅助（保持每函数 radon ≥ B）：

async def _judge_one_flip(
    q: GeneratedQuestion,
    flip_axis: str | None,
    *,
    agent: AgentRunner,
    vlm: VLMProvider,
    trees: dict[str, TreeIndex],
    config: AdversarialFilterConfig,
    run_id: str,
    session_id: str,
) -> tuple[FlipDecision, str | None]:
    """跑单题翻转判定，返回 (decision, 镜像预测字母)。"""
    tree = trees.get(q.video_id)
    if tree is None or flip_axis is None:
        return FlipDecision.FLIP_SKIPPED, None
    material = _rebuild_material(tree, q.source_nodes)
    mirror = await generate_mirror_question(
        q, flip_axis=flip_axis, vlm=vlm, material=material, session_id=session_id
    )
    if mirror is None:
        return FlipDecision.FLIP_SKIPPED, None
    preds = await agent.predict(
        [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_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

_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。实现时确保：

def _read_cheat_prediction(q: GeneratedQuestion) -> str | None:
    ...  # SELECT agent_prediction FROM adversarial_verdicts
        # WHERE question_id=? AND question_hash=? AND stage='cheat'

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

  • Step 4: 跑测试确认通过

Run: conda run -n Video-Tree-TRM pytest tests/unit/test_adversarial_flip_gate.py -v Expected: PASS

  • Step 5: 提交
git add app/question_gen/adversarial_filter.py tests/unit/test_adversarial_flip_gate.py
git commit -m "feat: add pairwise flip gate reusing P prediction and mirror agent run"

Task 9: final JSON 全量重写 + 补生成迭代循环 + 难度报告

编排两门 + 补生成迭代：accepted_questions_final.json 每轮全量原子重写（内容=所有 verdict=passed 题）；缺额>0 且轮次<上限→调 run_pipeline_v2 补生成（传 initial_used_node_ids/initial_embed_pool/seq_offset）；每轮记 agent 正确率，超阈值 logger.warning。

Files:

  • Modify: app/question_gen/adversarial_filter.py（write_final_bank + run_adversarial_rounds + _report_difficulty）

  • Test: tests/unit/test_adversarial_iteration.py（新建）

  • Step 1: 写失败测试

新建 tests/unit/test_adversarial_iteration.py，覆盖：

  • write_final_bank：全量重写（tmp+os.replace）、内容仅含 passed 题、可从空 verdicts 表重建为 []。
  • 缺额计算：deficit = target - passed；deficit≤0 或 round≥max → 停止（用假的 backfill 回调计数验证调用次数）。
  • 难度报告：agent 正确率 > 阈值 → caplog 捕获 warning。
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,
                         verdict="filtered_too_easy", pair_id=None, agent_config="c")
    all_qs = {"q1": _q("q1"), "q2": _q("q2")}
    out = tmp_path / "accepted_questions_final.json"
    write_final_bank(out, store, all_qs)
    data = json.loads(out.read_text(encoding="utf-8"))
    assert [d["question_id"] for d in data] == ["q1"]


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
        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")
    with caplog.at_level("WARNING"):
        _report_difficulty(store, round_no=0, threshold=0.85)
    assert any("太简单" in r.message or "简单" in r.message for r in caplog.records)
  • Step 2: 跑测试确认失败

Run: conda run -n Video-Tree-TRM pytest tests/unit/test_adversarial_iteration.py -v Expected: FAIL

  • Step 3: 实现 write_final_bank + _report_difficulty + run_adversarial_rounds

app/question_gen/adversarial_filter.py 追加。顶部补 import os。

def write_final_bank(
    final_path: Path,
    store: QuestionGenStore,
    all_questions: dict[str, GeneratedQuestion],
) -> int:
    """全量重写 accepted_questions_final.json（tmp+os.replace 原子）。

    内容 = store 中所有 verdict=passed 的题（可随时从 verdicts 表重建）。

    参数:
        final_path: 输出路径。
        store: verdict 来源。
        all_questions: question_id → GeneratedQuestion（重建 payload）。

    返回:
        写入的题数。
    """
    passed_ids = store.passed_question_ids()
    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")
    tmp.write_text(json.dumps(entries, ensure_ascii=False, indent=2), encoding="utf-8")
    os.replace(str(tmp), str(final_path))
    logger.info("final 题库全量重写: {} 题 → {}", len(entries), final_path)
    return len(entries)


def _question_to_final_entry(q: GeneratedQuestion) -> dict:
    """序列化为 final JSON entry（含 sub_pattern，与 accepted_questions.json 同构）。"""
    return {
        "question_id": q.question_id, "video_id": q.video_id,
        "task_type": q.task_type, "question": q.question,
        "options": list(q.options), "answer": q.answer,
        "source_nodes": list(q.source_nodes), "difficulty": q.difficulty,
        "family": q.family, "skill_target": q.skill_target,
        "sub_pattern": q.sub_pattern,
    }


def _report_difficulty(store: QuestionGenStore, *, round_no: int, threshold: float) -> float:
    """记录并按阈值告警本轮 agent 正确率（作弊门聚合）。"""
    acc = store.cheat_agent_accuracy(round_no)
    logger.info("难度报告 round={}: agent 正确率={:.2%}", round_no, acc)
    if acc > threshold:
        logger.warning(
            "出题太简单: round={} agent 正确率={:.2%} > 阈值 {:.2%}",
            round_no, acc, threshold,
        )
    return acc

run_adversarial_rounds 编排迭代（用 Protocol 化的 backfill 回调，便于测；真实实现由 Task 11 注入）：

async def run_adversarial_rounds(
    initial_questions: list[GeneratedQuestion],
    *,
    agent: AgentRunner,
    vlm: VLMProvider,
    store: QuestionGenStore,
    trees: dict[str, TreeIndex],
    config: AdversarialFilterConfig,
    final_path: Path,
    target: int,
    backfill: "BackfillFn",
    session_id: str,
) -> None:
    """两门 + 补生成迭代主循环，每轮全量重写 final 并做难度报告。

    参数:
        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。
    """
    all_questions: dict[str, GeneratedQuestion] = {q.question_id: q for q in initial_questions}
    pending = list(initial_questions)
    for round_no in range(config.adversarial_max_rounds):
        survivors = await run_cheater_gate(
            pending, agent=agent, store=store, config=config,
            round_no=round_no, run_id=f"{session_id}_cheat_{round_no}",
        )
        await run_flip_gate(
            survivors, agent=agent, vlm=vlm, store=store, trees=trees,
            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)
        _report_difficulty(
            store, round_no=round_no, threshold=config.difficulty_warn_threshold
        )
        deficit = target - passed_now
        if deficit <= 0 or round_no + 1 >= config.adversarial_max_rounds:
            break
        new_qs = await backfill(deficit, round_no, all_questions)
        for q in new_qs:
            all_questions[q.question_id] = q
        pending = new_qs  # 只对新补的题重新过滤
    logger.info("对抗过滤结束: final={} 题", len(store.passed_question_ids()))

补 BackfillFn Protocol：

class BackfillFn(Protocol):
    """补生成回调 — 缺额驱动，返回新增 AR 题。"""

    async def __call__(
        self,
        deficit: int,
        round_no: int,
        existing: dict[str, GeneratedQuestion],
    ) -> list[GeneratedQuestion]:
        ...
  • Step 4: 跑测试确认通过

Run: conda run -n Video-Tree-TRM pytest tests/unit/test_adversarial_iteration.py -v Expected: PASS

  • Step 5: 提交
git add app/question_gen/adversarial_filter.py tests/unit/test_adversarial_iteration.py
git commit -m "feat: add final bank rewrite, iteration loop and difficulty report"

Task 10: 顶层入口 run_adversarial_filter — 真实 agent/backfill 装配 + CLI

把 Phase B 拼成可运行入口：装配真实 AgentRunner（run_inference + RunLogImpl）与真实 backfill（run_pipeline_v2），从 accepted_questions.json 读题过滤 filter_task_types，调 run_adversarial_rounds。

Files:

  • Modify: app/question_gen/adversarial_filter.py（_RealAgentRunner + run_adversarial_filter 入口）

  • Modify: tools/generate_questions.py（新增 adversarial-filter 子命令，装配 adapters/router/store 后调入口）

  • Test: tests/integration/test_adversarial_filter_e2e.py（新建）

  • Step 1: 写端到端集成测试（mock agent + mock VLM）

新建 tests/integration/test_adversarial_filter_e2e.py：构造临时 accepted_questions.json（含 AR + 1 个非 AR 题）、临时树、mock AgentRunner/VLMProvider/backfill，调 run_adversarial_filter，断言：

  • 非 AR 题不进 agent 门（不出现在 verdicts 表）；

  • accepted_questions_final.json 仅含 passed 题；

  • 断点续跑：第二次调用不重跑已判题（agent 调用计数不变）。

  • Step 2: 跑测试确认失败

Run: conda run -n Video-Tree-TRM pytest tests/integration/test_adversarial_filter_e2e.py -v Expected: FAIL

  • Step 3: 实现 _RealAgentRunner

app/question_gen/adversarial_filter.py 追加（复用 Task 6 装配来源，注入 router 组件）：

class _RealAgentRunner:
    """AgentRunner 实现 — 复用 run_inference + RunLogImpl 读回预测。

    参数:
        llm: 推理 LLMProvider。
        tool_dispatch_fn / prompt_builder: 由 InferenceDepsRouter 提供。
        db_path: HarnessLog / RunLogImpl 的 sqlite 路径。
        concurrency / skill_mode / model: run_inference 参数与指纹来源。
    """

    def __init__(
        self, *, llm, tool_dispatch_fn, prompt_builder, 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, *, max_steps, run_id):
        """跑完整 agent，回读 predictions 表，返回 question_id → 预测字母。"""
        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=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}

指纹用 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"）。

  • Step 4: 实现 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 题数（见待确认项），调 run_adversarial_rounds。函数签名接收已装配好的 agent/vlm/trees/store/两个 config/路径，保持可测。

  • Step 5: 加 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: 跑测试确认通过

Run: conda run -n Video-Tree-TRM pytest tests/integration/test_adversarial_filter_e2e.py -v Expected: PASS

  • Step 7: 提交
git add app/question_gen/adversarial_filter.py tools/generate_questions.py tests/integration/test_adversarial_filter_e2e.py
git commit -m "feat: wire real agent runner and CLI entry for adversarial filter"

Task 11: 全量回归 + lint + radon + wiki 收口

Files: 无新代码；验证 + wiki 登记。

  • Step 1: 全量测试

Run: conda run -n Video-Tree-TRM pytest tests/ -q Expected: 全绿（含既有用例，证明 11 非 AR 题型与 Phase A 状态机行为不变）。有红回对应 Task 修。

  • Step 2: lint + 复杂度

Run: conda run -n Video-Tree-TRM ruff check app/ core/ tools/ --fix && conda run -n Video-Tree-TRM ruff format app/question_gen/adversarial_filter.py app/question_gen/adversarial_config.py Run: conda run -n Video-Tree-TRM radon cc app/question_gen/adversarial_filter.py -s -nc Expected: ruff 无剩余错误；radon 无 C 级及以下函数（有则拆分）。

  • Step 3: wiki 登记 plan 实体
conda run -n Video-Tree-TRM python3 .claude/tools/research_wiki.py add_entity research-wiki/ --type plan --id adversarial-question-gen-phaseB --title "Adversarial Question-Gen Phase B"
conda run -n Video-Tree-TRM python3 .claude/tools/research_wiki.py add_edge research-wiki/ --from "plan:adversarial-question-gen-phaseB" --to "design:adversarial-question-gen-phaseB" --type implements --evidence "Phase B 实现计划"
conda run -n Video-Tree-TRM python3 .claude/tools/research_wiki.py rebuild_index research-wiki/
  • Step 4: 提交
git add research-wiki/
git commit -m "docs: register Phase B plan in research wiki"

已确认的实现决策（原为 genuine ambiguities，现锁定）

  1. target 定义（已锁定）：target = 首轮 accepted_questions.json 中 filter_task_types 题数（即维持原始 AR 题库规模——过滤掉太简单/不翻转的题后，补生成回到同等题数但更难）。仅影响补生成停止条件，不影响门逻辑。ar30 场景下即首轮 AR 题数。
  2. agent "prediction" 语义（已锁定为字母 + 防御回退）：_run_single_question 落库 prediction = result_dict.get("answer")，与 qa.answer（字母 "A"/"B"/…）比较判对错，故按字母处理（canonical_answer_text 把字母映射为选项文本）。防御：若某 skill_mode 下 agent 返回选项全文而非字母，canonical_answer_text 返回 None → 保守判 flip_skipped（绝不误杀）。实现验证步骤（强制）：Task 6 实现时，先跑一次真实 agent 落一条 predictions 行、抽查 prediction 字段形态确认为字母；若为全文，给 canonical_answer_text 补"按文本匹配选项"回退分支后再继续。此验证已并入 Task 6 的实现约束。

Self-Review 与保真校验

Spec 覆盖（设计每节 → Task）：

  • §4.1 作弊者门 + adversarial_verdicts 表 → Task 2（表/续跑/聚合）+ Task 6（门逻辑）。
  • §4.2 配对翻转门（canonical 比较、无效→skipped、镜像正解校验、镜像不进库）→ Task 5（canonical/judge_flip）+ Task 7（镜像生成 + 正解校验）+ Task 8（门编排 + P 复用）。
  • §4.3 补生成与迭代（三参数、seq_offset 防撞、两份 JSON 时序、final 全量重写）→ Task 3（pipeline 参数）+ Task 9（final 重写 + 迭代）+ Task 10（真实 backfill 装配）。
  • §4.4 难度报告（agent_correct 聚合、阈值告警、不复用 difficulty_steps）→ Task 2（cheat_agent_accuracy）+ Task 9（_report_difficulty）。
  • §6 非功能（持久化/幂等/续跑/原子性）→ Task 2（每题立即落表、(qid,hash,stage) 续跑、config 作废）+ Task 9（final tmp+os.replace 原子、可从表重建）。
  • §8 配置（4 参数，filter 层非 strategy）→ Task 4。
  • SubPattern supports_flip/flip_axis 声明 → Task 1。
  • 路径隔离（仅 filter_task_types；11 题型 + Phase A 状态机零改动）→ Task 1/2/3 默认值 + 回归步骤，Task 10 按 filter_task_types 过滤，Task 11 全量回归。

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）一致。

核心算法保真（N/A）： Phase B 全部改动局限于 question_gen 后置过滤层（新模块 + 新表 + 3 个可选 pipeline 参数 + SubPattern 2 字段），不涉及 research-wiki/ARCHITECTURE.md §6 的 12 项核心算法（建树 4 + 训练 8）。作弊门/翻转门复用既有 run_inference（AgentLoop 完整树搜索）未改其内部。保真校验不适用。

Reference in New Issue View Git Blame Copy Permalink