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

feat(brief-11/11A): reader closure audit — per-case refusal taxonomy, graph-completeness helpers, regression tests (#343)

Browse source
This commit is contained in:
Shay 2026-05-27 05:14:42 -07:00 committed by GitHub
parent 60043973b0
commit aa53fcf78d
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 701 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

403
generate/comprehension/audit.py Normal file
View file

@ -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",
]

298
tests/test_brief_11_audit.py Normal file
View file

@ -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