From 4570c2c70e42539d7150b8dd6961d76e07bdbfce Mon Sep 17 00:00:00 2001 From: Shay Date: Tue, 26 May 2026 20:08:27 -0700 Subject: [PATCH] =?UTF-8?q?feat(comprehension):=20operational=20lexicon=20?= =?UTF-8?q?loader=20for=20en=5Fcore=5Fmath=5Fv1=20(ADR-0164=20=C2=A7Decisi?= =?UTF-8?q?on=20=C2=A71)=20(#325)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implements generate/comprehension/lexicon.py: loads per-category source files from en_core_math_v1/lexicon/*.jsonl (full schema including aliases), verifies manifest checksum against compiled lexicon.jsonl for pack integrity, and provides O(1) case-folded surface lookups. Module-level cache keyed on (path, mtime_ns, sha256) avoids redundant I/O. Exports: LexiconEntry, Lexicon, LexiconLoadError, load_lexicon(), lookup(). MappingProxyType over internal dicts prevents callers from mutating cached state. 29 tests cover load, checksum, category completeness, alias resolution, mutual-exclusion detection, determinism, and cache identity. --- generate/comprehension/lexicon.py | 198 ++++++++++++++++++++++ tests/test_lexicon.py | 273 ++++++++++++++++++++++++++++++ 2 files changed, 471 insertions(+) create mode 100644 generate/comprehension/lexicon.py create mode 100644 tests/test_lexicon.py diff --git a/generate/comprehension/lexicon.py b/generate/comprehension/lexicon.py new file mode 100644 index 00000000..f219bc44 --- /dev/null +++ b/generate/comprehension/lexicon.py @@ -0,0 +1,198 @@ +"""ADR-0164 §Decision §1 — operational lexicon loader for en_core_math_v1. + +Provides word → category lookups for the reader's apply_word step 2. +I/O happens only at load time; all lookups are O(1) dict hits. +""" + +from __future__ import annotations + +import hashlib +import json +import os +import types +from dataclasses import dataclass +from pathlib import Path +from typing import Mapping + +# --------------------------------------------------------------------------- +# Typed error +# --------------------------------------------------------------------------- + + +class LexiconLoadError(ValueError): + """Raised on any load-time failure (checksum mismatch, conflict, etc.).""" + + +# --------------------------------------------------------------------------- +# Data types +# --------------------------------------------------------------------------- + + +@dataclass(frozen=True, slots=True) +class LexiconEntry: + lemma: str + category: str + aliases: tuple[str, ...] + provenance: str + + +@dataclass(frozen=True, slots=True) +class Lexicon: + by_surface: Mapping[str, LexiconEntry] + by_category: Mapping[str, tuple[LexiconEntry, ...]] + pack_manifest_sha256: str + source_pack_id: str + + def __hash__(self) -> int: + # Content-derived hash: same pack bytes → same sha256 → same hash. + return hash((self.pack_manifest_sha256, self.source_pack_id)) + + def __eq__(self, other: object) -> bool: + if not isinstance(other, Lexicon): + return NotImplemented + return ( + dict(self.by_surface) == dict(other.by_surface) + and self.pack_manifest_sha256 == other.pack_manifest_sha256 + and self.source_pack_id == other.source_pack_id + ) + + +# --------------------------------------------------------------------------- +# Module-level cache — keyed on (resolved_path_str, mtime_ns, sha256) +# --------------------------------------------------------------------------- + +_CACHE: dict[tuple[str, int, str], Lexicon] = {} + +_DEFAULT_PACK_RELPATH = Path("language_packs") / "data" / "en_core_math_v1" + + +def _repo_root() -> Path: + here = Path(__file__).resolve() + # Walk up until we find a directory that has pyproject.toml or setup.cfg. + candidate = here + for _ in range(10): + candidate = candidate.parent + if (candidate / "pyproject.toml").exists() or (candidate / "setup.cfg").exists(): + return candidate + # Fallback: three levels up from this file + return here.parent.parent.parent + + +def _sha256_file(path: Path) -> str: + return hashlib.sha256(path.read_bytes()).hexdigest() + + +# --------------------------------------------------------------------------- +# Loading +# --------------------------------------------------------------------------- + + +def load_lexicon(pack_path: Path | None = None) -> Lexicon: + """Load and cache the operational lexicon. + + Reads per-category source files from ``pack_path/lexicon/*.jsonl`` for + full entry schema (including aliases). Verifies the manifest checksum + against the compiled ``lexicon.jsonl`` for pack integrity. + """ + resolved = ( + (pack_path if isinstance(pack_path, Path) else Path(pack_path)).resolve() + if pack_path is not None + else (_repo_root() / _DEFAULT_PACK_RELPATH).resolve() + ) + + compiled_path = resolved / "lexicon.jsonl" + manifest_path = resolved / "manifest.json" + + # Stable cache key based on compiled file mtime + sha256. + mtime_ns = os.stat(compiled_path).st_mtime_ns + actual_sha256 = _sha256_file(compiled_path) + cache_key = (str(resolved), mtime_ns, actual_sha256) + + cached = _CACHE.get(cache_key) + if cached is not None: + return cached + + # Verify manifest checksum against the compiled lexicon bytes. + manifest = json.loads(manifest_path.read_text(encoding="utf-8")) + declared_sha256: str = manifest.get("checksum", "") + if declared_sha256 != actual_sha256: + raise LexiconLoadError( + f"Manifest checksum mismatch for {resolved!r}: " + f"declared={declared_sha256!r}, actual={actual_sha256!r}" + ) + + pack_id: str = manifest.get("pack_id", resolved.name) + + # Load from source per-category files (they carry aliases). + source_dir = resolved / "lexicon" + if not source_dir.is_dir(): + raise LexiconLoadError( + f"Source lexicon directory not found: {source_dir}" + ) + + entries: list[LexiconEntry] = [] + for src_file in sorted(source_dir.glob("*.jsonl")): + for line in src_file.read_text(encoding="utf-8").splitlines(): + line = line.strip() + if not line: + continue + rec = json.loads(line) + entry = LexiconEntry( + lemma=rec["lemma"], + category=rec["category"], + aliases=tuple(rec.get("aliases", [])), + provenance=rec.get("provenance", ""), + ) + entries.append(entry) + + # Build by_surface (case-folded). Check for mutual-exclusion violations. + by_surface_mut: dict[str, LexiconEntry] = {} + conflicts: list[str] = [] + + for entry in entries: + surfaces = [entry.lemma.lower()] + [a.lower() for a in entry.aliases] + for surf in surfaces: + if surf in by_surface_mut: + existing = by_surface_mut[surf] + if existing.category != entry.category: + conflicts.append( + f"Surface {surf!r} is in both " + f"{existing.category!r} (lemma={existing.lemma!r}) " + f"and {entry.category!r} (lemma={entry.lemma!r})" + ) + else: + by_surface_mut[surf] = entry + + if conflicts: + raise LexiconLoadError( + "Mutual-exclusion violation in pack " + f"{pack_id!r}: {'; '.join(conflicts)}" + ) + + # Build by_category sorted by lemma within each group. + by_cat_mut: dict[str, list[LexiconEntry]] = {} + for entry in entries: + by_cat_mut.setdefault(entry.category, []).append(entry) + by_category_frozen: dict[str, tuple[LexiconEntry, ...]] = { + cat: tuple(sorted(lst, key=lambda e: e.lemma)) + for cat, lst in by_cat_mut.items() + } + + lexicon = Lexicon( + by_surface=types.MappingProxyType(by_surface_mut), + by_category=types.MappingProxyType(by_category_frozen), + pack_manifest_sha256=actual_sha256, + source_pack_id=pack_id, + ) + _CACHE[cache_key] = lexicon + return lexicon + + +# --------------------------------------------------------------------------- +# Lookup +# --------------------------------------------------------------------------- + + +def lookup(lexicon: Lexicon, surface: str) -> LexiconEntry | None: + """Case-fold surface and return its entry, or None if unknown.""" + return lexicon.by_surface.get(surface.lower()) diff --git a/tests/test_lexicon.py b/tests/test_lexicon.py new file mode 100644 index 00000000..263b3f44 --- /dev/null +++ b/tests/test_lexicon.py @@ -0,0 +1,273 @@ +"""Tests for generate.comprehension.lexicon (ADR-0164 §Decision §1).""" + +from __future__ import annotations + +import hashlib +import json +from pathlib import Path + +import pytest + +from generate.comprehension.lexicon import ( + Lexicon, + LexiconEntry, + LexiconLoadError, + load_lexicon, + lookup, +) + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +# Categories declared in ADR-0164 §Decision §1 that must be present in the +# en_core_math_v1 pack shipped with PR #322. +REQUIRED_CATEGORIES = { + "accumulation_verb", + "capacity_verb", + "currency_unit_noun", + "depletion_verb", + "entity_pronoun", + "possession_verb", + "proper_noun_entity_female", + "proper_noun_entity_male", + "question_open", + "residual_modifier", + "transfer_verb", +} + + +# --------------------------------------------------------------------------- +# Load +# --------------------------------------------------------------------------- + + +class TestLoad: + def test_load_succeeds(self) -> None: + lex = load_lexicon() + assert isinstance(lex, Lexicon) + + def test_all_required_categories_present(self) -> None: + lex = load_lexicon() + present = set(lex.by_category.keys()) + missing = REQUIRED_CATEGORIES - present + assert not missing, f"Missing categories: {sorted(missing)}" + + def test_by_category_values_are_sorted_tuples(self) -> None: + lex = load_lexicon() + for cat, entries in lex.by_category.items(): + assert isinstance(entries, tuple), f"{cat}: not a tuple" + lemmas = [e.lemma for e in entries] + assert lemmas == sorted(lemmas), f"{cat}: entries not sorted by lemma" + + def test_by_surface_is_mapping_proxy(self) -> None: + import types as _types + lex = load_lexicon() + assert isinstance(lex.by_surface, _types.MappingProxyType) + + def test_by_category_is_mapping_proxy(self) -> None: + import types as _types + lex = load_lexicon() + assert isinstance(lex.by_category, _types.MappingProxyType) + + def test_pack_id_is_math_v1(self) -> None: + lex = load_lexicon() + assert lex.source_pack_id == "en_core_math_v1" + + def test_sha256_field_is_populated(self) -> None: + lex = load_lexicon() + assert len(lex.pack_manifest_sha256) == 64 + + +# --------------------------------------------------------------------------- +# Checksum +# --------------------------------------------------------------------------- + + +class TestChecksum: + def test_checksum_mismatch_raises(self, tmp_path: Path) -> None: + """Tampering the manifest checksum must raise LexiconLoadError.""" + import shutil + + # Copy the real pack into a temp location. + real_pack = ( + Path(__file__).resolve().parent.parent + / "language_packs" / "data" / "en_core_math_v1" + ) + fake_pack = tmp_path / "en_core_math_v1" + shutil.copytree(real_pack, fake_pack) + + # Corrupt the manifest checksum. + manifest_path = fake_pack / "manifest.json" + manifest = json.loads(manifest_path.read_text()) + manifest["checksum"] = "0" * 64 + manifest_path.write_text(json.dumps(manifest)) + + with pytest.raises(LexiconLoadError, match="checksum"): + load_lexicon(fake_pack) + + +# --------------------------------------------------------------------------- +# Lookups +# --------------------------------------------------------------------------- + + +class TestLookups: + @pytest.mark.parametrize("lemma", ["buy", "earn", "get", "collect", "save"]) + def test_accumulation_verb_lemmas(self, lemma: str) -> None: + lex = load_lexicon() + entry = lookup(lex, lemma) + assert entry is not None, f"{lemma!r} not found" + assert entry.category == "accumulation_verb", ( + f"{lemma!r}: expected accumulation_verb, got {entry.category!r}" + ) + + def test_depletion_verb_lemma(self) -> None: + lex = load_lexicon() + entry = lookup(lex, "spend") + assert entry is not None + assert entry.category == "depletion_verb" + + def test_transfer_verb_lemma(self) -> None: + lex = load_lexicon() + entry = lookup(lex, "give") + assert entry is not None + assert entry.category == "transfer_verb" + + def test_currency_unit_noun(self) -> None: + lex = load_lexicon() + entry = lookup(lex, "money") + assert entry is not None + assert entry.category == "currency_unit_noun" + + def test_entity_pronoun(self) -> None: + lex = load_lexicon() + entry = lookup(lex, "she") + assert entry is not None + assert entry.category == "entity_pronoun" + + def test_proper_noun_female(self) -> None: + lex = load_lexicon() + entry = lookup(lex, "tina") + assert entry is not None + assert entry.category == "proper_noun_entity_female" + + def test_alias_resolves_to_lemma_entry(self) -> None: + lex = load_lexicon() + # "earned" is an alias of "earn" (accumulation_verb) + entry = lookup(lex, "earned") + assert entry is not None, "'earned' alias not found" + assert entry.lemma == "earn" + assert entry.category == "accumulation_verb" + + def test_alias_earns_resolves_to_earn(self) -> None: + lex = load_lexicon() + entry = lookup(lex, "earns") + assert entry is not None + assert entry.lemma == "earn" + + def test_alias_spent_resolves_to_spend(self) -> None: + lex = load_lexicon() + entry = lookup(lex, "spent") + assert entry is not None + assert entry.lemma == "spend" + + def test_unknown_surface_returns_none(self) -> None: + lex = load_lexicon() + assert lookup(lex, "xyzzy") is None + + def test_unknown_proper_noun_returns_none(self) -> None: + lex = load_lexicon() + assert lookup(lex, "Zaphod") is None + + def test_case_insensitive_lookup(self) -> None: + lex = load_lexicon() + lower = lookup(lex, "earn") + upper = lookup(lex, "EARN") + assert lower is not None + assert upper is not None + assert lower.lemma == upper.lemma + + +# --------------------------------------------------------------------------- +# Determinism +# --------------------------------------------------------------------------- + + +class TestDeterminism: + def test_two_loads_have_equal_hash(self) -> None: + lex1 = load_lexicon() + lex2 = load_lexicon() + assert hash(lex1) == hash(lex2) + + def test_two_loads_are_equal(self) -> None: + lex1 = load_lexicon() + lex2 = load_lexicon() + assert lex1 == lex2 + + +# --------------------------------------------------------------------------- +# Cache hit +# --------------------------------------------------------------------------- + + +class TestCacheHit: + def test_same_object_returned(self) -> None: + lex1 = load_lexicon() + lex2 = load_lexicon() + assert lex1 is lex2, "Second call must return the cached object" + + def test_explicit_path_cache_hit(self, tmp_path: Path) -> None: + import shutil + real_pack = ( + Path(__file__).resolve().parent.parent + / "language_packs" / "data" / "en_core_math_v1" + ) + fake_pack = tmp_path / "en_core_math_v1" + shutil.copytree(real_pack, fake_pack) + + a = load_lexicon(fake_pack) + b = load_lexicon(fake_pack) + assert a is b + + +# --------------------------------------------------------------------------- +# Mutual exclusion +# --------------------------------------------------------------------------- + + +class TestMutualExclusion: + def test_surface_in_two_categories_raises(self, tmp_path: Path) -> None: + """A surface appearing in two different categories raises LexiconLoadError.""" + import shutil + + real_pack = ( + Path(__file__).resolve().parent.parent + / "language_packs" / "data" / "en_core_math_v1" + ) + conflict_pack = tmp_path / "en_core_math_v1" + shutil.copytree(real_pack, conflict_pack) + + # Inject a conflicting entry: "earn" already exists as accumulation_verb; + # add it again as depletion_verb in the depletion_verb source file. + conflict_entry = json.dumps({ + "lemma": "earn", + "category": "depletion_verb", + "aliases": [], + "provenance": "test-conflict", + }) + depletion_file = conflict_pack / "lexicon" / "depletion_verb.jsonl" + with open(depletion_file, "a", encoding="utf-8") as f: + f.write("\n" + conflict_entry + "\n") + + # Recompute and patch the manifest checksum to match the (unmodified) + # compiled lexicon.jsonl — the conflict is in the source files only. + compiled_path = conflict_pack / "lexicon.jsonl" + actual_sha256 = hashlib.sha256(compiled_path.read_bytes()).hexdigest() + manifest_path = conflict_pack / "manifest.json" + manifest = json.loads(manifest_path.read_text()) + manifest["checksum"] = actual_sha256 + manifest_path.write_text(json.dumps(manifest)) + + with pytest.raises(LexiconLoadError, match="[Mm]utual"): + load_lexicon(conflict_pack)