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

feat(adr-0057): operator supersede CLI — retire active chain by appended replacement

Browse source 
`core teaching supersede <old_chain_id> --subject ... --intent ... --connective ...
--object ... --review-date YYYY-MM-DD` is the second corpus mutation surface
(alongside accept_proposal). No replay gate — it's a deliberate operator action
that replaces a hand-authored or previously discovery-promoted chain.

- teaching/supersede.py — `supersede_chain()` orchestrator with pre-checks
  (review_date format, intent whitelist, pack-consistency via re-audit,
  no double-supersede, no self-supersede, no new-chain-id collision) and
  byte-identical rollback on post-audit failure.
- teaching/proposals.py — extended `append_chain_to_corpus` with optional
  `superseded_by` kwarg; remains the only function in the codebase that
  writes to the active teaching corpus.
- core/cli.py — `core teaching supersede` subcommand wired to the live
  `_CORPUS_PATH`; EPILOG updated with example.
- tests/test_supersede.py — 13 tests pin every gate, byte-identical
  rollback on rejection, append-only at disk level, audit-and-runtime
  parity after supersession, hand_authored provenance with
  `supersede(<old_chain_id>)` tag.

Lane state: smoke 67 / cognition 121 / teaching 17 / supersede 13 / audit 23 /
proposals 16 / contemplation 16 / contemplation-wiring 6 / discovery 24 — green.
`core eval cognition`: intent 100% / surface 100% / term 91.7% / versor 100% — unchanged.
This commit is contained in:
Shay 2026-05-18 10:35:49 -07:00
parent e03ab4b609
commit 8d2c84a041
4 changed files with 555 additions and 5 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

59
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 teaching audit\n core teaching audit --json\n core teaching propose <candidate-jsonl-path>\n core teaching proposals --state pending\n core teaching review <proposal_id> --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 <candidate-jsonl-path>\n core teaching proposals --state pending\n core teaching review <proposal_id> --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: <intent>_<subject>_<connective>_<object>)",
)
teaching_supersede.set_defaults(func=cmd_teaching_supersede)
rust = subparsers.add_parser(
"rust",
help="build, test, and inspect the Rust backend",

17
teaching/proposals.py
View file

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

196
teaching/supersede.py Normal file
View file

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

288
tests/test_supersede.py Normal file
View file

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