diff --git a/teaching/math_lexical_ratification.py b/teaching/math_lexical_ratification.py new file mode 100644 index 00000000..fbe28e22 --- /dev/null +++ b/teaching/math_lexical_ratification.py @@ -0,0 +1,308 @@ +"""ADR-0167 W2-D — LexicalClaim ratification into math lexicon sources. + +This module is the explicit post-review mutation boundary for lexical math +reader evidence. It edits only per-category source files under +``en_core_math_v1/lexicon``; it does not regenerate the compiled +``lexicon.jsonl`` and therefore does not rewrite the manifest checksum. +""" + +from __future__ import annotations + +import hashlib +import json +import re +import string +from dataclasses import dataclass +from datetime import date +from pathlib import Path +from typing import Final + +from teaching.math_evidence import MathReaderRefusalEvidence, from_audit_row + + +SAFE_CATEGORIES: Final[frozenset[str]] = frozenset( + { + "drain_token", + } +) + +_FRAME_OPENER_CATEGORIES: Final[frozenset[str]] = frozenset( + { + "accumulation_verb", + "capacity_verb", + "copula_verb", + "depletion_verb", + "possession_verb", + "transfer_verb", + } +) +_REVIEWER_RE: Final[re.Pattern[str]] = re.compile(r"[^a-z0-9_-]+") +_TRIM_PUNCTUATION: Final[str] = string.punctuation + + +class RatificationError(ValueError): + """Base class for lexical ratification rejections.""" + + +class WrongClaimSubType(RatificationError): + """Raised when a non-lexical evidence record reaches this handler.""" + + +class WrongZeroViolationCandidate(RatificationError): + """Raised when a ratification could open a wrong>0 admission path.""" + + +class AlreadyRatified(RatificationError): + """Raised when the target surface is already covered by the source file.""" + + +class EvidenceTampering(RatificationError): + """Raised when the evidence hash no longer matches canonical bytes.""" + + +class UnknownCategory(RatificationError): + """Raised when the requested category is not a known lexicon source.""" + + +@dataclass(frozen=True, slots=True) +class RatificationReceipt: + target_file: str + lemma: str + category: str + provenance: str + file_sha256_before: str + file_sha256_after: str + evidence_hash: str + is_alias: bool + aliased_to: str | None + + +def _repo_root() -> Path: + here = Path(__file__).resolve() + for parent in here.parents: + if (parent / "pyproject.toml").exists() or (parent / "setup.cfg").exists(): + return parent + return here.parents[1] + + +def _default_pack_root() -> Path: + return _repo_root() / "language_packs" / "data" / "en_core_math_v1" + + +def _sha256_file(path: Path) -> str: + return hashlib.sha256(path.read_bytes()).hexdigest() + + +def _normalise_surface(surface: str) -> str: + return surface.lower().strip(_TRIM_PUNCTUATION) + + +def _reviewer_slug(reviewer: str) -> str: + slug = _REVIEWER_RE.sub("_", reviewer.lower()).strip("_-") + if not slug: + raise RatificationError("reviewer must contain at least one safe character") + return slug + + +def _read_entries(path: Path) -> list[dict[str, object]]: + entries: list[dict[str, object]] = [] + for line_number, line in enumerate(path.read_text(encoding="utf-8").splitlines(), start=1): + if not line.strip(): + continue + record = json.loads(line) + if not isinstance(record.get("lemma"), str): + raise RatificationError(f"{path}: line {line_number} missing string lemma") + if not isinstance(record.get("category"), str): + raise RatificationError(f"{path}: line {line_number} missing string category") + aliases = record.get("aliases", []) + if not isinstance(aliases, list) or not all( + isinstance(alias, str) for alias in aliases + ): + raise RatificationError(f"{path}: line {line_number} has invalid aliases") + entries.append(record) + return entries + + +def _write_entries(path: Path, entries: list[dict[str, object]]) -> None: + ordered = sorted(entries, key=lambda item: str(item["lemma"])) + lines = [json.dumps(entry, ensure_ascii=False, sort_keys=False) for entry in ordered] + path.write_text("\n".join(lines) + "\n", encoding="utf-8") + + +def _all_source_entries(pack_root: Path) -> list[dict[str, object]]: + source_dir = pack_root / "lexicon" + entries: list[dict[str, object]] = [] + for path in sorted(source_dir.glob("*.jsonl")): + entries.extend(_read_entries(path)) + return entries + + +def _entry_surfaces(entry: dict[str, object]) -> set[str]: + lemma = str(entry["lemma"]).lower() + aliases = entry.get("aliases", []) + return {lemma, *(str(alias).lower() for alias in aliases)} + + +def _find_frame_opener_conflict(pack_root: Path, lemma: str) -> tuple[str, str] | None: + for entry in _all_source_entries(pack_root): + category = str(entry["category"]) + if category not in _FRAME_OPENER_CATEGORIES: + continue + if lemma in _entry_surfaces(entry): + return str(entry["lemma"]), category + return None + + +def _stem_alias_parent( + entries: list[dict[str, object]], + lemma: str, +) -> dict[str, object] | None: + for entry in entries: + parent = str(entry["lemma"]).lower() + if parent == lemma: + return entry + if lemma in _entry_surfaces(entry): + return entry + if lemma in { + f"{parent}s", + f"{parent}es", + f"{parent}ed", + f"{parent}ing", + }: + return entry + if parent.endswith("y") and lemma == f"{parent[:-1]}ies": + return entry + return None + + +def _validate_evidence(claim: MathReaderRefusalEvidence) -> None: + if claim.sub_type != "lexical": + raise WrongClaimSubType( + f"Lexical ratification requires sub_type='lexical'; got {claim.sub_type!r}" + ) + recomputed = from_audit_row( + claim.audit_row, + claim.sub_type, + claim_signature=claim.claim_signature, + ) + if recomputed.evidence_hash != claim.evidence_hash: + raise EvidenceTampering( + "MathReaderRefusalEvidence hash does not match canonical audit row bytes" + ) + + +def apply_lexical_claim( + *, + claim: MathReaderRefusalEvidence, + category: str, + reviewer: str, + pack_root: Path | None = None, +) -> RatificationReceipt: + """Apply a reviewed lexical claim to a math source lexicon file. + + Preconditions: + - ``claim.sub_type == "lexical"`` or :class:`WrongClaimSubType` is raised. + - ``category`` is in :data:`SAFE_CATEGORIES`; frame-opener categories raise + :class:`WrongZeroViolationCandidate`. + - the target surface is not already a frame-opener lemma or alias. + - ``claim.evidence_hash`` matches a recomputation from ``claim.audit_row``. + + The write target is ``{pack_root}/lexicon/{category}.jsonl``. Entries are + sorted alphabetically by lemma. If the ratified surface is an inflectional + match for an existing lemma (for example ``weight`` -> ``weights``), it is + appended as an alias; otherwise a new lemma entry is written. + + The compiled ``lexicon.jsonl`` and ``manifest.json`` are intentionally not + regenerated. The loader verifies the manifest against compiled bytes but + reads per-category source files for alias-aware entries. + """ + _validate_evidence(claim) + + root = (pack_root if pack_root is not None else _default_pack_root()).resolve() + source_dir = root / "lexicon" + target_file = source_dir / f"{category}.jsonl" + + if category not in SAFE_CATEGORIES: + if category in _FRAME_OPENER_CATEGORIES: + raise WrongZeroViolationCandidate( + f"category {category!r} can open reader frames; LexicalClaim " + "ratification is restricted to drain-class categories" + ) + if not target_file.exists(): + raise UnknownCategory(f"unknown math lexicon category: {category!r}") + raise WrongZeroViolationCandidate( + f"category {category!r} is outside SAFE_CATEGORIES={sorted(SAFE_CATEGORIES)!r}" + ) + if not target_file.exists(): + raise UnknownCategory(f"missing source lexicon file for category: {category!r}") + + lemma = _normalise_surface(claim.audit_row.token_text) + if not lemma: + raise WrongZeroViolationCandidate("empty lexical surface cannot be ratified") + + conflict = _find_frame_opener_conflict(root, lemma) + if conflict is not None: + parent, conflict_category = conflict + raise WrongZeroViolationCandidate( + f"surface {lemma!r} already resolves through frame-opener " + f"{conflict_category!r} lemma {parent!r}" + ) + + provenance = f"phase_2_reader_ratified_{_reviewer_slug(reviewer)}_{date.today().isoformat()}" + before = _sha256_file(target_file) + entries = _read_entries(target_file) + target_relative = str(target_file.relative_to(_repo_root())) if target_file.is_relative_to(_repo_root()) else str(target_file) + + parent = _stem_alias_parent(entries, lemma) + is_alias = False + aliased_to: str | None = None + + if parent is not None: + parent_lemma = str(parent["lemma"]) + aliases = [str(alias) for alias in parent.get("aliases", [])] + surfaces = {parent_lemma.lower(), *(alias.lower() for alias in aliases)} + if lemma in surfaces: + raise AlreadyRatified( + f"lexical claim {lemma!r} is already ratified in {category!r}" + ) + aliases.append(lemma) + parent["aliases"] = sorted(set(aliases)) + is_alias = True + aliased_to = parent_lemma + else: + entries.append( + { + "lemma": lemma, + "category": category, + "aliases": [], + "provenance": provenance, + } + ) + + _write_entries(target_file, entries) + after = _sha256_file(target_file) + + return RatificationReceipt( + target_file=target_relative, + lemma=lemma, + category=category, + provenance=provenance, + file_sha256_before=before, + file_sha256_after=after, + evidence_hash=claim.evidence_hash, + is_alias=is_alias, + aliased_to=aliased_to, + ) + + +__all__ = [ + "AlreadyRatified", + "EvidenceTampering", + "RatificationError", + "RatificationReceipt", + "SAFE_CATEGORIES", + "UnknownCategory", + "WrongClaimSubType", + "WrongZeroViolationCandidate", + "apply_lexical_claim", +] diff --git a/tests/test_math_lexical_ratification.py b/tests/test_math_lexical_ratification.py new file mode 100644 index 00000000..8c79fedb --- /dev/null +++ b/tests/test_math_lexical_ratification.py @@ -0,0 +1,304 @@ +"""ADR-0167 W2-D — LexicalClaim ratification handler tests.""" + +from __future__ import annotations + +import functools +import hashlib +import json +import shutil +from pathlib import Path + +import pytest + +from generate.comprehension import lexicon as comprehension_lexicon +from generate.comprehension import lifecycle +from generate.comprehension.audit import AuditRow, audit_problem +from generate.comprehension.state import ReaderRefusal +from teaching.math_evidence import from_audit_row +from teaching.math_lexical_ratification import ( + AlreadyRatified, + EvidenceTampering, + SAFE_CATEGORIES, + UnknownCategory, + WrongClaimSubType, + WrongZeroViolationCandidate, + apply_lexical_claim, +) + + +REPO_ROOT = Path(__file__).resolve().parents[1] +PACK_ROOT = REPO_ROOT / "language_packs" / "data" / "en_core_math_v1" +CASES_PATH = REPO_ROOT / "evals" / "gsm8k_math" / "train_sample" / "v1" / "cases.jsonl" + + +@pytest.fixture() +def pack_copy(tmp_path: Path) -> Path: + target = tmp_path / "en_core_math_v1" + shutil.copytree(PACK_ROOT, target) + comprehension_lexicon._CACHE.clear() + lifecycle._get_lexicon.cache_clear() + yield target + comprehension_lexicon._CACHE.clear() + lifecycle._get_lexicon.cache_clear() + + +def _row(surface: str, *, missing_operator: str = "lexicon_entry") -> AuditRow: + return AuditRow( + case_id=f"case-{surface}", + sentence_index=0, + token_index=2, + token_text=surface, + recognized_terms=("Ava", "counts"), + skipped_frame=None, + missing_operator=missing_operator, + refusal_reason="unknown_word", + refusal_detail=f"no primitive or lexicon match for '{surface}'", + ) + + +def _claim(surface: str, *, sub_type: str = "lexical"): + return from_audit_row(_row(surface), sub_type) # type: ignore[arg-type] + + +def _entries(path: Path) -> list[dict]: + return [ + json.loads(line) + for line in path.read_text(encoding="utf-8").splitlines() + if line.strip() + ] + + +def _manifest_sha(pack_root: Path) -> str: + return hashlib.sha256((pack_root / "manifest.json").read_bytes()).hexdigest() + + +def _use_pack_for_reader(monkeypatch: pytest.MonkeyPatch, pack_root: Path) -> None: + comprehension_lexicon._CACHE.clear() + + @functools.cache + def _tmp_lexicon(): + return comprehension_lexicon.load_lexicon(pack_root) + + monkeypatch.setattr(lifecycle, "_get_lexicon", _tmp_lexicon) + + +def test_apply_lexical_claim_writes_lemma(pack_copy: Path) -> None: + receipt = apply_lexical_claim( + claim=_claim("widgets"), + category="drain_token", + reviewer="Ada", + pack_root=pack_copy, + ) + + assert receipt.lemma == "widgets" + assert receipt.category == "drain_token" + assert receipt.is_alias is False + assert receipt.aliased_to is None + entry = next(e for e in _entries(pack_copy / "lexicon" / "drain_token.jsonl") if e["lemma"] == "widgets") + assert entry == { + "lemma": "widgets", + "category": "drain_token", + "aliases": [], + "provenance": receipt.provenance, + } + lex = comprehension_lexicon.load_lexicon(pack_copy) + resolved = comprehension_lexicon.lookup(lex, "widgets") + assert resolved is not None + assert resolved.category == "drain_token" + + +def test_receipt_records_before_after_sha(pack_copy: Path) -> None: + receipt = apply_lexical_claim( + claim=_claim("widgets"), + category="drain_token", + reviewer="Ada", + pack_root=pack_copy, + ) + + assert len(receipt.file_sha256_before) == 64 + assert len(receipt.file_sha256_after) == 64 + assert receipt.file_sha256_before != receipt.file_sha256_after + assert receipt.evidence_hash == _claim("widgets").evidence_hash + + +def test_idempotent_raises_already_ratified(pack_copy: Path) -> None: + claim = _claim("widgets") + apply_lexical_claim( + claim=claim, + category="drain_token", + reviewer="Ada", + pack_root=pack_copy, + ) + + with pytest.raises(AlreadyRatified, match="already ratified"): + apply_lexical_claim( + claim=claim, + category="drain_token", + reviewer="Ada", + pack_root=pack_copy, + ) + + +def test_rejects_non_lexical_sub_type(pack_copy: Path) -> None: + with pytest.raises(WrongClaimSubType): + apply_lexical_claim( + claim=_claim("widgets", sub_type="frame"), + category="drain_token", + reviewer="Ada", + pack_root=pack_copy, + ) + + +def test_rejects_unsafe_category(pack_copy: Path) -> None: + with pytest.raises(WrongZeroViolationCandidate, match="open reader frames"): + apply_lexical_claim( + claim=_claim("widgets"), + category="accumulation_verb", + reviewer="Ada", + pack_root=pack_copy, + ) + + +def test_rejects_unknown_category(pack_copy: Path) -> None: + with pytest.raises(UnknownCategory): + apply_lexical_claim( + claim=_claim("widgets"), + category="not_a_category", + reviewer="Ada", + pack_root=pack_copy, + ) + + +def test_rejects_evidence_tampering(pack_copy: Path) -> None: + claim = _claim("widgets") + object.__setattr__(claim, "evidence_hash", "0" * 64) + + with pytest.raises(EvidenceTampering): + apply_lexical_claim( + claim=claim, + category="drain_token", + reviewer="Ada", + pack_root=pack_copy, + ) + + +def test_alphabetical_sort_preserved(pack_copy: Path) -> None: + apply_lexical_claim( + claim=_claim("aardwidgets"), + category="drain_token", + reviewer="Ada", + pack_root=pack_copy, + ) + + lemmas = [entry["lemma"] for entry in _entries(pack_copy / "lexicon" / "drain_token.jsonl")] + assert lemmas == sorted(lemmas) + + +def test_manifest_checksum_unchanged(pack_copy: Path) -> None: + manifest_bytes_before = (pack_copy / "manifest.json").read_bytes() + manifest_sha_before = _manifest_sha(pack_copy) + declared_before = json.loads(manifest_bytes_before)["checksum"] + + apply_lexical_claim( + claim=_claim("widgets"), + category="drain_token", + reviewer="Ada", + pack_root=pack_copy, + ) + + manifest_bytes_after = (pack_copy / "manifest.json").read_bytes() + assert manifest_bytes_after == manifest_bytes_before + assert _manifest_sha(pack_copy) == manifest_sha_before + assert json.loads(manifest_bytes_after)["checksum"] == declared_before + + +def test_hazard_case_0050_remains_refused(pack_copy: Path, monkeypatch: pytest.MonkeyPatch) -> None: + with pytest.raises(AlreadyRatified): + apply_lexical_claim( + claim=_claim("along"), + category="drain_token", + reviewer="Ada", + pack_root=pack_copy, + ) + _use_pack_for_reader(monkeypatch, pack_copy) + cases = [json.loads(line) for line in CASES_PATH.read_text().splitlines()] + case = next(c for c in cases if c["case_id"] == "gsm8k-train-sample-v1-0050") + + result, _rows = audit_problem(case["question"], case_id=case["case_id"]) + + assert isinstance(result, ReaderRefusal) + assert result.sentence_index == 0 + + +def test_audit_artifact_still_has_zero_admissions_after_tmp_ratification( + pack_copy: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + apply_lexical_claim( + claim=_claim("widgets"), + category="drain_token", + reviewer="Ada", + pack_root=pack_copy, + ) + _use_pack_for_reader(monkeypatch, pack_copy) + cases = [json.loads(line) for line in CASES_PATH.read_text().splitlines()] + + for case in cases: + result, _rows = audit_problem(case["question"], case_id=case["case_id"]) + assert isinstance(result, ReaderRefusal) or result is None, case["case_id"] + + +def test_load_lexicon_resolves_newly_ratified_lemma(pack_copy: Path) -> None: + apply_lexical_claim( + claim=_claim("ratifiables"), + category="drain_token", + reviewer="Ada", + pack_root=pack_copy, + ) + comprehension_lexicon._CACHE.clear() + + lex = comprehension_lexicon.load_lexicon(pack_copy) + resolved = comprehension_lexicon.lookup(lex, "ratifiables") + + assert resolved is not None + assert resolved.lemma == "ratifiables" + assert resolved.category == "drain_token" + + +def test_alias_path_when_existing_lemma_is_stem(pack_copy: Path) -> None: + apply_lexical_claim( + claim=_claim("widget"), + category="drain_token", + reviewer="Ada", + pack_root=pack_copy, + ) + + receipt = apply_lexical_claim( + claim=_claim("widgets"), + category="drain_token", + reviewer="Ada", + pack_root=pack_copy, + ) + + assert receipt.is_alias is True + assert receipt.aliased_to == "widget" + entry = next(e for e in _entries(pack_copy / "lexicon" / "drain_token.jsonl") if e["lemma"] == "widget") + assert "widgets" in entry["aliases"] + lex = comprehension_lexicon.load_lexicon(pack_copy) + resolved = comprehension_lexicon.lookup(lex, "widgets") + assert resolved is not None + assert resolved.lemma == "widget" + + +def test_rejects_existing_frame_opener_surface(pack_copy: Path) -> None: + with pytest.raises(WrongZeroViolationCandidate, match="frame-opener"): + apply_lexical_claim( + claim=_claim("earn"), + category="drain_token", + reviewer="Ada", + pack_root=pack_copy, + ) + + +def test_safe_categories_w2d_scope_is_drain_token_only() -> None: + assert SAFE_CATEGORIES == frozenset({"drain_token"})