feat(adr-0057): operator supersession history view — closes the supersede loop
`core teaching supersessions` (+ `--json`) pairs each retired chain with its active replacement. Derived view over `audit_corpus()`; pure, read-only. - teaching/audit.py — `SupersessionRecord` + `supersession_history(report)` returns retired→replacement pairs ordered by retired-line (disk order, oldest first). Orphan supersessions (retired with no live entry carrying the matching `superseded_by` — e.g. chained retirements where the middle link itself was retired) surface as `replacement=None` so silent corpus drift is inspectable. - core/cli.py — `core teaching supersessions [--json]`. Exit 1 if any orphan is detected (catches silent drift in CI); 0 otherwise. - tests/test_supersession_history.py — 7 tests pin empty-history, single-pair shape, chained-supersession surfaces both pairs, line-no ordering, orphan detection, JSON round-trip, no corpus mutation. Lane state: smoke 67 / cognition 121 / supersession-history 7 new / supersede 13 / audit 23 — green. `core eval cognition`: unchanged (intent 100% / surface 100% / term 91.7% / versor 100%). Real corpus today reports `(no supersessions)`.
This commit is contained in:
Shay
parent
8d2c84a041
commit
3cad6686cc
3 changed files with 312 additions and 1 deletions
59
core/cli.py
59
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 <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"
|
||||
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 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: <ORPHAN — no live entry carries this superseded_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",
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
190
tests/test_supersession_history.py
Normal file
190
tests/test_supersession_history.py
Normal file
|
|
@ -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
|
||||
Loading…
Reference in a new issue