core/generate/problem_frame_contracts.py

"""Diagnostic organ-contract readiness derived only from ProblemFrame evidence.

Contract dispatch is deliberately narrow:
- ``assess_contracts()`` routes to three diagnostic assessment functions.
- The ``_CONTRACT_REGISTRY`` provides catalog metadata for introspection and
  proposal-trace generation; it does not replace the structural logic inside
  each assessment function.
- All registered contracts have ``serving_allowed=False``; this module must
  never be imported from serving dispatch paths.
"""

from __future__ import annotations

import re
import numpy as np
from dataclasses import dataclass, field
from algebra.backend import versor_condition

from generate.construction_affordances import (
    ConstructionContract,
    _DECREASE_TO_FRACTION_FAMILY,
    _PERCENT_PARTITION_FAMILY,
    _QUANTITY_ENTITY_FAMILY,
    _UNARY_DELTA_FAMILY,
)
from generate.kernel_facts import BoundRelation, GroundedMention, SourceSpan
from generate.problem_frame import ProblemFrame
from chat.pack_resolver import resolve_geometric_signature


# ---------------------------------------------------------------------------
# Contract registry
#
# Maps candidate_organ -> ConstructionContract.  This registry provides
# metadata for proposal-trace generation and external introspection.  It does
# not replace the per-assessment dispatch logic in assess_contracts(); the
# structural proof obligations for each family live inside the dedicated
# assess_* functions below.
#
# Why not route dispatch through the registry?
# assess_fraction_decrease and assess_percent_partition each contain
# construction-specific structural logic (role iteration, topology proofs,
# hazard escalation) that cannot be generalised behind a single callable
# without introducing a forced abstraction boundary.  The registry expresses
# *what* a construction is; the assessment functions express *how* its
# obligations are checked.  Both layers are needed and should remain separate.
# ---------------------------------------------------------------------------

_CONTRACT_REGISTRY: dict[str, ConstructionContract] = {
    "fraction_decrease": ConstructionContract(
        family=_DECREASE_TO_FRACTION_FAMILY,
        assess_fn_name="assess_fraction_decrease",
    ),
    "percent_partition": ConstructionContract(
        family=_PERCENT_PARTITION_FAMILY,
        assess_fn_name="assess_percent_partition",
    ),
    "quantity_entity_binding": ConstructionContract(
        family=_QUANTITY_ENTITY_FAMILY,
        assess_fn_name="assess_quantity_entity",
    ),
    "unary_delta_transition": ConstructionContract(
        family=_UNARY_DELTA_FAMILY,
        assess_fn_name="assess_unary_delta",
    ),
}


def get_contract_family_id(candidate_organ: str) -> str | None:
    """Return the catalog family ID for a candidate organ, if registered."""
    contract = _CONTRACT_REGISTRY.get(candidate_organ)
    return contract.family.family_id if contract else None


@dataclass(frozen=True, slots=True)
class VersorBinding:
    source_span: tuple[int, int]
    semantic_identity: str
    geometric_payload: np.ndarray
    versor_error: float

    def __post_init__(self):
        if not isinstance(self.geometric_payload, np.ndarray):
            raise TypeError("geometric_payload must be a numpy.ndarray")
        if self.geometric_payload.dtype != np.float64:
            raise TypeError("geometric_payload must be float64")
        if not self.geometric_payload.flags.c_contiguous:
            raise ValueError("geometric_payload must be C-contiguous")
        if self.geometric_payload.shape != (32,):
            raise ValueError("geometric_payload must be of shape (32,)")
        
        # Mathematically assert that versor_error < 1e-6
        actual_error = float(versor_condition(self.geometric_payload))
        if self.versor_error >= 1e-6 or actual_error >= 1e-6:
            raise ValueError(f"versor_error {max(self.versor_error, actual_error)} exceeds 1e-6 threshold")


@dataclass(frozen=True, slots=True)
class ContractAssessment:
    candidate_organ: str = ""
    missing_bindings: tuple[str, ...] = ()
    unresolved_hazards: tuple[str, ...] = ()
    runnable: bool = False
    explanation: str = ""
    evidence_spans: tuple[SourceSpan, ...] = ()
    bindings: list[VersorBinding] = field(default_factory=list)


def _roles(frame: ProblemFrame, relation_type: str) -> set[str]:
    return {
        role.role
        for relation in frame.bound_relations
        if relation.relation_type == relation_type
        for role in relation.roles
    }


