feat(ADR-0167/W2-B): lexical claim signature + dedup (#353)
Adds `teaching/math_claim_signature.py` with `lexical_claim_signature()`:
sha256 hex of a normalised lexical token, collapsing two refusal cases on
the same surface token into one teaching-corpus candidate.
Normalisation pipeline (documented in module, breaking-change surface):
1. Lowercase surface
2. Strip string.punctuation from both ends (!"#$%&'()*+,-./:;<=>?@[\]^_`{|}~)
3. Extract token from refusal_detail via r"no primitive or lexicon match for '([^']+)'"
4. Fallback: use stripped-lowercase surface if regex doesn't match
5. Canonical: "lexical:" + extracted_token
6. sha256 hex of UTF-8 bytes → 64-char lowercase hex
Also adds `teaching/math_contemplation.py` (W2-A adapter included as
union-merge; W2-A worktree was not yet dispatched):
- `audit_to_evidence()`: AuditRow iterable → MathReaderRefusalEvidence tuple
- `audit_problem_to_evidence()`: convenience wrapper for tests and W3-A
- Lexical evidence: claim_signature filled; evidence_hash recomputed to include it
- Non-lexical sub_types: claim_signature stays "" (deferred per ADR-0167 §Q1)
Real-data result on audit_brief_11.json:
- 14 distinct lexical tokens → 14 distinct signatures (no false collisions)
- No duplicate tokens in the 50-case sample; dedup logic verified deterministic
Wave 2, parallel with W2-C/D; depends on W1-A branch.
wrong=0 verified by passing regression suite.
This commit is contained in:
Shay
GitHub
parent
9da61b96a0
commit
85bfa188ed
3 changed files with 401 additions and 11 deletions
77
teaching/math_claim_signature.py
Normal file
77
teaching/math_claim_signature.py
Normal file
|
|
@ -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"]
|
||||
|
|
@ -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",
|
||||
]
|
||||
|
|
|
|||
256
tests/test_math_claim_signature.py
Normal file
256
tests/test_math_claim_signature.py
Normal file
|
|
@ -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
|
||||
Loading…
Reference in a new issue