框架(评测隔离): 实现六原则的引擎强制层

- DDL 103：run_type 插入后不可变触发器(评测性锚点钉死)；生产只读角色 muse_read + RLS(评测候选对生产读默认不可见，app.eval_visible opt-in 才可见)；已验证 root可见/muse_read不可见/opt-in可见
- invariant_checks.py：常设不变量(从靠预防迁到靠发现)——I1评测候选未接受/I2正文来源不追溯评测候选/I3评测质量结果可过滤/I4 oracle红线/I5 COMPLETED轮次封存(声明==实际落库)，已跑通全PASS
- 原则1/2(口号升级/代价明账)已落 SoT；原则3/4/5/6 实现层完成

.claude/skills/runtime/scripts/invariant_checks.py

@@ -0,0 +1,96 @@
+#!/usr/bin/env python3
+"""常设不变量检查（harness 改造原则 5：从靠预防迁到靠发现）。
+
+逻辑隔离不能全靠预防，必须靠持续发现——从物理隔离迁到逻辑隔离的那一刻，
+必须同时从"靠预防"迁到"靠发现"。本脚本跑一组常设不变量，违反即报警（退出码非 0）。
+建议常设运行（CI / 看板健康面），违反即报警。
+
+检查项：
+I1 没有评测候选被接受（run_type=eval/diagnostic 且 state=accepted）——引擎已防，此为发现兜底。
+I2 没有正式正文的来源能追溯到评测候选（归因 source_object_id → 评测候选）。
+I3 评测候选的质量结果可被生产视图过滤（带评测标记，不混入生产质量视图）。
+I4 oracle 读侧红线：oracle/标准答案以 kind='oracle' 标记，生产模型输入不得包含（read-context 生产路径保证；此处校验标记完整）。
+I5 COMPLETED 轮次封存：一轮声明的 raw 集合 == 实际落库集合（lease 声明的 content_hashes 数 == 实际 raw_content 行数）。
+
+跑法：.venv/bin/python .claude/skills/runtime/scripts/invariant_checks.py [--run-id X]
+"""
+import argparse
+import json
+import sys
+from pathlib import Path
+
+DB_SCRIPTS = Path(__file__).resolve().parents[2] / "db" / "scripts"
+sys.path.insert(0, str(DB_SCRIPTS))
+from db import connect  # noqa: E402
+
+
+def _count(conn, sql, params=()):
+    return conn.execute(sql, params).fetchone()[0]
+
+
+def check_invariants(run_id=None):
+    """返回 [(检查项, 是否通过, 说明)]。"""
+    results = []
+    with connect(readonly=True) as conn:
+        # I1 没有评测候选被接受
+        n = _count(conn, "SELECT count(*) FROM example_candidate "
+                         "WHERE run_type IN ('eval','diagnostic') AND state='accepted'")
+        results.append(("I1 评测候选未被接受", n == 0,
+                        f"{n} 个评测候选被接受" if n else "无评测候选被接受"))
+
+        # I2 没有正式正文来源追溯到评测候选
+        n = _count(conn, "SELECT count(*) FROM muse_content_block_source_attribution a "
+                         "JOIN example_candidate c ON c.id::text = a.source_object_id "
+                         "WHERE c.run_type IN ('eval','diagnostic')")
+        results.append(("I2 正式正文来源不追溯到评测候选", n == 0,
+                        f"{n} 个正式正文来源追溯到评测候选" if n else "无正式正文来源追溯到评测候选"))
+
+        # I3 评测候选的质量结果可被生产视图过滤（评测候选的质量结果都能关联到评测候选，即可过滤）
+        n = _count(conn, "SELECT count(*) FROM example_quality_result q "
+                         "JOIN example_candidate c ON c.candidate_sha256 = q.candidate_sha256 "
+                         "WHERE c.run_type IN ('eval','diagnostic') AND q.candidate_sha256 IS NULL")
+        results.append(("I3 评测质量结果可过滤", n == 0,
+                        f"{n} 个评测质量结果无法关联评测候选（无法过滤）" if n else "评测质量结果均可关联过滤"))
+
+        # I4 oracle 读侧红线：oracle raw 行都带 kind='oracle' 标记（生产路径据此排除）
+        n = _count(conn, "SELECT count(*) FROM example_raw_content "
+                         "WHERE kind='oracle' AND content_sha256 IS NULL")
+        results.append(("I4 oracle 标记完整", n == 0,
+                        f"{n} 个 oracle raw 缺哈希标记" if n else "oracle raw 标记完整"))
+
+        # I5 COMPLETED 轮次封存：lease 声明的 content_hashes 数 == 实际 raw_content 行数
+        if run_id:
+            leases = conn.execute(
+                "SELECT id, content_hashes FROM example_raw_lease WHERE run_id=%s", (run_id,)).fetchall()
+        else:
+            leases = conn.execute("SELECT id, content_hashes FROM example_raw_lease").fetchall()
+        seal_violations = 0
+        for lease_id, content_hashes in leases:
+            declared = len(content_hashes) if isinstance(content_hashes, list) else 0
+            actual = _count(conn, "SELECT count(*) FROM example_raw_content WHERE lease_id=%s", (lease_id,))
+            if declared != actual:
+                seal_violations += 1
+        results.append(("I5 轮次封存(声明==实际落库)", seal_violations == 0,
+                        f"{seal_violations} 个租约声明与实际落库不符" if seal_violations else "所有租约封存一致"))
+
+    return results
+
+
+def main():
+    ap = argparse.ArgumentParser(description="常设不变量检查（从靠预防迁到靠发现）")
+    ap.add_argument("--run-id", default=None, help="只检查某轮的轮次封存")
+    args = ap.parse_args()
+    results = check_invariants(run_id=args.run_id)
+    all_pass = True
+    for name, ok, msg in results:
+        print(f"[{'PASS' if ok else 'VIOLATION'}] {name}：{msg}")
+        if not ok:
+            all_pass = False
+    if not all_pass:
+        print("\n有不变量被违反——逻辑隔离出现越界，须排查。", file=sys.stderr)
+        sys.exit(1)
+    print("\n所有不变量通过。")
+
+
+if __name__ == "__main__":
+    main()

