feat(comprehension): shared ComprehensionAttempt model + classify (N2)

core/comprehension_attempt/{model,classify}.py: a small frozen ComprehensionAttempt (organ, outcome, case_id, refusal_reason, family, setup_signature, answer, evidence) and classify_r1/classify_r2 that normalize each organ's heterogeneous output (typed setup | Refusal) into a uniform produce-mode setup outcome — setup_refused (with the organ's reason) or setup_correct (with a deterministic inline signature).

No solving, no gold comparison, no evals import (signatures computed inline) — runtime never depends on the harness. Ties to both gold corpora: R2 solved/solver_refuses->setup_correct, reader_refuses->setup_refused with the gold reason; R1 7 admitted / 3 refused. family left None (resolved by the N4 registry). 6 tests.
This commit is contained in:
Shay 2026-06-07 08:40:53 -07:00
parent 0b191055a8
commit 67c54d6813
4 changed files with 226 additions and 0 deletions

View file

@ -0,0 +1,12 @@
"""Shared typed record of a comprehension organ's attempt at a problem (N2).
The normalization layer the contemplation batch (N3 router, N4 registry, N6 pass manager) reasons
over uniform across the R1 and R2 setup compilers. Off-serving; imports no `evals`.
"""
from __future__ import annotations
from core.comprehension_attempt.classify import classify_r1, classify_r2
from core.comprehension_attempt.model import ComprehensionAttempt, Organ, Outcome
__all__ = ["ComprehensionAttempt", "Organ", "Outcome", "classify_r1", "classify_r2"]

View file

@ -0,0 +1,85 @@
"""Normalize R1/R2 organ output into a typed ``ComprehensionAttempt`` (N2, setup-level).
`classify_r1` / `classify_r2` run their organ and report **produce-mode** setup outcomes:
`setup_refused` (with the organ's typed reason) or `setup_correct` (an admissible setup was
produced, with a deterministic signature). They do NOT solve, do NOT compare to gold, and import
nothing from `evals` (signatures are computed inline) keeping this a thin, dependency-light
normalizer. Answer-level outcomes are reached downstream (N6) when the solver/verifier run.
"""
from __future__ import annotations
from typing import Any
from core.comprehension_attempt.model import ComprehensionAttempt
from generate.constraint_comprehension.model import ConstraintProblem
from generate.constraint_comprehension.reader import read_constraint_problem
from generate.meaning_graph.reader import Refusal
from generate.quantitative_comprehension import comprehend_quantitative, to_relational_metric
def _r1_signature(relations: list[dict[str, Any]]) -> str:
"""Deterministic, order-independent string signature of the projected R1 relations."""
items: list[tuple] = []
for r in relations:
kind = r["kind"]
if kind == "fact":
items.append((kind, r["entity"], int(r["value"])))
elif kind in ("more_than", "fewer_than"):
items.append((kind, r["entity"], r["ref"], int(r["delta"])))
elif kind == "times_as_many":
items.append((kind, r["entity"], r["ref"], int(r["factor"])))
elif kind == "divide_by":
items.append((kind, r["entity"], r["ref"], int(r["divisor"])))
elif kind == "sum_of":
items.append((kind, r["entity"], tuple(sorted(r["parts"]))))
else: # pragma: no cover - defensive
items.append(("unhandled", kind, r.get("entity", "")))
return repr(tuple(sorted(items, key=repr)))
def _r2_signature(problem: ConstraintProblem) -> str:
"""Deterministic, order-independent string signature of an R2 ConstraintProblem setup."""
unknowns = tuple(sorted((u.symbol, u.unit, u.domain) for u in problem.unknowns))
constraints: list[tuple] = []
for c in problem.constraints:
merged: dict[str, int] = {}
for symbol, coeff in c.lhs.terms:
merged[symbol] = merged.get(symbol, 0) + coeff
terms = tuple(sorted((s, v) for s, v in merged.items() if v != 0))
constraints.append((terms, c.relation, c.rhs - c.lhs.constant))
query = (problem.query.symbol, problem.query.unit)
return repr((unknowns, tuple(sorted(constraints, key=repr)), query))
def classify_r1(text: str, *, case_id: str | None = None) -> ComprehensionAttempt:
"""Attempt the R1 relational-arithmetic setup compiler on *text*."""
comp = comprehend_quantitative(text)
if isinstance(comp, Refusal):
return ComprehensionAttempt(
"r1_quantitative", "setup_refused", case_id=case_id, refusal_reason=comp.reason
)
projected = to_relational_metric(comp)
if projected is None:
return ComprehensionAttempt(
"r1_quantitative", "setup_refused", case_id=case_id, refusal_reason="unprojectable"
)
relations, _query = projected
return ComprehensionAttempt(
"r1_quantitative", "setup_correct", case_id=case_id, setup_signature=_r1_signature(relations)
)
def classify_r2(text: str, *, case_id: str | None = None) -> ComprehensionAttempt:
"""Attempt the R2 two-category constraint setup compiler on *text*."""
problem = read_constraint_problem(text)
if isinstance(problem, Refusal):
return ComprehensionAttempt(
"r2_constraints", "setup_refused", case_id=case_id, refusal_reason=problem.reason
)
return ComprehensionAttempt(
"r2_constraints", "setup_correct", case_id=case_id, setup_signature=_r2_signature(problem)
)
__all__ = ["classify_r1", "classify_r2"]

