import copy import json import pytest from yuqei_sdk.draft_semantic_review import DraftSemanticReviewRequest, draft_semantic_review_output_schema from yuqei_sdk.draft_semantic_wire import ( PROMPT_WIRE_V2_MARKER, WIRE_V1, WIRE_V2, decode_semantic_wire, semantic_wire_output_schema, semantic_wire_protocol, ) 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 = "甲方应付款。" candidate = "甲方应付款。" advisories = { "assumptions": [], "risk_notes": ["付款义务已保留。"], "change_explanations": [], "quality_findings": [{"severity": "warning", "code": "Q1", "message": "对象摘录须规范编码", "fields": []}], } return DraftSemanticReviewRequest.model_validate({ "schema_version": "draft-semantic-review-request.v1", "candidate_id": "candidate-1", "candidate_body_sha256": sha(candidate), "input_revision": 1, "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": [], "not_selected_clause_ids": [], "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 canonical(req): return { "schema_version": WIRE_V1, "status": "passed", "request_binding": { "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, }, "coverage": { "relation_categories": ["party", "duty", "trigger", "deadline", "written_form", "exception", "liability", "remedy"], "advisory_fields": ["assumptions", "risk_notes", "change_explanations", "quality_findings"], "target_checks": [{"target_id": "base-0001", "base_excerpt": req.base_document, "base_occurrence": 1, "candidate_excerpt": req.candidate_document, "candidate_occurrence": 1, "finding_ids": []}], "candidate_checks": [{"target_id": "candidate-0001", "candidate_excerpt": req.candidate_document, "candidate_occurrence": 1, "finding_ids": []}], "advisory_checks": [ {"field": "risk_notes", "index": 0, "exact_excerpt": "付款义务已保留。", "metadata_finding_ids": []}, {"field": "quality_findings", "index": 0, "exact_excerpt": '{"code":"Q1","fields":[],"message":"对象摘录须规范编码","severity":"warning"}', "metadata_finding_ids": []}, ], "statement": "已检查。", }, "findings": [], "repair_instructions": [], "metadata_findings": [], "limitations": ["不构成法律批准。"], } def wire(req): value = copy.deepcopy(canonical(req)) value["schema_version"] = WIRE_V2 for row in value["coverage"]["target_checks"]: row.pop("base_excerpt") for row in value["coverage"]["candidate_checks"]: row.pop("candidate_excerpt") for row in value["coverage"]["advisory_checks"]: row.pop("exact_excerpt") return value def test_protocol_is_snapshot_marker_only_and_never_falls_back(): assert semantic_wire_protocol("旧发布Prompt") == WIRE_V1 assert semantic_wire_protocol(PROMPT_WIRE_V2_MARKER + "\n规则") == WIRE_V2 for content in ( "规则\n" + PROMPT_WIRE_V2_MARKER, "[output-wire:unknown]", "[output-wire:", "[output-wire:]", PROMPT_WIRE_V2_MARKER + "\n" + PROMPT_WIRE_V2_MARKER, PROMPT_WIRE_V2_MARKER + "\n[output-wire:", PROMPT_WIRE_V2_MARKER + "\n[output-wire:]", PROMPT_WIRE_V2_MARKER + " trailing", ): with pytest.raises(ValueError): semantic_wire_protocol(content) with pytest.raises(ValueError): semantic_wire_protocol(None) def test_v1_schema_and_decode_are_unchanged(): req = request() assert semantic_wire_output_schema(req, WIRE_V1) == draft_semantic_review_output_schema(req) raw = json.dumps(canonical(req), ensure_ascii=False) assert decode_semantic_wire(req, raw, WIRE_V1).model_dump(mode="json") == canonical(req) def test_v2_schema_removes_only_three_mechanical_leaves_and_fixes_version(): schema = semantic_wire_output_schema(request(), WIRE_V2) assert schema["properties"]["schema_version"]["const"] == WIRE_V2 for definition, leaf in ( ("DraftSemanticTargetCheck", "base_excerpt"), ("DraftSemanticCandidateCheck", "candidate_excerpt"), ("DraftSemanticAdvisoryCheck", "exact_excerpt"), ): assert leaf not in schema["$defs"][definition]["properties"] assert leaf not in schema["$defs"][definition]["required"] assert "candidate_excerpt" in schema["$defs"]["DraftSemanticTargetCheck"]["properties"] def test_v2_schema_and_decoder_reject_case_changed_finding_aliases(): req = request() schema = semantic_wire_output_schema(req, WIRE_V2) identifier_pattern = schema["$defs"]["DraftSemanticFinding"]["properties"]["id"]["pattern"] assert schema["$defs"]["DraftSemanticTargetCheck"]["properties"]["finding_ids"]["items"]["pattern"] \ == identifier_pattern assert schema["$defs"]["DraftSemanticCandidateCheck"]["properties"]["finding_ids"]["items"]["pattern"] \ == identifier_pattern payload = wire(req) payload["findings"] = [{ "id": "find-001", "category": "duty", "disposition": "preserved", "severity": "info", "base_excerpt": req.base_document, "base_occurrence": 1, "candidate_excerpt": req.candidate_document, "candidate_occurrence": 1, "fact_fields": ["付款安排"], "clause_ids": [], "change_declaration_ids": [], "instruction_excerpt": "保留付款义务。", "explanation": "付款义务已保留。", "question_field": None, }] payload["coverage"]["target_checks"][0]["finding_ids"] = ["F-001"] payload["coverage"]["candidate_checks"][0]["finding_ids"] = ["F-001"] with pytest.raises(ValueError): decode_semantic_wire(req, json.dumps(payload, ensure_ascii=False), WIRE_V2) payload["coverage"]["target_checks"][0]["finding_ids"] = ["find-001"] payload["coverage"]["candidate_checks"][0]["finding_ids"] = ["find-001"] decoded = decode_semantic_wire(req, json.dumps(payload, ensure_ascii=False), WIRE_V2) assert decoded.coverage.target_checks[0].finding_ids == ["find-001"] assert decoded.coverage.candidate_checks[0].finding_ids == ["find-001"] def test_v2_hydrates_only_exact_frozen_leaf_values_and_preserves_judgment(): req = request() payload = wire(req) payload["coverage"]["statement"] = "模型自己的判断仍保留。" decoded = decode_semantic_wire(req, json.dumps(payload, ensure_ascii=False), WIRE_V2) assert decoded.schema_version == WIRE_V1 assert decoded.coverage.statement == "模型自己的判断仍保留。" assert decoded.coverage.target_checks[0].base_excerpt == req.base_document assert decoded.coverage.candidate_checks[0].candidate_excerpt == req.candidate_document assert decoded.coverage.advisory_checks[1].exact_excerpt == '{"code":"Q1","fields":[],"message":"对象摘录须规范编码","severity":"warning"}' @pytest.mark.parametrize("kind", ["missing", "duplicate", "unknown"]) def test_v2_rejects_incomplete_duplicate_or_unknown_target_references(kind): req = request() payload = wire(req) rows = payload["coverage"]["candidate_checks"] if kind == "missing": rows.clear() elif kind == "duplicate": rows.append(copy.deepcopy(rows[0])) else: rows[0]["target_id"] = "candidate-unknown" with pytest.raises(ValueError): decode_semantic_wire(req, json.dumps(payload, ensure_ascii=False), WIRE_V2) @pytest.mark.parametrize("mutation", ["wrong_version", "extra", "hydrated", "bad_occurrence"]) def test_v2_rejects_wrong_version_extra_fields_leaf_smuggling_and_bad_occurrence(mutation): req = request() payload = wire(req) if mutation == "wrong_version": payload["schema_version"] = WIRE_V1 elif mutation == "extra": payload["coverage"]["statement_extra"] = "x" elif mutation == "hydrated": payload["coverage"]["target_checks"][0]["base_excerpt"] = "伪造" else: payload["coverage"]["target_checks"][0]["base_occurrence"] = 0 with pytest.raises((ValueError, TypeError)): decode_semantic_wire(req, json.dumps(payload, ensure_ascii=False), WIRE_V2) def test_v2_rejects_missing_duplicate_unknown_and_out_of_range_advisory_refs(): req = request() for mutate in ( lambda rows: rows.pop(), lambda rows: rows.append(copy.deepcopy(rows[0])), lambda rows: rows[0].update(field="assumptions"), lambda rows: rows[0].update(index=99), ): payload = wire(req) mutate(payload["coverage"]["advisory_checks"]) with pytest.raises(ValueError): decode_semantic_wire(req, json.dumps(payload, ensure_ascii=False), WIRE_V2) @pytest.mark.parametrize("bad_index", [True, False, [], {}]) def test_v2_rejects_non_integer_advisory_indexes_without_type_errors(bad_index): req = request() payload = wire(req) payload["coverage"]["advisory_checks"][0]["index"] = bad_index with pytest.raises(ValueError): decode_semantic_wire(req, json.dumps(payload, ensure_ascii=False), WIRE_V2) @pytest.mark.parametrize("constant", ["NaN", "Infinity", "-Infinity"]) def test_v2_rejects_non_json_numeric_constants(constant): req = request() raw = json.dumps(wire(req), ensure_ascii=False).replace('"statement": "已检查。"', f'"statement": {constant}') with pytest.raises(ValueError): decode_semantic_wire(req, raw, WIRE_V2) def test_v2_preserves_repair_fields_without_server_generated_decisions(): req = request() payload = wire(req) payload["status"] = "repairable" payload["findings"] = [{ "id": "duty-gap", "category": "duty", "disposition": "omitted", "severity": "error", "base_excerpt": "甲方应付款", "base_occurrence": 1, "candidate_excerpt": "甲方应付款", "candidate_occurrence": 1, "fact_fields": ["付款安排"], "clause_ids": [], "change_declaration_ids": [], "instruction_excerpt": None, "explanation": "需修复", "question_field": None, }] payload["repair_instructions"] = [{ "finding_ids": ["duty-gap"], "metadata_finding_ids": [], "action": "保留付款义务", "constraints": ["不得新增事实"], }] decoded = decode_semantic_wire(req, json.dumps(payload, ensure_ascii=False), WIRE_V2) assert decoded.status == "repairable" assert decoded.findings[0].id == "duty-gap" assert decoded.repair_instructions[0].finding_ids == ["duty-gap"]