diff --git a/core/cli.py b/core/cli.py index 12fa5166..f2a854ff 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 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 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,41 @@ def cmd_teaching_review(args: argparse.Namespace) -> int: return 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``. + + Distinct from accept-a-proposal (no replay gate; this is a direct + operator action). Validates pack-consistency / intent / completeness + before the append, and rolls back the corpus byte-identically on any + post-audit failure. + """ + from chat.teaching_grounding import _CORPUS_PATH + from teaching.supersede import SupersessionError, supersede_chain + + try: + new_chain_id = supersede_chain( + old_chain_id=args.old_chain_id, + subject=args.subject, + intent=args.intent, + connective=args.connective, + object_=args.object, + review_date=args.review_date, + corpus_path=_CORPUS_PATH, + operator_note=args.note, + new_chain_id=args.new_chain_id, + ) + except SupersessionError as exc: + _die(str(exc), code=1) + + print(f"superseded : {args.old_chain_id}") + print(f"new chain_id : {new_chain_id}") + print(f"review_date : {args.review_date}") + if args.note: + print(f"note : {args.note}") + return 0 + + def cmd_pack_validate(args: argparse.Namespace) -> int: """Run executable source-pack validation gates.""" pack_id = _safe_pack_id(args.pack_id) @@ -1614,6 +1649,28 @@ def build_parser() -> argparse.ArgumentParser: ) teaching_review.set_defaults(func=cmd_teaching_review) + teaching_supersede = teaching_sub.add_parser( + "supersede", + help="retire an active corpus chain by appending a replacement (operator action)", + ) + teaching_supersede.add_argument( + "old_chain_id", + help="chain_id currently active in the corpus that this action retires", + ) + teaching_supersede.add_argument("--subject", required=True) + teaching_supersede.add_argument("--intent", required=True) + teaching_supersede.add_argument("--connective", required=True) + teaching_supersede.add_argument("--object", required=True) + teaching_supersede.add_argument( + "--review-date", required=True, help="YYYY-MM-DD", + ) + teaching_supersede.add_argument("--note", default="", help="operator note") + teaching_supersede.add_argument( + "--new-chain-id", default=None, + help="explicit new chain_id (default: ___)", + ) + teaching_supersede.set_defaults(func=cmd_teaching_supersede) + rust = subparsers.add_parser( "rust", help="build, test, and inspect the Rust backend", diff --git a/teaching/proposals.py b/teaching/proposals.py index 77f7badc..af53c067 100644 --- a/teaching/proposals.py +++ b/teaching/proposals.py @@ -323,14 +323,21 @@ def append_chain_to_corpus( corpus_path: Path, provenance: Provenance, chain_id: str | None = None, + superseded_by: str | None = None, ) -> str: """Append one reviewed chain JSON line to the active corpus. Returns the ``chain_id`` written. Trust boundary: this is the ONLY function in the codebase that writes to the active teaching - corpus, and it is reachable only from - ``accept_proposal`` after the replay-equivalence gate and - operator review. + corpus, and it is reachable only from ``accept_proposal`` (after + the replay-equivalence gate + operator review) or from + ``teaching.supersede.supersede_chain`` (operator-driven retire + of an existing chain — see ADR-0057). + + ``superseded_by`` records the ``chain_id`` of an earlier active + entry that this new entry retires. The earlier entry stays on + disk; ``teaching.audit`` and ``chat.teaching_grounding`` both + honour the supersession at load time. """ subject = str(chain["subject"]).strip().lower() intent = str(chain["intent"]).strip().lower() @@ -338,7 +345,7 @@ def append_chain_to_corpus( obj = str(chain["object"]).strip().lower() if not chain_id: chain_id = f"{intent}_{subject}_{connective}_{obj}" - entry = { + entry: dict[str, Any] = { "chain_id": chain_id, "subject": subject, "intent": intent, @@ -351,6 +358,8 @@ def append_chain_to_corpus( f"{provenance.review_date or ''}" ), } + if superseded_by: + entry["superseded_by"] = str(superseded_by).strip() line = json.dumps(entry, sort_keys=True, separators=(",", ":")) with corpus_path.open("a", encoding="utf-8") as fh: fh.write(line + "\n") diff --git a/teaching/supersede.py b/teaching/supersede.py new file mode 100644 index 00000000..2e256453 --- /dev/null +++ b/teaching/supersede.py @@ -0,0 +1,196 @@ +"""ADR-0057 follow-up — operator-driven supersession of an active corpus chain. + +Supersession is the **second** mutation surface on the reviewed +teaching corpus (alongside ``teaching.proposals.accept_proposal``). +It is *not* a proposal: there is no replay-equivalence gate and no +``ProposalLog`` round-trip. It is a direct operator action that +records: "this active chain is replaced by this new one." + +Trust boundary: + + - Append-only at the disk level. The earlier chain stays on disk; + the audit report and the runtime loader both honour the + ``superseded_by`` field to drop it from the active view. + - The single write surface remains ``proposals.append_chain_to_corpus``. + This module composes around it; it does not write its own JSONL. + - Validation gates (pack-consistency, intent whitelist, complete + fields, no double-supersede, not self-supersede) all run before + the append. Any gate failure raises ``SupersessionError`` and + leaves the corpus byte-identical. + - No clock-time read. ``review_date`` is operator-provided. + +This is distinct from ``TeachingChainProposal``: supersession is an +operator's deliberate replacement of a hand-authored or previously +discovery-promoted chain. It does not need a replay gate because +the operator is explicitly accepting any metric movement. +""" + +from __future__ import annotations + +import re +from pathlib import Path + +from teaching.audit import audit_corpus +from teaching.proposals import append_chain_to_corpus +from teaching.provenance import Provenance + +# Reused from chat.teaching_grounding to keep one definition. +from chat.teaching_grounding import _VALID_INTENTS + +_REVIEW_DATE_RE: re.Pattern[str] = re.compile(r"^\d{4}-\d{2}-\d{2}$") + + +class SupersessionError(ValueError): + """Raised when a supersession action fails a pre-condition gate.""" + + +def _validate_review_date(value: str) -> str: + value = (value or "").strip() + if not _REVIEW_DATE_RE.match(value): + raise SupersessionError( + f"review_date must be YYYY-MM-DD; got {value!r}" + ) + return value + + +def _validate_chain_fields( + subject: str, intent: str, connective: str, obj: str +) -> tuple[str, str, str, str]: + s = (subject or "").strip().lower() + i = (intent or "").strip().lower() + c = (connective or "").strip() + o = (obj or "").strip().lower() + if not s or not i or not c or not o: + raise SupersessionError( + "subject/intent/connective/object are all required" + ) + if i not in _VALID_INTENTS: + raise SupersessionError( + f"intent {i!r} is not in the supported whitelist " + f"({sorted(_VALID_INTENTS)})" + ) + return s, i, c, o + + +def supersede_chain( + *, + old_chain_id: str, + subject: str, + intent: str, + connective: str, + object_: str, + review_date: str, + corpus_path: Path, + adr_id: str = "adr-0057", + operator_note: str = "", + new_chain_id: str | None = None, +) -> str: + """Retire ``old_chain_id`` by appending a new chain that supersedes it. + + Returns the ``chain_id`` of the new active entry. Raises + ``SupersessionError`` on any pre-condition violation; in that + case the corpus on disk is unchanged. + + Pre-conditions (run in this order — cheapest first): + 1. ``review_date`` matches ``YYYY-MM-DD``. + 2. New chain fields are non-empty and ``intent`` is in + ``_VALID_INTENTS``. + 3. ``old_chain_id`` is currently *active* in the audit report + (it must exist and not already be superseded). + 4. The new chain itself passes the same audit gates that the + runtime loader applies (pack-consistency on subject/object, + non-self-supersede). This is verified by re-running + ``audit_corpus`` after the append and asserting the active + set has shifted exactly as expected; on any drift the + appended line is rolled back by truncation. + + Step 4 is the safety net: it makes silent introduction of a + pack-missing or otherwise invalid replacement impossible. + """ + _ = operator_note # reserved for future audit-log wiring; CLI surfaces it today + old_id = (old_chain_id or "").strip() + if not old_id: + raise SupersessionError("old_chain_id is required") + + _validate_review_date(review_date) + s, i, c, o = _validate_chain_fields(subject, intent, connective, object_) + + resolved_new_id = (new_chain_id or "").strip() or f"{i}_{s}_{c}_{o}" + if resolved_new_id == old_id: + raise SupersessionError( + "new chain_id is identical to old_chain_id — supersession " + "must produce a distinct active chain" + ) + + # -- Pre-append audit: old_chain_id must currently be active. + pre = audit_corpus(corpus_path) + active_chain_ids = {entry.chain_id for entry in pre.loaded} + if old_id not in active_chain_ids: + # Either the chain does not exist or it is already superseded. + if any(d.chain_id == old_id for d in pre.dropped): + raise SupersessionError( + f"old_chain_id {old_id!r} is already inactive " + f"(dropped by audit) — refusing to double-supersede" + ) + raise SupersessionError( + f"old_chain_id {old_id!r} is not in the active corpus" + ) + if resolved_new_id in active_chain_ids: + raise SupersessionError( + f"new chain_id {resolved_new_id!r} is already active; " + "choose a distinct connective/object or pass --new-chain-id" + ) + + chain = { + "subject": s, + "intent": i, + "connective": c, + "object": o, + } + review_date_clean = review_date.strip() + provenance = Provenance( + adr_id=adr_id, + source="hand_authored", + review_date=review_date_clean, + raw=f"{adr_id}:hand_authored:{review_date_clean}:supersede({old_id})", + ) + + # Snapshot bytes so we can roll back if the post-audit invariant + # is violated (defence in depth: should be impossible given the + # pre-checks, but corpus correctness is load-bearing). + bytes_before = corpus_path.read_bytes() if corpus_path.exists() else b"" + + written_chain_id = append_chain_to_corpus( + chain, + corpus_path=corpus_path, + provenance=provenance, + chain_id=resolved_new_id, + superseded_by=old_id, + ) + + # -- Post-append audit: confirm the active set shifted. + post = audit_corpus(corpus_path) + post_active = {entry.chain_id for entry in post.loaded} + expected_dropped = ( + f"superseded_by:{old_id}" in {d.reason for d in post.dropped} + ) + if ( + written_chain_id not in post_active + or old_id in post_active + or not expected_dropped + ): + # Roll back: truncate to bytes_before. + corpus_path.write_bytes(bytes_before) + raise SupersessionError( + f"post-append audit rejected the supersession " + f"(new={written_chain_id!r} active={written_chain_id in post_active}, " + f"old_still_active={old_id in post_active}); corpus rolled back" + ) + + return written_chain_id + + +__all__ = [ + "SupersessionError", + "supersede_chain", +] diff --git a/tests/test_supersede.py b/tests/test_supersede.py new file mode 100644 index 00000000..9eb3b60a --- /dev/null +++ b/tests/test_supersede.py @@ -0,0 +1,288 @@ +"""ADR-0057 follow-up — operator-driven supersession. + +Pins: + + - ``supersede_chain`` is the only path that emits a chain with a + ``superseded_by`` field. + - Single write surface preserved: it composes around + ``proposals.append_chain_to_corpus``, not its own writer. + - Pack-consistency, intent whitelist, and self-supersede gates + fire before any byte is written. + - Already-superseded targets cannot be double-retired. + - Post-audit shifts exactly one chain from active → dropped and + adds the replacement to active. + - Failure modes leave the corpus byte-identical. +""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from teaching.audit import audit_corpus +from teaching.supersede import SupersessionError, supersede_chain + + +def _seed(tmp_path: Path) -> Path: + """Two pack-consistent active chains; we'll retire the first.""" + 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 + + +# --------------------------------------------------------------------------- +# Happy path +# --------------------------------------------------------------------------- + + +def test_supersede_appends_one_line_and_retires_target(tmp_path: Path) -> None: + corpus = _seed(tmp_path) + pre_lines = corpus.read_text(encoding="utf-8").splitlines() + new_id = 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, + ) + assert new_id == "cause_light_grounds_truth" + + post_lines = corpus.read_text(encoding="utf-8").splitlines() + assert len(post_lines) == len(pre_lines) + 1 + # Earlier lines are byte-identical (append-only at disk level). + assert post_lines[: len(pre_lines)] == pre_lines + + last = json.loads(post_lines[-1]) + assert last["chain_id"] == "cause_light_grounds_truth" + assert last["superseded_by"] == "cause_light_reveals_truth" + assert "supersede(cause_light_reveals_truth)" in last["provenance"] + + +def test_audit_after_supersede_shifts_active_set(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) + active_ids = {c.chain_id for c in report.loaded} + dropped_ids = {d.chain_id for d in report.dropped} + assert "cause_light_reveals_truth" not in active_ids + assert "cause_light_reveals_truth" in dropped_ids + assert "cause_light_grounds_truth" in active_ids + # Unrelated chain untouched. + assert "verification_memory_requires_recall" in active_ids + + +def test_explicit_new_chain_id_honoured(tmp_path: Path) -> None: + corpus = _seed(tmp_path) + new_id = 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, + new_chain_id="cause_light_revised_v2", + ) + assert new_id == "cause_light_revised_v2" + + +# --------------------------------------------------------------------------- +# Eligibility / validation gates +# --------------------------------------------------------------------------- + + +def test_rejects_unknown_old_chain_id(tmp_path: Path) -> None: + corpus = _seed(tmp_path) + bytes_before = corpus.read_bytes() + with pytest.raises(SupersessionError, match="not in the active corpus"): + supersede_chain( + old_chain_id="does_not_exist", + subject="light", intent="cause", + connective="grounds", object_="truth", + review_date="2026-05-18", + corpus_path=corpus, + ) + assert corpus.read_bytes() == bytes_before + + +def test_rejects_double_supersede(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_after_first = corpus.read_bytes() + with pytest.raises(SupersessionError, match="already inactive"): + supersede_chain( + old_chain_id="cause_light_reveals_truth", + subject="light", intent="cause", + connective="orders", object_="truth", + review_date="2026-05-18", + corpus_path=corpus, + ) + assert corpus.read_bytes() == bytes_after_first + + +def test_rejects_pack_missing_subject(tmp_path: Path) -> None: + corpus = _seed(tmp_path) + bytes_before = corpus.read_bytes() + with pytest.raises(SupersessionError): + supersede_chain( + old_chain_id="cause_light_reveals_truth", + subject="zzznotalemma", intent="cause", + connective="grounds", object_="truth", + review_date="2026-05-18", + corpus_path=corpus, + ) + assert corpus.read_bytes() == bytes_before + + +def test_rejects_unsupported_intent(tmp_path: Path) -> None: + corpus = _seed(tmp_path) + bytes_before = corpus.read_bytes() + with pytest.raises(SupersessionError, match="whitelist"): + supersede_chain( + old_chain_id="cause_light_reveals_truth", + subject="light", intent="definition", + connective="grounds", object_="truth", + review_date="2026-05-18", + corpus_path=corpus, + ) + assert corpus.read_bytes() == bytes_before + + +def test_rejects_self_supersede(tmp_path: Path) -> None: + corpus = _seed(tmp_path) + bytes_before = corpus.read_bytes() + with pytest.raises(SupersessionError, match="identical"): + supersede_chain( + old_chain_id="cause_light_reveals_truth", + subject="light", intent="cause", + connective="reveals", object_="truth", + review_date="2026-05-18", + corpus_path=corpus, + ) + assert corpus.read_bytes() == bytes_before + + +def test_rejects_collision_with_active_chain_id(tmp_path: Path) -> None: + corpus = _seed(tmp_path) + bytes_before = corpus.read_bytes() + with pytest.raises(SupersessionError, match="already active"): + supersede_chain( + old_chain_id="cause_light_reveals_truth", + subject="memory", intent="verification", + connective="requires", object_="recall", + review_date="2026-05-18", + corpus_path=corpus, + ) + assert corpus.read_bytes() == bytes_before + + +def test_rejects_bad_review_date(tmp_path: Path) -> None: + corpus = _seed(tmp_path) + bytes_before = corpus.read_bytes() + with pytest.raises(SupersessionError, match="YYYY-MM-DD"): + supersede_chain( + old_chain_id="cause_light_reveals_truth", + subject="light", intent="cause", + connective="grounds", object_="truth", + review_date="May 18 2026", + corpus_path=corpus, + ) + assert corpus.read_bytes() == bytes_before + + +def test_rejects_empty_required_field(tmp_path: Path) -> None: + corpus = _seed(tmp_path) + bytes_before = corpus.read_bytes() + with pytest.raises(SupersessionError, match="required"): + supersede_chain( + old_chain_id="cause_light_reveals_truth", + subject="light", intent="cause", + connective=" ", object_="truth", + review_date="2026-05-18", + corpus_path=corpus, + ) + assert corpus.read_bytes() == bytes_before + + +# --------------------------------------------------------------------------- +# Runtime parity +# --------------------------------------------------------------------------- + + +def test_runtime_loader_honours_supersede_chain_output(tmp_path: Path) -> None: + """After supersede_chain runs, the live runtime loader (pointed + at this tmp corpus) sees the same active set as the audit report.""" + 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, + ) + from unittest.mock import patch + + from chat import teaching_grounding as tg + + with patch.object(tg, "_CORPUS_PATH", corpus): + tg._corpus_index.cache_clear() + try: + index = tg._corpus_index() + runtime_ids = {c.chain_id for c in index.values()} + finally: + tg._corpus_index.cache_clear() + + audit_ids = {c.chain_id for c in audit_corpus(corpus).loaded} + assert runtime_ids == audit_ids + assert "cause_light_reveals_truth" not in runtime_ids + assert "cause_light_grounds_truth" in runtime_ids + + +# --------------------------------------------------------------------------- +# Provenance shape +# --------------------------------------------------------------------------- + + +def test_provenance_is_hand_authored_with_supersede_tag(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) + new_entry = next( + c for c in report.loaded if c.chain_id == "cause_light_grounds_truth" + ) + assert new_entry.provenance.source == "hand_authored" + assert new_entry.provenance.review_date is not None + assert new_entry.provenance.review_date.startswith("2026-05-18") + assert "supersede(cause_light_reveals_truth)" in new_entry.provenance.raw