core-labs/core
1
0
Fork 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

Merge pull request #855 from AssetOverflow/feat/kernel-unary-delta-proposal-seam

Browse source 
feat(kernel): introduce diagnostic unary-delta proposal seam
This commit is contained in:
Shay 2026-06-21 05:57:58 -07:00 committed by GitHub
commit a5281df562
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
8 changed files with 714 additions and 17 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

70
generate/construction_affordances.py
View file

@ -458,11 +458,80 @@ _QUANTITY_ENTITY_FAMILY = ConstructionFamily(
serving_allowed=False,
)
_UNARY_DELTA_FAMILY = ConstructionFamily(
family_id="state_change.unary_delta",
display_name="Unary gained/lost delta",
signature=ConstructionSignature(
relation_type="unary_delta",
candidate_organ="unary_delta",
required_roles=(
RoleObligation(
"action_cue",
required=True,
description="Exact local gained/lost cue span from the original text.",
),
RoleObligation(
"delta_quantity",
required=True,
description="Exactly one grounded scalar quantity attached to the event.",
),
RoleObligation(
"changed_object",
required=True,
description="Exactly one grounded local object whose quantity changed.",
),
RoleObligation(
"direction",
required=True,
description="Increase for gained and decrease for lost.",
),
),
optional_roles=(),
),
hazards=(
ConstructionHazard(
hazard_category="quantity_entity_nonlocal",
blocking=True,
description=(
"Pronoun repair or cross-sentence binding would be required to "
"decide the changed object."
),
),
ConstructionHazard(
hazard_category="percent_change_vs_percent_of",
blocking=True,
description=(
"Percent/rate/comparison phrasing would widen this slice beyond "
"a local unary gained/lost event."
),
),
),
target_semantics=(
"authority:diagnostic_only",
"cue_inventory:gained|lost",
"locality:single_clause",
),
contract_labels=(
"unary_delta_proposal_required",
"unary_delta_relation_ambiguous",
"action_cue_unbound",
"delta_quantity_unbound",
"changed_object_unbound",
"direction_unbound",
"quantity_kind_unresolved",
"provenance_span_inexact",
"quantity_entity_nonlocal",
),
diagnostic_only=True,
serving_allowed=False,
)
# The catalog. Keys are family_id strings. Sorted for deterministic iteration.
_CATALOG: dict[str, ConstructionFamily] = {
_DECREASE_TO_FRACTION_FAMILY.family_id: _DECREASE_TO_FRACTION_FAMILY,
_PERCENT_PARTITION_FAMILY.family_id: _PERCENT_PARTITION_FAMILY,
_QUANTITY_ENTITY_FAMILY.family_id: _QUANTITY_ENTITY_FAMILY,
_UNARY_DELTA_FAMILY.family_id: _UNARY_DELTA_FAMILY,
}
# Secondary indices for O(1) lookup by organ or relation type.
@ -479,6 +548,7 @@ _PROPOSAL_FIRST_FAMILIES: frozenset[str] = frozenset({
"binding.quantity_entity",
"proportional_change.decrease_to_fraction",
"partition.percent_partition",
"state_change.unary_delta",
})

98
generate/foundational_families.py
View file

