From 3b427854abc9d3d6d00b08403eaff405defbeace Mon Sep 17 00:00:00 2001 From: zizi Date: Sun, 19 Jul 2026 15:48:19 +0800 Subject: [PATCH] =?UTF-8?q?=E6=A1=86=E6=9E=B6:=20=E6=8E=A5=E5=85=A5?= =?UTF-8?q?=E5=9B=9E=E6=94=BE=E5=86=85=E5=AE=B9=E5=AE=A1=E8=AE=A1=E4=B8=8E?= =?UTF-8?q?=E5=8F=82=E8=80=83=E4=BD=9C=E5=93=81=E5=8F=AA=E8=AF=BB=E9=80=82?= =?UTF-8?q?=E9=85=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .claude/skills/replay-eval/SKILL.md | 6 + .../replay-eval/scripts/audit_leakage.py | 213 +++++++ .../scripts/load_reference_work.py | 590 ++++++++++++++++++ .../skills/replay-eval/scripts/run_replay.py | 46 +- .../replay-eval/scripts/test_audit_leakage.py | 75 +++ .../scripts/test_load_reference_work.py | 131 ++++ .../replay-eval/scripts/test_run_replay.py | 55 +- .../2026-07-19-回放评测-细纲首跑设计与计划.md | 14 +- 8 files changed, 1125 insertions(+), 5 deletions(-) create mode 100644 .claude/skills/replay-eval/scripts/audit_leakage.py create mode 100644 .claude/skills/replay-eval/scripts/load_reference_work.py create mode 100644 .claude/skills/replay-eval/scripts/test_audit_leakage.py create mode 100644 .claude/skills/replay-eval/scripts/test_load_reference_work.py diff --git a/.claude/skills/replay-eval/SKILL.md b/.claude/skills/replay-eval/SKILL.md index e1a4402..707e634 100644 --- a/.claude/skills/replay-eval/SKILL.md +++ b/.claude/skills/replay-eval/SKILL.md @@ -8,6 +8,8 @@ disable-model-invocation: true 本 skill 只负责确定性的评测编排边界，不调用模型、不替代统一读取器，也不写正式规划或知识。评测目的、三臂定义和细纲评分合同见 `docs/2026-07-19-回放评测-细纲首跑设计与计划.md`。 +真实参考作品配置由 `scripts/load_reference_work.py` 从实验库只读组装；它只取作品元数据、窗级大纲、章级细纲摘要和预注册卡 ID 对应的候选卡历史。候选卡必须标记为 `eval_draft`，不能当作生产知识检索结果。 + ## 输入合同 - `reference_work`：参考作品标识和版本。 @@ -16,6 +18,7 @@ disable-model-invocation: true - 作品大纲窗口：使用 `from_order` / `to_order`，只允许完整窗口 `to_order <= as_of`。 - 实体卡和里程碑：必须带可验证的绝对章号；无明确上界的区间不进入快照。 - 来源合同：每个来源要有 `sourceId`、`sourceVersion`、用途授权快照和来源状态。 +- 内容审计合同：`leakageAudit.targetFacts.forbiddenFacts` 必须登记目标章结构化事实及首见章号；审计结果只允许输出路径和哈希。 ## 冻结规则 @@ -31,6 +34,8 @@ disable-model-invocation: true 授权、来源版本、目标章禁读或内容级泄露检查任一失败，都必须 fail-closed，返回明确的阻断状态，不靠重试绕过。 +`run_replay.py` 会同时审计公共快照和三臂卡注入区。没有 `leakageAudit` 配置、发现未来章记录或目标事实内容匹配时，不生成 `snapshot_manifest`，也不进入 planner。 + ## 产物边界 - 原始候选、标准事实摘要和完整输入只能留在临时运行目录或 `/tmp`。 @@ -40,6 +45,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/run_replay.py --mode execute`：在全部前置门通过后，使用无工具、无会话持久化的本地 planner CLI 逐臂生成候选；`--output-dir` 必须位于仓库外的临时目录。 - `scripts/write_report.py`：从 `run_result.json` 生成安全摘要；它不会读取候选正文，也不会把候选路径以外的原始响应写入报告。 diff --git a/.claude/skills/replay-eval/scripts/audit_leakage.py b/.claude/skills/replay-eval/scripts/audit_leakage.py new file mode 100644 index 0000000..f8bfc11 --- /dev/null +++ b/.claude/skills/replay-eval/scripts/audit_leakage.py @@ -0,0 +1,213 @@ +#!/usr/bin/env python3 +"""回放快照的内容级泄露审计。 + +审计只接收结构化目标事实和已经冻结的快照，不读取正文、不调用模型、不写数据库。 +审计结果只包含路径、哈希和原因，避免把目标章事实回显到最终报告。 +""" + +from __future__ import annotations + +import copy +import json +from typing import Any, Iterator, Mapping + +from build_snapshot import normalize_chapter, normalize_chapter_range, sha256_value + + +STATUS_READY = "ready" +STATUS_INVALID_SNAPSHOT = "invalid_snapshot" +STATUS_INVALID_INPUT = "invalid_audit_input" + +_CHAPTER_KEYS = frozenset( + {"chapter", "chapter_no", "order_no", "章", "章号", "from_order", "to_order"} +) + + +def _chapter_bounds(record: Mapping[str, Any]) -> tuple[int, int] | None: + """按快照冻结规则解析记录的绝对章号或完整章区间。""" + + for key in ("chapter", "chapter_no", "order_no", "章", "章号"): + if key in record: + return normalize_chapter_range(record[key]) + if "from_order" in record or "to_order" in record: + start = record.get("from_order") + end = record.get("to_order") + if start is None or end is None: + return None + return normalize_chapter_range(f"{start}-{end}") + return None + + +def _walk(value: Any, path: str = "$") -> Iterator[tuple[str, Any]]: + """以稳定顺序遍历快照，供审计定位泄露字段。""" + + yield path, value + if isinstance(value, Mapping): + for key in sorted(value, key=lambda item: str(item)): + yield from _walk(value[key], f"{path}.{key}") + elif isinstance(value, list): + for index, item in enumerate(value): + yield from _walk(item, f"{path}[{index}]") + + +def _validate_target_facts( + target_facts: Mapping[str, Any], as_of: int, target: int +) -> tuple[list[dict[str, Any]], list[str]]: + """校验目标事实登记，不把原始事实文本放进错误信息。""" + + errors: list[str] = [] + registered_target = normalize_chapter(target_facts.get("targetChapter")) + if registered_target != target: + errors.append("targetFacts.targetChapter 必须与 targetChapter 一致") + + raw_facts = target_facts.get("forbiddenFacts") + if not isinstance(raw_facts, list): + errors.append("targetFacts.forbiddenFacts 必须是数组") + return [], errors + + facts: list[dict[str, Any]] = [] + seen_ids: set[str] = set() + for index, item in enumerate(raw_facts): + if not isinstance(item, Mapping): + errors.append(f"targetFacts.forbiddenFacts[{index}] 必须是对象") + continue + fact_id = str(item.get("id") or "") + text = item.get("text") + first_chapter = normalize_chapter(item.get("firstChapter")) + if not fact_id or fact_id in seen_ids: + errors.append(f"targetFacts.forbiddenFacts[{index}].id 缺失或重复") + if not isinstance(text, str) or not text.strip(): + errors.append(f"targetFacts.forbiddenFacts[{index}].text 必须是非空字符串") + if first_chapter is None or first_chapter <= as_of: + errors.append(f"targetFacts.forbiddenFacts[{index}].firstChapter 必须晚于 as_of") + if fact_id and isinstance(text, str) and text.strip() and first_chapter is not None: + seen_ids.add(fact_id) + facts.append( + { + "id": fact_id, + "text": text, + "firstChapter": first_chapter, + "factHash": sha256_value(text), + } + ) + return facts, errors + + +def _finding( + *, + reason: str, + path: str, + value: Any, + fact: Mapping[str, Any] | None = None, +) -> dict[str, Any]: + """构造不回显事实文本的审计发现。""" + + result: dict[str, Any] = { + "reason": reason, + "path": path, + "valueHash": sha256_value(value), + } + if fact is not None: + result["factId"] = fact["id"] + result["factHash"] = fact["factHash"] + return result + + +def audit_snapshot( + snapshot: Mapping[str, Any], + target_facts: Mapping[str, Any], + *, + as_of: int, + target: int, +) -> dict[str, Any]: + """检查快照是否含有未来记录或目标章内容级事实。""" + + normalized_as_of = normalize_chapter(as_of) + normalized_target = normalize_chapter(target) + if normalized_as_of is None or normalized_target != normalized_as_of + 1: + return { + "status": STATUS_INVALID_INPUT, + "ok": False, + "errors": ["as_of/target 必须是连续章号"], + "warnings": [], + "findingCount": 0, + "findings": [], + } + if not isinstance(snapshot, Mapping) or not isinstance(target_facts, Mapping): + return { + "status": STATUS_INVALID_INPUT, + "ok": False, + "errors": ["snapshot/targetFacts 必须是对象"], + "warnings": [], + "findingCount": 0, + "findings": [], + } + + facts, errors = _validate_target_facts(target_facts, normalized_as_of, normalized_target) + if errors: + return { + "status": STATUS_INVALID_INPUT, + "ok": False, + "errors": errors, + "warnings": [], + "findingCount": 0, + "findings": [], + } + + findings: list[dict[str, Any]] = [] + for path, value in _walk(copy.deepcopy(dict(snapshot))): + if isinstance(value, Mapping) and _CHAPTER_KEYS.intersection(value): + bounds = _chapter_bounds(value) + if bounds is None: + findings.append( + _finding(reason="missing_or_invalid_chapter", path=path, value=value) + ) + elif bounds[1] > normalized_as_of: + findings.append( + _finding(reason="future_or_crosses_as_of", path=path, value=value) + ) + if isinstance(value, str): + for fact in facts: + if fact["text"] in value: + findings.append( + _finding(reason="target_fact_match", path=path, value=value, fact=fact) + ) + + return { + "status": STATUS_INVALID_SNAPSHOT if findings else STATUS_READY, + "ok": not findings, + "errors": [], + "warnings": [], + "findingCount": len(findings), + "findings": findings, + } + + +def _parse_args() -> Any: + """命令行参数解析保留给后续独立审计调用。""" + + import argparse + + parser = argparse.ArgumentParser(description="审计冻结快照的内容级目标事实泄露") + parser.add_argument("--snapshot", required=True) + parser.add_argument("--target-facts", required=True) + parser.add_argument("--as-of", type=int, required=True) + parser.add_argument("--target", type=int, required=True) + return parser.parse_args() + + +def main() -> int: + """执行一次文件级审计，只输出安全审计结果。""" + + args = _parse_args() + with open(args.snapshot, encoding="utf-8") as handle: + snapshot = json.load(handle) + with open(args.target_facts, encoding="utf-8") as handle: + target_facts = json.load(handle) + result = audit_snapshot(snapshot, target_facts, as_of=args.as_of, target=args.target) + print(json.dumps(result, ensure_ascii=False, sort_keys=True)) + return 0 if result["ok"] else 2 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/.claude/skills/replay-eval/scripts/load_reference_work.py b/.claude/skills/replay-eval/scripts/load_reference_work.py new file mode 100644 index 0000000..29527f5 --- /dev/null +++ b/.claude/skills/replay-eval/scripts/load_reference_work.py @@ -0,0 +1,590 @@ +#!/usr/bin/env python3 +"""从实验库只读组装回放评测配置。 + +本适配器只做 SELECT 和临时文件输出，不写数据库、不读取正文 block 的内容。 +参考作品的目标章只进入审计侧 proxy，历史规划上下文和三臂卡注入严格分开。 +""" + +from __future__ import annotations + +import argparse +import copy +import json +import re +from pathlib import Path +from typing import Any, Mapping, Sequence + +import psycopg +from psycopg.rows import dict_row + +from build_snapshot import filter_milestones, filter_outline_windows, normalize_chapter + + +DSN = ( + "postgresql://root:f6710e2d0294eb1c10e26a805a64bc54@100.64.0.8:5433/muse-example" + "?keepalives=1&keepalives_idle=15&keepalives_interval=5&keepalives_count=3" +) +TENANT_ID = 1 +REPO_ROOT = Path(__file__).resolve().parents[4] +DEFAULT_SNAPSHOT_VERSION = "next_fine_outline_replay_v0" + + +class AdapterError(ValueError): + """只读适配输入缺失、越界或不能证明安全时抛出。""" + + +def _safe_json(value: Any) -> str: + """用固定格式序列化配置，保证运行版本可复现。""" + + return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + + +def _required_chapter(value: Any, field: str) -> int: + """把章号限制为明确正整数，拒绝猜测性转换。""" + + chapter = normalize_chapter(value) + if chapter is None: + raise AdapterError(f"{field} 必须是正整数章号") + return chapter + + +def _reference_version(work: Mapping[str, Any], reference: Mapping[str, Any]) -> str: + """由数据库可见的修订和导入计数形成稳定来源版本。""" + + 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}" + + +def _row_id(row: Mapping[str, Any]) -> str: + """读取卡行主键并统一为来源 ID 字符串。""" + + value = row.get("id") + if value is None or str(value).strip() == "": + raise AdapterError("卡行缺少 id") + return str(value) + + +def _normalize_id_list(value: Any, field: str) -> list[str]: + """校验预注册卡 ID 列表，不根据目标章临时猜卡。""" + + if not isinstance(value, list) or not value: + raise AdapterError(f"{field} 必须是非空数组") + if any( + isinstance(item, bool) + or not isinstance(item, (int, str)) + or not str(item).strip().isdigit() + or int(item) <= 0 + for item in value + ): + raise AdapterError(f"{field} 只能包含正整数卡 ID") + result = [str(item) for item in value if str(item).strip()] + if len(result) != len(value) or len(result) != len(set(result)): + raise AdapterError(f"{field} 含空 ID 或重复 ID") + return result + + +def _card_history(payload: Mapping[str, Any]) -> list[Mapping[str, Any]]: + """只从卡的历史字段取里程碑，不使用终态摘要字段。""" + + fields = payload.get("字段") + if not isinstance(fields, Mapping): + raise AdapterError("卡缺少字段对象，拒绝使用静态卡内容") + for key in ("演变历程", "演变轨迹", "milestones"): + value = fields.get(key) + if isinstance(value, list): + return value + raise AdapterError("卡缺少可按绝对章号冻结的历史字段") + + +def project_card(row: Mapping[str, Any], *, as_of: int, source_version: str) -> dict[str, Any]: + """将候选卡投影为截至 as_of 的 eval-only 索引视图。""" + + normalized_as_of = _required_chapter(as_of, "as_of") + payload = row.get("draft_payload") + if not isinstance(payload, Mapping): + raise AdapterError(f"卡 {_row_id(row)} 的 draft_payload 不是对象") + card_type = str(payload.get("type") or "") + name = str(payload.get("名称") or "") + if not card_type or not name: + raise AdapterError(f"卡 {_row_id(row)} 缺少 type/名称") + + history, omitted = filter_milestones(_card_history(payload), normalized_as_of) + if not history: + raise AdapterError(f"卡 {_row_id(row)} 没有可证明落在 as_of 以前的历史") + + appearances: list[int] = [] + raw_appearances = payload.get("出场章") + if isinstance(raw_appearances, list): + for value in raw_appearances: + chapter = normalize_chapter(value) + if chapter is not None and chapter <= normalized_as_of: + appearances.append(chapter) + appearances = sorted(set(appearances)) + + card_id = _row_id(row) + latest = copy.deepcopy(history[-1]) + return { + "cardId": card_id, + "type": card_type, + "name": name, + "milestones": copy.deepcopy(history), + "appearanceChapters": appearances, + "derivedState": { + "asOfChapter": normalized_as_of, + "latestMilestone": latest, + }, + "source": { + "sourceId": f"eval-draft:{card_id}", + "sourceVersion": source_version, + "scope": "card_projection", + "chapterRange": f"1-{normalized_as_of}", + }, + "evaluationStatus": "eval_draft", + "upstreamStatus": str(row.get("status") or "unknown"), + "productionRetrievalEligible": False, + "omittedHistoryCount": len(omitted), + } + + +def _card_source_version(row: Mapping[str, Any], base_version: str) -> str: + """把卡行 revision 纳入来源版本，防止卡内容变更复用旧版本。""" + + return f"{base_version}:card-{_row_id(row)}-rev-{row.get('revision') or 0}" + + +def _project_outline(row: Mapping[str, Any]) -> dict[str, Any]: + """保留窗的结构化摘要和绝对边界，不使用 window_no 作为冻结键。""" + + start = _required_chapter(row.get("from_order"), "outline.from_order") + end = _required_chapter(row.get("to_order"), "outline.to_order") + if end < start: + raise AdapterError("大纲窗 from_order 大于 to_order") + return { + "from_order": start, + "to_order": end, + "windowNo": row.get("window_no"), + "outline": str(row.get("outline_text") or ""), + "checkStatus": str(row.get("check_status") or "unknown"), + "sourceId": f"outline-window:{row.get('id')}", + } + + +def _project_scaffold(row: Mapping[str, Any], source_version: str) -> dict[str, Any]: + """只取历史章细纲摘要，不查询或复制正文 block。""" + + chapter = _required_chapter(row.get("chapter"), "scaffold.chapter") + result: dict[str, Any] = { + "chapter": chapter, + "title": str(row.get("title") or ""), + "outline": str(row.get("outline_text") or ""), + "sourceId": f"scaffold:{row.get('id')}", + "sourceVersion": source_version, + } + if isinstance(row.get("pattern_hints"), list): + result["patternHints"] = copy.deepcopy(row["pattern_hints"]) + return result + + +def _target_facts_from_scaffold(target_scaffold: Mapping[str, Any], target: int) -> dict[str, Any]: + """从目标章 reference scaffold 生成审计侧 proxy，不送入 planner。""" + + target_id = target_scaffold.get("id") + text = str(target_scaffold.get("outline_text") or "").strip() + if not text: + raise AdapterError("目标章 scaffold 缺少结构化事实，不能执行内容级审计") + + fragments = [part.strip() for part in re.split(r"[。；;！？!?\n]+", text) if part.strip()] + facts: list[dict[str, Any]] = [] + seen: set[str] = set() + for index, fragment in enumerate([text, *fragments]): + if len(fragment) < 4 or fragment in seen: + continue + seen.add(fragment) + facts.append( + { + "id": f"target-scaffold:{target_id}:{index}", + "firstChapter": target, + "text": fragment, + } + ) + if not facts: + raise AdapterError("目标章 scaffold 没有可用于审计的结构化事实") + return { + "targetChapter": target, + "source": "reference_scaffold_proxy", + "forbiddenFacts": facts, + } + + +def _validate_selection(selection: Mapping[str, Any]) -> tuple[list[str], list[str]]: + """校验正确卡和 placebo 卡的预注册集合。""" + + if not isinstance(selection, Mapping): + raise AdapterError("card selection 必须是对象") + correct = _normalize_id_list(selection.get("correctCardIds"), "correctCardIds") + placebo = _normalize_id_list(selection.get("placeboCardIds"), "placeboCardIds") + if set(correct) & set(placebo): + raise AdapterError("correctCardIds 与 placeboCardIds 不能重叠") + return correct, placebo + + +def build_replay_config( + *, + work: Mapping[str, Any], + reference: Mapping[str, Any], + outline_rows: Sequence[Mapping[str, Any]], + scaffold_rows: Sequence[Mapping[str, Any]], + target_scaffold: Mapping[str, Any], + card_rows: Sequence[Mapping[str, Any]], + card_selection: Mapping[str, Any], + as_of: int, + target: int, + evaluation_set_version: str, + strategy_version: str, + run_id: str | None = None, + snapshot_version: str = DEFAULT_SNAPSHOT_VERSION, + history_chapter_limit: int = 6, +) -> dict[str, Any]: + """把只读查询结果组装为 run_replay 可消费的临时配置。""" + + normalized_as_of = _required_chapter(as_of, "as_of") + normalized_target = _required_chapter(target, "target") + if normalized_target != normalized_as_of + 1: + raise AdapterError("target 必须等于 as_of+1") + if not str(evaluation_set_version).strip() or not str(strategy_version).strip(): + raise AdapterError("evaluation_set_version/strategy_version 不能为空") + target_chapter = _required_chapter(target_scaffold.get("chapter"), "target_scaffold.chapter") + if target_chapter != normalized_target: + raise AdapterError("target scaffold 不是目标章，拒绝混用") + + source_version = _reference_version(work, reference) + kept_windows, _ = filter_outline_windows(outline_rows, normalized_as_of) + projected_windows = [_project_outline(row) for row in kept_windows] + + historical = [] + for row in scaffold_rows: + chapter = _required_chapter(row.get("chapter"), "scaffold.chapter") + if chapter <= normalized_as_of: + historical.append(row) + historical.sort(key=lambda row: _required_chapter(row.get("chapter"), "scaffold.chapter")) + if history_chapter_limit <= 0: + raise AdapterError("history_chapter_limit 必须为正数") + projected_scaffolds = [ + _project_scaffold(row, source_version) + for row in historical[-history_chapter_limit:] + ] + + correct_ids, placebo_ids = _validate_selection(card_selection) + if any(str(row.get("source_type") or "") != "upgrade_book" for row in card_rows): + raise AdapterError("卡选择包含非 upgrade_book 来源") + rows_by_id = {_row_id(row): row for row in card_rows} + if len(rows_by_id) != len(card_rows): + raise AdapterError("卡查询结果含重复 id") + selected_ids = set(correct_ids + placebo_ids) + if set(rows_by_id) != selected_ids: + missing = sorted(selected_ids - set(rows_by_id)) + unexpected = sorted(set(rows_by_id) - selected_ids) + raise AdapterError(f"卡查询结果与预注册不一致: missing={missing}, unexpected={unexpected}") + projected_cards = { + card_id: project_card( + rows_by_id[card_id], + as_of=normalized_as_of, + source_version=_card_source_version(rows_by_id[card_id], source_version), + ) + for card_id in sorted(selected_ids) + } + correct_cards = [projected_cards[card_id] for card_id in correct_ids] + placebo_cards = [projected_cards[card_id] for card_id in placebo_ids] + + source_catalog: list[dict[str, Any]] = [ + { + "sourceId": f"reference-work:{work.get('id')}", + "sourceVersion": source_version, + "scope": "metadata", + "sourceStatus": str(reference.get("parse_status") or "unknown"), + } + ] + for row in kept_windows: + source_catalog.append( + { + "sourceId": f"outline-window:{row.get('id')}", + "sourceVersion": source_version, + "from_order": _required_chapter(row.get("from_order"), "outline.from_order"), + "to_order": _required_chapter(row.get("to_order"), "outline.to_order"), + "scope": "outline_window", + } + ) + for row in historical[-history_chapter_limit:]: + source_catalog.append( + { + "sourceId": f"scaffold:{row.get('id')}", + "sourceVersion": source_version, + "chapter": _required_chapter(row.get("chapter"), "scaffold.chapter"), + "scope": "chapter", + } + ) + for card_id in sorted(selected_ids): + source_catalog.append( + { + "sourceId": f"eval-draft:{card_id}", + "sourceVersion": _card_source_version(rows_by_id[card_id], source_version), + "chapterRange": f"1-{normalized_as_of}", + "scope": "card_projection", + "sourceStatus": "eval_draft", + } + ) + + recent_source_ids = [item["sourceId"] for item in projected_scaffolds] + common_context = { + "L0": { + "purpose": "offline_evaluation", + "scenario": "fine_outline", + "targetChapter": normalized_target, + "outputContract": "fine_outline_v0", + }, + "L1": { + "asOfChapter": normalized_as_of, + "recentScaffoldSourceIds": recent_source_ids, + "historyChapterLimit": history_chapter_limit, + }, + "L2": { + "referenceWorkId": str(work.get("id")), + "outlineSourceCount": len(projected_windows), + "sourceVersion": source_version, + }, + "L3": { + "sourceMode": "eval_draft", + "authorizationRequired": True, + "targetChapterAvailableOnlyToAudit": True, + }, + } + target_facts = _target_facts_from_scaffold(target_scaffold, normalized_target) + reference_work = { + "id": str(work.get("id")), + "title": str(work.get("title") or ""), + "version": source_version, + "chapterCount": reference.get("imported_chapter_count"), + "declaredChapterCount": reference.get("declared_chapter_count"), + } + run_id = run_id or f"replay-work-{work.get('id')}-{normalized_target}" + return { + "runId": run_id, + "referenceWork": reference_work, + "evaluationSetVersion": str(evaluation_set_version), + "strategyVersion": str(strategy_version), + "targetChapter": normalized_target, + "snapshot": { + "asOfChapter": normalized_as_of, + "snapshotVersion": snapshot_version, + "data": { + "outlineWindows": projected_windows, + "chapters": projected_scaffolds, + "cards": [], + }, + }, + "sources": source_catalog, + "commonContext": common_context, + "arms": { + "outline_only": {"cards": [], "cardSourceIds": [], "cardStrategy": "none"}, + "outline_plus_cards": { + "cards": correct_cards, + "cardSourceIds": [f"eval-draft:{card_id}" for card_id in correct_ids], + "cardStrategy": "correct", + }, + "outline_plus_placebo_cards": { + "cards": placebo_cards, + "cardSourceIds": [f"eval-draft:{card_id}" for card_id in placebo_ids], + "cardStrategy": "placebo", + }, + }, + "authorization": { + "sourceStatus": "missing_authorization_snapshot", + "copyrightStatus": "unknown", + "sourceVersion": source_version, + "allowedPurpose": [], + "authorizationSnapshot": {}, + }, + "runPermissions": { + "purpose": "offline_evaluation", + "mode": "dry_run", + "sourceMode": "eval_draft", + "writesFormalData": False, + }, + "leakageAudit": { + "auditVersion": "content_fact_audit_v0", + "targetFacts": target_facts, + }, + } + + +def load_reference_rows( + *, + dsn: str, + tenant_id: int, + work_id: int, + as_of: int, + target: int, + card_selection: Mapping[str, Any], +) -> dict[str, Any]: + """在只读事务中读取组装所需的作品、摘要、卡和目标 proxy。""" + + normalized_as_of = _required_chapter(as_of, "as_of") + normalized_target = _required_chapter(target, "target") + 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") + work = conn.execute( + """ + SELECT id,title,revision,chapter_count,parse_status,import_status + FROM muse_content_work + WHERE tenant_id=%s AND id=%s AND deleted=FALSE + """, + (tenant_id, work_id), + ).fetchone() + reference = conn.execute( + """ + SELECT id,work_id,declared_chapter_count,imported_chapter_count, + parse_scope,parse_status,source_file,notes,update_time + FROM example_reference_work + WHERE tenant_id=%s AND work_id=%s AND deleted=FALSE + ORDER BY id DESC LIMIT 1 + """, + (tenant_id, work_id), + ).fetchone() + if work is None or reference is None: + raise AdapterError("作品或参考作品登记不存在") + if normalized_target > int(work.get("chapter_count") or 0) + 1: + raise AdapterError("目标章超出作品导入范围") + + outline_rows = conn.execute( + """ + SELECT id,window_no,from_order,to_order,outline_text,check_status + FROM example_parse_outline + WHERE tenant_id=%s AND work_id=%s AND deleted=FALSE AND from_order<=%s + ORDER BY from_order,to_order,id + """, + (tenant_id, work_id, normalized_as_of), + ).fetchall() + scaffold_rows = conn.execute( + """ + SELECT s.id,s.chapter_id,ch.order_no AS chapter,ch.title,s.outline_text, + s.entities,s.pattern_hints + FROM example_parse_scaffold s + JOIN muse_content_chapter ch ON ch.id=s.chapter_id + WHERE s.tenant_id=%s AND s.work_id=%s AND s.deleted=FALSE + AND ch.tenant_id=%s AND ch.deleted=FALSE AND ch.order_no<=%s + ORDER BY ch.order_no,s.id + """, + (tenant_id, work_id, tenant_id, normalized_as_of), + ).fetchall() + target_scaffold = conn.execute( + """ + SELECT s.id,s.chapter_id,ch.order_no AS chapter,ch.title,s.outline_text, + s.entities,s.pattern_hints + FROM example_parse_scaffold s + JOIN muse_content_chapter ch ON ch.id=s.chapter_id + WHERE s.tenant_id=%s AND s.work_id=%s AND s.deleted=FALSE + AND ch.tenant_id=%s AND ch.deleted=FALSE AND ch.order_no=%s + ORDER BY s.id LIMIT 1 + """, + (tenant_id, work_id, tenant_id, normalized_target), + ).fetchone() + if target_scaffold is None: + raise AdapterError("目标章没有 reference scaffold proxy") + + card_rows = conn.execute( + """ + SELECT id,status,source_type,source_id,revision,draft_payload + FROM muse_knowledge_draft + WHERE tenant_id=%s AND work_id=%s AND deleted=FALSE + AND source_type='upgrade_book' AND id=ANY(%s) + ORDER BY id + """, + (tenant_id, work_id, selected_ids), + ).fetchall() + return { + "work": work, + "reference": reference, + "outline_rows": outline_rows, + "scaffold_rows": scaffold_rows, + "target_scaffold": target_scaffold, + "card_rows": card_rows, + } + + +def _parse_args() -> argparse.Namespace: + """解析只读适配器命令行参数。""" + + parser = argparse.ArgumentParser(description="从实验库只读组装回放配置") + parser.add_argument("--dsn", default=DSN) + parser.add_argument("--tenant-id", type=int, default=TENANT_ID) + parser.add_argument("--work-id", type=int, required=True) + parser.add_argument("--as-of", type=int, required=True, dest="as_of") + parser.add_argument("--target-chapter", type=int, required=True, dest="target") + parser.add_argument("--card-selection", type=Path, required=True) + parser.add_argument("--output-dir", type=Path, required=True) + parser.add_argument("--evaluation-set-version", default="deep-space-v0") + parser.add_argument("--strategy-version", default="card-index-outline-v0") + parser.add_argument("--run-id") + parser.add_argument("--history-chapter-limit", type=int, default=6) + return parser.parse_args() + + +def main() -> int: + """执行只读查询并把配置写到仓库外临时目录。""" + + args = _parse_args() + output_dir = args.output_dir.resolve() + if output_dir.is_relative_to(REPO_ROOT.resolve()): + raise AdapterError("适配器输出目录必须位于仓库外") + selection = json.loads(args.card_selection.read_text(encoding="utf-8")) + rows = load_reference_rows( + dsn=args.dsn, + tenant_id=args.tenant_id, + work_id=args.work_id, + as_of=args.as_of, + target=args.target, + card_selection=selection, + ) + config = build_replay_config( + **rows, + card_selection=selection, + as_of=args.as_of, + target=args.target, + evaluation_set_version=args.evaluation_set_version, + strategy_version=args.strategy_version, + run_id=args.run_id, + history_chapter_limit=args.history_chapter_limit, + ) + output_dir.mkdir(parents=True, exist_ok=True) + (output_dir / "config.json").write_text(_safe_json(config) + "\n", encoding="utf-8") + (output_dir / "target_proxy.json").write_text( + _safe_json(rows["target_scaffold"]) + "\n", encoding="utf-8" + ) + summary = { + "runId": config["runId"], + "workId": args.work_id, + "asOfChapter": args.as_of, + "targetChapter": args.target, + "outlineWindowCount": len(config["snapshot"]["data"]["outlineWindows"]), + "historyChapterCount": len(config["snapshot"]["data"]["chapters"]), + "correctCardCount": len(config["arms"]["outline_plus_cards"]["cards"]), + "placeboCardCount": len(config["arms"]["outline_plus_placebo_cards"]["cards"]), + "cardSourceMode": "eval_draft", + "authorizationStatus": "missing_authorization_snapshot", + "configPath": str(output_dir / "config.json"), + } + (output_dir / "adapter_summary.json").write_text(_safe_json(summary) + "\n", encoding="utf-8") + print(json.dumps(summary, ensure_ascii=False, sort_keys=True)) + return 0 + + +if __name__ == "__main__": + try: + raise SystemExit(main()) + except (AdapterError, OSError, psycopg.Error) as error: + print(json.dumps({"status": "blocked_adapter", "error": str(error)}, ensure_ascii=False)) + raise SystemExit(2) diff --git a/.claude/skills/replay-eval/scripts/run_replay.py b/.claude/skills/replay-eval/scripts/run_replay.py index ec23eb1..c5952a2 100644 --- a/.claude/skills/replay-eval/scripts/run_replay.py +++ b/.claude/skills/replay-eval/scripts/run_replay.py @@ -15,6 +15,7 @@ from pathlib import Path from typing import Any, Mapping from build_snapshot import build_snapshot, normalize_chapter, sha256_value +from audit_leakage import audit_snapshot from check_snapshot import ( STATUS_READY, check_candidate_output, @@ -107,10 +108,15 @@ def _planner_prompt( skill = SKILL_PATH.read_text(encoding="utf-8") identity = PLANNER_PATH.read_text(encoding="utf-8") + # 公共快照不能携带知识卡；卡只能通过当前评测臂的独立注入区进入上下文， + # 否则 outline_only 臂会在“无卡”名义下偷看到全局卡片。 + public_snapshot = { + str(key): value for key, value in snapshot.items() if str(key) != "cards" + } context = { "targetChapter": target, "asOfChapter": as_of, - "frozenSnapshot": snapshot, + "frozenSnapshot": public_snapshot, "commonContext": common_input, "cardInjection": cards, } @@ -257,6 +263,22 @@ def run_replay( if not preflight["ok"]: return result + leakage_audit_config = config.get("leakageAudit") + if not isinstance(leakage_audit_config, Mapping): + # 没有目标事实审计就不能把 dry-run 说成可运行，避免结构冻结掩盖内容泄露。 + result["status"] = "blocked_leakage_audit" + result["ok"] = False + result["leakageAudit"] = { + "status": "not_configured", + "ok": False, + "errors": ["缺少 leakageAudit 配置"], + "warnings": [], + "findingCount": 0, + "findings": [], + } + (output_dir / "run_result.json").write_text(_safe_json(result) + "\n", encoding="utf-8") + return result + metadata = { "targetChapter": target, "referenceWork": reference_work, @@ -274,6 +296,28 @@ def run_replay( target_chapter=target, manifest_metadata=metadata, ) + + # 内容审计同时覆盖公共冻结快照和各臂卡注入区；卡不在公共区，不能因此逃过未来事实检查。 + audit_payload = { + "snapshot": frozen["snapshot"], + "armCardInjections": { + name: _require_mapping(arms, name).get("cards", []) for name in REQUIRED_ARMS + }, + } + audit_result = audit_snapshot( + audit_payload, + leakage_audit_config.get("targetFacts"), + as_of=as_of, + target=target, + ) + result["leakageAudit"] = audit_result + if not audit_result["ok"]: + # 审计失败的快照不生成 manifest，也不允许进入任何 planner 臂。 + result["status"] = audit_result["status"] + result["ok"] = False + (output_dir / "run_result.json").write_text(_safe_json(result) + "\n", encoding="utf-8") + return result + (output_dir / "snapshot_manifest.json").write_text(_safe_json(frozen["manifest"]) + "\n", encoding="utf-8") if mode == "dry_run": diff --git a/.claude/skills/replay-eval/scripts/test_audit_leakage.py b/.claude/skills/replay-eval/scripts/test_audit_leakage.py new file mode 100644 index 0000000..ae3f00f --- /dev/null +++ b/.claude/skills/replay-eval/scripts/test_audit_leakage.py @@ -0,0 +1,75 @@ +#!/usr/bin/env python3 +"""内容级快照泄露审计的纯函数测试。""" + +from __future__ import annotations + +import json +import pathlib +import sys +import unittest + +sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent)) +from audit_leakage import audit_snapshot # noqa: E402 + + +def target_facts(): + """提供只供审计侧使用的目标章事实，不进入规划器上下文。""" + + return { + "targetChapter": 489, + "forbiddenFacts": [ + { + "id": "betrayal-489", + "firstChapter": 489, + "text": "阿铎反水击毙加特朗", + } + ], + } + + +class LeakageAuditTest(unittest.TestCase): + def test_safe_snapshot_is_ready(self): + result = audit_snapshot( + {"events": [{"chapter": 488, "fact": "环星防御仍在"}]}, + target_facts(), + as_of=488, + target=489, + ) + self.assertTrue(result["ok"]) + self.assertEqual(result["status"], "ready") + self.assertEqual(result["findingCount"], 0) + + def test_future_nested_record_blocks_snapshot(self): + result = audit_snapshot( + {"events": [{"chapter": 489, "fact": "未来事件"}]}, + target_facts(), + as_of=488, + target=489, + ) + self.assertFalse(result["ok"]) + self.assertEqual(result["status"], "invalid_snapshot") + self.assertEqual(result["findings"][0]["reason"], "future_or_crosses_as_of") + + def test_mislabelled_target_fact_blocks_snapshot_without_echoing_text(self): + result = audit_snapshot( + {"events": [{"chapter": 488, "fact": "阿铎反水击毙加特朗"}]}, + target_facts(), + as_of=488, + target=489, + ) + encoded = json.dumps(result, ensure_ascii=False) + self.assertFalse(result["ok"]) + self.assertEqual(result["status"], "invalid_snapshot") + self.assertNotIn("阿铎反水击毙加特朗", encoded) + self.assertEqual(result["findings"][0]["reason"], "target_fact_match") + self.assertIn("factHash", result["findings"][0]) + self.assertIn("valueHash", result["findings"][0]) + + def test_target_and_as_of_must_form_next_chapter_pair(self): + result = audit_snapshot({}, target_facts(), as_of=487, target=489) + self.assertFalse(result["ok"]) + self.assertEqual(result["status"], "invalid_audit_input") + + +if __name__ == "__main__": + unittest.main() diff --git a/.claude/skills/replay-eval/scripts/test_load_reference_work.py b/.claude/skills/replay-eval/scripts/test_load_reference_work.py new file mode 100644 index 0000000..c536e8b --- /dev/null +++ b/.claude/skills/replay-eval/scripts/test_load_reference_work.py @@ -0,0 +1,131 @@ +#!/usr/bin/env python3 +"""参考作品只读适配器的纯数据组装测试。""" + +from __future__ import annotations + +import json +import pathlib +import sys +import unittest + +sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent)) +from load_reference_work import ( # noqa: E402 + AdapterError, + build_replay_config, + project_card, +) + + +WORK = { + "id": 8, + "title": "深空之影", + "revision": 12, + "chapter_count": 594, + "parse_status": "pending", +} +REFERENCE = { + "declared_chapter_count": 595, + "imported_chapter_count": 594, + "parse_scope": {"from": 1, "to": 594}, + "parse_status": "parsing", +} + + +def card_row(card_id=11126, name="苏铭"): + return { + "id": card_id, + "status": "pending", + "source_type": "upgrade_book", + "source_id": 8, + "revision": 3, + "update_time": "2026-07-19T00:00:00+00:00", + "draft_payload": { + "type": "character", + "名称": name, + "别名": [f"{name}别名"], + "出场章": [1, 430, 489, 550], + "字段": { + "演变历程": [ + {"章": 420, "台阶": "历史台阶"}, + {"章": 489, "台阶": "未来台阶"}, + ], + "演变概括": "全书终态摘要，不应透传", + "境界与能力": "终局能力，不应透传", + "说话方式": "可选静态字段", + }, + "一句话摘要": "终局摘要，不应透传", + }, + } + + +class LoadReferenceWorkTest(unittest.TestCase): + def test_window_freeze_uses_absolute_bounds_and_excludes_target_window(self): + config = build_replay_config( + work=WORK, + reference=REFERENCE, + outline_rows=[ + {"id": 1, "window_no": 20, "from_order": 401, "to_order": 429, "outline_text": "历史窗", "check_status": "revised"}, + {"id": 2, "window_no": 19, "from_order": 430, "to_order": 438, "outline_text": "目标后窗", "check_status": "revised"}, + ], + scaffold_rows=[ + {"id": 10, "chapter": 428, "title": "前章", "outline_text": "前情", "entities": None, "pattern_hints": []}, + ], + target_scaffold={"id": 11, "chapter": 430, "title": "目标", "outline_text": "目标章事实"}, + card_rows=[card_row(), card_row(11127, "赵宁")], + card_selection={"correctCardIds": [11126], "placeboCardIds": [11127]}, + as_of=429, + target=430, + evaluation_set_version="set-test", + strategy_version="strategy-test", + ) + windows = config["snapshot"]["data"]["outlineWindows"] + self.assertEqual([item["from_order"] for item in windows], [401]) + public = json.dumps( + { + "snapshot": config["snapshot"], + "commonContext": config["commonContext"], + "arms": config["arms"], + "sources": config["sources"], + }, + ensure_ascii=False, + ) + 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") + encoded = json.dumps(result, ensure_ascii=False) + self.assertEqual(result["evaluationStatus"], "eval_draft") + self.assertFalse(result["productionRetrievalEligible"]) + self.assertEqual([item["章"] for item in result["milestones"]], [420]) + self.assertNotIn("终局摘要", encoded) + self.assertNotIn("终局能力", encoded) + self.assertEqual(result["source"]["sourceId"], "eval-draft:11126") + self.assertEqual(result["source"]["sourceVersion"], "db-work-8-v12") + + 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") + + def test_sources_have_source_id_and_version(self): + config = build_replay_config( + work=WORK, + reference=REFERENCE, + 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", + ) + 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") + + +if __name__ == "__main__": + unittest.main() diff --git a/.claude/skills/replay-eval/scripts/test_run_replay.py b/.claude/skills/replay-eval/scripts/test_run_replay.py index c79b1d3..c23887f 100644 --- a/.claude/skills/replay-eval/scripts/test_run_replay.py +++ b/.claude/skills/replay-eval/scripts/test_run_replay.py @@ -11,7 +11,7 @@ import tempfile import unittest sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent)) -from run_replay import run_replay # noqa: E402 +from run_replay import _planner_prompt, run_replay # noqa: E402 from write_report import render_report # noqa: E402 @@ -42,6 +42,18 @@ def config(): "strategyVersion": "strategy-v1", "authorization": AUTH, "runPermissions": {"purpose": "offline_evaluation", "mode": "dry_run"}, + "leakageAudit": { + "targetFacts": { + "targetChapter": 489, + "forbiddenFacts": [ + { + "id": "future-fact-1", + "firstChapter": 489, + "text": "目标章未进入快照", + } + ], + } + }, "targetChapter": 489, "snapshot": { "asOfChapter": 488, @@ -62,6 +74,17 @@ def config(): class ReplayRunTest(unittest.TestCase): + def test_public_planner_context_does_not_include_snapshot_cards(self): + prompt = _planner_prompt( + target=489, + as_of=488, + snapshot={"cards": [{"name": "hidden-card"}], "safe": "public"}, + common_input={}, + cards=[{"name": "injected-card"}], + ) + self.assertNotIn("hidden-card", prompt) + self.assertIn("injected-card", prompt) + def test_dry_run_passes_and_writes_only_metadata(self): with tempfile.TemporaryDirectory() as directory: result = run_replay(config(), pathlib.Path(directory), mode="dry_run") @@ -80,6 +103,36 @@ class ReplayRunTest(unittest.TestCase): self.assertEqual(result["status"], "blocked_authorization") self.assertFalse((pathlib.Path(directory) / "snapshot_manifest.json").exists()) + def test_content_leak_stops_before_manifest(self): + bad = config() + bad["leakageAudit"]["targetFacts"]["forbiddenFacts"][0]["text"] = "安全历史" + with tempfile.TemporaryDirectory() as directory: + result = run_replay(bad, pathlib.Path(directory), mode="dry_run") + self.assertFalse(result["ok"]) + self.assertEqual(result["status"], "invalid_snapshot") + self.assertEqual(result["leakageAudit"]["findingCount"], 1) + self.assertFalse((pathlib.Path(directory) / "snapshot_manifest.json").exists()) + + def test_missing_content_audit_stops_before_model(self): + bad = config() + del bad["leakageAudit"] + with tempfile.TemporaryDirectory() as directory: + result = run_replay(bad, pathlib.Path(directory), mode="dry_run") + self.assertFalse(result["ok"]) + self.assertEqual(result["status"], "blocked_leakage_audit") + self.assertFalse((pathlib.Path(directory) / "snapshot_manifest.json").exists()) + + def test_arm_card_future_record_stops_before_model(self): + bad = config() + bad["arms"]["outline_plus_cards"]["cards"] = [ + {"name": "未来卡", "milestones": [{"chapter": 489, "fact": "未来"}]} + ] + with tempfile.TemporaryDirectory() as directory: + result = run_replay(bad, pathlib.Path(directory), mode="dry_run") + self.assertFalse(result["ok"]) + self.assertEqual(result["status"], "invalid_snapshot") + self.assertFalse((pathlib.Path(directory) / "snapshot_manifest.json").exists()) + def test_report_contains_hashes_but_not_raw_fields(self): result = run_replay(config(), pathlib.Path(tempfile.mkdtemp()), mode="dry_run") report = render_report(result) diff --git a/docs/2026-07-19-回放评测-细纲首跑设计与计划.md b/docs/2026-07-19-回放评测-细纲首跑设计与计划.md index b150394..93404a4 100644 --- a/docs/2026-07-19-回放评测-细纲首跑设计与计划.md +++ b/docs/2026-07-19-回放评测-细纲首跑设计与计划.md @@ -11,10 +11,12 @@ ## 执行状态（2026-07-19） - 已落地并同步到 `agent-example/main`：`fbb262c`（冻结/细纲/评分合同）、`a54a3a4`（审查反馈收紧与回放编排）、`1243d9b`（外部 planner execute 链路测试）。 -- 已验证：回放脚本 26 个离线测试、fine-outline 合同 3 个测试全部通过；fake planner 已跑通三臂 execute 链路。测试只使用合成 fixture，不代表参考作品评测结果。 +- 已验证：回放脚本 38 个离线测试、fine-outline 合同 3 个测试全部通过；fake planner 已跑通三臂 execute 链路。测试只使用合成 fixture，不代表参考作品评测结果。 +- 已新增并验证：`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`，三臂均未调用模型。 - `.claude/skills/llm/scripts/test_quota.py` 已统一到当前 `$24/6000` 契约：预算场景使用 `$24.5`，调用上限场景使用 `6000`，并增加策略常量断言。 -- 当前允许的结论是“冻结、变量控制、候选结构门和安全摘要机制已通”；不能说细纲智能体或知识卡已通过。内容级目标事实泄露审计、真实授权接入、detector/judge 双评和 430/489/550 实样仍待完成。 +- 当前允许的结论是“冻结、变量控制、内容级泄露审计、候选结构门和安全摘要机制已通”；不能说细纲智能体或知识卡已通过。真实授权接入、detector/judge 双评和 430/489/550 实样仍待完成。 --- @@ -286,14 +288,19 @@ arm ∈ {outline_only, outline_plus_cards, outline_plus_placebo_cards} - Create: `.claude/skills/replay-eval/SKILL.md` - Create: `.claude/skills/replay-eval/scripts/build_snapshot.py` - Create: `.claude/skills/replay-eval/scripts/check_snapshot.py` +- Create: `.claude/skills/replay-eval/scripts/audit_leakage.py` +- Create: `.claude/skills/replay-eval/scripts/load_reference_work.py` - Test: `.claude/skills/replay-eval/scripts/test_snapshot.py` +- Test: `.claude/skills/replay-eval/scripts/test_audit_leakage.py` +- Test: `.claude/skills/replay-eval/scripts/test_load_reference_work.py` - [x] 建立 `next_fine_outline_replay_v0` manifest：作品、N、目标章、评估集版本、策略版本、授权快照、来源版本、运行权限包和臂配置。 - [x] 按绝对章号构造 L0/L1/L2；完整窗口必须使用 `to_order <= N`，不能依赖 `window_no`。 - [x] 将卡生成 `as_of=N` 规划视图，剔除终态字段、未来里程碑、无绝对上界条目和无法溯源字段。 - [x] 预检授权双闸、来源状态、目标章禁读、候选卡选择来源和原书全文不落最终报告。 - [x] 用离线 fixture 覆盖未来里程碑、跨 N 窗口、章号归一化、嵌套章记录、终态裁剪、缺授权失败关闭和三臂公共区哈希一致。 -- [ ] 增加目标章内容级事实泄露审计，发现错位事件时整例作废；当前只完成结构化章号冻结。 +- [x] 增加目标章内容级事实泄露审计，发现错位事件或臂内卡未来记录时整例作废；审计结果只保留路径和哈希。 +- [x] 增加 PostgreSQL 只读参考作品适配器；只读取结构化大纲/细纲摘要和预注册卡，目标章仅进入临时审计 proxy，缺授权不放行。 ### Task 2：细纲智能体功能合同 @@ -336,6 +343,7 @@ arm ∈ {outline_only, outline_plus_cards, outline_plus_placebo_cards} - [x] 编排器支持每个目标章 A/B/C 三臂，planner 使用相同身份段、功能段、预算和输出合同；尚未运行参考作品目标章。 - [ ] 两个独立 judge 对每章候选做顺序互换盲评；detector 阻断样本不送 judge。 - [x] 运行时原始候选和标准事实摘要只放仓库外临时目录；安全报告只写摘要、定位、评分入口、哈希和失败类别。 +- [x] 接入真实数据库只读适配 smoke；授权缺失时维持 `blocked_authorization`，不启动 planner。 - [ ] 产出真实卡增量矩阵和样本级假阴/假阳说明；不能用机制 smoke 的合成结果代替。 ### Task 5：滚动评估门