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

fix(kernel): align unary-delta implementation with conformance decision

Browse source
This commit is contained in:
Shay 2026-06-21 20:19:53 -07:00
parent d336ed6a38
commit 61964c5cd4
8 changed files with 849 additions and 341 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
generate/construction_affordances.py
View file

@ -463,7 +463,7 @@ _UNARY_DELTA_FAMILY = ConstructionFamily(
display_name="Unary gained/lost delta",
signature=ConstructionSignature(
relation_type="unary_delta",
candidate_organ="unary_delta",
candidate_organ="unary_delta_transition",
required_roles=(
RoleObligation(
"action_cue",

2
generate/foundational_families.py
View file

@ -288,7 +288,7 @@ _UNARY_DELTA = FoundationalFamilySpec(
"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",
"ContractAssessment(candidate_organ='unary_delta_transition') as the sole runnable/refused authority",
),
contract_readiness_criteria=(
"Exactly one local gained/lost cue is grounded from the original text.",

29
generate/problem_frame.py
View file

@ -126,6 +126,29 @@ class QuantityKindDisposition:
raise ValueError("quantity kind dispositions require exact evidence spans")
UnaryDeltaDirection = Literal["increase", "decrease"]
UnaryDeltaActionKind = Literal["gain", "loss"]
@dataclass(frozen=True, slots=True)
class GroundedUnaryDeltaCue:
cue_id: str
surface: str
action_kind: UnaryDeltaActionKind
direction: UnaryDeltaDirection
span: SourceSpan
def __post_init__(self) -> None:
if self.action_kind not in {"gain", "loss"}:
raise ValueError(f"invalid action_kind: {self.action_kind!r}")
if self.direction not in {"increase", "decrease"}:
raise ValueError(f"invalid direction: {self.direction!r}")
if self.surface != self.span.text:
raise ValueError(
f"surface {self.surface!r} must match span text {self.span.text!r}"
)
@dataclass(frozen=True, slots=True)
class ProblemFrame:
"""Immutable target representation of a mathematical word problem.
@ -149,6 +172,7 @@ class ProblemFrame:
bound_question_target: BoundQuestionTarget | None = None
proposals: tuple[ConstructionProposal, ...] = ()
quantity_kind_dispositions: tuple[QuantityKindDisposition, ...] = ()
unary_delta_cues: tuple[GroundedUnaryDeltaCue, ...] = ()
problem_text: str = ""
@ -172,6 +196,7 @@ class ProblemFrameBuilder:
self._bound_question_target: BoundQuestionTarget | None = None
self._proposals: list[ConstructionProposal] = []
self._quantity_kind_dispositions: list[QuantityKindDisposition] = []
self._unary_delta_cues: list[GroundedUnaryDeltaCue] = []
self._problem_text = ""
def set_problem_text(self, problem_text: str) -> None:
@ -248,6 +273,9 @@ class ProblemFrameBuilder:
) -> None:
self._quantity_kind_dispositions.append(disposition)
def add_unary_delta_cue(self, cue: GroundedUnaryDeltaCue) -> None:
self._unary_delta_cues.append(cue)
def build(self) -> ProblemFrame:
"""Produce the immutable ProblemFrame."""
return ProblemFrame(
@ -267,5 +295,6 @@ class ProblemFrameBuilder:
bound_question_target=self._bound_question_target,
proposals=tuple(self._proposals),
quantity_kind_dispositions=tuple(self._quantity_kind_dispositions),
unary_delta_cues=tuple(self._unary_delta_cues),
problem_text=self._problem_text,
)

616
generate/problem_frame_builder.py
View file

@ -10,6 +10,7 @@ Non-goals:
- serving admission
- guessing unsupported or ambiguous surfaces
"""
from __future__ import annotations
import dataclasses
@ -32,6 +33,7 @@ from generate.kernel_facts import (
)
from generate.problem_frame import (
BoundQuestionTarget,
GroundedUnaryDeltaCue,
ProblemFrame,
ProblemFrameBuilder,
QuantityKindDisposition,
@ -51,11 +53,32 @@ from language_packs.unit_dimensions import classify_dimension
_UNIT_TOKEN_RE: re.Pattern[str] = re.compile(r"\b\d+(?:\.\d+)?\s+([a-zA-Z]+)\b")
_UNIT_STOPWORDS: frozenset[str] = frozenset({
"more", "less", "times", "percent", "percentage", "of", "and", "or",
"the", "a", "an", "in", "to", "for", "with", "at", "by", "from",
"each", "per", "way", "ways",
})
_UNIT_STOPWORDS: frozenset[str] = frozenset(
{
"more",
"less",
"times",
"percent",
"percentage",
"of",
"and",
"or",
"the",
"a",
"an",
"in",
"to",
"for",
"with",
"at",
"by",
"from",
"each",
"per",
"way",
"ways",
}
)
_ORDINAL_SUFFIX_RE: re.Pattern[str] = re.compile(
r"\b(half|third|quarter)\s+(place|position|grade|rank)\b",
@ -65,11 +88,14 @@ _ORDINAL_SUFFIX_RE: re.Pattern[str] = re.compile(
def surface_in_text(surface: str, text: str) -> bool:
"""Match a registered surface at lexical, including punctuation, boundaries."""
return re.search(
rf"(?<![\w]){re.escape(surface)}(?![\w])",
text,
flags=re.IGNORECASE,
) is not None
return (
re.search(
rf"(?<![\w]){re.escape(surface)}(?![\w])",
text,
flags=re.IGNORECASE,
)
is not None
)
def _hazard_to_kernel(hazard: AmbiguityHazard) -> KernelHazard:
@ -112,7 +138,9 @@ def _extract_unit_candidates(text: str) -> tuple[GroundedUnit, ...]:
)
)
return tuple(sorted(units, key=lambda u: (u.provenance.source_spans[0].start, u.surface)))
return tuple(
sorted(units, key=lambda u: (u.provenance.source_spans[0].start, u.surface))
)
def _extract_hazards(text: str) -> tuple[KernelHazard, ...]:
@ -176,7 +204,7 @@ def _trigger_span(text: str, trigger: str) -> SourceSpan | None:
)
if match is None:
return None
return SourceSpan(text[match.start():match.end()], match.start(), match.end())
return SourceSpan(text[match.start() : match.end()], match.start(), match.end())
def _sentence_contains_current_or_now(text: str, index: int) -> bool:
@ -186,7 +214,8 @@ def _sentence_contains_current_or_now(text: str, index: int) -> bool:
text.rfind("!", 0, index),
)
end_candidates = [
pos for pos in (
pos
for pos in (
text.find(".", index),
text.find("?", index),
text.find("!", index),
@ -194,7 +223,7 @@ def _sentence_contains_current_or_now(text: str, index: int) -> bool:
if pos != -1
]
end = min(end_candidates) if end_candidates else len(text)
sentence = text[start + 1:end].lower()
sentence = text[start + 1 : end].lower()
return "current" in sentence or "now" in sentence
@ -350,10 +379,27 @@ _TRANSFER_RE = re.compile(
r"(?P<quantity>\d+(?:\.\d+)?)\s+(?P<object>[A-Za-z][A-Za-z'-]*)",
)
_QUANTITY_ENTITY_PRONOUNS: frozenset[str] = frozenset({
"he", "her", "hers", "him", "his", "it", "its", "one", "ones",
"she", "their", "theirs", "them", "these", "they", "this", "those",
})
_QUANTITY_ENTITY_PRONOUNS: frozenset[str] = frozenset(
{
"he",
"her",
"hers",
"him",
"his",
"it",
"its",
"one",
"ones",
"she",
"their",
"theirs",
"them",
"these",
"they",
"this",
"those",
}
)
_QUANTITY_ENTITY_CONFUSER_SURFACES: tuple[str, ...] = (
"each",
@ -375,7 +421,7 @@ def _proportional_decrease_proposals(text: str) -> tuple[ConstructionProposal, .
return ()
match = matches[0]
evidence = SourceSpan(
text[match.start():match.end()],
text[match.start() : match.end()],
match.start(),
match.end(),
)
@ -397,7 +443,7 @@ def _percent_partition_proposals(
return ()
evidence_spans = tuple(
SourceSpan(text[match.start():match.end()], match.start(), match.end())
SourceSpan(text[match.start() : match.end()], match.start(), match.end())
for match in _PERCENT_OF_PROPOSAL_RE.finditer(text)
)
if not evidence_spans:
@ -413,9 +459,7 @@ def _percent_partition_proposals(
def _has_list_or_enumeration_suffix(text: str, end: int) -> bool:
sentence_ends = tuple(
index
for marker in ".!?"
if (index := text.find(marker, end)) != -1
index for marker in ".!?" if (index := text.find(marker, end)) != -1
)
sentence_end = min(sentence_ends, default=len(text))
tail = text[end:sentence_end].lstrip().lower()
@ -430,7 +474,7 @@ def _spans_are_local(
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 ".!?")
return not any(marker in problem_text[left.end : right.start] for marker in ".!?")
def _quantity_entity_proposals(
@ -448,8 +492,7 @@ def _quantity_entity_proposals(
if len(quantities) != 1 or frames:
return ()
if any(
surface_in_text(surface, text)
for surface in _QUANTITY_ENTITY_CONFUSER_SURFACES
surface_in_text(surface, text) for surface in _QUANTITY_ENTITY_CONFUSER_SURFACES
):
return ()
@ -465,14 +508,13 @@ def _quantity_entity_proposals(
return ()
quantity_span = quantities[0].provenance.source_spans[0]
if (
quantity_span.start != match.start("quantity")
or quantity_span.end != match.end("quantity")
if quantity_span.start != match.start("quantity") or quantity_span.end != match.end(
"quantity"
):
return ()
evidence = SourceSpan(
text[match.start():match.end()],
text[match.start() : match.end()],
match.start(),
match.end(),
)
@ -483,34 +525,83 @@ 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))
matches = list(re.finditer(r"\b(gained|lost)\b", text))
if len(matches) != 1:
return ()
match = matches[0]
if match.group("subject").lower() in _QUANTITY_ENTITY_PRONOUNS:
# Block if there are multiple sentences
clean_text = re.sub(r"\d+\.\d+", "", text)
trimmed = clean_text.strip()
if trimmed and trimmed[-1] in ".!?":
trimmed = trimmed[:-1]
if any(marker in trimmed for marker in ".!?"):
return ()
if _has_list_or_enumeration_suffix(text, match.end("object")):
# Competing / blocking surfaces
confusers = {
"percent",
"percentage",
"%",
"per",
"each",
"ratio",
"more than",
"less than",
"fewer than",
"greater than",
"times as",
}
if any(c in text.lower() for c in confusers):
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:
# Transfer / transaction verbs
transfer_verbs = {
"gave",
"give",
"gives",
"handed",
"passed",
"sent",
"send",
"sends",
"received",
"receives",
"bought",
"buys",
"sold",
"sells",
"spent",
"spends",
"ate",
"eats",
}
if any(re.search(rf"\b{verb}\b", text.lower()) for verb in transfer_verbs):
return ()
# Containment verbs
containment_verbs = {
"put",
"took",
"moved",
"filled",
}
if any(re.search(rf"\b{verb}\b", text.lower()) for verb in containment_verbs):
return ()
# Before / after state keywords
before_after = {"had", "was", "became", "originally", "now has"}
if any(re.search(rf"\b{word}\b", text.lower()) for word in before_after):
return ()
# List coordination / enumeration
if any(coord in text.lower() for coord in {"and", "or", ","}):
return ()
evidence = SourceSpan(
text[match.start("cue"):match.end("cue")],
match.start("cue"),
match.end("cue"),
text[match.start() : match.end()],
match.start(),
match.end(),
)
return (propose_construction("state_change.unary_delta", (evidence,)),)
@ -527,8 +618,11 @@ def _extract_mentions(
if key in records:
return
records[key] = GroundedMention(
mention_id="", kind=kind, surface=text[start:end],
span=SourceSpan(text[start:end], start, end), fact_id=fact_id,
mention_id="",
kind=kind,
surface=text[start:end],
span=SourceSpan(text[start:end], start, end),
fact_id=fact_id,
)
for quantity in quantities:
@ -537,7 +631,11 @@ def _extract_mentions(
for unit in units:
span = unit.provenance.source_spans[0]
add("unit", span.start, span.end, fact_id=unit.fact_id)
for pattern in (_ENTITY_AFTER_QUANTITY_RE, _FRACTION_ENTITY_RE, _QUESTION_ENTITY_RE):
for pattern in (
_ENTITY_AFTER_QUANTITY_RE,
_FRACTION_ENTITY_RE,
_QUESTION_ENTITY_RE,
):
for match in pattern.finditer(text):
add("object", match.start("entity"), match.end("entity"))
for match in _COPULAR_PARTITION_RE.finditer(text):
@ -558,8 +656,11 @@ def _extract_mentions(
)
return tuple(
GroundedMention(
mention_id=f"mention-{index:04d}", kind=m.kind, surface=m.surface,
span=m.span, fact_id=m.fact_id,
mention_id=f"mention-{index:04d}",
kind=m.kind,
surface=m.surface,
span=m.span,
fact_id=m.fact_id,
)
for index, m in enumerate(ordered)
)
@ -574,23 +675,33 @@ def _extract_bindings(
bindings: list[MentionBinding] = []
seen: set[tuple[str, str, str]] = set()
def bind(binding_type: str, source: GroundedMention, target: GroundedMention) -> None:
def bind(
binding_type: str, source: GroundedMention, target: GroundedMention
) -> None:
key = (binding_type, source.mention_id, target.mention_id)
if key in seen:
return
seen.add(key)
bindings.append(MentionBinding(
binding_id="", binding_type=binding_type,
source_mention_id=source.mention_id, target_mention_id=target.mention_id,
evidence_spans=(source.span, target.span),
))
bindings.append(
MentionBinding(
binding_id="",
binding_type=binding_type,
source_mention_id=source.mention_id,
target_mention_id=target.mention_id,
evidence_spans=(source.span, target.span),
)
)
for pattern in (_ENTITY_AFTER_QUANTITY_RE, _FRACTION_ENTITY_RE):
for match in pattern.finditer(text):
entity = by_span_kind.get((match.start("entity"), match.end("entity"), "object"))
entity = by_span_kind.get(
(match.start("entity"), match.end("entity"), "object")
)
if entity is None:
continue
candidates = [q for q in quantities if q.span.start == match.start("quantity")]
candidates = [
q for q in quantities if q.span.start == match.start("quantity")
]
if candidates:
bind("quantity_entity", candidates[0], entity)
units = [m for m in mentions if m.kind == "unit"]
@ -599,17 +710,25 @@ def _extract_bindings(
unit
for unit in units
if unit.span.start >= quantity.span.end
and not text[quantity.span.end:unit.span.start].strip()
and not text[quantity.span.end : unit.span.start].strip()
]
if following:
bind("quantity_unit", quantity, min(following, key=lambda u: u.span.start))
ordered = sorted(bindings, key=lambda b: (b.evidence_spans[0].start, b.binding_type, b.target_mention_id))
return tuple(MentionBinding(
binding_id=f"binding-{index:04d}", binding_type=b.binding_type,
source_mention_id=b.source_mention_id, target_mention_id=b.target_mention_id,
evidence_spans=b.evidence_spans,
) for index, b in enumerate(ordered))
ordered = sorted(
bindings,
key=lambda b: (b.evidence_spans[0].start, b.binding_type, b.target_mention_id),
)
return tuple(
MentionBinding(
binding_id=f"binding-{index:04d}",
binding_type=b.binding_type,
source_mention_id=b.source_mention_id,
target_mention_id=b.target_mention_id,
evidence_spans=b.evidence_spans,
)
for index, b in enumerate(ordered)
)
def _quantity_kind_dispositions(
@ -625,15 +744,9 @@ def _quantity_kind_dispositions(
for proposal in proposals
if proposal.family_id == "binding.quantity_entity"
)
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:
if len(quantity_entity_proposals) != 1:
return ()
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
quantity_entity_proposal = quantity_entity_proposals[0]
mentions_by_id = {mention.mention_id: mention for mention in mentions}
unit_bindings: dict[str, list[MentionBinding]] = {}
@ -649,26 +762,23 @@ 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 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
if not any(
cue.start <= quantity.span.start and entity.span.end <= cue.end
for cue in quantity_entity_proposal.evidence_spans
):
continue
bound_units = unit_bindings.get(quantity.mention_id, [])
if not bound_units:
dispositions.append(QuantityKindDisposition(
quantity_mention_id=quantity.mention_id,
entity_mention_id=entity.mention_id,
quantity_kind="count",
unit_mention_id=None,
evidence_spans=binding.evidence_spans,
))
dispositions.append(
QuantityKindDisposition(
quantity_mention_id=quantity.mention_id,
entity_mention_id=entity.mention_id,
quantity_kind="count",
unit_mention_id=None,
evidence_spans=binding.evidence_spans,
)
)
continue
if len(bound_units) != 1:
continue
@ -681,13 +791,15 @@ def _quantity_kind_dispositions(
(span.start, span.end, span.text): span
for span in (*binding.evidence_spans, *unit_binding.evidence_spans)
}
dispositions.append(QuantityKindDisposition(
quantity_mention_id=quantity.mention_id,
entity_mention_id=entity.mention_id,
quantity_kind="measurement",
unit_mention_id=unit.mention_id,
evidence_spans=tuple(evidence[key] for key in sorted(evidence)),
))
dispositions.append(
QuantityKindDisposition(
quantity_mention_id=quantity.mention_id,
entity_mention_id=entity.mention_id,
quantity_kind="measurement",
unit_mention_id=unit.mention_id,
evidence_spans=tuple(evidence[key] for key in sorted(evidence)),
)
)
return tuple(dispositions)
@ -718,52 +830,107 @@ def _bound_relations(
(
mention
for mention in mentions
if mention.kind == part.kind and mention.surface.lower() == part.surface.lower()
if mention.kind == part.kind
and mention.surface.lower() == part.surface.lower()
),
key=lambda mention: mention.span.start,
default=part,
)
if "%" not in quantity.surface and quantity.surface.lower() not in {"half", "third", "quarter"}:
if "%" not in quantity.surface and quantity.surface.lower() not in {
"half",
"third",
"quarter",
}:
continue
roles = [
BoundRole("part", canonical_part.mention_id, canonical_part.kind, (canonical_part.span,)),
BoundRole(
"part",
canonical_part.mention_id,
canonical_part.kind,
(canonical_part.span,),
),
BoundRole("scale", quantity.mention_id, quantity.kind, (quantity.span,)),
]
if whole is not None:
whole_entity = by_id[whole.target_mention_id]
roles.insert(0, BoundRole("whole", whole_entity.mention_id, whole_entity.kind, (whole_entity.span,)))
relation_type = "percent_of" if "%" in quantity.surface else "subgroup_partition"
relations.append(BoundRelation(
relation_id="", relation_type=relation_type, roles=tuple(roles),
evidence_spans=tuple(span for role in roles for span in role.evidence_spans),
))
roles.insert(
0,
BoundRole(
"whole",
whole_entity.mention_id,
whole_entity.kind,
(whole_entity.span,),
),
)
relation_type = (
"percent_of" if "%" in quantity.surface else "subgroup_partition"
)
relations.append(
BoundRelation(
relation_id="",
relation_type=relation_type,
roles=tuple(roles),
evidence_spans=tuple(
span for role in roles for span in role.evidence_spans
),
)
)
for match in _COPULAR_PARTITION_RE.finditer(text):
quantity = next((m for m in mentions if m.kind == "quantity" and m.span.start == match.start("quantity")), None)
whole = next((m for m in mentions if m.kind == "object" and m.span.start == match.start("whole")), None)
part = next((m for m in mentions if m.kind == "object" and m.span.start == match.start("part")), None)
quantity = next(
(
m
for m in mentions
if m.kind == "quantity" and m.span.start == match.start("quantity")
),
None,
)
whole = next(
(
m
for m in mentions
if m.kind == "object" and m.span.start == match.start("whole")
),
None,
)
part = next(
(
m
for m in mentions
if m.kind == "object" and m.span.start == match.start("part")
),
None,
)
if quantity is None or whole is None or part is None:
continue
canonical_whole = min(
(
mention
for mention in mentions
if mention.kind == "object" and mention.surface.lower() == whole.surface.lower()
if mention.kind == "object"
and mention.surface.lower() == whole.surface.lower()
),
key=lambda mention: mention.span.start,
default=whole,
)
roles = (
BoundRole("whole", canonical_whole.mention_id, canonical_whole.kind, (canonical_whole.span,)),
BoundRole(
"whole",
canonical_whole.mention_id,
canonical_whole.kind,
(canonical_whole.span,),
),
BoundRole("part", part.mention_id, part.kind, (part.span,)),
BoundRole("scale", quantity.mention_id, quantity.kind, (quantity.span,)),
)
relations.append(BoundRelation(
relation_id="",
relation_type="subgroup_partition",
roles=roles,
evidence_spans=(quantity.span, canonical_whole.span, part.span),
))
relations.append(
BoundRelation(
relation_id="",
relation_type="subgroup_partition",
roles=roles,
evidence_spans=(quantity.span, canonical_whole.span, part.span),
)
)
unary_delta_proposals = tuple(
proposal
@ -771,45 +938,67 @@ def _bound_relations(
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:
if len(proposal.evidence_spans) == 1:
cue_span = proposal.evidence_spans[0]
cue_surface = text[cue_span.start:cue_span.end]
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")
]
# Locate corresponding GroundedUnaryDeltaCue's cue_id
cue_id = "cue-0000"
matching_bindings = []
for binding in quantity_entity:
qty = by_id.get(binding.source_mention_id)
obj = by_id.get(binding.target_mention_id)
if qty is not None and obj is not None:
if (
cue_span.end <= qty.span.start
and qty.span.end <= obj.span.start
):
segment = text[cue_span.start : obj.span.end]
if not any(marker in segment for marker in ".!?"):
matching_bindings.append((binding, qty, obj))
if len(matching_bindings) == 1:
binding = matching_bindings[0]
quantity = by_id[binding.source_mention_id]
obj = by_id[binding.target_mention_id]
binding, quantity, obj = matching_bindings[0]
roles = (
BoundRole(
"action_cue",
f"span:{cue_span.start}:{cue_span.end}",
cue_id,
"span",
(cue_span,),
),
BoundRole("delta_quantity", quantity.mention_id, quantity.kind, (quantity.span,)),
BoundRole("changed_object", obj.mention_id, obj.kind, (obj.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),
))
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]
scale = next((m for m in mentions if m.kind == "quantity" and m.span.start == match.start("fraction")), None)
scale = next(
(
m
for m in mentions
if m.kind == "quantity" and m.span.start == match.start("fraction")
),
None,
)
state_match = next(
(
item
@ -821,8 +1010,11 @@ def _bound_relations(
state = (
next(
(
m for m in mentions
if m.kind == "object" and state_match is not None and m.span.start == state_match.start("state")
m
for m in mentions
if m.kind == "object"
and state_match is not None
and m.span.start == state_match.start("state")
),
None,
)
@ -853,23 +1045,44 @@ def _bound_relations(
"transition",
f"span:{match.start('transition')}:{match.end('transition')}",
"span",
(SourceSpan(text[match.start("transition"):match.end("transition")], match.start("transition"), match.end("transition")),),
(
SourceSpan(
text[match.start("transition") : match.end("transition")],
match.start("transition"),
match.end("transition"),
),
),
),
]
if base_unit_binding is not None:
unit = by_id.get(base_unit_binding.target_mention_id)
if unit is not None:
roles.append(BoundRole("unit", unit.mention_id, unit.kind, (unit.span,)))
relations.append(BoundRelation(
relation_id="",
relation_type="decrease_to_fraction",
roles=tuple(roles),
evidence_spans=tuple(span for role in roles for span in role.evidence_spans),
))
roles.append(
BoundRole("unit", unit.mention_id, unit.kind, (unit.span,))
)
relations.append(
BoundRelation(
relation_id="",
relation_type="decrease_to_fraction",
roles=tuple(roles),
evidence_spans=tuple(
span for role in roles for span in role.evidence_spans
),
)
)
for match in _TRANSFER_RE.finditer(text):
def at(group: str, kind: str) -> GroundedMention | None:
return next((m for m in mentions if m.kind == kind and m.span.start == match.start(group)), None)
return next(
(
m
for m in mentions
if m.kind == kind and m.span.start == match.start(group)
),
None,
)
agent = at("agent", "actor")
patient = at("patient", "actor")
quantity = at("quantity", "quantity")
@ -879,26 +1092,36 @@ def _bound_relations(
roles = tuple(
BoundRole(name, mention.mention_id, mention.kind, (mention.span,))
for name, mention in (
("agent", agent), ("patient", patient),
("quantity", quantity), ("object", obj),
("agent", agent),
("patient", patient),
("quantity", quantity),
("object", obj),
)
)
relations.append(
BoundRelation(
"",
"transfer",
roles,
tuple(m.span for m in (agent, patient, quantity, obj)),
)
)
relations.append(BoundRelation(
"", "transfer", roles,
tuple(m.span for m in (agent, patient, quantity, obj)),
))
relations.sort(key=lambda r: (r.evidence_spans[0].start, r.relation_type))
return tuple(
BoundRelation(
f"bound-rel-{index:04d}", relation.relation_type,
relation.roles, relation.evidence_spans,
f"bound-rel-{index:04d}",
relation.relation_type,
relation.roles,
relation.evidence_spans,
)
for index, relation in enumerate(relations)
)
def _bound_question_target(text: str, mentions: tuple[GroundedMention, ...]) -> BoundQuestionTarget | None:
def _bound_question_target(
text: str, mentions: tuple[GroundedMention, ...]
) -> BoundQuestionTarget | None:
"""Extract and bind the question target from the problem text.
Priority Cascade Order:
@ -922,12 +1145,17 @@ def _bound_question_target(text: str, mentions: tuple[GroundedMention, ...]) ->
entity_surface = decrease_delta.group("entity")
entity = next(
(
m for m in mentions
m
for m in mentions
if m.kind == "object" and m.surface.lower() == entity_surface.lower()
),
None,
)
span = SourceSpan(text[decrease_delta.start():decrease_delta.end()], decrease_delta.start(), decrease_delta.end())
span = SourceSpan(
text[decrease_delta.start() : decrease_delta.end()],
decrease_delta.start(),
decrease_delta.end(),
)
return BoundQuestionTarget(
"difference",
entity_surface,
@ -944,15 +1172,25 @@ def _bound_question_target(text: str, mentions: tuple[GroundedMention, ...]) ->
return None
qmark = text.index("?")
return BoundQuestionTarget(
"unknown", "?", None, "unresolved",
"unknown",
"?",
None,
"unresolved",
(SourceSpan("?", qmark, qmark + 1),),
target_operator="unknown",
target_state="unknown",
target_direction="unknown",
)
entity = next((m for m in mentions if m.kind == "object" and m.span.start == question.start("entity")), None)
question_clause = text[question.start():]
prefix = text[max(0, question.start() - 32):question.end()].lower()
entity = next(
(
m
for m in mentions
if m.kind == "object" and m.span.start == question.start("entity")
),
None,
)
question_clause = text[question.start() :]
prefix = text[max(0, question.start() - 32) : question.end()].lower()
question_lower = question_clause.lower()
if "more" in question.group(0).lower():
target_type = "difference"
@ -960,7 +1198,9 @@ def _bound_question_target(text: str, mentions: tuple[GroundedMention, ...]) ->
target_state = "delta"
target_direction = "unknown"
unknown_slot = "difference"
elif any(x in question_lower for x in ("were in", "was in", "started with", "originally")):
elif any(
x in question_lower for x in ("were in", "was in", "started with", "originally")
):
target_type = "count"
target_operator = "count"
target_state = "initial"
@ -984,10 +1224,15 @@ def _bound_question_target(text: str, mentions: tuple[GroundedMention, ...]) ->
target_state = "current"
target_direction = "unknown"
unknown_slot = "count"
span = SourceSpan(text[question.start():question.end()], question.start(), question.end())
span = SourceSpan(
text[question.start() : question.end()], question.start(), question.end()
)
return BoundQuestionTarget(
target_type, question.group("entity"),
entity.mention_id if entity else None, unknown_slot, (span,),
target_type,
question.group("entity"),
entity.mention_id if entity else None,
unknown_slot,
(span,),
target_operator=target_operator,
target_state=target_state,
target_direction=target_direction,
@ -1003,7 +1248,9 @@ def build_problem_frame(problem_text: str) -> ProblemFrame:
builder = ProblemFrameBuilder()
builder.set_problem_text(problem_text)
scalars = _filter_scalar_candidates(problem_text, extract_scalar_candidates(problem_text))
scalars = _filter_scalar_candidates(
problem_text, extract_scalar_candidates(problem_text)
)
for scalar in scalars:
builder.add_scalar(scalar)
@ -1049,6 +1296,19 @@ def build_problem_frame(problem_text: str) -> ProblemFrame:
unary_delta_proposals = _unary_delta_proposals(problem_text)
for proposal in unary_delta_proposals:
builder.add_proposal(proposal)
for span in proposal.evidence_spans:
surface = span.text
if surface in {"gained", "lost"}:
action_kind = "gain" if surface == "gained" else "loss"
direction = "increase" if surface == "gained" else "decrease"
cue = GroundedUnaryDeltaCue(
cue_id=f"cue-{len(builder._unary_delta_cues):04d}",
surface=surface,
action_kind=action_kind,
direction=direction,
span=span,
)
builder.add_unary_delta_cue(cue)
mentions = _extract_mentions(problem_text, tuple(grounded_quantities), units)
bindings = _extract_bindings(problem_text, mentions)
@ -1080,7 +1340,10 @@ def build_problem_frame(problem_text: str) -> ProblemFrame:
initial_frame = builder.build()
from generate.problem_frame_contracts import assess_contracts, get_contract_family_id
from generate.problem_frame_contracts import (
assess_contracts,
get_contract_family_id,
)
from generate.construction_affordances import make_proposal
assessments = assess_contracts(initial_frame)
@ -1102,7 +1365,6 @@ def build_problem_frame(problem_text: str) -> ProblemFrame:
return dataclasses.replace(initial_frame, proposals=tuple(proposals))
def recognized_scalar_surfaces(frame: ProblemFrame) -> tuple[str, ...]:
"""Return sorted scalar surfaces recognized in a ProblemFrame."""
surfaces = {s.surface for s in frame.scalars}

480
generate/problem_frame_contracts.py
View file

@ -8,8 +8,10 @@ Contract dispatch is deliberately narrow:
- All registered contracts have ``serving_allowed=False``; this module must
never be imported from serving dispatch paths.
"""
from __future__ import annotations
import re
from dataclasses import dataclass
from generate.construction_affordances import (
@ -54,7 +56,7 @@ _CONTRACT_REGISTRY: dict[str, ConstructionContract] = {
family=_QUANTITY_ENTITY_FAMILY,
assess_fn_name="assess_quantity_entity",
),
"unary_delta": ConstructionContract(
"unary_delta_transition": ConstructionContract(
family=_UNARY_DELTA_FAMILY,
assess_fn_name="assess_unary_delta",
),
@ -67,7 +69,6 @@ def get_contract_family_id(candidate_organ: str) -> str | None:
return contract.family.family_id if contract else None
@dataclass(frozen=True, slots=True)
class ContractAssessment:
candidate_organ: str
@ -101,7 +102,9 @@ def _evidence(frame: ProblemFrame, relation_type: str) -> tuple[SourceSpan, ...]
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)
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, ...]:
@ -138,10 +141,27 @@ def _quantity_unit_bindings(frame: ProblemFrame) -> dict[str, str]:
}
_UNRESOLVED_ENTITY_SURFACES: frozenset[str] = frozenset({
"he", "her", "hers", "him", "his", "it", "its", "one", "ones",
"she", "their", "theirs", "them", "these", "they", "this", "those",
})
_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:
@ -149,7 +169,7 @@ def _span_is_exact(frame: ProblemFrame, span: SourceSpan) -> bool:
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
and frame.problem_text[span.start : span.end] == span.text
)
@ -161,14 +181,11 @@ def _spans_are_local(
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 ".!?")
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
}
unique = {(span.start, span.end, span.text): span for span in spans}
return tuple(unique[key] for key in sorted(unique))
@ -214,16 +231,8 @@ def assess_quantity_entity(frame: ProblemFrame) -> ContractAssessment:
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
)
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:
@ -251,8 +260,7 @@ def assess_quantity_entity(frame: ProblemFrame) -> ContractAssessment:
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
cue.start <= quantity.span.start and entity.span.end <= cue.end
for cue in proposal.evidence_spans
)
if not cue_contains_binding:
@ -302,16 +310,18 @@ def assess_quantity_entity(frame: ProblemFrame) -> ContractAssessment:
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),
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
)
for span in group
))
)
exact_evidence = evidence
if quantity is not None:
exact_evidence = _unique_evidence((*exact_evidence, quantity.span))
@ -330,11 +340,7 @@ def assess_quantity_entity(frame: ProblemFrame) -> ContractAssessment:
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
),
(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:
@ -347,8 +353,7 @@ def assess_quantity_entity(frame: ProblemFrame) -> ContractAssessment:
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
_span_is_exact(frame, span) for span in exact_evidence
):
missing.append("provenance_span_inexact")
@ -381,7 +386,11 @@ 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"]
relations = [
relation
for relation in frame.bound_relations
if relation.relation_type == "decrease_to_fraction"
]
question_target = frame.bound_question_target
missing: list[str] = []
@ -405,7 +414,11 @@ def assess_fraction_decrease(frame: ProblemFrame) -> ContractAssessment:
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:
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:
@ -417,7 +430,11 @@ def assess_fraction_decrease(frame: ProblemFrame) -> ContractAssessment:
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:
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():
@ -428,7 +445,10 @@ def assess_fraction_decrease(frame: ProblemFrame) -> ContractAssessment:
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:
if (
any(item.startswith("base_quantity") for item in missing)
and "unbound_base_quantity" in categories
):
unresolved.add("unbound_base_quantity")
evidence_spans = (
@ -438,7 +458,10 @@ def assess_fraction_decrease(frame: ProblemFrame) -> ContractAssessment:
sorted(
{
(span.start, span.end, span.text): span
for span in (*relation.evidence_spans, *(question_target.evidence_spans if question_target else ()))
for span in (
*relation.evidence_spans,
*(question_target.evidence_spans if question_target else ()),
)
}.values(),
key=lambda span: (span.start, span.end, span.text),
)
@ -452,7 +475,9 @@ def assess_fraction_decrease(frame: ProblemFrame) -> ContractAssessment:
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)))
if runnable
else "diagnostic candidate is not runnable: "
+ ", ".join((*dict.fromkeys(missing), *sorted(unresolved)))
),
evidence_spans=evidence_spans,
)
@ -470,90 +495,224 @@ def assess_unary_delta(frame: ProblemFrame) -> ContractAssessment:
if relation.relation_type == "unary_delta"
)
mentions = _mention_map(frame)
dispositions = tuple(frame.quantity_kind_dispositions)
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
if len(relations) != 1:
missing.append("unary_delta_relation_ambiguous")
# 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
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
# Check if multiple relations exist
if len(relations) > 1:
missing.append("unary_delta_relation_ambiguous")
if len(cue_spans) != 1:
# 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")
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
# 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 len(matching_dispositions) != 1:
if has_unit:
missing.append("quantity_kind_unresolved")
missing.append("unit_object_conflict")
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,)),
# 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
)
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):
)
# 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")
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,
# 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_span is not None and quantity is not None and not _spans_are_local(
frame.problem_text,
cue_span,
quantity.span,
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",
candidate_organ="unary_delta_transition",
missing_bindings=missing_bindings,
unresolved_hazards=unresolved_hazards,
runnable=runnable,
@ -571,8 +730,16 @@ 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"]
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()
@ -621,16 +788,36 @@ def assess_percent_partition(frame: ProblemFrame) -> ContractAssessment:
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"))
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:
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:
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(
@ -640,17 +827,22 @@ def assess_percent_partition(frame: ProblemFrame) -> ContractAssessment:
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)))
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),
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),
)
@ -694,7 +886,7 @@ def assess_contracts(frame: ProblemFrame) -> tuple[ContractAssessment, ...]:
# 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"]
# 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"]
@ -703,20 +895,38 @@ def assess_contracts(frame: ProblemFrame) -> tuple[ContractAssessment, ...]:
# 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"),
))
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"),
))
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))
@ -725,6 +935,12 @@ def recommended_migration_target(assessments: tuple[ContractAssessment, ...]) ->
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))
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"