def _evidence(frame: ProblemFrame, relation_type: str) -> tuple[SourceSpan, ...]:
    spans = {
        (span.start, span.end, span.text): span
        for relation in frame.bound_relations
        if relation.relation_type == relation_type
        for span in relation.evidence_spans
    }
    if frame.bound_question_target is not None:
        for span in frame.bound_question_target.evidence_spans:
            spans[(span.start, span.end, span.text)] = span
    return tuple(spans[key] for key in sorted(spans))


def _role_target(relation: BoundRelation, role_name: str) -> str | None:
    return next(
        (role.target_id for role in relation.roles if role.role == role_name), None
    )


def _role_spans(relation: BoundRelation, role_name: str) -> tuple[SourceSpan, ...]:
    role = next((item for item in relation.roles if item.role == role_name), None)
    return () if role is None else role.evidence_spans


def _mention_map(frame: ProblemFrame) -> dict[str, object]:
    return {mention.mention_id: mention for mention in frame.mentions}


def _quantity_value_by_mention_id(frame: ProblemFrame) -> dict[str, object]:
    quantities = {quantity.fact_id: quantity for quantity in frame.quantities}
    return {
        mention.mention_id: quantities[mention.fact_id]
        for mention in frame.mentions
        if mention.fact_id is not None and mention.fact_id in quantities
    }


def _quantity_entity_bindings(frame: ProblemFrame) -> tuple[tuple[str, str], ...]:
    return tuple(
        (binding.source_mention_id, binding.target_mention_id)
        for binding in frame.bindings
        if binding.binding_type == "quantity_entity"
    )


def _quantity_unit_bindings(frame: ProblemFrame) -> dict[str, str]:
    return {
        binding.source_mention_id: binding.target_mention_id
        for binding in frame.bindings
        if binding.binding_type == "quantity_unit"
    }


_UNRESOLVED_ENTITY_SURFACES: frozenset[str] = frozenset(
    {
        "he",
        "her",
        "hers",
        "him",
        "his",
        "it",
        "its",
        "one",
        "ones",
        "she",
        "their",
        "theirs",
        "them",
        "these",
        "they",
        "this",
        "those",
    }
)


def _span_is_exact(frame: ProblemFrame, span: SourceSpan) -> bool:
    return (
        bool(frame.problem_text)
        and 0 <= span.start <= span.end <= len(frame.problem_text)
        and bool(span.text)
        and frame.problem_text[span.start : span.end] == span.text
    )


def _spans_are_local(
    problem_text: str,
    first: SourceSpan,
    second: SourceSpan,
) -> bool:
    left, right = sorted((first, second), key=lambda span: span.start)
    if left.end > right.start:
        return False
    return not any(marker in problem_text[left.end : right.start] for marker in ".!?")


def _unique_evidence(spans: tuple[SourceSpan, ...]) -> tuple[SourceSpan, ...]:
    unique = {(span.start, span.end, span.text): span for span in spans}
    return tuple(unique[key] for key in sorted(unique))


def _entity_mentions(frame: ProblemFrame) -> tuple[GroundedMention, ...]:
    return tuple(
        mention
        for mention in frame.mentions
        if mention.kind in {"entity", "object", "actor"}
    )


