import copy import json import pytest from pydantic import ValidationError from yuqei_sdk.draft_semantic_review import ( ADVISORY_FIELDS, RELATION_CATEGORIES, DraftSemanticReviewOutput, DraftSemanticReviewRequest, assess_draft_semantic_review, draft_semantic_review_output_schema, ) def sha(value): import hashlib text = value if isinstance(value, str) else json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False) return hashlib.sha256(text.encode()).hexdigest() def request(): base = "甲方应在十日内书面通知。违约方赔偿实际损失或侵权所得。" advisories = {"assumptions": [], "risk_notes": ["正文仅约定赔偿实际损失。"], "change_explanations": ["保留责任条款。"], "quality_findings": []} candidate = "甲方应在十日内书面通知。违约方赔偿实际损失或侵权所得。" return DraftSemanticReviewRequest.model_validate({"schema_version": "draft-semantic-review-request.v1", "candidate_id": "candidate-1", "candidate_body_sha256": sha(candidate), "input_revision": 2, "base_document": base, "candidate_document": candidate, "draft_request_sha256": "1" * 64, "facts": [{"field": "通知期限", "role": "body_fact", "value": "十日"}], "review_targets": [{"id": "base-0001", "text": base, "start": 0, "end": len(base)}], "candidate_review_targets": [{"id": "candidate-0001", "text": candidate, "start": 0, "end": len(candidate)}], "selected_clauses": [{"id": "notice", "title": "通知", "source_excerpt": "甲方应在十日内书面通知。"}], "not_selected_clause_ids": ["optional"], "pending_clause_ids": [], "conflict_clause_ids": [], "drafting_instructions": "不得缩短通知期限。", "change_declarations": {"schema_version": "draft-change-declarations.v1", "items": []}, "verified_change_declaration_ids": [], "change_assessment_sha256": "2" * 64, "candidate_advisories": advisories, "candidate_advisories_sha256": sha(advisories), "template": {"id": "tpl", "content_sha256": "3" * 64, "input_schema_sha256": "4" * 64, "input_fields": ["通知期限"]}, "playbook_sha256": "5" * 64, "prompt_execution_identity_sha256": "6" * 64}) def binding(req): return {"candidate_id": req.candidate_id, "candidate_body_sha256": req.candidate_body_sha256, "input_revision": req.input_revision, "draft_request_sha256": req.draft_request_sha256, "change_assessment_sha256": req.change_assessment_sha256, "candidate_advisories_sha256": req.candidate_advisories_sha256, "template_id": req.template.id, "template_content_sha256": req.template.content_sha256, "input_schema_sha256": req.template.input_schema_sha256, "playbook_sha256": req.playbook_sha256, "prompt_execution_identity_sha256": req.prompt_execution_identity_sha256} def output(req, **updates): value = {"schema_version": "draft-semantic-review.v1", "status": "passed", "request_binding": binding(req), "coverage": {"relation_categories": list(RELATION_CATEGORIES), "advisory_fields": list(ADVISORY_FIELDS), "target_checks": [{"target_id": "base-0001", "base_excerpt": req.base_document, "base_occurrence": None, "candidate_excerpt": req.base_document, "candidate_occurrence": None, "finding_ids": []}], "candidate_checks": [{"target_id": "candidate-0001", "candidate_excerpt": req.candidate_document, "candidate_occurrence": None, "finding_ids": []}], "advisory_checks": [ {"field": "risk_notes", "index": 0, "exact_excerpt": req.candidate_advisories.risk_notes[0], "metadata_finding_ids": []}, {"field": "change_explanations", "index": 0, "exact_excerpt": req.candidate_advisories.change_explanations[0], "metadata_finding_ids": []}], "statement": "逐类检查正文关系与候选说明。"}, "findings": [], "repair_instructions": [], "metadata_findings": [], "limitations": ["模型复核不是法律意见。"]} value.update(updates) return DraftSemanticReviewOutput.model_validate(value) def finding(**updates): value = {"id": "finding-1", "category": "deadline", "disposition": "preserved", "severity": "info", "base_excerpt": "十日内", "base_occurrence": None, "candidate_excerpt": "十日内", "candidate_occurrence": None, "fact_fields": ["通知期限"], "clause_ids": ["notice"], "change_declaration_ids": [], "instruction_excerpt": "不得缩短通知期限", "explanation": "期限被保留。", "question_field": None} value.update(updates) return value def test_passed_requires_complete_inspectable_coverage_even_with_no_findings(): req = request() assert assess_draft_semantic_review(req, output(req)).status == "valid" out = output(req) out.coverage.relation_categories.pop() assessment = assess_draft_semantic_review(req, out) assert assessment.status == "invalid" and "relation_coverage_incomplete" in assessment.errors assert assessment.semantic_correctness_verified is False assert assessment.legal_correctness_verified is False assert assessment.authorizes_adoption is False @pytest.mark.parametrize("field", ["candidate_body_sha256", "draft_request_sha256", "candidate_advisories_sha256"]) def test_forged_request_binding_is_invalid(field): req = request(); out = output(req) setattr(out.request_binding, field, "f" * 64) assert "request_binding_mismatch" in assess_draft_semantic_review(req, out).errors def test_request_recomputes_candidate_and_advisory_hashes(): payload = request().model_dump(mode="json") payload["candidate_body_sha256"] = "0" * 64 with pytest.raises(ValidationError): DraftSemanticReviewRequest.model_validate(payload) payload = request().model_dump(mode="json") payload["candidate_advisories"]["risk_notes"][0] = "篡改" with pytest.raises(ValidationError): DraftSemanticReviewRequest.model_validate(payload) @pytest.mark.parametrize("updates,error", [ ({"fact_fields": ["未知"]}, "fact_reference_invalid"), ({"clause_ids": ["optional"]}, "clause_reference_invalid"), ({"instruction_excerpt": "不存在"}, "instruction_excerpt_not_found"), ]) def test_unknown_refs_and_instruction_are_invalid(updates, error): req = request(); assessment = assess_draft_semantic_review(req, output(req, findings=[finding(**updates)])) assert error in assessment.errors def test_empty_instruction_cannot_be_used_as_evidence(): req = request(); req.drafting_instructions = "" assessment = assess_draft_semantic_review(req, output(req, findings=[finding()])) assert "instruction_excerpt_not_found" in assessment.errors def test_repeated_excerpt_requires_occurrence_and_then_locates_half_open_range(): req = request(); req.base_document = "十日内；十日内。" req.review_targets = [{"id": "base-0001", "text": req.base_document, "start": 0, "end": len(req.base_document)}] req = DraftSemanticReviewRequest.model_validate(req.model_dump(mode="json")) ambiguous = assess_draft_semantic_review(req, output(req, findings=[finding()])) assert "base_excerpt_ambiguous" in ambiguous.errors resolved = assess_draft_semantic_review(req, output(req, findings=[finding(base_occurrence=2)])) assert resolved.findings[0].errors == [] and resolved.findings[0].base_range.start == 4 def test_metadata_index_exact_excerpt_and_candidate_location_are_bound(): req = request() item = {"id": "metadata-1", "field": "risk_notes", "index": 0, "exact_excerpt": "错误概括", "candidate_excerpt": "实际损失或侵权所得", "candidate_occurrence": None, "disposition": "inaccurate", "explanation": "遗漏替代救济。"} assessment = assess_draft_semantic_review(req, output(req, status="blocked", findings=[finding(severity="error", disposition="unsupported_change")], metadata_findings=[item])) assert "metadata_exact_excerpt_mismatch" in assessment.errors @pytest.mark.parametrize("status,findings,repairs,error", [ ("passed", [finding(disposition="reversed", severity="error")], [], "passed_status_inconsistent"), ("needs_information", [], [], "needs_information_without_question"), ("repairable", [finding(disposition="omitted", severity="error")], [], "repairable_without_complete_instructions"), ("blocked", [finding()], [], "blocked_without_error_finding"), ]) def test_provider_cannot_self_sign_an_inconsistent_status(status, findings, repairs, error): req = request() assert error in assess_draft_semantic_review(req, output(req, status=status, findings=findings, repair_instructions=repairs)).errors def test_needs_information_question_must_reference_template_field(): req = request() invalid = output(req, status="needs_information", findings=[finding( disposition="ambiguous", severity="warning", question_field="未知字段")]) assessment = assess_draft_semantic_review(req, invalid) assert "question_field_invalid" in assessment.errors def test_provider_schema_is_strict_required_nullable_and_runtime_rejects_invalid_shape(): schema = draft_semantic_review_output_schema() for definition in schema["$defs"].values(): if "properties" in definition: assert set(definition["required"]) == set(definition["properties"]) assert definition["additionalProperties"] is False finding_schema = schema["$defs"]["DraftSemanticFinding"] assert any(item.get("type") == "null" for item in finding_schema["properties"]["base_excerpt"]["anyOf"]) with pytest.raises(ValidationError): DraftSemanticReviewOutput.model_validate({"schema_version": "draft-semantic-review.v1", "status": "passed"}) def test_finding_reference_runtime_rejects_case_changed_aliases_without_schema_drift(): schema = draft_semantic_review_output_schema() for definition, field in ( ("DraftSemanticTargetCheck", "finding_ids"), ("DraftSemanticCandidateCheck", "finding_ids"), ("DraftSemanticRepairInstruction", "finding_ids"), ("DraftSemanticAdvisoryCheck", "metadata_finding_ids"), ("DraftSemanticRepairInstruction", "metadata_finding_ids"), ): assert schema["$defs"][definition]["properties"][field]["items"] == { "type": "string" } req = request() payload = output(req, findings=[finding(id="find-001")]).model_dump(mode="json") payload["coverage"]["target_checks"][0]["finding_ids"] = ["F-001"] payload["coverage"]["candidate_checks"][0]["finding_ids"] = ["F-001"] with pytest.raises(ValidationError): DraftSemanticReviewOutput.model_validate(payload) metadata_payload = output(request()).model_dump(mode="json") metadata_payload["coverage"]["advisory_checks"][0]["metadata_finding_ids"] = ["M-001"] with pytest.raises(ValidationError): DraftSemanticReviewOutput.model_validate(metadata_payload) def test_lowercase_but_unknown_finding_references_remain_fail_closed(): req = request() out = output(req, findings=[finding(id="find-001")]) out.coverage.target_checks[0].finding_ids = ["find-002"] out.coverage.candidate_checks[0].finding_ids = ["find-002"] assessment = assess_draft_semantic_review(req, out) assert assessment.status == "invalid" assert "target_check_finding_invalid" in assessment.errors assert "candidate_check_finding_invalid" in assessment.errors def test_exact_case_sensitive_finding_references_remain_valid(): req = request() out = output(req, findings=[finding(id="find-001")]) out.coverage.target_checks[0].finding_ids = ["find-001"] out.coverage.candidate_checks[0].finding_ids = ["find-001"] assessment = assess_draft_semantic_review(req, out) assert assessment.status == "valid" assert "target_check_finding_invalid" not in assessment.errors assert "candidate_check_finding_invalid" not in assessment.errors def test_passed_cannot_omit_or_forge_target_and_advisory_item_coverage(): req = request(); out = output(req) out.coverage.target_checks = [] assert "semantic_review_dto_invalid" in assess_draft_semantic_review(req, out).errors out = output(req) out.coverage.advisory_checks[0].exact_excerpt = "伪造说明" assert "advisory_check_excerpt_invalid" in assess_draft_semantic_review(req, out).errors def test_request_targets_must_exactly_cover_frozen_base(): payload = request().model_dump(mode="json") payload["review_targets"][0]["end"] -= 1 with pytest.raises(ValidationError): DraftSemanticReviewRequest.model_validate(payload) def test_empty_excerpt_and_duplicate_metadata_location_are_rejected(): with pytest.raises(ValidationError): DraftSemanticReviewOutput.model_validate(output(request()).model_dump(mode="json") | { "findings": [finding(base_excerpt="")]}) req = request() item = {"id": "metadata-1", "field": "risk_notes", "index": 0, "exact_excerpt": req.candidate_advisories.risk_notes[0], "candidate_excerpt": "实际损失", "candidate_occurrence": None, "disposition": "inaccurate", "explanation": "概括范围不完整。"} assessment = assess_draft_semantic_review(req, output(req, status="blocked", metadata_findings=[item, copy.deepcopy(item)])) assert "metadata_finding_duplicate" in assessment.errors def test_passed_rejects_ambiguous_without_question_and_repairs_reject_unknown_ids(): req = request() ambiguous = output(req, findings=[finding(disposition="ambiguous", severity="warning")]) assert "passed_status_inconsistent" in assess_draft_semantic_review(req, ambiguous).errors repair = {"finding_ids": ["unknown"], "metadata_finding_ids": [], "action": "修正", "constraints": ["不新增事实"]} blocked = output(req, status="blocked", findings=[finding(severity="error")], repair_instructions=[repair]) assert "repair_instruction_reference_invalid" in assess_draft_semantic_review(req, blocked).errors def test_blank_coverage_excerpt_and_zero_length_range_are_rejected(): req = request(); payload = output(req).model_dump(mode="json") payload["coverage"]["target_checks"][0]["candidate_excerpt"] = " " with pytest.raises(ValidationError): DraftSemanticReviewOutput.model_validate(payload) from yuqei_sdk.draft_semantic_review import DraftSemanticRange with pytest.raises(ValidationError): DraftSemanticRange(start=2, end=2) def test_candidate_side_addition_cannot_be_omitted_from_coverage(): req = request(); out = output(req) out.coverage.candidate_checks = [] assessment = assess_draft_semantic_review(req, out) assert "semantic_review_dto_invalid" in assessment.errors def test_supported_change_requires_verified_grounded_declaration_and_can_pass(): req = request() addition = "并承担补救责任。" req.candidate_document += addition req.candidate_body_sha256 = sha(req.candidate_document) req.candidate_review_targets[0].text = req.candidate_document req.candidate_review_targets[0].end = len(req.candidate_document) declaration = {"id": "change-1", "target_id": "base-0001", "operation": "insert", "before_excerpt": None, "after_excerpt": addition, "before_occurrence": None, "after_occurrence": None, "preceding_anchor": None, "following_anchor": None, "basis": [{"type": "user_instruction", "ref": None}], "summary": "按指令补充", "pending_confirmation": False} from yuqei_sdk.draft_change_evidence import DraftChangeDeclaration req.change_declarations.items = [DraftChangeDeclaration.model_validate(declaration)] req.verified_change_declaration_ids = ["change-1"] req = DraftSemanticReviewRequest.model_validate(req.model_dump(mode="json")) supported = finding(disposition="supported_change", candidate_excerpt=addition, change_declaration_ids=["change-1"]) assert assess_draft_semantic_review(req, output(req, findings=[supported])).status == "valid" req.drafting_instructions = "" assert "supported_change_ungrounded" in assess_draft_semantic_review( req, output(req, findings=[supported])).errors def test_metadata_only_repair_and_cross_advisory_reference_rules(): req = request() metadata = {"id": "meta-risk", "field": "risk_notes", "index": 0, "exact_excerpt": req.candidate_advisories.risk_notes[0], "candidate_excerpt": "实际损失", "candidate_occurrence": None, "disposition": "inaccurate", "explanation": "概括不完整。"} repair = {"finding_ids": [], "metadata_finding_ids": ["meta-risk"], "action": "修正风险说明", "constraints": ["不得改变正文"]} out = output(req, status="repairable", metadata_findings=[metadata], repair_instructions=[repair]) out.coverage.advisory_checks[0].metadata_finding_ids = ["meta-risk"] assert assess_draft_semantic_review(req, out).status == "valid" crossed = copy.deepcopy(out) crossed.coverage.advisory_checks[0].metadata_finding_ids = [] crossed.coverage.advisory_checks[1].metadata_finding_ids = ["meta-risk"] assert "advisory_check_finding_unrelated" in assess_draft_semantic_review(req, crossed).errors def test_explicit_grounded_whole_target_deletion_can_pass_without_candidate_excerpt(): req = request() first, second = "删除此句。", "保留此句。" req.base_document, req.candidate_document = first + second, second req.candidate_body_sha256 = sha(second) req.review_targets = [ {"id": "base-0001", "text": first, "start": 0, "end": len(first)}, {"id": "base-0002", "text": second, "start": len(first), "end": len(first + second)}, ] req.candidate_review_targets = [ {"id": "candidate-0001", "text": second, "start": 0, "end": len(second)}] from yuqei_sdk.draft_change_evidence import DraftChangeDeclaration req.change_declarations.items = [DraftChangeDeclaration.model_validate({ "id": "delete-1", "target_id": "base-0001", "operation": "delete", "before_excerpt": first, "basis": [{"type": "user_instruction"}], "summary": "按用户明确要求删除", "pending_confirmation": False})] req.verified_change_declaration_ids = ["delete-1"] req = DraftSemanticReviewRequest.model_validate(req.model_dump(mode="json")) review = output(req, findings=[finding(id="delete-finding", disposition="supported_change", category="duty", base_excerpt=first, candidate_excerpt=None, fact_fields=[], clause_ids=[], change_declaration_ids=["delete-1"], explanation="用户明确要求删除。")]) review.coverage.target_checks = [ {"target_id": "base-0001", "base_excerpt": first, "base_occurrence": None, "candidate_excerpt": None, "candidate_occurrence": None, "finding_ids": ["delete-finding"]}, {"target_id": "base-0002", "base_excerpt": second, "base_occurrence": None, "candidate_excerpt": second, "candidate_occurrence": None, "finding_ids": []}, ] review.coverage.candidate_checks = [{"target_id": "candidate-0001", "candidate_excerpt": second, "candidate_occurrence": None, "finding_ids": []}] review = DraftSemanticReviewOutput.model_validate(review.model_dump(mode="json")) assert assess_draft_semantic_review(req, review).status == "valid" def test_unknown_or_pending_declaration_reference_is_stable_invalid(): req = request() unknown = output(req, findings=[finding(disposition="supported_change", change_declaration_ids=["unknown"])]) assessment = assess_draft_semantic_review(req, unknown) assert assessment.status == "invalid" assert "change_declaration_reference_invalid" in assessment.errors from yuqei_sdk.draft_change_evidence import DraftChangeDeclaration req.change_declarations.items = [DraftChangeDeclaration.model_validate({ "id": "pending-1", "target_id": "base-0001", "operation": "insert", "after_excerpt": "侵权所得", "basis": [{"type": "user_instruction"}], "summary": "待确认", "pending_confirmation": True})] req.verified_change_declaration_ids = ["pending-1"] req = DraftSemanticReviewRequest.model_validate(req.model_dump(mode="json")) pending = output(req, findings=[finding(disposition="supported_change", change_declaration_ids=["pending-1"])]) assert "supported_change_ungrounded" in assess_draft_semantic_review(req, pending).errors def test_actual_candidate_diff_needs_related_finding_but_synonymous_rewrite_can_be_preserved(): req = request() req.candidate_document += "新增内容。" req.candidate_body_sha256 = sha(req.candidate_document) req.candidate_review_targets[0].text = req.candidate_document req.candidate_review_targets[0].end = len(req.candidate_document) req = DraftSemanticReviewRequest.model_validate(req.model_dump(mode="json")) assessment = assess_draft_semantic_review(req, output(req)) assert "candidate_diff_after_unreviewed" in assessment.errors req = request() req.base_document, req.candidate_document = "甲方应书面通知。", "甲方须书面通知。" req.candidate_body_sha256 = sha(req.candidate_document) req.review_targets = [{"id": "base-0001", "text": req.base_document, "start": 0, "end": len(req.base_document)}] req.candidate_review_targets = [{"id": "candidate-0001", "text": req.candidate_document, "start": 0, "end": len(req.candidate_document)}] req = DraftSemanticReviewRequest.model_validate(req.model_dump(mode="json")) preserved = finding(disposition="preserved", base_excerpt="应", candidate_excerpt="须", fact_fields=[], clause_ids=[], instruction_excerpt=None) review = output(req, findings=[preserved]) review.coverage.target_checks[0].candidate_excerpt = req.candidate_document assert assess_draft_semantic_review(req, review).status == "valid" def test_one_character_finding_does_not_cover_long_replace_but_two_fragments_can_union(): req = request() req.base_document, req.candidate_document = "甲乙丙丁", "甲XY丁" req.candidate_body_sha256 = sha(req.candidate_document) req.review_targets = [{"id": "base-0001", "text": req.base_document, "start": 0, "end": len(req.base_document)}] req.candidate_review_targets = [{"id": "candidate-0001", "text": req.candidate_document, "start": 0, "end": len(req.candidate_document)}] req = DraftSemanticReviewRequest.model_validate(req.model_dump(mode="json")) first = finding(id="first", disposition="preserved", base_excerpt="乙", candidate_excerpt="X", fact_fields=[], clause_ids=[], instruction_excerpt=None) review = output(req, findings=[first]) review.coverage.target_checks[0].candidate_excerpt = req.candidate_document assessment = assess_draft_semantic_review(req, review) assert "candidate_diff_before_unreviewed" in assessment.errors assert "candidate_diff_after_unreviewed" in assessment.errors second = finding(id="second", disposition="preserved", base_excerpt="丙", candidate_excerpt="Y", fact_fields=[], clause_ids=[], instruction_excerpt=None) review = output(req, findings=[first, second]) review.coverage.target_checks[0].candidate_excerpt = req.candidate_document assert assess_draft_semantic_review(req, review).status == "valid"