"""SQLite 出题管线日志记录器 — 记录批次运行与逐题门判定。 写入频率低（每批次 ~240 题），使用同步 sqlite3 即可。 schema 定义参照 research-wiki/schemas/question-gen-runs.md 和 question-gen-items.md。 """ from __future__ import annotations import sqlite3 from dataclasses import dataclass from datetime import UTC, datetime from typing import TYPE_CHECKING, Protocol, runtime_checkable from loguru import logger if TYPE_CHECKING: from pathlib import Path # --------------------------------------------------------------------------- # GateReport Protocol（Task 6 尚未实现，此处声明 duck-type 接口） # --------------------------------------------------------------------------- @runtime_checkable class _GateResultLike(Protocol): """单门判定结果的最小接口。""" @property def verdict(self) -> object: """PASS / FAIL / SKIP 枚举值，.value 为小写字符串。""" ... @property def reason(self) -> str: """判定理由。""" ... @runtime_checkable class GateReportLike(Protocol): """GateReport 的最小 duck-type 接口。""" @property def key_verify(self) -> _GateResultLike: ... @property def blind_answer(self) -> _GateResultLike: ... @property def multi_true(self) -> _GateResultLike: ... @property def leak_test(self) -> _GateResultLike: ... @property def passed(self) -> bool: ... @property def reject_reason(self) -> str | None: ... # --------------------------------------------------------------------------- # 数据类 # --------------------------------------------------------------------------- @dataclass(frozen=True) class RunStats: """批次运行的统计摘要。 Attributes ---------- total_slots : int 目标题数（slot 总数）。 accepted : int 最终通过门判定的题数。 rejected : int 最终被拒绝的题数。 heavy_sampled : int 进行重量抽检的题数。 """ total_slots: int accepted: int rejected: int heavy_sampled: int # --------------------------------------------------------------------------- # DDL # --------------------------------------------------------------------------- # DEPRECATED(v3): v2 生成落库，Phase 2 生成替换后删；过渡期与 v3 表并存 _DDL_RUNS = """ CREATE TABLE IF NOT EXISTS question_gen_runs ( run_id TEXT PRIMARY KEY, git_sha TEXT NOT NULL, config_snapshot TEXT, started_at TEXT NOT NULL, ended_at TEXT, status TEXT NOT NULL DEFAULT 'running', total_slots INTEGER, accepted INTEGER, rejected INTEGER, heavy_sampled INTEGER ); """ # DEPRECATED(v3): v2 生成落库，Phase 2 生成替换后删；过渡期与 v3 表并存 _DDL_ITEMS = """ CREATE TABLE IF NOT EXISTS question_gen_items ( item_id TEXT PRIMARY KEY, run_id TEXT NOT NULL REFERENCES question_gen_runs(run_id), slot_id TEXT NOT NULL, video_id TEXT NOT NULL, family TEXT NOT NULL, task_type TEXT NOT NULL, skill_target TEXT NOT NULL, attempt INTEGER NOT NULL, question_text TEXT NOT NULL, sub_pattern TEXT, gate_key_verify TEXT, gate_blind_answer TEXT, gate_multi_true TEXT, gate_leak_test TEXT, gate_reject_reason TEXT, final_status TEXT NOT NULL DEFAULT 'pending', difficulty_steps INTEGER, selector_scores TEXT, created_at TEXT NOT NULL DEFAULT (datetime('now')) ); """ _DDL_INDEXES = [ "CREATE INDEX IF NOT EXISTS idx_qgr_status ON question_gen_runs(status);", "CREATE INDEX IF NOT EXISTS idx_qgi_run ON question_gen_items(run_id);", "CREATE INDEX IF NOT EXISTS idx_qgi_slot ON question_gen_items(slot_id);", "CREATE INDEX IF NOT EXISTS idx_qgi_status ON question_gen_items(final_status);", "CREATE INDEX IF NOT EXISTS idx_qgi_family ON question_gen_items(family);", "CREATE INDEX IF NOT EXISTS idx_qgi_task_type ON question_gen_items(task_type);", ] # DEPRECATED(v3): v2 生成落库，Phase 2 生成替换后删；过渡期与 v3 表并存 _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);", ] # --------------------------------------------------------------------------- # v3 出题领域观测表 DDL（facts/unit_verdict/collapse_metrics/quarantine/resume_state） # # 时间戳边界（Codex I-4）：v3 新表的 ts 列一律由 insert 方法外部传入，DDL 中不设 # DEFAULT (datetime('now'))、方法体不调 datetime.now()，以保证观测幂等可复现—— # 断点续跑重放时不因进程内时钟漂移产生脏数据。resume_state 无 ts 列（schema 未列）。 # --------------------------------------------------------------------------- _DDL_FACTS = """ CREATE TABLE IF NOT EXISTS facts ( fact_id TEXT PRIMARY KEY, video_id TEXT, segment_id TEXT, subject TEXT, action TEXT, object TEXT, frame_ids TEXT, polarity TEXT, fact_type TEXT, difficulty_tier INTEGER, verifier_refs TEXT, cross_agree INTEGER CHECK(cross_agree IN (0, 1)), negative_at_target TEXT, session_id TEXT, ts TEXT ); """ _DDL_UNIT_VERDICT = """ CREATE TABLE IF NOT EXISTS unit_verdict ( unit_id TEXT NOT NULL, pair_id TEXT, sub_pattern TEXT, stage INTEGER NOT NULL CHECK(stage BETWEEN 1 AND 6), verdict TEXT CHECK(verdict IN ('pass', 'fail', 'abstain')), reason TEXT, metric_value REAL, model TEXT, session_id TEXT, ts TEXT, PRIMARY KEY (unit_id, stage) ); """ _DDL_COLLAPSE_METRICS = """ CREATE TABLE IF NOT EXISTS collapse_metrics ( pair_id TEXT PRIMARY KEY, text_only_acc REAL, single_frame_acc REAL, placebo_drop REAL, majority_vote_hit REAL, slot_chi2 REAL, distractor_min_dist REAL, multiformat_consistency REAL, subtitle_answerability REAL, ts TEXT ); """ _DDL_QUARANTINE = """ CREATE TABLE IF NOT EXISTS quarantine ( content_fingerprint TEXT PRIMARY KEY, sub_pattern TEXT, quarantine_reason TEXT, round_no INTEGER, ts TEXT ); """ _DDL_RESUME_STATE = """ CREATE TABLE IF NOT EXISTS resume_state ( unit_id TEXT PRIMARY KEY, status TEXT CHECK(status IN ('pending', 'accepted', 'rejected')), config_fingerprint TEXT, seq_offset INTEGER ); """ _DDL_V3_TABLES = [ _DDL_FACTS, _DDL_UNIT_VERDICT, _DDL_COLLAPSE_METRICS, _DDL_QUARANTINE, _DDL_RESUME_STATE, ] _DDL_V3_INDEXES = [ "CREATE INDEX IF NOT EXISTS idx_facts_video ON facts(video_id);", "CREATE INDEX IF NOT EXISTS idx_facts_session ON facts(session_id);", # unit_verdict(unit_id) 无需独立索引：主键 (unit_id, stage) 的左前缀已覆盖 unit_id 查询。 "CREATE INDEX IF NOT EXISTS idx_uv_sub_pattern ON unit_verdict(sub_pattern);", "CREATE INDEX IF NOT EXISTS idx_quar_sub_pattern ON quarantine(sub_pattern);", "CREATE INDEX IF NOT EXISTS idx_resume_status ON resume_state(status);", ] # --------------------------------------------------------------------------- # Store 实现 # --------------------------------------------------------------------------- class QuestionGenStore: """SQLite 出题管线日志记录器。 记录每次出题批次的元数据（run）和每题的门判定结果（item）。 使用同步 sqlite3，写入频率低无需异步。 Parameters ---------- db_path : Path SQLite 数据库文件路径。父目录必须存在。 """ def __init__(self, db_path: Path) -> None: self._db_path = db_path self._conn = sqlite3.connect( str(db_path), check_same_thread=False, timeout=10.0, ) self._conn.execute("PRAGMA journal_mode=WAL") self._conn.execute("PRAGMA busy_timeout=5000") self._conn.execute("PRAGMA foreign_keys=ON") self._init_schema() logger.debug("QuestionGenStore 已初始化: {}", db_path) def _init_schema(self) -> None: """幂等创建表和索引。多次调用安全。""" self._conn.execute(_DDL_RUNS) 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) for ddl in _DDL_V3_TABLES: self._conn.execute(ddl) for idx_sql in _DDL_V3_INDEXES: self._conn.execute(idx_sql) self._conn.commit() # 幂等迁移：为已有表加 sub_pattern / selector_scores 列 cols = {r[1] for r in self._conn.execute("PRAGMA table_info(question_gen_items)")} if "sub_pattern" not in cols: self._conn.execute("ALTER TABLE question_gen_items ADD COLUMN sub_pattern TEXT") self._conn.commit() if "selector_scores" not in cols: self._conn.execute("ALTER TABLE question_gen_items ADD COLUMN selector_scores TEXT") self._conn.commit() def record_run_start(self, run_id: str, git_sha: str, config_snapshot: str) -> None: """记录批次开始。 Parameters ---------- run_id : str 批次唯一标识（UUID）。 git_sha : str 当前代码版本 HEAD commit。 config_snapshot : str 科研配置快照（JSON 序列化字符串）。 """ now = datetime.now(tz=UTC).isoformat(timespec="seconds") self._conn.execute( """ INSERT INTO question_gen_runs (run_id, git_sha, config_snapshot, started_at, status) VALUES (?, ?, ?, ?, 'running') """, (run_id, git_sha, config_snapshot, now), ) self._conn.commit() logger.info("出题批次已开始: run_id={}", run_id) def record_run_end(self, run_id: str, status: str, stats: RunStats) -> None: """记录批次结束，更新状态与统计。 Parameters ---------- run_id : str 批次唯一标识。 status : str 最终状态（completed / failed）。 stats : RunStats 批次统计摘要。 """ now = datetime.now(tz=UTC).isoformat(timespec="seconds") cursor = self._conn.execute( """ UPDATE question_gen_runs SET ended_at=?, status=?, total_slots=?, accepted=?, rejected=?, heavy_sampled=? WHERE run_id=? """, ( now, status, stats.total_slots, stats.accepted, stats.rejected, stats.heavy_sampled, run_id, ), ) self._conn.commit() if cursor.rowcount == 0: raise ValueError(f"run_id 不存在: {run_id}") logger.info( "出题批次已结束: run_id={}, status={}, accepted={}/{}", run_id, status, stats.accepted, stats.total_slots, ) def record_item( self, item_id: str, run_id: str, slot_id: str, video_id: str, family: str, task_type: str, skill_target: str, attempt: int, question_text: str, sub_pattern: str | None = None, ) -> None: """记录一道新生成的题目（初始状态 pending）。 Parameters ---------- item_id : str 题目唯一 ID（每轮独立）。 run_id : str 关联的批次 ID。 slot_id : str 逻辑 slot 标识（同 slot 多次重出共享）。 video_id : str 视频 ID。 family : str 题族（retrieval/reasoning/enumeration/visual/spatial）。 task_type : str Video-MME 12 类主标签。 skill_target : str M1-M5 + 题族子标签。 attempt : int 当前重出轮次（1-based）。 question_text : str 题目文本。 sub_pattern : str | None 子模式标识（如有）。 """ now = datetime.now(tz=UTC).isoformat(timespec="seconds") self._conn.execute( """ INSERT INTO question_gen_items (item_id, run_id, slot_id, video_id, family, task_type, skill_target, attempt, question_text, sub_pattern, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( item_id, run_id, slot_id, video_id, family, task_type, skill_target, attempt, question_text, sub_pattern, now, ), ) self._conn.commit() def update_gates(self, item_id: str, report: GateReportLike) -> None: """更新门判定结果及 final_status。 Parameters ---------- item_id : str 题目唯一 ID。 report : GateReportLike 门判定报告（duck-type，需有 key_verify/blind_answer/multi_true/leak_test 属性，每个属性具有 .verdict.value 和 .reason；以及 passed/reject_reason 属性）。 """ final_status = "accepted" if report.passed else "rejected" cursor = self._conn.execute( """ UPDATE question_gen_items SET gate_key_verify=?, gate_blind_answer=?, gate_multi_true=?, gate_leak_test=?, gate_reject_reason=?, final_status=? WHERE item_id=? """, ( report.key_verify.verdict.value, report.blind_answer.verdict.value, report.multi_true.verdict.value, report.leak_test.verdict.value, report.reject_reason, final_status, item_id, ), ) self._conn.commit() if cursor.rowcount == 0: raise ValueError(f"item_id 不存在: {item_id}") def mark_item_rejected(self, item_id: str, reason: str) -> None: """将已记录的 item 标记为 rejected（用于门控外的拒绝场景，如去重）。 Parameters ---------- item_id : str 题目唯一 ID。 reason : str 拒绝原因描述。 """ cursor = self._conn.execute( "UPDATE question_gen_items SET final_status='rejected', gate_reject_reason=? " "WHERE item_id=?", (reason, item_id), ) self._conn.commit() if cursor.rowcount == 0: raise ValueError(f"item_id 不存在: {item_id}") def update_difficulty(self, item_id: str, difficulty_steps: int) -> None: """更新重量抽检产出的 Agent 步数。 Parameters ---------- item_id : str 题目唯一 ID。 difficulty_steps : int Agent 完成该题所需步数。 """ cursor = self._conn.execute( "UPDATE question_gen_items SET difficulty_steps=? WHERE item_id=?", (difficulty_steps, item_id), ) self._conn.commit() 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 字符串）。 Parameters ---------- item_id : str 题目唯一 ID。 selector_scores_json : str 观测 JSON：correct_score / chosen / pool_size / anneal_rounds / hard_fail。 Raises ------ ValueError item_id 不存在时抛出。 """ cursor = self._conn.execute( "UPDATE question_gen_items SET selector_scores=? WHERE item_id=?", (selector_scores_json, item_id), ) self._conn.commit() if cursor.rowcount == 0: raise ValueError(f"item_id 不存在: {item_id}") def get_run_stats(self, run_id: str) -> RunStats: """查询批次统计摘要。 Parameters ---------- run_id : str 批次唯一标识。 Returns ------- RunStats 该批次的统计数据。 Raises ------ ValueError run_id 不存在时抛出。 """ row = self._conn.execute( "SELECT total_slots, accepted, rejected, heavy_sampled " "FROM question_gen_runs WHERE run_id=?", (run_id,), ).fetchone() if row is None: raise ValueError(f"run_id 不存在: {run_id}") return RunStats( total_slots=row[0] or 0, accepted=row[1] or 0, rejected=row[2] or 0, heavy_sampled=row[3] or 0, ) def load_progress(self) -> dict[str, str]: """加载已接受 slot 的进度映射（用于断点续跑）。 从最近一次 running 状态的批次中，只读取 accepted 的 slot。 rejected 的 slot 不纳入 progress，以便重跑时重新尝试。 Returns ------- dict[str, str] {slot_id: "accepted"} 映射。无进度时返回空 dict。 """ row = self._conn.execute( "SELECT run_id FROM question_gen_runs WHERE status='running' " "ORDER BY started_at DESC LIMIT 1", ).fetchone() if row is None: return {} run_id = row[0] rows = self._conn.execute( "SELECT DISTINCT slot_id FROM question_gen_items " "WHERE run_id=? AND final_status='accepted'", (run_id,), ).fetchall() return {row[0]: "accepted" for row in rows} # ----------------------------------------------------------------------- # v3 出题领域观测写入（ts 一律外部传入，禁进程内 now，保幂等可复现） # ----------------------------------------------------------------------- def insert_fact( self, *, fact_id: str, video_id: str, segment_id: str, subject: str, action: str, object: str, # noqa: A002 — 与设计列名 object 一致，仅 kwargs 传入无遮蔽风险 frame_ids: str, polarity: str, fact_type: str, difficulty_tier: int, verifier_refs: str, cross_agree: int, negative_at_target: str, session_id: str, ts: str, ) -> None: """写入一条帧感知抽取 Fact（facts 表，主键 fact_id）。 Parameters ---------- fact_id : str Fact 唯一标识（UUID）。 video_id, segment_id : str 溯源：所属视频与片段。 subject, action, object : str 结构化绑定三元组。 frame_ids : str 感知所用帧 ID 的 JSON 数组字符串。 polarity : str 事实极性（真/假）。 fact_type : str 事实类型（binding/state/manner/order/evidence）。 difficulty_tier : int 感知难度分层。 verifier_refs : str 双 VLM（qwen/MiniMax）各自裁决的 JSON 字符串。 cross_agree : int 双 VLM 是否一致（0/1）。 negative_at_target : str 目标点为假的核实结果。 session_id : str epoch/step 关联 ID。 ts : str 观测时间戳 ISO 字符串，由调用方外部传入（禁进程内 now）。 """ self._conn.execute( """ INSERT INTO facts (fact_id, video_id, segment_id, subject, action, object, frame_ids, polarity, fact_type, difficulty_tier, verifier_refs, cross_agree, negative_at_target, session_id, ts) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( fact_id, video_id, segment_id, subject, action, object, frame_ids, polarity, fact_type, difficulty_tier, verifier_refs, cross_agree, negative_at_target, session_id, ts, ), ) self._conn.commit() logger.debug("facts 已写入: fact_id={}", fact_id) def insert_unit_verdict( self, *, unit_id: str, pair_id: str | None, sub_pattern: str, stage: int, verdict: str, reason: str, metric_value: float | None, model: str | None, session_id: str, ts: str, ) -> None: """写入六层验证某一层的裁决（unit_verdict 表，主键 (unit_id, stage)）。 同一 (unit_id, stage) 重跑时以 upsert 覆盖旧行——每 unit 每层仅保留最新一条， 便于断点续跑重放而不残留过期裁决。 覆盖语义：ON CONFLICT DO UPDATE 用 ``excluded.*`` **整行覆盖**全部非键列 （含可空的 metric_value/model 及 ts）——只保留每层最新完整裁决，不支持部分字段 更新。调用方每次必须传完整行，否则会用 metric_value=None/model=None 误抹先前非空值。 Parameters ---------- unit_id : str pair/single 单元标识。 pair_id : str | None 所属 pair 关联标识（single 单元可为 None）。 sub_pattern : str 6 子模式之一。 stage : int 验证层号（1-6）。 verdict : str 该层裁决（pass/fail/abstain）。 reason : str 拒因描述。 metric_value : float | None 该层量化值（无量化值时为 None）。 model : str | None 裁判模型名（无模型判定时为 None）。 session_id : str epoch/step 关联 ID。 ts : str 观测时间戳 ISO 字符串，由调用方外部传入（禁进程内 now）。 """ self._conn.execute( """ INSERT INTO unit_verdict (unit_id, pair_id, sub_pattern, stage, verdict, reason, metric_value, model, session_id, ts) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(unit_id, stage) DO UPDATE SET pair_id=excluded.pair_id, sub_pattern=excluded.sub_pattern, verdict=excluded.verdict, reason=excluded.reason, metric_value=excluded.metric_value, model=excluded.model, session_id=excluded.session_id, ts=excluded.ts """, ( unit_id, pair_id, sub_pattern, stage, verdict, reason, metric_value, model, session_id, ts, ), ) self._conn.commit() logger.debug("unit_verdict 已写入: unit_id={}, stage={}", unit_id, stage) def insert_collapse_metrics( self, *, pair_id: str, text_only_acc: float, single_frame_acc: float, placebo_drop: float, majority_vote_hit: float, slot_chi2: float, distractor_min_dist: float, multiformat_consistency: float, subtitle_answerability: float, ts: str, ) -> None: """写入配对坍缩度量（collapse_metrics 表，主键 pair_id）。 同一 pair_id 重算时以 upsert 覆盖旧行——每 pair 仅保留最新一组度量。 覆盖语义：ON CONFLICT DO UPDATE 用 ``excluded.*`` **整行覆盖**全部非键列（含 ts）， 不支持部分字段更新；调用方每次必须传完整度量行。 Parameters ---------- pair_id : str 配对唯一标识。 text_only_acc, single_frame_acc, placebo_drop : float 模型探针类度量。 majority_vote_hit, slot_chi2, distractor_min_dist : float 纯结构类度量。 multiformat_consistency, subtitle_answerability : float 探针类度量。 ts : str 观测时间戳 ISO 字符串，由调用方外部传入（禁进程内 now）。 """ self._conn.execute( """ INSERT INTO collapse_metrics (pair_id, text_only_acc, single_frame_acc, placebo_drop, majority_vote_hit, slot_chi2, distractor_min_dist, multiformat_consistency, subtitle_answerability, ts) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(pair_id) DO UPDATE SET text_only_acc=excluded.text_only_acc, single_frame_acc=excluded.single_frame_acc, placebo_drop=excluded.placebo_drop, majority_vote_hit=excluded.majority_vote_hit, slot_chi2=excluded.slot_chi2, distractor_min_dist=excluded.distractor_min_dist, multiformat_consistency=excluded.multiformat_consistency, subtitle_answerability=excluded.subtitle_answerability, ts=excluded.ts """, ( pair_id, text_only_acc, single_frame_acc, placebo_drop, majority_vote_hit, slot_chi2, distractor_min_dist, multiformat_consistency, subtitle_answerability, ts, ), ) self._conn.commit() logger.debug("collapse_metrics 已写入: pair_id={}", pair_id) def quarantine( self, *, content_fingerprint: str, sub_pattern: str, quarantine_reason: str, round_no: int, ts: str, ) -> None: """将失败题的内容指纹写入隔离区黑名单（quarantine 表，主键 content_fingerprint）。 同一 content_fingerprint 重复调用以 upsert 覆盖——保证语义相同题面只占一行， 补构造前查此表当黑名单去重。覆盖语义：ON CONFLICT DO UPDATE 用 ``excluded.*`` **整行覆盖**全部非键列（含 ts），不支持部分字段更新。 Parameters ---------- content_fingerprint : str 题面语义内容指纹（去重键）。 sub_pattern : str 题目所属子模式。 quarantine_reason : str 隔离原因。 round_no : int 隔离发生的过滤轮次。 ts : str 观测时间戳 ISO 字符串，由调用方外部传入（禁进程内 now）。 """ self._conn.execute( """ INSERT INTO quarantine (content_fingerprint, sub_pattern, quarantine_reason, round_no, ts) VALUES (?, ?, ?, ?, ?) ON CONFLICT(content_fingerprint) DO UPDATE SET sub_pattern=excluded.sub_pattern, quarantine_reason=excluded.quarantine_reason, round_no=excluded.round_no, ts=excluded.ts """, (content_fingerprint, sub_pattern, quarantine_reason, round_no, ts), ) self._conn.commit() logger.debug("quarantine 已写入: fingerprint={}", content_fingerprint) def upsert_resume_state( self, *, unit_id: str, status: str, config_fingerprint: str, seq_offset: int, ) -> None: """写入/更新断点续跑状态（resume_state 表，主键 unit_id）。 同一 unit_id 覆盖更新——每单元仅保留最新续跑状态。config_fingerprint 变化时 由调用方据此判定旧进度作废并重跑。本表无 ts 列（schema 未定义），但方法体仍禁 进程内 now，保持 v3 观测一致的幂等可复现语义。 覆盖语义：ON CONFLICT DO UPDATE 用 ``excluded.*`` **整行覆盖**全部非键列，不支持 部分字段更新；调用方每次必须传完整状态行。 Parameters ---------- unit_id : str 单元唯一标识。 status : str 续跑状态（pending/accepted/rejected）。 config_fingerprint : str 求解器/裁判 config 指纹，变更时旧进度作废。 seq_offset : int 补构造续编偏移，防止编号相撞。 """ self._conn.execute( """ INSERT INTO resume_state (unit_id, status, config_fingerprint, seq_offset) VALUES (?, ?, ?, ?) ON CONFLICT(unit_id) DO UPDATE SET status=excluded.status, config_fingerprint=excluded.config_fingerprint, seq_offset=excluded.seq_offset """, (unit_id, status, config_fingerprint, seq_offset), ) self._conn.commit() logger.debug("resume_state 已写入: unit_id={}, status={}", unit_id, status) def close(self) -> None: """关闭数据库连接。""" self._conn.close() logger.debug("QuestionGenStore 已关闭")