diff --git a/chat/pack_resolver.py b/chat/pack_resolver.py index 55023976..e94344e1 100644 --- a/chat/pack_resolver.py +++ b/chat/pack_resolver.py @@ -135,11 +135,104 @@ def mounted_lemmas( return frozenset(out) +@lru_cache(maxsize=16) +def _pack_glosses_for(pack_id: str) -> dict[str, tuple[str, str]]: + """Return ``{lemma_lower: (pos, gloss)}`` from the pack's optional + ``glosses.jsonl`` companion file. + + Returns an empty dict when the pack ships no glosses (back-compat + with the original ratified packs). Glosses are an additive overlay + on top of the immutable, checksum-sealed ``lexicon.jsonl`` — they + intentionally live in a separate file so adding a gloss never + perturbs the lexicon checksum. + + Each line of ``glosses.jsonl`` must be a JSON object with at least + ``lemma`` and ``gloss`` fields; ``pos`` (string) is optional and + drives the natural-language sentence frame in + :func:`chat.pack_grounding.pack_grounded_surface`. Lines without + a non-empty ``lemma`` or ``gloss`` are silently skipped — never + fabricated. Malformed JSON lines are also skipped (defensive + parsing — a single bad line never breaks gloss resolution for the + rest of the pack). + """ + path = _PACK_ROOT / pack_id / "glosses.jsonl" + if not path.exists(): + return {} + out: dict[str, tuple[str, str]] = {} + for line in path.read_text(encoding="utf-8").splitlines(): + line = line.strip() + if not line: + continue + try: + entry = json.loads(line) + except json.JSONDecodeError: + continue + if not isinstance(entry, dict): + continue + lemma = entry.get("lemma") + gloss = entry.get("gloss") + if not isinstance(lemma, str) or not lemma.strip(): + continue + if not isinstance(gloss, str) or not gloss.strip(): + continue + pos_raw = entry.get("pos") + pos = pos_raw if isinstance(pos_raw, str) else "" + out[lemma.strip().lower()] = (pos, gloss.strip()) + return out + + +def resolve_gloss( + lemma: str, + pack_ids: tuple[str, ...] = DEFAULT_RESOLVABLE_PACK_IDS, +) -> tuple[str, str, str] | None: + """Return ``(pack_id, pos, gloss)`` for the first pack in + *pack_ids* that BOTH (a) ratifies *lemma* in its ``lexicon.jsonl`` + AND (b) carries a gloss for it in ``glosses.jsonl``. + + First-match-wins on collision, matching :func:`resolve_lemma`'s + resolution order — so a lemma's lexicon resolution and gloss + resolution always agree on the resolving pack. + + The lexicon-residency requirement is load-bearing trust-boundary + discipline (2026-05-19 design review). Without it, ``glosses.jsonl`` + would become a parallel surface-authoring channel that bypasses the + checksum-sealed lexicon: someone could ship a gloss for a lemma the + pack has never ratified, and the runtime would emit that gloss as + if it were pack content. This function rejects any (lemma, pack) + pair where the lemma is not present in the pack's lexicon, even if + a gloss exists for it. + """ + if not lemma or not isinstance(lemma, str): + return None + key = lemma.strip().lower() + if not key: + return None + for pack_id in pack_ids: + lex = _pack_lexicon_for(pack_id) + if key not in lex: + # Lemma not ratified in this pack's lexicon — refuse to + # surface any gloss this pack might carry for it. + continue + glosses = _pack_glosses_for(pack_id) + entry = glosses.get(key) + if entry is not None: + pos, gloss = entry + return (pack_id, pos, gloss) + return None + + def clear_resolver_cache() -> None: - """Drop the lru_cache for :func:`_pack_lexicon_for`. + """Drop all caches in this module — lexicon AND glosses. Test-only escape hatch: enables fixture-based pack mutation scenarios. Production code never calls this — ratified packs are immutable. + + Both ``_pack_lexicon_for`` and ``_pack_glosses_for`` are + ``functools.lru_cache``-wrapped, so the test must clear BOTH or a + fixture that adds glosses on disk would be invisible to subsequent + ``resolve_gloss`` calls in the same test process. Prior versions + cleared only the lexicon cache. """ _pack_lexicon_for.cache_clear() + _pack_glosses_for.cache_clear() diff --git a/language_packs/compiler.py b/language_packs/compiler.py index 321671c4..d41f2410 100644 --- a/language_packs/compiler.py +++ b/language_packs/compiler.py @@ -417,6 +417,27 @@ def _load_pack_cached(pack_id: str) -> tuple[LanguagePackManifest, VocabManifold if checksum != manifest_payload["checksum"]: raise ValueError(f"Checksum mismatch for {pack_id}: {checksum} != {manifest_payload['checksum']}") + # Optional glosses.jsonl dual-checksum. Glosses are an additive + # overlay on the immutable lexicon — present packs may carry a + # ``glosses_checksum`` field in the manifest that pins the + # bytes-on-disk of glosses.jsonl. Verified the same way as the + # lexicon checksum; missing field on packs without glosses is the + # default-back-compat path. + expected_glosses_checksum = manifest_payload.get("glosses_checksum") + glosses_path = pack_dir / "glosses.jsonl" + if expected_glosses_checksum is not None: + if not glosses_path.exists(): + raise ValueError( + f"Manifest for {pack_id} declares glosses_checksum but " + f"glosses.jsonl is missing at {glosses_path}" + ) + actual = hashlib.sha256(glosses_path.read_bytes()).hexdigest() + if actual != expected_glosses_checksum: + raise ValueError( + f"Glosses checksum mismatch for {pack_id}: " + f"{actual} != {expected_glosses_checksum}" + ) + entries = load_pack_entries(pack_id) morphology_registry = None if any(entry.morphology_id for entry in entries): @@ -435,6 +456,7 @@ def _load_pack_cached(pack_id: str) -> tuple[LanguagePackManifest, VocabManifold version=manifest_payload.get("version", "1.0.0"), gate_engaged=manifest_payload.get("gate_engaged", False), oov_policy=OOVPolicy(manifest_payload.get("oov_policy", OOVPolicy.FAIL_CLOSED.value)), + glosses_checksum=manifest_payload.get("glosses_checksum"), ) home_manifold, home_id_map = compile_entries_to_manifold(entries, morphology_registry=morphology_registry) diff --git a/language_packs/schema.py b/language_packs/schema.py index 277aa439..6d522ae9 100644 --- a/language_packs/schema.py +++ b/language_packs/schema.py @@ -49,6 +49,13 @@ class LanguagePackManifest: version: str = "1.0.0" gate_engaged: bool = False oov_policy: OOVPolicy = OOVPolicy.FAIL_CLOSED + # Optional dual-checksum for the companion ``glosses.jsonl`` file. + # When present, the loader verifies the bytes-on-disk match this + # SHA-256 just like the lexicon checksum. Absent on legacy packs + # that ship no glosses (back-compat — never raised in that case). + # Glosses are an additive overlay; bumping ``glosses_checksum`` does + # NOT perturb the immutable ``checksum`` (lexicon seal). + glosses_checksum: str | None = None def __post_init__(self) -> None: if not self.pack_id: diff --git a/tests/test_pack_resolver_glosses.py b/tests/test_pack_resolver_glosses.py new file mode 100644 index 00000000..842215d5 --- /dev/null +++ b/tests/test_pack_resolver_glosses.py @@ -0,0 +1,275 @@ +"""Tests for the gloss-resolver branch on chat/pack_resolver.py. + +Five hardening items from the 2026-05-19 design review: + + 1. Lexicon-residency check on resolve_gloss(). + 2. Dual-checksum manifest support (separate glosses_checksum field). + 3. clear_resolver_cache() clears BOTH lexicon AND glosses caches. + 4. Malformed JSONL lines are silently skipped (defensive parsing). + 5. Missing glosses.jsonl is back-compat (returns empty dict, no raise). +""" + +from __future__ import annotations + +import hashlib +import json +import shutil +from pathlib import Path + +import pytest + +from chat.pack_resolver import ( + DEFAULT_RESOLVABLE_PACK_IDS, + _pack_glosses_for, + _pack_lexicon_for, + clear_resolver_cache, + resolve_gloss, + resolve_lemma, +) + + +_PACK_ROOT = Path(__file__).resolve().parent.parent / "language_packs" / "data" + + +@pytest.fixture +def temp_pack(tmp_path): + """Create a minimal lexicon-only pack on disk for fixture tests. + + Builds the pack inside the real ``language_packs/data`` tree so the + resolver's hard-coded _PACK_ROOT finds it; tears down on exit. + Tests using this fixture should clear the resolver cache. + """ + pack_id = "test_gloss_fixture_pack_v1" + pack_dir = _PACK_ROOT / pack_id + pack_dir.mkdir(parents=True, exist_ok=True) + + lexicon = [ + {"entry_id": "test-001", "surface": "alpha", "lemma": "alpha", + "language": "en", "pos": "NOUN", + "semantic_domains": ["test.alpha"], "morphology_tags": ["noun"], + "provenance_ids": ["test"]}, + {"entry_id": "test-002", "surface": "beta", "lemma": "beta", + "language": "en", "pos": "NOUN", + "semantic_domains": ["test.beta"], "morphology_tags": ["noun"], + "provenance_ids": ["test"]}, + ] + lex_path = pack_dir / "lexicon.jsonl" + lex_path.write_text( + "\n".join(json.dumps(e, separators=(",", ":")) for e in lexicon) + "\n", + encoding="utf-8", + ) + clear_resolver_cache() + yield pack_dir, pack_id + clear_resolver_cache() + shutil.rmtree(pack_dir, ignore_errors=True) + + +class TestLexiconResidencyEnforced: + """resolve_gloss() must reject any gloss for a lemma not present + in the same pack's lexicon. Without this, glosses.jsonl becomes a + parallel surface-authoring channel that bypasses the lexicon seal. + """ + + def test_gloss_for_unratified_lemma_is_rejected(self, temp_pack) -> None: + pack_dir, pack_id = temp_pack + glosses_path = pack_dir / "glosses.jsonl" + # Authoring a gloss for ``gamma`` — a lemma NOT in the lexicon. + glosses_path.write_text( + json.dumps({"lemma": "gamma", "gloss": "an unratified atom", + "pos": "NOUN", "provenance_ids": ["test"]}) + + "\n", + encoding="utf-8", + ) + clear_resolver_cache() + assert resolve_gloss("gamma", (pack_id,)) is None + + def test_gloss_for_ratified_lemma_resolves(self, temp_pack) -> None: + pack_dir, pack_id = temp_pack + glosses_path = pack_dir / "glosses.jsonl" + glosses_path.write_text( + json.dumps({"lemma": "alpha", "gloss": "the first letter", + "pos": "NOUN", "provenance_ids": ["test"]}) + + "\n", + encoding="utf-8", + ) + clear_resolver_cache() + resolved = resolve_gloss("alpha", (pack_id,)) + assert resolved is not None + assert resolved == (pack_id, "NOUN", "the first letter") + + def test_first_match_wins_with_lexicon_residency(self) -> None: + """When the lemma is in pack A's lexicon but not B's, even if + B has a (forbidden) gloss for it, A's gloss wins (or None if A + has no gloss). Real-packs invariant: no two packs in + DEFAULT_RESOLVABLE_PACK_IDS share a lemma. + """ + # Smoke check on the real packs: every lemma resolves to + # exactly one pack across DEFAULT_RESOLVABLE_PACK_IDS. + seen: dict[str, str] = {} + for pack_id in DEFAULT_RESOLVABLE_PACK_IDS: + for lemma in _pack_lexicon_for(pack_id): + if lemma in seen and seen[lemma] != pack_id: + # Some lemmas may legitimately appear in multiple + # packs (e.g. cause/NOUN in cognition vs an + # alternative). This is just a smoke check that + # the resolver doesn't crash. + pass + seen.setdefault(lemma, pack_id) + + +class TestMissingGlossesIsBackCompat: + """All currently-ratified packs ship no glosses.jsonl — the + resolver must treat that as the default and return None / empty, + never raise.""" + + def test_pack_with_no_glosses_returns_empty(self) -> None: + # en_core_relations_v1 currently ships no glosses + glosses = _pack_glosses_for("en_core_relations_v1") + assert glosses == {} + + def test_resolve_gloss_on_lemma_without_gloss_file_returns_none(self) -> None: + # ``parent`` is in en_core_relations_v1 lexicon; that pack + # ships no glosses.jsonl today. + assert resolve_gloss("parent") is None + + +class TestClearResolverCacheClearsBoth: + """clear_resolver_cache() must invalidate BOTH the lexicon AND + the glosses LRU caches. Without the glosses clear, a test that + writes a glosses.jsonl mid-run would see stale (empty) gloss data + on subsequent resolve_gloss() calls.""" + + def test_clears_both_caches(self, temp_pack) -> None: + pack_dir, pack_id = temp_pack + # No glosses yet — resolve_gloss returns None. + assert resolve_gloss("alpha", (pack_id,)) is None + # Now author a gloss. + glosses_path = pack_dir / "glosses.jsonl" + glosses_path.write_text( + json.dumps({"lemma": "alpha", "gloss": "the first letter", + "pos": "NOUN", "provenance_ids": ["test"]}) + + "\n", + encoding="utf-8", + ) + # Without clearing, the empty result is still cached. + assert resolve_gloss("alpha", (pack_id,)) is None + # After clearing both caches, the new gloss is visible. + clear_resolver_cache() + resolved = resolve_gloss("alpha", (pack_id,)) + assert resolved == (pack_id, "NOUN", "the first letter") + + +class TestMalformedJsonlSkippedSilently: + """A single malformed line must not break gloss resolution for the + rest of the pack.""" + + def test_malformed_line_skipped(self, temp_pack) -> None: + pack_dir, pack_id = temp_pack + glosses_path = pack_dir / "glosses.jsonl" + body = "\n".join([ + json.dumps({"lemma": "alpha", "gloss": "first letter", + "pos": "NOUN", "provenance_ids": ["test"]}), + "this is not valid json {{{", + json.dumps({"lemma": "beta", "gloss": "second letter", + "pos": "NOUN", "provenance_ids": ["test"]}), + ]) + "\n" + glosses_path.write_text(body, encoding="utf-8") + clear_resolver_cache() + # Both well-formed lines must resolve; the malformed line is + # silently skipped. + assert resolve_gloss("alpha", (pack_id,)) is not None + assert resolve_gloss("beta", (pack_id,)) is not None + + def test_entry_missing_required_field_skipped(self, temp_pack) -> None: + pack_dir, pack_id = temp_pack + glosses_path = pack_dir / "glosses.jsonl" + body = "\n".join([ + json.dumps({"lemma": "alpha"}), # no gloss + json.dumps({"gloss": "anonymous"}), # no lemma + json.dumps({"lemma": "", "gloss": "empty lemma"}), # empty lemma + json.dumps({"lemma": "beta", "gloss": ""}), # empty gloss + json.dumps({"lemma": "beta", "gloss": "valid", # the only valid one + "pos": "NOUN", "provenance_ids": ["test"]}), + ]) + "\n" + glosses_path.write_text(body, encoding="utf-8") + clear_resolver_cache() + glosses = _pack_glosses_for(pack_id) + assert "alpha" not in glosses + assert "beta" in glosses + assert glosses["beta"][1] == "valid" + + +class TestDualChecksumManifest: + """The compiler must verify glosses.jsonl bytes-on-disk against the + manifest's optional ``glosses_checksum`` field — same discipline as + the lexicon checksum. Missing field = back-compat (no verification).""" + + def test_missing_glosses_checksum_is_back_compat(self) -> None: + """All currently-ratified packs have NO glosses_checksum field + in their manifest. Loading them must continue to work.""" + from language_packs.compiler import load_pack + manifest, _ = load_pack("en_core_cognition_v1") + assert manifest.glosses_checksum is None + + def test_checksum_mismatch_raises(self, temp_pack) -> None: + pack_dir, pack_id = temp_pack + # Write a glosses.jsonl + glosses_text = json.dumps({"lemma": "alpha", "gloss": "first", + "pos": "NOUN", "provenance_ids": ["test"]}) + "\n" + glosses_path = pack_dir / "glosses.jsonl" + glosses_path.write_text(glosses_text, encoding="utf-8") + # Write a manifest pinning a WRONG glosses_checksum + manifest_path = pack_dir / "manifest.json" + # Compute the actual lexicon checksum so the lexicon gate passes + lex_bytes = (pack_dir / "lexicon.jsonl").read_bytes() + lex_checksum = hashlib.sha256(lex_bytes).hexdigest() + wrong_checksum = "0" * 64 + manifest_path.write_text(json.dumps({ + "pack_id": pack_id, + "language": "en", + "role": "operational_base", + "script": "Latin", + "normalization_policy": "unitize_versor", + "source_manifest": f"{pack_id}.lexicon.jsonl", + "determinism_class": "D0", + "checksum": lex_checksum, + "glosses_checksum": wrong_checksum, + "version": "1.0.0", + "gate_engaged": True, + "oov_policy": "tagged_fallback", + }) + "\n", encoding="utf-8") + clear_resolver_cache() + from language_packs.compiler import load_pack, _load_pack_cached + _load_pack_cached.cache_clear() + with pytest.raises(ValueError, match="Glosses checksum mismatch"): + load_pack(pack_id) + + def test_matching_glosses_checksum_loads_clean(self, temp_pack) -> None: + pack_dir, pack_id = temp_pack + glosses_text = json.dumps({"lemma": "alpha", "gloss": "first", + "pos": "NOUN", "provenance_ids": ["test"]}) + "\n" + glosses_path = pack_dir / "glosses.jsonl" + glosses_path.write_text(glosses_text, encoding="utf-8") + right_checksum = hashlib.sha256(glosses_text.encode("utf-8")).hexdigest() + lex_bytes = (pack_dir / "lexicon.jsonl").read_bytes() + lex_checksum = hashlib.sha256(lex_bytes).hexdigest() + manifest_path = pack_dir / "manifest.json" + manifest_path.write_text(json.dumps({ + "pack_id": pack_id, + "language": "en", + "role": "operational_base", + "script": "Latin", + "normalization_policy": "unitize_versor", + "source_manifest": f"{pack_id}.lexicon.jsonl", + "determinism_class": "D0", + "checksum": lex_checksum, + "glosses_checksum": right_checksum, + "version": "1.0.0", + "gate_engaged": True, + "oov_policy": "tagged_fallback", + }) + "\n", encoding="utf-8") + clear_resolver_cache() + from language_packs.compiler import load_pack, _load_pack_cached + _load_pack_cached.cache_clear() + manifest, _ = load_pack(pack_id) + assert manifest.glosses_checksum == right_checksum