326 lines
9.8 KiB
Python
326 lines
9.8 KiB
Python
"""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",
|
|
]
|