@ -6,9 +6,9 @@ documents, recognize constructions, assess ``ProblemFrame`` contracts, or route
proposals. Authorization fields mirror reviewed implementation boundaries;
they never authorize serving.
The current registry deliberately contains only the two specifications approved
by the initial family-spec gate. General substrate/CGA retrieval for these
families is not live.
The current registry deliberately contains only the reviewed foundational
families approved by the family-spec gate. General substrate/CGA retrieval for
these families is not live.
"""
from __future__ import annotations
@ -249,7 +249,97 @@ _STATE_CHANGE = FoundationalFamilySpec(
)
_FAMILIES = (_QUANTITY_ENTITY_BINDING, _STATE_CHANGE)
_UNARY_DELTA = FoundationalFamilySpec(
family_id="state_change.unary_delta",
display_name="Unary Delta State Change",
status="Implemented (diagnostic-only; non-serving)",
related_adrs=("ADR-0223", "ADR-0224"),
domains=(
"arithmetic_quantitative",
"physical_science",
"life_science",
"reading_comprehension",
),
summary=(
"Represents one exact local gained/lost event as a unary quantity "
"delta over a changed object, without asserting owner, before-state, "
"after-state, or arithmetic closure."
),
surface_chunk_patterns=(
"<subject> gained <number> <object>",
"<subject> lost <number> <object>",
),
semantic_neighborhood=(
"inventory_change",
"resource_delta",
"count_change",
"local_event_delta",
),
construction_signatures=("unary_delta",),
required_roles=("action_cue", "delta_quantity", "changed_object", "direction"),
optional_roles=("subject_surface",),
hazards_confusers=(
"PF-EN-002 quantity_entity_unbound",
"PF-EN-005 role_alias_collision",
"PF-HZ-003 hazard_ignored_by_contract",
"PF-TG-004 target_direction_unknown",
),
frame_representation=(
"ConstructionProposal(status='proposed', family_id='state_change.unary_delta')",
"BoundRelation(relation_type='unary_delta') with action_cue, delta_quantity, changed_object, and direction roles",
"Exact SourceSpan evidence for cue, quantity, and object",
"ContractAssessment(candidate_organ='unary_delta') as the sole runnable/refused authority",
),
contract_readiness_criteria=(
"Exactly one local gained/lost cue is grounded from the original text.",
"Exactly one grounded scalar quantity and one grounded changed object are bound.",
"Quantity/object grounding composes with exact local quantity_entity evidence.",
"Direction is increase for gained and decrease for lost.",
"No transfer, containment, comparison, rate, percent, passive, modal, negated, or cross-sentence reasoning is required.",
),
verification_style=(
"Cue substitution between gained and lost must flip direction without widening other roles.",
"Synthetic or widened cue/quantity/object spans must refuse rather than normalize.",
),
refusal_conditions=(
"Cue, quantity, or changed object is missing or ambiguous.",
"Object grounding would require pronoun repair, cross-sentence binding, or legacy semantic-state logic.",
),
cross_domain_evidence=(
FoundationalDomainEvidence(
domain="physical_science",
example="The jar lost 2 cookies.",
expected_roles=("action_cue", "delta_quantity", "changed_object", "direction"),
),
FoundationalDomainEvidence(
domain="life_science",
example="The sapling gained 3 leaves.",
expected_roles=("action_cue", "delta_quantity", "changed_object", "direction"),
),
FoundationalDomainEvidence(
domain="reading_comprehension",
example="Ana gained 3 marbles.",
expected_roles=("action_cue", "delta_quantity", "changed_object", "direction"),
),
),
current_state=(
"Implemented only as a bounded diagnostic proposal-first gained/lost "
"relation. No transfer semantics, owner assertion, state ledger, or "
"serving adapter is authorized."
),
target_state=(
"Proposal-first unary-delta relation with exact cue/quantity/object grounding, "
"typed refusal, and cross-domain evidence."
),
serving_status="Diagnostic-only implementation / not serving.",
serving_allowed=False,
implementation_authorized=True,
primary_relation_type="unary_delta",
future_adapter="unary_delta_adapter",
)
_FAMILIES = (_QUANTITY_ENTITY_BINDING, _STATE_CHANGE, _UNARY_DELTA)
_BY_ID = MappingProxyType({family.family_id: family for family in _FAMILIES})
_BY_RELATION_TYPE = MappingProxyType(
{family.primary_relation_type: family for family in _FAMILIES}

134
generate/problem_frame_builder.py
View file

@ -334,6 +334,12 @@ _DECREASE_DELTA_QUESTION_RE = re.compile(
r"\bwhat\s+will\s+the\s+(?P<entity>[A-Za-z][A-Za-z'-]*)\s+decrease\s+by\??",
re.IGNORECASE,
)
_UNARY_DELTA_RE = re.compile(
r"\b(?P<subject>(?:[A-Z][A-Za-z'-]*|[Tt]he\s+[A-Za-z][A-Za-z'-]*))\s+"
r"(?P<cue>gained|lost)\s+"
r"(?P<quantity>\d+(?:\.\d+)?)\s+"
r"(?P<object>[A-Za-z][A-Za-z'-]*)\b"
)
_ACTOR_VERB_RE = re.compile(
r"\b(?P<actor>[A-Z][A-Za-z'-]*)\s+"
r"(?:gave|gives|give|received|receives|spent|spends|ate|eats|bought|buys|sold|sells)\b"
@ -416,6 +422,17 @@ def _has_list_or_enumeration_suffix(text: str, end: int) -> bool:
return tail.startswith((",", ";", "and ", "or "))
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 _quantity_entity_proposals(
text: str,
quantities: tuple[GroundedScalar, ...],
@ -462,6 +479,42 @@ def _quantity_entity_proposals(
return (propose_construction("binding.quantity_entity", (evidence,)),)
def _unary_delta_proposals(
text: str,
) -> tuple[ConstructionProposal, ...]:
"""Propose the narrow gained/lost unary-delta slice from exact local cues."""
if any(
surface_in_text(surface, text)
for surface in _QUANTITY_ENTITY_CONFUSER_SURFACES
):
return ()
matches = tuple(_UNARY_DELTA_RE.finditer(text))
if len(matches) != 1:
return ()
match = matches[0]
if match.group("subject").lower() in _QUANTITY_ENTITY_PRONOUNS:
return ()
if _has_list_or_enumeration_suffix(text, match.end("object")):
return ()
sentence_ends = tuple(
index
for marker in ".!?"
if (index := text.find(marker, match.end("object"))) != -1
)
sentence_end = min(sentence_ends, default=len(text))
trailing = text[match.end("object"):sentence_end].strip()
if trailing:
return ()
evidence = SourceSpan(
text[match.start("cue"):match.end("cue")],
match.start("cue"),
match.end("cue"),
)
return (propose_construction("state_change.unary_delta", (evidence,)),)
def _extract_mentions(
text: str,
quantities: tuple[GroundedScalar, ...],
@ -560,6 +613,7 @@ def _extract_bindings(
def _quantity_kind_dispositions(
text: str,
mentions: tuple[GroundedMention, ...],
bindings: tuple[MentionBinding, ...],
proposals: tuple[ConstructionProposal, ...],
@ -571,9 +625,15 @@ def _quantity_kind_dispositions(
for proposal in proposals
if proposal.family_id == "binding.quantity_entity"
)
if len(quantity_entity_proposals) != 1:
unary_delta_proposals = tuple(
proposal
for proposal in proposals
if proposal.family_id == "state_change.unary_delta"
)
if len(quantity_entity_proposals) + len(unary_delta_proposals) != 1:
return ()
proposal = quantity_entity_proposals[0]
quantity_entity_proposal = quantity_entity_proposals[0] if quantity_entity_proposals else None
unary_delta_proposal = unary_delta_proposals[0] if unary_delta_proposals else None
mentions_by_id = {mention.mention_id: mention for mention in mentions}
unit_bindings: dict[str, list[MentionBinding]] = {}
@ -589,11 +649,16 @@ def _quantity_kind_dispositions(
entity = mentions_by_id.get(binding.target_mention_id)
if quantity is None or entity is None or quantity.fact_id is None:
continue
if not any(
cue.start <= quantity.span.start and entity.span.end <= cue.end
for cue in proposal.evidence_spans
):
continue
if quantity_entity_proposal is not None:
if not any(
cue.start <= quantity.span.start and entity.span.end <= cue.end
for cue in quantity_entity_proposal.evidence_spans
):
continue
elif unary_delta_proposal is not None:
cue = unary_delta_proposal.evidence_spans[0]
if quantity.span.start < cue.end or not _spans_are_local(text, cue, entity.span):
continue
bound_units = unit_bindings.get(quantity.mention_id, [])
if not bound_units:
@ -631,6 +696,7 @@ def _bound_relations(
text: str,
mentions: tuple[GroundedMention, ...],
bindings: tuple[MentionBinding, ...],
proposals: tuple[ConstructionProposal, ...],
) -> tuple[BoundRelation, ...]:
by_id = {m.mention_id: m for m in mentions}
relations: list[BoundRelation] = []
@ -699,6 +765,47 @@ def _bound_relations(
evidence_spans=(quantity.span, canonical_whole.span, part.span),
))
unary_delta_proposals = tuple(
proposal
for proposal in proposals
if proposal.family_id == "state_change.unary_delta"
)
if len(unary_delta_proposals) == 1:
unary_match = _UNARY_DELTA_RE.search(text)
proposal = unary_delta_proposals[0]
if unary_match is not None and len(proposal.evidence_spans) == 1:
cue_span = proposal.evidence_spans[0]
cue_surface = text[cue_span.start:cue_span.end]
if cue_span.text == cue_surface and cue_surface in {"gained", "lost"}:
direction = "increase" if cue_surface == "gained" else "decrease"
matching_bindings = [
binding
for binding in quantity_entity
if by_id[binding.source_mention_id].span.start == unary_match.start("quantity")
and by_id[binding.target_mention_id].span.start == unary_match.start("object")
]
if len(matching_bindings) == 1:
binding = matching_bindings[0]
quantity = by_id[binding.source_mention_id]
obj = by_id[binding.target_mention_id]
roles = (
BoundRole(
"action_cue",
f"span:{cue_span.start}:{cue_span.end}",
"span",
(cue_span,),
),
BoundRole("delta_quantity", quantity.mention_id, quantity.kind, (quantity.span,)),
BoundRole("changed_object", obj.mention_id, obj.kind, (obj.span,)),
BoundRole("direction", direction, "direction", (cue_span,)),
)
relations.append(BoundRelation(
relation_id="",
relation_type="unary_delta",
roles=roles,
evidence_spans=(cue_span, quantity.span, obj.span),
))
decrease_matches = list(_DECREASE_TO_FRACTION_RE.finditer(text))
if len(decrease_matches) == 1:
match = decrease_matches[0]
@ -939,6 +1046,9 @@ def build_problem_frame(problem_text: str) -> ProblemFrame:
)
for proposal in quantity_entity_proposals:
builder.add_proposal(proposal)
unary_delta_proposals = _unary_delta_proposals(problem_text)
for proposal in unary_delta_proposals:
builder.add_proposal(proposal)
mentions = _extract_mentions(problem_text, tuple(grounded_quantities), units)
bindings = _extract_bindings(problem_text, mentions)
@ -951,12 +1061,18 @@ def build_problem_frame(problem_text: str) -> ProblemFrame:
for binding in bindings:
builder.add_binding(binding)
for disposition in _quantity_kind_dispositions(
problem_text,
mentions,
bindings,
quantity_entity_proposals,
(*quantity_entity_proposals, *unary_delta_proposals),
):
builder.add_quantity_kind_disposition(disposition)
for relation in _bound_relations(problem_text, mentions, bindings):
for relation in _bound_relations(
problem_text,
mentions,
bindings,
(*quantity_entity_proposals, *unary_delta_proposals),
):
builder.add_bound_relation(relation)
bound_target = _bound_question_target(problem_text, mentions)
if bound_target is not None:

130
generate/problem_frame_contracts.py
View file

@ -17,6 +17,7 @@ from generate.construction_affordances import (
_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
@ -53,6 +54,10 @@ _CONTRACT_REGISTRY: dict[str, ConstructionContract] = {
family=_QUANTITY_ENTITY_FAMILY,
assess_fn_name="assess_quantity_entity",
),
"unary_delta": ConstructionContract(
family=_UNARY_DELTA_FAMILY,
assess_fn_name="assess_unary_delta",
),
}
@ -99,6 +104,11 @@ 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}
@ -448,6 +458,115 @@ def assess_fraction_decrease(frame: ProblemFrame) -> ContractAssessment:
)
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)
dispositions = tuple(frame.quantity_kind_dispositions)
missing: list[str] = []
unresolved: set[str] = set()
if len(proposals) != 1:
missing.append("unary_delta_proposal_required")
proposal = proposals[0] if len(proposals) == 1 else None
if len(relations) != 1:
missing.append("unary_delta_relation_ambiguous")
relation = relations[0] if len(relations) == 1 else None
cue_spans = _role_spans(relation, "action_cue") if relation is not None else ()
quantity_id = _role_target(relation, "delta_quantity") if relation is not None else None
object_id = _role_target(relation, "changed_object") if relation is not None else None
direction = _role_target(relation, "direction") if relation is not None else None
if len(cue_spans) != 1:
missing.append("action_cue_unbound")
cue_span = cue_spans[0] if len(cue_spans) == 1 else None
if cue_span is not None and cue_span.text not in {"gained", "lost"}:
missing.append("action_cue_unbound")
quantity = mentions.get(quantity_id) if quantity_id is not None else None
if quantity is None or quantity.kind != "quantity":
missing.append("delta_quantity_unbound")
changed_object = mentions.get(object_id) if object_id is not None else None
if changed_object is None or changed_object.kind != "object":
missing.append("changed_object_unbound")
expected_direction = None
if cue_span is not None:
expected_direction = "increase" if cue_span.text == "gained" else "decrease"
if direction is None or direction != expected_direction:
missing.append("direction_unbound")
if quantity is not None and changed_object is not None:
matching_dispositions = tuple(
disposition
for disposition in dispositions
if disposition.quantity_mention_id == quantity.mention_id
and disposition.entity_mention_id == changed_object.mention_id
)
if len(matching_dispositions) != 1:
missing.append("quantity_kind_unresolved")
exact_evidence = _unique_evidence(tuple(
span
for group in (
(() if proposal is None else proposal.evidence_spans),
(() if cue_span 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
))
if proposal is not None and cue_span is not None and proposal.evidence_spans != (cue_span,):
missing.append("provenance_span_inexact")
if relation is not None and quantity is not None and changed_object is not None and cue_span 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")
if cue_span 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_span 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")
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",
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)
@ -547,11 +666,15 @@ def assess_contracts(frame: ProblemFrame) -> tuple[ContractAssessment, ...]:
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
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"]``.
4. ``container_packing`` / ``labor_rate`` — inline skeleton assessments;
5. ``container_packing`` / ``labor_rate`` — inline skeleton assessments;
not yet in the catalog registry (added to registry when obligations are
fully specified).
@ -570,6 +693,9 @@ def assess_contracts(frame: ProblemFrame) -> tuple[ContractAssessment, ...]:
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"]
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))

15
tests/test_construction_affordances.py
View file

@ -16,7 +16,7 @@ from generate.kernel_facts import SourceSpan
def test_catalog_entries_are_diagnostic_only_and_serving_forbidden() -> None:
families = all_diagnostic_families()
assert len(families) == 3
assert len(families) == 4
for family in families:
assert family.diagnostic_only is True
assert family.serving_allowed is False
@ -30,6 +30,7 @@ def test_catalog_ordering_is_deterministic() -> None:
"binding.quantity_entity",
"partition.percent_partition",
"proportional_change.decrease_to_fraction",
"state_change.unary_delta",
]
@ -49,6 +50,11 @@ def test_lookups_correctness() -> None:
assert lookup_by_organ("percent_partition") is partition_family
assert lookup_by_relation_type("percent_of") is partition_family
unary_delta = lookup_family("state_change.unary_delta")
assert unary_delta is not None
assert lookup_by_organ("unary_delta") is unary_delta
assert lookup_by_relation_type("unary_delta") is unary_delta
assert lookup_family("invalid_family_id") is None
assert lookup_by_organ("invalid_organ") is None
assert lookup_by_relation_type("invalid_relation") is None
@ -81,6 +87,12 @@ def test_lookups_correctness() -> None:
"percent_partition",
{"whole", "part", "scale"},
),
(
"state_change.unary_delta",
"unary_delta",
"unary_delta",
{"action_cue", "delta_quantity", "changed_object", "direction"},
),
),
)
def test_propose_construction_is_preassessment_and_catalog_backed(
@ -111,6 +123,7 @@ def test_propose_construction_is_preassessment_and_catalog_backed(
"binding.quantity_entity",
"proportional_change.decrease_to_fraction",
"partition.percent_partition",
"state_change.unary_delta",
),
)
def test_make_proposal_rejects_migrated_proposal_first_families(

11
tests/test_construction_proposal_seam.py
View file

@ -36,6 +36,7 @@ PERCENT_PARTITION_CASE = (
)
QUANTITY_ENTITY_CASE = "A school has 100 students."
UNARY_DELTA_CASE = "Ana gained 3 marbles."
MIGRATED_CASES = (
(
@ -53,6 +54,11 @@ MIGRATED_CASES = (
"partition.percent_partition",
"percent_partition",
),
(
UNARY_DELTA_CASE,
"state_change.unary_delta",
"unary_delta",
),
)
@ -202,6 +208,11 @@ def test_migrated_families_bypass_legacy_make_proposal(
"partition.percent_partition",
("20% of", "10% of"),
),
(
UNARY_DELTA_CASE,
"state_change.unary_delta",
("gained",),
),
),
)
def test_migrated_proposals_contain_only_exact_motivating_surface_evidence(

10
tests/test_foundational_families.py
View file

@ -19,6 +19,7 @@ from generate.foundational_families import (
EXPECTED_FAMILY_IDS = (
"binding.quantity_entity",
"state_change.transition",
"state_change.unary_delta",
)
@ -35,10 +36,12 @@ def test_registry_contains_only_the_two_approved_specs_in_deterministic_order()
def test_public_family_accessors() -> None:
quantity = get_foundational_family("binding.quantity_entity")
state_change = require_foundational_family("state_change.transition")
unary_delta = require_foundational_family("state_change.unary_delta")
assert quantity is not None
assert quantity.family_id == "binding.quantity_entity"
assert state_change.family_id == "state_change.transition"
assert unary_delta.family_id == "state_change.unary_delta"
assert get_foundational_family("missing.family") is None
with pytest.raises(KeyError, match="missing.family"):
@ -48,9 +51,11 @@ def test_public_family_accessors() -> None:
def test_lookup_by_primary_relation_type() -> None:
quantity = get_foundational_family_by_relation_type("quantity_entity")
state_change = get_foundational_family_by_relation_type("state_change")
unary_delta = get_foundational_family_by_relation_type("unary_delta")
assert quantity is get_foundational_family("binding.quantity_entity")
assert state_change is get_foundational_family("state_change.transition")
assert unary_delta is get_foundational_family("state_change.unary_delta")
assert get_foundational_family_by_relation_type("missing_relation") is None
@ -63,7 +68,7 @@ def test_registry_keys_are_unique() -> None:
assert len(relation_types) == len(set(relation_types))
def test_specs_are_frozen_and_only_quantity_entity_is_authorized() -> None:
def test_specs_are_frozen_and_only_authorized_slices_are_implemented() -> None:
for family in iter_foundational_families():
assert isinstance(family, FoundationalFamilySpec)
assert family.serving_allowed is False
@ -78,6 +83,9 @@ def test_specs_are_frozen_and_only_quantity_entity_is_authorized() -> None:
assert require_foundational_family(
"state_change.transition"
).implementation_authorized is False
assert require_foundational_family(
"state_change.unary_delta"
).implementation_authorized is True
def test_required_adr_0224_fields_are_populated() -> None:

263
tests/test_unary_delta_proposal.py Normal file
View file

@ -0,0 +1,263 @@
"""Diagnostic-only contract tests for ``state_change.unary_delta``."""
from __future__ import annotations
import ast
import dataclasses
from pathlib import Path
import generate.problem_frame_contracts as problem_frame_contracts
import pytest
from generate.kernel_facts import SourceSpan
from generate.problem_frame_builder import build_problem_frame
from generate.problem_frame_contracts import assess_contracts, assess_unary_delta
FAMILY_ID = "state_change.unary_delta"
CANDIDATE_ORGAN = "unary_delta"
def _proposal(frame):
proposals = tuple(
proposal
for proposal in frame.proposals
if proposal.family_id == FAMILY_ID
)
assert len(proposals) == 1
return proposals[0]
def _relation(frame):
relations = tuple(
relation
for relation in frame.bound_relations
if relation.relation_type == "unary_delta"
)
assert len(relations) == 1
return relations[0]
def _assessment(frame):
assessments = tuple(
assessment
for assessment in assess_contracts(frame)
if assessment.candidate_organ == CANDIDATE_ORGAN
)
assert len(assessments) == 1
return assessments[0]
@pytest.mark.parametrize(
("problem_text", "cue", "direction"),
(
("Ana gained 3 marbles.", "gained", "increase"),
("The jar lost 2 cookies.", "lost", "decrease"),
),
)
def test_exact_local_gained_lost_event_is_proposed_bound_and_assessed(
problem_text: str,
cue: str,
direction: str,
) -> None:
frame = build_problem_frame(problem_text)
proposal = _proposal(frame)
relation = _relation(frame)
assessment = _assessment(frame)
assert proposal.status == "proposed"
assert proposal.diagnostic_only is True
assert proposal.serving_allowed is False
assert not hasattr(proposal, "runnable")
roles = {role.role: role for role in relation.roles}
assert set(roles) == {"action_cue", "delta_quantity", "changed_object", "direction"}
assert roles["direction"].target_id == direction
assert roles["action_cue"].evidence_spans[0].text == cue
assert assessment.runnable is True
assert assessment.missing_bindings == ()
assert assessment.unresolved_hazards == ()
assert all(
span.text == problem_text[span.start:span.end]
for span in (*proposal.evidence_spans, *relation.evidence_spans, *assessment.evidence_spans)
)
def test_builder_publishes_unary_delta_proposal_before_assessment(
monkeypatch: pytest.MonkeyPatch,
) -> None:
observed_statuses: list[str] = []
original = problem_frame_contracts.assess_contracts
def observe(frame):
observed_statuses.append(_proposal(frame).status)
return original(frame)
monkeypatch.setattr(problem_frame_contracts, "assess_contracts", observe)
frame = build_problem_frame("Ana gained 3 marbles.")
assert observed_statuses == ["proposed"]
assert _assessment(frame).runnable is True
def test_proposal_free_frame_does_not_dispatch_unary_delta_contract() -> None:
frame = build_problem_frame("Ana gained 3 marbles.")
proposal_free = dataclasses.replace(frame, proposals=())
assert CANDIDATE_ORGAN not in {
assessment.candidate_organ
for assessment in assess_contracts(proposal_free)
}
assessment = assess_unary_delta(proposal_free)
assert assessment.runnable is False
assert "unary_delta_proposal_required" in assessment.missing_bindings
@pytest.mark.parametrize(
"problem_text",
(
"There are 12 apples.",
"Tom has 12 apples.",
"Tom gave Ana 3 apples.",
"Tom sent Ana 3 apples.",
"Tom received 3 apples.",
"Tom bought 3 apples.",
"Tom put 3 apples in the box.",
"Tom took 3 apples from the box.",
"Tom had 12 apples and lost 3.",
"There are 12 apples. Tom lost 3.",
"She gained 3 apples.",
"Tom gained apples.",
"Tom gained 3 apples and oranges.",
"Tom gained 3 and 4 apples.",
"Tom gained 3 more apples than Ana.",
"20% of apples were gained.",
"3 apples per child were gained.",
"Tom did not gain 3 apples.",
"Tom may have gained 3 apples.",
"Tom gained 3 apples and lost 2 apples.",
),
)
def test_confusers_do_not_dispatch_unary_delta(problem_text: str) -> None:
frame = build_problem_frame(problem_text)
assert FAMILY_ID not in {proposal.family_id for proposal in frame.proposals}
assert not any(
relation.relation_type == "unary_delta"
for relation in frame.bound_relations
)
assert CANDIDATE_ORGAN not in {
assessment.candidate_organ
for assessment in assess_contracts(frame)
}
def test_missing_object_literal_surface_stays_unbound_and_non_runnable() -> None:
frame = build_problem_frame("Tom gained 3.")
assert FAMILY_ID not in {proposal.family_id for proposal in frame.proposals}
assert not any(
relation.relation_type == "unary_delta"
for relation in frame.bound_relations
)
assert CANDIDATE_ORGAN not in {
assessment.candidate_organ
for assessment in assess_contracts(frame)
}
def test_missing_quantity_blocks_runnable_unary_delta() -> None:
frame = build_problem_frame("Tom gained 3 apples.")
relation = _relation(frame)
stripped_roles = tuple(role for role in relation.roles if role.role != "delta_quantity")
broken = dataclasses.replace(frame, bound_relations=(dataclasses.replace(relation, roles=stripped_roles),))
assessment = assess_unary_delta(broken)
assert assessment.runnable is False
assert "delta_quantity_unbound" in assessment.missing_bindings
def test_quantity_kind_conflict_blocks_runnable_unary_delta() -> None:
frame = build_problem_frame("Tom gained 3 degrees.")
assessment = _assessment(frame)
assert assessment.runnable is False
assert "quantity_kind_unresolved" in assessment.missing_bindings
def test_synthetic_or_mismatched_unary_delta_evidence_refuses() -> None:
frame = build_problem_frame("Ana gained 3 marbles.")
proposal = _proposal(frame)
relation = _relation(frame)
synthetic = SourceSpan("synthetic", 0, 9)
cue = SourceSpan("Ana gained", 0, 10)
bad_relation = dataclasses.replace(
relation,
roles=tuple(
dataclasses.replace(role, evidence_spans=(synthetic,))
if role.role == "action_cue"
else role
for role in relation.roles
),
evidence_spans=(cue, *relation.evidence_spans[1:]),
)
broken = dataclasses.replace(
frame,
proposals=(dataclasses.replace(proposal, evidence_spans=(synthetic,)),),
bound_relations=(bad_relation,),
)
assessment = assess_unary_delta(broken)
assert assessment.runnable is False
assert "provenance_span_inexact" in assessment.missing_bindings
def test_unary_delta_replay_is_deterministic() -> None:
problem_text = "Ana gained 3 marbles."
first = build_problem_frame(problem_text)
second = build_problem_frame(problem_text)
assert first.proposals == second.proposals
assert first.bound_relations == second.bound_relations
assert assess_contracts(first) == assess_contracts(second)
def test_existing_proposal_first_families_do_not_gain_unary_delta_dispatch() -> None:
percent_partition = (
"A school has 100 students. Half of the students are girls, the other half are boys. "
"20% of the girls have dogs and 10% of the boys have dogs. "
"How many students own dogs?"
)
fraction_decrease = (
"A tank's temperature will decrease to 3/4 of its current temperature. "
"If the current temperature is 84 degrees, how many degrees will it decrease by?"
)
quantity_entity = "A school has 100 students."
for problem_text in (percent_partition, fraction_decrease, quantity_entity):
frame = build_problem_frame(problem_text)
assert FAMILY_ID not in {proposal.family_id for proposal in frame.proposals}
assert CANDIDATE_ORGAN not in {
assessment.candidate_organ
for assessment in assess_contracts(frame)
}
def test_unary_delta_path_does_not_import_legacy_semantic_state() -> None:
for module in (
"generate.problem_frame_builder",
"generate.problem_frame_contracts",
):
path = Path(__import__(module, fromlist=["__file__"]).__file__)
tree = ast.parse(path.read_text(encoding="utf-8"))
imported = {
node.module
for node in ast.walk(tree)
if isinstance(node, ast.ImportFrom) and node.module is not None
}
assert "generate.derivation.state" not in imported