框架: 建立参考作品授权快照与原文件版本

新增 append-only 授权快照 DDL，并以原文件 SHA-256 固定参考作品来源版本。收紧 loader 唯一来源证明链、授权用途和时效失败关闭，保持数据库待 apply、真实记录未写。

.claude/skills/db/scripts/test_authorization_snapshot_ddl.py

@@ -0,0 +1,29 @@
+#!/usr/bin/env python3
+"""参考作品授权快照 DDL 的 append-only 静态门禁。"""
+
+import pathlib
+import re
+import unittest
+
+
+DDL_PATH = pathlib.Path(__file__).resolve().parents[4] / "db" / "ddl" / "96-example参考作品授权快照.sql"
+
+
+class AuthorizationSnapshotDdlTest(unittest.TestCase):
+    def test_authorization_snapshot_is_append_only_and_version_unique(self):
+        self.assertTrue(DDL_PATH.is_file(), "缺少授权快照 DDL")
+        ddl = DDL_PATH.read_text(encoding="utf-8")
+        self.assertIn("CREATE TABLE example_reference_authorization_snapshot", ddl)
+        self.assertRegex(ddl, r"UNIQUE\s*\(tenant_id,\s*work_id,\s*snapshot_version\)")
+        self.assertRegex(ddl, r"BEFORE\s+UPDATE\s+OR\s+DELETE")
+        self.assertIn("source_hash", ddl)
+        self.assertIn("source_version", ddl)
+        self.assertRegex(ddl, r"research_only.*public_domain.*licensed.*unauthorized")
+        self.assertRegex(ddl, r"jsonb_typeof\(allowed_purpose\)\s*=\s*'array'")
+        self.assertIn("authorization_basis <> 'user_authorization' OR copyright_status = 'research_only'", ddl)
+        self.assertIn("allowed_purpose = '[\"offline_evaluation\"]'::jsonb", ddl)
+        self.assertIsNotNone(re.search(r"RAISE\s+EXCEPTION", ddl, re.IGNORECASE))
+
+
+if __name__ == "__main__":
+    unittest.main()

.claude/skills/replay-eval/SKILL.md

@@ -8,7 +8,9 @@ disable-model-invocation: true
 
 本 skill 只负责确定性的评测编排边界，不调用模型、不替代统一读取器，也不写正式规划或知识。评测目的、三臂定义和细纲评分合同见 `docs/2026-07-19-回放评测-细纲首跑设计与计划.md`。
 
-真实参考作品配置由 `scripts/load_reference_work.py` 从实验库只读组装；它只取作品元数据、窗级大纲、章级细纲摘要和预注册卡 ID 对应的候选卡历史。候选卡必须标记为 `eval_draft`，不能当作生产知识检索结果。
+真实参考作品配置由 `scripts/load_reference_work.py` 在 `REPEATABLE READ READ ONLY` 事务中从实验库组装；它只取作品元数据、窗级大纲、章级细纲摘要和预注册卡 ID 对应的候选卡历史。候选卡必须标记为 `eval_draft`，不能当作生产知识检索结果。
+
+来源版本只认原文件证明链：`example_reference_work.source_file` 必须唯一匹配成功 import task 和未软删 knowledge document，三方文件名一致，文档 `file_hash` 为 64 位小写 SHA-256，且 import `command_id` 以该 hash 前 16 位开头。`sourceHash=sha256:<hash>`，`sourceVersion=raw-file-v1:sha256:<hash>`；作品 revision 和导入章数不得改变原文件版本。
 
 ## 输入合同
 
@@ -45,7 +47,7 @@ disable-model-invocation: true
 ## 编排入口
 
 - `scripts/run_replay.py --mode dry_run`：只执行授权、来源、冻结和三臂 manifest 预检，不调用模型；这是首个机制 smoke 入口。
-- `scripts/load_reference_work.py`：从 PostgreSQL 只读事务组装仓库外临时配置；缺授权字段仍会生成可审计配置，但送入 `run_replay` 后必须保持 `blocked_authorization`。
+- `scripts/load_reference_work.py`：从 PostgreSQL 只读事务组装仓库外临时配置；读取 `example_reference_authorization_snapshot` 当前原文件版本的最新快照，组装 authorization 外层与 snapshot。缺授权记录仍生成可审计配置，但送入 `run_replay` 后必须保持 `blocked_authorization`。
 - `scripts/run_replay.py --mode execute`：在全部前置门通过后，依次执行三臂 planner、整组 schema、逐臂盲 detector、两个独立盲 judge、rubric 校验、稳定性门和去盲汇总；`--output-dir` 必须位于仓库外的临时目录。
 - detector 输入输出均为 JSON；输入只有匿名候选 ID、候选和公共冻结到 `as_of` 的规划上下文，不含 arm 名、`cardInjection`、`cardManifest`、任何臂特有卡内容、目标章 proxy 或其他评委结果。卡注入合法性只由确定性预检负责。任一 `high` 严重度发现整组标记 `detector_blocked`，judge 调用数必须为 0；报告不合约时标记 `detector_invalid`。
 - 两个 judge 使用不同 `judgeId` 和独立无会话进程。第二个 judge 的匿名候选顺序必须与第一个完全相反；任何 rubric 不合约标记 `judge_invalid`，任一同维差值大于 `0.5` 标记 `judge_unstable`，两者都不得标记 `completed`。
@@ -53,4 +55,4 @@ disable-model-invocation: true
 - planner、detector、judge 子进程统一受 `--timeout-seconds` 限制，默认 300 秒；任一超时分别落盘 `planner_timeout`、`detector_timeout`、`judge_timeout`，不得继续进入后续阶段或标记 `completed`。
 - `scripts/write_report.py`：从 `run_result.json` 生成独立严格 schema 的安全摘要，只接受受限标识符、枚举、数字、短安全摘要和 SHA-256；不会读取候选正文，也不会把候选路径以外的原始响应写入报告。
 
-真实作品运行前必须先从权威来源取得不可变授权快照。数据库没有该字段时，使用 dry-run 证明机制并保持 `blocked_authorization`，不得用本地配置或口头许可伪造放行。
+真实作品运行前必须先从权威来源取得不可变授权快照。当前状态：DDL 已实现待 apply，真实记录未写，实跑未开始。用户授权只能登记为 `research_only` 且 `allowedPurpose=["offline_evaluation"]`，不得伪造为 `licensed`；缺记录继续保持 `blocked_authorization`。

.claude/skills/replay-eval/scripts/check_snapshot.py

@@ -8,6 +8,8 @@ from __future__ import annotations
 
 import argparse
 import json
+import re
+from datetime import datetime, timezone
 from pathlib import Path
 from typing import Any, Mapping, Sequence
 
@@ -25,10 +27,12 @@ FORBIDDEN_SOURCE_STATUSES = frozenset(
     {"revoked", "delisted", "recalled", "blocked", "owner_missing", "unauthorized"}
 )
 ALLOWED_SOURCE_STATUSES = frozenset({"active", "approved", "authorized", "licensed"})
