core-labs/core
1
0
Fork 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

feat(adr-0055): Phase A — teaching corpus audit, supersession, typed provenance

Browse source 
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.
This commit is contained in:
Shay 2026-05-18 08:15:23 -07:00
parent 0f797e2940
commit 7aa77806f9
5 changed files with 681 additions and 3 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

25
chat/teaching_grounding.py
View file

@ -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,

45
core/cli.py
View file

@ -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 "<unknown>"
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",

249
teaching/audit.py Normal file
View file

@ -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:<field>"``
- ``"unsupported_intent:<intent>"``
- ``"pack_missing_subject:<lemma>"``
- ``"pack_missing_object:<lemma>"``
- ``"superseded_by:<chain_id>"``
"""
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),
)

88
teaching/provenance.py Normal file
View file

@ -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: ``"<adr_id>:<source>:<review_date>"``
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,
)

277
tests/test_teaching_audit.py Normal file
View file

@ -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