import copy import json from pathlib import Path import pytest from pydantic import ValidationError from test_support.external_qa import require_external_qa from yuqei_sdk.contract_drafting import ( ContractDraftOutput, ContractDraftRequest, draft_digest, draft_output_schema, validate_draft_output, ) from yuqei_sdk.draft_change_evidence import ( CHANGE_EVIDENCE_POLICY, DraftChangeDeclarations, assess_draft_changes, build_change_evidence_context, ) BASE = "第一条：价款为100元。\n\n第二条：通知应书面送达。" def request(*, enabled=True, base=BASE): context = build_change_evidence_context(base) snapshot = {"id": "tpl-1", "clause_assembly": {"items": [{"clause_id": "price"}]}} if enabled: snapshot.update(change_evidence_policy=CHANGE_EVIDENCE_POLICY, change_evidence_context=context) return ContractDraftRequest(contract_title="测试", contract_type="service", base_document=base, facts=[{"name": "价款", "value": "100元"}], template_snapshot=snapshot, playbook_snapshot={"rules": [{"id": "rule-1"}]}, retrieved_sources=[{"id": "source-1", "title": "来源", "content": "来源正文"}]) def output(body, declarations=None): return ContractDraftOutput(schema_version="contract-draft.v1", status="candidate_ready", draft_document=body, missing_fields=[], assumptions=[], risk_notes=[], fact_bindings=[{"field": "价款", "value": "100元", "document_excerpt": "100元"}], source_bindings=[], change_explanations=["说明"], quality_findings=[], change_declarations=declarations) def declaration(**updates): payload = {"schema_version": "draft-change-declarations.v1", "items": [{ "id": "change-1", "target_id": "base-0002", "operation": "replace", "before_excerpt": "书面送达", "after_excerpt": "电子邮件送达", "basis": [{"type": "ai_suggestion"}], "summary": "调整通知方式", "pending_confirmation": True, }]} payload["items"][0].update(updates) return payload def test_context_covers_every_codepoint_and_is_deterministic(): context = build_change_evidence_context(BASE) assert "".join(item["text"] for item in context["targets"]) == BASE assert [(item["start"], item["end"]) for item in context["targets"]] == [(0, 14), (14, len(BASE))] assert context == build_change_evidence_context(BASE) assert context["range_unit"] == "unicode_codepoint_half_open" def test_verified_replace_produces_server_hashes_ranges_and_false_disclaimers(): candidate = BASE.replace("书面送达", "电子邮件送达") assessment = assess_draft_changes(request(), output(candidate, declaration())) assert assessment.status == "complete" assert assessment.items[0].status == "verified" assert assessment.items[0].before_range is not None and assessment.items[0].after_range is not None assert assessment.uncovered_diff_ranges == [] assert assessment.semantic_equivalence_verified is False assert assessment.legal_correctness_verified is False assert assessment.authorizes_adoption is False @pytest.mark.parametrize("candidate", [ BASE.replace("通知应书面送达", "通知应书面或电子邮件送达"), BASE.replace("通知应书面送达", "通知应送达"), ]) def test_whole_paragraph_replace_accepts_internal_pure_insert_or_delete(candidate): evidence = declaration( before_excerpt="第二条：通知应书面送达。", after_excerpt=candidate.split("\n\n", 1)[1], basis=[{"type": "template", "ref": "tpl-1"}], pending_confirmation=False, ) assessment = assess_draft_changes(request(), output(candidate, evidence)) assert assessment.status == "complete" assert assessment.items[0].status == "verified" assert assessment.uncovered_diff_ranges == [] def test_saved_real_whole_paragraph_replacements_bind_to_their_internal_edits(): require_external_qa( "multiscene-closeout-20260910/draft-timeout-diagnostic-real-v1/01-draft-before.json", "multiscene-closeout-20260910/draft-timeout-diagnostic-real-v1/01-draft-after.json", ) root = Path(__file__).resolve().parents[2] evidence_dir = root / "docs/qa/multiscene-closeout-20260910/draft-timeout-diagnostic-real-v1" before = json.loads((evidence_dir / "01-draft-before.json").read_text(encoding="utf-8")) after = json.loads((evidence_dir / "01-draft-after.json").read_text(encoding="utf-8")) req = ContractDraftRequest.model_validate(before["request"]) out = ContractDraftOutput.model_validate(after["job"]["result"]["output"]) assessment = assess_draft_changes(req, out) affected = {item.declaration_id: item for item in assessment.items if item.declaration_id in {"chg-0003", "chg-0005"}} assert set(affected) == {"chg-0003", "chg-0005"} assert all(item.status == "verified" for item in affected.values()) assert all("replace_ranges_not_same_hunk" not in item.errors for item in affected.values()) def test_whole_paragraph_replace_cannot_bind_an_edit_from_an_unrelated_target(): candidate = BASE.replace("价款为100元", "价款为200元") evidence = declaration( before_excerpt="第二条：通知应书面送达。", after_excerpt="第一条：价款为200元。", basis=[{"type": "template", "ref": "tpl-1"}], pending_confirmation=False, ) assessment = assess_draft_changes(request(), output(candidate, evidence)) assert assessment.status == "invalid" assert assessment.items[0].status == "invalid" @pytest.mark.parametrize("operation,fields", [ ("insert", {"before_excerpt": "旧", "after_excerpt": "新"}), ("delete", {"before_excerpt": "旧", "after_excerpt": "新"}), ("replace", {"before_excerpt": "相同", "after_excerpt": "相同"}), ]) def test_operation_shape_is_strict(operation, fields): with pytest.raises(ValidationError): DraftChangeDeclarations.model_validate(declaration(operation=operation, **fields)) def test_repeated_excerpt_requires_independent_occurrence_or_anchor(): base = "重复。重复。" req = request(base=base) ambiguous = output("改。重复。", declaration(target_id="base-0001", before_excerpt="重复", after_excerpt="改", pending_confirmation=True)) assessment = assess_draft_changes(req, ambiguous) assert assessment.status == "invalid" assert assessment.items[0].status == "ambiguous" resolved = ambiguous.model_copy(deep=True) resolved.change_declarations.items[0].before_occurrence = 1 resolved.change_declarations.items[0].after_occurrence = 1 assert assess_draft_changes(req, resolved).status == "complete" @pytest.mark.parametrize("basis", [ {"type": "fact", "ref": "missing"}, {"type": "template", "ref": "unknown"}, {"type": "playbook_rule", "ref": "unknown"}, {"type": "retrieved_source", "ref": "unknown"}, {"type": "ai_suggestion", "ref": "fake"}, ]) def test_forged_basis_is_invalid_and_validate_returns_hard_error(basis): candidate = BASE.replace("书面送达", "电子邮件送达") out = output(candidate, declaration(basis=[basis])) assessment = assess_draft_changes(request(), out) assert assessment.status == "invalid" assert "change_declaration_invalid" in assessment.errors assert "draft_change_evidence_invalid" in [item.code for item in validate_draft_output(request(), out)] def test_uncovered_diff_is_incomplete_warning_not_hard_error(): candidate = BASE.replace("书面送达", "电子邮件送达") + "\n新增句。" out = output(candidate, declaration()) assessment = assess_draft_changes(request(), out) assert assessment.status == "incomplete" and assessment.uncovered_diff_ranges findings = validate_draft_output(request(), out) warning = next(item for item in findings if item.code == "draft_change_evidence_incomplete") assert warning.severity == "warning" assert "不代表语义或法律正确性" in warning.message def test_unknown_target_overlap_and_unrelated_insert_fail_closed(): candidate = BASE + "新增甲。新增乙。" bad = declaration(target_id="base-9999", operation="insert", before_excerpt=None, after_excerpt="新增甲", pending_confirmation=True) assert assess_draft_changes(request(), output(candidate, bad)).status == "invalid" unrelated = declaration(target_id="base-0001", operation="insert", before_excerpt=None, after_excerpt="新增甲", pending_confirmation=True) assert assess_draft_changes(request(), output(candidate, unrelated)).status == "invalid" def test_context_tamper_and_missing_declarations_are_invalid(): req = request() req.template_snapshot["change_evidence_context"]["targets"][0]["text"] = "tampered" assessment = assess_draft_changes(req, output(BASE)) assert assessment.status == "invalid" assert "change_evidence_context_invalid" in assessment.errors assert "change_declarations_missing" in assessment.errors def test_legacy_output_json_and_provider_schema_omit_new_contract(): legacy = request(enabled=False) out = output(BASE) dumped = out.model_dump(mode="json") assert "change_declarations" not in dumped assert "change_declarations" not in json.loads(out.model_dump_json()) schema = draft_output_schema() assert "change_declarations" not in schema["properties"] assert not any(name.startswith("DraftChange") for name in schema["$defs"]) assert draft_output_schema(legacy) == schema def test_policy_provider_schema_requires_declarations_without_changing_legacy_schema(): legacy = copy.deepcopy(draft_output_schema()) policy = draft_output_schema(request()) assert "change_declarations" in policy["properties"] assert "change_declarations" in policy["required"] assert policy["properties"]["change_declarations"].get("$ref") assert draft_output_schema() == legacy assert draft_digest(draft_output_schema()) == draft_digest(legacy) for definition in policy["$defs"].values(): if "properties" in definition: assert set(definition["required"]) == set(definition["properties"]) @pytest.mark.parametrize("base,candidate,expected", [ ("原文", "新增原文", [("after", 0, 2)]), ("原文", "原文新增", [("after", 2, 4)]), ("新增原文", "原文", [("before", 0, 2)]), ("原文新增", "原文", [("before", 2, 4)]), ]) def test_head_and_tail_insert_delete_ranges(base, candidate, expected): from yuqei_sdk.draft_change_evidence import _diff_ranges assert _diff_ranges(base, candidate) == expected def test_malformed_context_is_invalid_but_never_traversed(): req = request() req.template_snapshot["change_evidence_context"] = {"targets": [None, "secret"]} candidate = BASE.replace("书面送达", "电子邮件送达") assessment = assess_draft_changes(req, output(candidate, declaration())) assert assessment.status == "invalid" assert "change_evidence_context_invalid" in assessment.errors assert assessment.items[0].status == "verified" def test_partial_excerpt_does_not_cover_the_rest_of_one_diff_hunk(): candidate = BASE.replace("书面送达", "电子邮件及短信送达") evidence = declaration(after_excerpt="电子邮件") assessment = assess_draft_changes(request(), output(candidate, evidence)) assert assessment.status == "incomplete" assert any(item.side == "after" and item.end > item.start for item in assessment.uncovered_diff_ranges) def test_cross_paragraph_and_oversized_excerpts_are_rejected(): with pytest.raises(ValidationError): DraftChangeDeclarations.model_validate(declaration(before_excerpt="甲\n\n乙")) with pytest.raises(ValidationError): DraftChangeDeclarations.model_validate(declaration(before_excerpt="甲" * 2001)) @pytest.mark.parametrize("snapshot_patch,basis", [ ({"clause_assembly": "not-an-object"}, {"type": "template", "ref": "None"}), ({"id": "", "clause_assembly": []}, {"type": "template", "ref": "tpl-1"}), ]) def test_arbitrary_template_snapshot_shapes_and_empty_ids_fail_closed(snapshot_patch, basis): req = request() req.template_snapshot.update(snapshot_patch) candidate = BASE.replace("书面送达", "电子邮件送达") assessment = assess_draft_changes(req, output(candidate, declaration(basis=[basis]))) assert assessment.status == "invalid" assert "basis_invalid:template" in assessment.items[0].errors def test_missing_and_disabled_rule_ids_are_not_valid_bases(): req = request() req.playbook_snapshot = {"rules": [{"id": None}, {"id": "off", "enabled": False}]} candidate = BASE.replace("书面送达", "电子邮件送达") for ref in ("None", "off"): assessment = assess_draft_changes(req, output(candidate, declaration(basis=[{"type": "playbook_rule", "ref": ref}]))) assert assessment.status == "invalid" assert "basis_invalid:playbook_rule" in assessment.items[0].errors def test_ai_suggestion_cannot_be_silently_confirmed_and_pending_is_correlated(): candidate = BASE.replace("书面送达", "电子邮件送达") silent = assess_draft_changes(request(), output(candidate, declaration(pending_confirmation=False))) assert "ai_suggestion_requires_confirmation" in silent.items[0].errors unrelated = assess_draft_changes(request(), output(candidate, declaration(basis=[{"type": "template", "ref": "tpl-1"}], pending_confirmation=True))) assert "pending_confirmation_unjustified" in unrelated.items[0].errors def test_large_distant_edits_are_refined_into_their_own_targets(): base = "开头旧甲" + ("中" * 5000) + "结尾旧乙" candidate = "开头新甲" + ("中" * 5000) + "结尾新乙" req = request(base=base) changes = DraftChangeDeclarations.model_validate({ "schema_version": "draft-change-declarations.v1", "items": [ {"id": "change-a", "target_id": "base-0001", "operation": "replace", "before_excerpt": "旧甲", "after_excerpt": "新甲", "basis": [{"type": "ai_suggestion"}], "summary": "调整甲", "pending_confirmation": True}, {"id": "change-b", "target_id": "base-0003", "operation": "replace", "before_excerpt": "旧乙", "after_excerpt": "新乙", "basis": [{"type": "ai_suggestion"}], "summary": "调整乙", "pending_confirmation": True}, ], }) assessment = assess_draft_changes(req, output(candidate, changes)) assert assessment.status == "complete" assert [item.status for item in assessment.items] == ["verified", "verified"] def test_user_instruction_basis_requires_an_actual_request_instruction(): candidate = BASE.replace("书面送达", "电子邮件送达") evidence = declaration(basis=[{"type": "user_instruction"}]) absent = assess_draft_changes(request(), output(candidate, evidence)) assert "basis_invalid:user_instruction" in absent.items[0].errors req = request() req.drafting_instructions = "请调整通知方式。" assert assess_draft_changes(req, output(candidate, evidence)).status == "complete" def test_maximum_fragmented_context_ids_remain_valid_declaration_ids(): base = "x\n\n" * 33_333 context = build_change_evidence_context(base) assert len(base) == 99_999 assert len(context["targets"]) == 33_333 assert context["targets"][-1]["target_id"] == "base-33333" DraftChangeDeclarations.model_validate(declaration( target_id=context["targets"][-1]["target_id"])) def test_many_repeated_short_lines_use_bounded_token_matching(): from yuqei_sdk.draft_change_evidence import _diff_hunks base = "相同\n" * 25_000 candidate = base[:50_000] + "改" + base[50_001:] hunks = _diff_hunks(base, candidate) assert hunks assert all(0 <= left[0] <= left[1] <= len(base) for left, _ in hunks) def test_policy_disabled_returns_none_and_does_not_validate_declarations(): legacy = request(enabled=False) assert assess_draft_changes(legacy, output(BASE)) is None assert not any(item.code.startswith("draft_change_evidence") for item in validate_draft_output(legacy, output(BASE)))