def assess_quantity_entity(frame: ProblemFrame) -> ContractAssessment:
    """Assess one proposal-backed local quantity/entity edge.

    This contract is deliberately stricter than generic mention extraction: it
    closes only one exact scalar, one exact entity, one exact local edge, and a
    positively grounded count/measurement disposition.  It derives neither an
    answer nor serving authority.
    """

    proposals = tuple(
        proposal
        for proposal in frame.proposals
        if proposal.family_id == _QUANTITY_ENTITY_FAMILY.family_id
    )
    bindings = tuple(
        binding
        for binding in frame.bindings
        if binding.binding_type == "quantity_entity"
    )
    mentions = {mention.mention_id: mention for mention in frame.mentions}
    quantity_facts = {quantity.fact_id: quantity for quantity in frame.quantities}

    missing: list[str] = []
    unresolved: set[str] = set()
    if len(proposals) != 1:
        missing.append("quantity_entity_proposal_required")
    proposal = proposals[0] if len(proposals) == 1 else None

    if not bindings:
        missing.append("local_binding_relation_unbound")
    elif len(bindings) != 1:
        missing.append("local_binding_relation_ambiguous")
    binding = bindings[0] if len(bindings) == 1 else None

    quantity = mentions.get(binding.source_mention_id) if binding is not None else None
    entity = mentions.get(binding.target_mention_id) if binding is not None else None
    if quantity is None or quantity.kind != "quantity":
        missing.append("quantity_unbound")
    elif quantity.fact_id is None or quantity.fact_id not in quantity_facts:
        missing.append("quantity_unbound")
    if len(frame.quantities) != 1:
        missing.append("quantity_ambiguous")

    if entity is None or entity.kind not in {"entity", "object"}:
        missing.append("entity_unbound")
    elif entity.surface.lower() in _UNRESOLVED_ENTITY_SURFACES:
        missing.append("entity_unbound")
        unresolved.add("quantity_entity_nonlocal")

    if quantity is not None and entity is not None:
        competing_entities = tuple(
            mention
            for mention in _entity_mentions(frame)
            if mention.mention_id != entity.mention_id
            and _spans_are_local(frame.problem_text, entity.span, mention.span)
        )
        if competing_entities:
            missing.append("entity_ambiguous")
        if not _spans_are_local(frame.problem_text, quantity.span, entity.span):
            missing.append("quantity_entity_nonlocal")

    if proposal is not None and quantity is not None and entity is not None:
        cue_contains_binding = any(
            cue.start <= quantity.span.start and entity.span.end <= cue.end
            for cue in proposal.evidence_spans
        )
        if not cue_contains_binding:
            missing.append("local_binding_relation_unbound")

    dispositions = tuple(
        disposition
        for disposition in frame.quantity_kind_dispositions
        if quantity is not None
        and entity is not None
        and disposition.quantity_mention_id == quantity.mention_id
        and disposition.entity_mention_id == entity.mention_id
    )
    if len(dispositions) != 1:
        missing.append("quantity_kind_unresolved")
    disposition = dispositions[0] if len(dispositions) == 1 else None

    unit_bindings = tuple(
        binding
        for binding in frame.bindings
        if quantity is not None
        and binding.binding_type == "quantity_unit"
        and binding.source_mention_id == quantity.mention_id
    )
    if len(unit_bindings) > 1:
        missing.append("unit_kind_conflict")
    unit_binding = unit_bindings[0] if len(unit_bindings) == 1 else None
    unit = (
        mentions.get(unit_binding.target_mention_id)
        if unit_binding is not None
        else None
    )
    if disposition is not None:
        if disposition.quantity_kind == "count" and unit_binding is not None:
            missing.append("unit_kind_conflict")
        elif disposition.quantity_kind == "measurement":
            if (
                unit_binding is None
                or disposition.unit_mention_id != unit_binding.target_mention_id
                or unit is None
                or unit.kind != "unit"
            ):
                missing.append("unit_kind_conflict")
    elif unit_binding is not None:
        missing.append("unit_kind_conflict")

    if unit is not None and entity is not None and unit.span == entity.span:
        missing.append("unit_kind_conflict")

    evidence = _unique_evidence(
        tuple(
            span
            for group in (
                (() if proposal is None else proposal.evidence_spans),
                (() if binding is None else binding.evidence_spans),
                (() if disposition is None else disposition.evidence_spans),
                (() if unit_binding is None else unit_binding.evidence_spans),
            )
            for span in group
        )
    )
    exact_evidence = evidence
    if quantity is not None:
        exact_evidence = _unique_evidence((*exact_evidence, quantity.span))
        quantity_fact = (
            quantity_facts.get(quantity.fact_id)
            if quantity.fact_id is not None
            else None
        )
        if (
            quantity_fact is None
            or quantity.span not in quantity_fact.provenance.source_spans
        ):
            missing.append("provenance_span_inexact")
    if entity is not None:
        exact_evidence = _unique_evidence((*exact_evidence, entity.span))
    if unit is not None:
        exact_evidence = _unique_evidence((*exact_evidence, unit.span))
        unit_fact = next(
            (grounded for grounded in frame.units if unit.fact_id == grounded.fact_id),
            None,
        )
        if unit_fact is None or unit.span not in unit_fact.provenance.source_spans:
            missing.append("provenance_span_inexact")

    if binding is not None and quantity is not None and entity is not None:
        if binding.evidence_spans != (quantity.span, entity.span):
            missing.append("provenance_span_inexact")
    if unit_binding is not None and quantity is not None and unit is not None:
        if unit_binding.evidence_spans != (quantity.span, unit.span):
            missing.append("provenance_span_inexact")
    if not exact_evidence or not all(
        _span_is_exact(frame, span) for span in exact_evidence
    ):
        missing.append("provenance_span_inexact")

    competing_families = {candidate.name for candidate in frame.process_frames}
    if competing_families:
        missing.append("competing_family_context")
    categories = {hazard.category for hazard in frame.hazards}
    if "percent_change_vs_percent_of" in categories:
        unresolved.add("percent_change_vs_percent_of")

    missing_bindings = tuple(dict.fromkeys(missing))
    unresolved_hazards = tuple(sorted(unresolved))
    runnable = not missing_bindings and not unresolved_hazards
    return ContractAssessment(
        candidate_organ="quantity_entity_binding",
        missing_bindings=missing_bindings,
        unresolved_hazards=unresolved_hazards,
        runnable=runnable,
        explanation=(
            "one exact local quantity/entity binding is grounded diagnostically"
            if runnable
            else "diagnostic candidate is not runnable: "
            + ", ".join((*missing_bindings, *unresolved_hazards))
        ),
        evidence_spans=exact_evidence,
    )


