From d77cbc95eb61c4a06b51ebc317a503544bb2260f Mon Sep 17 00:00:00 2001 From: iomgaa Date: Tue, 14 Jul 2026 15:35:11 -0400 Subject: [PATCH] feat: add adversarial_verdicts table with resume and terminal-verdict query Co-Authored-By: Claude Opus 4.8 (1M context) --- app/question_gen/run_store.py | 185 ++++++++++++++++++ research-wiki/schemas/adversarial-verdicts.md | 86 ++++++++ tests/unit/test_adversarial_verdicts_store.py | 93 +++++++++ 3 files changed, 364 insertions(+) create mode 100644 research-wiki/schemas/adversarial-verdicts.md create mode 100644 tests/unit/test_adversarial_verdicts_store.py diff --git a/app/question_gen/run_store.py b/app/question_gen/run_store.py index 2bca609..77b78c5 100644 --- a/app/question_gen/run_store.py +++ b/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 字符串）。 diff --git a/research-wiki/schemas/adversarial-verdicts.md b/research-wiki/schemas/adversarial-verdicts.md new file mode 100644 index 0000000..85f57a2 --- /dev/null +++ b/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)` +天然被排除，绝不泄漏进最终题库。 diff --git a/tests/unit/test_adversarial_verdicts_store.py b/tests/unit/test_adversarial_verdicts_store.py new file mode 100644 index 0000000..da60d27 --- /dev/null +++ b/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()