fix(kernel): harden quantity-entity proposal seam

This commit is contained in:
Shay 2026-06-20 21:37:57 -07:00
parent b5c4a39b56
commit c28581c7d0
4 changed files with 76 additions and 63 deletions

View file

@ -24,8 +24,8 @@ Design doctrine (from ADR-0223):
""" """
from __future__ import annotations from __future__ import annotations
from dataclasses import dataclass, replace from dataclasses import dataclass
from typing import TYPE_CHECKING from typing import TYPE_CHECKING, ClassVar, Literal
if TYPE_CHECKING: if TYPE_CHECKING:
from generate.kernel_facts import SourceSpan from generate.kernel_facts import SourceSpan
@ -170,11 +170,8 @@ class ConstructionProposal:
relation_type: The ``BoundRelation.relation_type`` that was searched. relation_type: The ``BoundRelation.relation_type`` that was searched.
candidate_organ: The organ this proposal targets. candidate_organ: The organ this proposal targets.
evidence_spans: Source spans that motivated the proposal. evidence_spans: Source spans that motivated the proposal.
status: One of ``"proposed"``, ``"partial"``, ``"closed"``, status: Always ``"proposed"``. Runnable/refused authority
``"refused"``. ``"closed"`` means the contract belongs exclusively to ``ContractAssessment``.
assessment was runnable.
missing_roles: Role obligation names that were absent at assessment time.
active_hazards: Hazard categories that were unresolved at assessment time.
role_obligations: Catalog-declared roles that downstream binding and role_obligations: Catalog-declared roles that downstream binding and
assessment must ground or explicitly refuse. assessment must ground or explicitly refuse.
diagnostic_only: True for every proposal in the current catalog. diagnostic_only: True for every proposal in the current catalog.
@ -185,28 +182,36 @@ class ConstructionProposal:
relation_type: str relation_type: str
candidate_organ: str candidate_organ: str
evidence_spans: tuple[SourceSpan, ...] evidence_spans: tuple[SourceSpan, ...]
status: str status: Literal["proposed"]
missing_roles: tuple[str, ...]
active_hazards: tuple[str, ...]
role_obligations: tuple[RoleObligation, ...] = () role_obligations: tuple[RoleObligation, ...] = ()
diagnostic_only: bool = True diagnostic_only: bool = True
serving_allowed: bool = False serving_allowed: bool = False
_VALID_STATUSES: frozenset[str] = frozenset({ _VALID_STATUS: ClassVar[Literal["proposed"]] = "proposed"
"proposed", "partial", "closed", "refused"
})
def __post_init__(self) -> None: def __post_init__(self) -> None:
if self.status not in self._VALID_STATUSES: if self.status != self._VALID_STATUS:
raise ValueError( raise ValueError(
f"ConstructionProposal.status must be one of " "ConstructionProposal.status must remain 'proposed'; "
f"{sorted(self._VALID_STATUSES)}, got {self.status!r}" "ContractAssessment is the sole runnable/refused authority"
) )
if not self.diagnostic_only or self.serving_allowed: if not self.diagnostic_only or self.serving_allowed:
raise ValueError( raise ValueError(
"ConstructionProposal must remain diagnostic-only and serving-disallowed" "ConstructionProposal must remain diagnostic-only and serving-disallowed"
) )
@property
def missing_roles(self) -> tuple[str, ...]:
"""Compatibility view; assessment blockers are not proposal state."""
return ()
@property
def active_hazards(self) -> tuple[str, ...]:
"""Compatibility view; assessment hazards are not proposal state."""
return ()
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# Catalog entries # Catalog entries
@ -527,8 +532,6 @@ def propose_construction(
candidate_organ=family.signature.candidate_organ, candidate_organ=family.signature.candidate_organ,
evidence_spans=evidence_spans, evidence_spans=evidence_spans,
status="proposed", status="proposed",
missing_roles=(),
active_hazards=(),
role_obligations=( role_obligations=(
*family.signature.required_roles, *family.signature.required_roles,
*family.signature.optional_roles, *family.signature.optional_roles,
@ -545,11 +548,11 @@ def make_proposal(
missing_roles: tuple[str, ...], missing_roles: tuple[str, ...],
active_hazards: tuple[str, ...], active_hazards: tuple[str, ...],
) -> ConstructionProposal: ) -> ConstructionProposal:
"""Map assessment evidence onto a proposal for legacy catalog paths. """Reject assessment-backed proposal synthesis.
Migrated proposal-first families must enter through Every catalog family is proposal-first. The assessment-shaped signature
:func:`propose_construction`. This adapter remains only for explicitly remains temporarily for callers that need a loud migration failure, but no
unmigrated catalog paths that still synthesize proposals from assessments. ``ConstructionProposal`` may encode assessment output.
Args: Args:
family_id: Catalog family identifier. family_id: Catalog family identifier.
@ -559,35 +562,14 @@ def make_proposal(
missing_roles: ContractAssessment.missing_bindings contents. missing_roles: ContractAssessment.missing_bindings contents.
active_hazards: ContractAssessment.unresolved_hazards contents. active_hazards: ContractAssessment.unresolved_hazards contents.
Returns:
A ConstructionProposal with status derived from the assessment:
- ``"closed"`` if runnable and no missing roles/hazards;
- ``"partial"`` if some roles are bound but closure is incomplete;
- ``"refused"`` if active hazards block closure;
- ``"proposed"`` otherwise (construction proposed but not evaluated).
Raises: Raises:
KeyError: If *family_id* is not registered in the catalog. KeyError: If *family_id* is not registered in the catalog.
ValueError: If *family_id* has already migrated to the proposal-first seam. ValueError: For every catalog family; use ``propose_construction``
before assessment instead.
""" """
if family_id in _PROPOSAL_FIRST_FAMILIES: _ = (evidence_spans, assessment_runnable, missing_roles, active_hazards)
if family_id not in _CATALOG:
raise KeyError(family_id)
raise ValueError( raise ValueError(
f"{family_id} is proposal-first; use propose_construction before assessment" f"{family_id} is proposal-first; use propose_construction before assessment"
) )
proposal = propose_construction(family_id, evidence_spans)
if assessment_runnable:
status = "closed"
elif active_hazards:
status = "refused"
elif missing_roles:
status = "partial"
else:
status = "proposed"
return replace(
proposal,
status=status,
missing_roles=missing_roles,
active_hazards=active_hazards,
)

View file

@ -562,8 +562,18 @@ def _extract_bindings(
def _quantity_kind_dispositions( def _quantity_kind_dispositions(
mentions: tuple[GroundedMention, ...], mentions: tuple[GroundedMention, ...],
bindings: tuple[MentionBinding, ...], bindings: tuple[MentionBinding, ...],
proposals: tuple[ConstructionProposal, ...],
) -> tuple[QuantityKindDisposition, ...]: ) -> tuple[QuantityKindDisposition, ...]:
"""Close count/measurement kind only from existing local bindings.""" """Close kind only for the exact proposal-backed local binding."""
quantity_entity_proposals = tuple(
proposal
for proposal in proposals
if proposal.family_id == "binding.quantity_entity"
)
if len(quantity_entity_proposals) != 1:
return ()
proposal = quantity_entity_proposals[0]
mentions_by_id = {mention.mention_id: mention for mention in mentions} mentions_by_id = {mention.mention_id: mention for mention in mentions}
unit_bindings: dict[str, list[MentionBinding]] = {} unit_bindings: dict[str, list[MentionBinding]] = {}
@ -579,6 +589,11 @@ def _quantity_kind_dispositions(
entity = mentions_by_id.get(binding.target_mention_id) entity = mentions_by_id.get(binding.target_mention_id)
if quantity is None or entity is None or quantity.fact_id is None: if quantity is None or entity is None or quantity.fact_id is None:
continue continue
if not any(
cue.start <= quantity.span.start and entity.span.end <= cue.end
for cue in proposal.evidence_spans
):
continue
bound_units = unit_bindings.get(quantity.mention_id, []) bound_units = unit_bindings.get(quantity.mention_id, [])
if not bound_units: if not bound_units:
@ -917,11 +932,12 @@ def build_problem_frame(problem_text: str) -> ProblemFrame:
builder.add_proposal(proposal) builder.add_proposal(proposal)
for proposal in _percent_partition_proposals(problem_text, frames): for proposal in _percent_partition_proposals(problem_text, frames):
builder.add_proposal(proposal) builder.add_proposal(proposal)
for proposal in _quantity_entity_proposals( quantity_entity_proposals = _quantity_entity_proposals(
problem_text, problem_text,
tuple(grounded_quantities), tuple(grounded_quantities),
frames, frames,
): )
for proposal in quantity_entity_proposals:
builder.add_proposal(proposal) builder.add_proposal(proposal)
mentions = _extract_mentions(problem_text, tuple(grounded_quantities), units) mentions = _extract_mentions(problem_text, tuple(grounded_quantities), units)
@ -934,7 +950,11 @@ def build_problem_frame(problem_text: str) -> ProblemFrame:
builder.add_object(mention.surface) builder.add_object(mention.surface)
for binding in bindings: for binding in bindings:
builder.add_binding(binding) builder.add_binding(binding)
for disposition in _quantity_kind_dispositions(mentions, bindings): for disposition in _quantity_kind_dispositions(
mentions,
bindings,
quantity_entity_proposals,
):
builder.add_quantity_kind_disposition(disposition) builder.add_quantity_kind_disposition(disposition)
for relation in _bound_relations(problem_text, mentions, bindings): for relation in _bound_relations(problem_text, mentions, bindings):
builder.add_bound_relation(relation) builder.add_bound_relation(relation)

View file

@ -138,25 +138,35 @@ def test_make_proposal_still_rejects_unknown_family_ids() -> None:
propose_construction("invalid_family_id", (span,)) propose_construction("invalid_family_id", (span,))
def test_construction_proposal_status_validation() -> None: def test_construction_proposal_cannot_carry_assessment_authority() -> None:
span = SourceSpan("test text", 0, 9) span = SourceSpan("test text", 0, 9)
ConstructionProposal( ConstructionProposal(
family_id="proportional_change.decrease_to_fraction", family_id="proportional_change.decrease_to_fraction",
relation_type="decrease_to_fraction", relation_type="decrease_to_fraction",
candidate_organ="fraction_decrease", candidate_organ="fraction_decrease",
evidence_spans=(span,), evidence_spans=(span,),
status="closed", status="proposed",
missing_roles=(),
active_hazards=(),
) )
with pytest.raises(ValueError, match="status must be one of"): for status in ("partial", "closed", "refused"):
with pytest.raises(ValueError, match="must remain 'proposed'"):
ConstructionProposal( ConstructionProposal(
family_id="proportional_change.decrease_to_fraction", family_id="proportional_change.decrease_to_fraction",
relation_type="decrease_to_fraction", relation_type="decrease_to_fraction",
candidate_organ="fraction_decrease", candidate_organ="fraction_decrease",
evidence_spans=(span,), evidence_spans=(span,),
status="invalid_status", status=status, # type: ignore[arg-type]
missing_roles=(), )
active_hazards=(),
assert "missing_roles" not in ConstructionProposal.__dataclass_fields__
assert "active_hazards" not in ConstructionProposal.__dataclass_fields__
with pytest.raises(TypeError, match="unexpected keyword argument 'missing_roles'"):
ConstructionProposal(
family_id="proportional_change.decrease_to_fraction",
relation_type="decrease_to_fraction",
candidate_organ="fraction_decrease",
evidence_spans=(span,),
status="proposed",
missing_roles=("base_quantity_unbound",),
) )

View file

@ -120,6 +120,7 @@ def test_confusers_do_not_dispatch_quantity_entity(problem_text: str) -> None:
frame = build_problem_frame(problem_text) frame = build_problem_frame(problem_text)
assert FAMILY_ID not in {proposal.family_id for proposal in frame.proposals} assert FAMILY_ID not in {proposal.family_id for proposal in frame.proposals}
assert frame.quantity_kind_dispositions == ()
assert CANDIDATE_ORGAN not in { assert CANDIDATE_ORGAN not in {
assessment.candidate_organ assessment.candidate_organ
for assessment in assess_contracts(frame) for assessment in assess_contracts(frame)