feat: add adversarial_verdicts table with resume and terminal-verdict query

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

app/question_gen/run_store.py

@@ -139,6 +139,28 @@ _DDL_INDEXES = [
     "CREATE INDEX IF NOT EXISTS idx_qgi_task_type ON question_gen_items(task_type);",
 ]
 
+_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);",
+]
+
 
 # ---------------------------------------------------------------------------
 # Store 实现
@@ -176,6 +198,9 @@ class QuestionGenStore:
         self._conn.execute(_DDL_ITEMS)
         for idx_sql in _DDL_INDEXES:
             self._conn.execute(idx_sql)
+        self._conn.execute(_DDL_VERDICTS)
+        for idx_sql in _DDL_VERDICTS_INDEXES:
+            self._conn.execute(idx_sql)
         self._conn.commit()
 
         # 幂等迁移：为已有表加 sub_pattern / selector_scores 列
@@ -382,6 +407,166 @@ class QuestionGenStore:
         if cursor.rowcount == 0:
             raise ValueError(f"item_id 不存在: {item_id}")
 
+    def record_verdict(
+        self,
+        *,
+        question_id: str,
+        question_hash: str,
+        stage: str,
+        round: int,  # noqa: A002 — 与设计列名一致，仅 kwargs 传入无遮蔽风险
+        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）。
+
+        每次写入立即 commit，保证崩溃安全（进程中断最多丢失当前未提交的一条）。
+
+        Parameters
+        ----------
+        question_id, question_hash, stage : str
+            续跑主键三元组（stage ∈ cheat|flip_original|flip_mirror）。
+        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 集合（续跑用）。
+
+        Parameters
+        ----------
+        question_id : str
+            题目唯一标识。
+        question_hash : str
+            当前题面指纹；hash 不匹配的旧行视为未完成，需重跑试答。
+        agent_config : str
+            当前 agent 配置指纹。
+
+        Returns
+        -------
+        set[str]
+            已完成的 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。
+
+        Parameters
+        ----------
+        question_id : str
+            题目唯一标识。
+        agent_config : str
+            当前 agent 配置指纹；保留该 config 行，其余全部删除。
+        """
+        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。
+
+        Parameters
+        ----------
+        round_no : int
+            过滤轮次。
+
+        Returns
+        -------
+        float
+            该轮 stage='cheat' 的 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 final_passed_question_ids(
+        self, hash_by_qid: dict[str, str], agent_config: str
+    ) -> set[str]:
+        """在当前 hash+config 下通过两门的 question_id 集合（final JSON 全量重建用）。
+
+        终判规则（防 stale 泄漏）：仅当该题在 **当前 question_hash + 当前
+        agent_config** 下同时满足——存在 stage='cheat' 且 verdict='passed'
+        （agent 答错=不太简单），且不存在任何 stage 的 verdict='filtered_no_flip'
+        （未被翻转门剔除）——才计入 final-passed。stale hash / stale config 的旧行
+        因不匹配传入的 (qid, hash, config) 天然被排除，绝不泄漏进最终题库。
+
+        Parameters
+        ----------
+        hash_by_qid : dict[str, str]
+            question_id → 当前 question_hash 映射（来自本轮 all_questions）。
+        agent_config : str
+            当前 agent 配置指纹。
+
+        Returns
+        -------
+        set[str]
+            终判 passed 的 question_id 集合。
+        """
+        passed: set[str] = set()
+        for qid, qhash in hash_by_qid.items():
+            rows = self._conn.execute(
+                "SELECT stage, verdict FROM adversarial_verdicts "
+                "WHERE question_id=? AND question_hash=? AND agent_config=?",
+                (qid, qhash, agent_config),
+            ).fetchall()
+            if not rows:
+                continue
+            cheat_passed = any(
+                stage == "cheat" and verdict == "passed" for stage, verdict in rows
+            )
+            no_flip = any(verdict == "filtered_no_flip" for _, verdict in rows)
+            if cheat_passed and not no_flip:
+                passed.add(qid)
+        return passed
+
     def update_selector_scores(self, item_id: str, selector_scores_json: str) -> None:
         """写入 grounded selector 打分观测（JSON 字符串）。
 

research-wiki/schemas/adversarial-verdicts.md

@@ -0,0 +1,86 @@
+---
+type: schema
+node_id: schema:adversarial-verdicts
+title: "表结构: adversarial_verdicts（Phase B agent 门判定）"
+date: 2026-07-14
+---
+
+# 表结构: adversarial_verdicts（Phase B agent 门判定）
+
+出题管线 Phase B 后置过滤层的持久化底座。记录 agent 对每题每 stage 的一次试答判定，支持
+按 `(question_id, question_hash, stage)` 续跑、按 `agent_config` 变化作废、聚合作弊门正确率，
+以及在当前 hash+config 下重建最终题库。与 Phase A 的 `question_gen_items` 表正交（互不影响）。
+
+## 列定义
+
+| 列名 | 类型 | 约束 | 说明 |
+|------|------|------|------|
+| `question_id` | TEXT | PK(1/3) NOT NULL | 题目唯一标识 |
+| `question_hash` | TEXT | PK(2/3) NOT NULL | 当前题面指纹；hash 不匹配的旧行视为未完成，需重跑试答 |
+| `stage` | TEXT | PK(3/3) NOT NULL | cheat / flip_original / flip_mirror |
+| `round` | INTEGER | NOT NULL | 过滤轮次 |
+| `agent_prediction` | TEXT | | agent 预测答案字母（可 NULL） |
+| `agent_correct` | INTEGER | | 作弊门是否答对（1/0；翻转门 stage 可为 NULL） |
+| `verdict` | TEXT | NOT NULL | passed / filtered_too_easy / filtered_no_flip / flip_skipped |
+| `pair_id` | TEXT | | 关联原题与镜像题（可 NULL） |
+| `agent_config` | TEXT | NOT NULL | agent 配置指纹（skill_mode/max_steps/model） |
+| `created_at` | TEXT | NOT NULL | ISO8601（写入/覆盖时刷新） |
+
+主键 `(question_id, question_hash, stage)`：同三元组重复写入即 upsert 覆盖（幂等）。
+
+### `verdict` 四枚举值
+
+| 值 | 语义 |
+|----|------|
+| `passed` | 通过本 stage（作弊门：agent 答错=不太简单；翻转门：镜像题翻转成立） |
+| `filtered_too_easy` | 作弊门剔除：agent 无视频即答对，题目太简单 |
+| `filtered_no_flip` | 翻转门剔除：镜像题答案未按预期翻转 |
+| `flip_skipped` | 翻转门跳过（未构造镜像题等） |
+
+## DDL
+
+```sql
+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)
+);
+
+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);
+```
+
+## 非功能性行为
+
+| 维度 | 策略 |
+|------|------|
+| 持久化 | 每次 `record_verdict` 立即 commit，崩溃最多丢失当前未提交的一条 |
+| 幂等性 | 主键 upsert；`CREATE TABLE IF NOT EXISTS` 天然幂等，重复 `_init_schema` 安全 |
+| 断点续跑 | `completed_stages(qid, hash, config)` 返回已完成 stage，重启只补跑未完成的 |
+| config 作废 | `invalidate_stale_config(qid, config)` 真正 DELETE 非当前 config 的旧行 |
+
+## Store 方法
+
+| 方法 | 用途 |
+|------|------|
+| `record_verdict(...)` | upsert 一条 agent 门判定（Phase B Task 6 记 verdict） |
+| `completed_stages(qid, hash, config)` | 续跑：当前 hash+config 下已完成的 stage 集合 |
+| `invalidate_stale_config(qid, config)` | agent_config 变化时删除该题所有旧 config 行 |
+| `cheat_agent_accuracy(round_no)` | 某轮 stage='cheat' 的 agent_correct 平均值（无数据返 0.0） |
+| `final_passed_question_ids(hash_by_qid, config)` | 终判 passed 集合，final JSON 全量重建用 |
+
+### 终判规则（防 stale 泄漏）
+
+`final_passed_question_ids` 仅当该题在 **当前 question_hash + 当前 agent_config** 下同时满足：
+存在 `stage='cheat'` 且 `verdict='passed'`，且不存在任何 stage 的 `verdict='filtered_no_flip'`，
+才计入 final-passed。stale hash / stale config 的旧行因不匹配传入的 `(qid, hash, config)`
+天然被排除，绝不泄漏进最终题库。

tests/unit/test_adversarial_verdicts_store.py

@@ -0,0 +1,93 @@
+"""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 = {
+        "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")
+    # 旧 config 行必须被真正删除（不能只靠 cfg2 查空——no-op 也满足那个弱断言）
+    assert store.completed_stages("v1_Action Recognition_0001", "h1", "cfg1") == set()
+    cfg1_rows = store._conn.execute(
+        "SELECT COUNT(*) FROM adversarial_verdicts WHERE agent_config='cfg1'"
+    ).fetchone()[0]
+    assert cfg1_rows == 0
+    assert store.completed_stages("v1_Action Recognition_0001", "h1", "cfg2") == set()
+    store.close()
+
+
+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_final_passed_question_ids_survives_both_gates(tmp_path):
+    store = _store(tmp_path)
+    # q1 太简单被作弊门剔除；q2 过两门；q3 被翻转门剔除（filtered_no_flip）
+    store.record_verdict(**_row(question_id="q1", question_hash="a", stage="cheat",
+                                verdict="filtered_too_easy"))
+    store.record_verdict(**_row(question_id="q2", question_hash="b", stage="cheat",
+                                verdict="passed"))
+    store.record_verdict(**_row(question_id="q3", question_hash="c", stage="cheat",
+                                verdict="passed"))
+    store.record_verdict(**_row(question_id="q3", question_hash="c", stage="flip_mirror",
+                                verdict="filtered_no_flip"))
+    passed = store.final_passed_question_ids(
+        {"q1": "a", "q2": "b", "q3": "c"}, "cfg1"
+    )
+    assert passed == {"q2"}  # 仅 q2：cheat=passed 且无 filtered_no_flip
+    # stale hash 不泄漏（当前 hash 不匹配旧行）
+    assert store.final_passed_question_ids({"q2": "stale"}, "cfg1") == set()
+    # stale config 不泄漏
+    assert store.final_passed_question_ids({"q2": "b"}, "cfgX") == set()
+    store.close()