diff --git a/generate/problem_frame_bound_relations.py b/generate/problem_frame_bound_relations.py new file mode 100644 index 00000000..7036ee5f --- /dev/null +++ b/generate/problem_frame_bound_relations.py @@ -0,0 +1,574 @@ +"""ProblemFrame bound relation and target helpers. + +This module owns the phase that turns grounded mentions, bindings, proposals, and +unary-delta cues into quantity-kind dispositions, ``BoundRelation`` records, and +bound question targets. It does not assess contracts or create proposals. +""" + +from __future__ import annotations + +import re + +from generate.construction_affordances import ConstructionProposal +from generate.kernel_facts import ( + BoundRelation, + BoundRole, + GroundedMention, + MentionBinding, + SourceSpan, +) +from generate.problem_frame import ( + BoundQuestionTarget, + GroundedUnaryDeltaCue, + QuantityKindDisposition, +) +from generate.problem_frame_extractors import _sentence_contains_current_or_now + +_QUESTION_ENTITY_RE = re.compile( + r"\bhow\s+(?:many|much)\s+(?:more\s+)?(?P[A-Za-z][A-Za-z'-]*)", + re.IGNORECASE, +) +_DECREASE_STATE_RE = re.compile( + r"(?P[A-Za-z][A-Za-z'-]*)\s+will\s+decrease\s+to", + re.IGNORECASE, +) +_DECREASE_DELTA_QUESTION_RE = re.compile( + r"\bwhat\s+will\s+the\s+(?P[A-Za-z][A-Za-z'-]*)\s+decrease\s+by\??", + re.IGNORECASE, +) + +# Duplicated intentionally to preserve phase-local ownership. +# Do not import another phase's internals just to share this regex. +_COPULAR_PARTITION_RE = re.compile( + r"\b(?Phalf|third|quarter)\b\s+of\s+(?:the\s+)?" + r"(?P[A-Za-z][A-Za-z'-]*)\s+(?:are|is)\s+(?P[A-Za-z][A-Za-z'-]*)", + re.IGNORECASE, +) + +# Duplicated intentionally to preserve phase-local ownership. +# Do not import another phase's internals just to share this regex. +_DECREASE_TO_FRACTION_RE = re.compile( + r"(?Pdecrease\s+to)\s+(?P\d+\s*/\s*\d+)\s+of", + re.IGNORECASE, +) + +# Duplicated intentionally to preserve phase-local ownership. +# Do not import another phase's internals just to share this regex. +_TRANSFER_RE = re.compile( + r"\b(?P[A-Z][A-Za-z'-]*)\s+(?:gave|gives|give|handed|passed)\s+" + r"(?P[A-Z][A-Za-z'-]*)\s+" + r"(?P\d+(?:\.\d+)?)\s+(?P[A-Za-z][A-Za-z'-]*)", +) + + +def _quantity_kind_dispositions( + text: str, + mentions: tuple[GroundedMention, ...], + bindings: tuple[MentionBinding, ...], + proposals: tuple[ConstructionProposal, ...], +) -> tuple[QuantityKindDisposition, ...]: + """Close kind only for the exact proposal-backed local binding.""" + + quantity_entity_proposals = tuple( + proposal + for proposal in proposals + if proposal.family_id == "binding.quantity_entity" + ) + if len(quantity_entity_proposals) != 1: + return () + quantity_entity_proposal = quantity_entity_proposals[0] + + mentions_by_id = {mention.mention_id: mention for mention in mentions} + unit_bindings: dict[str, list[MentionBinding]] = {} + for binding in bindings: + if binding.binding_type == "quantity_unit": + unit_bindings.setdefault(binding.source_mention_id, []).append(binding) + + dispositions: list[QuantityKindDisposition] = [] + for binding in bindings: + if binding.binding_type != "quantity_entity": + continue + quantity = mentions_by_id.get(binding.source_mention_id) + entity = mentions_by_id.get(binding.target_mention_id) + if quantity is None or entity is None or quantity.fact_id is None: + 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, + ) + ) + continue + if len(bound_units) != 1: + continue + + unit_binding = bound_units[0] + unit = mentions_by_id.get(unit_binding.target_mention_id) + if unit is None or unit.span == entity.span: + continue + evidence = { + (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)), + ) + ) + + return tuple(dispositions) + + +def _bound_relations( + text: str, + 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] = [] + quantity_entity = [b for b in bindings if b.binding_type == "quantity_entity"] + whole = next( + ( + binding + for binding in quantity_entity + if "%" not in by_id[binding.source_mention_id].surface + and by_id[binding.source_mention_id].surface.lower() + not in {"half", "third", "quarter"} + ), + None, + ) + for binding in quantity_entity: + quantity = by_id[binding.source_mention_id] + part = by_id[binding.target_mention_id] + canonical_part = min( + ( + mention + for mention in mentions + 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", + }: + continue + roles = [ + 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 + ), + ) + ) + + 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, + ) + 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() + ), + key=lambda mention: mention.span.start, + default=whole, + ) + roles = ( + 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), + ) + ) + + unary_delta_proposals = tuple( + proposal + for proposal in proposals + if proposal.family_id == "state_change.unary_delta" + ) + if len(unary_delta_proposals) == 1: + proposal = unary_delta_proposals[0] + if 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" + # Locate corresponding GroundedUnaryDeltaCue's cue_id + 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: + match = decrease_matches[0] + scale = next( + ( + m + for m in mentions + if m.kind == "quantity" and m.span.start == match.start("fraction") + ), + None, + ) + state_match = next( + ( + item + for item in _DECREASE_STATE_RE.finditer(text) + if item.start("state") < match.start("transition") + ), + None, + ) + 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") + ), + None, + ) + if state_match is not None + else None + ) + unit_binding_by_quantity = { + binding.source_mention_id: binding + for binding in bindings + if binding.binding_type == "quantity_unit" + } + base_candidates = [ + mention + for mention in mentions + if mention.kind == "quantity" + and mention.mention_id != (scale.mention_id if scale else None) + and mention.mention_id in unit_binding_by_quantity + and _sentence_contains_current_or_now(text, mention.span.start) + ] + if len(base_candidates) == 1 and scale is not None and state is not None: + base = base_candidates[0] + base_unit_binding = unit_binding_by_quantity.get(base.mention_id) + roles = [ + BoundRole("base_quantity", base.mention_id, base.kind, (base.span,)), + BoundRole("scale", scale.mention_id, scale.kind, (scale.span,)), + BoundRole("state_entity", state.mention_id, state.kind, (state.span,)), + BoundRole( + "transition", + f"span:{match.start('transition')}:{match.end('transition')}", + "span", + ( + 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 + ), + ) + ) + + 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, + ) + + agent = at("agent", "actor") + patient = at("patient", "actor") + quantity = at("quantity", "quantity") + obj = at("object", "object") + if all((agent, patient, quantity, obj)): + assert agent and patient and quantity and obj + roles = tuple( + BoundRole(name, mention.mention_id, mention.kind, (mention.span,)) + for name, mention in ( + ("agent", agent), + ("patient", patient), + ("quantity", quantity), + ("object", 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, + ) + for index, relation in enumerate(relations) + ) + + +def _bound_question_target( + text: str, mentions: tuple[GroundedMention, ...] +) -> BoundQuestionTarget | None: + """Extract and bind the question target from the problem text. + + Priority Cascade Order: + 1. Specific regex-based triggers: + - Proportional decrease delta: checked first using ``_DECREASE_DELTA_QUESTION_RE``. + If matched, returns a difference/delta/decrease target. + 2. General question clause extraction: + - Triggers on ``_QUESTION_ENTITY_RE``. + - If no match, but "?" is present in the text, returns an "unknown" target. + 3. Target classification of the question clause: + - "more" -> difference / delta / unknown direction. + - Initial state indicators ("were in", "was in", "started with", "originally") -> count / initial / inverse. + - Remaining indicators ("remaining", "left" in context) -> count / final / remaining. + - Aggregate indicators ("total", "altogether", "own") -> count / aggregate / forward. + - Portion percentage ("percent", "percentage") -> portion / final / forward. + - Portion fraction ("ratio", "fraction") -> portion / final / forward. + - Fallback -> count / final / forward. + """ + decrease_delta = _DECREASE_DELTA_QUESTION_RE.search(text) + if decrease_delta is not None: + entity_surface = decrease_delta.group("entity") + entity = next( + ( + 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(), + ) + return BoundQuestionTarget( + "difference", + entity_surface, + entity.mention_id if entity else None, + "delta_quantity", + (span,), + target_operator="difference", + target_state="delta", + target_direction="decrease", + ) + question = _QUESTION_ENTITY_RE.search(text) + if question is None: + if "?" not in text: + return None + qmark = text.index("?") + return BoundQuestionTarget( + "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() + question_lower = question_clause.lower() + if "more" in question.group(0).lower(): + target_type = "difference" + target_operator = "difference" + 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") + ): + target_type = "count" + target_operator = "count" + target_state = "initial" + target_direction = "inverse" + unknown_slot = "initial" + elif any(x in prefix for x in ("remaining", "left")): + target_type = "remaining" + target_operator = "count" + target_state = "final" + target_direction = "remaining" + unknown_slot = "remaining" + elif any(x in question_lower for x in ("total", "altogether", "own")): + target_type = "count" + target_operator = "count" + target_state = "aggregate" + target_direction = "forward" + unknown_slot = "count" + else: + target_type = "count" + target_operator = "count" + target_state = "current" + target_direction = "unknown" + unknown_slot = "count" + 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_operator=target_operator, + target_state=target_state, + target_direction=target_direction, + ) diff --git a/generate/problem_frame_builder.py b/generate/problem_frame_builder.py index dda08ad1..8582cee2 100644 --- a/generate/problem_frame_builder.py +++ b/generate/problem_frame_builder.py @@ -13,1236 +13,32 @@ Non-goals: from __future__ import annotations -import dataclasses -import re from fractions import Fraction -from generate.construction_affordances import ConstructionProposal, propose_construction -from generate.kernel_facts import ( - BoundRelation, - BoundRole, - CandidateRelation, - GroundedMention, - GroundedScalar, - GroundedUnit, - KernelHazard, - KernelProvenance, - MentionBinding, - RelationRole, - SourceSpan, +from generate.kernel_facts import GroundedScalar +from generate.problem_frame import GroundedUnaryDeltaCue, ProblemFrame, ProblemFrameBuilder +from generate.problem_frame_bound_relations import ( + _bound_question_target, + _bound_relations, + _quantity_kind_dispositions, ) -from generate.problem_frame import ( - BoundQuestionTarget, - GroundedUnaryDeltaCue, - ProblemFrame, - ProblemFrameBuilder, - QuantityKindDisposition, - QuestionTarget, +from generate.problem_frame_extractors import ( + _detect_question_target, + _extract_candidate_relations, + _extract_hazards, + _extract_process_frame_candidates, + _extract_unit_candidates, + _filter_scalar_candidates, + _scalar_to_grounded, ) -from generate.process_frames import ProcessFrame, all_frames -from language_packs.ambiguity_hazards import ( - AmbiguityHazard, - all_registered_surfaces, - lookup_hazards, +from generate.problem_frame_mentions import _extract_bindings, _extract_mentions +from generate.problem_frame_proposals import ( + _percent_partition_proposals, + _proportional_decrease_proposals, + _quantity_entity_proposals, + _unary_delta_proposals, ) -from language_packs.scalar_equivalence import ( - ScalarCandidate, - extract_scalar_candidates, -) -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", - } -) - -_ORDINAL_SUFFIX_RE: re.Pattern[str] = re.compile( - r"\b(half|third|quarter)\s+(place|position|grade|rank)\b", - re.IGNORECASE, -) - - -def surface_in_text(surface: str, text: str) -> bool: - """Match a registered surface at lexical, including punctuation, boundaries.""" - return ( - re.search( - rf"(? KernelHazard: - return KernelHazard( - hazard_id=hazard.hazard_id, - category=hazard.category, - surface=hazard.surface, - description=hazard.description, - context_required=hazard.context_required, - ) - - -def _extract_unit_candidates(text: str) -> tuple[GroundedUnit, ...]: - units: list[GroundedUnit] = [] - seen: set[tuple[str, int, int]] = set() - - for match in _UNIT_TOKEN_RE.finditer(text): - token = match.group(1) - token_lower = token.lower() - if token_lower in _UNIT_STOPWORDS: - continue - dim_fact = classify_dimension(token_lower) - if dim_fact is None: - continue - start = match.start(1) - end = match.end(1) - key = (token_lower, start, end) - if key in seen: - continue - seen.add(key) - span = SourceSpan(text[start:end], start, end) - provenance = KernelProvenance(kind="problem_text", source_spans=(span,)) - units.append( - GroundedUnit( - fact_id=f"unit-{len(units):04d}", - surface=token_lower, - dimension=dim_fact.dimension, - singular=dim_fact.singular, - provenance=provenance, - ) - ) - - return tuple( - sorted(units, key=lambda u: (u.provenance.source_spans[0].start, u.surface)) - ) - - -def _extract_hazards(text: str) -> tuple[KernelHazard, ...]: - text_lower = text.lower() - hazards: list[KernelHazard] = [] - seen: set[str] = set() - - for surface in all_registered_surfaces(): - if not surface_in_text(surface, text_lower): - continue - for hazard in lookup_hazards(surface): - if hazard.hazard_id in seen: - continue - seen.add(hazard.hazard_id) - hazards.append(_hazard_to_kernel(hazard)) - - if "%" in text: - for hazard in lookup_hazards("percent"): - if hazard.hazard_id in seen: - continue - seen.add(hazard.hazard_id) - hazards.append(_hazard_to_kernel(hazard)) - - return tuple(sorted(hazards, key=lambda h: h.hazard_id)) - - -def _is_ordinal_scalar_span(text: str, start: int, end: int) -> bool: - """Refuse fraction readings for ordinals like ``third place``.""" - window_start = max(0, start - 20) - window_end = min(len(text), end + 20) - window = text[window_start:window_end] - for match in _ORDINAL_SUFFIX_RE.finditer(window): - abs_start = window_start + match.start() - abs_end = window_start + match.end() - if start >= abs_start and end <= abs_end: - return True - return False - - -def _filter_scalar_candidates( - text: str, - candidates: tuple[ScalarCandidate, ...], -) -> tuple[ScalarCandidate, ...]: - kept: list[ScalarCandidate] = [] - for candidate in candidates: - if candidate.source_span is None: - kept.append(candidate) - continue - start, end = candidate.source_span - if _is_ordinal_scalar_span(text, start, end): - continue - kept.append(candidate) - return tuple(kept) - - -def _trigger_span(text: str, trigger: str) -> SourceSpan | None: - match = re.search( - rf"(? bool: - start = max( - text.rfind(".", 0, index), - text.rfind("?", 0, index), - text.rfind("!", 0, index), - ) - end_candidates = [ - pos - for pos in ( - text.find(".", index), - text.find("?", index), - text.find("!", index), - ) - if pos != -1 - ] - end = min(end_candidates) if end_candidates else len(text) - sentence = text[start + 1 : end].lower() - return "current" in sentence or "now" in sentence - - -def _extract_process_frame_candidates(text: str) -> tuple[ProcessFrame, ...]: - text_lower = text.lower() - matched: dict[str, ProcessFrame] = {} - - for frame in all_frames(): - for trigger in frame.trigger_surfaces: - if surface_in_text(trigger, text_lower): - matched[frame.name] = frame - break - - return tuple(matched[name] for name in sorted(matched)) - - -def _frame_roles(frame: ProcessFrame) -> tuple[RelationRole, ...]: - roles: list[RelationRole] = [] - for role in frame.required_roles: - roles.append(RelationRole(role.name, True, role.description)) - for role in frame.optional_roles: - roles.append(RelationRole(role.name, False, role.description)) - return tuple(roles) - - -def _extract_candidate_relations( - text: str, - frames: tuple[ProcessFrame, ...], -) -> tuple[CandidateRelation, ...]: - relations: list[CandidateRelation] = [] - - for frame in frames: - span: SourceSpan | None = None - for trigger in frame.trigger_surfaces: - span = _trigger_span(text, trigger) - if span is not None: - break - provenance = ( - KernelProvenance(kind="problem_text", source_spans=(span,)) - if span is not None - else None - ) - frame_hazards = tuple( - KernelHazard( - hazard_id=f"frame-{frame.name}-{category}", - category=category, - surface=frame.name, - description=f"Process frame {frame.name} hazard {category}", - ) - for category in frame.hazards - ) - relations.append( - CandidateRelation( - relation_id=f"rel-{frame.name}", - relation_type=frame.candidate_relation, - roles=_frame_roles(frame), - provenance=provenance, - hazards=frame_hazards, - ) - ) - - return tuple(relations) - - -def _scalar_to_grounded( - candidate: ScalarCandidate, - text: str, - index: int, -) -> GroundedScalar | None: - if candidate.source_span is None or candidate.source_surface is None: - return None - - start, end = candidate.source_span - span = SourceSpan(candidate.source_surface, start, end) - provenance = KernelProvenance(kind="problem_text", source_spans=(span,)) - hazards = tuple( - KernelHazard( - hazard_id=hid, - category=hid, - surface=candidate.surface, - description=f"Scalar hazard {hid}", - ) - for hid in candidate.hazards - ) - return GroundedScalar( - fact_id=f"scalar-{index:04d}", - surface=candidate.surface, - value=candidate.canonical, - provenance=provenance, - hazards=hazards, - ) - - -def _detect_question_target(text: str) -> QuestionTarget | None: - text_lower = text.lower() - if "how many" in text_lower: - return QuestionTarget("how many", "count") - if "how much" in text_lower: - return QuestionTarget("how much", "quantity") - if "?" in text: - return QuestionTarget("?", "unknown") - return None - - -_ENTITY_AFTER_QUANTITY_RE = re.compile( - r"(?P\d+(?:\.\d+)?\s*%?)\s+(?:of\s+(?:the\s+)?)?" - r"(?P[A-Za-z][A-Za-z'-]*)", - re.IGNORECASE, -) -_FRACTION_ENTITY_RE = re.compile( - r"\b(?Phalf|third|quarter)\b\s+(?:of\s+(?:the\s+)?|are\s+|the\s+)?" - r"(?P[A-Za-z][A-Za-z'-]*)", - re.IGNORECASE, -) -_QUESTION_ENTITY_RE = re.compile( - r"\bhow\s+(?:many|much)\s+(?:more\s+)?(?P[A-Za-z][A-Za-z'-]*)", - re.IGNORECASE, -) -_COPULAR_PARTITION_RE = re.compile( - r"\b(?Phalf|third|quarter)\b\s+of\s+(?:the\s+)?" - r"(?P[A-Za-z][A-Za-z'-]*)\s+(?:are|is)\s+(?P[A-Za-z][A-Za-z'-]*)", - re.IGNORECASE, -) -_DECREASE_TO_FRACTION_RE = re.compile( - r"(?Pdecrease\s+to)\s+(?P\d+\s*/\s*\d+)\s+of", - re.IGNORECASE, -) -_PERCENT_OF_PROPOSAL_RE = re.compile( - r"\b\d+(?:\.\d+)?\s*%\s+of\b", - re.IGNORECASE, -) -_DECREASE_STATE_RE = re.compile( - r"(?P[A-Za-z][A-Za-z'-]*)\s+will\s+decrease\s+to", - re.IGNORECASE, -) -_DECREASE_DELTA_QUESTION_RE = re.compile( - r"\bwhat\s+will\s+the\s+(?P[A-Za-z][A-Za-z'-]*)\s+decrease\s+by\??", - re.IGNORECASE, -) - -_ACTOR_VERB_RE = re.compile( - r"\b(?P[A-Z][A-Za-z'-]*)\s+" - r"(?:gave|gives|give|received|receives|spent|spends|ate|eats|bought|buys|sold|sells)\b" -) -_TRANSFER_RE = re.compile( - r"\b(?P[A-Z][A-Za-z'-]*)\s+(?:gave|gives|give|handed|passed)\s+" - r"(?P[A-Z][A-Za-z'-]*)\s+" - r"(?P\d+(?:\.\d+)?)\s+(?P[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_CONFUSER_SURFACES: tuple[str, ...] = ( - "each", - "fewer than", - "greater than", - "less than", - "more than", - "per", - "percent", - "percentage", - "ratio", -) - - -def _proportional_decrease_proposals(text: str) -> tuple[ConstructionProposal, ...]: - """Propose the one authorized proposal-first construction from its chunk.""" - matches = tuple(_DECREASE_TO_FRACTION_RE.finditer(text)) - if len(matches) != 1: - return () - match = matches[0] - evidence = SourceSpan( - text[match.start() : match.end()], - match.start(), - match.end(), - ) - return ( - propose_construction( - "proportional_change.decrease_to_fraction", - (evidence,), - ), - ) - - -def _percent_partition_proposals( - text: str, - frames: tuple[ProcessFrame, ...], -) -> tuple[ConstructionProposal, ...]: - """Propose percent partition from a process cue plus explicit percent-of.""" - frame_names = {frame.name for frame in frames} - if not frame_names & {"partition", "consumption"}: - return () - - evidence_spans = tuple( - SourceSpan(text[match.start() : match.end()], match.start(), match.end()) - for match in _PERCENT_OF_PROPOSAL_RE.finditer(text) - ) - if not evidence_spans: - return () - - return ( - propose_construction( - "partition.percent_partition", - evidence_spans, - ), - ) - - -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 - ) - sentence_end = min(sentence_ends, default=len(text)) - tail = text[end:sentence_end].lstrip().lower() - 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, ...], - frames: tuple[ProcessFrame, ...], -) -> tuple[ConstructionProposal, ...]: - """Propose one narrow local quantity/entity cue from existing extraction. - - The family is intentionally unavailable when another process frame or a - rate/comparison/percent surface is active. Such text needs a different - family to interpret it; this seam never selects the nearest noun. - """ - - if len(quantities) != 1 or frames: - return () - if any( - surface_in_text(surface, text) for surface in _QUANTITY_ENTITY_CONFUSER_SURFACES - ): - return () - - matches = tuple(_ENTITY_AFTER_QUANTITY_RE.finditer(text)) - if len(matches) != 1: - return () - match = matches[0] - if "%" in match.group("quantity"): - return () - if match.group("entity").lower() in _QUANTITY_ENTITY_PRONOUNS: - return () - if _has_list_or_enumeration_suffix(text, match.end("entity")): - return () - - quantity_span = quantities[0].provenance.source_spans[0] - if quantity_span.start != match.start("quantity") or quantity_span.end != match.end( - "quantity" - ): - return () - - evidence = SourceSpan( - text[match.start() : match.end()], - match.start(), - match.end(), - ) - 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.""" - matches = list(re.finditer(r"\b(gained|lost)\b", text)) - if len(matches) != 1: - return () - match = matches[0] - - # 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 () - - # Competing / blocking surfaces - confusers = { - "percent", - "percentage", - "%", - "per", - "each", - "ratio", - "than", - "more than", - "less than", - "fewer than", - "greater than", - "times as", - } - 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 = { - "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 - for coord in {"and", "or"}: - if re.search(rf"\b{coord}\b", text, re.IGNORECASE): - return () - if "," in text: - return () - - evidence = SourceSpan( - text[match.start() : match.end()], - match.start(), - match.end(), - ) - return (propose_construction("state_change.unary_delta", (evidence,)),) - - -def _extract_mentions( - text: str, - quantities: tuple[GroundedScalar, ...], - units: tuple[GroundedUnit, ...], -) -> tuple[GroundedMention, ...]: - records: dict[tuple[str, int, int], GroundedMention] = {} - - def add(kind: str, start: int, end: int, *, fact_id: str | None = None) -> None: - key = (kind, start, end) - 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, - ) - - for quantity in quantities: - span = quantity.provenance.source_spans[0] - add("quantity", span.start, span.end, fact_id=quantity.fact_id) - 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 match in pattern.finditer(text): - add("object", match.start("entity"), match.end("entity")) - for match in _COPULAR_PARTITION_RE.finditer(text): - add("object", match.start("whole"), match.end("whole")) - add("object", match.start("part"), match.end("part")) - for match in _DECREASE_STATE_RE.finditer(text): - add("object", match.start("state"), match.end("state")) - for match in _ACTOR_VERB_RE.finditer(text): - add("actor", match.start("actor"), match.end("actor")) - for match in _TRANSFER_RE.finditer(text): - add("actor", match.start("agent"), match.end("agent")) - add("actor", match.start("patient"), match.end("patient")) - add("object", match.start("object"), match.end("object")) - - ordered = sorted( - records.values(), - key=lambda m: (m.span.start, m.span.end, m.kind, m.surface.lower()), - ) - return tuple( - GroundedMention( - 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) - ) - - -def _extract_bindings( - text: str, - mentions: tuple[GroundedMention, ...], -) -> tuple[MentionBinding, ...]: - by_span_kind = {(m.span.start, m.span.end, m.kind): m for m in mentions} - quantities = [m for m in mentions if m.kind == "quantity"] - bindings: list[MentionBinding] = [] - seen: set[tuple[str, str, str]] = set() - - 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), - ) - ) - - 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") - ) - if entity is None: - continue - 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"] - for quantity in quantities: - following = [ - unit - for unit in units - if unit.span.start >= quantity.span.end - 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) - ) - - -def _quantity_kind_dispositions( - text: str, - mentions: tuple[GroundedMention, ...], - bindings: tuple[MentionBinding, ...], - proposals: tuple[ConstructionProposal, ...], -) -> tuple[QuantityKindDisposition, ...]: - """Close kind only for the exact proposal-backed local binding.""" - - quantity_entity_proposals = tuple( - proposal - for proposal in proposals - if proposal.family_id == "binding.quantity_entity" - ) - if len(quantity_entity_proposals) != 1: - return () - quantity_entity_proposal = quantity_entity_proposals[0] - - mentions_by_id = {mention.mention_id: mention for mention in mentions} - unit_bindings: dict[str, list[MentionBinding]] = {} - for binding in bindings: - if binding.binding_type == "quantity_unit": - unit_bindings.setdefault(binding.source_mention_id, []).append(binding) - - dispositions: list[QuantityKindDisposition] = [] - for binding in bindings: - if binding.binding_type != "quantity_entity": - continue - quantity = mentions_by_id.get(binding.source_mention_id) - entity = mentions_by_id.get(binding.target_mention_id) - if quantity is None or entity is None or quantity.fact_id is None: - 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, - ) - ) - continue - if len(bound_units) != 1: - continue - - unit_binding = bound_units[0] - unit = mentions_by_id.get(unit_binding.target_mention_id) - if unit is None or unit.span == entity.span: - continue - evidence = { - (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)), - ) - ) - - return tuple(dispositions) - - -def _bound_relations( - text: str, - 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] = [] - quantity_entity = [b for b in bindings if b.binding_type == "quantity_entity"] - whole = next( - ( - binding - for binding in quantity_entity - if "%" not in by_id[binding.source_mention_id].surface - and by_id[binding.source_mention_id].surface.lower() - not in {"half", "third", "quarter"} - ), - None, - ) - for binding in quantity_entity: - quantity = by_id[binding.source_mention_id] - part = by_id[binding.target_mention_id] - canonical_part = min( - ( - mention - for mention in mentions - 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", - }: - continue - roles = [ - 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 - ), - ) - ) - - 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, - ) - 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() - ), - key=lambda mention: mention.span.start, - default=whole, - ) - roles = ( - 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), - ) - ) - - unary_delta_proposals = tuple( - proposal - for proposal in proposals - if proposal.family_id == "state_change.unary_delta" - ) - if len(unary_delta_proposals) == 1: - proposal = unary_delta_proposals[0] - if 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" - # Locate corresponding GroundedUnaryDeltaCue's cue_id - 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: - match = decrease_matches[0] - scale = next( - ( - m - for m in mentions - if m.kind == "quantity" and m.span.start == match.start("fraction") - ), - None, - ) - state_match = next( - ( - item - for item in _DECREASE_STATE_RE.finditer(text) - if item.start("state") < match.start("transition") - ), - None, - ) - 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") - ), - None, - ) - if state_match is not None - else None - ) - unit_binding_by_quantity = { - binding.source_mention_id: binding - for binding in bindings - if binding.binding_type == "quantity_unit" - } - base_candidates = [ - mention - for mention in mentions - if mention.kind == "quantity" - and mention.mention_id != (scale.mention_id if scale else None) - and mention.mention_id in unit_binding_by_quantity - and _sentence_contains_current_or_now(text, mention.span.start) - ] - if len(base_candidates) == 1 and scale is not None and state is not None: - base = base_candidates[0] - base_unit_binding = unit_binding_by_quantity.get(base.mention_id) - roles = [ - BoundRole("base_quantity", base.mention_id, base.kind, (base.span,)), - BoundRole("scale", scale.mention_id, scale.kind, (scale.span,)), - BoundRole("state_entity", state.mention_id, state.kind, (state.span,)), - BoundRole( - "transition", - f"span:{match.start('transition')}:{match.end('transition')}", - "span", - ( - 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 - ), - ) - ) - - 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, - ) - - agent = at("agent", "actor") - patient = at("patient", "actor") - quantity = at("quantity", "quantity") - obj = at("object", "object") - if all((agent, patient, quantity, obj)): - assert agent and patient and quantity and obj - roles = tuple( - BoundRole(name, mention.mention_id, mention.kind, (mention.span,)) - for name, mention in ( - ("agent", agent), - ("patient", patient), - ("quantity", quantity), - ("object", 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, - ) - for index, relation in enumerate(relations) - ) - - -def _bound_question_target( - text: str, mentions: tuple[GroundedMention, ...] -) -> BoundQuestionTarget | None: - """Extract and bind the question target from the problem text. - - Priority Cascade Order: - 1. Specific regex-based triggers: - - Proportional decrease delta: checked first using ``_DECREASE_DELTA_QUESTION_RE``. - If matched, returns a difference/delta/decrease target. - 2. General question clause extraction: - - Triggers on ``_QUESTION_ENTITY_RE``. - - If no match, but "?" is present in the text, returns an "unknown" target. - 3. Target classification of the question clause: - - "more" -> difference / delta / unknown direction. - - Initial state indicators ("were in", "was in", "started with", "originally") -> count / initial / inverse. - - Remaining indicators ("remaining", "left" in context) -> count / final / remaining. - - Aggregate indicators ("total", "altogether", "own") -> count / aggregate / forward. - - Portion percentage ("percent", "percentage") -> portion / final / forward. - - Portion fraction ("ratio", "fraction") -> portion / final / forward. - - Fallback -> count / final / forward. - """ - decrease_delta = _DECREASE_DELTA_QUESTION_RE.search(text) - if decrease_delta is not None: - entity_surface = decrease_delta.group("entity") - entity = next( - ( - 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(), - ) - return BoundQuestionTarget( - "difference", - entity_surface, - entity.mention_id if entity else None, - "delta_quantity", - (span,), - target_operator="difference", - target_state="delta", - target_direction="decrease", - ) - question = _QUESTION_ENTITY_RE.search(text) - if question is None: - if "?" not in text: - return None - qmark = text.index("?") - return BoundQuestionTarget( - "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() - question_lower = question_clause.lower() - if "more" in question.group(0).lower(): - target_type = "difference" - target_operator = "difference" - 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") - ): - target_type = "count" - target_operator = "count" - target_state = "initial" - target_direction = "inverse" - unknown_slot = "initial" - elif any(x in prefix for x in ("remaining", "left")): - target_type = "remaining" - target_operator = "count" - target_state = "final" - target_direction = "remaining" - unknown_slot = "remaining" - elif any(x in question_lower for x in ("total", "altogether", "own")): - target_type = "count" - target_operator = "count" - target_state = "aggregate" - target_direction = "forward" - unknown_slot = "count" - else: - target_type = "count" - target_operator = "count" - target_state = "current" - target_direction = "unknown" - unknown_slot = "count" - 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_operator=target_operator, - target_state=target_state, - target_direction=target_direction, - ) +from language_packs.scalar_equivalence import extract_scalar_candidates def build_problem_frame(problem_text: str) -> ProblemFrame: @@ -1326,18 +122,20 @@ def build_problem_frame(problem_text: str) -> ProblemFrame: builder.add_object(mention.surface) for binding in bindings: builder.add_binding(binding) + + proposals_for_grounding = (*quantity_entity_proposals, *unary_delta_proposals) for disposition in _quantity_kind_dispositions( problem_text, mentions, bindings, - (*quantity_entity_proposals, *unary_delta_proposals), + proposals_for_grounding, ): builder.add_quantity_kind_disposition(disposition) for relation in _bound_relations( problem_text, mentions, bindings, - (*quantity_entity_proposals, *unary_delta_proposals), + proposals_for_grounding, builder.unary_delta_cues, ): builder.add_bound_relation(relation) @@ -1345,31 +143,7 @@ def build_problem_frame(problem_text: str) -> ProblemFrame: if bound_target is not None: builder.set_bound_question_target(bound_target) - initial_frame = builder.build() - - from generate.problem_frame_contracts import ( - assess_contracts, - get_contract_family_id, - ) - from generate.construction_affordances import make_proposal - - assessments = assess_contracts(initial_frame) - proposals = list(initial_frame.proposals) - proposed_family_ids = {proposal.family_id for proposal in proposals} - for assessment in assessments: - family_id = get_contract_family_id(assessment.candidate_organ) - if family_id is not None and family_id not in proposed_family_ids: - proposal = make_proposal( - family_id=family_id, - evidence_spans=assessment.evidence_spans, - assessment_runnable=assessment.runnable, - missing_roles=assessment.missing_bindings, - active_hazards=assessment.unresolved_hazards, - ) - proposals.append(proposal) - proposed_family_ids.add(family_id) - - return dataclasses.replace(initial_frame, proposals=tuple(proposals)) + return builder.build() def recognized_scalar_surfaces(frame: ProblemFrame) -> tuple[str, ...]: diff --git a/generate/problem_frame_extractors.py b/generate/problem_frame_extractors.py new file mode 100644 index 00000000..de3653d1 --- /dev/null +++ b/generate/problem_frame_extractors.py @@ -0,0 +1,305 @@ +"""ProblemFrame extraction helpers. + +This module owns raw/evidenced surface observation for ProblemFrame construction. +It is intentionally phase-local: extraction observes text and substrate facts; it +does not propose constructions, bind mentions, assess contracts, or serve. +""" + +from __future__ import annotations + +import re + +from generate.kernel_facts import ( + CandidateRelation, + GroundedScalar, + GroundedUnit, + KernelHazard, + KernelProvenance, + RelationRole, + SourceSpan, +) +from generate.problem_frame import QuestionTarget +from generate.process_frames import ProcessFrame, all_frames +from language_packs.ambiguity_hazards import ( + AmbiguityHazard, + all_registered_surfaces, + lookup_hazards, +) +from language_packs.scalar_equivalence import ScalarCandidate +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", + } +) + +_ORDINAL_SUFFIX_RE: re.Pattern[str] = re.compile( + r"\b(half|third|quarter)\s+(place|position|grade|rank)\b", + re.IGNORECASE, +) + + +def surface_in_text(surface: str, text: str) -> bool: + """Match a registered surface at lexical, including punctuation, boundaries.""" + return ( + re.search( + rf"(? KernelHazard: + return KernelHazard( + hazard_id=hazard.hazard_id, + category=hazard.category, + surface=hazard.surface, + description=hazard.description, + context_required=hazard.context_required, + ) + + +def _extract_unit_candidates(text: str) -> tuple[GroundedUnit, ...]: + units: list[GroundedUnit] = [] + seen: set[tuple[str, int, int]] = set() + + for match in _UNIT_TOKEN_RE.finditer(text): + token = match.group(1) + token_lower = token.lower() + if token_lower in _UNIT_STOPWORDS: + continue + dim_fact = classify_dimension(token_lower) + if dim_fact is None: + continue + start = match.start(1) + end = match.end(1) + key = (token_lower, start, end) + if key in seen: + continue + seen.add(key) + span = SourceSpan(text[start:end], start, end) + provenance = KernelProvenance(kind="problem_text", source_spans=(span,)) + units.append( + GroundedUnit( + fact_id=f"unit-{len(units):04d}", + surface=token_lower, + dimension=dim_fact.dimension, + singular=dim_fact.singular, + provenance=provenance, + ) + ) + + return tuple( + sorted(units, key=lambda u: (u.provenance.source_spans[0].start, u.surface)) + ) + + +def _extract_hazards(text: str) -> tuple[KernelHazard, ...]: + text_lower = text.lower() + hazards: list[KernelHazard] = [] + seen: set[str] = set() + + for surface in all_registered_surfaces(): + if not surface_in_text(surface, text_lower): + continue + for hazard in lookup_hazards(surface): + if hazard.hazard_id in seen: + continue + seen.add(hazard.hazard_id) + hazards.append(_hazard_to_kernel(hazard)) + + if "%" in text: + for hazard in lookup_hazards("percent"): + if hazard.hazard_id in seen: + continue + seen.add(hazard.hazard_id) + hazards.append(_hazard_to_kernel(hazard)) + + return tuple(sorted(hazards, key=lambda h: h.hazard_id)) + + +def _is_ordinal_scalar_span(text: str, start: int, end: int) -> bool: + """Refuse fraction readings for ordinals like ``third place``.""" + window_start = max(0, start - 20) + window_end = min(len(text), end + 20) + window = text[window_start:window_end] + for match in _ORDINAL_SUFFIX_RE.finditer(window): + abs_start = window_start + match.start() + abs_end = window_start + match.end() + if start >= abs_start and end <= abs_end: + return True + return False + + +def _filter_scalar_candidates( + text: str, + candidates: tuple[ScalarCandidate, ...], +) -> tuple[ScalarCandidate, ...]: + kept: list[ScalarCandidate] = [] + for candidate in candidates: + if candidate.source_span is None: + kept.append(candidate) + continue + start, end = candidate.source_span + if _is_ordinal_scalar_span(text, start, end): + continue + kept.append(candidate) + return tuple(kept) + + +def _trigger_span(text: str, trigger: str) -> SourceSpan | None: + match = re.search( + rf"(? bool: + start = max( + text.rfind(".", 0, index), + text.rfind("?", 0, index), + text.rfind("!", 0, index), + ) + end_candidates = [ + pos + for pos in ( + text.find(".", index), + text.find("?", index), + text.find("!", index), + ) + if pos != -1 + ] + end = min(end_candidates) if end_candidates else len(text) + sentence = text[start + 1 : end].lower() + return "current" in sentence or "now" in sentence + + +def _extract_process_frame_candidates(text: str) -> tuple[ProcessFrame, ...]: + text_lower = text.lower() + matched: dict[str, ProcessFrame] = {} + + for frame in all_frames(): + for trigger in frame.trigger_surfaces: + if surface_in_text(trigger, text_lower): + matched[frame.name] = frame + break + + return tuple(matched[name] for name in sorted(matched)) + + +def _frame_roles(frame: ProcessFrame) -> tuple[RelationRole, ...]: + roles: list[RelationRole] = [] + for role in frame.required_roles: + roles.append(RelationRole(role.name, True, role.description)) + for role in frame.optional_roles: + roles.append(RelationRole(role.name, False, role.description)) + return tuple(roles) + + +def _extract_candidate_relations( + text: str, + frames: tuple[ProcessFrame, ...], +) -> tuple[CandidateRelation, ...]: + relations: list[CandidateRelation] = [] + + for frame in frames: + span: SourceSpan | None = None + for trigger in frame.trigger_surfaces: + span = _trigger_span(text, trigger) + if span is not None: + break + provenance = ( + KernelProvenance(kind="problem_text", source_spans=(span,)) + if span is not None + else None + ) + frame_hazards = tuple( + KernelHazard( + hazard_id=f"frame-{frame.name}-{category}", + category=category, + surface=frame.name, + description=f"Process frame {frame.name} hazard {category}", + ) + for category in frame.hazards + ) + relations.append( + CandidateRelation( + relation_id=f"rel-{frame.name}", + relation_type=frame.candidate_relation, + roles=_frame_roles(frame), + provenance=provenance, + hazards=frame_hazards, + ) + ) + + return tuple(relations) + + +def _scalar_to_grounded( + candidate: ScalarCandidate, + text: str, + index: int, +) -> GroundedScalar | None: + if candidate.source_span is None or candidate.source_surface is None: + return None + + start, end = candidate.source_span + span = SourceSpan(candidate.source_surface, start, end) + provenance = KernelProvenance(kind="problem_text", source_spans=(span,)) + hazards = tuple( + KernelHazard( + hazard_id=hid, + category=hid, + surface=candidate.surface, + description=f"Scalar hazard {hid}", + ) + for hid in candidate.hazards + ) + return GroundedScalar( + fact_id=f"scalar-{index:04d}", + surface=candidate.surface, + value=candidate.canonical, + provenance=provenance, + hazards=hazards, + ) + + +def _detect_question_target(text: str) -> QuestionTarget | None: + text_lower = text.lower() + if "how many" in text_lower: + return QuestionTarget("how many", "count") + if "how much" in text_lower: + return QuestionTarget("how much", "quantity") + if "?" in text: + return QuestionTarget("?", "unknown") + return None diff --git a/generate/problem_frame_mentions.py b/generate/problem_frame_mentions.py new file mode 100644 index 00000000..b5551299 --- /dev/null +++ b/generate/problem_frame_mentions.py @@ -0,0 +1,175 @@ +"""ProblemFrame mention and binding helpers. + +This module owns grounded mention extraction and mention-binding edges. It does +not create construction proposals, assess contracts, or mutate builder state. +""" + +from __future__ import annotations + +import re + +from generate.kernel_facts import ( + GroundedMention, + GroundedScalar, + GroundedUnit, + MentionBinding, + SourceSpan, +) + +_ENTITY_AFTER_QUANTITY_RE = re.compile( + r"(?P\d+(?:\.\d+)?\s*%?)\s+(?:of\s+(?:the\s+)?)?" + r"(?P[A-Za-z][A-Za-z'-]*)", + re.IGNORECASE, +) +_FRACTION_ENTITY_RE = re.compile( + r"\b(?Phalf|third|quarter)\b\s+(?:of\s+(?:the\s+)?|are\s+|the\s+)?" + r"(?P[A-Za-z][A-Za-z'-]*)", + re.IGNORECASE, +) +_QUESTION_ENTITY_RE = re.compile( + r"\bhow\s+(?:many|much)\s+(?:more\s+)?(?P[A-Za-z][A-Za-z'-]*)", + re.IGNORECASE, +) +_COPULAR_PARTITION_RE = re.compile( + r"\b(?Phalf|third|quarter)\b\s+of\s+(?:the\s+)?" + r"(?P[A-Za-z][A-Za-z'-]*)\s+(?:are|is)\s+(?P[A-Za-z][A-Za-z'-]*)", + re.IGNORECASE, +) +_DECREASE_STATE_RE = re.compile( + r"(?P[A-Za-z][A-Za-z'-]*)\s+will\s+decrease\s+to", + re.IGNORECASE, +) +_ACTOR_VERB_RE = re.compile( + r"\b(?P[A-Z][A-Za-z'-]*)\s+" + r"(?:gave|gives|give|received|receives|spent|spends|ate|eats|bought|buys|sold|sells)\b" +) +_TRANSFER_RE = re.compile( + r"\b(?P[A-Z][A-Za-z'-]*)\s+(?:gave|gives|give|handed|passed)\s+" + r"(?P[A-Z][A-Za-z'-]*)\s+" + r"(?P\d+(?:\.\d+)?)\s+(?P[A-Za-z][A-Za-z'-]*)", +) + + +def _extract_mentions( + text: str, + quantities: tuple[GroundedScalar, ...], + units: tuple[GroundedUnit, ...], +) -> tuple[GroundedMention, ...]: + records: dict[tuple[str, int, int], GroundedMention] = {} + + def add(kind: str, start: int, end: int, *, fact_id: str | None = None) -> None: + key = (kind, start, end) + 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, + ) + + for quantity in quantities: + span = quantity.provenance.source_spans[0] + add("quantity", span.start, span.end, fact_id=quantity.fact_id) + 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 match in pattern.finditer(text): + add("object", match.start("entity"), match.end("entity")) + for match in _COPULAR_PARTITION_RE.finditer(text): + add("object", match.start("whole"), match.end("whole")) + add("object", match.start("part"), match.end("part")) + for match in _DECREASE_STATE_RE.finditer(text): + add("object", match.start("state"), match.end("state")) + for match in _ACTOR_VERB_RE.finditer(text): + add("actor", match.start("actor"), match.end("actor")) + for match in _TRANSFER_RE.finditer(text): + add("actor", match.start("agent"), match.end("agent")) + add("actor", match.start("patient"), match.end("patient")) + add("object", match.start("object"), match.end("object")) + + ordered = sorted( + records.values(), + key=lambda m: (m.span.start, m.span.end, m.kind, m.surface.lower()), + ) + return tuple( + GroundedMention( + 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) + ) + + +def _extract_bindings( + text: str, + mentions: tuple[GroundedMention, ...], +) -> tuple[MentionBinding, ...]: + by_span_kind = {(m.span.start, m.span.end, m.kind): m for m in mentions} + quantities = [m for m in mentions if m.kind == "quantity"] + bindings: list[MentionBinding] = [] + seen: set[tuple[str, str, str]] = set() + + 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), + ) + ) + + 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") + ) + if entity is None: + continue + 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"] + for quantity in quantities: + following = [ + unit + for unit in units + if unit.span.start >= quantity.span.end + 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) + ) diff --git a/generate/problem_frame_proposals.py b/generate/problem_frame_proposals.py new file mode 100644 index 00000000..da3a8de9 --- /dev/null +++ b/generate/problem_frame_proposals.py @@ -0,0 +1,265 @@ +"""ProblemFrame construction proposal helpers. + +This module owns pre-assessment construction hypotheses. It may create +``ConstructionProposal`` records from exact surface/process evidence, but it does +not bind roles, assess contracts, or serve. +""" + +from __future__ import annotations + +import re + +from generate.construction_affordances import ConstructionProposal, propose_construction +from generate.kernel_facts import GroundedScalar, SourceSpan +from generate.process_frames import ProcessFrame + +from generate.problem_frame_extractors import surface_in_text + +_DECREASE_TO_FRACTION_RE = re.compile( + r"(?Pdecrease\s+to)\s+(?P\d+\s*/\s*\d+)\s+of", + re.IGNORECASE, +) +_PERCENT_OF_PROPOSAL_RE = re.compile( + r"\b\d+(?:\.\d+)?\s*%\s+of\b", + re.IGNORECASE, +) + +# Duplicated intentionally to preserve phase-local ownership. +# Do not import another phase's internals just to share this regex. +_ENTITY_AFTER_QUANTITY_RE = re.compile( + r"(?P\d+(?:\.\d+)?\s*%?)\s+(?:of\s+(?:the\s+)?)?" + r"(?P[A-Za-z][A-Za-z'-]*)", + re.IGNORECASE, +) + +_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", + "fewer than", + "greater than", + "less than", + "more than", + "per", + "percent", + "percentage", + "ratio", +) + + +def _proportional_decrease_proposals(text: str) -> tuple[ConstructionProposal, ...]: + """Propose the one authorized proposal-first construction from its chunk.""" + matches = tuple(_DECREASE_TO_FRACTION_RE.finditer(text)) + if len(matches) != 1: + return () + match = matches[0] + evidence = SourceSpan( + text[match.start() : match.end()], + match.start(), + match.end(), + ) + return ( + propose_construction( + "proportional_change.decrease_to_fraction", + (evidence,), + ), + ) + + +def _percent_partition_proposals( + text: str, + frames: tuple[ProcessFrame, ...], +) -> tuple[ConstructionProposal, ...]: + """Propose percent partition from a process cue plus explicit percent-of.""" + frame_names = {frame.name for frame in frames} + if not frame_names & {"partition", "consumption"}: + return () + + evidence_spans = tuple( + SourceSpan(text[match.start() : match.end()], match.start(), match.end()) + for match in _PERCENT_OF_PROPOSAL_RE.finditer(text) + ) + if not evidence_spans: + return () + + return ( + propose_construction( + "partition.percent_partition", + evidence_spans, + ), + ) + + +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 + ) + sentence_end = min(sentence_ends, default=len(text)) + tail = text[end:sentence_end].lstrip().lower() + 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, ...], + frames: tuple[ProcessFrame, ...], +) -> tuple[ConstructionProposal, ...]: + """Propose one narrow local quantity/entity cue from existing extraction. + + The family is intentionally unavailable when another process frame or a + rate/comparison/percent surface is active. Such text needs a different + family to interpret it; this seam never selects the nearest noun. + """ + + if len(quantities) != 1 or frames: + return () + if any( + surface_in_text(surface, text) for surface in _QUANTITY_ENTITY_CONFUSER_SURFACES + ): + return () + + matches = tuple(_ENTITY_AFTER_QUANTITY_RE.finditer(text)) + if len(matches) != 1: + return () + match = matches[0] + if "%" in match.group("quantity"): + return () + if match.group("entity").lower() in _QUANTITY_ENTITY_PRONOUNS: + return () + if _has_list_or_enumeration_suffix(text, match.end("entity")): + return () + + quantity_span = quantities[0].provenance.source_spans[0] + if quantity_span.start != match.start("quantity") or quantity_span.end != match.end( + "quantity" + ): + return () + + evidence = SourceSpan( + text[match.start() : match.end()], + match.start(), + match.end(), + ) + 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.""" + matches = list(re.finditer(r"\b(gained|lost)\b", text)) + if len(matches) != 1: + return () + match = matches[0] + + # 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 () + + # Competing / blocking surfaces + confusers = { + "percent", + "percentage", + "%", + "per", + "each", + "ratio", + "than", + "more than", + "less than", + "fewer than", + "greater than", + "times as", + } + 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 = { + "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 + for coord in {"and", "or"}: + if re.search(rf"\b{coord}\b", text, re.IGNORECASE): + return () + if "," in text: + return () + + evidence = SourceSpan( + text[match.start() : match.end()], + match.start(), + match.end(), + ) + return (propose_construction("state_change.unary_delta", (evidence,)),) diff --git a/tests/test_problem_frame_phase_boundaries.py b/tests/test_problem_frame_phase_boundaries.py new file mode 100644 index 00000000..8ab029ea --- /dev/null +++ b/tests/test_problem_frame_phase_boundaries.py @@ -0,0 +1,92 @@ +from __future__ import annotations + +import ast +from pathlib import Path + +from generate.problem_frame_builder import build_problem_frame + +ROOT = Path(__file__).resolve().parents[1] + + +def _tree(path: str) -> ast.AST: + return ast.parse((ROOT / path).read_text(), filename=path) + + +def _imported_names(tree: ast.AST) -> set[str]: + names: set[str] = set() + for node in ast.walk(tree): + if isinstance(node, ast.ImportFrom): + if node.module is not None: + names.add(node.module) + names.update(alias.name for alias in node.names) + elif isinstance(node, ast.Import): + names.update(alias.name for alias in node.names) + return names + + +def _called_names(tree: ast.AST) -> set[str]: + names: set[str] = set() + for node in ast.walk(tree): + if isinstance(node, ast.Call): + if isinstance(node.func, ast.Name): + names.add(node.func.id) + elif isinstance(node.func, ast.Attribute): + names.add(node.func.attr) + return names + + +def _defined_function_names(tree: ast.AST) -> set[str]: + return {node.name for node in ast.walk(tree) if isinstance(node, ast.FunctionDef)} + + +def test_builder_has_no_assessment_backed_proposal_imports_or_calls() -> None: + tree = _tree("generate/problem_frame_builder.py") + forbidden = {"make_proposal", "assess_contracts", "get_contract_family_id"} + + assert forbidden.isdisjoint(_imported_names(tree)) + assert forbidden.isdisjoint(_called_names(tree)) + + +def test_proposal_phase_does_not_import_contracts_or_builder() -> None: + imports = _imported_names(_tree("generate/problem_frame_proposals.py")) + + assert "generate.problem_frame_contracts" not in imports + assert "problem_frame_contracts" not in imports + assert "ProblemFrameBuilder" not in imports + + +def test_contract_phase_does_not_import_builder() -> None: + imports = _imported_names(_tree("generate/problem_frame_contracts.py")) + + assert "generate.problem_frame_builder" not in imports + assert "problem_frame_builder" not in imports + + +def test_builder_no_longer_defines_phase_helpers() -> None: + defined = _defined_function_names(_tree("generate/problem_frame_builder.py")) + + assert not {name for name in defined if name.startswith("_extract_")} + assert not {name for name in defined if name.endswith("_proposals")} + assert "_quantity_kind_dispositions" not in defined + assert "_bound_relations" not in defined + assert "_bound_question_target" not in defined + + +def test_builder_smoke_shapes_remain_grounded() -> None: + simple = build_problem_frame("Mia has 7 apples. How many apples does Mia have?") + assert tuple(proposal.family_id for proposal in simple.proposals) == ( + "binding.quantity_entity", + ) + assert {mention.surface.lower() for mention in simple.mentions} >= {"7", "apples"} + assert simple.bindings + + gained = build_problem_frame("Tom gained 3 apples") + assert "state_change.unary_delta" in { + proposal.family_id for proposal in gained.proposals + } + assert gained.unary_delta_cues + assert any(relation.relation_type == "unary_delta" for relation in gained.bound_relations) + + measurement = build_problem_frame("The tank has 3 liters. How much liquid is in the tank?") + assert {unit.surface for unit in measurement.units} == {"liters"} + assert any(mention.surface == "3" for mention in measurement.mentions)