diff --git a/core/cli.py b/core/cli.py index f2a854ff..2ef88a4d 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 teaching audit\n core teaching audit --json\n core teaching propose \n core teaching proposals --state pending\n core teaching review --accept --review-date 2026-05-18\n core teaching supersede cause_light_reveals_truth --subject light --intent cause --connective grounds --object truth --review-date 2026-05-18\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 teaching propose \n core teaching proposals --state pending\n core teaching review --accept --review-date 2026-05-18\n core teaching supersede cause_light_reveals_truth --subject light --intent cause --connective grounds --object truth --review-date 2026-05-18\n core teaching supersessions\n core teaching supersessions --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": ( @@ -622,6 +622,54 @@ def cmd_teaching_review(args: argparse.Namespace) -> int: return 0 +def cmd_teaching_supersessions(args: argparse.Namespace) -> int: + """Pair each retired chain with its active replacement. + + Derived view over ``teaching.audit.audit_corpus`` — pure, read-only. + Surfaces orphan supersessions (retired chain with no live replacement + carrying the matching ``superseded_by``) so silent corpus drift is + inspectable. + """ + from teaching.audit import audit_corpus, supersession_history + + report = audit_corpus() + records = supersession_history(report) + + if args.json: + print(json.dumps( + { + "corpus_id": report.corpus_id, + "corpus_path": report.corpus_path, + "supersessions": [r.as_dict() for r in records], + }, + ensure_ascii=False, indent=2, sort_keys=True, + )) + return 0 + + if not records: + print("(no supersessions)") + return 0 + + has_orphan = False + for r in records: + if r.replacement is None: + has_orphan = True + print( + f"retired: {r.retired_chain_id} (line {r.retired_line_no})\n" + f" replaced_by: " + ) + continue + rep = r.replacement + prov = rep.provenance.raw or "(unknown)" + print( + f"retired: {r.retired_chain_id} (line {r.retired_line_no})\n" + f" replaced_by: {rep.chain_id} (line {rep.line_no})\n" + f" {rep.subject} {rep.connective} {rep.object} [{rep.intent}]\n" + f" provenance: {prov}" + ) + return 1 if has_orphan else 0 + + def cmd_teaching_supersede(args: argparse.Namespace) -> int: """ADR-0057 follow-up — retire an active corpus chain by appending a new chain marked ``superseded_by``. @@ -1671,6 +1719,15 @@ def build_parser() -> argparse.ArgumentParser: ) teaching_supersede.set_defaults(func=cmd_teaching_supersede) + teaching_supersessions = teaching_sub.add_parser( + "supersessions", + help="pair each retired chain with its active replacement (derived view)", + ) + teaching_supersessions.add_argument( + "--json", action="store_true", help="emit machine-readable JSON", + ) + teaching_supersessions.set_defaults(func=cmd_teaching_supersessions) + rust = subparsers.add_parser( "rust", help="build, test, and inspect the Rust backend", diff --git a/teaching/audit.py b/teaching/audit.py index 73691029..97e3e853 100644 --- a/teaching/audit.py +++ b/teaching/audit.py @@ -247,3 +247,67 @@ def audit_corpus(path: Path | None = None) -> AuditReport: loaded=tuple(loaded), dropped=tuple(dropped), ) + + +@dataclass(frozen=True, slots=True) +class SupersessionRecord: + """One retired→replacement pair derived from an ``AuditReport``. + + ``replacement`` is ``None`` when the corpus claims a chain was + superseded but no currently-active entry carries the matching + ``superseded_by`` field — i.e. an orphan supersession. This + surfaces silent corpus drift to the operator. + """ + + retired_chain_id: str + retired_line_no: int + replacement: LoadedChain | None + + def as_dict(self) -> dict[str, Any]: + return { + "retired_chain_id": self.retired_chain_id, + "retired_line_no": self.retired_line_no, + "replacement": ( + { + "chain_id": self.replacement.chain_id, + "line_no": self.replacement.line_no, + "subject": self.replacement.subject, + "intent": self.replacement.intent, + "connective": self.replacement.connective, + "object": self.replacement.object, + "provenance": { + "adr_id": self.replacement.provenance.adr_id, + "source": self.replacement.provenance.source, + "review_date": self.replacement.provenance.review_date, + "raw": self.replacement.provenance.raw, + }, + } + if self.replacement is not None + else None + ), + } + + +def supersession_history(report: AuditReport) -> tuple[SupersessionRecord, ...]: + """Derive retired→replacement pairs from an audit report. + + Pure function of the report; deterministic, ordered by the + retired entry's line number (disk order, oldest first). + """ + by_supersedes: dict[str, LoadedChain] = { + entry.superseded_by: entry + for entry in report.loaded + if entry.superseded_by + } + records: list[SupersessionRecord] = [] + for dropped in report.dropped: + if not dropped.reason.startswith("superseded_by:"): + continue + retired_id = dropped.chain_id or dropped.reason.split(":", 1)[1] + records.append(SupersessionRecord( + retired_chain_id=retired_id, + retired_line_no=dropped.line_no, + replacement=by_supersedes.get(retired_id), + )) + records.sort(key=lambda r: r.retired_line_no) + return tuple(records) diff --git a/tests/test_supersession_history.py b/tests/test_supersession_history.py new file mode 100644 index 00000000..90951b09 --- /dev/null +++ b/tests/test_supersession_history.py @@ -0,0 +1,190 @@ +"""Supersession history view over the audit report. + +Pins: + + - Pure derived view; no corpus mutation. + - Deterministic order (retired-line ascending). + - Each retired→replacement pair is surfaced; orphans + (retired with no matching live ``superseded_by``) are flagged + with ``replacement=None`` instead of being silently dropped. + - Chained supersessions (A retired by B, B retired by C) surface + both pairs. +""" + +from __future__ import annotations + +import json +from pathlib import Path + +from teaching.audit import audit_corpus, supersession_history +from teaching.supersede import supersede_chain + + +def _seed(tmp_path: Path) -> Path: + p = tmp_path / "cognition_chains_v1.jsonl" + lines = [ + json.dumps({ + "chain_id": "cause_light_reveals_truth", + "subject": "light", "intent": "cause", "connective": "reveals", + "object": "truth", + "provenance": "adr-test:hand_authored:2026-05-18", + }), + json.dumps({ + "chain_id": "verification_memory_requires_recall", + "subject": "memory", "intent": "verification", "connective": "requires", + "object": "recall", + "provenance": "adr-test:hand_authored:2026-05-18", + }), + ] + p.write_text("\n".join(lines) + "\n", encoding="utf-8") + return p + + +def test_empty_history_on_clean_corpus(tmp_path: Path) -> None: + corpus = _seed(tmp_path) + report = audit_corpus(corpus) + assert supersession_history(report) == () + + +def test_history_pairs_retired_with_replacement(tmp_path: Path) -> None: + corpus = _seed(tmp_path) + supersede_chain( + old_chain_id="cause_light_reveals_truth", + subject="light", intent="cause", + connective="grounds", object_="truth", + review_date="2026-05-18", + corpus_path=corpus, + ) + report = audit_corpus(corpus) + records = supersession_history(report) + assert len(records) == 1 + r = records[0] + assert r.retired_chain_id == "cause_light_reveals_truth" + assert r.replacement is not None + assert r.replacement.chain_id == "cause_light_grounds_truth" + assert r.replacement.connective == "grounds" + assert "supersede(cause_light_reveals_truth)" in r.replacement.provenance.raw + + +def test_chained_supersession_surfaces_both_pairs(tmp_path: Path) -> None: + corpus = _seed(tmp_path) + supersede_chain( + old_chain_id="cause_light_reveals_truth", + subject="light", intent="cause", + connective="grounds", object_="truth", + review_date="2026-05-18", + corpus_path=corpus, + ) + supersede_chain( + old_chain_id="cause_light_grounds_truth", + subject="light", intent="cause", + connective="orders", object_="truth", + review_date="2026-05-19", + corpus_path=corpus, + ) + report = audit_corpus(corpus) + records = supersession_history(report) + retired_ids = [r.retired_chain_id for r in records] + assert "cause_light_reveals_truth" in retired_ids + assert "cause_light_grounds_truth" in retired_ids + # Latest replacement is the only thing still active for (light, cause). + by_retired = {r.retired_chain_id: r for r in records} + final = by_retired["cause_light_grounds_truth"].replacement + assert final is not None + assert final.chain_id == "cause_light_orders_truth" + + +def test_records_sorted_by_retired_line_no(tmp_path: Path) -> None: + corpus = _seed(tmp_path) + supersede_chain( + old_chain_id="cause_light_reveals_truth", + subject="light", intent="cause", + connective="grounds", object_="truth", + review_date="2026-05-18", + corpus_path=corpus, + ) + supersede_chain( + old_chain_id="verification_memory_requires_recall", + subject="memory", intent="verification", + connective="grounds", object_="recall", + review_date="2026-05-19", + corpus_path=corpus, + ) + report = audit_corpus(corpus) + records = supersession_history(report) + line_nos = [r.retired_line_no for r in records] + assert line_nos == sorted(line_nos) + + +def test_orphan_supersession_surfaces_replacement_none(tmp_path: Path) -> None: + """A corpus where a chain claims supersession but the retiring entry + itself is later retired by something with a different ``superseded_by``.""" + p = tmp_path / "cognition_chains_v1.jsonl" + # B retires A. C retires B (so B drops too). C carries + # superseded_by=B only — A is now an orphan: dropped (because B + # said so) but no live entry carries superseded_by=A. + lines = [ + json.dumps({ + "chain_id": "cause_light_reveals_truth", + "subject": "light", "intent": "cause", "connective": "reveals", + "object": "truth", + "provenance": "adr-test:hand_authored:2026-05-18", + }), + json.dumps({ + "chain_id": "cause_light_grounds_truth", + "subject": "light", "intent": "cause", "connective": "grounds", + "object": "truth", + "provenance": "adr-test:hand_authored:2026-05-18", + "superseded_by": "cause_light_reveals_truth", + }), + json.dumps({ + "chain_id": "cause_light_orders_truth", + "subject": "light", "intent": "cause", "connective": "orders", + "object": "truth", + "provenance": "adr-test:hand_authored:2026-05-19", + "superseded_by": "cause_light_grounds_truth", + }), + ] + p.write_text("\n".join(lines) + "\n", encoding="utf-8") + report = audit_corpus(p) + records = supersession_history(report) + by_retired = {r.retired_chain_id: r for r in records} + # A is dropped (B's superseded_by points at it), but no LIVE entry + # carries superseded_by=A — B itself was retired. Orphan. + assert by_retired["cause_light_reveals_truth"].replacement is None + # B has a live replacement: C. + rep_b = by_retired["cause_light_grounds_truth"].replacement + assert rep_b is not None + assert rep_b.chain_id == "cause_light_orders_truth" + + +def test_as_dict_round_trips_through_json(tmp_path: Path) -> None: + corpus = _seed(tmp_path) + supersede_chain( + old_chain_id="cause_light_reveals_truth", + subject="light", intent="cause", + connective="grounds", object_="truth", + review_date="2026-05-18", + corpus_path=corpus, + ) + report = audit_corpus(corpus) + records = supersession_history(report) + blob = json.dumps([r.as_dict() for r in records], sort_keys=True) + assert "cause_light_reveals_truth" in blob + assert "cause_light_grounds_truth" in blob + + +def test_pure_function_no_corpus_mutation(tmp_path: Path) -> None: + corpus = _seed(tmp_path) + supersede_chain( + old_chain_id="cause_light_reveals_truth", + subject="light", intent="cause", + connective="grounds", object_="truth", + review_date="2026-05-18", + corpus_path=corpus, + ) + bytes_before = corpus.read_bytes() + report = audit_corpus(corpus) + supersession_history(report) + supersession_history(report) + assert corpus.read_bytes() == bytes_before