diff --git a/teaching/math_claim_signature.py b/teaching/math_claim_signature.py new file mode 100644 index 00000000..b58a83a1 --- /dev/null +++ b/teaching/math_claim_signature.py @@ -0,0 +1,77 @@ +"""ADR-0167 W2-B — Deterministic claim-signature normalisation for LexicalClaim. + +Produces a stable sha256 hex that collapses two GSM8K cases refusing on the +same lexical token into one teaching-corpus candidate. Only ``sub_type == +"lexical"`` evidence gets a non-empty signature in this PR; all other sub_types +are deferred to a follow-up ADR. + +Normalisation pipeline (applied in order, documented here as a breaking-change +surface — modifying these rules changes all existing signatures): + + 1. Lowercase the ``surface`` string. + 2. Strip leading and trailing characters that are members of the fixed + punctuation set ``string.punctuation``: + ``!"#$%&'()*+,-./:;<=>?@[\\]^_`{|}~`` + This set is frozen; expansions are breaking changes. + 3. Extract the unknown-token from ``refusal_detail`` via the canonical + regex r"no primitive or lexicon match for '([^']+)'" + 4. If the regex does not match (e.g. fraction/percentage detail format), + use the stripped-lowercase ``surface`` from step 2 as the token. + 5. Canonical string: ``"lexical:" + extracted_token`` + 6. sha256 hex of canonical string encoded as UTF-8. + +Returns a 64-character lowercase hex string. Pure function. Deterministic +across machines (sha256 of well-defined UTF-8 bytes). +""" + +from __future__ import annotations + +import hashlib +import re +import string + +# Frozen punctuation set — documented as breaking-change surface. +_PUNCT_SET: str = string.punctuation # !"#$%&'()*+,-./:;<=>?@[\]^_`{|}~ + +# Canonical regex matching lexicon-entry refusal details. +_LEXICON_DETAIL_RE: re.Pattern[str] = re.compile( + r"no primitive or lexicon match for '([^']+)'" +) + + +def lexical_claim_signature( + *, + surface: str, + refusal_detail: str, +) -> str: + """Deterministic sha256 hex of a normalised lexical claim. + + See module docstring for the full normalisation pipeline. + + Parameters + ---------- + surface: + The raw token surface form (e.g. ``"crayons,"``). + refusal_detail: + The ``AuditRow.refusal_detail`` string verbatim. + + Returns + ------- + str + 64-character lowercase hex. + """ + # Step 1 — lowercase + lowered = surface.lower() + # Step 2 — strip punctuation set members from both ends + stripped = lowered.strip(_PUNCT_SET) + # Step 3 — attempt canonical extraction from refusal_detail + match = _LEXICON_DETAIL_RE.search(refusal_detail) + # Step 4 — fallback to stripped surface if regex doesn't match + token = match.group(1) if match else stripped + # Step 5 — canonical string + canonical = "lexical:" + token + # Step 6 — sha256 hex + return hashlib.sha256(canonical.encode("utf-8")).hexdigest() + + +__all__ = ["lexical_claim_signature"] diff --git a/teaching/math_contemplation.py b/teaching/math_contemplation.py index 04973a9c..0e897db8 100644 --- a/teaching/math_contemplation.py +++ b/teaching/math_contemplation.py @@ -1,7 +1,16 @@ -"""ADR-0167 W2-A — audit-to-evidence adapter. +"""ADR-0167 W2-A / W2-B — Audit-to-evidence adapter + lexical claim signature. -Pure deterministic conversion from comprehension audit rows into typed -``MathReaderRefusalEvidence`` records for the teaching corridor. +W2-A deliverable: :func:`audit_to_evidence` and +:func:`audit_problem_to_evidence` convert :class:`AuditRow` sequences into +typed :class:`MathReaderRefusalEvidence` teaching-corridor records. + +W2-B deliverable: For ``sub_type == "lexical"`` evidence, ``claim_signature`` +is computed via :func:`lexical_claim_signature` and the ``evidence_hash`` is +recomputed to incorporate the signature. All other sub_types leave +``claim_signature == ""``. + +Pure module. No filesystem writes, no network calls, no global mutation. +Deterministic: same inputs → byte-identical output across all reruns. """ from __future__ import annotations @@ -9,6 +18,7 @@ from __future__ import annotations from typing import Iterable from generate.comprehension.audit import AuditRow, audit_problem +from teaching.math_claim_signature import lexical_claim_signature from teaching.math_evidence import ( MathReaderRefusalEvidence, SUB_TYPE_FOR_OPERATOR, @@ -21,17 +31,43 @@ def audit_to_evidence( ) -> tuple[MathReaderRefusalEvidence, ...]: """Convert audit rows into typed teaching-corridor evidence records. - Pure function. Deterministic. No filesystem, no network, no mutation. - Sub-type assignment from ``teaching.math_evidence.SUB_TYPE_FOR_OPERATOR``. + Pure function. Deterministic. No filesystem, no network, no mutation. + Sub-type assignment from :data:`teaching.math_evidence.SUB_TYPE_FOR_OPERATOR`. Skips rows with ``missing_operator=None`` (no sub_type → no candidate). + + For ``sub_type == "lexical"``, :func:`lexical_claim_signature` fills the + ``claim_signature`` field and the ``evidence_hash`` incorporates it. + For all other sub_types, ``claim_signature`` remains ``""`` (deferred). + + Input order is preserved. + + Parameters + ---------- + audit_rows: + Iterable of :class:`AuditRow` instances (e.g. from :func:`audit_problem`). + + Returns + ------- + tuple[MathReaderRefusalEvidence, ...] + One record per row whose ``missing_operator`` maps to a known sub_type. """ - out: list[MathReaderRefusalEvidence] = [] + results: list[MathReaderRefusalEvidence] = [] for row in audit_rows: if row.missing_operator is None: continue - sub_type = SUB_TYPE_FOR_OPERATOR[row.missing_operator] - out.append(from_audit_row(row, sub_type, claim_signature="")) - return tuple(out) + sub_type = SUB_TYPE_FOR_OPERATOR.get(row.missing_operator) + if sub_type is None: + continue + if sub_type == "lexical": + sig = lexical_claim_signature( + surface=row.token_text, + refusal_detail=row.refusal_detail, + ) + else: + sig = "" + evidence = from_audit_row(row, sub_type, claim_signature=sig) + results.append(evidence) + return tuple(results) def audit_problem_to_evidence( @@ -39,9 +75,30 @@ def audit_problem_to_evidence( *, case_id: str, ) -> tuple[MathReaderRefusalEvidence, ...]: - """Run reader audit on ``problem_text`` and return mapped evidence rows.""" + """Run the reader over *problem_text* and return evidence records. + + Convenience wrapper that calls :func:`audit_problem` and pipes the + resulting :class:`AuditRow` list through :func:`audit_to_evidence`. + Useful for tests and downstream pipeline work (W3-A). + + Parameters + ---------- + problem_text: + Raw GSM8K-style problem string. + case_id: + Identifier attached to every :class:`AuditRow` (e.g. ``"probe"``). + + Returns + ------- + tuple[MathReaderRefusalEvidence, ...] + Evidence records for any refusals encountered. Empty tuple on full + admission or if the text produced no sentences. + """ _result, rows = audit_problem(problem_text, case_id=case_id) return audit_to_evidence(rows) -__all__ = ["audit_problem_to_evidence", "audit_to_evidence"] +__all__ = [ + "audit_to_evidence", + "audit_problem_to_evidence", +] diff --git a/tests/test_math_claim_signature.py b/tests/test_math_claim_signature.py new file mode 100644 index 00000000..879038c1 --- /dev/null +++ b/tests/test_math_claim_signature.py @@ -0,0 +1,256 @@ +"""Tests for ADR-0167 W2-B: lexical claim signature normalisation. + +Verifies: +- Determinism (same input → same hex) +- Punctuation stripping +- Case insensitivity +- Format invariant (64-char lowercase hex) +- Fallback when refusal_detail doesn't match the canonical regex +- Real-data sanity: no false collisions on audit_brief_11.json +- Real-data dedup: duplicate tokens collapse to one signature +- Non-lexical evidence pins the W2-A invariant (claim_signature stays "") +""" + +from __future__ import annotations + +import hashlib +import json +from pathlib import Path + +import pytest + +from teaching.math_claim_signature import lexical_claim_signature +from teaching.math_contemplation import audit_to_evidence +from teaching.math_evidence import SUB_TYPE_FOR_OPERATOR + +# --------------------------------------------------------------------------- +# Fixture: load audit_brief_11.json once +# --------------------------------------------------------------------------- + +_ARTIFACT_PATH = ( + Path(__file__).parent.parent + / "evals/gsm8k_math/train_sample/v1/audit_brief_11.json" +) + +_LEXICAL_OPS: frozenset[str] = frozenset( + op + for op, sub in SUB_TYPE_FOR_OPERATOR.items() + if sub == "lexical" +) + + +@pytest.fixture(scope="module") +def artifact() -> dict: + return json.loads(_ARTIFACT_PATH.read_text(encoding="utf-8")) + + +@pytest.fixture(scope="module") +def lexical_cases(artifact) -> list[dict]: + return [ + c + for c in artifact["per_case"] + if c.get("missing_operator") in _LEXICAL_OPS + ] + + +# --------------------------------------------------------------------------- +# Core invariant tests +# --------------------------------------------------------------------------- + + +def test_identical_surface_identical_signature(): + sig1 = lexical_claim_signature( + surface="crayons", + refusal_detail="no primitive or lexicon match for 'crayons'", + ) + sig2 = lexical_claim_signature( + surface="crayons", + refusal_detail="no primitive or lexicon match for 'crayons'", + ) + assert sig1 == sig2 + + +def test_different_surface_different_signature(): + sig_a = lexical_claim_signature( + surface="crayons", + refusal_detail="no primitive or lexicon match for 'crayons'", + ) + sig_b = lexical_claim_signature( + surface="oysters", + refusal_detail="no primitive or lexicon match for 'oysters'", + ) + assert sig_a != sig_b + + +def test_punctuation_strip(): + """crayons, crayons. and crayons should all collapse to the same signature.""" + base_detail = "no primitive or lexicon match for 'crayons'" + sig_plain = lexical_claim_signature(surface="crayons", refusal_detail=base_detail) + sig_comma = lexical_claim_signature(surface="crayons,", refusal_detail=base_detail) + sig_period = lexical_claim_signature(surface="crayons.", refusal_detail=base_detail) + # Note: when refusal_detail matches the regex, the extracted token wins + # (which is always un-punctuated). All three should produce the same sig. + assert sig_plain == sig_comma == sig_period + + +def test_case_insensitive(): + detail = "no primitive or lexicon match for 'crayons'" + sig_lower = lexical_claim_signature(surface="crayons", refusal_detail=detail) + sig_upper = lexical_claim_signature(surface="Crayons", refusal_detail=detail) + assert sig_lower == sig_upper + + +def test_signature_is_64_char_lowercase_hex(): + sig = lexical_claim_signature( + surface="widgets", + refusal_detail="no primitive or lexicon match for 'widgets'", + ) + assert len(sig) == 64 + assert sig == sig.lower() + assert all(c in "0123456789abcdef" for c in sig) + + +def test_extraction_falls_back_to_surface(): + """When refusal_detail doesn't match the regex, surface is used verbatim.""" + sig_fallback = lexical_claim_signature( + surface="10%", + refusal_detail="fraction/percentage literal at position 7 is out-of-scope (eval only)", + ) + # Manually compute expected: strip punctuation from "10%", lowercase → "10" + # Wait — "%" is in string.punctuation, so strip yields "10" + # canonical = "lexical:10" + expected = hashlib.sha256("lexical:10".encode("utf-8")).hexdigest() + assert sig_fallback == expected + + +def test_extraction_fallback_is_still_deterministic(): + sig1 = lexical_claim_signature( + surface="1/4", + refusal_detail="fraction/percentage literal at position 11 is out-of-scope (eval only)", + ) + sig2 = lexical_claim_signature( + surface="1/4", + refusal_detail="fraction/percentage literal at position 11 is out-of-scope (eval only)", + ) + assert sig1 == sig2 + + +# --------------------------------------------------------------------------- +# Real-data tests +# --------------------------------------------------------------------------- + + +def test_real_data_no_false_collisions(lexical_cases): + """Distinct token surfaces must produce distinct signatures.""" + # Build {token_text → signature} using the canonical regex path where applicable + token_to_sig: dict[str, str] = {} + for case in lexical_cases: + token = case["token_text"] + sig = lexical_claim_signature( + surface=token, + refusal_detail=case.get("refusal_detail", ""), + ) + if token in token_to_sig: + assert token_to_sig[token] == sig, ( + f"Same token '{token}' produced different signatures" + ) + else: + token_to_sig[token] = sig + # All distinct tokens should map to distinct signatures + sigs = list(token_to_sig.values()) + assert len(sigs) == len(set(sigs)), ( + "False collision: distinct tokens collapsed to same signature" + ) + + +def test_real_data_collapses_duplicates(lexical_cases): + """Two cases with the same extracted token collapse to one signature. + + In audit_brief_11.json each token is unique, so we simulate by taking + a single token and verifying two invocations with different case context + produce the same signature (dedup works across case boundaries). + """ + # Pick the first lexical case and duplicate it with a different "caller" + case = lexical_cases[0] + sig_a = lexical_claim_signature( + surface=case["token_text"], + refusal_detail=case.get("refusal_detail", ""), + ) + sig_b = lexical_claim_signature( + surface=case["token_text"], + refusal_detail=case.get("refusal_detail", ""), + ) + assert sig_a == sig_b, "Same token should collapse to the same signature" + + +# --------------------------------------------------------------------------- +# W2-A invariant pin: non-lexical evidence has empty claim_signature +# --------------------------------------------------------------------------- + + +def test_non_lexical_evidence_has_empty_signature(artifact): + """audit_to_evidence must leave claim_signature == '' for non-lexical rows.""" + from generate.comprehension.audit import AuditRow + + non_lexical_cases = [ + c + for c in artifact["per_case"] + if c.get("missing_operator") is not None + and c.get("missing_operator") not in _LEXICAL_OPS + ] + assert non_lexical_cases, "Expected at least one non-lexical case in artifact" + + # Build AuditRow instances for non-lexical cases + rows = [ + AuditRow( + case_id=c["case_id"], + sentence_index=c["sentence_index"], + token_index=c.get("token_index", 0), + token_text=c.get("token_text", ""), + recognized_terms=tuple(c.get("recognized_terms", [])), + skipped_frame=c.get("skipped_frame"), + missing_operator=c["missing_operator"], + refusal_reason=c["refusal_reason"], + refusal_detail=c.get("refusal_detail", ""), + ) + for c in non_lexical_cases + ] + evidence_records = audit_to_evidence(rows) + for ev in evidence_records: + assert ev.claim_signature == "", ( + f"Non-lexical sub_type '{ev.sub_type}' should have empty claim_signature, " + f"got {ev.claim_signature!r}" + ) + + +# --------------------------------------------------------------------------- +# Lexical evidence in contemplation gets non-empty signature +# --------------------------------------------------------------------------- + + +def test_lexical_evidence_gets_non_empty_signature(lexical_cases): + """audit_to_evidence fills claim_signature for lexical rows.""" + from generate.comprehension.audit import AuditRow + + rows = [ + AuditRow( + case_id=c["case_id"], + sentence_index=c["sentence_index"], + token_index=c.get("token_index", 0), + token_text=c.get("token_text", ""), + recognized_terms=tuple(c.get("recognized_terms", [])), + skipped_frame=c.get("skipped_frame"), + missing_operator=c["missing_operator"], + refusal_reason=c["refusal_reason"], + refusal_detail=c.get("refusal_detail", ""), + ) + for c in lexical_cases + ] + evidence_records = audit_to_evidence(rows) + assert evidence_records, "Expected at least one lexical evidence record" + for ev in evidence_records: + assert ev.sub_type == "lexical" + assert ev.claim_signature != "", ( + f"Lexical evidence for case {ev.case_id} must have non-empty claim_signature" + ) + assert len(ev.claim_signature) == 64