Merge pull request #862 from AssetOverflow/codex/feat/kernel-contract-residual-projection
feat(kernel): implement diagnostic-only ContractResidual projection
This commit is contained in:
Shay
GitHub
commit
c66ef3ee26
2 changed files with 856 additions and 0 deletions
326
generate/contract_residuals.py
Normal file
326
generate/contract_residuals.py
Normal file
|
|
@ -0,0 +1,326 @@
|
|||
"""Diagnostic-only ContractResidual projection over ContractAssessment outputs.
|
||||
|
||||
This module is a leaf read-model projector. It must not assess contracts, build
|
||||
frames, search, repair, serve, mutate artifacts, or allocate compute. The only
|
||||
dependency direction is:
|
||||
|
||||
ContractAssessment -> ContractResidual
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
from dataclasses import dataclass
|
||||
from enum import Enum, unique
|
||||
|
||||
from generate.kernel_facts import SourceSpan
|
||||
from generate.problem_frame_contracts import (
|
||||
ContractAssessment,
|
||||
get_contract_family_id,
|
||||
)
|
||||
|
||||
|
||||
@unique
|
||||
class ResidualKind(str, Enum):
|
||||
MISSING_PROPOSAL = "missing_proposal"
|
||||
MISSING_RELATION = "missing_relation"
|
||||
MISSING_ROLE = "missing_role"
|
||||
AMBIGUOUS_RELATION = "ambiguous_relation"
|
||||
AMBIGUOUS_ROLE = "ambiguous_role"
|
||||
INEXACT_PROVENANCE = "inexact_provenance"
|
||||
NONLOCAL_BINDING = "nonlocal_binding"
|
||||
UNSUPPORTED_TOPOLOGY = "unsupported_topology"
|
||||
UNIT_OBJECT_CONFLICT = "unit_object_conflict"
|
||||
HAZARD_BLOCKED = "hazard_blocked"
|
||||
TARGET_UNBOUND = "target_unbound"
|
||||
CONTRACT_GAP_UNCLASSIFIED = "contract_gap_unclassified"
|
||||
|
||||
|
||||
@unique
|
||||
class ResidualSourceAxis(str, Enum):
|
||||
PROPOSAL = "proposal"
|
||||
RELATION = "relation"
|
||||
ROLE = "role"
|
||||
PROVENANCE = "provenance"
|
||||
LOCALITY = "locality"
|
||||
TOPOLOGY = "topology"
|
||||
UNIT_OBJECT = "unit_object"
|
||||
HAZARD = "hazard"
|
||||
TARGET = "target"
|
||||
UNKNOWN = "unknown"
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class ContractResidual:
|
||||
residual_id: str
|
||||
candidate_organ: str
|
||||
family_id: str | None
|
||||
residual_kind: ResidualKind
|
||||
residual_code: str
|
||||
source_axis: ResidualSourceAxis
|
||||
evidence_spans: tuple[SourceSpan, ...]
|
||||
explanation: str
|
||||
|
||||
|
||||
_FALLBACK_CLASSIFICATION: tuple[ResidualKind, ResidualSourceAxis] = (
|
||||
ResidualKind.CONTRACT_GAP_UNCLASSIFIED,
|
||||
ResidualSourceAxis.UNKNOWN,
|
||||
)
|
||||
|
||||
_LABEL_CLASSIFICATIONS: dict[str, tuple[ResidualKind, ResidualSourceAxis]] = {
|
||||
"quantity_entity_proposal_required": (
|
||||
ResidualKind.MISSING_PROPOSAL,
|
||||
ResidualSourceAxis.PROPOSAL,
|
||||
),
|
||||
"unary_delta_proposal_required": (
|
||||
ResidualKind.MISSING_PROPOSAL,
|
||||
ResidualSourceAxis.PROPOSAL,
|
||||
),
|
||||
"local_binding_relation_unbound": (
|
||||
ResidualKind.MISSING_RELATION,
|
||||
ResidualSourceAxis.RELATION,
|
||||
),
|
||||
"quantity_unbound": (ResidualKind.MISSING_ROLE, ResidualSourceAxis.ROLE),
|
||||
"entity_unbound": (ResidualKind.MISSING_ROLE, ResidualSourceAxis.ROLE),
|
||||
"quantity_kind_unresolved": (
|
||||
ResidualKind.MISSING_ROLE,
|
||||
ResidualSourceAxis.ROLE,
|
||||
),
|
||||
"base_quantity_unbound": (ResidualKind.MISSING_ROLE, ResidualSourceAxis.ROLE),
|
||||
"scale_unbound": (ResidualKind.MISSING_ROLE, ResidualSourceAxis.ROLE),
|
||||
"state_entity_unbound": (ResidualKind.MISSING_ROLE, ResidualSourceAxis.ROLE),
|
||||
"grounded_partition_subgroup": (
|
||||
ResidualKind.MISSING_ROLE,
|
||||
ResidualSourceAxis.ROLE,
|
||||
),
|
||||
"grounded_whole_entity": (
|
||||
ResidualKind.MISSING_ROLE,
|
||||
ResidualSourceAxis.ROLE,
|
||||
),
|
||||
"original_whole_unbound": (
|
||||
ResidualKind.MISSING_ROLE,
|
||||
ResidualSourceAxis.ROLE,
|
||||
),
|
||||
"action_cue_unbound": (ResidualKind.MISSING_ROLE, ResidualSourceAxis.ROLE),
|
||||
"delta_quantity_unbound": (
|
||||
ResidualKind.MISSING_ROLE,
|
||||
ResidualSourceAxis.ROLE,
|
||||
),
|
||||
"changed_object_unbound": (
|
||||
ResidualKind.MISSING_ROLE,
|
||||
ResidualSourceAxis.ROLE,
|
||||
),
|
||||
"direction_unbound": (ResidualKind.MISSING_ROLE, ResidualSourceAxis.ROLE),
|
||||
"local_binding_relation_ambiguous": (
|
||||
ResidualKind.AMBIGUOUS_RELATION,
|
||||
ResidualSourceAxis.RELATION,
|
||||
),
|
||||
"decrease_relation_ambiguous": (
|
||||
ResidualKind.AMBIGUOUS_RELATION,
|
||||
ResidualSourceAxis.RELATION,
|
||||
),
|
||||
"unary_delta_relation_ambiguous": (
|
||||
ResidualKind.AMBIGUOUS_RELATION,
|
||||
ResidualSourceAxis.RELATION,
|
||||
),
|
||||
"quantity_ambiguous": (ResidualKind.AMBIGUOUS_ROLE, ResidualSourceAxis.ROLE),
|
||||
"entity_ambiguous": (ResidualKind.AMBIGUOUS_ROLE, ResidualSourceAxis.ROLE),
|
||||
"multiple_original_whole_candidates": (
|
||||
ResidualKind.AMBIGUOUS_ROLE,
|
||||
ResidualSourceAxis.ROLE,
|
||||
),
|
||||
"delta_quantity_ambiguous": (
|
||||
ResidualKind.AMBIGUOUS_ROLE,
|
||||
ResidualSourceAxis.ROLE,
|
||||
),
|
||||
"changed_object_ambiguous": (
|
||||
ResidualKind.AMBIGUOUS_ROLE,
|
||||
ResidualSourceAxis.ROLE,
|
||||
),
|
||||
"provenance_span_inexact": (
|
||||
ResidualKind.INEXACT_PROVENANCE,
|
||||
ResidualSourceAxis.PROVENANCE,
|
||||
),
|
||||
"base_quantity_provenance_missing": (
|
||||
ResidualKind.INEXACT_PROVENANCE,
|
||||
ResidualSourceAxis.PROVENANCE,
|
||||
),
|
||||
"scale_provenance_missing": (
|
||||
ResidualKind.INEXACT_PROVENANCE,
|
||||
ResidualSourceAxis.PROVENANCE,
|
||||
),
|
||||
"quantity_entity_nonlocal": (
|
||||
ResidualKind.NONLOCAL_BINDING,
|
||||
ResidualSourceAxis.LOCALITY,
|
||||
),
|
||||
"pronoun_antecedent_unresolved": (
|
||||
ResidualKind.NONLOCAL_BINDING,
|
||||
ResidualSourceAxis.LOCALITY,
|
||||
),
|
||||
"percent_subgroup_links_incomplete": (
|
||||
ResidualKind.UNSUPPORTED_TOPOLOGY,
|
||||
ResidualSourceAxis.TOPOLOGY,
|
||||
),
|
||||
"inverse_topology_unlicensed": (
|
||||
ResidualKind.UNSUPPORTED_TOPOLOGY,
|
||||
ResidualSourceAxis.TOPOLOGY,
|
||||
),
|
||||
"event_assertion_unlicensed": (
|
||||
ResidualKind.UNSUPPORTED_TOPOLOGY,
|
||||
ResidualSourceAxis.TOPOLOGY,
|
||||
),
|
||||
"passive_voice_unsupported": (
|
||||
ResidualKind.UNSUPPORTED_TOPOLOGY,
|
||||
ResidualSourceAxis.TOPOLOGY,
|
||||
),
|
||||
"multiple_actor_surface": (
|
||||
ResidualKind.UNSUPPORTED_TOPOLOGY,
|
||||
ResidualSourceAxis.TOPOLOGY,
|
||||
),
|
||||
"unit_kind_conflict": (
|
||||
ResidualKind.UNIT_OBJECT_CONFLICT,
|
||||
ResidualSourceAxis.UNIT_OBJECT,
|
||||
),
|
||||
"unit_continuity_unproven": (
|
||||
ResidualKind.UNIT_OBJECT_CONFLICT,
|
||||
ResidualSourceAxis.UNIT_OBJECT,
|
||||
),
|
||||
"unit_object_conflict": (
|
||||
ResidualKind.UNIT_OBJECT_CONFLICT,
|
||||
ResidualSourceAxis.UNIT_OBJECT,
|
||||
),
|
||||
"competing_family_context": (
|
||||
ResidualKind.HAZARD_BLOCKED,
|
||||
ResidualSourceAxis.HAZARD,
|
||||
),
|
||||
"percent_change_vs_percent_of": (
|
||||
ResidualKind.HAZARD_BLOCKED,
|
||||
ResidualSourceAxis.HAZARD,
|
||||
),
|
||||
"state_entity_continuity_unproven": (
|
||||
ResidualKind.HAZARD_BLOCKED,
|
||||
ResidualSourceAxis.HAZARD,
|
||||
),
|
||||
"scale_out_of_range": (
|
||||
ResidualKind.HAZARD_BLOCKED,
|
||||
ResidualSourceAxis.HAZARD,
|
||||
),
|
||||
"partition_subgroups_not_distinct": (
|
||||
ResidualKind.HAZARD_BLOCKED,
|
||||
ResidualSourceAxis.HAZARD,
|
||||
),
|
||||
"delta_decrease_target_unbound": (
|
||||
ResidualKind.TARGET_UNBOUND,
|
||||
ResidualSourceAxis.TARGET,
|
||||
),
|
||||
"delta_decrease_target_required": (
|
||||
ResidualKind.TARGET_UNBOUND,
|
||||
ResidualSourceAxis.TARGET,
|
||||
),
|
||||
"grounded_question_target": (
|
||||
ResidualKind.TARGET_UNBOUND,
|
||||
ResidualSourceAxis.TARGET,
|
||||
),
|
||||
"forward_aggregate_target_required": (
|
||||
ResidualKind.TARGET_UNBOUND,
|
||||
ResidualSourceAxis.TARGET,
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
def _classify_label(label: str) -> tuple[ResidualKind, ResidualSourceAxis]:
|
||||
return _LABEL_CLASSIFICATIONS.get(label, _FALLBACK_CLASSIFICATION)
|
||||
|
||||
|
||||
def _dedupe_codes(codes: tuple[str, ...]) -> tuple[str, ...]:
|
||||
seen: set[str] = set()
|
||||
unique: list[str] = []
|
||||
for code in codes:
|
||||
if code in seen:
|
||||
continue
|
||||
seen.add(code)
|
||||
unique.append(code)
|
||||
return tuple(unique)
|
||||
|
||||
|
||||
def _span_payload(span: SourceSpan) -> dict[str, object]:
|
||||
return {
|
||||
"text": span.text,
|
||||
"start": span.start,
|
||||
"end": span.end,
|
||||
"sentence_index": span.sentence_index,
|
||||
}
|
||||
|
||||
|
||||
def _residual_id(
|
||||
*,
|
||||
candidate_organ: str,
|
||||
residual_kind: ResidualKind,
|
||||
residual_code: str,
|
||||
evidence_spans: tuple[SourceSpan, ...],
|
||||
) -> str:
|
||||
payload = {
|
||||
"candidate_organ": candidate_organ,
|
||||
"residual_kind": residual_kind.value,
|
||||
"residual_code": residual_code,
|
||||
"evidence_spans": [_span_payload(span) for span in evidence_spans],
|
||||
}
|
||||
encoded = json.dumps(
|
||||
payload,
|
||||
ensure_ascii=False,
|
||||
separators=(",", ":"),
|
||||
sort_keys=True,
|
||||
).encode("utf-8")
|
||||
return hashlib.sha256(encoded).hexdigest()
|
||||
|
||||
|
||||
def _project_one(
|
||||
assessment: ContractAssessment,
|
||||
residual_code: str,
|
||||
) -> ContractResidual:
|
||||
residual_kind, source_axis = _classify_label(residual_code)
|
||||
evidence_spans = assessment.evidence_spans
|
||||
return ContractResidual(
|
||||
residual_id=_residual_id(
|
||||
candidate_organ=assessment.candidate_organ,
|
||||
residual_kind=residual_kind,
|
||||
residual_code=residual_code,
|
||||
evidence_spans=evidence_spans,
|
||||
),
|
||||
candidate_organ=assessment.candidate_organ,
|
||||
family_id=get_contract_family_id(assessment.candidate_organ),
|
||||
residual_kind=residual_kind,
|
||||
residual_code=residual_code,
|
||||
source_axis=source_axis,
|
||||
evidence_spans=evidence_spans,
|
||||
explanation=assessment.explanation,
|
||||
)
|
||||
|
||||
|
||||
def project_contract_residuals(
|
||||
assessments: tuple[ContractAssessment, ...],
|
||||
) -> tuple[ContractResidual, ...]:
|
||||
"""Project refused contract assessments into non-authoritative residuals.
|
||||
|
||||
Runnable assessments produce no residuals, even if a malformed assessment carries
|
||||
missing or unresolved labels. The residual projection is diagnostic only and
|
||||
must not be used as serving/search/mutation authority.
|
||||
"""
|
||||
residuals: list[ContractResidual] = []
|
||||
for assessment in assessments:
|
||||
if assessment.runnable:
|
||||
continue
|
||||
codes = _dedupe_codes(
|
||||
(*assessment.missing_bindings, *assessment.unresolved_hazards)
|
||||
)
|
||||
residuals.extend(_project_one(assessment, code) for code in codes)
|
||||
return tuple(sorted(residuals, key=lambda residual: residual.residual_id))
|
||||
|
||||
|
||||
__all__ = [
|
||||
"ResidualKind",
|
||||
"ResidualSourceAxis",
|
||||
"ContractResidual",
|
||||
"project_contract_residuals",
|
||||
]
|
||||
530
tests/test_contract_residuals.py
Normal file
530
tests/test_contract_residuals.py
Normal file
|
|
@ -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
|
||||
Loading…
Reference in a new issue