371 lines
15 KiB
Python
371 lines
15 KiB
Python
#!/usr/bin/env python3
|
||
"""正文写手输入输出合同的确定性与失败关闭测试。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import copy
|
||
import hashlib
|
||
import pathlib
|
||
import sys
|
||
import unittest
|
||
|
||
sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent))
|
||
|
||
from writer_contract import ( # noqa: E402
|
||
ContractError,
|
||
calculate_target_chars,
|
||
canonical_json,
|
||
han_count,
|
||
normalize_text,
|
||
retrieval_identity,
|
||
validate_writer_context,
|
||
validate_writer_output,
|
||
)
|
||
|
||
|
||
def valid_context() -> dict:
|
||
"""构造覆盖全部必填字段的最小合法上下文。"""
|
||
|
||
prose_evidence = []
|
||
for chapter in range(485, 489):
|
||
text = f"第{chapter}章冻结原文。"
|
||
prose_evidence.append(
|
||
{
|
||
"evidenceId": f"prose:{chapter}",
|
||
"chapter": chapter,
|
||
"sourceRef": {
|
||
"sourceId": f"chapter:{chapter}",
|
||
"sourceVersion": f"chapter-{chapter}-v1",
|
||
"chapter": chapter,
|
||
"blockId": 1,
|
||
"startCodePoint": 0,
|
||
"endCodePoint": len(text),
|
||
},
|
||
"contentSha256": "sha256:" + hashlib.sha256(text.encode("utf-8")).hexdigest(),
|
||
"purpose": "recent_full_chapter",
|
||
"text": text,
|
||
"isRecentBaseline": True,
|
||
}
|
||
)
|
||
context = {
|
||
"schemaVersion": "writer-context-v1",
|
||
"runId": "writer-run-001",
|
||
"attempt": 1,
|
||
"mode": "production",
|
||
"purpose": "production",
|
||
"qualityPolicyVersion": "writer-production-v1",
|
||
"workId": 8,
|
||
"targetChapter": 489,
|
||
"asOf": 488,
|
||
"contextSnapshot": {
|
||
"manifestId": "sha256:" + "1" * 64,
|
||
"contextSha256": "sha256:" + "2" * 64,
|
||
"generatedAt": "2026-07-20T00:00:00Z",
|
||
},
|
||
"sourceVersion": "raw-file-v1:sha256:" + "3" * 64,
|
||
"authorizationSnapshot": {
|
||
"snapshotId": "auth-001",
|
||
"allowedPurpose": "generation",
|
||
"verifiedAt": "2026-07-20T00:00:00Z",
|
||
},
|
||
"sourceStatus": "active",
|
||
"retrievalPlan": {
|
||
"planVersion": "writer-retrieval-plan-v1",
|
||
"planId": "sha256:" + "4" * 64,
|
||
"runId": "writer-run-001",
|
||
"asOf": 488,
|
||
"queries": [],
|
||
"cardIndexVersion": "card-v1",
|
||
"proseIndexVersion": "prose-v1",
|
||
"filters": {
|
||
"workId": 8,
|
||
"asOfChapter": 488,
|
||
"sourceStatus": "active",
|
||
"authorizationRequired": True,
|
||
},
|
||
"tieBreak": "score DESC, sourceVersion ASC, sourceId ASC, sourceOffset ASC",
|
||
"tokenBudget": {"maxContextChars": 20000},
|
||
},
|
||
"retrievalManifest": {
|
||
"manifestVersion": "writer-retrieval-manifest-v1",
|
||
"manifestId": "sha256:" + "1" * 64,
|
||
"planId": "sha256:" + "4" * 64,
|
||
"sources": [],
|
||
"omittedSources": [],
|
||
},
|
||
"fineOutline": {
|
||
"sourceRef": {
|
||
"sourceId": "fine-outline:489",
|
||
"sourceVersion": "fine-outline-v3",
|
||
"chapter": 489,
|
||
},
|
||
"hardConstraints": ["必须完成围攻突围"],
|
||
"adjustableBeats": [],
|
||
"declaredNewFacts": [],
|
||
},
|
||
"narrativeState": {
|
||
"time": "围攻当日",
|
||
"location": "圣蒂曼",
|
||
"characterPositions": {},
|
||
"immediateSituation": "战斗持续",
|
||
},
|
||
"factEvidence": [],
|
||
"proseEvidence": prose_evidence,
|
||
"patternReferences": [],
|
||
"evidenceCoverage": [],
|
||
"outputContract": {
|
||
"targetChars": 4000,
|
||
"minChars": 3600,
|
||
"maxChars": 4400,
|
||
"frontmatterRequired": False,
|
||
"newSettingDeclarationRequired": True,
|
||
},
|
||
"tokenBudget": {"maxContextChars": 20000, "usedContextChars": 0},
|
||
"omittedSources": [],
|
||
"acceptanceEligible": True,
|
||
}
|
||
plan_payload = {key: value for key, value in context["retrievalPlan"].items() if key != "planId"}
|
||
context["retrievalPlan"]["planId"] = retrieval_identity(plan_payload)
|
||
context["retrievalManifest"]["planId"] = context["retrievalPlan"]["planId"]
|
||
manifest_payload = {key: value for key, value in context["retrievalManifest"].items() if key != "manifestId"}
|
||
context["retrievalManifest"]["manifestId"] = retrieval_identity(manifest_payload)
|
||
context["contextSnapshot"]["manifestId"] = context["retrievalManifest"]["manifestId"]
|
||
context["contextSnapshot"]["contextSha256"] = retrieval_identity(context)
|
||
return context
|
||
|
||
|
||
def valid_output() -> dict:
|
||
"""构造覆盖全部必填字段的最小合法写手输出。"""
|
||
|
||
body = normalize_text("第一段正文。")
|
||
digest = "sha256:" + hashlib.sha256(body.encode("utf-8")).hexdigest()
|
||
return {
|
||
"schemaVersion": "writer-output-v1",
|
||
"runId": "writer-run-001",
|
||
"attempt": 1,
|
||
"mode": "production",
|
||
"qualityPolicyVersion": "writer-production-v1",
|
||
"contextSnapshotId": "sha256:" + "1" * 64,
|
||
"contextSnapshotSha256": "sha256:" + "2" * 64,
|
||
"candidateVersion": 1,
|
||
"candidateSha256": digest,
|
||
"acceptanceEligible": True,
|
||
"candidateBody": body,
|
||
"claimLedger": [],
|
||
"evidenceRequests": [],
|
||
"newSettingDeclarations": [],
|
||
"selfCheck": {"hardConstraintsCovered": True, "notes": []},
|
||
}
|
||
|
||
|
||
class WriterContractTest(unittest.TestCase):
|
||
def test_index_hints_are_strict_diagnostic_only_and_frozen(self):
|
||
"""诊断卡索引提示只能进入不可接受的诊断上下文。"""
|
||
|
||
hint = {
|
||
"cardId": "card-1",
|
||
"name": "甲",
|
||
"type": "character",
|
||
"content": "甲曾在城门出现",
|
||
"sourceId": "upgrade-book:card-1",
|
||
"sourceVersion": "upgrade-book-v1",
|
||
"asOf": 488,
|
||
}
|
||
context = valid_context()
|
||
context.update(
|
||
{
|
||
"mode": "diagnostic_only",
|
||
"purpose": "evaluation",
|
||
"qualityPolicyVersion": "writer-eval-v1",
|
||
"acceptanceEligible": False,
|
||
"evidenceStrategy": "card_index_only",
|
||
"indexHints": [hint],
|
||
}
|
||
)
|
||
context["proseEvidence"] = []
|
||
context["contextSnapshot"]["contextSha256"] = retrieval_identity(context)
|
||
validate_writer_context(context)
|
||
|
||
for mutation in ("production", "unknown_field", "future"):
|
||
invalid = copy.deepcopy(context)
|
||
if mutation == "production":
|
||
invalid.update({"mode": "production", "purpose": "production", "acceptanceEligible": True})
|
||
elif mutation == "unknown_field":
|
||
invalid["indexHints"][0]["reason"] = "不得透传"
|
||
else:
|
||
invalid["indexHints"][0]["asOf"] = 489
|
||
invalid["contextSnapshot"]["contextSha256"] = retrieval_identity(invalid)
|
||
with self.subTest(mutation=mutation), self.assertRaises(ContractError):
|
||
validate_writer_context(invalid)
|
||
|
||
def test_card_index_only_requires_explicit_strategy_and_no_fact_or_prose(self):
|
||
"""B 臂跳过四章基线必须留在合同字段中，不能隐式旁路。"""
|
||
|
||
context = valid_context()
|
||
context.update(
|
||
{
|
||
"mode": "diagnostic_only",
|
||
"purpose": "diagnostic",
|
||
"qualityPolicyVersion": "writer-eval-v1",
|
||
"acceptanceEligible": False,
|
||
"evidenceStrategy": "card_index_only",
|
||
"indexHints": [{
|
||
"cardId": "card-1", "name": "甲", "type": "character",
|
||
"content": "甲的冻结状态", "sourceId": "card-source:1",
|
||
"sourceVersion": "cards-v1", "asOf": 488,
|
||
}],
|
||
}
|
||
)
|
||
context["proseEvidence"] = []
|
||
context["factEvidence"] = [{
|
||
"evidenceId": "fact-1", "fact": "甲在场", "sourceType": "formal_setting",
|
||
"sourceRef": {"sourceId": "setting:1", "sourceVersion": "setting-v1"},
|
||
"contentSha256": "sha256:" + hashlib.sha256("甲在场".encode()).hexdigest(),
|
||
"riskLevel": "high",
|
||
}]
|
||
context["contextSnapshot"]["contextSha256"] = retrieval_identity(context)
|
||
with self.assertRaises(ContractError):
|
||
validate_writer_context(context)
|
||
|
||
def test_diagnostic_prose_strategies_are_explicit_and_keep_baseline(self):
|
||
"""A/C 臂保留连续历史原文，C 臂只额外加入诊断卡索引提示。"""
|
||
|
||
hint = {
|
||
"cardId": "card-1", "name": "甲", "type": "character",
|
||
"content": "甲的冻结状态", "sourceId": "card-source:1",
|
||
"sourceVersion": "cards-v1", "asOf": 488,
|
||
}
|
||
for strategy, hints in (
|
||
("historical_prose_only", []),
|
||
("card_index_plus_prose", [hint]),
|
||
):
|
||
context = valid_context()
|
||
context.update(
|
||
{
|
||
"mode": "diagnostic_only",
|
||
"purpose": "evaluation",
|
||
"qualityPolicyVersion": "writer-eval-v1",
|
||
"acceptanceEligible": False,
|
||
"evidenceStrategy": strategy,
|
||
"indexHints": hints,
|
||
}
|
||
)
|
||
context["contextSnapshot"]["contextSha256"] = retrieval_identity(context)
|
||
with self.subTest(strategy=strategy):
|
||
self.assertEqual(
|
||
[item["chapter"] for item in validate_writer_context(context)["proseEvidence"]],
|
||
[485, 486, 487, 488],
|
||
)
|
||
|
||
def test_production_cannot_skip_or_break_recent_baseline(self):
|
||
"""生产上下文不能借诊断策略或不连续近章绕过四章基线。"""
|
||
|
||
for mutation in ("diagnostic_strategy", "missing_baseline"):
|
||
context = valid_context()
|
||
if mutation == "diagnostic_strategy":
|
||
context["evidenceStrategy"] = "historical_prose_only"
|
||
else:
|
||
context["proseEvidence"].pop(1)
|
||
context["contextSnapshot"]["contextSha256"] = retrieval_identity(context)
|
||
with self.subTest(mutation=mutation), self.assertRaises(ContractError):
|
||
validate_writer_context(context)
|
||
def test_run_id_does_not_change_retrieval_identity(self):
|
||
first = {"runId": "run-a", "query": "咖啡\u0301", "nested": {"value": 1}}
|
||
second = {"runId": "run-b", "query": "咖啡\u0301", "nested": {"value": 1}}
|
||
self.assertEqual(retrieval_identity(first), retrieval_identity(second))
|
||
|
||
def test_text_is_nfc_and_lf_before_offsets_and_hash(self):
|
||
decomposed = "Cafe\u0301\r\n第二行\r第三行"
|
||
normalized = "Caf\u00e9\n第二行\n第三行"
|
||
self.assertEqual(normalize_text(decomposed), normalized)
|
||
self.assertEqual(canonical_json({"text": decomposed}), canonical_json({"text": normalized}))
|
||
|
||
def test_han_count_does_not_count_markdown_or_non_han_text(self):
|
||
self.assertEqual(han_count("# **正文** 123 ABC，扩展𠀀"), 5)
|
||
|
||
def test_target_chars_use_half_up_and_hard_bounds(self):
|
||
self.assertEqual(
|
||
calculate_target_chars(
|
||
recent_chapter_han_counts=[2501, 2502, 2503, 2504],
|
||
hard_event_count=3,
|
||
foreshadowing_action_count=0,
|
||
required_scene_count=0,
|
||
),
|
||
2100,
|
||
)
|
||
self.assertEqual(calculate_target_chars(explicit_target_chars=1500), 2000)
|
||
self.assertEqual(calculate_target_chars(explicit_target_chars=11000), 10000)
|
||
self.assertEqual(
|
||
calculate_target_chars(
|
||
recent_chapter_han_counts=[3000, 3000, 3000],
|
||
hard_event_count=0,
|
||
foreshadowing_action_count=0,
|
||
required_scene_count=0,
|
||
),
|
||
2600,
|
||
)
|
||
|
||
def test_evaluation_and_diagnostic_contexts_are_never_acceptable(self):
|
||
for purpose in ("evaluation", "diagnostic"):
|
||
context = valid_context()
|
||
context["mode"] = "diagnostic_only"
|
||
context["purpose"] = purpose
|
||
context["qualityPolicyVersion"] = "writer-eval-v1"
|
||
context["acceptanceEligible"] = True
|
||
with self.assertRaises(ContractError):
|
||
validate_writer_context(context)
|
||
|
||
context["acceptanceEligible"] = False
|
||
context["contextSnapshot"]["contextSha256"] = retrieval_identity(context)
|
||
validate_writer_context(context)
|
||
|
||
def test_unknown_missing_and_wrong_version_fail_closed(self):
|
||
for mutation in ("unknown", "missing", "version"):
|
||
context = copy.deepcopy(valid_context())
|
||
if mutation == "unknown":
|
||
context["unexpected"] = True
|
||
elif mutation == "missing":
|
||
del context["fineOutline"]
|
||
else:
|
||
context["schemaVersion"] = "writer-context-v2"
|
||
with self.subTest(mutation=mutation), self.assertRaises(ContractError):
|
||
validate_writer_context(context)
|
||
|
||
output = valid_output()
|
||
output["selfCheck"]["unknown"] = True
|
||
with self.assertRaises(ContractError):
|
||
validate_writer_output(output)
|
||
|
||
def test_candidate_hash_and_diagnostic_acceptance_are_checked(self):
|
||
output = valid_output()
|
||
validate_writer_output(output)
|
||
|
||
output["candidateBody"] = "被修改的正文"
|
||
with self.assertRaises(ContractError):
|
||
validate_writer_output(output)
|
||
|
||
diagnostic = valid_output()
|
||
diagnostic["mode"] = "diagnostic_only"
|
||
diagnostic["qualityPolicyVersion"] = "writer-eval-v1"
|
||
diagnostic["acceptanceEligible"] = True
|
||
with self.assertRaises(ContractError):
|
||
validate_writer_output(diagnostic)
|
||
|
||
def test_plan_manifest_and_context_identity_tampering_fails_closed(self):
|
||
for path in ("plan", "manifest", "context"):
|
||
context = valid_context()
|
||
if path == "plan":
|
||
context["retrievalPlan"]["cardIndexVersion"] = "tampered"
|
||
elif path == "manifest":
|
||
context["retrievalManifest"]["sources"].append(
|
||
{"sourceId": "setting:1", "sourceVersion": "setting-v1"}
|
||
)
|
||
else:
|
||
context["fineOutline"]["hardConstraints"].append("被篡改的约束")
|
||
with self.subTest(path=path), self.assertRaises(ContractError):
|
||
validate_writer_context(context)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|