from yuqei_sdk.contract_drafting import ( ContractDraftOutput, ContractDraftRequest, DraftFact, DraftFactBinding, DraftInputConflictAssessment, DraftMissingField, DraftQualityFinding, assess_draft_input_conflicts, draft_digest, draft_output_schema, validate_draft_output, ) def request(policy=True): return ContractDraftRequest(contract_title="测试", contract_type="服务", base_document="正文", facts=[DraftFact(name="控制", value="不接触", role="control"), DraftFact(name="范围", value="接触合成文件")], template_snapshot={"input_conflict_policy": "draft-input-conflict.v1"} if policy else {}) def output(*, evidence=True, status="incomplete", gap=False): findings = [{"code": "FACT_CONTROL_CONFLICT", "severity": "error", "message": "输入冲突", "fields": ["控制", "范围"], "fact_evidence": ([{"field": "控制", "value": "不接触"}, {"field": "范围", "value": "接触合成文件"}] if evidence else [])}] if gap: findings.append({"code": "TEMPLATE_FIELD_GAP", "severity": "warning", "message": "gap", "fields": ["控制"]}) return ContractDraftOutput(schema_version="contract-draft.v1", status=status, draft_document="正文 接触合成文件", missing_fields=[{"field": "控制", "question": "请确认", "stage": "before_signature"}], assumptions=[], risk_notes=[], fact_bindings=[{"field": "范围", "value": "接触合成文件", "document_excerpt": "接触合成文件"}], source_bindings=[], change_explanations=[], quality_findings=findings) def codes(req, out): return {item.code for item in validate_draft_output(req, out)} def test_valid_structured_conflict_covers_only_its_missing_target(): assessment = assess_draft_input_conflicts(request(), output()) assert assessment == DraftInputConflictAssessment(policy_enabled=True, valid_finding_indexes=[0], covered_missing_fields=["控制"]) assert "draft_unmodeled_blocking_fact" not in codes(request(), output()) def test_legacy_policy_does_not_reclassify_same_output(): assert not assess_draft_input_conflicts(request(False), output()).policy_enabled assert "draft_unmodeled_blocking_fact" in codes(request(False), output()) def test_missing_or_changed_evidence_and_candidate_ready_are_invalid(): assert "draft_input_conflict_evidence_invalid" in codes(request(), output(evidence=False)) changed = output(); changed.quality_findings[0].fact_evidence[0].value = "别的值" assert "draft_input_conflict_evidence_invalid" in codes(request(), changed) assert "draft_input_conflict_evidence_invalid" in codes(request(), output(status="candidate_ready")) def test_template_gap_cannot_mask_a_provided_fact_under_new_policy(): assert "draft_template_gap_masks_provided_fact" in codes(request(), output(gap=True)) def test_evidence_is_forbidden_on_other_finding_codes_and_empty_legacy_serialization_is_stable(): out = output(); out.quality_findings[0].code = "OTHER" assert assess_draft_input_conflicts(request(), out).invalid_finding_indexes == [0] finding = out.quality_findings[0].model_copy(update={"fact_evidence": []}) assert "fact_evidence" not in finding.model_dump(mode="json") def test_legacy_finding_serialization_and_digest_remain_unchanged_when_evidence_is_empty(): legacy = {"code": "OLD", "severity": "warning", "message": "旧输出", "fields": ["范围"]} finding = DraftQualityFinding.model_validate(legacy) assert finding.model_dump(mode="json") == legacy assert draft_digest(finding.model_dump(mode="json")) == draft_digest(legacy) out = output() out.quality_findings = [finding] assert "fact_evidence" not in out.model_dump(mode="json")["quality_findings"][0] def test_provider_schema_requires_evidence_only_for_opted_in_conflict_policy(): runtime_finding = ContractDraftOutput.model_json_schema()["$defs"]["DraftQualityFinding"] legacy = draft_output_schema() provider_finding = draft_output_schema(request())["$defs"]["DraftQualityFinding"] assert "fact_evidence" in runtime_finding["properties"] assert "fact_evidence" not in runtime_finding["required"] assert "fact_evidence" in provider_finding["required"] assert "fact_evidence" not in legacy["$defs"]["DraftQualityFinding"]["properties"] assert "DraftConflictFactEvidence" not in legacy["$defs"] assert draft_digest(legacy) == "0d59e4969533b5dfe961df6005f80cea56d419c837f77022e7b2c965fd953e2d" assert draft_digest(draft_output_schema(request())) == "2bdf2d71cd5d73c1c513902d1f85a59cce168cb8a835ebfdf2d1986998d18b5e" def test_conflict_evidence_remains_strict_when_change_tracking_is_also_enabled(): req = request() req.template_snapshot["change_evidence_policy"] = "draft-change-evidence.v1" finding = draft_output_schema(req)["$defs"]["DraftQualityFinding"] assert "fact_evidence" in finding["properties"] assert "fact_evidence" in finding["required"] def test_instruction_and_evidence_roles_cannot_certify_input_conflicts(): req = request() req.facts[1].role = "instruction" assert assess_draft_input_conflicts(req, output()).invalid_finding_indexes == [0] req.facts[1].role = "evidence" assert assess_draft_input_conflicts(req, output()).invalid_finding_indexes == [0] def test_duplicate_fields_or_evidence_cannot_certify_input_conflicts(): duplicate_fields = output() duplicate_fields.quality_findings[0].fields = ["控制", "范围", "范围"] assert assess_draft_input_conflicts(request(), duplicate_fields).invalid_finding_indexes == [0] duplicate_evidence = output() duplicate_evidence.quality_findings[0].fact_evidence.append( duplicate_evidence.quality_findings[0].fact_evidence[-1].model_copy()) assert assess_draft_input_conflicts(request(), duplicate_evidence).invalid_finding_indexes == [0] def test_valid_conflict_does_not_explain_an_unrelated_missing_field(): out = output() out.missing_fields.append(DraftMissingField(field="任意逃逸字段", question="请补充", stage="before_signature")) findings = validate_draft_output(request(), out) unexplained = next(item for item in findings if item.code == "draft_unmodeled_blocking_fact") assert unexplained.fields == ["任意逃逸字段"] assert unexplained.message == "新增阻断缺项既没有 TEMPLATE_FIELD_GAP 采集缺口说明，也没有有效冲突补问关联。" def test_legacy_unmodeled_blocking_fact_message_is_unchanged(): out = output() out.missing_fields.append(DraftMissingField(field="任意逃逸字段", question="请补充", stage="before_signature")) findings = validate_draft_output(request(False), out) unexplained = next(item for item in findings if item.code == "draft_unmodeled_blocking_fact") assert unexplained.message == "模型新增了模板未声明的阻断缺项，但没有以 TEMPLATE_FIELD_GAP 说明模板采集缺口。" def test_template_gap_can_still_cover_an_unprovided_optional_field(): out = output() out.missing_fields = [DraftMissingField(field="可选但未填字段", question="请确认是否需要", stage="before_signature")] out.quality_findings = [DraftQualityFinding(code="TEMPLATE_FIELD_GAP", severity="warning", message="模板未采集该可选字段。", fields=["可选但未填字段"])] assert "draft_unmodeled_blocking_fact" not in codes(request(), out) def test_legacy_runtime_shape_parses_and_serializes_without_new_member(): legacy = {"code": "OLD", "severity": "warning", "message": "旧结构", "fields": []} assert DraftQualityFinding.model_validate(legacy).model_dump(mode="json") == legacy runtime = ContractDraftOutput.model_json_schema()["$defs"]["DraftQualityFinding"] assert set(runtime["required"]) == set(legacy) def test_composite_question_target_is_distinct_from_valid_fact_evidence(): out = output() out.missing_fields[0] = DraftMissingField( field="控制、范围和一致口径", question="请统一确认", stage="before_signature") assessment = assess_draft_input_conflicts(request(), out) assert assessment.invalid_finding_indexes == [0] assert assessment.invalid_evidence_finding_indexes == [] assert assessment.invalid_question_target_finding_indexes == [0] result = codes(request(), out) assert "draft_input_conflict_question_target_invalid" in result assert "draft_input_conflict_evidence_invalid" not in result def test_unrelated_real_field_is_still_an_invalid_question_target(): req = request() req.facts.append(DraftFact(name="无关事实", value="无关值", role="body_fact")) out = output() out.draft_document += " 无关值" out.fact_bindings.append(DraftFactBinding(field="无关事实", value="无关值", document_excerpt="无关值")) out.missing_fields[0] = DraftMissingField(field="无关事实", question="请确认", stage="before_signature") assessment = assess_draft_input_conflicts(req, out) assert assessment.invalid_question_target_finding_indexes == [0] assert assessment.covered_missing_fields == [] def test_exact_conflict_field_remains_a_valid_question_target(): assessment = assess_draft_input_conflicts(request(), output()) assert assessment.valid_finding_indexes == [0] assert assessment.invalid_finding_indexes == [] assert assessment.invalid_evidence_finding_indexes == [] assert assessment.invalid_question_target_finding_indexes == [] def test_legacy_policy_has_no_new_diagnostics_for_composite_target(): out = output() out.missing_fields[0] = DraftMissingField(field="控制和范围", question="请确认", stage="before_signature") assessment = assess_draft_input_conflicts(request(False), out) assert not assessment.policy_enabled assert assessment.invalid_finding_indexes == [] assert assessment.invalid_evidence_finding_indexes == [] assert assessment.invalid_question_target_finding_indexes == []