def assess_fraction_decrease(frame: ProblemFrame) -> ContractAssessment:
    mentions = _mention_map(frame)
    quantities = _quantity_value_by_mention_id(frame)
    quantity_units = _quantity_unit_bindings(frame)
    relations = [
        relation
        for relation in frame.bound_relations
        if relation.relation_type == "decrease_to_fraction"
    ]
    question_target = frame.bound_question_target

    missing: list[str] = []
    unresolved: set[str] = set()
    if len(relations) != 1:
        missing.append("decrease_relation_ambiguous")
    relation = relations[0] if len(relations) == 1 else None

    base_id = _role_target(relation, "base_quantity") if relation is not None else None
    scale_id = _role_target(relation, "scale") if relation is not None else None
    state_id = _role_target(relation, "state_entity") if relation is not None else None
    unit_id = _role_target(relation, "unit") if relation is not None else None

    if base_id is None:
        missing.append("base_quantity_unbound")
    if scale_id is None:
        missing.append("scale_unbound")
    if state_id is None:
        missing.append("state_entity_unbound")
    if base_id is not None and base_id not in quantities:
        missing.append("base_quantity_provenance_missing")
    if scale_id is not None and scale_id not in quantities:
        missing.append("scale_provenance_missing")
    if (
        unit_id is not None
        and base_id is not None
        and quantity_units.get(base_id) != unit_id
    ):
        missing.append("unit_continuity_unproven")

    if question_target is None or not question_target.grounded:
        missing.append("delta_decrease_target_unbound")
    else:
        if not (
            question_target.target_operator == "difference"
            and question_target.target_state == "delta"
            and question_target.target_direction == "decrease"
        ):
            missing.append("delta_decrease_target_required")
        if (
            state_id is not None
            and state_id in mentions
            and question_target.target_mention_id in mentions
        ):
            relation_state = mentions[state_id]
            target_state = mentions[question_target.target_mention_id]
            if relation_state.surface.lower() != target_state.surface.lower():
                missing.append("state_entity_continuity_unproven")

    scale = quantities.get(scale_id) if scale_id is not None else None
    if scale is not None and not (0 < scale.value < 1):
        missing.append("scale_out_of_range")

    categories = {hazard.category for hazard in frame.hazards}
    if (
        any(item.startswith("base_quantity") for item in missing)
        and "unbound_base_quantity" in categories
    ):
        unresolved.add("unbound_base_quantity")

    evidence_spans = (
        _evidence(frame, "decrease_to_fraction")
        if relation is None
        else tuple(
            sorted(
                {
                    (span.start, span.end, span.text): span
                    for span in (
                        *relation.evidence_spans,
                        *(question_target.evidence_spans if question_target else ()),
                    )
                }.values(),
                key=lambda span: (span.start, span.end, span.text),
            )
        )
    )
    runnable = not missing and not unresolved
    return ContractAssessment(
        candidate_organ="fraction_decrease",
        missing_bindings=tuple(dict.fromkeys(missing)),
        unresolved_hazards=tuple(sorted(unresolved)),
        runnable=runnable,
        explanation=(
            "all fraction-decrease roles and delta target obligations are grounded"
            if runnable
            else "diagnostic candidate is not runnable: "
            + ", ".join((*dict.fromkeys(missing), *sorted(unresolved)))
        ),
        evidence_spans=evidence_spans,
    )


