#!/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: """构造覆盖全部必填字段的最小合法上下文。""" 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": [], "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_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()