refactor: add CHECK constraints and drop redundant index in v3 tables

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,
+    stage        INTEGER NOT NULL CHECK(stage BETWEEN 1 AND 6),
-    verdict      TEXT,
+    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

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 旧表未受影响
 # ---------------------------------------------------------------------------