def assess_unary_delta(frame: ProblemFrame) -> ContractAssessment:
    proposals = tuple(
        proposal
        for proposal in frame.proposals
        if proposal.family_id == _UNARY_DELTA_FAMILY.family_id
    )
    relations = tuple(
        relation
        for relation in frame.bound_relations
        if relation.relation_type == "unary_delta"
    )
    mentions = _mention_map(frame)

    missing: list[str] = []
    unresolved: set[str] = set()

    # 1. require exactly one unary-delta proposal
    if len(proposals) != 1:
        missing.append("unary_delta_proposal_required")
    proposal = proposals[0] if len(proposals) == 1 else None

    # 2. require exactly one typed cue in frame.unary_delta_cues
    if len(frame.unary_delta_cues) != 1:
        missing.append("action_cue_unbound")
    cue = frame.unary_delta_cues[0] if len(frame.unary_delta_cues) == 1 else None

    # 3. require exact proposal cue span == typed cue span
    if proposal is not None and cue is not None:
        if proposal.evidence_spans != (cue.span,):
            missing.append("provenance_span_inexact")

    relation = relations[0] if len(relations) == 1 else None

    # Check if multiple relations exist
    if len(relations) > 1:
        missing.append("unary_delta_relation_ambiguous")

    # Roles / targets
    cue_id_role = None
    quantity_id = None
    object_id = None
    direction_role = None

    if relation is not None:
        cue_id_role = _role_target(relation, "action_cue")
        quantity_id = _role_target(relation, "delta_quantity")
        object_id = _role_target(relation, "changed_object")
        direction_role = _role_target(relation, "direction")

        # 4. require relation action_cue target to equal cue.cue_id
        if cue is not None and cue_id_role != cue.cue_id:
            missing.append("action_cue_unbound")

        # 5. derive expected direction from the typed cue and check it
        expected_direction = cue.direction if cue is not None else None
        if direction_role is None or direction_role != expected_direction:
            missing.append("direction_unbound")
    else:
        # If relation is absent, all roles are unbound
        missing.append("action_cue_unbound")
        missing.append("direction_unbound")

    # Get mentions if they are referenced by the relation
    quantity = mentions.get(quantity_id) if quantity_id is not None else None
    changed_object = mentions.get(object_id) if object_id is not None else None

    # 6. require quantity and changed_object roles only when a relation is present
    if relation is not None:
        if quantity is None or quantity.kind != "quantity":
            missing.append("delta_quantity_unbound")
        if changed_object is None or changed_object.kind != "object":
            missing.append("changed_object_unbound")
    else:
        # Refuse missing quantity/object with stable missing labels from frame evidence
        quantities_in_frame = [m for m in frame.mentions if m.kind == "quantity"]
        objects_in_frame = [m for m in frame.mentions if m.kind == "object"]

        if not quantities_in_frame:
            missing.append("delta_quantity_unbound")
        elif len(quantities_in_frame) > 1:
            missing.append("delta_quantity_ambiguous")
        else:
            quantity = quantities_in_frame[0]

        if not objects_in_frame:
            missing.append("changed_object_unbound")
        elif len(objects_in_frame) > 1:
            missing.append("changed_object_ambiguous")
        else:
            changed_object = objects_in_frame[0]

        # Check local binding relation if both exist but relation is absent
        if quantity is not None and changed_object is not None:
            has_binding = any(
                b.binding_type == "quantity_entity"
                and b.source_mention_id == quantity.mention_id
                and b.target_mention_id == changed_object.mention_id
                for b in frame.bindings
            )
            if not has_binding:
                missing.append("local_binding_relation_unbound")

    # 7. refuse unit/object conflict such as `Tom gained 3 degrees.`
    if quantity is not None:
        has_unit = any(
            b.binding_type == "quantity_unit"
            and b.source_mention_id == quantity.mention_id
            for b in frame.bindings
        )
        if has_unit:
            missing.append("quantity_kind_unresolved")
            missing.append("unit_object_conflict")

    # 8. use exact spans only
    exact_evidence = _unique_evidence(
        tuple(
            span
            for group in (
                (() if proposal is None else proposal.evidence_spans),
                (() if cue is None else (cue.span,)),
                (() if quantity is None else (quantity.span,)),
                (() if changed_object is None else (changed_object.span,)),
            )
            for span in group
        )
    )

    # Span check for proposal cue vs cue
    if (
        proposal is not None
        and cue is not None
        and proposal.evidence_spans != (cue.span,)
    ):
        missing.append("provenance_span_inexact")

    # Span check for relation evidence
    if (
        relation is not None
        and quantity is not None
        and changed_object is not None
        and cue is not None
    ):
        if relation.evidence_spans != (cue.span, quantity.span, changed_object.span):
            missing.append("provenance_span_inexact")

    if not exact_evidence or not all(
        _span_is_exact(frame, span) for span in exact_evidence
    ):
        missing.append("provenance_span_inexact")

    # Order/containment check
    if (
        cue is not None
        and changed_object is not None
        and not _spans_are_local(
            frame.problem_text,
            cue.span,
            changed_object.span,
        )
    ):
        missing.append("quantity_entity_nonlocal")
    if (
        cue is not None
        and quantity is not None
        and not _spans_are_local(
            frame.problem_text,
            cue.span,
            quantity.span,
        )
    ):
        missing.append("quantity_entity_nonlocal")

    # Check for pronoun antecedent unresolved
    pronouns = {
        "he",
        "her",
        "hers",
        "him",
        "his",
        "it",
        "its",
        "one",
        "ones",
        "she",
        "their",
        "theirs",
        "them",
        "these",
        "they",
        "this",
        "those",
    }
    if any(re.search(rf"\b{p}\b", frame.problem_text.lower()) for p in pronouns):
        unresolved.add("pronoun_antecedent_unresolved")

    # Check negation / modality
    negation_modality = {
        "not",
        "never",
        "may",
        "might",
        "could",
        "would",
        "should",
        "can",
    }
    if any(
        re.search(rf"\b{word}\b", frame.problem_text.lower())
        for word in negation_modality
    ):
        unresolved.add("event_assertion_unlicensed")

    # Check passive voice
    if cue is not None:
        passive_pattern = rf"\b(was|were|been|be)\s+{cue.surface}\b"
        by_pattern = rf"\b{cue.surface}\s+by\b"
        if re.search(passive_pattern, frame.problem_text.lower()) or re.search(
            by_pattern, frame.problem_text.lower()
        ):
            unresolved.add("passive_voice_unsupported")

    # Check for multiple actors
    if len(frame.actors) > 1:
        unresolved.add("multiple_actor_surface")

    missing_bindings = tuple(dict.fromkeys(missing))
    unresolved_hazards = tuple(sorted(unresolved))
    runnable = not missing_bindings and not unresolved_hazards
    return ContractAssessment(
        candidate_organ="unary_delta_transition",
        missing_bindings=missing_bindings,
        unresolved_hazards=unresolved_hazards,
        runnable=runnable,
        explanation=(
            "one exact local unary gained/lost delta is grounded diagnostically"
            if runnable
            else "diagnostic candidate is not runnable: "
            + ", ".join((*missing_bindings, *unresolved_hazards))
        ),
        evidence_spans=exact_evidence,
    )


