diff --git a/.claude/skills/read-context/scripts/retrieve_writer_sources.py b/.claude/skills/read-context/scripts/retrieve_writer_sources.py index 8491188..1eff4fd 100644 --- a/.claude/skills/read-context/scripts/retrieve_writer_sources.py +++ b/.claude/skills/read-context/scripts/retrieve_writer_sources.py @@ -392,9 +392,22 @@ def _index_hint(card: Mapping[str, Any], *, as_of: int) -> dict[str, Any]: latest = state_as_of[-1] if not isinstance(latest, Mapping): raise RetrievalError(f"卡 {card.get('cardId')} 最新冻结状态非法") - content = normalize_text( - str(latest.get("fact") or latest.get("台阶") or canonical_json(latest)) - ) + if card.get("sourceKind") == "eval_draft": + # 诊断卡只能公开索引元数据；里程碑台阶属于被测信息，不能再透传给 B/C 臂。 + content = canonical_json( + { + "name": str(card.get("name") or ""), + "type": str(card.get("type") or ""), + "sourceChapters": sorted( + {_chapter(ref.get("chapter"), "card.sourceRefs.chapter") for ref in card.get("sourceRefs", [])} + ), + } + ) + else: + # 生产 Canonical 卡保持原有冻结状态提示行为。 + content = normalize_text( + str(latest.get("fact") or latest.get("台阶") or canonical_json(latest)) + ) hint = { "cardId": str(card.get("cardId") or ""), "name": str(card.get("name") or ""), @@ -491,7 +504,11 @@ def retrieve_writer_sources( "chapter": _chapter(row.get("chapter"), "prose.chapter"), "sourceRef": copy.deepcopy(ref), "contentSha256": _content_hash(text), - "purpose": str(row.get("purpose") or "card_source"), + "purpose": ( + "card_chapter_proxy" + if ref.get("sourceType") == "card_chapter_proxy" + else str(row.get("purpose") or "card_source") + ), "text": text, "isRecentBaseline": False, } diff --git a/.claude/skills/read-context/scripts/test_retrieve_writer_sources.py b/.claude/skills/read-context/scripts/test_retrieve_writer_sources.py index 70b99eb..35edfbf 100644 --- a/.claude/skills/read-context/scripts/test_retrieve_writer_sources.py +++ b/.claude/skills/read-context/scripts/test_retrieve_writer_sources.py @@ -292,6 +292,29 @@ class RetrieveWriterSourcesTest(unittest.TestCase): [{"sourceId": "canonical-entity:2", "reason": "missing_source_refs"}], ) + def test_eval_draft_hint_only_contains_index_metadata_and_proxy_is_explicit(self): + """评测卡不透传里程碑台阶，整章代理必须保留降级来源和用途。""" + + replay_card = card("1", 0.9) + replay_card["sourceKind"] = "eval_draft" + replay_card["sourceRefs"][0]["sourceType"] = "card_chapter_proxy" + result = retrieve_writer_sources( + plan=self.plan, + card_repository=FakeCardRepository([replay_card]), + prose_repository=FakeProseRepository(), + ) + hint_content = result["indexHints"][0]["content"] + self.assertEqual( + hint_content, + '{"name":"角色1","sourceChapters":[3],"type":"character"}', + ) + self.assertNotIn("冻结线内状态", hint_content) + self.assertEqual( + result["proseEvidence"][0]["sourceRef"]["sourceType"], + "card_chapter_proxy", + ) + self.assertEqual(result["proseEvidence"][0]["purpose"], "card_chapter_proxy") + def test_snapshot_transaction_is_repeatable_read_and_read_only(self): connection = FakeConnection() begin_read_snapshot(connection) diff --git a/.claude/skills/replay-eval/SKILL.md b/.claude/skills/replay-eval/SKILL.md index f5c4d39..aa12244 100644 --- a/.claude/skills/replay-eval/SKILL.md +++ b/.claude/skills/replay-eval/SKILL.md @@ -61,6 +61,8 @@ disable-model-invocation: true 正文 Gate A 固定配置为 `configs/writer-gate-a-deep-space-v1.json`，预注册深空之影五章：489 战斗、321 人物对话、544 转折、199 信息揭示、523 老角色回归。不得根据候选结果换章、换场景分类或修改冻结点；每个冻结点必须等于 `targetChapter-1`。 +真实五章装配还必须在 `commonControls` 预注册选择器版本、选择器规范 JSON 的原始字节 SHA-256、`inputProvenance=oracle_reference_scaffold` 和统一 `maxContextChars=140000`。loader 在数据库读取前机械核对这些值：选择器篡改、来源标记变化或样本预算与公共控制不一致都失败关闭；loader 不得根据实际文本抬高预算。连续四章基线装不下时由 assembler 阻断，弱补充原文由 assembler 按预算裁剪。 + 每个样本必须提供 `writerContextInput`。仓内配置只允许 `contentMode=sanitized_contract_fixture`，`recentChapters` 只能放连续四章的脱敏合成短文本，用于证明 WriterContext 合同、章号冻结和 A/B/C 差异策略；它不是历史原文，不证明真实正文完整性或文学质量。配置可以记录细纲硬约束、主要实体、预期篇幅、新角色比例和泄漏哨兵，但不得包含原书全文、完整目标细纲或标准答案。 三臂都必须构造并校验 `WriterContext v1`，且固定 `mode=diagnostic_only`、`purpose=evaluation`、`acceptanceEligible=false`： @@ -69,13 +71,15 @@ disable-model-invocation: true - B：`evidenceStrategy=card_index_only`，只放 `retrievalResult.indexHints`，`proseEvidence` 为空；这是合同内可审计的诊断基线例外，不是生产旁路。 - C：`evidenceStrategy=card_index_plus_prose`，保留连续四章脱敏基线，并放冻结 `indexHints` 与配置中的补充原文证据。 -`indexHints` 每项只能包含 `cardId/name/type/content/sourceId/sourceVersion/asOf`，只能用于 `diagnostic_only` 的 `evaluation/diagnostic`，不得进入生产上下文；`claimLedger` 不得引用它。所有 `asOf` 和来源章号不得超过样本冻结点。 +`indexHints` 每项只能包含 `cardId/name/type/content/sourceId/sourceVersion/asOf`，只能用于 `diagnostic_only` 的 `evaluation/diagnostic`，不得进入生产上下文；`claimLedger` 不得引用它。eval_draft 的 `content` 只能编码卡名、类型和实际回读章号，不得透传里程碑台阶；生产 Canonical 卡行为不变。所有 `asOf` 和来源章号不得超过样本冻结点。 + +卡没有持久化 `sourceRefs` 时，允许把冻结线内最近三个里程碑章的整章 Canonical block 作为降级回读，但每个引用必须标记 `sourceType=card_chapter_proxy`，对应 `proseEvidence.purpose=card_chapter_proxy`，不得冒充精确片段。每个代理章必须实际出现卡规范名或选择器预注册的合法别名，否则失败关闭。 每个样本必须记录 `frozenRecentHanCounts` 和 `targetLengthBasis`。篇幅只能使用冻结点前连续四章汉字数，经 `calculate_target_chars` 复算；当前 Gate A 固定 `hardEventCount=1`、`foreshadowingActionCount=0`、`requiredSceneCount=0`，禁止读取目标章实际长度。`expectedLength` 与 WriterContext 的输出合同必须等于机械目标的正负 10%，再受 2000-10000 边界限制。 -`newCharacterRatio` 定义为：具名 `requiredCharacters` 中，在 `asOfChapter` 前无记录的角色比例。泛称角色不参与猜测；例如第 544 章的“内应”无法确定具体身份，必须记录 `newCharacterRatio=null` 和 `newCharacterRatioStatus=unresolved_generic_role`，不得默认成 0。 +`newCharacterRatio` 定义为：具名 `requiredCharacters` 中，在 `asOfChapter` 前无记录的角色比例。真实 loader 必须在同一 `REPEATABLE READ READ ONLY` 事务内直接扫描冻结历史 Canonical 正文，按名字返回首次命中章和命中章集合，再重算 `knownBeforeAsOf/absentBeforeAsOf/newCharacterRatio`；卡内容不得参与这个判定。泛称角色不参与猜测；例如第 544 章的“内应”无法确定具体身份，必须记录 `newCharacterRatio=null` 和 `newCharacterRatioStatus=unresolved_generic_role`，不得默认成 0。 -盲评使用独立的 `writer-blind-input-v1` 内容边界。judge 只能看到 `blind-1/2/3` 候选正文/哈希，以及所有臂完全相同的 `sharedEvaluationReference`：目标章细纲硬约束、实体、必需角色、伏笔、章末钩子和冻结点前连续四章历史原文基准。共同参考可以在真实运行时临时传给 judge，但不能进入安全摘要。它不得包含 `indexHints`、各臂补充原文、卡 manifest、`evidenceStrategy`、真实 A/B/C 映射、各臂 WriterContext、`raw` 目录或任何包含 `candidate-A/B/C` 的路径；卡的正确性只能由共同历史原文基准验证，不能把被测卡本身当裁判。映射只保留在编排器内存中，评分完成后才去盲。 +盲评使用独立的 `writer-blind-input-v1` 内容边界。judge 只能看到 `blind-1/2/3` 候选正文/哈希，以及所有臂完全相同的 `sharedEvaluationReference`：细纲严格四字段 `sourceRef/hardConstraints/adjustableBeats/declaredNewFacts`、必需角色、伏笔、章末钩子和冻结点前连续四章历史原文基准；不得增加 `entities`。共同参考可以在真实运行时临时传给 judge，但不能进入安全摘要。它不得包含 `indexHints`、各臂补充原文、卡 manifest、`evidenceStrategy`、真实 A/B/C 映射、各臂 WriterContext、`raw` 目录或任何包含 `candidate-A/B/C` 的路径；卡的正确性只能由共同历史原文基准验证，不能把被测卡本身当裁判。映射只保留在编排器内存中，评分完成后才去盲。 正文 dry-run 命令： diff --git a/.claude/skills/replay-eval/configs/writer-gate-a-deep-space-card-selectors-v1.json b/.claude/skills/replay-eval/configs/writer-gate-a-deep-space-card-selectors-v1.json new file mode 100644 index 0000000..c7c648a --- /dev/null +++ b/.claude/skills/replay-eval/configs/writer-gate-a-deep-space-card-selectors-v1.json @@ -0,0 +1,64 @@ +{ + "selectorVersion": "writer-gate-a-deep-space-card-selectors-v1", + "evaluationSetVersion": "writer-gate-a-deep-space-v1", + "workId": 8, + "samples": [ + { + "sampleId": "deep-space-489-battle", + "targetChapter": 489, + "cards": [ + { + "type": "location", + "name": "圣蒂曼行星" + } + ] + }, + { + "sampleId": "deep-space-321-character-dialogue", + "targetChapter": 321, + "cards": [ + { + "type": "character", + "name": "莫妮卡" + } + ] + }, + { + "sampleId": "deep-space-544-turning-point", + "targetChapter": 544, + "cards": [ + { + "type": "location", + "name": "迷途之地" + } + ] + }, + { + "sampleId": "deep-space-199-information-reveal", + "targetChapter": 199, + "cards": [ + { + "type": "character", + "name": "赛莉丝" + }, + { + "type": "event", + "name": "联赛系统被机械一族入侵", + "sourceAliases": [ + "联赛系统" + ] + } + ] + }, + { + "sampleId": "deep-space-523-returning-character", + "targetChapter": 523, + "cards": [ + { + "type": "character", + "name": "伊蕾莉雅" + } + ] + } + ] +} diff --git a/.claude/skills/replay-eval/configs/writer-gate-a-deep-space-v1.json b/.claude/skills/replay-eval/configs/writer-gate-a-deep-space-v1.json index a0446c9..30d0cd9 100644 --- a/.claude/skills/replay-eval/configs/writer-gate-a-deep-space-v1.json +++ b/.claude/skills/replay-eval/configs/writer-gate-a-deep-space-v1.json @@ -43,6 +43,10 @@ }, "commonControls": { "workId": 8, + "selectorVersion": "writer-gate-a-deep-space-card-selectors-v1", + "selectorSha256": "sha256:9df06f7d5864696deda3dd119f8cbded01ba9a9c7e5bf5655553df938f267691", + "inputProvenance": "oracle_reference_scaffold", + "maxContextChars": 140000, "modelVersion": "writer-model-unwired", "sampling": { "temperature": 0, @@ -248,7 +252,7 @@ "newSettingDeclarationRequired": true }, "tokenBudget": { - "maxContextChars": 50000 + "maxContextChars": 140000 }, "generatedAt": "2026-07-20T00:00:00Z", "requirements": { @@ -475,7 +479,7 @@ "newSettingDeclarationRequired": true }, "tokenBudget": { - "maxContextChars": 50000 + "maxContextChars": 140000 }, "generatedAt": "2026-07-20T00:00:00Z", "requirements": { @@ -709,7 +713,7 @@ "newSettingDeclarationRequired": true }, "tokenBudget": { - "maxContextChars": 50000 + "maxContextChars": 140000 }, "generatedAt": "2026-07-20T00:00:00Z", "requirements": { @@ -943,7 +947,7 @@ "newSettingDeclarationRequired": true }, "tokenBudget": { - "maxContextChars": 50000 + "maxContextChars": 140000 }, "generatedAt": "2026-07-20T00:00:00Z", "requirements": { @@ -1177,7 +1181,7 @@ "newSettingDeclarationRequired": true }, "tokenBudget": { - "maxContextChars": 50000 + "maxContextChars": 140000 }, "generatedAt": "2026-07-20T00:00:00Z", "requirements": { diff --git a/.claude/skills/replay-eval/scripts/load_writer_reference_work.py b/.claude/skills/replay-eval/scripts/load_writer_reference_work.py new file mode 100644 index 0000000..70c0f3c --- /dev/null +++ b/.claude/skills/replay-eval/scripts/load_writer_reference_work.py @@ -0,0 +1,1283 @@ +#!/usr/bin/env python3 +"""从实验库只读装配 Writer Gate A 五章真实临时配置。 + +本适配器只执行 SELECT，并把来源证明、授权、目标 scaffold、冻结近章正文和 +预注册 upgrade_book 卡固定在同一个 REPEATABLE READ READ ONLY 事务中。 +目标章正文不查询；目标 scaffold 只作为本层合法细纲和泄漏审计 proxy 使用。 +""" + +from __future__ import annotations + +import argparse +import copy +import hashlib +import json +import sys +import uuid +from pathlib import Path +from typing import Any, Mapping, Sequence + +import psycopg +from psycopg.rows import dict_row + +SCRIPT_DIR = Path(__file__).resolve().parent +READ_CONTEXT_SCRIPTS = SCRIPT_DIR.parents[1] / "read-context" / "scripts" +sys.path.insert(0, str(READ_CONTEXT_SCRIPTS)) + +from build_snapshot import normalize_chapter_range # noqa: E402 +from load_reference_work import ( # noqa: E402 + DSN, + TENANT_ID, + AdapterError, + _target_facts_from_scaffold, + begin_read_snapshot, + project_authorization, + project_card, + validate_source_records, +) +from retrieve_writer_sources import ( # noqa: E402 + ReplayCardIndexRepository, + RetrievalError, + build_retrieval_plan, + retrieve_writer_sources, +) +from writer_contract import han_count, normalize_text # noqa: E402 + + +PRIVATE_TMP = Path("/private/tmp").resolve() +DEFAULT_BASE_CONFIG = SCRIPT_DIR.parent / "configs" / "writer-gate-a-deep-space-v1.json" +DEFAULT_SELECTOR_CONFIG = ( + SCRIPT_DIR.parent / "configs" / "writer-gate-a-deep-space-card-selectors-v1.json" +) +CANONICAL_CHAPTER_STATUSES = frozenset({"published", "confirmed", "canonical"}) +EXPECTED_INPUT_PROVENANCE = "oracle_reference_scaffold" +PREREGISTERED_MAX_CONTEXT_CHARS = 140_000 + + +class WriterReferenceWorkError(AdapterError): + """真实正文装配输入缺失、歧义、漂移或越过冻结线时抛出。""" + + +def _safe_json(value: Any) -> str: + """稳定序列化临时配置，便于重复运行后比较哈希。""" + + return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + + +def _positive_chapter(value: Any, field: str) -> int: + """严格接受正整数章号，不把 bool、浮点或模糊文本猜成章号。""" + + if isinstance(value, bool): + raise WriterReferenceWorkError(f"{field} 必须是正整数章号") + if isinstance(value, str) and value.isdigit(): + value = int(value) + if not isinstance(value, int) or value <= 0: + raise WriterReferenceWorkError(f"{field} 必须是正整数章号") + return value + + +def _card_payload(row: Mapping[str, Any]) -> Mapping[str, Any]: + """读取卡 payload；选择器只信任结构化 type/name/alias。""" + + payload = row.get("draft_payload") + if not isinstance(payload, Mapping): + raise WriterReferenceWorkError(f"卡 {row.get('id')} 缺少 draft_payload 对象") + return payload + + +def _selector_samples(selector_config: Mapping[str, Any]) -> list[Mapping[str, Any]]: + """校验稳定选择器顶层结构和样本唯一性。""" + + if not isinstance(selector_config, Mapping): + raise WriterReferenceWorkError("卡选择器配置必须是对象") + samples = selector_config.get("samples") + if not isinstance(samples, list) or not samples or any( + not isinstance(item, Mapping) for item in samples + ): + raise WriterReferenceWorkError("卡选择器 samples 必须是非空对象数组") + sample_ids = [str(item.get("sampleId") or "") for item in samples] + if any(not item for item in sample_ids) or len(sample_ids) != len(set(sample_ids)): + raise WriterReferenceWorkError("卡选择器 sampleId 必须非空且唯一") + targets = [ + _positive_chapter(item.get("targetChapter"), f"{item['sampleId']}.targetChapter") + for item in samples + ] + if len(targets) != len(set(targets)): + raise WriterReferenceWorkError("卡选择器 targetChapter 必须唯一") + return samples + + +def selector_sha256(content: bytes) -> str: + """对选择器规范文件原始字节计算带算法前缀的 SHA-256。""" + + if not isinstance(content, bytes) or not content: + raise WriterReferenceWorkError("卡选择器规范文件不能为空") + return "sha256:" + hashlib.sha256(content).hexdigest() + + +def _validate_loader_controls( + base_config: Mapping[str, Any], + selector_config: Mapping[str, Any], + *, + selector_digest: str, +) -> Mapping[str, Any]: + """在读取数据库前校验预注册选择器、输入来源和统一上下文上限。""" + + if not isinstance(base_config, Mapping): + raise WriterReferenceWorkError("基础配置必须是对象") + common = base_config.get("commonControls") + if not isinstance(common, Mapping): + raise WriterReferenceWorkError("基础配置缺少 commonControls") + selector_version = str(selector_config.get("selectorVersion") or "") + if not selector_version or common.get("selectorVersion") != selector_version: + raise WriterReferenceWorkError("selectorVersion 未与预注册公共控制绑定") + if common.get("selectorSha256") != selector_digest: + raise WriterReferenceWorkError("选择器规范 JSON 的 SHA-256 与预注册公共控制不一致") + if common.get("inputProvenance") != EXPECTED_INPUT_PROVENANCE: + raise WriterReferenceWorkError("inputProvenance 必须固定为 oracle_reference_scaffold") + max_context_chars = common.get("maxContextChars") + if max_context_chars != PREREGISTERED_MAX_CONTEXT_CHARS: + raise WriterReferenceWorkError( + "commonControls.maxContextChars 必须严格等于预注册固定值 140000" + ) + samples = base_config.get("samples") + if not isinstance(samples, list) or not samples: + raise WriterReferenceWorkError("基础配置 samples 不能为空") + for index, sample in enumerate(samples): + if not isinstance(sample, Mapping): + raise WriterReferenceWorkError(f"samples[{index}] 必须是对象") + configured = sample.get("writerContextInput", {}).get("tokenBudget", {}) + if not isinstance(configured, Mapping): + raise WriterReferenceWorkError(f"samples[{index}] tokenBudget 必须是对象") + if configured.get("maxContextChars") != max_context_chars: + raise WriterReferenceWorkError( + f"samples[{index}] maxContextChars 必须原样使用预注册公共控制值" + ) + return common + + +def _required_character_probes(base_config: Mapping[str, Any]) -> list[dict[str, Any]]: + """只从冻结细纲要求提取具名角色；卡内角色状态不参与判定。""" + + probes: list[dict[str, Any]] = [] + for raw_sample in base_config.get("samples", []): + sample_id = str(raw_sample.get("sampleId") or "") + as_of = _positive_chapter(raw_sample.get("asOfChapter"), f"{sample_id}.asOfChapter") + requirements = raw_sample.get("writerContextInput", {}).get("requirements", {}) + basis = raw_sample.get("newCharacterBasis") + if not isinstance(requirements, Mapping) or not isinstance(basis, Mapping): + raise WriterReferenceWorkError(f"{sample_id} 缺少角色要求或冻结判定基线") + required = requirements.get("requiredCharacters") + generic = basis.get("genericRoles") + if ( + not isinstance(required, list) + or not required + or any(not isinstance(name, str) or not name.strip() for name in required) + or len(required) != len(set(required)) + ): + raise WriterReferenceWorkError(f"{sample_id}.requiredCharacters 必须是无重复非空字符串数组") + if ( + not isinstance(generic, list) + or any(not isinstance(name, str) or not name.strip() for name in generic) + or not set(generic).issubset(required) + ): + raise WriterReferenceWorkError(f"{sample_id}.genericRoles 必须是 requiredCharacters 子集") + named = [name for name in required if name not in set(generic)] + if generic and named: + raise WriterReferenceWorkError(f"{sample_id} 暂不允许具名角色与泛称角色混合计算比例") + probes.extend( + {"sampleId": sample_id, "name": name, "asOfChapter": as_of} + for name in named + ) + identities = [(item["sampleId"], item["name"]) for item in probes] + if len(identities) != len(set(identities)): + raise WriterReferenceWorkError("具名角色 Canonical 查询包含重复项") + return probes + + +def _canonical_character_index( + rows: Sequence[Mapping[str, Any]], + probes: Sequence[Mapping[str, Any]], +) -> dict[str, dict[str, dict[str, Any]]]: + """校验 Canonical 正文命中结果，并按样本和角色名建立只含章号的索引。""" + + expected = { + (str(item["sampleId"]), str(item["name"])): int(item["asOfChapter"]) + for item in probes + } + indexed: dict[str, dict[str, dict[str, Any]]] = {} + seen: set[tuple[str, str]] = set() + for index, raw in enumerate(rows): + if not isinstance(raw, Mapping): + raise WriterReferenceWorkError(f"canonical_character_mentions[{index}] 不是对象") + sample_id = str(raw.get("sample_id") or raw.get("sampleId") or "") + name = str(raw.get("name") or "") + identity = (sample_id, name) + if identity not in expected or identity in seen: + raise WriterReferenceWorkError("Canonical 角色命中结果含额外项或重复项") + seen.add(identity) + as_of = _positive_chapter( + raw.get("as_of_chapter", raw.get("asOfChapter")), + f"{sample_id}.{name}.asOfChapter", + ) + if as_of != expected[identity]: + raise WriterReferenceWorkError("Canonical 角色命中结果冻结点漂移") + raw_hits = raw.get("hit_chapters", raw.get("hitChapters")) or [] + if not isinstance(raw_hits, list): + raise WriterReferenceWorkError("Canonical 角色命中章必须是数组") + hits = sorted({_positive_chapter(item, f"{sample_id}.{name}.hitChapters") for item in raw_hits}) + if any(chapter > as_of for chapter in hits): + raise WriterReferenceWorkError("Canonical 角色命中结果越过冻结点") + raw_first = raw.get("first_chapter", raw.get("firstChapter")) + first = None if raw_first is None else _positive_chapter(raw_first, f"{sample_id}.{name}.firstChapter") + if first != (hits[0] if hits else None): + raise WriterReferenceWorkError("Canonical 角色首次命中章与命中章集合不一致") + indexed.setdefault(sample_id, {})[name] = { + "firstChapter": first, + "hitChapters": hits, + } + if seen != set(expected): + raise WriterReferenceWorkError("Canonical 角色命中结果缺少预注册具名角色") + return indexed + + +def _recompute_new_character_ratio( + sample: dict[str, Any], + character_index: Mapping[str, Mapping[str, Mapping[str, Any]]], +) -> None: + """依据冻结 Canonical 正文重写具名角色分区；不读取或信任卡内容。""" + + sample_id = str(sample.get("sampleId") or "") + as_of = _positive_chapter(sample.get("asOfChapter"), f"{sample_id}.asOfChapter") + requirements = sample.get("writerContextInput", {}).get("requirements", {}) + basis = sample.get("newCharacterBasis") + if not isinstance(requirements, Mapping) or not isinstance(basis, Mapping): + raise WriterReferenceWorkError(f"{sample_id} 缺少角色比例输入") + required = list(requirements.get("requiredCharacters") or []) + generic = list(basis.get("genericRoles") or []) + if generic: + sample["newCharacterRatio"] = None + sample["newCharacterRatioStatus"] = "unresolved_generic_role" + sample["newCharacterBasis"] = { + "definition": "named_required_characters_absent_before_as_of_ratio", + "asOfChapter": as_of, + "requiredCharacters": required, + "knownBeforeAsOf": [], + "absentBeforeAsOf": [], + "genericRoles": generic, + } + return + mentions = character_index.get(sample_id, {}) + known = [name for name in required if mentions.get(name, {}).get("hitChapters")] + absent = [name for name in required if name not in known] + sample["newCharacterRatio"] = len(absent) / len(required) + sample["newCharacterRatioStatus"] = "resolved" + sample["newCharacterBasis"] = { + "definition": "named_required_characters_absent_before_as_of_ratio", + "asOfChapter": as_of, + "requiredCharacters": required, + "knownBeforeAsOf": known, + "absentBeforeAsOf": absent, + "genericRoles": [], + } + + +def resolve_card_selectors( + card_rows: Sequence[Mapping[str, Any]], + selector_config: Mapping[str, Any], +) -> dict[str, list[dict[str, Any]]]: + """按 type + canonical name/alias 精确唯一解析预注册卡。 + + 这里只读取稳定配置，不接收候选分数、回放结果或目标章正文，因此运行结果 + 不可能反向改变选卡。canonical name 和 alias 都是全字符串相等匹配。 + """ + + if not isinstance(card_rows, Sequence) or isinstance(card_rows, (str, bytes)): + raise WriterReferenceWorkError("卡查询结果必须是数组") + result: dict[str, list[dict[str, Any]]] = {} + used_card_ids: set[str] = set() + for sample in _selector_samples(selector_config): + sample_id = str(sample["sampleId"]) + selectors = sample.get("cards") + if not isinstance(selectors, list) or not selectors or any( + not isinstance(item, Mapping) for item in selectors + ): + raise WriterReferenceWorkError(f"{sample_id}.cards 必须是非空对象数组") + identities: list[tuple[str, str]] = [] + selected: list[dict[str, Any]] = [] + for index, selector in enumerate(selectors): + card_type = str(selector.get("type") or "").strip() + requested_name = str(selector.get("name") or "").strip() + if not card_type or not requested_name: + raise WriterReferenceWorkError(f"{sample_id}.cards[{index}] 缺少 type/name") + identity = (card_type, requested_name) + if identity in identities: + raise WriterReferenceWorkError(f"{sample_id} 含重复卡选择器 {identity}") + identities.append(identity) + + matches: list[dict[str, Any]] = [] + for raw_row in card_rows: + if not isinstance(raw_row, Mapping): + raise WriterReferenceWorkError("卡查询结果含非对象行") + payload = _card_payload(raw_row) + if str(payload.get("type") or "") != card_type: + continue + canonical_name = str(payload.get("名称") or "") + raw_aliases = payload.get("别名") + if raw_aliases is None: + aliases: list[str] = [] + elif isinstance(raw_aliases, list) and all( + isinstance(alias, str) for alias in raw_aliases + ): + aliases = raw_aliases + else: + raise WriterReferenceWorkError(f"卡 {raw_row.get('id')} 的别名不是字符串数组") + if canonical_name == requested_name or requested_name in aliases: + matches.append(copy.deepcopy(dict(raw_row))) + if len(matches) != 1: + raise WriterReferenceWorkError( + f"{sample_id} 选择器 ({card_type},{requested_name}) 必须唯一匹配，实际 {len(matches)} 张" + ) + card_id = str(matches[0].get("id") or "") + if not card_id or card_id in used_card_ids: + raise WriterReferenceWorkError(f"卡 {card_id or ''} 被重复选择") + used_card_ids.add(card_id) + selected.append(matches[0]) + result[sample_id] = selected + return result + + +def milestone_reference_chapters( + milestones: Sequence[Mapping[str, Any]], + *, + as_of: int, + limit: int = 3, +) -> list[int]: + """展开明确里程碑区间，去重后返回冻结线内最近最多三章。""" + + freeze = _positive_chapter(as_of, "as_of") + if isinstance(limit, bool) or not isinstance(limit, int) or limit <= 0: + raise WriterReferenceWorkError("里程碑引用上限必须是正整数") + chapters: set[int] = set() + for index, milestone in enumerate(milestones): + if not isinstance(milestone, Mapping): + raise WriterReferenceWorkError(f"milestones[{index}] 不是对象") + raw_chapter = next( + ( + milestone[key] + for key in ("chapter", "chapter_no", "order_no", "章", "章号") + if key in milestone + ), + None, + ) + bounds = normalize_chapter_range(raw_chapter) + if bounds is None: + raise WriterReferenceWorkError(f"milestones[{index}] 缺少明确绝对章号边界") + # 卡可包含未来演变，但来源定位只展开完整落在冻结线内的里程碑。 + if bounds[1] > freeze: + continue + chapters.update(range(bounds[0], bounds[1] + 1)) + if not chapters: + raise WriterReferenceWorkError("卡在冻结线内没有可定位 Canonical 原文的里程碑") + return sorted(chapters)[-limit:] + + +def _normalize_card_milestones(card: dict[str, Any], *, as_of: int) -> None: + """把区间里程碑状态归一到明确结束章，供既有冻结器严格消费。""" + + normalized: list[dict[str, Any]] = [] + for index, raw in enumerate(card.get("milestones") or []): + if not isinstance(raw, Mapping): + raise WriterReferenceWorkError(f"卡 {card.get('cardId')} 里程碑[{index}] 非法") + raw_chapter = next( + (raw[key] for key in ("chapter", "chapter_no", "order_no", "章", "章号") if key in raw), + None, + ) + bounds = normalize_chapter_range(raw_chapter) + if bounds is None or bounds[1] > as_of: + raise WriterReferenceWorkError(f"卡 {card.get('cardId')} 含未冻结或无边界里程碑") + item = copy.deepcopy(dict(raw)) + for alias in ("chapter_no", "order_no", "章", "章号"): + item.pop(alias, None) + item["chapter"] = bounds[1] + normalized.append(item) + normalized.sort(key=lambda item: (item["chapter"], str(item.get("id") or ""))) + card["milestones"] = normalized + card["stateAsOf"] = copy.deepcopy(normalized) + + +def index_unique_canonical_blocks( + block_rows: Sequence[Mapping[str, Any]], + *, + allowed_chapters: set[int], +) -> dict[int, dict[str, Any]]: + """校验每个请求章恰好一个 Canonical block，并拒绝额外未来章。""" + + if not allowed_chapters: + raise WriterReferenceWorkError("Canonical block 请求章集合不能为空") + grouped: dict[int, list[dict[str, Any]]] = {} + for index, raw in enumerate(block_rows): + if not isinstance(raw, Mapping): + raise WriterReferenceWorkError(f"block_rows[{index}] 不是对象") + chapter = _positive_chapter(raw.get("chapter"), f"block_rows[{index}].chapter") + if chapter not in allowed_chapters: + raise WriterReferenceWorkError(f"读取到未请求或目标/未来章正文: {chapter}") + if str(raw.get("chapter_status") or "") not in CANONICAL_CHAPTER_STATUSES: + raise WriterReferenceWorkError(f"第 {chapter} 章不是 Canonical 状态") + text = raw.get("content_text") + if not isinstance(text, str) or not text: + raise WriterReferenceWorkError(f"第 {chapter} 章 Canonical block 正文为空") + grouped.setdefault(chapter, []).append(copy.deepcopy(dict(raw))) + missing = sorted(allowed_chapters - set(grouped)) + duplicates = sorted(chapter for chapter, rows in grouped.items() if len(rows) != 1) + if missing or duplicates: + raise WriterReferenceWorkError( + f"Canonical block 必须逐章唯一: missing={missing}, non_unique={duplicates}" + ) + return {chapter: rows[0] for chapter, rows in grouped.items()} + + +def select_recent_canonical_blocks( + block_rows: Sequence[Mapping[str, Any]], + *, + as_of: int, +) -> list[dict[str, Any]]: + """按冻结点选择连续四章，缺章或重复 block 均失败关闭。""" + + freeze = _positive_chapter(as_of, "as_of") + expected = list(range(max(1, freeze - 3), freeze + 1)) + relevant = [row for row in block_rows if row.get("chapter") in expected] + indexed = index_unique_canonical_blocks(relevant, allowed_chapters=set(expected)) + return [indexed[chapter] for chapter in expected] + + +def _block_source_ref(row: Mapping[str, Any]) -> dict[str, Any]: + """为唯一 Canonical block 生成完整代码点区间来源引用。""" + + chapter = _positive_chapter(row.get("chapter"), "block.chapter") + block_id = row.get("block_id") + if isinstance(block_id, bool) or not isinstance(block_id, int) or block_id <= 0: + raise WriterReferenceWorkError(f"第 {chapter} 章 block_id 非法") + text = str(row.get("content_text") or "") + revision = row.get("revision") or 0 + source_version = f"chapter:{chapter}:block:{block_id}:revision:{revision}" + return { + "sourceId": f"chapter:{chapter}:block:{block_id}", + "sourceVersion": source_version, + "chapter": chapter, + "blockId": block_id, + "startCodePoint": 0, + "endCodePoint": len(text), + } + + +class SnapshotProseRepository: + """只从同一数据库事务已冻结的 block 行展开来源引用。""" + + def __init__(self, block_index: Mapping[int, Mapping[str, Any]]): + self._by_chapter = { + int(chapter): copy.deepcopy(dict(row)) for chapter, row in block_index.items() + } + self._by_block = {int(row["block_id"]): row for row in self._by_chapter.values()} + + def read_source_refs( + self, + *, + work_id: int, + as_of: int, + source_refs: Sequence[Mapping[str, Any]], + ) -> list[dict[str, Any]]: + """逐引用校验章、块与代码点区间，不建立第二个数据库连接。""" + + if work_id != 8: + raise WriterReferenceWorkError("Writer Gate A 只允许预注册 work=8") + freeze = _positive_chapter(as_of, "as_of") + result: list[dict[str, Any]] = [] + for index, ref in enumerate(source_refs): + if not isinstance(ref, Mapping): + raise WriterReferenceWorkError(f"source_refs[{index}] 不是对象") + chapter = _positive_chapter(ref.get("chapter"), f"source_refs[{index}].chapter") + if chapter > freeze: + raise WriterReferenceWorkError(f"source_refs[{index}] 包含目标章或未来章") + block_id = ref.get("blockId") + row = self._by_block.get(block_id) if isinstance(block_id, int) else None + if row is None or int(row["chapter"]) != chapter: + raise WriterReferenceWorkError(f"source_refs[{index}] 不能定位唯一 Canonical block") + text = str(row["content_text"]) + start = ref.get("startCodePoint") + end = ref.get("endCodePoint") + if ( + isinstance(start, bool) + or isinstance(end, bool) + or not isinstance(start, int) + or not isinstance(end, int) + or start < 0 + or end <= start + or end > len(text) + ): + raise WriterReferenceWorkError(f"source_refs[{index}] 代码点区间越界") + fragment = text[start:end] + result.append( + { + "chapter": chapter, + "blockId": block_id, + "blockOrder": int(row.get("block_order") or 0), + "sourceRef": copy.deepcopy(dict(ref)), + "text": fragment, + "contentSha256": "sha256:" + + hashlib.sha256(fragment.encode("utf-8")).hexdigest(), + "purpose": str(ref.get("sourceType") or "card_source"), + } + ) + return result + + +def _target_scaffold_index( + rows: Sequence[Mapping[str, Any]], targets: set[int] +) -> dict[int, dict[str, Any]]: + """目标 scaffold 也要求逐章唯一，禁止 ORDER BY/LIMIT 猜选。""" + + grouped: dict[int, list[dict[str, Any]]] = {} + for index, raw in enumerate(rows): + if not isinstance(raw, Mapping): + raise WriterReferenceWorkError(f"target_scaffolds[{index}] 不是对象") + chapter = _positive_chapter(raw.get("chapter"), f"target_scaffolds[{index}].chapter") + if chapter not in targets: + raise WriterReferenceWorkError(f"读取到未预注册目标 scaffold: {chapter}") + grouped.setdefault(chapter, []).append(copy.deepcopy(dict(raw))) + missing = sorted(targets - set(grouped)) + duplicates = sorted(chapter for chapter, values in grouped.items() if len(values) != 1) + if missing or duplicates: + raise WriterReferenceWorkError( + f"目标 scaffold 必须逐章唯一: missing={missing}, non_unique={duplicates}" + ) + return {chapter: values[0] for chapter, values in grouped.items()} + + +def _selected_card_entities(cards: Sequence[Mapping[str, Any]]) -> list[dict[str, str]]: + """把稳定选择器结果转换为检索计划实体，不从运行结果追加查询。""" + + return [ + { + "id": f"selected-card:{card['cardId']}", + "type": str(card["type"]), + "name": str(card["name"]), + } + for card in cards + ] + + +def _context_token_budget( + configured: Mapping[str, Any], + *, + preregistered_max: int, +) -> dict[str, int]: + """原样使用预注册上限；基线超限由组装器失败，补充证据由组装器裁剪。""" + + configured_max = configured.get("maxContextChars") + if ( + isinstance(configured_max, bool) + or not isinstance(configured_max, int) + or configured_max <= 0 + ): + raise WriterReferenceWorkError("tokenBudget.maxContextChars 必须是正整数") + if configured_max != preregistered_max: + raise WriterReferenceWorkError("loader 禁止改写或扩张预注册 maxContextChars") + return {"maxContextChars": preregistered_max} + + +def _context_source_status(authorization: Mapping[str, Any]) -> str: + """按既有 WriterContext 授权绑定规则投影运行期来源状态。""" + + source_status = str(authorization.get("sourceStatus") or "").lower() + if source_status in {"active", "approved", "licensed"}: + return "active" + if source_status == "authorized": + return "authorized" + raise WriterReferenceWorkError(f"授权来源状态不能进入 WriterContext: {source_status}") + + +def _project_card_for_sample( + row: Mapping[str, Any], + *, + as_of: int, + source_version: str, + block_index: Mapping[int, Mapping[str, Any]], + source_aliases: Sequence[str] = (), +) -> dict[str, Any]: + """冻结卡；缺精确引用时只生成可识别且显式降级的整章代理。""" + + card_id = str(row.get("id") or "") + card_version = f"{source_version}:card-{card_id}-rev-{row.get('revision') or 0}" + projected = project_card(row, as_of=as_of, source_version=card_version) + milestones = copy.deepcopy(projected.get("milestones") or []) + uses_chapter_proxy = not projected.get("sourceRefs") + if uses_chapter_proxy: + chapters = milestone_reference_chapters(milestones, as_of=as_of) + try: + projected["sourceRefs"] = [] + payload = _card_payload(row) + raw_aliases = payload.get("别名") or [] + if not isinstance(raw_aliases, list) or any( + not isinstance(alias, str) for alias in raw_aliases + ): + raise WriterReferenceWorkError(f"卡 {card_id} 的别名不是字符串数组") + legal_names = { + str(projected.get("name") or "").strip(), + *(alias.strip() for alias in raw_aliases), + *(str(alias).strip() for alias in source_aliases), + } + legal_names.discard("") + for chapter in chapters: + row_text = normalize_text(str(block_index[chapter]["content_text"])) + if not any(name in row_text for name in legal_names): + raise WriterReferenceWorkError( + f"卡 {card_id} 的整章代理第 {chapter} 章未出现规范名或合法别名" + ) + ref = _block_source_ref(block_index[chapter]) + ref["sourceType"] = "card_chapter_proxy" + projected["sourceRefs"].append(ref) + except KeyError as error: + raise WriterReferenceWorkError( + f"卡 {card_id} 的里程碑章 {error.args[0]} 缺少唯一 Canonical block" + ) from error + for index, ref in enumerate(projected.get("sourceRefs") or []): + chapter = _positive_chapter(ref.get("chapter"), f"卡 {card_id}.sourceRefs[{index}].chapter") + if chapter > as_of: + raise WriterReferenceWorkError(f"卡 {card_id} sourceRef 包含目标章或未来章") + if ref.get("blockId") not in {row["block_id"] for row in block_index.values()}: + raise WriterReferenceWorkError(f"卡 {card_id} sourceRef 未绑定本次 Canonical 快照") + if uses_chapter_proxy and ref.get("sourceType") != "card_chapter_proxy": + raise WriterReferenceWorkError(f"卡 {card_id} 整章代理缺少降级来源类型") + _normalize_card_milestones(projected, as_of=as_of) + return projected + + +def _sample_sources( + recent_rows: Sequence[Mapping[str, Any]], + cards: Sequence[Mapping[str, Any]], + *, + as_of: int, +) -> list[dict[str, Any]]: + """生成只含冻结历史的可验证来源目录，不登记目标 scaffold 为历史来源。""" + + sources = [_block_source_ref(row) for row in recent_rows] + for card in cards: + sources.append( + { + "sourceId": str(card["sourceId"]), + "sourceVersion": str(card["sourceVersion"]), + "chapterRange": f"1-{as_of}", + "scope": "card_projection", + } + ) + return sources + + +def _recent_chapter_input(rows: Sequence[Mapping[str, Any]]) -> list[dict[str, Any]]: + """把连续四章唯一 block 投影成 WriterContext 的完整历史基线。""" + + return [ + { + "chapter": int(row["chapter"]), + "sourceRef": _block_source_ref(row), + "text": normalize_text(str(row["content_text"])), + } + for row in rows + ] + + +def _required_block_chapters( + base_config: Mapping[str, Any], + selected: Mapping[str, Sequence[Mapping[str, Any]]], +) -> set[int]: + """在正文查询前计算固定章集合，保证 SQL 不会读取目标章。""" + + required: set[int] = set() + for sample in base_config.get("samples", []): + sample_id = str(sample.get("sampleId") or "") + target = _positive_chapter(sample.get("targetChapter"), f"{sample_id}.targetChapter") + as_of = _positive_chapter(sample.get("asOfChapter"), f"{sample_id}.asOfChapter") + if target != as_of + 1: + raise WriterReferenceWorkError(f"{sample_id} targetChapter 必须等于 asOfChapter+1") + required.update(range(max(1, as_of - 3), as_of + 1)) + for row in selected.get(sample_id, []): + projected = project_card( + row, + as_of=as_of, + source_version=f"prequery:card-{row.get('id')}", + ) + refs = projected.get("sourceRefs") or [] + if refs: + for index, ref in enumerate(refs): + chapter = _positive_chapter( + ref.get("chapter"), f"{sample_id}.sourceRefs[{index}].chapter" + ) + if chapter > as_of: + raise WriterReferenceWorkError(f"{sample_id} 卡引用包含目标章或未来章") + required.add(chapter) + else: + required.update( + milestone_reference_chapters(projected.get("milestones") or [], as_of=as_of) + ) + return required + + +def load_writer_reference_rows( + *, + dsn: str, + tenant_id: int, + work_id: int, + targets: Sequence[int], + base_config: Mapping[str, Any], + selector_config: Mapping[str, Any], + selector_digest: str, +) -> dict[str, Any]: + """在同一只读可重复读事务读取五章装配所需全部数据。""" + + _validate_loader_controls( + base_config, + selector_config, + selector_digest=selector_digest, + ) + character_probes = _required_character_probes(base_config) + normalized_targets = sorted({_positive_chapter(item, "targets[]") for item in targets}) + if work_id != 8 or int(selector_config.get("workId") or 0) != work_id: + raise WriterReferenceWorkError("Writer Gate A 只允许预注册 work=8") + if not normalized_targets: + raise WriterReferenceWorkError("目标章集合不能为空") + selector_targets = sorted( + _positive_chapter(item.get("targetChapter"), f"{item['sampleId']}.targetChapter") + for item in _selector_samples(selector_config) + ) + if selector_targets != normalized_targets: + raise WriterReferenceWorkError("调用目标章必须与稳定选择器完全一致") + selector_names = sorted( + { + str(card["name"]) + for sample in _selector_samples(selector_config) + for card in sample["cards"] + } + ) + selector_types = sorted( + { + str(card["type"]) + for sample in _selector_samples(selector_config) + for card in sample["cards"] + } + ) + + with psycopg.connect(dsn, row_factory=dict_row) as conn: + begin_read_snapshot(conn) + 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_rows = conn.execute( + """ + SELECT id,work_id,declared_chapter_count,imported_chapter_count, + 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 + """, + (tenant_id, work_id), + ).fetchall() + if work is None or len(reference_rows) != 1: + raise WriterReferenceWorkError("作品或唯一参考作品登记不存在") + reference = reference_rows[0] + 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 + AND source_snapshot->>'file'=%s + ORDER BY id + """, + (tenant_id, work_id, reference.get("source_file")), + ).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) + + target_scaffolds = 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.work_id=%s AND ch.deleted=FALSE + AND ch.order_no=ANY(%s) + ORDER BY ch.order_no,s.id + """, + (tenant_id, work_id, tenant_id, work_id, normalized_targets), + ).fetchall() + card_rows = conn.execute( + """ + SELECT id,status,source_type,source_id,revision,draft_payload,deleted + FROM muse_knowledge_draft + WHERE tenant_id=%s AND work_id=%s AND deleted=FALSE + AND source_type='upgrade_book' + AND draft_payload->>'type'=ANY(%s) + AND ( + draft_payload->>'名称'=ANY(%s) + OR COALESCE(draft_payload->'别名','[]'::jsonb) ?| %s + ) + ORDER BY id + """, + (tenant_id, work_id, selector_types, selector_names, selector_names), + ).fetchall() + selected = resolve_card_selectors(card_rows, selector_config) + character_mentions = conn.execute( + """ + WITH probes AS ( + SELECT * + FROM unnest(%s::text[],%s::text[],%s::integer[]) + AS probe(sample_id,name,as_of_chapter) + ) + SELECT probe.sample_id,probe.name,probe.as_of_chapter, + MIN(ch.order_no) FILTER (WHERE b.id IS NOT NULL) AS first_chapter, + COALESCE( + ARRAY_AGG(DISTINCT ch.order_no ORDER BY ch.order_no) + FILTER (WHERE b.id IS NOT NULL), + ARRAY[]::integer[] + ) AS hit_chapters + FROM probes probe + LEFT JOIN muse_content_chapter ch + ON ch.tenant_id=%s AND ch.work_id=%s AND ch.deleted=FALSE + AND ch.status IN ('published','confirmed','canonical') + AND ch.order_no<=probe.as_of_chapter + LEFT JOIN muse_content_block b + ON b.chapter_id=ch.id AND b.tenant_id=%s AND b.work_id=%s AND b.deleted=FALSE + AND POSITION(probe.name IN b.content_text)>0 + GROUP BY probe.sample_id,probe.name,probe.as_of_chapter + ORDER BY probe.sample_id,probe.name + """, + ( + [str(item["sampleId"]) for item in character_probes], + [str(item["name"]) for item in character_probes], + [int(item["asOfChapter"]) for item in character_probes], + tenant_id, + work_id, + tenant_id, + work_id, + ), + ).fetchall() + required_chapters: set[int] = set() + for target in normalized_targets: + as_of = target - 1 + required_chapters.update(range(max(1, as_of - 3), as_of + 1)) + target_by_sample = { + str(item["sampleId"]): _positive_chapter( + item.get("targetChapter"), f"{item['sampleId']}.targetChapter" + ) + for item in selector_config["samples"] + } + for sample_id, rows in selected.items(): + as_of = target_by_sample[sample_id] - 1 + for row in rows: + projected = project_card( + row, + as_of=as_of, + source_version=f"prequery:card-{row.get('id')}", + ) + refs = projected.get("sourceRefs") or [] + if refs: + required_chapters.update( + _positive_chapter(ref.get("chapter"), "card.sourceRef.chapter") + for ref in refs + ) + else: + required_chapters.update( + milestone_reference_chapters(projected.get("milestones") or [], as_of=as_of) + ) + if any(chapter in normalized_targets for chapter in required_chapters): + raise WriterReferenceWorkError("正文读取集合包含目标章") + block_rows = conn.execute( + """ + SELECT ch.order_no AS chapter,ch.id AS chapter_id,ch.status AS chapter_status, + b.id AS block_id,b.order_no AS block_order,b.revision,b.content_text + FROM muse_content_chapter ch + JOIN muse_content_block b ON b.chapter_id=ch.id + WHERE ch.tenant_id=%s AND ch.work_id=%s AND ch.deleted=FALSE + AND b.tenant_id=%s AND b.work_id=%s AND b.deleted=FALSE + AND ch.status IN ('published','confirmed','canonical') + AND ch.order_no=ANY(%s) + ORDER BY ch.order_no,b.order_no,b.id + """, + (tenant_id, work_id, tenant_id, work_id, sorted(required_chapters)), + ).fetchall() + + index_unique_canonical_blocks(block_rows, allowed_chapters=required_chapters) + _target_scaffold_index(target_scaffolds, set(normalized_targets)) + _canonical_character_index(character_mentions, character_probes) + return { + "work": work, + "reference": reference, + "source": source, + "authorization": authorization, + "target_scaffolds": target_scaffolds, + "card_rows": [row for rows in selected.values() for row in rows], + "block_rows": block_rows, + "canonical_character_mentions": character_mentions, + } + + +def assemble_writer_gate_config( + *, + base_config: Mapping[str, Any], + selector_config: Mapping[str, Any], + selector_digest: str, + rows: Mapping[str, Any], +) -> dict[str, Any]: + """把同一事务快照装配成 canonical_frozen_prose 五样本配置。""" + + common_controls = _validate_loader_controls( + base_config, + selector_config, + selector_digest=selector_digest, + ) + if not isinstance(base_config, Mapping) or base_config.get("profile") != "writer_replay": + raise WriterReferenceWorkError("基础配置 profile 必须是 writer_replay") + samples = base_config.get("samples") + if not isinstance(samples, list) or not samples: + raise WriterReferenceWorkError("基础配置 samples 不能为空") + sample_ids = [str(item.get("sampleId") or "") for item in samples] + selector_samples = _selector_samples(selector_config) + selector_ids = [str(item["sampleId"]) for item in selector_samples] + if sample_ids != selector_ids: + raise WriterReferenceWorkError("稳定卡选择器样本顺序必须与预注册配置完全一致") + for sample, selector in zip(samples, selector_samples, strict=True): + if sample.get("targetChapter") != selector.get("targetChapter"): + raise WriterReferenceWorkError( + f"{sample.get('sampleId')} targetChapter 未绑定稳定选择器" + ) + if selector_config.get("evaluationSetVersion") != base_config.get("evaluationSetVersion"): + raise WriterReferenceWorkError("卡选择器 evaluationSetVersion 未绑定预注册配置") + work_id = int(selector_config.get("workId") or 0) + if work_id != 8 or base_config.get("referenceWork", {}).get("id") != work_id: + raise WriterReferenceWorkError("基础配置与卡选择器必须共同绑定 work=8") + + selected = resolve_card_selectors(rows.get("card_rows", []), selector_config) + selector_by_sample = {str(item["sampleId"]): item for item in selector_samples} + targets = {_positive_chapter(item.get("targetChapter"), "sample.targetChapter") for item in samples} + scaffolds = _target_scaffold_index(rows.get("target_scaffolds", []), targets) + required_chapters = _required_block_chapters(base_config, selected) + block_index = index_unique_canonical_blocks( + rows.get("block_rows", []), allowed_chapters=required_chapters + ) + source = rows.get("source") + authorization = rows.get("authorization") + work = rows.get("work") + if not all(isinstance(item, Mapping) for item in (source, authorization, work)): + raise WriterReferenceWorkError("作品、来源或授权投影缺失") + source_version = str(source.get("sourceVersion") or "") + if not source_version.startswith("raw-file-v1:sha256:"): + raise WriterReferenceWorkError("真实 Writer 配置必须绑定原文件版本") + + config = copy.deepcopy(dict(base_config)) + config["referenceWork"] = { + "id": work_id, + "title": str(work.get("title") or ""), + "version": source_version, + } + config["authorization"] = copy.deepcopy(dict(authorization)) + assembled_samples: list[dict[str, Any]] = [] + prose_repository = SnapshotProseRepository(block_index) + top_snapshot = authorization.get("authorizationSnapshot") + if not isinstance(top_snapshot, Mapping): + raise WriterReferenceWorkError("授权缺少不可变 authorizationSnapshot") + character_index = _canonical_character_index( + rows.get("canonical_character_mentions", []), + _required_character_probes(base_config), + ) + + for raw_sample in samples: + sample = copy.deepcopy(dict(raw_sample)) + sample_id = str(sample["sampleId"]) + target = _positive_chapter(sample.get("targetChapter"), f"{sample_id}.targetChapter") + as_of = _positive_chapter(sample.get("asOfChapter"), f"{sample_id}.asOfChapter") + if target != as_of + 1: + raise WriterReferenceWorkError(f"{sample_id} targetChapter 必须等于 asOfChapter+1") + scaffold = scaffolds[target] + outline_text = str(scaffold.get("outline_text") or "").strip() + if not outline_text: + raise WriterReferenceWorkError(f"{sample_id} 目标 scaffold 为空") + recent_rows = [block_index[chapter] for chapter in range(max(1, as_of - 3), as_of + 1)] + actual_counts = [han_count(str(row["content_text"])) for row in recent_rows] + expected_counts = sample.get("frozenRecentHanCounts") + if actual_counts != expected_counts: + raise WriterReferenceWorkError( + f"{sample_id} 冻结近章 Han 计数漂移: expected={expected_counts}, actual={actual_counts}" + ) + + projected_cards = [ + _project_card_for_sample( + row, + as_of=as_of, + source_version=source_version, + block_index=block_index, + source_aliases=selector_by_sample[sample_id]["cards"][index].get( + "sourceAliases", [] + ), + ) + for index, row in enumerate(selected[sample_id]) + ] + fine_outline = { + "sourceRef": { + "sourceId": f"scaffold:{scaffold.get('id')}", + "sourceVersion": source_version, + "chapter": target, + }, + "hardConstraints": [outline_text], + "adjustableBeats": copy.deepcopy( + sample.get("writerContextInput", {}).get("fineOutline", {}).get( + "adjustableBeats", [] + ) + ), + "declaredNewFacts": [], + "entities": _selected_card_entities(projected_cards), + } + token_budget = _context_token_budget( + sample.get("writerContextInput", {}).get("tokenBudget", {}), + preregistered_max=int(common_controls["maxContextChars"]), + ) + plan = build_retrieval_plan( + run_id=f"writer-loader:{sample_id}", + work_id=work_id, + target_chapter=target, + as_of=as_of, + fine_outline=fine_outline, + card_index_version=f"upgrade-book:{source_version}", + prose_index_version=source_version, + token_budget=token_budget, + ) + sources = _sample_sources(recent_rows, projected_cards, as_of=as_of) + leakage = { + "method": "target-scaffold-proxy-and-chapter-bound-audit", + "targetFacts": _target_facts_from_scaffold(scaffold, target), + } + replay_repository = ReplayCardIndexRepository.from_replay_config( + { + "targetChapter": target, + "snapshot": { + "asOfChapter": as_of, + "snapshotVersion": f"writer-gate-a-canonical-{target}-v1", + "data": { + "chapters": [ + { + "chapter": int(row["chapter"]), + "sourceId": _block_source_ref(row)["sourceId"], + "contentSha256": "sha256:" + + hashlib.sha256( + str(row["content_text"]).encode("utf-8") + ).hexdigest(), + } + for row in recent_rows + ], + "cards": [], + }, + }, + "authorization": authorization, + "sources": sources, + "leakageAudit": leakage, + }, + cards=projected_cards, + preregistered_card_ids=[str(card["cardId"]) for card in projected_cards], + ) + retrieval_result = retrieve_writer_sources( + plan=plan, + card_repository=replay_repository, + prose_repository=prose_repository, + ) + # 既有检索器会为带 sourceRefs 的卡附加卡摘要事实。Writer Gate A 明确把卡 + # 限定为索引，因此真实配置只保留索引提示和回读原文，不把摘要升级为权威事实。 + retrieval_result["factEvidence"] = [] + + context_input = copy.deepcopy(dict(sample.get("writerContextInput") or {})) + context_input.update( + { + "contentMode": "canonical_frozen_prose", + "sourceVersion": source_version, + "sourceStatus": _context_source_status(authorization), + "authorizationSnapshot": { + "snapshotId": str(top_snapshot.get("id") or ""), + "allowedPurpose": "offline_evaluation", + "verifiedAt": str(top_snapshot.get("checkedAt") or ""), + "sourceVersion": source_version, + }, + "generatedAt": str(top_snapshot.get("checkedAt") or ""), + "fineOutline": fine_outline, + "narrativeState": { + "time": "", + "location": "", + "characterPositions": {}, + "immediateSituation": "", + }, + "recentChapters": _recent_chapter_input(recent_rows), + "retrievalResult": retrieval_result, + "retrievalQueries": copy.deepcopy(plan["queries"]), + "cardIndexVersion": plan["cardIndexVersion"], + "proseIndexVersion": plan["proseIndexVersion"], + "tokenBudget": token_budget, + } + ) + sample.update( + { + "workId": work_id, + "targetTitle": str(scaffold.get("title") or sample.get("targetTitle") or ""), + "snapshotVersion": f"writer-gate-a-canonical-{target}-v1", + "snapshotData": { + "chapters": [ + { + "chapter": int(row["chapter"]), + "sourceId": _block_source_ref(row)["sourceId"], + "contentSha256": "sha256:" + + hashlib.sha256( + str(row["content_text"]).encode("utf-8") + ).hexdigest(), + } + for row in recent_rows + ], + "cards": [], + }, + "sources": sources, + "outlineSource": f"scaffold:{scaffold.get('id')}", + "fineOutlineSource": f"scaffold:{scaffold.get('id')}", + "writerContextInput": context_input, + "leakageAudit": leakage, + } + ) + _recompute_new_character_ratio(sample, character_index) + assembled_samples.append(sample) + config["samples"] = assembled_samples + return config + + +def write_temporary_config(config: Mapping[str, Any], output_dir: Path) -> Path: + """排他创建 /private/tmp 独立子目录并写入唯一完整配置。""" + + resolved = output_dir.expanduser().resolve() + if resolved == PRIVATE_TMP or not resolved.is_relative_to(PRIVATE_TMP): + raise WriterReferenceWorkError("输出目录必须位于 /private/tmp 的独立子目录") + try: + resolved.mkdir(parents=False, exist_ok=False) + except FileExistsError as error: + raise WriterReferenceWorkError("输出目录必须是尚不存在的独立子目录") from error + except FileNotFoundError as error: + raise WriterReferenceWorkError("输出目录父目录必须已存在") from error + config_path = resolved / "config.json" + config_path.write_text(_safe_json(config) + "\n", encoding="utf-8") + return config_path + + +def _parse_args() -> argparse.Namespace: + """解析真实 Writer Gate A 装配器参数。""" + + parser = argparse.ArgumentParser(description="装配 Writer Gate A 五章真实临时配置") + parser.add_argument("--dsn", default=DSN) + parser.add_argument("--tenant-id", type=int, default=TENANT_ID) + parser.add_argument("--base-config", type=Path, default=DEFAULT_BASE_CONFIG) + parser.add_argument("--card-selectors", type=Path, default=DEFAULT_SELECTOR_CONFIG) + parser.add_argument("--output-dir", type=Path) + return parser.parse_args() + + +def main() -> int: + """执行单事务只读装配并只回显安全摘要与临时配置路径。""" + + args = _parse_args() + base_config = json.loads(args.base_config.read_text(encoding="utf-8")) + selector_bytes = args.card_selectors.read_bytes() + selectors = json.loads(selector_bytes.decode("utf-8")) + selector_digest = selector_sha256(selector_bytes) + samples = base_config.get("samples") + if not isinstance(samples, list): + raise WriterReferenceWorkError("基础配置 samples 非法") + targets = [_positive_chapter(item.get("targetChapter"), "sample.targetChapter") for item in samples] + rows = load_writer_reference_rows( + dsn=args.dsn, + tenant_id=args.tenant_id, + work_id=int(selectors.get("workId") or 0), + targets=targets, + base_config=base_config, + selector_config=selectors, + selector_digest=selector_digest, + ) + config = assemble_writer_gate_config( + base_config=base_config, + selector_config=selectors, + selector_digest=selector_digest, + rows=rows, + ) + output_dir = args.output_dir or ( + PRIVATE_TMP / f"writer-gate-a-{uuid.uuid4().hex}" + ) + config_path = write_temporary_config(config, output_dir) + summary = { + "status": "ready_for_writer_dry_run", + "workId": config["referenceWork"]["id"], + "sampleCount": len(config["samples"]), + "contentMode": "canonical_frozen_prose", + "configPath": str(config_path), + } + print(json.dumps(summary, ensure_ascii=False, sort_keys=True)) + return 0 + + +if __name__ == "__main__": + try: + raise SystemExit(main()) + except ( + WriterReferenceWorkError, + RetrievalError, + OSError, + json.JSONDecodeError, + psycopg.Error, + ) as error: + print( + json.dumps( + {"status": "blocked_writer_reference_adapter", "error": str(error)}, + ensure_ascii=False, + ) + ) + raise SystemExit(2) + + +__all__ = [ + "WriterReferenceWorkError", + "SnapshotProseRepository", + "assemble_writer_gate_config", + "index_unique_canonical_blocks", + "load_writer_reference_rows", + "milestone_reference_chapters", + "resolve_card_selectors", + "selector_sha256", + "select_recent_canonical_blocks", + "write_temporary_config", +] diff --git a/.claude/skills/replay-eval/scripts/run_writer_replay.py b/.claude/skills/replay-eval/scripts/run_writer_replay.py index b44b8ee..12af405 100644 --- a/.claude/skills/replay-eval/scripts/run_writer_replay.py +++ b/.claude/skills/replay-eval/scripts/run_writer_replay.py @@ -67,6 +67,13 @@ SNAPSHOT_COMPAT_ARMS = ( "outline_plus_placebo_cards", ) CONTENT_MODES = frozenset({"sanitized_contract_fixture", "canonical_frozen_prose"}) +# Judge 只能复用写手细纲合同中的四个字段，不能直接信任原始评测配置。 +FINE_OUTLINE_JUDGE_FIELDS = ( + "sourceRef", + "hardConstraints", + "adjustableBeats", + "declaredNewFacts", +) class WriterReplayError(ValueError): @@ -768,6 +775,7 @@ def _candidate_summary(candidate: Mapping[str, Any], arm: str) -> dict[str, Any] def _build_blind_input( *, raw_sample: Mapping[str, Any], + writer_fine_outline: Mapping[str, Any], candidates: Mapping[str, Mapping[str, Any]], mapping: Mapping[str, str], blind_id: str, @@ -798,10 +806,14 @@ def _build_blind_input( ) context_input = _mapping(raw_sample.get("writerContextInput"), "writerContextInput") requirements = _mapping(context_input.get("requirements"), "writerContextInput.requirements") + if set(writer_fine_outline) != set(FINE_OUTLINE_JUDGE_FIELDS): + raise WriterReplayError("写手细纲字段必须严格匹配 judge 共同参考合同") shared_reference = { - "fineOutline": copy.deepcopy( - _mapping(context_input.get("fineOutline"), "writerContextInput.fineOutline") - ), + # 只复制写手已通过合同校验的四字段视图，禁止从原始 fineOutline 倾倒评委不可见字段。 + "fineOutline": { + field: copy.deepcopy(writer_fine_outline[field]) + for field in FINE_OUTLINE_JUDGE_FIELDS + }, "requirements": { "requiredEvents": copy.deepcopy( _sequence(requirements.get("requiredEvents"), "requirements.requiredEvents") @@ -905,11 +917,24 @@ def _execute_sample_for_test( } mapping, blind_order = _blind_mapping(evaluation_set_version, str(prepared["sampleId"])) + # 共同参考取自实际写手上下文，并在进入 judge 前确认三臂视图完全一致。 + writer_fine_outline = _mapping( + prepared["_contexts"][REQUIRED_ARMS[0]].get("fineOutline"), + "writerContext.fineOutline", + ) + for arm in REQUIRED_ARMS[1:]: + current_fine_outline = _mapping( + prepared["_contexts"][arm].get("fineOutline"), + f"writerContext.{arm}.fineOutline", + ) + if current_fine_outline != writer_fine_outline: + raise WriterReplayError("三臂写手细纲不一致，不能构造共同评测参考") reviews: dict[str, Any] = {} first_reports: list[Mapping[str, Any]] = [] for blind_id in blind_order: first_input = _build_blind_input( raw_sample=raw_sample, + writer_fine_outline=writer_fine_outline, candidates=raw_candidates, mapping=mapping, blind_id=blind_id, @@ -919,6 +944,7 @@ def _execute_sample_for_test( first_reports.append(first) second_input = _build_blind_input( raw_sample=raw_sample, + writer_fine_outline=writer_fine_outline, candidates=raw_candidates, mapping=mapping, blind_id=blind_id, @@ -929,6 +955,7 @@ def _execute_sample_for_test( if adjudication.get("status") == "needs_third_reviewer": third_input = _build_blind_input( raw_sample=raw_sample, + writer_fine_outline=writer_fine_outline, candidates=raw_candidates, mapping=mapping, blind_id=blind_id, diff --git a/.claude/skills/replay-eval/scripts/test_load_writer_reference_work.py b/.claude/skills/replay-eval/scripts/test_load_writer_reference_work.py new file mode 100644 index 0000000..a544587 --- /dev/null +++ b/.claude/skills/replay-eval/scripts/test_load_writer_reference_work.py @@ -0,0 +1,558 @@ +#!/usr/bin/env python3 +"""Writer Gate A 真实五章临时配置装配器测试。""" + +from __future__ import annotations + +import json +import pathlib +import sys +import tempfile +import unittest +from unittest.mock import patch + +SCRIPT_DIR = pathlib.Path(__file__).resolve().parent +READ_CONTEXT_SCRIPTS = SCRIPT_DIR.parents[1] / "read-context" / "scripts" +sys.path.insert(0, str(SCRIPT_DIR)) +sys.path.insert(0, str(READ_CONTEXT_SCRIPTS)) + +import load_writer_reference_work as loader # noqa: E402 +from load_writer_reference_work import ( # noqa: E402 + WriterReferenceWorkError, + assemble_writer_gate_config, + load_writer_reference_rows, + milestone_reference_chapters, + resolve_card_selectors, + write_temporary_config, +) +from run_writer_replay import run_writer_replay # noqa: E402 + + +BASE_CONFIG_PATH = SCRIPT_DIR.parent / "configs" / "writer-gate-a-deep-space-v1.json" +SELECTOR_PATH = ( + SCRIPT_DIR.parent / "configs" / "writer-gate-a-deep-space-card-selectors-v1.json" +) +SELECTOR_DIGEST = loader.selector_sha256(SELECTOR_PATH.read_bytes()) +FILE_HASH = "02cf1f8c1ca03c26e0b839d88fe536e83c0af20fd8972235b7aedca6a33becf4" +SOURCE_HASH = f"sha256:{FILE_HASH}" +SOURCE_VERSION = f"raw-file-v1:{SOURCE_HASH}" + + +def _card_row( + card_id: int, + card_type: str, + name: str, + *, + aliases: list[str] | None = None, + milestones: list[dict[str, object]] | None = None, +) -> dict[str, object]: + """构造符合 upgrade_book 冻结投影合同的卡行。""" + + return { + "id": card_id, + "status": "pending", + "source_type": "upgrade_book", + "source_id": 8, + "revision": 1, + "deleted": False, + "draft_payload": { + "type": card_type, + "名称": name, + "别名": aliases or [], + "出场章": [], + "字段": { + "演变历程": milestones or [{"章": 1, "台阶": f"{name}历史事实"}], + }, + }, + } + + +def _selected_cards() -> list[dict[str, object]]: + """构造五样本固定选择所需的六张唯一卡。""" + + return [ + _card_row(101, "location", "圣蒂曼行星", milestones=[{"章": 487, "台阶": "历史地点"}]), + _card_row(102, "character", "莫妮卡", milestones=[{"章": 317, "台阶": "历史人物"}]), + _card_row(103, "location", "迷途之地", milestones=[{"章": "539-542", "台阶": "历史地点"}]), + _card_row(104, "character", "璐茜", aliases=["赛莉丝"], milestones=[{"章": 190, "台阶": "历史身份"}]), + _card_row(105, "event", "联赛系统被机械一族入侵", milestones=[{"章": "188-189", "台阶": "历史事件"}]), + _card_row(106, "character", "伊蕾莉雅", milestones=[{"章": "460-462", "台阶": "历史人物"}]), + ] + + +def _authorization() -> dict[str, object]: + """构造通过既有授权门禁的真实原文件授权投影。""" + + snapshot = { + "id": "1", + "version": "auth-work-8-test-v1", + "immutable": True, + "sourceHash": SOURCE_HASH, + "sourceVersion": SOURCE_VERSION, + "sourceStatus": "active", + "copyrightStatus": "research_only", + "allowedPurpose": ["offline_evaluation"], + "forbiddenPurpose": ["external_distribution"], + "authorizationBasis": "user_authorization", + "authorizedBy": "user:1", + "displaySummary": "仅供离线评测", + "checkedAt": "2026-07-19T00:00:00+00:00", + "expiresAt": None, + "revalidationAt": "2099-07-19T00:00:00+00:00", + } + return { + "sourceStatus": "active", + "copyrightStatus": "research_only", + "sourceHash": SOURCE_HASH, + "sourceVersion": SOURCE_VERSION, + "allowedPurpose": ["offline_evaluation"], + "forbiddenPurpose": ["external_distribution"], + "authorizationSnapshot": snapshot, + } + + +def _block_row(chapter: int, text: str | None = None) -> dict[str, object]: + """构造单章唯一 Canonical block；默认正文长度只用于小型测试。""" + + body = text if text is not None else f"第{chapter}章历史正文" + return { + "chapter": chapter, + "chapter_id": 10000 + chapter, + "chapter_status": "published", + "block_id": 20000 + chapter, + "block_order": 1, + "revision": 1, + "content_text": body, + } + + +def _assembly_rows(base_config: dict[str, object]) -> dict[str, object]: + """按预注册 Han 计数构造可通过真实 dry-run 的纯数据快照。""" + + recent_counts: dict[int, int] = {} + target_scaffolds: list[dict[str, object]] = [] + for sample in base_config["samples"]: + for chapter, count in zip( + sample["targetLengthBasis"]["sourceChapters"], + sample["frozenRecentHanCounts"], + strict=True, + ): + recent_counts[chapter] = count + target = sample["targetChapter"] + target_scaffolds.append( + { + "id": 30000 + target, + "chapter": target, + "title": f"第{target}章目标", + "outline_text": f"第{target}章合法目标细纲，完成预注册场景。", + "entities": [], + "pattern_hints": [], + } + ) + + extra_card_chapters = {188, 189, 190, 317, 460, 461, 462, 487, 540, 541, 542} + proxy_names = { + 188: "联赛系统", + 189: "联赛系统", + 190: "赛莉丝", + 317: "莫妮卡", + 460: "伊蕾莉雅", + 461: "伊蕾莉雅", + 462: "伊蕾莉雅", + 487: "圣蒂曼行星", + 540: "迷途之地", + 541: "迷途之地", + 542: "迷途之地", + } + block_rows = [] + for chapter, count in recent_counts.items(): + prefix = proxy_names.get(chapter, "") + block_rows.append(_block_row(chapter, prefix + "文" * (count - loader.han_count(prefix)))) + block_rows.extend( + _block_row(chapter, f"第{chapter}章{proxy_names[chapter]}历史正文") + for chapter in sorted(extra_card_chapters - set(recent_counts)) + ) + return { + "work": {"id": 8, "title": "深空之影", "chapter_count": 594}, + "reference": {"id": 8, "imported_chapter_count": 594}, + "source": { + "sourceHash": SOURCE_HASH, + "sourceVersion": SOURCE_VERSION, + "fileName": "深空之影_远瞳.txt", + }, + "authorization": _authorization(), + "target_scaffolds": target_scaffolds, + "card_rows": _selected_cards(), + "block_rows": block_rows, + "canonical_character_mentions": [ + { + "sample_id": "deep-space-489-battle", + "name": "加特朗", + "as_of_chapter": 488, + "first_chapter": None, + "hit_chapters": [], + }, + { + "sample_id": "deep-space-321-character-dialogue", + "name": "莫妮卡", + "as_of_chapter": 320, + "first_chapter": 317, + "hit_chapters": [317], + }, + { + "sample_id": "deep-space-199-information-reveal", + "name": "赛莉丝", + "as_of_chapter": 198, + "first_chapter": 190, + "hit_chapters": [190], + }, + { + "sample_id": "deep-space-523-returning-character", + "name": "伊蕾莉雅", + "as_of_chapter": 522, + "first_chapter": 460, + "hit_chapters": [460, 461, 462], + }, + ], + } + + +class FakeQueryResult: + """提供 psycopg 结果对象的最小 fetch 接口。""" + + def __init__(self, rows: object): + self.rows = rows if isinstance(rows, list) else [rows] + + def fetchone(self): + """返回第一行或空值。""" + + return self.rows[0] if self.rows else None + + def fetchall(self): + """返回全部测试行。""" + + return self.rows + + +class FakeReadOnlyConnection: + """记录全部 SQL，证明所有读取都处在同一只读事务。""" + + def __init__(self, rows: dict[str, object]): + self.rows = rows + self.queries: list[str] = [] + + def __enter__(self): + """模拟 psycopg 连接上下文。""" + + return self + + def __exit__(self, exc_type, exc_value, traceback): + """测试连接不吞掉异常。""" + + return False + + def execute(self, query, params=None): + """按表名路由固定结果，并保留事务语句供断言。""" + + del params + sql = " ".join(str(query).split()) + self.queries.append(sql) + if sql.startswith("SET TRANSACTION"): + return FakeQueryResult([]) + if "FROM muse_content_work" in sql: + return FakeQueryResult(self.rows["work"]) + if "FROM example_reference_work" in sql: + return FakeQueryResult([{**self.rows["reference"], "source_file": "深空之影_远瞳.txt", "deleted": False}]) + if "FROM muse_content_import_task" in sql: + return FakeQueryResult( + [{ + "id": 11, + "status": "succeeded", + "command_id": f"import-{FILE_HASH[:16]}", + "source_snapshot": {"file": "深空之影_远瞳.txt"}, + "deleted": False, + }] + ) + if "FROM muse_knowledge_document" in sql: + return FakeQueryResult( + [{"id": 12, "file_name": "深空之影_远瞳.txt", "file_hash": FILE_HASH, "deleted": False}] + ) + if "FROM example_reference_authorization_snapshot" in sql: + snapshot = self.rows["authorization"]["authorizationSnapshot"] + return FakeQueryResult( + { + "id": 1, + "snapshot_version": snapshot["version"], + "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": snapshot["checkedAt"], + "expires_at": None, + "revalidation_at": snapshot["revalidationAt"], + } + ) + if "FROM example_parse_scaffold" in sql: + return FakeQueryResult(self.rows["target_scaffolds"]) + if "FROM muse_knowledge_draft" in sql: + return FakeQueryResult(self.rows["card_rows"]) + if "WITH probes AS" in sql: + return FakeQueryResult(self.rows["canonical_character_mentions"]) + if "FROM muse_content_chapter ch" in sql and "muse_content_block" in sql: + return FakeQueryResult(self.rows["block_rows"]) + raise AssertionError(f"未处理 SQL: {sql}") + + +class LoadWriterReferenceWorkTest(unittest.TestCase): + """覆盖选择、冻结、事务和真实 Writer dry-run 合同。""" + + def setUp(self): + """每个测试都从仓内预注册输入与稳定选择器开始。""" + + self.base_config = json.loads(BASE_CONFIG_PATH.read_text(encoding="utf-8")) + self.selectors = json.loads(SELECTOR_PATH.read_text(encoding="utf-8")) + + def test_resolves_exact_canonical_name_and_alias(self): + """canonical name 与精确 alias 都必须唯一解析，不能模糊挑卡。""" + + resolved = resolve_card_selectors(_selected_cards(), self.selectors) + self.assertEqual(resolved["deep-space-489-battle"][0]["id"], 101) + self.assertEqual(resolved["deep-space-199-information-reveal"][0]["id"], 104) + self.assertEqual( + resolved["deep-space-199-information-reveal"][0]["draft_payload"]["名称"], + "璐茜", + ) + + def test_duplicate_or_missing_selector_match_fails_closed(self): + """同一选择器命中零张或多张卡时都不能猜测。""" + + duplicate = _selected_cards() + [_card_row(999, "character", "莫妮卡")] + with self.assertRaises(WriterReferenceWorkError): + resolve_card_selectors(duplicate, self.selectors) + missing = [row for row in _selected_cards() if row["id"] != 106] + with self.assertRaises(WriterReferenceWorkError): + resolve_card_selectors(missing, self.selectors) + + def test_selector_hash_version_and_frozen_controls_fail_closed_on_tamper(self): + """选择器内容、版本、scaffold 来源和上下文上限都必须与 base 公共控制一致。""" + + self.assertEqual(self.base_config["commonControls"]["selectorSha256"], SELECTOR_DIGEST) + tampered = json.loads(json.dumps(self.selectors, ensure_ascii=False)) + tampered["samples"][0]["cards"][0]["name"] = "被篡改" + with self.assertRaisesRegex(WriterReferenceWorkError, "SHA-256"): + assemble_writer_gate_config( + base_config=self.base_config, + selector_config=tampered, + selector_digest=loader.selector_sha256( + json.dumps(tampered, ensure_ascii=False).encode("utf-8") + ), + rows=_assembly_rows(self.base_config), + ) + + invalid = json.loads(json.dumps(self.base_config, ensure_ascii=False)) + invalid["commonControls"]["inputProvenance"] = "runtime_scaffold" + with self.assertRaisesRegex(WriterReferenceWorkError, "inputProvenance"): + assemble_writer_gate_config( + base_config=invalid, + selector_config=self.selectors, + selector_digest=SELECTOR_DIGEST, + rows=_assembly_rows(self.base_config), + ) + + invalid = json.loads(json.dumps(self.base_config, ensure_ascii=False)) + invalid["samples"][0]["writerContextInput"]["tokenBudget"]["maxContextChars"] += 1 + with self.assertRaisesRegex(WriterReferenceWorkError, "原样使用"): + assemble_writer_gate_config( + base_config=invalid, + selector_config=self.selectors, + selector_digest=SELECTOR_DIGEST, + rows=_assembly_rows(self.base_config), + ) + + synchronized_tamper = json.loads(json.dumps(self.base_config, ensure_ascii=False)) + synchronized_tamper["commonControls"]["maxContextChars"] = 150000 + for sample in synchronized_tamper["samples"]: + sample["writerContextInput"]["tokenBudget"]["maxContextChars"] = 150000 + with self.assertRaisesRegex(WriterReferenceWorkError, "严格等于预注册固定值 140000"): + assemble_writer_gate_config( + base_config=synchronized_tamper, + selector_config=self.selectors, + selector_digest=SELECTOR_DIGEST, + rows=_assembly_rows(self.base_config), + ) + + def test_milestone_ranges_expand_and_keep_latest_three_unique_chapters(self): + """明确区间按绝对章展开，重复章去重后只保留最近三章。""" + + chapters = milestone_reference_chapters( + [ + {"章": 3, "台阶": "早期"}, + {"章": "5-8", "台阶": "区间"}, + {"章": 8, "台阶": "重复"}, + ], + as_of=8, + ) + self.assertEqual(chapters, [6, 7, 8]) + + def test_future_block_or_non_continuous_recent_chapters_fail_closed(self): + """目标章正文和缺章近章都不能进入装配结果。""" + + with self.assertRaises(WriterReferenceWorkError): + loader.index_unique_canonical_blocks([_block_row(9)], allowed_chapters={8}) + with self.assertRaises(WriterReferenceWorkError): + loader.select_recent_canonical_blocks( + [_block_row(5), _block_row(6), _block_row(8)], + as_of=8, + ) + repository = loader.SnapshotProseRepository({8: _block_row(8)}) + with self.assertRaises(WriterReferenceWorkError): + repository.read_source_refs( + work_id=8, + as_of=8, + source_refs=[ + { + "sourceId": "chapter:9:block:20009", + "sourceVersion": "chapter:9:block:20009:revision:1", + "chapter": 9, + "blockId": 20009, + "startCodePoint": 0, + "endCodePoint": 1, + } + ], + ) + + def test_han_count_drift_blocks_assembly(self): + """真实近章 Han 计数只要一章漂移就必须阻断。""" + + rows = _assembly_rows(self.base_config) + rows["block_rows"][0]["content_text"] += "文" + with self.assertRaisesRegex(WriterReferenceWorkError, "Han 计数漂移"): + assemble_writer_gate_config( + base_config=self.base_config, + selector_config=self.selectors, + selector_digest=SELECTOR_DIGEST, + rows=rows, + ) + + def test_all_database_reads_share_one_repeatable_read_only_transaction(self): + """来源、授权、scaffold、卡与正文必须由同一连接快照读取。""" + + rows = _assembly_rows(self.base_config) + connection = FakeReadOnlyConnection(rows) + with patch.object(loader.psycopg, "connect", return_value=connection) as connect: + loaded = load_writer_reference_rows( + dsn="postgresql://unused", + tenant_id=1, + work_id=8, + targets=[199, 321, 489, 523, 544], + base_config=self.base_config, + selector_config=self.selectors, + selector_digest=SELECTOR_DIGEST, + ) + connect.assert_called_once() + self.assertEqual( + connection.queries.count("SET TRANSACTION ISOLATION LEVEL REPEATABLE READ READ ONLY"), + 1, + ) + self.assertEqual(len(loaded["target_scaffolds"]), 5) + self.assertEqual(len(loaded["card_rows"]), 6) + mentions = {item["name"]: item for item in loaded["canonical_character_mentions"]} + self.assertEqual(mentions["加特朗"]["hit_chapters"], []) + self.assertEqual(mentions["莫妮卡"]["first_chapter"], 317) + + def test_complete_real_content_config_passes_writer_dry_run(self): + """纯数据装配出的 canonical_frozen_prose 五样本配置通过真实 dry-run。""" + + config = assemble_writer_gate_config( + base_config=self.base_config, + selector_config=self.selectors, + selector_digest=SELECTOR_DIGEST, + rows=_assembly_rows(self.base_config), + ) + self.assertEqual(len(config["samples"]), 5) + self.assertTrue( + all( + sample["writerContextInput"]["contentMode"] == "canonical_frozen_prose" + for sample in config["samples"] + ) + ) + for sample in config["samples"]: + context = sample["writerContextInput"] + retrieval = context["retrievalResult"] + self.assertEqual(retrieval["factEvidence"], []) + self.assertEqual(context["tokenBudget"]["maxContextChars"], 140000) + self.assertTrue(all(len(card["sourceRefs"]) <= 3 for card in retrieval["cards"])) + self.assertTrue( + all( + ref["sourceType"] == "card_chapter_proxy" + for card in retrieval["cards"] + for ref in card["sourceRefs"] + ) + ) + self.assertTrue( + all( + evidence["purpose"] == "card_chapter_proxy" + for evidence in retrieval["proseEvidence"] + ) + ) + self.assertTrue( + all( + evidence["chapter"] <= sample["asOfChapter"] + for evidence in [*context["recentChapters"], *retrieval["proseEvidence"]] + ) + ) + result = run_writer_replay(config, run_id="writer-loader-test") + self.assertTrue(result["ok"], result) + self.assertEqual(result["status"], "ready") + + def test_character_ratio_comes_from_canonical_text_not_contaminated_card(self): + """卡内塞入角色名不能把 Canonical 正文从未出现的角色伪装成已知。""" + + rows = _assembly_rows(self.base_config) + rows["card_rows"][0]["draft_payload"]["别名"].append("加特朗") + config = assemble_writer_gate_config( + base_config=self.base_config, + selector_config=self.selectors, + selector_digest=SELECTOR_DIGEST, + rows=rows, + ) + samples = {item["targetChapter"]: item for item in config["samples"]} + self.assertEqual(samples[489]["newCharacterBasis"]["absentBeforeAsOf"], ["加特朗"]) + self.assertEqual(samples[489]["newCharacterRatio"], 1.0) + self.assertEqual(samples[321]["newCharacterBasis"]["knownBeforeAsOf"], ["莫妮卡"]) + + def test_derived_chapter_proxy_requires_name_or_preregistered_alias_in_each_chapter(self): + """整章代理若不能在对应正文命中规范名或预注册别名，必须失败关闭。""" + + rows = _assembly_rows(self.base_config) + target = next(row for row in rows["block_rows"] if row["chapter"] == 188) + target["content_text"] = "第188章无关历史正文" + with self.assertRaisesRegex(WriterReferenceWorkError, "规范名或合法别名"): + assemble_writer_gate_config( + base_config=self.base_config, + selector_config=self.selectors, + selector_digest=SELECTOR_DIGEST, + rows=rows, + ) + + def test_output_must_be_new_private_tmp_child(self): + """配置只能写入 /private/tmp 的全新独立子目录。""" + + config = {"profile": "writer_replay"} + with tempfile.TemporaryDirectory(dir="/private/tmp") as parent: + output_dir = pathlib.Path(parent) / "writer-config" + path = write_temporary_config(config, output_dir) + self.assertEqual(path, output_dir / "config.json") + self.assertEqual(json.loads(path.read_text(encoding="utf-8")), config) + with self.assertRaises(WriterReferenceWorkError): + write_temporary_config(config, output_dir) + with self.assertRaises(WriterReferenceWorkError): + write_temporary_config(config, SCRIPT_DIR / "forbidden-output") + + +if __name__ == "__main__": + unittest.main() diff --git a/.claude/skills/replay-eval/scripts/test_run_writer_replay.py b/.claude/skills/replay-eval/scripts/test_run_writer_replay.py index 00e3277..9d57d94 100644 --- a/.claude/skills/replay-eval/scripts/test_run_writer_replay.py +++ b/.claude/skills/replay-eval/scripts/test_run_writer_replay.py @@ -547,6 +547,15 @@ class WriterReplayDryRunTest(unittest.TestCase): gate_config = json.loads(gate_config_path.read_text(encoding="utf-8")) expected_targets = {489: 7500, 321: 7600, 544: 6700, 199: 2000, 523: 6100} expected_ratios = {489: 1.0, 321: 0.0, 544: None, 199: 0.0, 523: 0.0} + self.assertEqual( + gate_config["commonControls"]["selectorSha256"], + "sha256:9df06f7d5864696deda3dd119f8cbded01ba9a9c7e5bf5655553df938f267691", + ) + self.assertEqual( + gate_config["commonControls"]["inputProvenance"], + "oracle_reference_scaffold", + ) + self.assertEqual(gate_config["commonControls"]["maxContextChars"], 140000) for sample in gate_config["samples"]: basis = sample["targetLengthBasis"] @@ -562,6 +571,10 @@ class WriterReplayDryRunTest(unittest.TestCase): self.assertEqual(recalculated, expected_targets[target_chapter]) self.assertEqual(sample["targetChars"], recalculated) self.assertEqual(sample["newCharacterRatio"], expected_ratios[target_chapter]) + self.assertEqual( + sample["writerContextInput"]["tokenBudget"]["maxContextChars"], + gate_config["commonControls"]["maxContextChars"], + ) if target_chapter == 544: self.assertEqual(sample["newCharacterRatioStatus"], "unresolved_generic_role") else: @@ -686,10 +699,21 @@ class WriterReplayExecuteBoundaryTest(unittest.TestCase): detector = FakeSemanticDetector() judge = MaliciousJudge() adapters = WriterReplayTestAdapters(runner, detector, judge) + evaluation_config = config() + fine_outline = evaluation_config["samples"][0]["writerContextInput"]["fineOutline"] + # 恶意配置模拟把目标原文、索引、文件路径和真实臂塞进细纲对象；这些字段写手看不到，评委也不能看到。 + fine_outline.update( + { + "targetOriginal": "目标章原文泄漏哨兵", + "indexHints": [{"content": "被测卡索引泄漏哨兵"}], + "path": "/private/tmp/candidate-A.json", + "arm": "A", + } + ) with tempfile.TemporaryDirectory(dir="/private/tmp") as directory: output_dir = pathlib.Path(directory) / "run" result = run_writer_replay( - config(), + evaluation_config, run_id="blind-isolation", output_dir=output_dir, execute=True, @@ -698,6 +722,20 @@ class WriterReplayExecuteBoundaryTest(unittest.TestCase): self.assertTrue(result["ok"]) self.assertTrue(judge.visible_inputs) + expected_fine_outline = { + "sourceRef": { + "sourceId": "fixture:fine-outline:489", + "sourceVersion": "fine-outline-fixture-v1", + "chapter": 489, + }, + "hardConstraints": ["必须完成围攻突围"], + "adjustableBeats": [], + "declaredNewFacts": [], + } + self.assertTrue( + all(context["fineOutline"] == expected_fine_outline for context in runner.contexts), + "三臂写手必须只看到严格细纲合同字段", + ) shared_references = [] for visible in judge.visible_inputs: self.assertEqual( @@ -714,7 +752,8 @@ class WriterReplayExecuteBoundaryTest(unittest.TestCase): self.assertEqual(set(visible["candidateOrder"]), {"blind-1", "blind-2", "blind-3"}) shared = visible["sharedEvaluationReference"] shared_references.append(copy.deepcopy(shared)) - self.assertEqual(shared["fineOutline"]["hardConstraints"], ["必须完成围攻突围"]) + self.assertEqual(shared["fineOutline"], expected_fine_outline) + self.assertNotIn("entities", shared["fineOutline"]) self.assertEqual(shared["requirements"]["requiredCharacters"], ["林澈"]) self.assertEqual( shared["requirements"]["chapterEndHook"]["anchors"], diff --git a/docs/2026-07-20-正文智能体正式优化设计与计划.md b/docs/2026-07-20-正文智能体正式优化设计与计划.md index 0195068..f583136 100644 --- a/docs/2026-07-20-正文智能体正式优化设计与计划.md +++ b/docs/2026-07-20-正文智能体正式优化设计与计划.md @@ -136,6 +136,8 @@ writer 可以创造非 Canonical 的环境细节、动作、过渡、无名配 正向创作没有回放冻结点时，`asOf` 等于当前 Canonical 最新章。 +离线 eval_draft 的 `indexHint.content` 只允许保存卡名、类型和实际回读章号，不得透传里程碑台阶。这个收紧只作用于诊断评测卡，生产 Canonical 卡继续使用既有冻结状态提示。 + ### 5.3 根据卡回读原文 首版必须保留 run3 已验证条件：**目标章之前连续 4 章全文**。选择性裁剪只能作为后续 A/B 实验变量；未证明不降质前，不得替代四章全文基线。 @@ -147,6 +149,8 @@ writer 可以创造非 Canonical 的环境细节、动作、过渡、无名配 3. 能力、物品或关系的代表性表现； 4. 与本章细纲同类的历史场景。 +若 eval_draft 卡没有持久化 `sourceRefs`，Gate A 可把冻结线内最近三个里程碑章的整章 Canonical block 作为降级代理。该来源必须同时标记 `sourceRef.sourceType=card_chapter_proxy` 和 `proseEvidence.purpose=card_chapter_proxy`，不能伪装成精确片段；每个代理章正文必须出现卡规范名或预注册合法别名，否则失败关闭。连续四章基线不可裁剪，代理等弱补充由 assembler 在固定预算内裁剪。 + `RetrievalManifest` 保存 `planId`、查询、过滤、排序、来源版本、章号、片段 offset、内容哈希和裁剪原因。`manifestId` 对规范化来源集合计算 SHA-256，排除 `runId`、时间戳和执行节点；同一 `planId` 与索引版本必须产生相同 `manifestId`。 ## 6. 双证据模型 @@ -206,6 +210,8 @@ writer 可以创造非 Canonical 的环境细节、动作、过渡、无名配 所有对象使用严格 schema，额外字段失败；引用的 ID、版本和 hash 必须存在且一致。 +Gate A 的公共控制还固定 `selectorVersion`、选择器规范 JSON 原始字节 SHA-256、`inputProvenance=oracle_reference_scaffold` 和 `maxContextChars=140000`。loader 在任何数据库读取前核对并原样沿用这些值，不得按真实输入扩大预算；若固定预算连细纲硬约束和连续四章全文基线都装不下，assembler 必须失败关闭。 + `evidenceStrategy` 是兼容字段：生产上下文未填写时按 `production_dual_evidence` 校验；正文 A/B/C 回放不得使用缺省值，必须分别显式填写 `historical_prose_only`、`card_index_only`、`card_index_plus_prose`。`indexHints` 只能在 `mode=diagnostic_only`、`purpose=evaluation/diagnostic` 且 `acceptanceEligible=false` 时出现；生产上下文硬拒绝。其 `asOf` 不得超过上下文冻结点，`claimLedger` 不得引用 `indexHints`。只有显式填写 `evidenceStrategy=card_index_only` 的 B 臂允许在合同内跳过连续四章基线，并且必须保持 `proseEvidence=[]`；该例外不能用于生产。 ### 7.2 `WriterOutput v1` @@ -273,7 +279,7 @@ detector 只阻断可定位、可验证的问题：细纲硬约束漏项、事 - 离线优化允许最多 5 轮，且每轮只改一个变量；与生产两轮返修是两套状态机。 - 三臂使用相同细纲、冻结点、模型、篇幅算法和最大生成预算。 - 候选臂名映射为随机化 ID；顺序种子由预注册 `evaluationSetVersion + sampleId` 派生并固定。 -- judge 的可信输入边界是独立 `writer-blind-input-v1` 内容，包含 `blind-1/2/3` 候选正文/哈希，以及所有臂完全相同的 `sharedEvaluationReference`。共同参考只取目标章细纲硬约束、实体、必需角色、伏笔、章末钩子和冻结点前连续四章历史原文基准，用于评价细纲忠实、设定、文风和卡索引正确性；它可在真实运行时临时传递，但不进入安全摘要。共同参考严禁包含 `indexHints`、各臂补充原文、卡 manifest、`evidenceStrategy`、真实映射或臂名，不能用被测卡本身给被测候选背书。真实 A/B/C 映射仅在编排器内存中存在，judge 不得访问各臂 WriterContext、原始运行目录或含 `candidate-A/B/C` 的路径。 +- judge 的可信输入边界是独立 `writer-blind-input-v1` 内容，包含 `blind-1/2/3` 候选正文/哈希，以及所有臂完全相同的 `sharedEvaluationReference`。共同参考的细纲只保留 `sourceRef/hardConstraints/adjustableBeats/declaredNewFacts` 四字段，不增加 `entities`；另带必需角色、伏笔、章末钩子和冻结点前连续四章历史原文基准，用于评价细纲忠实、设定、文风和卡索引正确性。它可在真实运行时临时传递，但不进入安全摘要。共同参考严禁包含 `indexHints`、各臂补充原文、卡 manifest、`evidenceStrategy`、真实映射或臂名，不能用被测卡本身给被测候选背书。真实 A/B/C 映射仅在编排器内存中存在，judge 不得访问各臂 WriterContext、原始运行目录或含 `candidate-A/B/C` 的路径。 - 两个评委使用独立无会话实例，第二评委反转顺序；评分步长为 0.5。 - 同维分差 > 0.5 时判不稳定，最多增加一次第三评委。三评分中若至少一对差值 <=0.5，则该维最终分取三者中位数；若不存在稳定配对，则样本进入 `invalid_unstable`，不参与方向结论。 - 五维：设定与实体保真、情节与细纲忠实、叙事完整与张力、文风一致、文笔质量。 @@ -307,7 +313,9 @@ detector 只阻断可定位、可验证的问题：细纲硬约束漏项、事 预期长度不是人工填写：配置记录目标章之前连续四章的 `frozenRecentHanCounts`，统一以 `hardEventCount=1`、`foreshadowingActionCount=0`、`requiredSceneCount=0` 调用 `calculate_target_chars`，且 `usesTargetChapterLength=false`。五章机械结果固定为 489=7500、321=7600、544=6700、199=2000、523=6100；上下限取正负 10% 后再受 2000-10000 限幅。 -新角色比例只统计具名 `requiredCharacters` 在冻结点前无记录的比例。489 的加特朗在 488 前无记录，因此为 1.0；321/199/523 的具名角色已有记录，因此为 0；544 的“内应”是泛称、无法机械判定具体身份，必须为 `null + unresolved_generic_role`，不允许默认成 0。 +新角色比例只统计具名 `requiredCharacters` 在冻结点前无记录的比例。真实 loader 在来源、授权、scaffold、卡和正文所在的同一个 `REPEATABLE READ READ ONLY` 事务内，直接扫描冻结历史 Canonical 正文并返回名字对应的首次命中章与命中章集合，再独立重算 `knownBeforeAsOf/absentBeforeAsOf/newCharacterRatio`，不得相信卡内是否出现该名字。489 的加特朗在 488 前无记录，因此为 1.0；321/199/523 的具名角色已有记录，因此为 0；544 的“内应”是泛称、无法机械判定具体身份，必须为 `null + unresolved_generic_role`，不允许默认成 0。 + +真实装配保留目标章 reference scaffold，并以 `inputProvenance=oracle_reference_scaffold` 明示来源。它只用于固定本次正文层细纲输入和目标事实泄漏代理；这项验证不覆盖上游清洗、抽卡、范式、细纲生成或完整创作链，不能据此宣称整条创作链完成。 仓内配置不含原文全文。`writerContextInput.contentMode=sanitized_contract_fixture`，连续四章只放脱敏合成短文本，用于 dry-run 机械证明 WriterContext 合同、冻结边界、三臂差异、manifest 可复现和候选不可接受；不能据此声称历史原文完整、生成质量通过或 real-run 完成。