diff --git a/app/question_gen/run_store.py b/app/question_gen/run_store.py index 9473d1f..4412bb6 100644 --- a/app/question_gen/run_store.py +++ b/app/question_gen/run_store.py @@ -183,7 +183,7 @@ CREATE TABLE IF NOT EXISTS facts ( fact_type TEXT, difficulty_tier INTEGER, verifier_refs TEXT, - cross_agree INTEGER, + cross_agree INTEGER CHECK(cross_agree IN (0, 1)), negative_at_target TEXT, session_id TEXT, ts TEXT @@ -195,8 +195,8 @@ CREATE TABLE IF NOT EXISTS unit_verdict ( unit_id TEXT NOT NULL, pair_id TEXT, sub_pattern TEXT, - stage INTEGER NOT NULL, - verdict 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, @@ -234,7 +234,7 @@ CREATE TABLE IF NOT EXISTS quarantine ( _DDL_RESUME_STATE = """ CREATE TABLE IF NOT EXISTS resume_state ( unit_id TEXT PRIMARY KEY, - status TEXT, + status TEXT CHECK(status IN ('pending', 'accepted', 'rejected')), config_fingerprint TEXT, seq_offset INTEGER ); @@ -251,7 +251,7 @@ _DDL_V3_TABLES = [ _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);", - "CREATE INDEX IF NOT EXISTS idx_uv_unit ON unit_verdict(unit_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);", @@ -848,6 +848,10 @@ class QuestionGenStore: 同一 (unit_id, stage) 重跑时以 upsert 覆盖旧行——每 unit 每层仅保留最新一条， 便于断点续跑重放而不残留过期裁决。 + 覆盖语义：ON CONFLICT DO UPDATE 用 ``excluded.*`` **整行覆盖**全部非键列 + （含可空的 metric_value/model 及 ts）——只保留每层最新完整裁决，不支持部分字段 + 更新。调用方每次必须传完整行，否则会用 metric_value=None/model=None 误抹先前非空值。 + Parameters ---------- unit_id : str @@ -921,6 +925,9 @@ class QuestionGenStore: 同一 pair_id 重算时以 upsert 覆盖旧行——每 pair 仅保留最新一组度量。 + 覆盖语义：ON CONFLICT DO UPDATE 用 ``excluded.*`` **整行覆盖**全部非键列（含 ts）， + 不支持部分字段更新；调用方每次必须传完整度量行。 + Parameters ---------- pair_id : str @@ -980,7 +987,8 @@ class QuestionGenStore: """将失败题的内容指纹写入隔离区黑名单（quarantine 表，主键 content_fingerprint）。 同一 content_fingerprint 重复调用以 upsert 覆盖——保证语义相同题面只占一行， - 补构造前查此表当黑名单去重。 + 补构造前查此表当黑名单去重。覆盖语义：ON CONFLICT DO UPDATE 用 ``excluded.*`` + **整行覆盖**全部非键列（含 ts），不支持部分字段更新。 Parameters ---------- @@ -1025,6 +1033,9 @@ class QuestionGenStore: 由调用方据此判定旧进度作废并重跑。本表无 ts 列（schema 未定义），但方法体仍禁 进程内 now，保持 v3 观测一致的幂等可复现语义。 + 覆盖语义：ON CONFLICT DO UPDATE 用 ``excluded.*`` **整行覆盖**全部非键列，不支持 + 部分字段更新；调用方每次必须传完整状态行。 + Parameters ---------- unit_id : str diff --git a/tests/unit/test_run_store_v3_tables.py b/tests/unit/test_run_store_v3_tables.py index 2e57375..0e6c6b6 100644 --- a/tests/unit/test_run_store_v3_tables.py +++ b/tests/unit/test_run_store_v3_tables.py @@ -6,6 +6,7 @@ from __future__ import annotations +import sqlite3 from typing import TYPE_CHECKING import pytest @@ -173,19 +174,22 @@ def test_insert_unit_verdict_rerun_same_stage_overwrites(store: QuestionGenStore ts="2026-07-15T11:10:00+00:00", **common, ) + new_ts = "2026-07-15T11:20:00+00:00" store.insert_unit_verdict( verdict="pass", reason="second", metric_value=0.9, model="m2", - ts="2026-07-15T11:20:00+00:00", + ts=new_ts, **common, ) rows = store._conn.execute( - "SELECT verdict, reason, metric_value FROM unit_verdict WHERE unit_id='unit-3' AND stage=2" + "SELECT verdict, reason, metric_value, ts " + "FROM unit_verdict WHERE unit_id='unit-3' AND stage=2" ).fetchall() assert len(rows) == 1 - assert rows[0] == ("pass", "second", 0.9) + # ts 随整行覆盖为第二次（新）ts，锁死 DO UPDATE SET ts=excluded.ts 语义。 + assert rows[0] == ("pass", "second", 0.9, new_ts) # --------------------------------------------------------------------------- @@ -241,6 +245,7 @@ def test_insert_collapse_metrics_recompute_upserts(store: QuestionGenStore) -> N subtitle_answerability=0.1, ts="2026-07-15T12:00:00+00:00", ) + new_ts = "2026-07-15T12:30:00+00:00" store.insert_collapse_metrics( pair_id="pair-2", text_only_acc=0.05, @@ -251,13 +256,14 @@ def test_insert_collapse_metrics_recompute_upserts(store: QuestionGenStore) -> N distractor_min_dist=0.9, multiformat_consistency=0.95, subtitle_answerability=0.0, - ts="2026-07-15T12:30:00+00:00", + ts=new_ts, ) rows = store._conn.execute( - "SELECT text_only_acc, slot_chi2 FROM collapse_metrics WHERE pair_id='pair-2'" + "SELECT text_only_acc, slot_chi2, ts FROM collapse_metrics WHERE pair_id='pair-2'" ).fetchall() assert len(rows) == 1 - assert rows[0] == (0.05, 9.0) + # ts 随整行覆盖为第二次（新）ts，锁死 DO UPDATE SET ts=excluded.ts 语义。 + assert rows[0] == (0.05, 9.0, new_ts) # --------------------------------------------------------------------------- @@ -346,6 +352,78 @@ def test_upsert_resume_state_same_unit_overwrites(store: QuestionGenStore) -> No assert rows[0] == ("accepted", "cfg-2", 5) +# --------------------------------------------------------------------------- +# CHECK 约束 fail-loud（非法枚举写入直接报错，符合 §4.2/P5） +# --------------------------------------------------------------------------- + + +def test_unit_verdict_invalid_stage_raises(store: QuestionGenStore) -> None: + """非法 stage（超出 1-6）触发 CHECK 约束，写入直接 IntegrityError。""" + with pytest.raises(sqlite3.IntegrityError): + store.insert_unit_verdict( + unit_id="unit-bad", + pair_id=None, + sub_pattern="order", + stage=7, + verdict="pass", + reason="r", + metric_value=None, + model=None, + session_id="sess-1", + ts="2026-07-15T11:00:00+00:00", + ) + + +def test_unit_verdict_invalid_verdict_raises(store: QuestionGenStore) -> None: + """非法 verdict（不在 pass/fail/abstain）触发 CHECK 约束，写入直接 IntegrityError。""" + with pytest.raises(sqlite3.IntegrityError): + store.insert_unit_verdict( + unit_id="unit-bad2", + pair_id=None, + sub_pattern="order", + stage=1, + verdict="maybe", + reason="r", + metric_value=None, + model=None, + session_id="sess-1", + ts="2026-07-15T11:00:00+00:00", + ) + + +def test_facts_invalid_cross_agree_raises(store: QuestionGenStore) -> None: + """非法 cross_agree（非 0/1）触发 CHECK 约束，写入直接 IntegrityError。""" + with pytest.raises(sqlite3.IntegrityError): + store.insert_fact( + fact_id="fact-bad", + video_id="v", + segment_id="s", + subject="a", + action="b", + object="c", + frame_ids="[]", + polarity="真", + fact_type="state", + difficulty_tier=1, + verifier_refs="{}", + cross_agree=2, + negative_at_target="x", + session_id="sess", + ts="2026-07-15T11:00:00+00:00", + ) + + +def test_resume_state_invalid_status_raises(store: QuestionGenStore) -> None: + """非法 status（不在 pending/accepted/rejected）触发 CHECK 约束，写入直接 IntegrityError。""" + with pytest.raises(sqlite3.IntegrityError): + store.upsert_resume_state( + unit_id="unit-bad", + status="unknown", + config_fingerprint="cfg", + seq_offset=0, + ) + + # --------------------------------------------------------------------------- # v2 旧表未受影响 # ---------------------------------------------------------------------------