def assess_percent_partition(frame: ProblemFrame) -> ContractAssessment:
    mentions = {mention.mention_id: mention for mention in frame.mentions}
    quantities = _quantity_value_by_mention_id(frame)
    quantity_entity = _quantity_entity_bindings(frame)
    subgroups = [
        relation
        for relation in frame.bound_relations
        if relation.relation_type == "subgroup_partition"
    ]
    percentages = [
        relation
        for relation in frame.bound_relations
        if relation.relation_type == "percent_of"
    ]
    linked_pairs: list[tuple[BoundRelation, BoundRelation]] = []
    subgroup_part_ids: set[str] = set()
    shared_whole_ids: set[str] = set()
    original_whole_quantities: set[str] = set()

    for subgroup in subgroups:
        subgroup_part = _role_target(subgroup, "part")
        subgroup_whole = _role_target(subgroup, "whole")
        if subgroup_part is None or subgroup_whole is None:
            continue
        subgroup_part_ids.add(subgroup_part)
        shared_whole_ids.add(subgroup_whole)
        relation_start = min(span.start for span in subgroup.evidence_spans)
        original_whole_quantities.update(
            quantity_id
            for quantity_id, entity_id in quantity_entity
            if entity_id == subgroup_whole
            and quantity_id in mentions
            and quantity_id in quantities
            and mentions[quantity_id].span.start < relation_start
        )
        for percent in percentages:
            percent_part = _role_target(percent, "part")
            percent_whole = _role_target(percent, "whole")
            if percent_part == subgroup_part and percent_whole == subgroup_whole:
                linked_pairs.append((subgroup, percent))

    missing: list[str] = []
    if not subgroups:
        missing.append("grounded_partition_subgroup")
    if not shared_whole_ids:
        missing.append("grounded_whole_entity")
    if not original_whole_quantities:
        missing.append("original_whole_unbound")
    elif len(original_whole_quantities) != 1:
        missing.append("multiple_original_whole_candidates")
    if len(subgroup_part_ids) < 2:
        missing.append("partition_subgroups_not_distinct")
    if len(linked_pairs) < 2:
        missing.append("percent_subgroup_links_incomplete")
    question_target = frame.bound_question_target
    if question_target is None or not question_target.grounded:
        missing.append("grounded_question_target")
    elif not (
        question_target.target_operator == "count"
        and question_target.target_state == "aggregate"
        and question_target.target_direction == "forward"
    ):
        if (
            question_target.target_state == "initial"
            and question_target.target_direction == "inverse"
        ):
            missing.extend(
                ("inverse_topology_unlicensed", "forward_aggregate_target_required")
            )
        else:
            missing.append("forward_aggregate_target_required")

    unresolved: set[str] = set()
    categories = {hazard.category for hazard in frame.hazards}
    if (
        any(
            item in missing
            for item in ("grounded_whole_entity", "original_whole_unbound")
        )
        and "unbound_base_quantity" in categories
    ):
        unresolved.add("unbound_base_quantity")
    if (
        any(
            item in missing
            for item in (
                "grounded_partition_subgroup",
                "percent_subgroup_links_incomplete",
            )
        )
        and "percent_change_vs_percent_of" in categories
    ):
        unresolved.add("percent_change_vs_percent_of")
    runnable = not missing and not unresolved
    return ContractAssessment(
        candidate_organ="percent_partition",
        missing_bindings=tuple(dict.fromkeys(missing)),
        unresolved_hazards=tuple(sorted(unresolved)),
        runnable=runnable,
        explanation=(
            "all percent-partition roles and the question target are grounded"
            if runnable
            else "diagnostic candidate is not runnable: "
            + ", ".join((*dict.fromkeys(missing), *sorted(unresolved)))
        ),
        evidence_spans=tuple(
            sorted(
                {
                    (span.start, span.end, span.text): span
                    for pair in linked_pairs
                    for relation in pair
                    for span in relation.evidence_spans
                }.values(),
                key=lambda span: (span.start, span.end, span.text),
            )
        )
        + (() if question_target is None else question_target.evidence_spans),
    )


