From 7aa77806f9f93f800c33e3bf06e18008be785575 Mon Sep 17 00:00:00 2001 From: Shay Date: Mon, 18 May 2026 08:15:23 -0700 Subject: [PATCH] =?UTF-8?q?feat(adr-0055):=20Phase=20A=20=E2=80=94=20teach?= =?UTF-8?q?ing=20corpus=20audit,=20supersession,=20typed=20provenance?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Lands the three load-bearing pieces of ADR-0055 Phase A so later phases (DiscoveryCandidate, TeachingChainProposal) have a safe substrate to write into. - teaching/audit.py: pure, deterministic re-parse of the reviewed corpus with same gates as the runtime loader but keeps drop reasons (invalid_json, missing_required_field:*, unsupported_intent, pack_missing_subject, pack_missing_object, superseded_by:*). - teaching/provenance.py: typed Provenance(adr_id, source, review_date, raw); legacy "reviewed" maps to "hand_authored" so current corpus reports the canonical enum without a file rewrite. - chat/teaching_grounding._corpus_index honors superseded_by — active view drops superseded entries while disk preserves history. - core teaching audit CLI subcommand (--json optional); exits 1 on any drop so CI catches silent corpus shrinkage from pack swaps. Observable behaviour unchanged: corpus is 10/10 loaded, all five core lanes green (smoke 67, cognition 121, runtime 19, teaching 17, packs 6), cognition eval metrics identical on dev / public / holdout splits. versor_condition < 1e-6 invariant untouched. Tests: tests/test_teaching_audit.py — 23 tests covering provenance parser, real-corpus determinism, every drop-reason path, supersession semantics, runtime/audit parity, read-only contract. --- chat/teaching_grounding.py | 25 +++- core/cli.py | 45 +++++- teaching/audit.py | 249 +++++++++++++++++++++++++++++++ teaching/provenance.py | 88 +++++++++++ tests/test_teaching_audit.py | 277 +++++++++++++++++++++++++++++++++++ 5 files changed, 681 insertions(+), 3 deletions(-) create mode 100644 teaching/audit.py create mode 100644 teaching/provenance.py create mode 100644 tests/test_teaching_audit.py diff --git a/chat/teaching_grounding.py b/chat/teaching_grounding.py index 1843313d..63a54137 100644 --- a/chat/teaching_grounding.py +++ b/chat/teaching_grounding.py @@ -99,11 +99,20 @@ def _corpus_index() -> dict[tuple[str, str], TeachingChain]: is reviewed memory but the runtime still verifies pack consistency on load so a pack-corpus skew cannot leak a non-pack atom into a surface. + + ADR-0055 Phase A: an entry whose ``chain_id`` appears as another + entry's ``superseded_by`` is also dropped from the active view. + Append-only history on disk is preserved; the loader derives the + active set. """ if not _CORPUS_PATH.exists(): return {} pack = _pack_index() - out: dict[tuple[str, str], TeachingChain] = {} + # First sweep: collect supersession claims. Only well-formed + # entries (parseable JSON object) can retire other entries — a + # malformed line cannot supersede a good one. + superseded_ids: set[str] = set() + parsed_lines: list[dict] = [] for line in _CORPUS_PATH.read_text(encoding="utf-8").splitlines(): line = line.strip() if not line: @@ -112,6 +121,15 @@ def _corpus_index() -> dict[tuple[str, str], TeachingChain]: entry = json.loads(line) except json.JSONDecodeError: continue + if not isinstance(entry, dict): + continue + parsed_lines.append(entry) + sup = entry.get("superseded_by") + if isinstance(sup, str) and sup.strip(): + superseded_ids.add(sup.strip()) + + out: dict[tuple[str, str], TeachingChain] = {} + for entry in parsed_lines: subject = (entry.get("subject") or "").strip().lower() intent = (entry.get("intent") or "").strip().lower() obj = (entry.get("object") or "").strip().lower() @@ -124,9 +142,12 @@ def _corpus_index() -> dict[tuple[str, str], TeachingChain]: # surface atom is pack-sourced. if subject not in pack or obj not in pack: continue + chain_id = str(entry.get("chain_id") or f"{subject}_{intent}") + if chain_id in superseded_ids: + continue try: chain = TeachingChain( - chain_id=str(entry.get("chain_id") or f"{subject}_{intent}"), + chain_id=chain_id, subject=subject, intent=intent, connective=connective, diff --git a/core/cli.py b/core/cli.py index 02635fcf..d13d5a28 100644 --- a/core/cli.py +++ b/core/cli.py @@ -23,7 +23,7 @@ _CORE_RS_DIR = _REPO_ROOT / "core-rs" _CORE_RS_MANIFEST = _CORE_RS_DIR / "Cargo.toml" DESCRIPTION = "CORE versor engine command suite." -EPILOG = "Examples:\n core chat\n core pulse \"What is truth?\"\n core pulse --no-glove --json \"Compare knowledge and wisdom\"\n core bench\n core bench --suite determinism --runs 50\n core bench --suite speedup --json\n core trace \"word beginning truth\"\n core trace --output-language grc --frame-pack grc --json \"logos\"\n core rust status\n core rust build\n core oov covenant\n core pack list\n core pack verify en_minimal_v1\n core test --suite fast -q\n core test --suite pulse -q\n core test --suite proof -q\n core test --suite cognition -q\n core test -- tests/test_alignment_graph.py -q\n core demo audit-tour\n core demo pack-measurements\n core demo long-context-comparison\n core eval --list\n core eval cognition\n core eval cognition --json --save\n core eval cognition --split dev --version v1\n core eval cognition --split holdout" +EPILOG = "Examples:\n core chat\n core pulse \"What is truth?\"\n core pulse --no-glove --json \"Compare knowledge and wisdom\"\n core bench\n core bench --suite determinism --runs 50\n core bench --suite speedup --json\n core trace \"word beginning truth\"\n core trace --output-language grc --frame-pack grc --json \"logos\"\n core rust status\n core rust build\n core oov covenant\n core pack list\n core pack verify en_minimal_v1\n core teaching audit\n core teaching audit --json\n core test --suite fast -q\n core test --suite pulse -q\n core test --suite proof -q\n core test --suite cognition -q\n core test -- tests/test_alignment_graph.py -q\n core demo audit-tour\n core demo pack-measurements\n core demo long-context-comparison\n core eval --list\n core eval cognition\n core eval cognition --json --save\n core eval cognition --split dev --version v1\n core eval cognition --split holdout" _TEST_SUITES: dict[str, tuple[str, ...]] = { "fast": ( @@ -465,6 +465,32 @@ def _safe_pack_id(pack_id: str) -> str: return pack_id +def cmd_teaching_audit(args: argparse.Namespace) -> int: + """ADR-0055 Phase A — surface load decisions on the reviewed teaching corpus. + + Re-parses the cognition-chains JSONL with the same gates as the + runtime loader, but keeps drop reasons so silent shrinkage (pack + skew, supersession, schema drift) is inspectable. Pure read. + """ + from teaching.audit import audit_corpus + + report = audit_corpus() + if args.json: + print(json.dumps(report.as_dict(), ensure_ascii=False, indent=2, sort_keys=True)) + return 0 if not report.dropped else 1 + print(f"corpus_id : {report.corpus_id}") + print(f"corpus_path : {report.corpus_path}") + print(f"lines_on_disk : {report.lines_on_disk}") + print(f"lines_loaded : {report.lines_loaded}") + if report.dropped: + print(f"\ndropped ({len(report.dropped)}):") + for d in report.dropped: + cid = d.chain_id or "" + print(f" L{d.line_no:>4} {cid:<40} {d.reason}") + return 1 + return 0 + + def cmd_pack_validate(args: argparse.Namespace) -> int: """Run executable source-pack validation gates.""" pack_id = _safe_pack_id(args.pack_id) @@ -1386,6 +1412,23 @@ def build_parser() -> argparse.ArgumentParser: ) pack_validate.set_defaults(func=cmd_pack_validate) + teaching = subparsers.add_parser( + "teaching", + help="inspect the reviewed teaching corpus", + ) + teaching_sub = teaching.add_subparsers( + dest="teaching_command", metavar="teaching-command", required=True, + ) + teaching_audit = teaching_sub.add_parser( + "audit", + help="surface load decisions and drop reasons for the cognition-chains corpus", + ) + teaching_audit.add_argument( + "--json", action="store_true", + help="emit machine-readable JSON", + ) + teaching_audit.set_defaults(func=cmd_teaching_audit) + rust = subparsers.add_parser( "rust", help="build, test, and inspect the Rust backend", diff --git a/teaching/audit.py b/teaching/audit.py new file mode 100644 index 00000000..73691029 --- /dev/null +++ b/teaching/audit.py @@ -0,0 +1,249 @@ +"""ADR-0055 Phase A — corpus audit for the reviewed teaching chains. + +Re-parses ``teaching/cognition_chains/cognition_chains_v1.jsonl`` with +the same gates as ``chat.teaching_grounding._corpus_index`` (schema, +intent whitelist, pack-consistency, supersession), but **keeps the +drop reasons** so an operator can inspect silent corpus shrinkage — +e.g. a pack swap that left a chain referencing a missing lemma, or +a supersession chain that retired older entries. + +Pure-function, non-mutating, deterministic. Output is order-stable +(input line order) so it is safe to wire into CI and diff against +prior runs. + +This module is **read-only**. It must never write to the corpus, +to the pack, or to any runtime state. Trust-boundary: the +``raw_line`` field surfaces the verbatim JSONL line; callers that +log this should route through ``core._safe_display.safe_display``. +""" + +from __future__ import annotations + +import json +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +from chat.pack_grounding import _pack_index +from chat.teaching_grounding import ( + _CORPUS_PATH, + _VALID_INTENTS, + TEACHING_CORPUS_ID, +) +from teaching.provenance import Provenance, parse_provenance + + +@dataclass(frozen=True, slots=True) +class DroppedChain: + """One JSONL line that did not enter the active corpus.""" + + line_no: int + reason: str + raw_line: str + chain_id: str | None = None + + +@dataclass(frozen=True, slots=True) +class LoadedChain: + """One JSONL line that entered the active corpus.""" + + line_no: int + chain_id: str + subject: str + intent: str + connective: str + object: str + provenance: Provenance + superseded_by: str | None + + +@dataclass(frozen=True, slots=True) +class AuditReport: + corpus_id: str + corpus_path: str + lines_on_disk: int + lines_loaded: int + loaded: tuple[LoadedChain, ...] = field(default_factory=tuple) + dropped: tuple[DroppedChain, ...] = field(default_factory=tuple) + + def as_dict(self) -> dict[str, Any]: + return { + "corpus_id": self.corpus_id, + "corpus_path": self.corpus_path, + "lines_on_disk": self.lines_on_disk, + "lines_loaded": self.lines_loaded, + "loaded": [ + { + "line_no": c.line_no, + "chain_id": c.chain_id, + "subject": c.subject, + "intent": c.intent, + "connective": c.connective, + "object": c.object, + "provenance": { + "adr_id": c.provenance.adr_id, + "source": c.provenance.source, + "review_date": c.provenance.review_date, + "raw": c.provenance.raw, + }, + "superseded_by": c.superseded_by, + } + for c in self.loaded + ], + "dropped": [ + { + "line_no": d.line_no, + "reason": d.reason, + "chain_id": d.chain_id, + "raw_line": d.raw_line, + } + for d in self.dropped + ], + } + + +def _read_entries(path: Path) -> list[tuple[int, str, dict | None]]: + """Read JSONL lines as ``(line_no, raw, parsed_or_none)`` tuples. + + Line numbers are 1-based to match editor conventions. Blank lines + are skipped silently (they were never on the disk-loaded count + either — they are not chains). + """ + out: list[tuple[int, str, dict | None]] = [] + if not path.exists(): + return out + for idx, raw in enumerate(path.read_text(encoding="utf-8").splitlines(), start=1): + line = raw.strip() + if not line: + continue + try: + entry = json.loads(line) + except json.JSONDecodeError: + out.append((idx, line, None)) + continue + out.append((idx, line, entry)) + return out + + +def audit_corpus(path: Path | None = None) -> AuditReport: + """Re-parse the teaching corpus and surface load decisions. + + Returns an ``AuditReport`` with one entry per non-empty line on + disk, classified as loaded or dropped with a deterministic reason + string. Pack consistency is checked against the same pack the + runtime loads. + + Reasons (stable strings, safe to assert against in tests): + - ``"invalid_json"`` + - ``"missing_required_field:"`` + - ``"unsupported_intent:"`` + - ``"pack_missing_subject:"`` + - ``"pack_missing_object:"`` + - ``"superseded_by:"`` + """ + corpus_path = path or _CORPUS_PATH + entries = _read_entries(corpus_path) + pack = _pack_index() + + # First sweep — collect supersession claims from entries that + # otherwise look well-formed enough to assert one. An invalid + # entry cannot supersede anything; this is intentional so a bad + # line cannot retire a good one. + superseded_ids: set[str] = set() + for _line_no, _raw, entry in entries: + if not isinstance(entry, dict): + continue + sup = entry.get("superseded_by") + if isinstance(sup, str) and sup.strip(): + # The current entry claims to supersede ``sup``. + superseded_ids.add(sup.strip()) + + loaded: list[LoadedChain] = [] + dropped: list[DroppedChain] = [] + + for line_no, raw, entry in entries: + if entry is None: + dropped.append(DroppedChain(line_no=line_no, reason="invalid_json", raw_line=raw)) + continue + + chain_id_raw = entry.get("chain_id") + chain_id_for_audit = str(chain_id_raw) if isinstance(chain_id_raw, str) else None + + subject = (entry.get("subject") or "").strip().lower() if isinstance(entry.get("subject"), str) else "" + intent = (entry.get("intent") or "").strip().lower() if isinstance(entry.get("intent"), str) else "" + obj = (entry.get("object") or "").strip().lower() if isinstance(entry.get("object"), str) else "" + connective = (entry.get("connective") or "").strip() if isinstance(entry.get("connective"), str) else "" + + if not subject: + dropped.append(DroppedChain( + line_no=line_no, reason="missing_required_field:subject", + raw_line=raw, chain_id=chain_id_for_audit, + )) + continue + if not intent: + dropped.append(DroppedChain( + line_no=line_no, reason="missing_required_field:intent", + raw_line=raw, chain_id=chain_id_for_audit, + )) + continue + if not obj: + dropped.append(DroppedChain( + line_no=line_no, reason="missing_required_field:object", + raw_line=raw, chain_id=chain_id_for_audit, + )) + continue + if not connective: + dropped.append(DroppedChain( + line_no=line_no, reason="missing_required_field:connective", + raw_line=raw, chain_id=chain_id_for_audit, + )) + continue + if intent not in _VALID_INTENTS: + dropped.append(DroppedChain( + line_no=line_no, reason=f"unsupported_intent:{intent}", + raw_line=raw, chain_id=chain_id_for_audit, + )) + continue + if subject not in pack: + dropped.append(DroppedChain( + line_no=line_no, reason=f"pack_missing_subject:{subject}", + raw_line=raw, chain_id=chain_id_for_audit, + )) + continue + if obj not in pack: + dropped.append(DroppedChain( + line_no=line_no, reason=f"pack_missing_object:{obj}", + raw_line=raw, chain_id=chain_id_for_audit, + )) + continue + + chain_id = chain_id_for_audit or f"{subject}_{intent}" + if chain_id in superseded_ids: + dropped.append(DroppedChain( + line_no=line_no, reason=f"superseded_by:{chain_id}", + raw_line=raw, chain_id=chain_id, + )) + continue + + sup_raw = entry.get("superseded_by") + sup = sup_raw.strip() if isinstance(sup_raw, str) and sup_raw.strip() else None + + loaded.append(LoadedChain( + line_no=line_no, + chain_id=chain_id, + subject=subject, + intent=intent, + connective=connective, + object=obj, + provenance=parse_provenance(entry.get("provenance")), + superseded_by=sup, + )) + + return AuditReport( + corpus_id=TEACHING_CORPUS_ID, + corpus_path=str(corpus_path), + lines_on_disk=len(entries), + lines_loaded=len(loaded), + loaded=tuple(loaded), + dropped=tuple(dropped), + ) diff --git a/teaching/provenance.py b/teaching/provenance.py new file mode 100644 index 00000000..9160db0f --- /dev/null +++ b/teaching/provenance.py @@ -0,0 +1,88 @@ +"""ADR-0055 Phase A — typed provenance parser for reviewed teaching entries. + +Today's corpus provenance is a free-text string like +``"adr-0052:reviewed:2026-05-17"``. This module parses that shape +into a typed ``Provenance`` without rewriting the JSONL on disk. +Existing entries keep their raw string; the parser is lenient — if +the string does not match the expected shape, the typed fields fall +back to ``None`` / ``"unknown"`` and the raw is preserved. + +Future entries written by ``TeachingChainProposal`` (Phase C) should +emit the canonical shape directly. + +Source enum follows ADR-0055 §A3: + + - ``"hand_authored"`` — written by an operator in an ADR PR + - ``"discovery_promoted"`` — accepted ``TeachingChainProposal`` (Phase C+) + - ``"imported"`` — bulk-imported from an external reviewed source + +Anything else maps to ``"unknown"``. The legacy ``"reviewed"`` token +seen in existing entries is treated as ``"hand_authored"`` so today's +corpus reports the right enum without a file rewrite. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any + + +_KNOWN_SOURCES: frozenset[str] = frozenset({ + "hand_authored", + "discovery_promoted", + "imported", +}) + +# Legacy free-text tokens that have appeared in the corpus prior to +# Phase A. Each maps to its canonical typed enum value. +_LEGACY_SOURCE_ALIASES: dict[str, str] = { + "reviewed": "hand_authored", +} + + +@dataclass(frozen=True, slots=True) +class Provenance: + adr_id: str | None + source: str + review_date: str | None + raw: str + + +def _coerce_source(token: str) -> str: + token = token.strip().lower() + if token in _KNOWN_SOURCES: + return token + if token in _LEGACY_SOURCE_ALIASES: + return _LEGACY_SOURCE_ALIASES[token] + return "unknown" + + +def parse_provenance(value: Any) -> Provenance: + """Parse a free-text provenance string into the typed shape. + + Recognised shape: ``"::"`` + where each segment is non-empty. Extra trailing segments are + folded into ``review_date`` so ``"adr-0052:reviewed:2026-05-17"`` + parses cleanly even if a future writer adds a fourth field by + accident (defensive — the parser does not crash on drift). + + Non-string or empty input produces a ``Provenance`` with all + typed fields ``None`` / ``"unknown"`` and ``raw=""``. + """ + if not isinstance(value, str): + return Provenance(adr_id=None, source="unknown", review_date=None, raw="") + raw = value.strip() + if not raw: + return Provenance(adr_id=None, source="unknown", review_date=None, raw="") + parts = raw.split(":") + if len(parts) < 3: + return Provenance(adr_id=None, source="unknown", review_date=None, raw=raw) + adr_id = parts[0].strip() or None + source = _coerce_source(parts[1]) + review_date_token = ":".join(parts[2:]).strip() or None + return Provenance( + adr_id=adr_id, + source=source, + review_date=review_date_token, + raw=raw, + ) diff --git a/tests/test_teaching_audit.py b/tests/test_teaching_audit.py new file mode 100644 index 00000000..3252188b --- /dev/null +++ b/tests/test_teaching_audit.py @@ -0,0 +1,277 @@ +"""ADR-0055 Phase A — teaching corpus audit + supersession + typed provenance. + +Pinned contracts: + + - ``teaching.audit.audit_corpus`` is pure, deterministic, and never + mutates the corpus or the pack. + - The active runtime loader (``chat.teaching_grounding._corpus_index``) + and the audit module agree on which entries are dropped and why. + - ``superseded_by`` retires an earlier chain from the active view but + leaves it on disk. + - Legacy ``"reviewed"`` provenance source token maps to + ``"hand_authored"`` so the current corpus reports the typed enum + without a file rewrite. + - Pack-consistency drop is surfaced with the specific lemma name in + the reason. +""" + +from __future__ import annotations + +import json +from pathlib import Path +from unittest.mock import patch + +from teaching.audit import audit_corpus +from teaching.provenance import Provenance, parse_provenance + + +# --------------------------------------------------------------------------- +# parse_provenance — typed shape +# --------------------------------------------------------------------------- + + +def test_parse_legacy_reviewed_token_maps_to_hand_authored() -> None: + p = parse_provenance("adr-0052:reviewed:2026-05-17") + assert p.adr_id == "adr-0052" + assert p.source == "hand_authored" + assert p.review_date == "2026-05-17" + assert p.raw == "adr-0052:reviewed:2026-05-17" + + +def test_parse_canonical_hand_authored() -> None: + p = parse_provenance("adr-9999:hand_authored:2099-12-31") + assert p.source == "hand_authored" + + +def test_parse_discovery_promoted() -> None: + p = parse_provenance("adr-9999:discovery_promoted:2099-12-31") + assert p.source == "discovery_promoted" + + +def test_parse_imported() -> None: + p = parse_provenance("adr-9999:imported:2099-12-31") + assert p.source == "imported" + + +def test_parse_unknown_source_token_falls_back() -> None: + p = parse_provenance("adr-9999:gibberish:2099-12-31") + assert p.source == "unknown" + # adr_id and review_date are still captured. + assert p.adr_id == "adr-9999" + assert p.review_date == "2099-12-31" + + +def test_parse_non_string_input_safe() -> None: + assert parse_provenance(None).source == "unknown" + assert parse_provenance(42).source == "unknown" + assert parse_provenance({"adr": "x"}).source == "unknown" + + +def test_parse_empty_string_safe() -> None: + p = parse_provenance("") + assert p == Provenance(adr_id=None, source="unknown", review_date=None, raw="") + + +def test_parse_short_string_no_crash() -> None: + p = parse_provenance("adr-0052") + assert p.source == "unknown" + assert p.raw == "adr-0052" + + +def test_parse_extra_trailing_colons_folded_into_date() -> None: + p = parse_provenance("adr-9999:hand_authored:2099-12-31:extra") + # Folding into review_date keeps drift safe — no crash, no silent loss. + assert p.review_date == "2099-12-31:extra" + + +# --------------------------------------------------------------------------- +# audit_corpus — real corpus, no mutations +# --------------------------------------------------------------------------- + + +def test_audit_real_corpus_runs_clean() -> None: + report = audit_corpus() + assert report.lines_on_disk == report.lines_loaded + assert report.dropped == () + assert len(report.loaded) >= 10 + assert report.corpus_id == "cognition_chains_v1" + + +def test_audit_loaded_entries_have_typed_provenance() -> None: + report = audit_corpus() + for entry in report.loaded: + assert entry.provenance.source in { + "hand_authored", "discovery_promoted", "imported", "unknown", + } + + +def test_audit_is_deterministic() -> None: + a = audit_corpus() + b = audit_corpus() + assert a.as_dict() == b.as_dict() + + +def test_audit_as_dict_is_json_serialisable() -> None: + report = audit_corpus() + blob = json.dumps(report.as_dict(), sort_keys=True) + assert "cognition_chains_v1" in blob + + +# --------------------------------------------------------------------------- +# audit_corpus — synthetic corpora exercising drop reasons +# --------------------------------------------------------------------------- + + +def _write_corpus(tmp_path: Path, lines: list[str]) -> Path: + p = tmp_path / "cognition_chains_v1.jsonl" + p.write_text("\n".join(lines) + "\n", encoding="utf-8") + return p + + +def test_audit_surfaces_invalid_json(tmp_path: Path) -> None: + path = _write_corpus(tmp_path, ["not json at all"]) + report = audit_corpus(path) + assert report.lines_on_disk == 1 + assert report.lines_loaded == 0 + assert len(report.dropped) == 1 + assert report.dropped[0].reason == "invalid_json" + + +def test_audit_surfaces_unsupported_intent(tmp_path: Path) -> None: + path = _write_corpus(tmp_path, [json.dumps({ + "chain_id": "x", "subject": "light", "intent": "definition", + "connective": "is", "object": "truth", + "provenance": "adr-test:hand_authored:2026-05-18", + })]) + report = audit_corpus(path) + assert len(report.dropped) == 1 + assert report.dropped[0].reason == "unsupported_intent:definition" + + +def test_audit_surfaces_pack_missing_subject(tmp_path: Path) -> None: + path = _write_corpus(tmp_path, [json.dumps({ + "chain_id": "x", "subject": "zzznotalemma", "intent": "cause", + "connective": "reveals", "object": "truth", + "provenance": "adr-test:hand_authored:2026-05-18", + })]) + report = audit_corpus(path) + assert len(report.dropped) == 1 + assert report.dropped[0].reason == "pack_missing_subject:zzznotalemma" + + +def test_audit_surfaces_pack_missing_object(tmp_path: Path) -> None: + path = _write_corpus(tmp_path, [json.dumps({ + "chain_id": "x", "subject": "light", "intent": "cause", + "connective": "reveals", "object": "zzznotalemma", + "provenance": "adr-test:hand_authored:2026-05-18", + })]) + report = audit_corpus(path) + assert len(report.dropped) == 1 + assert report.dropped[0].reason == "pack_missing_object:zzznotalemma" + + +def test_audit_surfaces_missing_required_field(tmp_path: Path) -> None: + path = _write_corpus(tmp_path, [json.dumps({ + "chain_id": "x", "subject": "", "intent": "cause", + "connective": "reveals", "object": "truth", + })]) + report = audit_corpus(path) + assert len(report.dropped) == 1 + assert report.dropped[0].reason == "missing_required_field:subject" + + +# --------------------------------------------------------------------------- +# Supersession — disk preserves history, active view drops superseded +# --------------------------------------------------------------------------- + + +def test_supersession_drops_earlier_chain_from_active_view(tmp_path: Path) -> None: + older = { + "chain_id": "cause_light_reveals_truth", + "subject": "light", "intent": "cause", "connective": "reveals", + "object": "truth", "domains_subject_k": 2, "domains_object_k": 1, + "provenance": "adr-test:hand_authored:2026-05-18", + } + newer = { + "chain_id": "cause_light_grounds_truth", + "subject": "light", "intent": "cause", "connective": "grounds", + "object": "truth", "domains_subject_k": 2, "domains_object_k": 1, + "provenance": "adr-test:hand_authored:2026-05-19", + "superseded_by": "cause_light_reveals_truth", + } + path = _write_corpus(tmp_path, [json.dumps(older), json.dumps(newer)]) + report = audit_corpus(path) + assert report.lines_on_disk == 2 + assert report.lines_loaded == 1 + # The newer entry retires the older one. + assert report.dropped[0].chain_id == "cause_light_reveals_truth" + assert report.dropped[0].reason == "superseded_by:cause_light_reveals_truth" + # Newer entry is active. + assert report.loaded[0].chain_id == "cause_light_grounds_truth" + + +def test_default_superseded_by_is_null_in_loaded_entries() -> None: + """Existing corpus uses default null — must round-trip unchanged.""" + report = audit_corpus() + assert all(entry.superseded_by is None for entry in report.loaded) + + +# --------------------------------------------------------------------------- +# Runtime parity — runtime loader and audit agree +# --------------------------------------------------------------------------- + + +def test_runtime_loader_and_audit_agree_on_active_chain_ids() -> None: + """Whatever audit_corpus says is loaded must also be what the + runtime loader has indexed (modulo the keying — runtime keys by + (subject, intent); audit lists chain_ids).""" + from chat.teaching_grounding import _corpus_index + + _corpus_index.cache_clear() + runtime_chains = {c.chain_id for c in _corpus_index().values()} + audit_chains = {c.chain_id for c in audit_corpus().loaded} + assert runtime_chains == audit_chains + + +def test_runtime_loader_honors_superseded_by(tmp_path: Path) -> None: + """If a corpus on disk has supersession, the runtime loader's + active set must match the audit report.""" + older = { + "chain_id": "cause_light_reveals_truth", + "subject": "light", "intent": "cause", "connective": "reveals", + "object": "truth", + "provenance": "adr-test:hand_authored:2026-05-18", + } + newer = { + "chain_id": "cause_light_grounds_truth", + "subject": "light", "intent": "cause", "connective": "grounds", + "object": "truth", + "provenance": "adr-test:hand_authored:2026-05-19", + "superseded_by": "cause_light_reveals_truth", + } + path = _write_corpus(tmp_path, [json.dumps(older), json.dumps(newer)]) + + from chat import teaching_grounding as tg + + with patch.object(tg, "_CORPUS_PATH", path): + tg._corpus_index.cache_clear() + try: + index = tg._corpus_index() + active = {c.chain_id for c in index.values()} + assert active == {"cause_light_grounds_truth"} + finally: + tg._corpus_index.cache_clear() + + +# --------------------------------------------------------------------------- +# Audit is read-only (trust boundary) +# --------------------------------------------------------------------------- + + +def test_audit_does_not_mutate_corpus_on_disk() -> None: + from chat.teaching_grounding import _CORPUS_PATH + + before = _CORPUS_PATH.read_bytes() + audit_corpus() + after = _CORPUS_PATH.read_bytes() + assert before == after