From d4e9852864138f80530e40b9a60706cb2b301162 Mon Sep 17 00:00:00 2001 From: iomgaa Date: Tue, 14 Jul 2026 16:00:14 -0400 Subject: [PATCH] feat: add cheater gate with resume-safe survivor recovery --- app/question_gen/adversarial_filter.py | 119 ++++++++++++++++++++ tests/unit/test_adversarial_cheater_gate.py | 87 ++++++++++++++ 2 files changed, 206 insertions(+) create mode 100644 tests/unit/test_adversarial_cheater_gate.py diff --git a/app/question_gen/adversarial_filter.py b/app/question_gen/adversarial_filter.py index 334be91..ec8eef3 100644 --- a/app/question_gen/adversarial_filter.py +++ b/app/question_gen/adversarial_filter.py @@ -12,6 +12,14 @@ from __future__ import annotations import enum import hashlib import json +from typing import TYPE_CHECKING, Protocol + +from loguru import logger + +if TYPE_CHECKING: + from app.question_gen.adversarial_config import AdversarialFilterConfig + from app.question_gen.run_store import QuestionGenStore + from core.types import GeneratedQuestion _VALID_LETTERS = ("A", "B", "C", "D") @@ -94,3 +102,114 @@ def judge_flip(*, p_text: str | None, q_text: str | None) -> FlipDecision: if p_text.strip() != q_text.strip(): return FlipDecision.PASSED return FlipDecision.FILTERED_NO_FLIP + + +class AgentRunner(Protocol): + """完整 inference agent 试答端口 — Phase B 只依赖此接口（便于 mock）。 + + 实现见 Task 10 的 _RealAgentRunner（复用 run_inference + RunLogImpl）。 + """ + + model: str + skill_mode: str # 从首次定义即入 Protocol，保证 Task 6/8/10 指纹口径一致 + + async def predict( + self, + questions: list[GeneratedQuestion], + *, + max_steps: int, + run_id: str, + ) -> dict[str, str | None]: + """跑完整 agent，返回 question_id → 预测答案字母（无预测为 None）。""" + ... + + +def _cheat_hash(question: GeneratedQuestion) -> str: + """按散参数签名计算作弊门题面 hash（question/options/answer）。""" + return question_hash(question.question, question.options, question.answer) + + +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, _cheat_hash(q), cfg_fp), + ).fetchall() + if rows and rows[0][0] == 0: + survivors.append(q) + return survivors + + +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 预测字母已落 adversarial_verdicts 表 + （stage='cheat'），翻转门经 `_read_cheat_prediction` 从表读取复用（不重跑）。 + """ + cfg_fp = agent_config_fingerprint( + skill_mode=agent.skill_mode, + max_steps=config.adversarial_agent_max_steps, + model=agent.model, + ) + todo: list[GeneratedQuestion] = [] + completed: list[GeneratedQuestion] = [] + for q in questions: + h = _cheat_hash(q) + store.invalidate_stale_config(q.question_id, cfg_fp) + if "cheat" in store.completed_stages(q.question_id, h, cfg_fp): + completed.append(q) + else: + todo.append(q) + + # C1: 无条件先从已完成题恢复存活者（agent 答错），再对未判题跑 agent 追加。 + # 两者都流向翻转门——绝不因 todo 非空而丢掉已完成的存活者（混合续跑正确性）。 + survivors: list[GeneratedQuestion] = _recover_survivors(completed, store, cfg_fp) + + if todo: + preds = await agent.predict( + todo, max_steps=config.adversarial_agent_max_steps, run_id=run_id + ) + 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=_cheat_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(questions), len(completed), len(todo), len(survivors), + ) + return survivors diff --git a/tests/unit/test_adversarial_cheater_gate.py b/tests/unit/test_adversarial_cheater_gate.py new file mode 100644 index 0000000..62f7a40 --- /dev/null +++ b/tests/unit/test_adversarial_cheater_gate.py @@ -0,0 +1,87 @@ +"""作弊门：agent 答对=filtered_too_easy 并落表；答错=cheat verdict=passed 待翻转。""" + +import pytest + +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 +from core.types import GeneratedQuestion + + +class _FakeAgent: + """按 question_id → 预测字母返回的 mock AgentRunner。""" + + def __init__(self, preds: dict[str, str], model: str = "m1", skill_mode: str = "auto"): + self._preds = preds + self.model = model + self.skill_mode = skill_mode + self.calls: list[str] = [] + + async def predict(self, questions, *, max_steps, run_id): + 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() + + +@pytest.mark.asyncio +async def test_cheater_gate_resume_mixed_keeps_all_survivors(tmp_path): + """混合续跑：部分题已有 cheat verdict、部分未判——已完成的存活者不得被丢。""" + store = QuestionGenStore(str(tmp_path / "q.db")) + cfg = AdversarialFilterConfig() + # 第一轮：先只判 done_hard（答错=存活），落表 + agent1 = _FakeAgent({"done_hard": "B"}) # 答错(正解 A) + await run_cheater_gate([_q("done_hard")], agent=agent1, store=store, + config=cfg, round_no=0, run_id="r0") + # 第二轮：done_hard 已判 + 新题 new_hard 未判混在一起 + agent2 = _FakeAgent({"new_hard": "C"}) # 新题答错(正解 A)=存活 + survivors = await run_cheater_gate( + [_q("done_hard"), _q("new_hard")], agent=agent2, store=store, + config=cfg, round_no=0, run_id="r1", + ) + ids = {q.question_id for q in survivors} + assert ids == {"done_hard", "new_hard"} # 已完成存活者 done_hard 未被丢 + assert agent2.calls == ["new_hard"] # 只对未判题跑 agent + store.close()