View file

@ -0,0 +1,63 @@
"""Typed, immutable record of one comprehension organ's attempt at a problem (N2).
A small normalization layer over the R1 (`generate.quantitative_comprehension`) and R2
(`generate.constraint_comprehension`) setup compilers: it turns each organ's heterogeneous
output (a typed setup, or a typed `Refusal`) into one uniform, frozen `ComprehensionAttempt`.
Nothing here changes reader behavior it only *describes* an outcome so the router (N3),
failure-family registry (N4), and contemplation pass manager (N6) can reason over both organs
uniformly.
Outcome semantics. `classify` (N2) produces **produce-mode** outcomes what the organ did on
its own gates, with no gold in hand: `setup_refused` (the organ refused) or `setup_correct`
(an admissible setup was produced). The gold-relative outcomes (`setup_wrong`, `answer_wrong`)
are representable here but are emitted only in **eval mode** by the lanes that hold gold never
fabricated by `classify`. `answer_*` / `contradiction` are reached when the solver / answer-choice
verifier run downstream (N6), not at setup classification time.
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Literal
from generate.binding_graph.model import SourceSpanLink
Organ = Literal["r1_quantitative", "r2_constraints"]
Outcome = Literal[
"setup_correct", # an admissible setup was produced (produce-mode) / matches gold (eval-mode)
"setup_refused", # the organ refused to assemble a setup
"setup_wrong", # eval-mode only: produced setup diverges from gold (a wrong=0 breach)
"answer_correct", # a value was produced and self-verified / matches gold
"answer_refused", # setup produced but the solver/verifier refused
"answer_wrong", # eval-mode only: produced value diverges from gold
"contradiction", # a verified value contradicts a supplied answer key
]
@dataclass(frozen=True, slots=True)
class ComprehensionAttempt:
"""One organ's attempt at one problem. Immutable; carries the outcome, the refusal reason
(if any), a deterministic setup signature (for cross-organ comparison), the answer (if a
value was produced), and source-span evidence. ``family`` is left ``None`` by ``classify``
and resolved later by the N4 failure-family registry."""
organ: Organ
outcome: Outcome
case_id: str | None = None
refusal_reason: str | None = None
family: str | None = None
setup_signature: str | None = None
answer: int | None = None
evidence: tuple[SourceSpanLink, ...] = ()
@property
def is_setup_correct(self) -> bool:
return self.outcome == "setup_correct"
@property
def is_refusal(self) -> bool:
return self.outcome in ("setup_refused", "answer_refused")
__all__ = ["ComprehensionAttempt", "Organ", "Outcome"]

View file

@ -0,0 +1,66 @@
"""Tests for the shared ComprehensionAttempt normalizer (N2).
Pins that `classify_r1` / `classify_r2` faithfully normalize each organ's output against the
existing gold: well-formed fixtures `setup_correct` with a deterministic signature; refused
fixtures `setup_refused` carrying the organ's typed reason. No gold comparison, no solving.
"""
from __future__ import annotations
import dataclasses
import pytest
from core.comprehension_attempt import classify_r1, classify_r2
from evals.constraint_oracle.runner import _load_r2_gold
from evals.setup_oracle.runner import _load_r1_gold
def test_classify_r2_matches_gold_expect() -> None:
for fx in _load_r2_gold():
att = classify_r2(fx["text"], case_id=fx["id"])
assert att.organ == "r2_constraints"
if fx["expect"] in ("solved", "solver_refuses"):
assert att.outcome == "setup_correct", f"{fx['id']}: {att.refusal_reason}"
assert att.setup_signature is not None
else: # reader_refuses
assert att.outcome == "setup_refused"
assert att.refusal_reason == fx["reader_reason"], fx["id"]
def test_classify_r1_admits_seven_refuses_three() -> None:
attempts = [classify_r1(fx["text"], case_id=fx["id"]) for fx in _load_r1_gold()]
correct = [a for a in attempts if a.outcome == "setup_correct"]
refused = [a for a in attempts if a.outcome == "setup_refused"]
assert len(correct) == 7 and len(refused) == 3
assert all(a.organ == "r1_quantitative" for a in attempts)
assert all(a.setup_signature is not None for a in correct)
assert all(a.refusal_reason for a in refused)
def test_classify_r1_specific_inverse_and_pronoun() -> None:
inverse = classify_r1("Nia has 9 more beads than Omar. Nia has 15 beads. How many beads does Omar have?")
assert inverse.outcome == "setup_correct"
pronoun = classify_r1("Pat has 5 marbles. He has 3 more than her. How many marbles does she have?")
assert pronoun.outcome == "setup_refused" and pronoun.refusal_reason is not None
def test_signatures_are_deterministic() -> None:
text = next(f for f in _load_r2_gold() if f["expect"] == "solved")["text"]
assert classify_r2(text).setup_signature == classify_r2(text).setup_signature
def test_attempt_is_frozen() -> None:
att = classify_r2("A school rents 6 buses. How many large buses are there?")
with pytest.raises(dataclasses.FrozenInstanceError):
att.outcome = "setup_correct" # type: ignore[misc]
def test_classify_does_not_import_evals() -> None:
# core/comprehension_attempt must not depend on evals (runtime must not import the harness).
import inspect
import core.comprehension_attempt.classify as m
source = inspect.getsource(m)
assert "import evals" not in source and "from evals" not in source