feat(ADR-0167/W1-A): MathReaderRefusalEvidence schema + canonical-bytes (#350)

Foundation type for routing comprehension-reader refusals into the
teaching corridor.  Frozen dataclass with sha256 evidence_hash computed
from deterministic canonical bytes (mirrors state.to_canonical_bytes
pattern).  Includes SUB_TYPE_FOR_OPERATOR mapping table covering all 13
missing_operator values in the current audit artifact.

Wave 1 only — no runtime mutation, no teaching-store integration, no
admission path.  Downstream W2-A/B/C/D type-import from this module.
This commit is contained in:
Shay 2026-05-27 06:30:21 -07:00 committed by GitHub
parent 4f0815ef9a
commit 99c11d160a
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 365 additions and 0 deletions

168
teaching/math_evidence.py Normal file
View file

@ -0,0 +1,168 @@
"""ADR-0167 W1-A — MathReaderRefusalEvidence schema + canonical-bytes.
Typed record wrapping an :class:`AuditRow` from the comprehension reader's
refusal taxonomy. Every downstream Wave 2 brief (W2-A through W2-D)
type-imports :class:`MathReaderRefusalEvidence` and
:data:`SUB_TYPE_FOR_OPERATOR` from this module.
This module is a standalone schema it does not mutate any teaching store,
pack, or runtime state.
"""
from __future__ import annotations
import hashlib
import json
from dataclasses import dataclass
from typing import Any, Final, Literal
from generate.comprehension.audit import AuditRow
# ---------------------------------------------------------------------------
# Sub-type literal
# ---------------------------------------------------------------------------
SubType = Literal["lexical", "frame", "composition", "reference", "slot"]
SUB_TYPE_FOR_OPERATOR: Final[dict[str, SubType]] = {
"lexicon_entry": "lexical",
"compound_numeric_literal": "lexical",
"compound_time_literal": "lexical",
"pre_frame_filler_sentence": "frame",
"multi_quantity_composition": "composition",
"quantity_extraction": "composition",
"pronoun_resolution": "reference",
"question_frame_slot": "slot",
"unit_binding": "slot",
"fraction_percentage_literal": "lexical", # provisional
"multi_subject_sentence": "frame",
"question_target_slot": "slot",
"descriptive_frame_question": "slot",
}
# ---------------------------------------------------------------------------
# Canonical-bytes helpers (mirrors generate/comprehension/state.py pattern)
# ---------------------------------------------------------------------------
class _OmitSentinel:
__slots__ = ()
_OMIT = _OmitSentinel()
def _canonical_value(obj: Any) -> Any:
"""Recursively convert to a canonical JSON-serialisable value.
None sentinel _OMIT (caller drops from dict).
Tuples/lists JSON arrays. Dataclasses sorted-key dicts.
"""
if obj is None:
return _OMIT
if isinstance(obj, bool):
return obj
if isinstance(obj, int):
return obj
if isinstance(obj, str):
return obj
if isinstance(obj, (tuple, list)):
return [_canonical_value(item) for item in obj]
if hasattr(obj, "__dataclass_fields__"):
out: dict[str, Any] = {}
for key in sorted(obj.__dataclass_fields__.keys()):
val = _canonical_value(getattr(obj, key))
if val is not _OMIT:
out[key] = val
return out
raise TypeError(
f"_canonical_value: cannot serialise {type(obj).__name__}"
)
def _to_canonical_bytes_for_evidence(
evidence: MathReaderRefusalEvidence,
) -> bytes:
"""Produce deterministic canonical bytes, excluding evidence_hash itself."""
out: dict[str, Any] = {}
for key in sorted(evidence.__dataclass_fields__.keys()):
if key == "evidence_hash":
continue
val = _canonical_value(getattr(evidence, key))
if val is not _OMIT:
out[key] = val
return json.dumps(
out,
ensure_ascii=False,
sort_keys=True,
separators=(",", ":"),
).encode("utf-8")
# ---------------------------------------------------------------------------
# Evidence record
# ---------------------------------------------------------------------------
@dataclass(frozen=True, slots=True)
class MathReaderRefusalEvidence:
"""Typed record binding a comprehension-reader refusal to the teaching corridor.
Construct via :func:`from_audit_row` do not instantiate directly.
"""
case_id: str
sentence_index: int
token_index: int
refusal_reason: str
missing_operator: str | None
claim_signature: str
evidence_hash: str
audit_row: AuditRow
sub_type: SubType
def to_canonical_bytes(self) -> bytes:
"""Deterministic canonical bytes (evidence_hash excluded)."""
return _to_canonical_bytes_for_evidence(self)
# ---------------------------------------------------------------------------
# Factory
# ---------------------------------------------------------------------------
def from_audit_row(
audit_row: AuditRow,
sub_type: SubType,
*,
claim_signature: str = "",
) -> MathReaderRefusalEvidence:
"""Build a :class:`MathReaderRefusalEvidence` from an audit row.
The ``evidence_hash`` is computed from the canonical bytes of all other
fields. ``claim_signature`` defaults to ``""`` in W1-A; W2-B fills it.
"""
placeholder_hash = ""
record = MathReaderRefusalEvidence(
case_id=audit_row.case_id,
sentence_index=audit_row.sentence_index,
token_index=audit_row.token_index,
refusal_reason=audit_row.refusal_reason,
missing_operator=audit_row.missing_operator,
claim_signature=claim_signature,
evidence_hash=placeholder_hash,
audit_row=audit_row,
sub_type=sub_type,
)
canonical = record.to_canonical_bytes()
real_hash = hashlib.sha256(canonical).hexdigest()
object.__setattr__(record, "evidence_hash", real_hash)
return record
__all__ = [
"MathReaderRefusalEvidence",
"SUB_TYPE_FOR_OPERATOR",
"SubType",
"from_audit_row",
]

View file

@ -0,0 +1,197 @@
"""Tests for ADR-0167 W1-A — MathReaderRefusalEvidence schema + canonical-bytes."""
from __future__ import annotations
import dataclasses
import hashlib
import json
from pathlib import Path
import pytest
from generate.comprehension.audit import AuditRow, audit_problem
from teaching.math_evidence import (
MathReaderRefusalEvidence,
SUB_TYPE_FOR_OPERATOR,
from_audit_row,
)
# ---------------------------------------------------------------------------
# Fixture: a concrete AuditRow for reuse
# ---------------------------------------------------------------------------
_SAMPLE_ROW = AuditRow(
case_id="test-001",
sentence_index=0,
token_index=3,
token_text="hefty",
recognized_terms=("James", "has", "5"),
skipped_frame="initial_state",
missing_operator="lexicon_entry",
refusal_reason="unknown_word",
refusal_detail="unknown word: hefty",
)
# ---------------------------------------------------------------------------
# test_evidence_is_frozen
# ---------------------------------------------------------------------------
def test_evidence_is_frozen() -> None:
ev = from_audit_row(_SAMPLE_ROW, "lexical")
with pytest.raises(dataclasses.FrozenInstanceError):
ev.case_id = "mutated" # type: ignore[misc]
with pytest.raises(dataclasses.FrozenInstanceError):
ev.evidence_hash = "mutated" # type: ignore[misc]
with pytest.raises(dataclasses.FrozenInstanceError):
ev.sub_type = "frame" # type: ignore[misc]
# ---------------------------------------------------------------------------
# test_canonical_bytes_deterministic
# ---------------------------------------------------------------------------
def test_canonical_bytes_deterministic() -> None:
ev1 = from_audit_row(_SAMPLE_ROW, "lexical")
ev2 = from_audit_row(_SAMPLE_ROW, "lexical")
assert ev1.to_canonical_bytes() == ev2.to_canonical_bytes()
# ---------------------------------------------------------------------------
# test_evidence_hash_matches_canonical_bytes
# ---------------------------------------------------------------------------
def test_evidence_hash_matches_canonical_bytes() -> None:
ev = from_audit_row(_SAMPLE_ROW, "lexical")
expected = hashlib.sha256(ev.to_canonical_bytes()).hexdigest()
assert ev.evidence_hash == expected
# ---------------------------------------------------------------------------
# test_evidence_hash_excludes_itself
# ---------------------------------------------------------------------------
def test_evidence_hash_excludes_itself() -> None:
ev = from_audit_row(_SAMPLE_ROW, "lexical")
canonical_before = ev.to_canonical_bytes()
object.__setattr__(ev, "evidence_hash", "tampered")
canonical_after = ev.to_canonical_bytes()
assert canonical_before == canonical_after
# ---------------------------------------------------------------------------
# test_sub_type_table_covers_all_missing_operators
# ---------------------------------------------------------------------------
def test_sub_type_table_covers_all_missing_operators() -> None:
artifact_path = Path(
"evals/gsm8k_math/train_sample/v1/audit_brief_11.json"
)
data = json.loads(artifact_path.read_text())
operators_in_artifact: set[str] = set()
for case in data["per_case"]:
op = case.get("missing_operator")
if op is not None:
operators_in_artifact.add(op)
assert operators_in_artifact, "artifact has no missing_operator values"
missing = operators_in_artifact - set(SUB_TYPE_FOR_OPERATOR)
assert not missing, f"operators not in SUB_TYPE_FOR_OPERATOR: {sorted(missing)}"
# ---------------------------------------------------------------------------
# test_sub_type_table_covers_live_audit (against cases.jsonl)
# ---------------------------------------------------------------------------
def test_sub_type_table_covers_live_audit() -> None:
cases_path = Path("evals/gsm8k_math/train_sample/v1/cases.jsonl")
operators_found: set[str] = set()
for line in cases_path.read_text().splitlines():
c = json.loads(line)
_result, rows = audit_problem(c["question"], case_id=c["case_id"])
for row in rows:
if row.missing_operator is not None:
operators_found.add(row.missing_operator)
assert row.missing_operator in SUB_TYPE_FOR_OPERATOR, (
f"live audit operator {row.missing_operator!r} "
f"not in SUB_TYPE_FOR_OPERATOR"
)
assert operators_found, "no refusals found in live audit — data may have changed"
# ---------------------------------------------------------------------------
# test_distinct_sub_types_have_distinct_hashes
# ---------------------------------------------------------------------------
def test_distinct_sub_types_have_distinct_hashes() -> None:
ev_lexical = from_audit_row(_SAMPLE_ROW, "lexical")
ev_frame = from_audit_row(_SAMPLE_ROW, "frame")
assert ev_lexical.evidence_hash != ev_frame.evidence_hash
assert ev_lexical.to_canonical_bytes() != ev_frame.to_canonical_bytes()
# ---------------------------------------------------------------------------
# test_from_audit_row_factory
# ---------------------------------------------------------------------------
def test_from_audit_row_factory() -> None:
ev = from_audit_row(_SAMPLE_ROW, "lexical")
assert ev.case_id == _SAMPLE_ROW.case_id
assert ev.sentence_index == _SAMPLE_ROW.sentence_index
assert ev.token_index == _SAMPLE_ROW.token_index
assert ev.refusal_reason == _SAMPLE_ROW.refusal_reason
assert ev.missing_operator == _SAMPLE_ROW.missing_operator
assert ev.audit_row is _SAMPLE_ROW
assert ev.sub_type == "lexical"
assert isinstance(ev.evidence_hash, str)
assert len(ev.evidence_hash) == 64 # sha256 hex
# ---------------------------------------------------------------------------
# test_claim_signature_default_empty_in_w1a
# ---------------------------------------------------------------------------
def test_claim_signature_default_empty_in_w1a() -> None:
ev = from_audit_row(_SAMPLE_ROW, "lexical")
assert ev.claim_signature == ""
# ---------------------------------------------------------------------------
# test_determinism_across_n_builds
# ---------------------------------------------------------------------------
def test_determinism_across_n_builds() -> None:
hashes = {from_audit_row(_SAMPLE_ROW, "lexical").evidence_hash for _ in range(20)}
assert len(hashes) == 1, f"non-deterministic hashes: {hashes}"
# ---------------------------------------------------------------------------
# test_from_audit_row_with_none_missing_operator
# ---------------------------------------------------------------------------
def test_from_audit_row_with_none_missing_operator() -> None:
row = AuditRow(
case_id="test-nil",
sentence_index=1,
token_index=0,
token_text="?",
recognized_terms=(),
skipped_frame=None,
missing_operator=None,
refusal_reason="no_question_target",
refusal_detail="no target found",
)
ev = from_audit_row(row, "slot")
assert ev.missing_operator is None
canonical = json.loads(ev.to_canonical_bytes())
assert "missing_operator" not in canonical