diff --git a/research-wiki/plans/2026-07-14-adversarial-question-gen-phaseB-plan.md b/research-wiki/plans/2026-07-14-adversarial-question-gen-phaseB-plan.md new file mode 100644 index 0000000..1a80d08 --- /dev/null +++ b/research-wiki/plans/2026-07-14-adversarial-question-gen-phaseB-plan.md @@ -0,0 +1,1806 @@ +# 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、`: `，**禁止任何 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`： + +```python +"""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，带默认值）： + +```python + 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=(...)` 之后）加： + +```python + supports_flip=True, + flip_axis="before/after", +``` + +`_CROSS_SEGMENT_ENTITY_TRACKING = SubPattern(...)` 的构造末尾加： + +```python + 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: 提交** + +```bash +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`： + +```python +"""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` 之后追加： + +```python +_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()` 之前插入： + +```python + 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` 之后追加： + +```python + 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: 提交** + +```bash +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`： + +```python +"""补生成参数：_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`，改签名与循环： + +```python +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` 之后）追加： + +```python + 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 行）： + +```python + slots = _assign_slots(video_ids, task_types, config.per_type, seq_offset=seq_offset) +``` + +Phase 3 初始化处（约 954 行）改为继承传入值（默认空，不传=现状）： + +```python + 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: 提交** + +```bash +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`： + +```python +"""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`： + +```python +"""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` 追加顶层区段： + +```yaml +adversarial_filter: + filter_task_types: [Action Recognition] + adversarial_max_rounds: 5 + adversarial_agent_max_steps: 40 + difficulty_warn_threshold: 0.85 +``` + +- [ ] **Step 6: 提交** + +```bash +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`： + +```python +"""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`： + +```python +"""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: 提交** + +```bash +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=, embed_provider=embed, llm=llm, vlm=vlm, ocr=ocr, default_prompts_dir=store/prompts/, default_skills_dir=store/skills/, skill_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=, skill_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`： + +```python +"""作弊门：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`： + +```python +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`（续跑恢复）： + +```python +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: 提交** + +```bash +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`： + +```markdown +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`。 + +```python +_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: 提交** + +```bash +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"` 有独立行。示例断言骨架： + +```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"} +``` + +（其余三例类比：镜像 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`： + +```python +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）： + +```python +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`。实现时确保： + +```python +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: 提交** + +```bash +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。 + +```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, + 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`。 + +```python +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 注入）： + +```python +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： + +```python +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: 提交** + +```bash +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 组件）： + +```python +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: 提交** + +```bash +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 实体** + +```bash +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: 提交** + +```bash +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 完整树搜索）**未改其内部**。**保真校验不适用。** + +---