"""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 from dataclasses import dataclass from generate.construction_affordances import ( ConstructionContract, _DECREASE_TO_FRACTION_FAMILY, _PERCENT_PARTITION_FAMILY, _QUANTITY_ENTITY_FAMILY, ) from generate.kernel_facts import BoundRelation, GroundedMention, SourceSpan from generate.problem_frame import ProblemFrame # --------------------------------------------------------------------------- # 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", ), } 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 ContractAssessment: candidate_organ: str missing_bindings: tuple[str, ...] unresolved_hazards: tuple[str, ...] runnable: bool explanation: str evidence_spans: tuple[SourceSpan, ...] 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 _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_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 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. ``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"]``. 4. ``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 _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"