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

fix(kernel): complete unary-delta conformance alignment

Browse source
This commit is contained in:
Shay 2026-06-21 22:07:12 -07:00
parent 61964c5cd4
commit 59a919e347
4 changed files with 133 additions and 55 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
generate/construction_affordances.py
View file

@ -516,11 +516,19 @@ _UNARY_DELTA_FAMILY = ConstructionFamily(
"unary_delta_relation_ambiguous",
"action_cue_unbound",
"delta_quantity_unbound",
"delta_quantity_ambiguous",
"changed_object_unbound",
"changed_object_ambiguous",
"local_binding_relation_unbound",
"direction_unbound",
"quantity_kind_unresolved",
"unit_object_conflict",
"provenance_span_inexact",
"quantity_entity_nonlocal",
"pronoun_antecedent_unresolved",
"event_assertion_unlicensed",
"passive_voice_unsupported",
"multiple_actor_surface",
),
diagnostic_only=True,
serving_allowed=False,

20
generate/problem_frame.py
View file

@ -139,10 +139,14 @@ class GroundedUnaryDeltaCue:
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 not in {"gained", "lost"}:
raise ValueError(f"invalid surface: {self.surface!r}")
if self.surface == "gained":
if self.action_kind != "gain" or self.direction != "increase":
raise ValueError("invalid gained triple")
elif self.surface == "lost":
if self.action_kind != "loss" or self.direction != "decrease":
raise ValueError("invalid lost triple")
if self.surface != self.span.text:
raise ValueError(
f"surface {self.surface!r} must match span text {self.span.text!r}"
@ -276,6 +280,14 @@ class ProblemFrameBuilder:
def add_unary_delta_cue(self, cue: GroundedUnaryDeltaCue) -> None:
self._unary_delta_cues.append(cue)
@property
def unary_delta_cue_count(self) -> int:
return len(self._unary_delta_cues)
@property
def unary_delta_cues(self) -> tuple[GroundedUnaryDeltaCue, ...]:
return tuple(self._unary_delta_cues)
def build(self) -> ProblemFrame:
"""Produce the immutable ProblemFrame."""
return ProblemFrame(

109
generate/problem_frame_builder.py
View file

@ -363,12 +363,7 @@ _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"
@ -546,14 +541,17 @@ def _unary_delta_proposals(
"per",
"each",
"ratio",
"than",
"more than",
"less than",
"fewer than",
"greater than",
"times as",
}
if any(c in text.lower() for c in confusers):
return ()
for c in confusers:
pattern = rf"\b{re.escape(c)}\b" if c[0].isalnum() and c[-1].isalnum() else re.escape(c)
if re.search(pattern, text, re.IGNORECASE):
return ()
# Transfer / transaction verbs
transfer_verbs = {
@ -595,7 +593,10 @@ def _unary_delta_proposals(
return ()
# List coordination / enumeration
if any(coord in text.lower() for coord in {"and", "or", ","}):
for coord in {"and", "or"}:
if re.search(rf"\b{coord}\b", text, re.IGNORECASE):
return ()
if "," in text:
return ()
evidence = SourceSpan(
@ -809,6 +810,7 @@ def _bound_relations(
mentions: tuple[GroundedMention, ...],
bindings: tuple[MentionBinding, ...],
proposals: tuple[ConstructionProposal, ...],
unary_delta_cues: tuple[GroundedUnaryDeltaCue, ...],
) -> tuple[BoundRelation, ...]:
by_id = {m.mention_id: m for m in mentions}
relations: list[BoundRelation] = []
@ -945,48 +947,52 @@ def _bound_relations(
if cue_span.text == cue_surface and cue_surface in {"gained", "lost"}:
direction = "increase" if cue_surface == "gained" else "decrease"
# 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, quantity, obj = matching_bindings[0]
roles = (
BoundRole(
"action_cue",
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("direction", direction, "direction", (cue_span,)),
)
relations.append(
BoundRelation(
relation_id="",
relation_type="unary_delta",
roles=roles,
evidence_spans=(cue_span, quantity.span, obj.span),
cue_id = None
for cue in unary_delta_cues:
if cue.span.start == cue_span.start and cue.span.end == cue_span.end:
cue_id = cue.cue_id
break
if cue_id is not None:
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, quantity, obj = matching_bindings[0]
roles = (
BoundRole(
"action_cue",
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("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:
@ -1302,7 +1308,7 @@ def build_problem_frame(problem_text: str) -> ProblemFrame:
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}",
cue_id=f"cue-{builder.unary_delta_cue_count:04d}",
surface=surface,
action_kind=action_kind,
direction=direction,
@ -1332,6 +1338,7 @@ def build_problem_frame(problem_text: str) -> ProblemFrame:
mentions,
bindings,
(*quantity_entity_proposals, *unary_delta_proposals),
builder.unary_delta_cues,
):
builder.add_bound_relation(relation)
bound_target = _bound_question_target(problem_text, mentions)

51
tests/test_unary_delta_proposal.py
View file

@ -561,3 +561,54 @@ def test_forbidden_import_coupling_checks() -> None:
assert forbid not in imported
for imp in imported:
assert not imp.startswith(forbid)
def test_unary_delta_cue_conformance() -> None:
from generate.problem_frame import GroundedUnaryDeltaCue
from generate.kernel_facts import SourceSpan
# 1. Invalid surface/action_kind/direction triples must raise ValueError
span = SourceSpan("gained", 0, 6)
with pytest.raises(ValueError):
GroundedUnaryDeltaCue("cue-0000", "gained", "loss", "increase", span)
with pytest.raises(ValueError):
GroundedUnaryDeltaCue("cue-0000", "gained", "gain", "decrease", span)
with pytest.raises(ValueError):
GroundedUnaryDeltaCue("cue-0000", "lost", "gain", "decrease", span)
with pytest.raises(ValueError):
GroundedUnaryDeltaCue("cue-0000", "lost", "loss", "increase", span)
with pytest.raises(ValueError):
GroundedUnaryDeltaCue("cue-0000", "gained", "gain", "increase", SourceSpan("lost", 0, 4))
# Valid triples must succeed
c1 = GroundedUnaryDeltaCue("cue-0001", "gained", "gain", "increase", span)
assert c1.cue_id == "cue-0001"
# 2. Test exact span resolution in _bound_relations
from generate.problem_frame_builder import _bound_relations, GroundedMention, MentionBinding, ConstructionProposal
mentions = (
GroundedMention("m-0000", "quantity", "3", SourceSpan("3", 7, 8)),
GroundedMention("m-0001", "object", "apples", SourceSpan("apples", 9, 15)),
)
bindings = (
MentionBinding("b-0000", "quantity_entity", "m-0000", "m-0001", (SourceSpan("3 apples", 7, 15),)),
)
proposals = (
ConstructionProposal(
family_id="state_change.unary_delta",
relation_type="unary_delta",
candidate_organ="unary_delta_transition",
evidence_spans=(span,),
status="proposed",
diagnostic_only=True,
serving_allowed=False,
),
)
# If the cue has a different ID and is passed in, _bound_relations must resolve and use it
cues = (
GroundedUnaryDeltaCue("cue-1234", "gained", "gain", "increase", span),
)
relations = _bound_relations("gained 3 apples", mentions, bindings, proposals, cues)
assert len(relations) == 1
action_cue_role = next(r for r in relations[0].roles if r.role == "action_cue")
assert action_cue_role.target_id == "cue-1234"