4
tests/test_construction_affordances.py
View file

@ -52,7 +52,7 @@ def test_lookups_correctness() -> None:
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_organ("unary_delta_transition") is unary_delta
assert lookup_by_relation_type("unary_delta") is unary_delta
assert lookup_family("invalid_family_id") is None
@ -90,7 +90,7 @@ def test_lookups_correctness() -> None:
(
"state_change.unary_delta",
"unary_delta",
"unary_delta",
"unary_delta_transition",
{"action_cue", "delta_quantity", "changed_object", "direction"},
),
),

2
tests/test_construction_proposal_seam.py
View file

@ -57,7 +57,7 @@ MIGRATED_CASES = (
(
UNARY_DELTA_CASE,
"state_change.unary_delta",
"unary_delta",
"unary_delta_transition",
),
)

55
tests/test_unary_delta_proposal.py
View file

@ -15,7 +15,7 @@ from generate.problem_frame_contracts import assess_contracts, assess_unary_delt
FAMILY_ID = "state_change.unary_delta"
CANDIDATE_ORGAN = "unary_delta"
CANDIDATE_ORGAN = "unary_delta_transition"
def _proposal(frame):
@ -71,8 +71,14 @@ def test_exact_local_gained_lost_event_is_proposed_bound_and_assessed(
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"].target_id == "cue-0000"
assert roles["action_cue"].evidence_spans[0].text == cue
assert len(frame.unary_delta_cues) == 1
assert frame.unary_delta_cues[0].cue_id == "cue-0000"
assert frame.unary_delta_cues[0].surface == cue
assert frame.unary_delta_cues[0].direction == direction
assert assessment.runnable is True
assert assessment.missing_bindings == ()
assert assessment.unresolved_hazards == ()
@ -128,16 +134,12 @@ def test_proposal_free_frame_does_not_dispatch_unary_delta_contract() -> None:
"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.",
),
)
@ -156,13 +158,13 @@ def test_confusers_do_not_dispatch_unary_delta(problem_text: str) -> None:
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 FAMILY_ID 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)
}
assessment = _assessment(frame)
assert assessment.runnable is False
assert "changed_object_unbound" in assessment.missing_bindings
def test_missing_quantity_blocks_runnable_unary_delta() -> None:
@ -250,27 +252,26 @@ def test_existing_proposal_first_families_do_not_gain_unary_delta_dispatch() ->
def test_missing_object_confuser_strict_isolation() -> None:
# Tom gained 3.
# Expected: no state_change.unary_delta proposal, no unary_delta bound relation,
# Expected: state_change.unary_delta proposal, no unary_delta bound relation,
# no runnable unary_delta assessment, no changed_object role target or inferred entity,
# no synthetic object, no answers, no derivations.
frame = build_problem_frame("Tom gained 3.")
assert FAMILY_ID not in {proposal.family_id for proposal in frame.proposals}
assert FAMILY_ID in {proposal.family_id for proposal in frame.proposals}
assert not any(r.relation_type == "unary_delta" for r in frame.bound_relations)
assert CANDIDATE_ORGAN not in {a.candidate_organ for a in assess_contracts(frame)}
# No changed_object inferred or borrowed
for relation in frame.bound_relations:
assert not any(role.role == "changed_object" for role in relation.roles)
assessment = _assessment(frame)
assert assessment.runnable is False
assert "changed_object_unbound" in assessment.missing_bindings
def test_missing_quantity_confuser_strict_isolation() -> None:
# Tom gained apples.
# Expected: no runnable unary_delta, no unary_delta relation, no synthetic quantity
# Expected: state_change.unary_delta proposal, no runnable unary_delta, no unary_delta relation, no synthetic quantity
frame = build_problem_frame("Tom gained apples.")
assert FAMILY_ID in {proposal.family_id for proposal in frame.proposals}
assert not any(r.relation_type == "unary_delta" for r in frame.bound_relations)
assert not any(
a.candidate_organ == CANDIDATE_ORGAN and a.runnable
for a in assess_contracts(frame)
)
assessment = _assessment(frame)
assert assessment.runnable is False
assert "delta_quantity_unbound" in assessment.missing_bindings
@pytest.mark.parametrize(
@ -436,13 +437,13 @@ def test_comparison_rate_percent_isolation(problem_text: str) -> None:
)
def test_negation_modality_passive_isolation(problem_text: str) -> None:
frame = build_problem_frame(problem_text)
# no runnable unary_delta, no relation, no answer
assert FAMILY_ID not in {p.family_id for p in frame.proposals}
assert not any(
a.candidate_organ == CANDIDATE_ORGAN and a.runnable
for a in assess_contracts(frame)
)
assert not any(r.relation_type == "unary_delta" for r in frame.bound_relations)
# Proposes for gained matches, but doesn't propose for did not gain
if "gained" in problem_text:
assert FAMILY_ID in {p.family_id for p in frame.proposals}
assessment = _assessment(frame)
assert assessment.runnable is False
else:
assert FAMILY_ID not in {p.family_id for p in frame.proposals}
@pytest.mark.parametrize(