def _build_fraction_decrease_payload_and_bind(span_text: str) -> tuple[np.ndarray, str] | None:
    """Construct CGA dilation versor payload and bind text for 'decrease to N/M of' evidence.

    Computes the multiplicative scaling versor using hyperbolic functions on the
    log-ratio (standard CGA encoding for dilations in the relevant plane).
    Falls back gracefully if no fraction match.
    """
    m = re.search(r"decrease to (\d+)/(\d+)\s+of", span_text)
    if not m:
        return None
    n, d = int(m.group(1)), int(m.group(2))
    k = n / d
    ln_k_half = np.log(k) / 2.0
    payload = np.zeros(32, dtype=np.float64)
    payload[0] = np.cosh(ln_k_half)
    payload[15] = -np.sinh(ln_k_half)  # appropriate bivector component for the dilation
    frac_match = re.search(r"(\d+\s*/\s*\d+)", span_text)
    bind_text = frac_match.group(1) if frac_match else span_text
    return payload, bind_text


def assess_geometric_proposals(frame: ProblemFrame) -> list[ContractAssessment]:
    assessments = []
    
    # Pre-compute reverse mapping from family_id to candidate_organ
    family_to_organ = {}
    for organ, contract in _CONTRACT_REGISTRY.items():
        if contract.family and contract.family.family_id:
            family_to_organ[contract.family.family_id] = organ

    for prop in frame.proposals:
        candidate_organ = family_to_organ.get(prop.family_id)
        if not candidate_organ:
            continue
            
        bindings = []
        for span in prop.evidence_spans:
            signature = resolve_geometric_signature(span.text)
            payload = None
            bind_text = span.text
            if signature:
                _, geom = signature
                # Per current integration for VersorBinding payload shape, use unit
                # default. Full use of geom dict and parameterized phrase support
                # in resolve_geometric_signature can be extended in followup work
                # for richer constructions.
                payload = np.zeros(32, dtype=np.float64)
                payload[0] = 1.0
            elif candidate_organ == "fraction_decrease":
                frac_result = _build_fraction_decrease_payload_and_bind(span.text)
                if frac_result:
                    payload, bind_text = frac_result

            if payload is not None:
                binding = VersorBinding(
                    source_span=(span.start, span.end),
                    semantic_identity=bind_text,
                    geometric_payload=payload,
                    versor_error=versor_condition(payload),
                )
                bindings.append(binding)
                
        runnable = False
        if bindings and all(b.versor_error < 1e-6 for b in bindings):
            runnable = True

        assessment = ContractAssessment(
            bindings=bindings,
            candidate_organ=candidate_organ,
            runnable=runnable,
            explanation="Assessed via geometric algebra.",
            evidence_spans=prop.evidence_spans
        )
        assessments.append(assessment)
        
    return assessments