-ALLOWED_COPYRIGHT_STATUSES = frozenset({"active", "approved", "authorized", "licensed", "owned"})
+ALLOWED_COPYRIGHT_STATUSES = frozenset({"licensed", "public_domain", "research_only"})
 FORBIDDEN_COPYRIGHT_STATUSES = frozenset(
     {"unauthorized", "unlicensed", "revoked", "expired", "blocked"}
 )
+SOURCE_HASH_PATTERN = re.compile(r"^sha256:[0-9a-f]{64}$")
+SOURCE_VERSION_PATTERN = re.compile(r"^raw-file-v1:sha256:[0-9a-f]{64}$")
 CARD_KEYS = frozenset(
     {
         "arm",
@@ -97,6 +101,20 @@ def _field(value: Mapping[str, Any], *keys: str) -> Any:
     return None
 
 
+def _parse_utc_time(value: Any, field: str) -> tuple[datetime | None, str | None]:
+    """解析带时区的 ISO-8601 时间；格式含糊时按失败关闭处理。"""
+
+    if not isinstance(value, str) or not value.strip():
+        return None, f"授权快照 {field} 不是有效时间"
+    try:
+        parsed = datetime.fromisoformat(value.strip().replace("Z", "+00:00"))
+    except ValueError:
+        return None, f"授权快照 {field} 不是有效 ISO-8601 时间"
+    if parsed.tzinfo is None:
+        return None, f"授权快照 {field} 缺少时区"
+    return parsed.astimezone(timezone.utc), None
+
+
 def check_authorization(authorization: Mapping[str, Any] | None) -> dict[str, Any]:
     """授权信息缺失、用途不符或来源进入危险状态时关闭评测。"""
 
@@ -110,8 +128,11 @@ def check_authorization(authorization: Mapping[str, Any] | None) -> dict[str, An
         "id",
         "version",
         "immutable",
+        "sourceHash",
         "sourceVersion",
         "sourceStatus",
+        "copyrightStatus",
+        "authorizationBasis",
         "allowedPurpose",
         "checkedAt",
     )
@@ -126,6 +147,22 @@ def check_authorization(authorization: Mapping[str, Any] | None) -> dict[str, An
     if not snapshot.get("expiresAt") and not snapshot.get("revalidationAt"):
         return _result(STATUS_BLOCKED_AUTHORIZATION, ["授权快照缺少过期或重验时间"])
 
+    now = datetime.now(timezone.utc)
+    checked_at, error = _parse_utc_time(snapshot.get("checkedAt"), "checkedAt")
+    if error:
+        return _result(STATUS_BLOCKED_AUTHORIZATION, [error])
+    if checked_at is not None and checked_at > now:
+        return _result(STATUS_BLOCKED_AUTHORIZATION, ["授权快照 checkedAt 晚于当前时间"])
+    for field in ("expiresAt", "revalidationAt"):
+        raw_time = snapshot.get(field)
+        if not raw_time:
+            continue
+        deadline, error = _parse_utc_time(raw_time, field)
+        if error:
+            return _result(STATUS_BLOCKED_AUTHORIZATION, [error])
+        if deadline is not None and deadline <= now:
+            return _result(STATUS_BLOCKED_AUTHORIZATION, [f"授权快照 {field} 已到期"])
+
     source_status = str(_field(authorization, "sourceStatus", "source_status") or "").lower()
     if not source_status:
         return _result(STATUS_BLOCKED_AUTHORIZATION, ["缺少 sourceStatus"])
@@ -143,10 +180,20 @@ def check_authorization(authorization: Mapping[str, Any] | None) -> dict[str, An
         return _result(STATUS_BLOCKED_AUTHORIZATION, [f"版权状态禁止评测: {copyright_status}"])
     if copyright_status not in ALLOWED_COPYRIGHT_STATUSES:
         return _result(STATUS_BLOCKED_AUTHORIZATION, [f"版权状态未登记，拒绝评测: {copyright_status}"])
+    if str(snapshot["copyrightStatus"]).lower() != copyright_status:
+        return _result(STATUS_BLOCKED_AUTHORIZATION, ["授权快照 copyrightStatus 不一致"])
+
+    source_hash = str(_field(authorization, "sourceHash", "source_hash") or "")
+    if not SOURCE_HASH_PATTERN.fullmatch(source_hash):
+        return _result(STATUS_BLOCKED_AUTHORIZATION, ["sourceHash 必须是 sha256:<64位小写十六进制>"])
+    if str(snapshot.get("sourceHash")) != source_hash:
+        return _result(STATUS_BLOCKED_AUTHORIZATION, ["授权快照 sourceHash 不一致"])
 
     source_version = str(_field(authorization, "sourceVersion", "source_version") or "")
-    if not source_version:
-        return _result(STATUS_BLOCKED_AUTHORIZATION, ["缺少 sourceVersion"])
+    if not SOURCE_VERSION_PATTERN.fullmatch(source_version):
+        return _result(STATUS_BLOCKED_AUTHORIZATION, ["sourceVersion 必须是 raw-file-v1:sha256:<64位小写十六进制>"])
+    if source_version != f"raw-file-v1:{source_hash}":
+        return _result(STATUS_BLOCKED_AUTHORIZATION, ["sourceVersion 与 sourceHash 不一致"])
     if str(snapshot.get("sourceVersion")) != source_version:
         return _result(STATUS_BLOCKED_AUTHORIZATION, ["授权快照 sourceVersion 不一致"])
     if str(snapshot["sourceStatus"]).lower() != source_status:
@@ -166,6 +213,12 @@ def check_authorization(authorization: Mapping[str, Any] | None) -> dict[str, An
         return _result(STATUS_BLOCKED_AUTHORIZATION, ["授权快照用途不包含 offline_evaluation"])
     if set(snapshot_allowed) != set(allowed):
         return _result(STATUS_BLOCKED_AUTHORIZATION, ["授权快照用途与运行用途不一致"])
+    authorization_basis = str(snapshot.get("authorizationBasis") or "")
+    if authorization_basis == "user_authorization":
+        if copyright_status != "research_only":
+            return _result(STATUS_BLOCKED_AUTHORIZATION, ["用户授权不能登记为 licensed"])
+        if list(allowed) != ["offline_evaluation"]:
+            return _result(STATUS_BLOCKED_AUTHORIZATION, ["用户授权仅允许 offline_evaluation"])
     return _result(STATUS_READY)
 
 

.claude/skills/replay-eval/scripts/load_reference_work.py

@@ -11,6 +11,7 @@ import argparse
 import copy
 import json
 import re
+from datetime import date, datetime
 from pathlib import Path
 from typing import Any, Mapping, Sequence
 
@@ -27,6 +28,7 @@ DSN = (
 TENANT_ID = 1
 REPO_ROOT = Path(__file__).resolve().parents[4]
 DEFAULT_SNAPSHOT_VERSION = "next_fine_outline_replay_v0"
+FILE_HASH_PATTERN = re.compile(r"^[0-9a-f]{64}$")
 
 
 class AdapterError(ValueError):
@@ -48,13 +50,130 @@ def _required_chapter(value: Any, field: str) -> int:
     return chapter
 
 
-def _reference_version(work: Mapping[str, Any], reference: Mapping[str, Any]) -> str:
-    """由数据库可见的修订和导入计数形成稳定来源版本。"""
+def _unique_record(rows: Sequence[Mapping[str, Any]], label: str) -> Mapping[str, Any]:
+    """来源证明链要求唯一行；缺失或重复都不能猜测选取。"""
 
-    work_id = work.get("id")
-    revision = work.get("revision") or 0
-    imported = reference.get("imported_chapter_count") or 0
-    return f"db-work-{work_id}-rev-{revision}-imported-{imported}"
+    if not isinstance(rows, Sequence) or isinstance(rows, (str, bytes)) or len(rows) != 1:
+        count = len(rows) if isinstance(rows, Sequence) and not isinstance(rows, (str, bytes)) else 0
+        raise AdapterError(f"{label} 必须唯一匹配，实际 {count} 行")
+    row = rows[0]
+    if not isinstance(row, Mapping):
+        raise AdapterError(f"{label} 不是对象")
+    if row.get("deleted") is True:
+        raise AdapterError(f"{label} 已软删")
+    return row
+
+
+def validate_source_records(
+    reference_rows: Sequence[Mapping[str, Any]],
+    import_task_rows: Sequence[Mapping[str, Any]],
+    document_rows: Sequence[Mapping[str, Any]],
+) -> dict[str, Any]:
+    """交叉核验原文件登记、成功导入任务和知识文档，生成稳定文件版本。"""
+
+    reference = _unique_record(reference_rows, "reference.source_file")
+    import_task = _unique_record(import_task_rows, "成功 import task")
+    document = _unique_record(document_rows, "未删 knowledge document")
+
+    source_file = str(reference.get("source_file") or "").strip()
+    if not source_file:
+        raise AdapterError("reference.source_file 为空")
+    if str(import_task.get("status") or "") != "succeeded":
+        raise AdapterError("import task 不是 succeeded")
+    source_snapshot = import_task.get("source_snapshot")
+    if not isinstance(source_snapshot, Mapping):
+        raise AdapterError("import task 缺少 source_snapshot")
+    if str(source_snapshot.get("file") or "") != source_file:
+        raise AdapterError("import task filename 与 reference.source_file 不一致")
+    if str(document.get("file_name") or "") != source_file:
+        raise AdapterError("knowledge document filename 与 reference.source_file 不一致")
+
+    file_hash = str(document.get("file_hash") or "")
+    if not FILE_HASH_PATTERN.fullmatch(file_hash):
+        raise AdapterError("knowledge document file_hash 必须是 64 位小写十六进制")
+    expected_command_prefix = f"import-{file_hash[:16]}"
+    command_id = str(import_task.get("command_id") or "")
+    if not command_id.startswith(expected_command_prefix):
+        raise AdapterError("import task command_id 前缀与原文件 hash 不一致")
+
+    source_hash = f"sha256:{file_hash}"
+    return {
+        "referenceWorkId": str(reference.get("id") or ""),
+        "importTaskId": str(import_task.get("id") or ""),
+        "documentId": str(document.get("id") or ""),
+        "fileName": source_file,
+        "sourceHash": source_hash,
+        "sourceVersion": f"raw-file-v1:{source_hash}",
+        "importCommandId": command_id,
+    }
+
+
+def _iso_time(value: Any) -> str | None:
+    """把数据库时间统一为带时区的 ISO 字符串，空值保持为空。"""
+
+    if value is None:
+        return None
+    if isinstance(value, (datetime, date)):
+        return value.isoformat()
+    text = str(value).strip()
+    return text or None
+
+
+def project_authorization(
+    row: Mapping[str, Any] | None,
+    source: Mapping[str, Any],
+) -> dict[str, Any]:
+    """把最新授权行投影为外层授权与不可变快照；无记录时保持阻断。"""
+
+    source_hash = str(source.get("sourceHash") or "")
+    source_version = str(source.get("sourceVersion") or "")
+    if row is None:
+        return {
+            "sourceStatus": "missing_authorization_snapshot",
+            "copyrightStatus": "unknown",
+            "sourceHash": source_hash,
+            "sourceVersion": source_version,
+            "allowedPurpose": [],
+            "authorizationSnapshot": {},
+        }
+    if not isinstance(row, Mapping):
+        raise AdapterError("授权快照行不是对象")
+    if str(row.get("source_hash") or "") != source_hash:
+        raise AdapterError("授权快照 source_hash 与原文件不一致")
+    if str(row.get("source_version") or "") != source_version:
+        raise AdapterError("授权快照 source_version 与原文件不一致")
+
+    allowed = row.get("allowed_purpose")
+    forbidden = row.get("forbidden_purpose")
+    if not isinstance(allowed, list) or not isinstance(forbidden, list):
+        raise AdapterError("授权快照用途字段必须是数组")
+    source_status = str(row.get("source_status") or "")
+    copyright_status = str(row.get("copyright_status") or "")
+    snapshot = {
+        "id": str(row.get("id") or ""),
+        "version": str(row.get("snapshot_version") or ""),
+        "immutable": True,
+        "sourceHash": source_hash,
+        "sourceVersion": source_version,
+        "sourceStatus": source_status,
+        "copyrightStatus": copyright_status,
+        "allowedPurpose": copy.deepcopy(allowed),
+        "forbiddenPurpose": copy.deepcopy(forbidden),
+        "authorizationBasis": str(row.get("authorization_basis") or ""),
+        "authorizedBy": str(row.get("authorized_by") or ""),
+        "displaySummary": str(row.get("display_summary") or ""),
+        "checkedAt": _iso_time(row.get("checked_at")),
+        "expiresAt": _iso_time(row.get("expires_at")),
+        "revalidationAt": _iso_time(row.get("revalidation_at")),
+    }
+    return {
+        "sourceStatus": source_status,
+        "copyrightStatus": copyright_status,
+        "sourceHash": source_hash,
+        "sourceVersion": source_version,
+        "allowedPurpose": copy.deepcopy(allowed),
+        "authorizationSnapshot": snapshot,
+    }
 
 
 def _row_id(row: Mapping[str, Any]) -> str:
@@ -243,6 +362,8 @@ def build_replay_config(
     target: int,
     evaluation_set_version: str,
     strategy_version: str,
+    source: Mapping[str, Any],
+    authorization: Mapping[str, Any],
     run_id: str | None = None,
     snapshot_version: str = DEFAULT_SNAPSHOT_VERSION,
     history_chapter_limit: int = 6,
@@ -259,7 +380,10 @@ def build_replay_config(
     if target_chapter != normalized_target:
         raise AdapterError("target scaffold 不是目标章，拒绝混用")
 
-    source_version = _reference_version(work, reference)
+    source_hash = str(source.get("sourceHash") or "")
+    source_version = str(source.get("sourceVersion") or "")
+    if not source_hash.startswith("sha256:") or source_version != f"raw-file-v1:{source_hash}":
+        raise AdapterError("来源 hash/version 不符合原文件版本合同")
     kept_windows, _ = filter_outline_windows(outline_rows, normalized_as_of)
     projected_windows = [_project_outline(row) for row in kept_windows]
 
@@ -302,6 +426,7 @@ def build_replay_config(
         {
             "sourceId": f"reference-work:{work.get('id')}",
             "sourceVersion": source_version,
+            "sourceHash": source_hash,
             "scope": "metadata",
             "sourceStatus": str(reference.get("parse_status") or "unknown"),
         }
@@ -353,6 +478,7 @@ def build_replay_config(
             "referenceWorkId": str(work.get("id")),
             "outlineSourceCount": len(projected_windows),
             "sourceVersion": source_version,
+            "sourceHash": source_hash,
         },
         "L3": {
             "sourceMode": "eval_draft",
@@ -399,13 +525,7 @@ def build_replay_config(
                 "cardStrategy": "placebo",
             },
         },
-        "authorization": {
-            "sourceStatus": "missing_authorization_snapshot",
-            "copyrightStatus": "unknown",
-            "sourceVersion": source_version,
-            "allowedPurpose": [],
-            "authorizationSnapshot": {},
-        },
+        "authorization": copy.deepcopy(dict(authorization)),
         "runPermissions": {
             "purpose": "offline_evaluation",
             "mode": "dry_run",
@@ -435,7 +555,7 @@ def load_reference_rows(
     correct_ids, placebo_ids = _validate_selection(card_selection)
     selected_ids = [int(item) if str(item).isdigit() else item for item in correct_ids + placebo_ids]
     with psycopg.connect(dsn, row_factory=dict_row) as conn:
-        conn.execute("SET TRANSACTION READ ONLY")
+        conn.execute("SET TRANSACTION ISOLATION LEVEL REPEATABLE READ READ ONLY")
         work = conn.execute(
             """
             SELECT id,title,revision,chapter_count,parse_status,import_status
@@ -444,18 +564,52 @@ def load_reference_rows(
             """,
             (tenant_id, work_id),
         ).fetchone()
-        reference = conn.execute(
+        reference_rows = conn.execute(
             """
             SELECT id,work_id,declared_chapter_count,imported_chapter_count,
-                   parse_scope,parse_status,source_file,notes,update_time
+                   parse_scope,parse_status,source_file,notes,update_time,deleted
             FROM example_reference_work
             WHERE tenant_id=%s AND work_id=%s AND deleted=FALSE
-            ORDER BY id DESC LIMIT 1
+            ORDER BY id
             """,
             (tenant_id, work_id),
-        ).fetchone()
-        if work is None or reference is None:
+        ).fetchall()
+        if work is None:
             raise AdapterError("作品或参考作品登记不存在")
+        reference = _unique_record(reference_rows, "reference.source_file")
+
+        import_task_rows = conn.execute(
+            """
+            SELECT id,status,command_id,source_snapshot,deleted
+            FROM muse_content_import_task
+            WHERE tenant_id=%s AND work_id=%s AND status='succeeded' AND deleted=FALSE
+            ORDER BY id
+            """,
+            (tenant_id, work_id),
+        ).fetchall()
+        document_rows = conn.execute(
+            """
+            SELECT id,file_name,file_hash,deleted
+            FROM muse_knowledge_document
+            WHERE tenant_id=%s AND file_name=%s AND deleted=FALSE
+            ORDER BY id
+            """,
+            (tenant_id, reference.get("source_file")),
+        ).fetchall()
+        source = validate_source_records(reference_rows, import_task_rows, document_rows)
+        authorization_row = conn.execute(
+            """
+            SELECT id,snapshot_version,source_hash,source_version,copyright_status,source_status,
+                   allowed_purpose,forbidden_purpose,authorization_basis,authorized_by,
+                   display_summary,checked_at,expires_at,revalidation_at
+            FROM example_reference_authorization_snapshot
+            WHERE tenant_id=%s AND work_id=%s AND source_version=%s
+            ORDER BY checked_at DESC,id DESC
+            LIMIT 1
+            """,
+            (tenant_id, work_id, source["sourceVersion"]),
+        ).fetchone()
+        authorization = project_authorization(authorization_row, source)
         if normalized_target > int(work.get("chapter_count") or 0) + 1:
             raise AdapterError("目标章超出作品导入范围")
 
@@ -508,6 +662,8 @@ def load_reference_rows(
     return {
         "work": work,
         "reference": reference,
+        "source": source,
+        "authorization": authorization,
         "outline_rows": outline_rows,
         "scaffold_rows": scaffold_rows,
         "target_scaffold": target_scaffold,
@@ -574,7 +730,7 @@ def main() -> int:
         "correctCardCount": len(config["arms"]["outline_plus_cards"]["cards"]),
         "placeboCardCount": len(config["arms"]["outline_plus_placebo_cards"]["cards"]),
         "cardSourceMode": "eval_draft",
-        "authorizationStatus": "missing_authorization_snapshot",
+        "authorizationStatus": config["authorization"]["sourceStatus"],
         "configPath": str(output_dir / "config.json"),
     }
     (output_dir / "adapter_summary.json").write_text(_safe_json(summary) + "\n", encoding="utf-8")

.claude/skills/replay-eval/scripts/test_check_snapshot.py

@@ -22,18 +22,22 @@ from check_snapshot import (  # noqa: E402
 
 AUTH = {
     "sourceStatus": "active",
-    "copyrightStatus": "licensed",
-    "sourceVersion": "work-v1",
+    "copyrightStatus": "research_only",
+    "sourceHash": "sha256:02cf1f8c1ca03c26e0b839d88fe536e83c0af20fd8972235b7aedca6a33becf4",
+    "sourceVersion": "raw-file-v1:sha256:02cf1f8c1ca03c26e0b839d88fe536e83c0af20fd8972235b7aedca6a33becf4",
     "allowedPurpose": ["offline_evaluation"],
     "authorizationSnapshot": {
         "id": "auth-1",
         "version": "v1",
         "immutable": True,
-        "sourceVersion": "work-v1",
+        "sourceHash": "sha256:02cf1f8c1ca03c26e0b839d88fe536e83c0af20fd8972235b7aedca6a33becf4",
+        "sourceVersion": "raw-file-v1:sha256:02cf1f8c1ca03c26e0b839d88fe536e83c0af20fd8972235b7aedca6a33becf4",
         "sourceStatus": "active",
+        "copyrightStatus": "research_only",
+        "authorizationBasis": "user_authorization",
         "allowedPurpose": ["offline_evaluation"],
         "checkedAt": "2026-07-19T00:00:00Z",
-        "revalidationAt": "2026-07-20T00:00:00Z",
+        "revalidationAt": "2099-07-20T00:00:00Z",
     },
 }
 
@@ -51,8 +55,57 @@ class CheckSnapshotTest(unittest.TestCase):
         self.assertEqual(check_authorization(denied)["status"], STATUS_BLOCKED_AUTHORIZATION)
         unknown_status = {**AUTH, "sourceStatus": "temporary"}
         self.assertEqual(check_authorization(unknown_status)["status"], STATUS_BLOCKED_AUTHORIZATION)
-        unlicensed = {**AUTH, "copyrightStatus": "unlicensed"}
-        self.assertEqual(check_authorization(unlicensed)["status"], STATUS_BLOCKED_AUTHORIZATION)
+        unauthorized = {**AUTH, "copyrightStatus": "unauthorized"}
+        self.assertEqual(check_authorization(unauthorized)["status"], STATUS_BLOCKED_AUTHORIZATION)
+
+    def test_research_only_and_public_domain_allow_offline_evaluation(self):
+        self.assertTrue(check_authorization(AUTH)["ok"])
+        public_domain = {
+            **AUTH,
+            "copyrightStatus": "public_domain",
+            "authorizationSnapshot": {
+                **AUTH["authorizationSnapshot"],
+                "copyrightStatus": "public_domain",
+                "authorizationBasis": "public_domain_record",
+            },
+        }
+        self.assertTrue(check_authorization(public_domain)["ok"])
+
+    def test_user_authorization_cannot_be_forged_as_licensed(self):
+        forged = {
+            **AUTH,
+            "copyrightStatus": "licensed",
+            "authorizationSnapshot": {
+                **AUTH["authorizationSnapshot"],
+                "copyrightStatus": "licensed",
+            },
+        }
+        self.assertEqual(check_authorization(forged)["status"], STATUS_BLOCKED_AUTHORIZATION)
+
+    def test_source_hash_and_version_must_match_snapshot(self):
+        bad_hash = {**AUTH, "sourceHash": "sha256:" + "0" * 64}
+        self.assertEqual(check_authorization(bad_hash)["status"], STATUS_BLOCKED_AUTHORIZATION)
+        bad_version = {**AUTH, "sourceVersion": "raw-file-v1:sha256:" + "0" * 64}
+        self.assertEqual(check_authorization(bad_version)["status"], STATUS_BLOCKED_AUTHORIZATION)
+
+    def test_expired_or_due_revalidation_is_blocked(self):
+        expired = {
+            **AUTH,
+            "authorizationSnapshot": {
+                **AUTH["authorizationSnapshot"],
+                "expiresAt": "2020-01-01T00:00:00Z",
+                "revalidationAt": None,
+            },
+        }
+        self.assertEqual(check_authorization(expired)["status"], STATUS_BLOCKED_AUTHORIZATION)
+        due = {
+            **AUTH,
+            "authorizationSnapshot": {
+                **AUTH["authorizationSnapshot"],
+                "revalidationAt": "2020-01-01T00:00:00Z",
+            },
+        }
+        self.assertEqual(check_authorization(due)["status"], STATUS_BLOCKED_AUTHORIZATION)
 
     def test_target_source_is_blocked(self):
         allowed = check_target_sources(

.claude/skills/replay-eval/scripts/test_load_reference_work.py

@@ -4,11 +4,13 @@
 from __future__ import annotations
 
 import json
+import inspect
 import pathlib
 import sys
 import unittest
 
 sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent))
+import load_reference_work as loader  # noqa: E402
 from load_reference_work import (  # noqa: E402
     AdapterError,
     build_replay_config,
@@ -16,6 +18,9 @@ from load_reference_work import (  # noqa: E402
 )
 
 
+FILE_HASH = "02cf1f8c1ca03c26e0b839d88fe536e83c0af20fd8972235b7aedca6a33becf4"
+SOURCE_HASH = f"sha256:{FILE_HASH}"
+SOURCE_VERSION = f"raw-file-v1:{SOURCE_HASH}"
 WORK = {
     "id": 8,
     "title": "深空之影",
@@ -28,6 +33,37 @@ REFERENCE = {
     "imported_chapter_count": 594,
     "parse_scope": {"from": 1, "to": 594},
     "parse_status": "parsing",
+    "source_file": "深空之影_远瞳.txt",
+    "deleted": False,
+}
+IMPORT_TASK = {
+    "id": 19,
+    "status": "succeeded",
+    "command_id": f"import-{FILE_HASH[:16]}",
+    "source_snapshot": {"file": REFERENCE["source_file"]},
+    "deleted": False,
+}
+DOCUMENT = {
+    "id": 7,
+    "file_name": REFERENCE["source_file"],
+    "file_hash": FILE_HASH,
+    "deleted": False,
+}
+AUTHORIZATION_ROW = {
+    "id": 1,
+    "snapshot_version": "auth-work-8-v1",
+    "source_hash": SOURCE_HASH,
+    "source_version": SOURCE_VERSION,
+    "copyright_status": "research_only",
+    "source_status": "active",
+    "allowed_purpose": ["offline_evaluation"],
+    "forbidden_purpose": ["external_distribution"],
+    "authorization_basis": "user_authorization",
+    "authorized_by": "user:1",
+    "display_summary": "用户授权仅用于内部离线评测",
+    "checked_at": "2026-07-19T00:00:00Z",
+    "expires_at": None,
+    "revalidation_at": "2099-07-19T00:00:00Z",
 }
 
 
@@ -59,6 +95,10 @@ def card_row(card_id=11126, name="苏铭"):
 
 
 class LoadReferenceWorkTest(unittest.TestCase):
+    def setUp(self):
+        self.source = loader.validate_source_records([REFERENCE], [IMPORT_TASK], [DOCUMENT])
+        self.authorization = loader.project_authorization(AUTHORIZATION_ROW, self.source)
+
     def test_window_freeze_uses_absolute_bounds_and_excludes_target_window(self):
         config = build_replay_config(
             work=WORK,
@@ -77,6 +117,8 @@ class LoadReferenceWorkTest(unittest.TestCase):
             target=430,
             evaluation_set_version="set-test",
             strategy_version="strategy-test",
+            source=self.source,
+            authorization=self.authorization,
         )
         windows = config["snapshot"]["data"]["outlineWindows"]
         self.assertEqual([item["from_order"] for item in windows], [401])
@@ -92,7 +134,7 @@ class LoadReferenceWorkTest(unittest.TestCase):
         self.assertNotIn("目标章事实", public)
 
     def test_card_is_eval_draft_and_history_is_frozen(self):
-        result = project_card(card_row(), as_of=488, source_version="db-work-8-v12")
+        result = project_card(card_row(), as_of=488, source_version=SOURCE_VERSION)
         encoded = json.dumps(result, ensure_ascii=False)
         self.assertEqual(result["evaluationStatus"], "eval_draft")
         self.assertFalse(result["productionRetrievalEligible"])
@@ -100,13 +142,13 @@ class LoadReferenceWorkTest(unittest.TestCase):
         self.assertNotIn("终局摘要", encoded)
         self.assertNotIn("终局能力", encoded)
         self.assertEqual(result["source"]["sourceId"], "eval-draft:11126")
-        self.assertEqual(result["source"]["sourceVersion"], "db-work-8-v12")
+        self.assertEqual(result["source"]["sourceVersion"], SOURCE_VERSION)
 
     def test_missing_history_fails_closed_instead_of_using_static_card_fields(self):
         row = card_row()
         row["draft_payload"]["字段"].pop("演变历程")
         with self.assertRaises(AdapterError):
-            project_card(row, as_of=488, source_version="db-work-8-v12")
+            project_card(row, as_of=488, source_version=SOURCE_VERSION)
 
     def test_sources_have_source_id_and_version(self):
         config = build_replay_config(
@@ -121,10 +163,101 @@ class LoadReferenceWorkTest(unittest.TestCase):
             target=489,
             evaluation_set_version="set-test",
             strategy_version="strategy-test",
+            source=self.source,
+            authorization=self.authorization,
         )
         self.assertTrue(config["sources"])
         self.assertTrue(all(item["sourceId"] and item["sourceVersion"] for item in config["sources"]))
-        self.assertEqual(config["referenceWork"]["version"], "db-work-8-rev-12-imported-594")
+        self.assertEqual(config["referenceWork"]["version"], SOURCE_VERSION)
+        self.assertEqual(config["authorization"], self.authorization)
+
+    def test_source_records_normalize_original_file_version(self):
+        self.assertEqual(self.source["documentId"], "7")
+        self.assertEqual(self.source["fileName"], REFERENCE["source_file"])
+        self.assertEqual(self.source["sourceHash"], SOURCE_HASH)
+        self.assertEqual(self.source["sourceVersion"], SOURCE_VERSION)
+
+    def test_missing_source_record_fails_closed(self):
+        for references, tasks, documents in (
+            ([], [IMPORT_TASK], [DOCUMENT]),
+            ([REFERENCE], [], [DOCUMENT]),
+            ([REFERENCE], [IMPORT_TASK], []),
+        ):
+            with self.subTest(references=len(references), tasks=len(tasks), documents=len(documents)):
+                with self.assertRaises(AdapterError):
+                    loader.validate_source_records(references, tasks, documents)
+
+    def test_duplicate_source_record_fails_closed(self):
+        for references, tasks, documents in (
+            ([REFERENCE, REFERENCE], [IMPORT_TASK], [DOCUMENT]),
+            ([REFERENCE], [IMPORT_TASK, IMPORT_TASK], [DOCUMENT]),
+            ([REFERENCE], [IMPORT_TASK], [DOCUMENT, DOCUMENT]),
+        ):
+            with self.subTest(references=len(references), tasks=len(tasks), documents=len(documents)):
+                with self.assertRaises(AdapterError):
+                    loader.validate_source_records(references, tasks, documents)
+
+    def test_invalid_hash_fails_closed(self):
+        with self.assertRaises(AdapterError):
+            loader.validate_source_records([REFERENCE], [IMPORT_TASK], [{**DOCUMENT, "file_hash": "xyz"}])
+
+    def test_filename_mismatch_fails_closed(self):
+        bad_task = {**IMPORT_TASK, "source_snapshot": {"file": "别的文件.txt"}}
+        with self.assertRaises(AdapterError):
+            loader.validate_source_records([REFERENCE], [bad_task], [DOCUMENT])
+        with self.assertRaises(AdapterError):
+            loader.validate_source_records([REFERENCE], [IMPORT_TASK], [{**DOCUMENT, "file_name": "别的文件.txt"}])
+
+    def test_command_prefix_mismatch_fails_closed(self):
+        with self.assertRaises(AdapterError):
+            loader.validate_source_records(
+                [REFERENCE],
+                [{**IMPORT_TASK, "command_id": "import-0000000000000000"}],
+                [DOCUMENT],
+            )
+
+    def test_soft_deleted_source_record_fails_closed(self):
+        for references, tasks, documents in (
+            ([{**REFERENCE, "deleted": True}], [IMPORT_TASK], [DOCUMENT]),
+            ([REFERENCE], [{**IMPORT_TASK, "deleted": True}], [DOCUMENT]),
+            ([REFERENCE], [IMPORT_TASK], [{**DOCUMENT, "deleted": True}]),
+        ):
+            with self.subTest(references=references, tasks=tasks, documents=documents):
+                with self.assertRaises(AdapterError):
+                    loader.validate_source_records(references, tasks, documents)
+
+    def test_authorization_is_projected_or_kept_blocked(self):
+        self.assertEqual(self.authorization["copyrightStatus"], "research_only")
+        self.assertEqual(self.authorization["sourceHash"], SOURCE_HASH)
+        self.assertEqual(self.authorization["authorizationSnapshot"]["sourceVersion"], SOURCE_VERSION)
+        blocked = loader.project_authorization(None, self.source)
+        self.assertEqual(blocked["sourceStatus"], "missing_authorization_snapshot")
+        self.assertEqual(blocked["allowedPurpose"], [])
+        self.assertEqual(blocked["authorizationSnapshot"], {})
+
+    def test_work_revision_change_does_not_change_source_version(self):
+        changed_work = {**WORK, "revision": 999}
+        config = build_replay_config(
+            work=changed_work,
+            reference={**REFERENCE, "imported_chapter_count": 593},
+            outline_rows=[{"id": 1, "from_order": 1, "to_order": 10, "outline_text": "历史窗"}],
+            scaffold_rows=[],
+            target_scaffold={"id": 11, "chapter": 489, "title": "目标", "outline_text": "目标事实"},
+            card_rows=[card_row(), card_row(11127, "赵宁")],
+            card_selection={"correctCardIds": [11126], "placeboCardIds": [11127]},
+            as_of=488,
+            target=489,
+            evaluation_set_version="set-test",
+            strategy_version="strategy-test",
+            source=self.source,
+            authorization=self.authorization,
+        )
+        self.assertEqual(config["referenceWork"]["version"], SOURCE_VERSION)
+
+    def test_database_transaction_is_repeatable_read_only(self):
+        source = inspect.getsource(loader.load_reference_rows)
+        self.assertIn("SET TRANSACTION ISOLATION LEVEL REPEATABLE READ READ ONLY", source)
+        self.assertIn("ORDER BY checked_at DESC,id DESC\n            LIMIT 1", source)
 
 
 if __name__ == "__main__":

.claude/skills/replay-eval/scripts/test_run_replay.py

@@ -16,20 +16,26 @@ from run_replay import _parse_args, _planner_prompt, run_replay  # noqa: E402
 from write_report import render_report  # noqa: E402
 
 
+SOURCE_HASH = "sha256:02cf1f8c1ca03c26e0b839d88fe536e83c0af20fd8972235b7aedca6a33becf4"
+SOURCE_VERSION = f"raw-file-v1:{SOURCE_HASH}"
 AUTH = {
     "sourceStatus": "active",
-    "copyrightStatus": "licensed",
-    "sourceVersion": "work-v1",
+    "copyrightStatus": "research_only",
+    "sourceHash": SOURCE_HASH,
+    "sourceVersion": SOURCE_VERSION,
     "allowedPurpose": ["offline_evaluation"],
     "authorizationSnapshot": {
         "id": "auth-1",
         "version": "v1",
         "immutable": True,
-        "sourceVersion": "work-v1",
+        "sourceHash": SOURCE_HASH,
+        "sourceVersion": SOURCE_VERSION,
         "sourceStatus": "active",
+        "copyrightStatus": "research_only",
+        "authorizationBasis": "user_authorization",
         "allowedPurpose": ["offline_evaluation"],
         "checkedAt": "2026-07-19T00:00:00Z",
-        "revalidationAt": "2026-07-20T00:00:00Z",
+        "revalidationAt": "2099-07-20T00:00:00Z",
     },
 }
 
@@ -38,7 +44,7 @@ def config():
     common = {"l0": {"targetChapter": 489}, "l1": {"asOfChapter": 488}, "l2": {"mainline": "安全公共输入"}}
     return {
         "runId": "smoke-001",
-        "referenceWork": {"id": "deep-space", "version": "work-v1"},
+        "referenceWork": {"id": "deep-space", "version": SOURCE_VERSION},
         "evaluationSetVersion": "set-v1",
         "strategyVersion": "strategy-v1",
         "authorization": AUTH,

.claude/skills/replay-eval/scripts/write_report.py

@@ -30,6 +30,7 @@ REPORT_FORBIDDEN_KEYS = frozenset(
     }
 )
 IDENTIFIER_PATTERN = re.compile(r"[A-Za-z0-9][A-Za-z0-9._:-]{0,63}\Z")
+SOURCE_VERSION_PATTERN = re.compile(r"[A-Za-z0-9][A-Za-z0-9._:-]{0,127}\Z")
 SHA256_PATTERN = re.compile(r"[0-9a-f]{64}\Z")
 ARM_NAMES = frozenset(
     {"outline_only", "outline_plus_cards", "outline_plus_placebo_cards"}
@@ -130,6 +131,15 @@ def _identifier(value: Any, path: str, default: str = "unknown") -> str:
     return candidate
 
 
+def _source_version(value: Any, path: str) -> str:
+    """来源版本允许容纳带算法前缀的完整 SHA-256，仍只接受安全标识字符。"""
+
+    candidate = "unknown" if value is None or value == "" else value
+    if not isinstance(candidate, str) or SOURCE_VERSION_PATTERN.fullmatch(candidate) is None:
+        raise ValueError(f"{path} 必须是长度不超过 128 的安全来源版本")
+    return candidate
+
+
 def _enum(value: Any, allowed: frozenset[str], path: str, default: str) -> str:
     candidate = default if value is None or value == "" else value
     if not isinstance(candidate, str) or candidate not in allowed:
@@ -257,7 +267,10 @@ def _project_report(result: Mapping[str, Any]) -> dict[str, Any]:
         "status": _enum(result.get("status"), RUN_STATUSES, "status", "unknown"),
         "mode": _enum(result.get("mode"), frozenset({"unknown", "dry_run", "execute"}), "mode", "unknown"),
         "referenceWork": _identifier(result.get("referenceWork"), "referenceWork"),
-        "referenceWorkVersion": _identifier(result.get("referenceWorkVersion"), "referenceWorkVersion"),
+        "referenceWorkVersion": _source_version(
+            result.get("referenceWorkVersion"),
+            "referenceWorkVersion",
+        ),
         "asOfChapter": _integer(result.get("asOfChapter"), "asOfChapter", minimum=1),
         "targetChapter": _integer(result.get("targetChapter"), "targetChapter", minimum=1),
         "snapshotVersion": _identifier(result.get("snapshotVersion"), "snapshotVersion"),

db/ddl/96-example参考作品授权快照.sql

@@ -0,0 +1,83 @@
+-- example_reference_authorization_snapshot：参考作品用途授权的不可变快照。
+-- 本表只允许 INSERT；授权变化通过新增版本表达，禁止覆盖或删除历史证据。
+CREATE TABLE example_reference_authorization_snapshot (
+    id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
+    reference_work_id BIGINT NOT NULL,               -- → example_reference_work.id（软引用）
+    work_id BIGINT NOT NULL,                         -- → muse_content_work.id（软引用）
+    knowledge_document_id BIGINT NOT NULL,           -- → muse_knowledge_document.id（软引用）
+    import_task_id BIGINT NOT NULL,                  -- → muse_content_import_task.id（软引用）
+    source_file VARCHAR(500) NOT NULL,                -- 授权对应的原文件名
+    source_hash VARCHAR(71) NOT NULL,                 -- sha256:<64位小写十六进制>
+    source_version VARCHAR(96) NOT NULL,              -- raw-file-v1:sha256:<64位小写十六进制>
+    snapshot_version VARCHAR(160) NOT NULL,           -- 授权快照业务版本，不使用物理主键充当合同版本
+    copyright_status VARCHAR(30) NOT NULL,            -- licensed/public_domain/research_only/unauthorized
+    source_status VARCHAR(30) NOT NULL DEFAULT 'active',
+    allowed_purpose JSONB NOT NULL DEFAULT '[]',      -- 允许用途数组，例如 ["offline_evaluation"]
+    forbidden_purpose JSONB NOT NULL DEFAULT '[]',    -- 明确禁止用途数组，例如 ["external_distribution"]
+    authorization_basis VARCHAR(40) NOT NULL,         -- user_authorization/public_domain_record/license_contract
+    authorized_by VARCHAR(128) NOT NULL,              -- 授权主体或登记责任人
+    authorization_evidence JSONB NOT NULL,            -- 授权原文摘要、时间和证据定位，不存敏感全文
+    display_summary VARCHAR(500) NOT NULL,            -- 面向审计面的脱敏摘要
+    checked_at TIMESTAMPTZ NOT NULL,                  -- 本次授权核验时间
+    expires_at TIMESTAMPTZ,                           -- 到期即阻断
+    revalidation_at TIMESTAMPTZ,                      -- 到点必须重验
+    creator VARCHAR(64) NOT NULL DEFAULT '',
+    create_time TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    tenant_id BIGINT NOT NULL DEFAULT 0,
+    CONSTRAINT uk_example_reference_auth_work_version UNIQUE (tenant_id, work_id, snapshot_version),
+    CONSTRAINT chk_example_reference_auth_source_hash
+        CHECK (source_hash ~ '^sha256:[0-9a-f]{64}$'),
+    CONSTRAINT chk_example_reference_auth_source_version
+        CHECK (source_version = 'raw-file-v1:' || source_hash),
+    CONSTRAINT chk_example_reference_auth_copyright
+        CHECK (copyright_status IN ('research_only','public_domain','licensed','unauthorized')),
+    CONSTRAINT chk_example_reference_auth_source_status
+        CHECK (source_status IN ('active','stale','revoked','delisted','recalled','blocked','owner_missing','unauthorized')),
+    CONSTRAINT chk_example_reference_auth_allowed_purpose
+        CHECK (jsonb_typeof(allowed_purpose) = 'array'),
+    CONSTRAINT chk_example_reference_auth_forbidden_purpose
+        CHECK (jsonb_typeof(forbidden_purpose) = 'array'),
+    CONSTRAINT chk_example_reference_auth_basis
+        CHECK (authorization_basis IN ('user_authorization','public_domain_record','license_contract')),
+    CONSTRAINT chk_example_reference_auth_basis_copyright
+        CHECK (
+            (authorization_basis <> 'user_authorization' OR copyright_status = 'research_only')
+            AND (authorization_basis <> 'public_domain_record' OR copyright_status = 'public_domain')
+            AND (authorization_basis <> 'license_contract' OR copyright_status = 'licensed')
+        ),
+    CONSTRAINT chk_example_reference_auth_user_purpose
+        CHECK (
+            authorization_basis <> 'user_authorization'
+            OR allowed_purpose = '["offline_evaluation"]'::jsonb
+        ),
+    CONSTRAINT chk_example_reference_auth_evidence
+        CHECK (jsonb_typeof(authorization_evidence) = 'object'),
+    CONSTRAINT chk_example_reference_auth_purpose
+        CHECK (
+            (copyright_status = 'unauthorized' AND jsonb_array_length(allowed_purpose) = 0)
+            OR (copyright_status <> 'unauthorized' AND jsonb_array_length(allowed_purpose) > 0)
+        ),
+    CONSTRAINT chk_example_reference_auth_recheck
+        CHECK (expires_at IS NOT NULL OR revalidation_at IS NOT NULL),
+    CONSTRAINT chk_example_reference_auth_expiry_order
+        CHECK (expires_at IS NULL OR expires_at > checked_at),
+    CONSTRAINT chk_example_reference_auth_revalidation_order
+        CHECK (revalidation_at IS NULL OR revalidation_at > checked_at)
+);
+
+CREATE INDEX idx_example_reference_auth_latest
+    ON example_reference_authorization_snapshot(tenant_id, work_id, checked_at DESC, id DESC);
+
+-- 授权快照是审计证据；撤销、到期或用途变化都必须插入新版本，不能改写旧记录。
+CREATE FUNCTION reject_example_reference_authorization_snapshot_mutation()
+RETURNS TRIGGER
+LANGUAGE plpgsql
+AS $$
+BEGIN
+    RAISE EXCEPTION 'example_reference_authorization_snapshot 是 append-only 表，禁止 UPDATE/DELETE';
+END;
+$$;
+
+CREATE TRIGGER trg_example_reference_auth_append_only
+BEFORE UPDATE OR DELETE ON example_reference_authorization_snapshot
+FOR EACH ROW EXECUTE FUNCTION reject_example_reference_authorization_snapshot_mutation();

db/表映射.md

@@ -2,7 +2,8 @@
 
 > 口径（创始人拍板③ 2026-07-10）：主仓表**原样不改列**；实验私货全进 `example_*` 前缀。
 > 建表方式：`db/ddl/` 下文件经 db skill `apply`，主仓部分为 `muse-cloud/sql/muse/` 原文拷贝或逐字摘录。
-> 应用顺序：V1 → V3 → V5 → 90-ALTER摘录 → V26 → 91-example。已于 2026-07-13 应用，共 **24 张表**。
+> 已应用顺序：V1 → V3 → V5 → 90-ALTER摘录 → V26 → 91-example。已于 2026-07-13 应用，共 **24 张表**。
+> 待应用：`96-example参考作品授权快照.sql` 已实现但尚未通过 db skill `apply`，库内尚无该表和真实授权记录。
 
 ## 主仓一致表（20 张）
 
@@ -57,3 +58,9 @@
 - 软删照主仓：`deleted=TRUE`，不物理删。**例外**：meta 种子行（`muse_meta_field`）幂等重跑=删旧插新（种子演练场景，豁免软删）。
 - meta 字段行 sort_order 段位约定：1–9 基础字段，11 起特有字段。
 - 双轨对应：B2 产出全落 `muse_knowledge_draft(status='pending')`；B5 管理员确认（仅创始人触发）= draft 翻 `confirmed` + 落 `muse_knowledge_entity(status='active')`；丢弃= draft 翻 `ignored`。
+
+## 待应用实验表
+
+| DDL | 表 | 用途 | 当前状态 |
+|---|---|---|---|
+| 96 | example_reference_authorization_snapshot | 参考作品原文件版本对应的不可变用途授权快照；同作品快照版本唯一，禁止 UPDATE/DELETE | DDL 已实现待 apply；真实记录未写 |

docs/2026-07-19-回放评测-细纲首跑设计与计划.md

@@ -15,8 +15,10 @@
 - 已新增并验证：`audit_leakage.py` 对公共快照和三臂卡注入区执行内容级事实审计；`load_reference_work.py` 通过 PostgreSQL 只读事务组装仓库外临时配置，候选卡标记为 `eval_draft`，不进入生产检索。
 - 已执行真实适配 smoke：work=8、冻结 488、目标 489，组装 31 个完整历史大纲窗、6 个近章细纲摘要、正确/错配卡各 2 张；送入回放仍为 `blocked_authorization`（`example_reference_work` 没有授权快照），未生成 `snapshot_manifest`，未调用模型。
 - 已执行真实数据库前置 smoke：深空之影 `work_id=8` 的 `example_reference_work` 表没有不可变授权快照/版权用途字段，结果为 `blocked_authorization`，三臂均未调用模型。
+- 任务 A 已实现原文件版本与 append-only 授权快照 DDL：work=8 的权威原文件 `document_id=7`、SHA-256=`02cf1f8c1ca03c26e0b839d88fe536e83c0af20fd8972235b7aedca6a33becf4`，来源版本固定为 `raw-file-v1:sha256:<hash>`，不再使用作品 revision+导入章数。
+- 当前状态：**DDL 已实现待 apply，真实记录未写，实跑未开始**。用户授权只能登记为 `research_only + offline_evaluation`，不得伪造 `licensed`；apply 与 INSERT 均不在本任务执行范围。
 - `.claude/skills/llm/scripts/test_quota.py` 已统一到当前 `$24/6000` 契约：预算场景使用 `$24.5`，调用上限场景使用 `6000`，并增加策略常量断言。
-- 当前允许的结论是“冻结、变量控制、内容级泄露审计、候选结构门、detector/judge 双评编排和安全摘要机制已通”；不能说细纲智能体或知识卡已通过。真实授权接入和 430/489/550 实样仍待完成。
+- 当前允许的结论是“冻结、变量控制、内容级泄露审计、候选结构门、detector/judge 双评编排、安全摘要机制和授权快照代码合同已通”；不能说数据库表已应用、真实授权已写入、细纲智能体或知识卡已通过。真实授权接入和 430/489/550 实样仍待完成。
 
 ---