diff --git a/tests/test_contract_residuals.py b/tests/test_contract_residuals.py new file mode 100644 index 00000000..948e8ea4 --- /dev/null +++ b/tests/test_contract_residuals.py @@ -0,0 +1,530 @@ +from __future__ import annotations + +import ast +import dataclasses +from pathlib import Path + +import pytest + +from generate.contract_residuals import ( + ContractResidual, + ResidualKind, + ResidualSourceAxis, + project_contract_residuals, +) +from generate.kernel_facts import SourceSpan +from generate.problem_frame_contracts import ContractAssessment + + +def _assessment( + *, + candidate_organ: str = "unary_delta_transition", + missing_bindings: tuple[str, ...] = (), + unresolved_hazards: tuple[str, ...] = (), + runnable: bool = False, + explanation: str = "diagnostic candidate is not runnable", + evidence_spans: tuple[SourceSpan, ...] = (), +) -> ContractAssessment: + return ContractAssessment( + candidate_organ=candidate_organ, + missing_bindings=missing_bindings, + unresolved_hazards=unresolved_hazards, + runnable=runnable, + explanation=explanation, + evidence_spans=evidence_spans, + ) + + +def _entries( + kind: ResidualKind, + axis: ResidualSourceAxis, + *labels: str, +) -> tuple[tuple[str, ResidualKind, ResidualSourceAxis], ...]: + return tuple((label, kind, axis) for label in labels) + + +EXPECTED_LABEL_MAP: tuple[ + tuple[str, ResidualKind, ResidualSourceAxis], + ..., +] = ( + *_entries( + ResidualKind.MISSING_PROPOSAL, + ResidualSourceAxis.PROPOSAL, + "quantity_entity_proposal_required", + "unary_delta_proposal_required", + ), + *_entries( + ResidualKind.MISSING_RELATION, + ResidualSourceAxis.RELATION, + "local_binding_relation_unbound", + ), + *_entries( + ResidualKind.MISSING_ROLE, + ResidualSourceAxis.ROLE, + "quantity_unbound", + "entity_unbound", + "quantity_kind_unresolved", + "base_quantity_unbound", + "scale_unbound", + "state_entity_unbound", + "grounded_partition_subgroup", + "grounded_whole_entity", + "original_whole_unbound", + "action_cue_unbound", + "delta_quantity_unbound", + "changed_object_unbound", + "direction_unbound", + ), + *_entries( + ResidualKind.AMBIGUOUS_RELATION, + ResidualSourceAxis.RELATION, + "local_binding_relation_ambiguous", + "decrease_relation_ambiguous", + "unary_delta_relation_ambiguous", + ), + *_entries( + ResidualKind.AMBIGUOUS_ROLE, + ResidualSourceAxis.ROLE, + "quantity_ambiguous", + "entity_ambiguous", + "multiple_original_whole_candidates", + "delta_quantity_ambiguous", + "changed_object_ambiguous", + ), + *_entries( + ResidualKind.INEXACT_PROVENANCE, + ResidualSourceAxis.PROVENANCE, + "provenance_span_inexact", + "base_quantity_provenance_missing", + "scale_provenance_missing", + ), + *_entries( + ResidualKind.NONLOCAL_BINDING, + ResidualSourceAxis.LOCALITY, + "quantity_entity_nonlocal", + "pronoun_antecedent_unresolved", + ), + *_entries( + ResidualKind.UNSUPPORTED_TOPOLOGY, + ResidualSourceAxis.TOPOLOGY, + "percent_subgroup_links_incomplete", + "inverse_topology_unlicensed", + "event_assertion_unlicensed", + "passive_voice_unsupported", + "multiple_actor_surface", + ), + *_entries( + ResidualKind.UNIT_OBJECT_CONFLICT, + ResidualSourceAxis.UNIT_OBJECT, + "unit_kind_conflict", + "unit_continuity_unproven", + "unit_object_conflict", + ), + *_entries( + ResidualKind.HAZARD_BLOCKED, + ResidualSourceAxis.HAZARD, + "competing_family_context", + "percent_change_vs_percent_of", + "state_entity_continuity_unproven", + "scale_out_of_range", + "partition_subgroups_not_distinct", + ), + *_entries( + ResidualKind.TARGET_UNBOUND, + ResidualSourceAxis.TARGET, + "delta_decrease_target_unbound", + "delta_decrease_target_required", + "grounded_question_target", + "forward_aggregate_target_required", + ), + *_entries( + ResidualKind.CONTRACT_GAP_UNCLASSIFIED, + ResidualSourceAxis.UNKNOWN, + "unbound_base_quantity", + "container", + "content", + "count_per", + "worker", + "rate", + "duration", + ), +) + + +@pytest.mark.parametrize(("label", "kind", "axis"), EXPECTED_LABEL_MAP) +def test_contract_residual_exact_label_mapping( + label: str, + kind: ResidualKind, + axis: ResidualSourceAxis, +) -> None: + residuals = project_contract_residuals( + (_assessment(missing_bindings=(label,)),) + ) + + assert len(residuals) == 1 + residual = residuals[0] + assert residual.residual_kind is kind + assert residual.source_axis is axis + assert residual.residual_code == label + + +def test_all_currently_emitted_labels_are_accounted_for() -> None: + assert len(EXPECTED_LABEL_MAP) == 53 + assert len({label for label, _, _ in EXPECTED_LABEL_MAP}) == 53 + + +@pytest.mark.parametrize( + "label", + ("Quantity_Unbound", "quantity_unbound ", "unknown_label"), +) +def test_arbitrary_and_case_altered_labels_fall_back_without_fuzzy_match( + label: str, +) -> None: + residuals = project_contract_residuals( + (_assessment(missing_bindings=(label,)),) + ) + + assert len(residuals) == 1 + assert residuals[0].residual_code == label + assert residuals[0].residual_kind is ResidualKind.CONTRACT_GAP_UNCLASSIFIED + assert residuals[0].source_axis is ResidualSourceAxis.UNKNOWN + + +def test_duplicate_code_across_missing_and_hazard_fields_emits_once() -> None: + residuals = project_contract_residuals( + ( + _assessment( + missing_bindings=("quantity_unbound",), + unresolved_hazards=("quantity_unbound",), + ), + ) + ) + + assert len(residuals) == 1 + assert residuals[0].residual_code == "quantity_unbound" + + +def test_runnable_assessment_produces_no_residual_even_with_labels() -> None: + residuals = project_contract_residuals( + ( + _assessment( + missing_bindings=("changed_object_unbound",), + unresolved_hazards=("unit_object_conflict",), + runnable=True, + ), + ) + ) + + assert residuals == () + + +def test_refused_assessment_without_labels_produces_no_invented_residual() -> None: + assert project_contract_residuals((_assessment(),)) == () + + +def test_residual_preserves_code_spans_and_explanation_verbatim() -> None: + span_a = SourceSpan("gained", 4, 10, 0) + span_b = SourceSpan("3", 11, 12, 0) + explanation = "diagnostic candidate is not runnable: changed_object_unbound" + residuals = project_contract_residuals( + ( + _assessment( + missing_bindings=("changed_object_unbound",), + explanation=explanation, + evidence_spans=(span_a, span_b), + ), + ) + ) + + assert len(residuals) == 1 + residual = residuals[0] + assert residual.residual_code == "changed_object_unbound" + assert residual.evidence_spans == (span_a, span_b) + assert residual.explanation == explanation + + +def test_family_id_is_resolved_from_candidate_organ() -> None: + residual = project_contract_residuals( + ( + _assessment( + candidate_organ="unary_delta_transition", + missing_bindings=("changed_object_unbound",), + ), + ) + )[0] + + assert residual.family_id == "state_change.unary_delta" + + +def test_unknown_candidate_organ_keeps_none_family_id() -> None: + residual = project_contract_residuals( + ( + _assessment( + candidate_organ="future_unknown_organ", + missing_bindings=("unknown_label",), + ), + ) + )[0] + + assert residual.family_id is None + + +def test_replay_stability_and_sorted_output_are_independent_of_input_order() -> None: + assessment_a = _assessment(missing_bindings=("changed_object_unbound",)) + assessment_b = _assessment( + candidate_organ="quantity_entity_binding", + missing_bindings=("unit_kind_conflict",), + ) + + forward = project_contract_residuals((assessment_a, assessment_b)) + reversed_input = project_contract_residuals((assessment_b, assessment_a)) + + assert forward == reversed_input + assert tuple(residual.residual_id for residual in forward) == tuple( + sorted(residual.residual_id for residual in forward) + ) + + +def test_known_sha256_vector_and_explanation_independence() -> None: + span = SourceSpan("gained", 4, 10, 0) + residual_a = project_contract_residuals( + ( + _assessment( + missing_bindings=("changed_object_unbound",), + explanation="first explanation", + evidence_spans=(span,), + ), + ) + )[0] + residual_b = project_contract_residuals( + ( + _assessment( + missing_bindings=("changed_object_unbound",), + explanation="second explanation", + evidence_spans=(span,), + ), + ) + )[0] + + assert residual_a.residual_id == ( + "e56fa420ce0acc64071fc5ef819d740d54f1ab17f1a268728a0ef490784ffe32" + ) + assert residual_a.residual_id == residual_b.residual_id + + +def test_evidence_span_order_and_sentence_index_participate_in_identity() -> None: + span_a = SourceSpan("gained", 4, 10, 0) + span_b = SourceSpan("3", 11, 12, 0) + residual_ordered = project_contract_residuals( + ( + _assessment( + missing_bindings=("changed_object_unbound",), + evidence_spans=(span_a, span_b), + ), + ) + )[0] + residual_reversed = project_contract_residuals( + ( + _assessment( + missing_bindings=("changed_object_unbound",), + evidence_spans=(span_b, span_a), + ), + ) + )[0] + residual_no_sentence = project_contract_residuals( + ( + _assessment( + missing_bindings=("changed_object_unbound",), + evidence_spans=(SourceSpan("gained", 4, 10), span_b), + ), + ) + )[0] + + assert residual_ordered.residual_id != residual_reversed.residual_id + assert residual_ordered.residual_id != residual_no_sentence.residual_id + + +def test_empty_evidence_is_lawful_and_does_not_synthesize_span() -> None: + residual = project_contract_residuals( + ( + _assessment( + missing_bindings=("changed_object_unbound",), + evidence_spans=(), + ), + ) + )[0] + + assert residual.evidence_spans == () + + +def test_projection_leaves_assessment_unchanged() -> None: + span = SourceSpan("lost", 8, 12, 0) + assessment = _assessment( + missing_bindings=("delta_quantity_unbound",), + evidence_spans=(span,), + ) + before = assessment + + project_contract_residuals((assessment,)) + + assert assessment == before + + +def test_contract_residual_has_no_authority_fields() -> None: + disallowed = { + "runnable", + "verdict", + "answer", + "proof", + "search_eligible", + "retryable", + "priority", + "budget", + "action", + "serving_allowed", + "mutation", + "recommended_migration_target", + } + + assert disallowed.isdisjoint( + {field.name for field in dataclasses.fields(ContractResidual)} + ) + + +def test_public_api_is_exactly_the_read_model_projection() -> None: + import generate.contract_residuals as contract_residuals + + assert tuple(contract_residuals.__all__) == ( + "ResidualKind", + "ResidualSourceAxis", + "ContractResidual", + "project_contract_residuals", + ) + + +def test_family_examples_project_expected_residuals() -> None: + examples = ( + ( + _assessment( + candidate_organ="unary_delta_transition", + missing_bindings=( + "action_cue_unbound", + "direction_unbound", + "changed_object_unbound", + ), + ), + { + "action_cue_unbound": ResidualKind.MISSING_ROLE, + "direction_unbound": ResidualKind.MISSING_ROLE, + "changed_object_unbound": ResidualKind.MISSING_ROLE, + }, + ), + ( + _assessment( + candidate_organ="quantity_entity_binding", + missing_bindings=("quantity_kind_unresolved",), + unresolved_hazards=("unit_kind_conflict",), + ), + { + "quantity_kind_unresolved": ResidualKind.MISSING_ROLE, + "unit_kind_conflict": ResidualKind.UNIT_OBJECT_CONFLICT, + }, + ), + ( + _assessment( + candidate_organ="fraction_decrease", + missing_bindings=("delta_decrease_target_unbound",), + ), + {"delta_decrease_target_unbound": ResidualKind.TARGET_UNBOUND}, + ), + ( + _assessment( + candidate_organ="percent_partition", + missing_bindings=("percent_subgroup_links_incomplete",), + unresolved_hazards=( + "partition_subgroups_not_distinct", + "percent_change_vs_percent_of", + ), + ), + { + "partition_subgroups_not_distinct": ResidualKind.HAZARD_BLOCKED, + "percent_subgroup_links_incomplete": ( + ResidualKind.UNSUPPORTED_TOPOLOGY + ), + "percent_change_vs_percent_of": ResidualKind.HAZARD_BLOCKED, + }, + ), + ) + + for assessment, expected in examples: + residuals = project_contract_residuals((assessment,)) + assert {residual.residual_code for residual in residuals} == set(expected) + for residual in residuals: + assert residual.residual_kind is expected[residual.residual_code] + + +def test_positive_family_controls_project_to_empty_tuple() -> None: + runnable_assessments = ( + _assessment(candidate_organ="unary_delta_transition", runnable=True), + _assessment(candidate_organ="quantity_entity_binding", runnable=True), + _assessment(candidate_organ="fraction_decrease", runnable=True), + _assessment(candidate_organ="percent_partition", runnable=True), + ) + + assert project_contract_residuals(runnable_assessments) == () + + +def test_module_imports_are_leaf_read_model_allowlist() -> None: + source = Path("generate/contract_residuals.py").read_text() + tree = ast.parse(source) + imports: set[str] = set() + calls: set[str] = set() + + for node in ast.walk(tree): + if isinstance(node, ast.Import): + imports.update(alias.name for alias in node.names) + elif isinstance(node, ast.ImportFrom): + imports.add(node.module or "") + elif isinstance(node, ast.Call): + if isinstance(node.func, ast.Name): + calls.add(node.func.id) + elif isinstance(node.func, ast.Attribute): + calls.add(node.func.attr) + + assert imports <= { + "__future__", + "hashlib", + "json", + "dataclasses", + "enum", + "generate.kernel_facts", + "generate.problem_frame_contracts", + } + + forbidden_calls = { + "build_problem_frame", + "assess_contracts", + "determine", + "store", + "write", + "repair", + "search", + "serve", + "SourceSpan", + } + assert forbidden_calls.isdisjoint(calls) + + +def test_no_reverse_imports_into_contract_residuals() -> None: + for path in Path("generate").glob("*.py"): + if path.name == "contract_residuals.py": + continue + assert "contract_residuals" not in path.read_text() + + +def test_enum_values_are_lower_snake_case() -> None: + for enum_cls in (ResidualKind, ResidualSourceAxis): + for member in enum_cls: + assert member.value == member.value.lower() + assert " " not in member.value + assert "-" not in member.value