db/ddl/103-example评测隔离引擎强制.sql

@@ -0,0 +1,48 @@
+-- 评测隔离引擎强制（harness 改造原则 4：强制收拢到引擎单点 + run_type 钉死）。
+-- 设计拍板 2026-08-01（docs/2026-08-01-评测harness改造设计.md 原则 4/6）。
+-- 核心：评测与生产同库，靠逻辑隔离（访问分离）；强制必须压到引擎单点，不靠每条查询自觉过滤。
+--   1) run_type 插入后不可变——评测性这个锚点钉死，不能先改成 production 再接受。
+--   2) (eval, accepted) 永不可达——既有 trg_example_candidate_no_eval_accept 已守；run_type 不可变后更硬。
+--   3) 生产只读角色 muse_read + RLS——评测行对生产读默认不可见，显式 opt-in（app.eval_visible）才可见。
+-- 注意：root 是 superuser，RLS 对其无效；muse_read 是非超级用户角色，RLS 对它生效。
+--       生产读路径应改用 muse_read 连接（切换为后续代码改动），admin/看板用 root 可见全部（含评测，带标记）。
+
+-- ── 1) run_type 插入后不可变 ──
+CREATE OR REPLACE FUNCTION reject_run_type_change()
+RETURNS TRIGGER
+LANGUAGE plpgsql
+AS $$
+BEGIN
+    IF NEW.run_type IS DISTINCT FROM OLD.run_type THEN
+        RAISE EXCEPTION 'run_type 插入后不可变（评测隔离引擎强制，原 % 新 %）', OLD.run_type, NEW.run_type;
+    END IF;
+    RETURN NEW;
+END;
+$$;
+
+CREATE OR REPLACE TRIGGER trg_example_candidate_run_type_immutable
+BEFORE UPDATE ON example_candidate
+FOR EACH ROW EXECUTE FUNCTION reject_run_type_change();
+
+-- ── 2) 生产只读角色 muse_read（非超级用户，RLS 对其生效）──
+DO $$
+BEGIN
+    IF NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'muse_read') THEN
+        CREATE ROLE muse_read NOLOGIN;
+    END IF;
+END
+$$;
+
+GRANT SELECT ON example_candidate TO muse_read;
+GRANT SELECT ON example_run TO muse_read;
+GRANT SELECT ON example_run_receipt TO muse_read;
+GRANT SELECT ON example_quality_result TO muse_read;
+
+-- ── 3) RLS：评测候选对生产读默认不可见，显式 opt-in（app.eval_visible='true'）才可见 ──
+ALTER TABLE example_candidate ENABLE ROW LEVEL SECURITY;
+ALTER TABLE example_candidate FORCE ROW LEVEL SECURITY;
+
+DROP POLICY IF EXISTS eval_isolation_candidate ON example_candidate;
+CREATE POLICY eval_isolation_candidate ON example_candidate
+    FOR SELECT
+    USING (run_type IS DISTINCT FROM 'eval' OR current_setting('app.eval_visible', true) = 'true');