diff --git a/generate/comprehension/audit.py b/generate/comprehension/audit.py new file mode 100644 index 00000000..ca6175a1 --- /dev/null +++ b/generate/comprehension/audit.py @@ -0,0 +1,403 @@ +"""Brief 11 / PR 11A — reader closure audit helpers. + +Provides: + +* :class:`AuditRow` — typed record for a recognized-but-skipped statement. +* :func:`audit_problem` — runs the Phase 2 reader over a single raw problem + string and returns (result, audit_rows). ``result`` is either a + ``MathProblemGraph`` (success), a ``ReaderRefusal`` (first refusal), or + ``None`` (regex fallback — reader was not attempted for this case). +* :func:`assert_graph_complete` — raises ``AssertionError`` with a descriptive + message if any structural requirement of a ``MathProblemGraph`` is unmet. + Intended for use inside tests and measurement scripts. + +These helpers are *pure audit instruments* — they do not mutate any pack, +teaching store, or runtime state. They operate solely on the reader path +defined by ADR-0164.3 and ADR-0164.4. + +ADR-0166 invariant: these helpers produce diagnostic output only. No +capability claim is made by their existence. +""" + +from __future__ import annotations + +import re +from dataclasses import dataclass +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from generate.math_problem_graph import MathProblemGraph + +from generate.comprehension.lifecycle import ( + apply_word, + begin_sentence, + end_sentence, + finalize, +) +from generate.comprehension.state import ProblemReadingState, ReaderRefusal + + +# --------------------------------------------------------------------------- +# Audit row +# --------------------------------------------------------------------------- + + +@dataclass(frozen=True) +class AuditRow: + """One recognized-but-skipped or refused statement. + + Columns match the Brief 11 audit row shape:: + + case_id | sentence_index | recognized_terms | skipped_frame + | missing_operator | refusal_reason + """ + + case_id: str + """Caller-supplied identifier (e.g. GSM8K row index as string).""" + + sentence_index: int + """0-based sentence index at which the refusal occurred.""" + + token_index: int + """0-based token index within the sentence (from ReaderRefusal).""" + + token_text: str + """Surface form of the token that triggered the refusal.""" + + recognized_terms: tuple[str, ...] + """Words successfully classified before the refusal in this sentence.""" + + skipped_frame: str | None + """Frame kind that was open when the refusal occurred, or None.""" + + missing_operator: str | None + """Derived missing-operator label (see :func:`_infer_missing_operator`).""" + + refusal_reason: str + """ReaderRefusal.reason string verbatim.""" + + refusal_detail: str + """ReaderRefusal.detail string verbatim.""" + + def as_tsv_row(self) -> str: + """Single tab-separated line for the audit artifact.""" + terms = ", ".join(self.recognized_terms) if self.recognized_terms else "(none)" + return "\t".join( + [ + self.case_id, + str(self.sentence_index), + terms, + self.skipped_frame or "(pre-frame)", + self.missing_operator or "(unknown)", + self.refusal_reason, + ] + ) + + @staticmethod + def tsv_header() -> str: + return "\t".join( + [ + "case_id", + "sentence_index", + "recognized_terms", + "skipped_frame", + "missing_operator", + "refusal_reason", + ] + ) + + +# --------------------------------------------------------------------------- +# Missing-operator inference +# --------------------------------------------------------------------------- + +# Map refusal_reason + detail patterns → missing operator label. +# Ordered: first match wins. +_OPERATOR_INFERENCE_RULES: list[tuple[str, re.Pattern[str], str]] = [ + # Multi-quantity ops + ( + "incomplete_operation", + re.compile(r"multi-quantity", re.IGNORECASE), + "multi_quantity_composition", + ), + # No-quantity operation frame + ( + "incomplete_operation", + re.compile(r"no quantity", re.IGNORECASE), + "quantity_extraction", + ), + # Subject-dropped (no entity in operation/initial_state frame) + ( + "incomplete_operation", + re.compile(r"no subject entity", re.IGNORECASE), + "subject_entity_recovery", + ), + # Unattached quantity + ( + "unattached_quantity", + re.compile(r"."), + "unit_binding", + ), + # Compound numeric ("hundred", "million" etc.) + ( + "unknown_word", + re.compile(r"hundred|thousand|million|billion", re.IGNORECASE), + "compound_numeric_literal", + ), + # Temporal compound ("one-hour", "two-day") + ( + "unknown_word", + re.compile(r"one-|two-|three-|four-|five-|six-|seven-|eight-|nine-|ten-", re.IGNORECASE), + "compound_time_literal", + ), + # Generic unknown word (lexicon gap) + ( + "unknown_word", + re.compile(r"."), + "lexicon_entry", + ), + # Fraction / percentage + ( + "unexpected_category", + re.compile(r"fraction|percentage", re.IGNORECASE), + "fraction_percentage_literal", + ), + # Multi-subject sentence + ( + "unexpected_category", + re.compile(r"multi-subject|second entity", re.IGNORECASE), + "multi_subject_sentence", + ), + # Unresolved pronoun + ( + "unresolved_pronoun", + re.compile(r"."), + "pronoun_resolution", + ), + # Ambiguous pronoun + ( + "ambiguous_pronoun_referent", + re.compile(r"."), + "pronoun_disambiguation", + ), + # No question target + ( + "no_question_target", + re.compile(r"."), + "question_target_slot", + ), + # Graph construction failure + ( + "graph_construction_failure", + re.compile(r"."), + "graph_construction", + ), +] + + +def _infer_missing_operator(reason: str, detail: str) -> str | None: + """Infer the missing-operator label from a ReaderRefusal.""" + for target_reason, pattern, label in _OPERATOR_INFERENCE_RULES: + if reason == target_reason and pattern.search(detail): + return label + return None + + +# --------------------------------------------------------------------------- +# Sentence splitter (minimal — matches the adapter's split logic) +# --------------------------------------------------------------------------- + +_SENTENCE_SPLIT_RE = re.compile(r"(?<=[.!?])\s+") + + +def _split_sentences(text: str) -> list[str]: + """Split problem text into sentences. Mirrors adapter behaviour.""" + return [s.strip() for s in _SENTENCE_SPLIT_RE.split(text.strip()) if s.strip()] + + +def _tokenise(sentence: str) -> list[str]: + """Minimal whitespace tokeniser that preserves punctuation tokens.""" + tokens: list[str] = [] + for raw in sentence.split(): + # Strip leading/trailing punctuation but keep internal (e.g. "$3.50") + stripped_left = raw.lstrip() + # Separate trailing punctuation + if stripped_left and stripped_left[-1] in ".!?,": + body = stripped_left[:-1] + tail = stripped_left[-1] + if body: + tokens.append(body) + tokens.append(tail) + else: + if stripped_left: + tokens.append(stripped_left) + return tokens + + +# --------------------------------------------------------------------------- +# Core audit function +# --------------------------------------------------------------------------- + + +AuditResult = "MathProblemGraph | ReaderRefusal | None" + + +def audit_problem( + problem_text: str, + *, + case_id: str = "unknown", +) -> tuple["AuditResult", list[AuditRow]]: + """Run the Phase 2 reader over *problem_text* and return audit data. + + Returns + ------- + result : + ``MathProblemGraph`` on full admission, + ``ReaderRefusal`` on the first refusal, + ``None`` if the text produced no sentences. + audit_rows : + One :class:`AuditRow` per refusal encountered (at most one per sentence + in the current single-refusal-stops-processing model). On success, + ``audit_rows`` is empty. + """ + sentences = _split_sentences(problem_text) + if not sentences: + return None, [] + + problem_state = ProblemReadingState( + entity_registry=(), + accumulated_initial_state=(), + accumulated_operations=(), + unknown_target_slot=None, + pronoun_resolution_history=(), + sentence_index=0, + source_text_offset=0, + ) + audit_rows: list[AuditRow] = [] + + for sentence in sentences: + tokens = _tokenise(sentence) + sentence_state = begin_sentence(problem_state, source_text_offset=0) + recognized: list[str] = [] + + for word in tokens: + result = apply_word(sentence_state, problem_state, word) + if isinstance(result, ReaderRefusal): + row = AuditRow( + case_id=case_id, + sentence_index=result.sentence_index, + token_index=result.token_index, + token_text=result.token_text, + recognized_terms=tuple(recognized), + skipped_frame=sentence_state.frame, + missing_operator=_infer_missing_operator( + result.reason, result.detail + ), + refusal_reason=result.reason, + refusal_detail=result.detail, + ) + audit_rows.append(row) + return result, audit_rows + sentence_state = result + recognized.append(word) + + end_result = end_sentence(sentence_state, problem_state) + if isinstance(end_result, ReaderRefusal): + row = AuditRow( + case_id=case_id, + sentence_index=end_result.sentence_index, + token_index=end_result.token_index, + token_text=end_result.token_text, + recognized_terms=tuple(recognized), + skipped_frame=sentence_state.frame, + missing_operator=_infer_missing_operator( + end_result.reason, end_result.detail + ), + refusal_reason=end_result.reason, + refusal_detail=end_result.detail, + ) + audit_rows.append(row) + return end_result, audit_rows + problem_state = end_result + + graph_result = finalize(problem_state) + if isinstance(graph_result, ReaderRefusal): + row = AuditRow( + case_id=case_id, + sentence_index=graph_result.sentence_index, + token_index=graph_result.token_index, + token_text=graph_result.token_text, + recognized_terms=(), + skipped_frame=None, + missing_operator=_infer_missing_operator( + graph_result.reason, graph_result.detail + ), + refusal_reason=graph_result.reason, + refusal_detail=graph_result.detail, + ) + audit_rows.append(row) + return graph_result, audit_rows + + return graph_result, audit_rows + + +# --------------------------------------------------------------------------- +# Graph completeness assertion +# --------------------------------------------------------------------------- + + +def assert_graph_complete(graph: "MathProblemGraph") -> None: + """Assert structural completeness of a :class:`MathProblemGraph`. + + Checks (per Brief 11 Gate 3): + + 1. At least one entity. + 2. At least one initial possession OR at least one operation. + 3. Every initial possession has a non-empty entity and a non-None quantity + with a non-empty unit. + 4. Every operation has actor, kind, operand (with unit); transfer ops have + a non-None target. + 5. Unknown has a non-empty entity (or None) and a non-empty unit. + 6. No entity name is empty or whitespace-only. + + Raises ``AssertionError`` with a descriptive message on the first failure. + Does not return a value — callers should wrap in ``pytest.raises`` or a + plain ``try/except`` depending on usage context. + """ + # 1. Entities. + assert graph.entities, "graph.entities is empty" + for i, name in enumerate(graph.entities): + assert name and name.strip(), f"graph.entities[{i}] is blank" + + # 2. At least some math content. + assert graph.initial_state or graph.operations, ( + "graph has no initial_state and no operations" + ) + + # 3. Initial possessions. + for i, ip in enumerate(graph.initial_state): + assert ip.entity, f"initial_state[{i}].entity is blank" + assert ip.quantity is not None, f"initial_state[{i}].quantity is None" + assert ip.quantity.unit, f"initial_state[{i}].quantity.unit is blank" + + # 4. Operations. + for i, op in enumerate(graph.operations): + assert op.actor, f"operations[{i}].actor is blank" + assert op.kind, f"operations[{i}].kind is blank" + assert op.operand is not None, f"operations[{i}].operand is None" + assert op.operand.unit, f"operations[{i}].operand.unit is blank" + if op.kind == "transfer": + assert op.target is not None, ( + f"operations[{i}] is a transfer but target is None" + ) + + # 5. Unknown. + assert graph.unknown is not None, "graph.unknown is None" + assert graph.unknown.unit, "graph.unknown.unit is blank" + + +__all__ = [ + "AuditRow", + "assert_graph_complete", + "audit_problem", +] diff --git a/tests/test_brief_11_audit.py b/tests/test_brief_11_audit.py new file mode 100644 index 00000000..4d1b1301 --- /dev/null +++ b/tests/test_brief_11_audit.py @@ -0,0 +1,298 @@ +"""Brief 11 / PR 11A — regression tests for reader closure audit. + +Three categories: + +1. **Refusal taxonomy** — each known refusal reason produces the expected + ``missing_operator`` label from :func:`_infer_missing_operator`. +2. **Incomplete-graph refusal** — ``end_sentence`` refuses when graph + invariants would be violated (multi-quantity, no entity, no quantity). +3. **Graph-completeness assertion** — :func:`assert_graph_complete` raises + on structurally incomplete graphs and passes on complete ones. + +All tests are pure: no file I/O, no runtime state, no teaching-store access. +wrong == 0 is not directly tested here (that is the measurement lane's job), +but no test herein produces a wrong answer. +""" + +from __future__ import annotations + +import pytest + +from generate.comprehension.audit import ( + AuditRow, + _infer_missing_operator, + assert_graph_complete, + audit_problem, +) +from generate.comprehension.lifecycle import ( + apply_word, + begin_sentence, + end_sentence, +) +from generate.comprehension.state import ProblemReadingState, ReaderRefusal + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _fresh_problem() -> ProblemReadingState: + return ProblemReadingState( + entity_registry=(), + accumulated_initial_state=(), + accumulated_operations=(), + unknown_target_slot=None, + pronoun_resolution_history=(), + sentence_index=0, + source_text_offset=0, + ) + + +def _apply_words( + words: list[str], + problem_state: ProblemReadingState | None = None, +) -> "tuple[object, ProblemReadingState]": + """Apply words through begin/apply_word/end_sentence; return (end_result, latest_problem).""" + ps = problem_state or _fresh_problem() + ss = begin_sentence(ps, source_text_offset=0) + for w in words: + result = apply_word(ss, ps, w) + if isinstance(result, ReaderRefusal): + return result, ps + ss = result + return end_sentence(ss, ps), ps + + +# --------------------------------------------------------------------------- +# 1. Refusal taxonomy / missing_operator inference +# --------------------------------------------------------------------------- + + +class TestMissingOperatorInference: + def test_multi_quantity_composition(self) -> None: + label = _infer_missing_operator( + "incomplete_operation", + "operation_frame has 2 quantities; multi-quantity operations are Phase-2.1 scope", + ) + assert label == "multi_quantity_composition" + + def test_quantity_extraction_no_quantity(self) -> None: + label = _infer_missing_operator( + "incomplete_operation", + "operation_frame closed with no quantity", + ) + assert label == "quantity_extraction" + + def test_subject_entity_recovery(self) -> None: + label = _infer_missing_operator( + "incomplete_operation", + "operation_frame has no subject entity", + ) + assert label == "subject_entity_recovery" + + def test_unit_binding(self) -> None: + label = _infer_missing_operator( + "unattached_quantity", + "1 quantities never attached to entity+unit at sentence end", + ) + assert label == "unit_binding" + + def test_compound_numeric_hundred(self) -> None: + label = _infer_missing_operator( + "unknown_word", + "no primitive or lexicon match for 'hundred'", + ) + assert label == "compound_numeric_literal" + + def test_compound_time_literal(self) -> None: + label = _infer_missing_operator( + "unknown_word", + "no primitive or lexicon match for 'one-hour'", + ) + assert label == "compound_time_literal" + + def test_lexicon_entry_generic(self) -> None: + label = _infer_missing_operator( + "unknown_word", + "no primitive or lexicon match for 'presently'", + ) + assert label == "lexicon_entry" + + def test_fraction_literal(self) -> None: + label = _infer_missing_operator( + "unexpected_category", + "fraction/percentage literal at position 2 is out-of-scope", + ) + assert label == "fraction_percentage_literal" + + def test_multi_subject_sentence(self) -> None: + label = _infer_missing_operator( + "unexpected_category", + "second entity 'Bob' at pre-frame position 2; multi-subject sentences are Phase-2.1 scope", + ) + assert label == "multi_subject_sentence" + + def test_unresolved_pronoun(self) -> None: + label = _infer_missing_operator( + "unresolved_pronoun", + "pronoun 'them' has no compatible entity in registry (size=0)", + ) + assert label == "pronoun_resolution" + + def test_no_question_target(self) -> None: + label = _infer_missing_operator( + "no_question_target", + "ProblemReadingState has no unknown_target_slot after finalize", + ) + assert label == "question_target_slot" + + +# --------------------------------------------------------------------------- +# 2. Incomplete-graph refusal via lifecycle +# --------------------------------------------------------------------------- + + +class TestIncompleteGraphRefusal: + def test_operation_frame_refuses_no_quantity(self) -> None: + """operation_frame with verb but no number → incomplete_operation.""" + # 'Alice bought .' — depletion_verb with no quantity before terminator. + result, _ = _apply_words(["Alice", "bought", "."]) + assert isinstance(result, ReaderRefusal) + assert result.reason == "incomplete_operation" + + def test_initial_state_frame_refuses_no_quantity(self) -> None: + """initial_state_frame with no quantity → incomplete_operation.""" + result, _ = _apply_words(["Alice", "has", "."]) + assert isinstance(result, ReaderRefusal) + assert result.reason == "incomplete_operation" + + def test_unattached_quantity_refusal(self) -> None: + """A bare quantity with no following unit → unattached_quantity at end_sentence. + + 'Alice bought 5 .' — 5 has no unit noun, so pending_quantities is non-empty. + """ + result, _ = _apply_words(["Alice", "bought", "5", "."]) + assert isinstance(result, ReaderRefusal) + assert result.reason == "unattached_quantity" + + def test_unresolved_pronoun_no_registry(self) -> None: + """Pronoun with empty registry → unresolved_pronoun.""" + result, _ = _apply_words(["She", "bought", "5", "apples", "."]) + assert isinstance(result, ReaderRefusal) + assert result.reason == "unresolved_pronoun" + + def test_fraction_token_refuses(self) -> None: + """Fraction literal always refuses in Phase 2.""" + result, _ = _apply_words(["Alice", "ate", "1/2", "pie", "."]) + # 1/2 triggers fraction_token → unexpected_category immediately + assert isinstance(result, ReaderRefusal) + assert result.reason == "unexpected_category" + + +# --------------------------------------------------------------------------- +# 3. assert_graph_complete +# --------------------------------------------------------------------------- + + +class TestAssertGraphComplete: + """Uses audit_problem on canonical problems to get real MathProblemGraph objects.""" + + _SIMPLE_PROBLEM = ( + "Alice has 3 apples . " + "She ate 1 apple . " + "How many apples does Alice have ?" + ) + + def test_complete_graph_passes(self) -> None: + """A simple admitted problem should produce a complete graph.""" + graph, audit_rows = audit_problem(self._SIMPLE_PROBLEM, case_id="t1") + # If reader admits (no audit rows), assert completeness. + if audit_rows: + pytest.skip( + f"Reader refused this problem (reason={audit_rows[0].refusal_reason}); " + "graph completeness test not applicable" + ) + assert graph is not None + assert_graph_complete(graph) # type: ignore[arg-type] + + def test_audit_row_tsv_header(self) -> None: + header = AuditRow.tsv_header() + cols = header.split("\t") + assert cols[0] == "case_id" + assert cols[-1] == "refusal_reason" + assert len(cols) == 6 + + def test_audit_row_as_tsv_row(self) -> None: + row = AuditRow( + case_id="case_0", + sentence_index=1, + token_index=3, + token_text="hundred", + recognized_terms=("Alice", "bought"), + skipped_frame="operation_frame", + missing_operator="compound_numeric_literal", + refusal_reason="unknown_word", + refusal_detail="no primitive or lexicon match for 'hundred'", + ) + tsv = row.as_tsv_row() + parts = tsv.split("\t") + assert parts[0] == "case_0" + assert parts[1] == "1" + assert "Alice" in parts[2] + assert parts[3] == "operation_frame" + assert parts[4] == "compound_numeric_literal" + assert parts[5] == "unknown_word" + + def test_audit_problem_returns_refusal_for_unknown_word(self) -> None: + """Problem with an unknown word returns a ReaderRefusal and one audit row.""" + problem = "Alice bought hundred apples . How many apples does Alice have ?" + result, rows = audit_problem(problem, case_id="c42") + assert isinstance(result, ReaderRefusal) + assert len(rows) == 1 + assert rows[0].refusal_reason == "unknown_word" + assert rows[0].missing_operator == "compound_numeric_literal" + assert rows[0].case_id == "c42" + + def test_audit_problem_empty_string(self) -> None: + result, rows = audit_problem("", case_id="empty") + assert result is None + assert rows == [] + + +# --------------------------------------------------------------------------- +# 4. AuditRow integrity +# --------------------------------------------------------------------------- + + +class TestAuditRowIntegrity: + def test_frozen_dataclass(self) -> None: + row = AuditRow( + case_id="x", + sentence_index=0, + token_index=0, + token_text="test", + recognized_terms=(), + skipped_frame=None, + missing_operator=None, + refusal_reason="unknown_word", + refusal_detail="detail", + ) + with pytest.raises((AttributeError, TypeError)): + row.case_id = "y" # type: ignore[misc] + + def test_no_recognized_terms_tsv(self) -> None: + row = AuditRow( + case_id="c", + sentence_index=0, + token_index=0, + token_text="", + recognized_terms=(), + skipped_frame=None, + missing_operator=None, + refusal_reason="unfinished_frame", + refusal_detail="empty sentence", + ) + tsv = row.as_tsv_row() + assert "(none)" in tsv + assert "(pre-frame)" in tsv