diff --git a/teaching/math_evidence.py b/teaching/math_evidence.py new file mode 100644 index 00000000..a4290483 --- /dev/null +++ b/teaching/math_evidence.py @@ -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", +] diff --git a/tests/test_math_evidence_schema.py b/tests/test_math_evidence_schema.py new file mode 100644 index 00000000..884a050e --- /dev/null +++ b/tests/test_math_evidence_schema.py @@ -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