def assess_contracts(frame: ProblemFrame) -> tuple[ContractAssessment, ...]:
    """Return deterministic diagnostic assessments; never admits serving.

    Dispatch order:
    1. ``quantity_entity`` — triggered by its proposal-first foundational
       family in ``frame.proposals``.  Routes to
       ``assess_quantity_entity``, which closes only exact local evidence.
       Registry key: ``_CONTRACT_REGISTRY["quantity_entity_binding"]``.
    2. ``decrease_to_fraction`` — triggered by its proposal-first catalog
       family in ``frame.proposals``.  Routes to ``assess_fraction_decrease``,
       which still determines closure from bound frame evidence.
       Registry key: ``_CONTRACT_REGISTRY["fraction_decrease"]``.
    3. ``unary_delta`` — triggered by its proposal-first catalog family in
       ``frame.proposals``. Routes to ``assess_unary_delta``, which closes
       only exact local gained/lost cue, quantity, and object evidence.
       Registry key: ``_CONTRACT_REGISTRY["unary_delta"]``.
    4. ``percent_partition`` — triggered by its proposal-first catalog family
       in ``frame.proposals``.  Routes to ``assess_percent_partition``, which
       still determines closure from bound frame evidence.
       Registry key: ``_CONTRACT_REGISTRY["percent_partition"]``.
    5. ``container_packing`` / ``labor_rate`` — inline skeleton assessments;
       not yet in the catalog registry (added to registry when obligations are
       fully specified).

    The registry provides catalog metadata for proposal traces; it does not
    replace the structural logic inside each assess_* function.  See module
    docstring for rationale.
    """
    frame_names = {candidate.name for candidate in frame.process_frames}
    results: list[ContractAssessment] = []

    # Registry-backed diagnostic families
    proposed_family_ids = {proposal.family_id for proposal in frame.proposals}
    if _QUANTITY_ENTITY_FAMILY.family_id in proposed_family_ids:
        # Catalog: _CONTRACT_REGISTRY["quantity_entity_binding"]
        results.append(assess_quantity_entity(frame))
    if _DECREASE_TO_FRACTION_FAMILY.family_id in proposed_family_ids:
        # Catalog: _CONTRACT_REGISTRY["fraction_decrease"]
        results.append(assess_fraction_decrease(frame))
    if _UNARY_DELTA_FAMILY.family_id in proposed_family_ids:
        # Catalog: _CONTRACT_REGISTRY["unary_delta_transition"]
        results.append(assess_unary_delta(frame))
    if _PERCENT_PARTITION_FAMILY.family_id in proposed_family_ids:
        # Catalog: _CONTRACT_REGISTRY["percent_partition"]
        results.append(assess_percent_partition(frame))

    # Skeleton families not yet in the catalog registry
    if "container_packing" in frame_names and frame.bound_question_target is not None:
        roles = _roles(frame, "container_packing")
        missing = tuple(
            name for name in ("container", "content", "count_per") if name not in roles
        )
        results.append(
            ContractAssessment(
                "nested_fraction_remainder_total",
                missing,
                (),
                not missing,
                "container contract grounded"
                if not missing
                else "missing container bindings: " + ", ".join(missing),
                _evidence(frame, "container_packing"),
            )
        )
    if "labor_rate" in frame_names:
        roles = _roles(frame, "labor_rate")
        missing = tuple(
            name for name in ("worker", "rate", "duration") if name not in roles
        )
        results.append(
            ContractAssessment(
                "temporal_tariff",
                missing,
                (),
                not missing,
                "temporal tariff contract grounded"
                if not missing
                else "missing tariff bindings: " + ", ".join(missing),
                _evidence(frame, "labor_rate"),
            )
        )
    return tuple(sorted(results, key=lambda item: item.candidate_organ))


def recommended_migration_target(assessments: tuple[ContractAssessment, ...]) -> str:
    runnable = [item.candidate_organ for item in assessments if item.runnable]
    if runnable:
        return sorted(runnable)[0]
    if assessments:
        best = min(
            assessments,
            key=lambda item: (
                len(item.missing_bindings) + len(item.unresolved_hazards),
                item.candidate_organ,
            ),
        )
        return f"substrate:contract_gap:{best.candidate_organ}"
    return "substrate:problem_frame_builder"