From 6f4b2b7b2c8352e985ae4ba1599088d85e50176f Mon Sep 17 00:00:00 2001 From: Shay Date: Mon, 18 May 2026 10:52:23 -0700 Subject: [PATCH] =?UTF-8?q?feat(adr-0057):=20anti-regression=20demo=20?= =?UTF-8?q?=E2=80=94=20three-gate=20defense=20against=20learning=20harm?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `core demo anti-regression` (+ `--json`) is a self-contained walkthrough of the three independent gates that every reviewed-corpus extension must pass. Designed for showcasing CORE's epistemic discipline to reviewers / industry observers — no LLM provider has a published equivalent. Scenes: - S1. Eligibility predicate refuses an undetermined-polarity candidate before any replay is invoked. ProposalError raised; no log row. - S2. Replay-equivalence gate auto-rejects a regressing candidate with the named regressed metrics in the operator note. Uses the documented `run_replay=` kwarg of `propose_from_candidate` to inject a controlled regression of the same `ReplayEvidence` shape the real gate produces. - S3. Real `teaching.replay.run_replay_equivalence` runs the cognition public lane. A replay-equivalent candidate reaches 'pending' — operator `--accept` is still required to write. Each scene asserts the active corpus is byte-identical pre/post. - evals/anti_regression/run_demo.py — `run_demo(emit_json=False)` returns a structured `DemoReport`; verbose human output by default, JSON on flag. - core/cli.py — `core demo anti-regression` target wired alongside audit-tour / pack-measurements / long-context-comparison. - tests/test_anti_regression_demo.py — 5 tests pin each scene's load-bearing claim + the corpus-byte-identical invariant. Lane state: anti-regression-demo 5 new — green. Demo runs in ~10s end-to-end. --- core/cli.py | 13 +- evals/anti_regression/__init__.py | 1 + evals/anti_regression/run_demo.py | 430 +++++++++++++++++++++++++++++ tests/test_anti_regression_demo.py | 63 +++++ 4 files changed, 506 insertions(+), 1 deletion(-) create mode 100644 evals/anti_regression/__init__.py create mode 100644 evals/anti_regression/run_demo.py create mode 100644 tests/test_anti_regression_demo.py diff --git a/core/cli.py b/core/cli.py index 2ef88a4d..8635463f 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 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" +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 demo anti-regression\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": ( @@ -1397,6 +1397,14 @@ def cmd_demo(args: argparse.Namespace) -> int: _print_human(report) return 0 + if target == "anti-regression": + from evals.anti_regression.run_demo import run_demo + + report = run_demo(emit_json=args.json) + if args.json: + print(json.dumps(report, indent=2, sort_keys=True)) + return 0 + if target == "long-context-comparison": from evals.long_context_cost.comparison_runner import ( run_comparison, @@ -1786,6 +1794,7 @@ def build_parser() -> argparse.ArgumentParser: "audit-tour", "pack-measurements", "long-context-comparison", + "anti-regression", "list-results", ], help=( @@ -1798,6 +1807,8 @@ def build_parser() -> argparse.ArgumentParser: "numbers across the three ratified identity packs. " "long-context-comparison: ADR-0045 — CORE exact recall NIAH at " "N∈{100,1k,10k,100k} paired with frozen transformer baselines. " + "anti-regression: ADR-0057 — three-gate defense against learning " + "harmful chains (eligibility / replay-equivalence / operator). " "list-results: index every JSON report in the results directory." ), ) diff --git a/evals/anti_regression/__init__.py b/evals/anti_regression/__init__.py new file mode 100644 index 00000000..4cf6edc0 --- /dev/null +++ b/evals/anti_regression/__init__.py @@ -0,0 +1 @@ +"""Anti-regression demo — the replay-equivalence gate in action.""" diff --git a/evals/anti_regression/run_demo.py b/evals/anti_regression/run_demo.py new file mode 100644 index 00000000..4362e16a --- /dev/null +++ b/evals/anti_regression/run_demo.py @@ -0,0 +1,430 @@ +"""Anti-regression demo — three scenes showing how CORE refuses to learn +something that would make it worse. + +The thesis: when a system extends its own knowledge, **the gate that +decides what to admit is the load-bearing part** — not the proposer. +CORE's reviewed-corpus extension path (ADR-0057) has three independent +gates that must each pass before any byte is written: + + S1. Eligibility predicate (mechanical, pre-replay). + Five mechanical checks on the candidate's shape (polarity, + evidence-floor, claim-domain, boundary-clean, chain-complete). + Ineligible candidates raise ``ProposalError`` and never enter + the proposal log. + + S2. Replay-equivalence gate (mechanical, post-eligibility). + The full cognition lane runs against the active corpus AND + against a transient copy with the proposed chain appended. + Any strict-decrease in a watched metric auto-rejects the + proposal with the metrics named in the operator note. + Active corpus file bytes are byte-identical pre/post. + + S3. Operator review (manual, post-replay). + Even a replay-equivalent proposal only reaches the *pending* + state — explicit ``core teaching review --accept`` is + required to write to the active corpus. + +This demo runs each scene end-to-end against the real ``ProposalLog`` +in an isolated temp directory. No active corpus or production log is +touched. + +Scenes 1 and 3 use the **real** ``teaching.replay.run_replay_equivalence`` +function. Scene 2 injects a controlled replay function (via the +documented ``run_replay=`` kwarg of ``propose_from_candidate``) that +returns a regressed ``ReplayEvidence`` of the same shape the real gate +produces — demonstrating the auto-rejection lifecycle on a synthetic +regression deterministically. In production the real gate produces +this same shape when a real regression is detected. +""" + +from __future__ import annotations + +import tempfile +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +from teaching.discovery import DiscoveryCandidate, EvidencePointer +from teaching.proposals import ( + ProposalError, + ProposalLog, + ReplayEvidence, + propose_from_candidate, +) + + +_VERBOSE = True + + +def _say(*args: Any, **kwargs: Any) -> None: + if _VERBOSE: + print(*args, **kwargs) + + +def _print_header(title: str, claim: str) -> None: + _say() + _say("─" * 72) + _say(f" {title}") + _say("─" * 72) + _say(f" CLAIM: {claim}") + _say() + + +# --------------------------------------------------------------------------- +# Synthetic ReplayEvidence builder for Scene 2 +# --------------------------------------------------------------------------- + + +def _make_regressed_replay( + *, regressed_metrics: tuple[str, ...] +) -> Any: + """Return a ``run_replay`` function that emits a regressed + ``ReplayEvidence`` with the same shape the real gate produces. + """ + baseline = { + "intent_accuracy": 1.0, + "surface_groundedness": 1.0, + "term_capture_rate": 0.9167, + "versor_closure_rate": 1.0, + } + candidate = dict(baseline) + for m in regressed_metrics: + candidate[m] = round(candidate[m] - 0.0833, 4) + + def _fn(chain: dict[str, Any]) -> ReplayEvidence: # noqa: ARG001 + return ReplayEvidence( + baseline=baseline, + candidate=candidate, + regressed_metrics=tuple(sorted(regressed_metrics)), + replay_equivalent=False, + ) + + return _fn + + +# --------------------------------------------------------------------------- +# Candidate builders +# --------------------------------------------------------------------------- + + +def _candidate_undetermined() -> DiscoveryCandidate: + """A candidate that fails the eligibility predicate at the polarity + gate. Used for Scene 1.""" + return DiscoveryCandidate( + candidate_id="demo_undetermined_001", + proposed_chain={ + "subject": "wisdom", "intent": "cause", + "connective": "informs", "object": "judgment", + }, + trigger="would_have_grounded", + source_turn_trace="demo_trace_001", + pack_consistent=True, + boundary_clean=True, + polarity="undetermined", + claim_domain="factual", + evidence=( + EvidencePointer( + source="corpus", + ref="cause_wisdom_orders_judgment", + polarity="affirms", + epistemic_status="reviewed", + ), + ), + ) + + +def _candidate_for_regression() -> DiscoveryCandidate: + """A candidate that passes eligibility but (under the injected + regression replay) is auto-rejected for regressing + ``surface_groundedness`` and ``term_capture_rate``.""" + return DiscoveryCandidate( + candidate_id="demo_regression_002", + proposed_chain={ + "subject": "knowledge", "intent": "cause", + "connective": "obscures", "object": "wisdom", + }, + trigger="would_have_grounded", + source_turn_trace="demo_trace_002", + pack_consistent=True, + boundary_clean=True, + polarity="affirms", + claim_domain="factual", + evidence=( + EvidencePointer( + source="corpus", + ref="cause_knowledge_requires_evidence", + polarity="affirms", + epistemic_status="reviewed", + ), + ), + ) + + +def _candidate_pass_through() -> DiscoveryCandidate: + """A candidate that passes both eligibility and the real + replay-equivalence gate. Lands in ``pending`` awaiting + operator review.""" + return DiscoveryCandidate( + candidate_id="demo_pass_003", + proposed_chain={ + "subject": "judgment", "intent": "verification", + "connective": "requires", "object": "evidence", + }, + trigger="would_have_grounded", + source_turn_trace="demo_trace_003", + pack_consistent=True, + boundary_clean=True, + polarity="affirms", + claim_domain="factual", + evidence=( + EvidencePointer( + source="corpus", + ref="verification_truth_requires_evidence", + polarity="affirms", + epistemic_status="reviewed", + ), + ), + ) + + +# --------------------------------------------------------------------------- +# Scene results +# --------------------------------------------------------------------------- + + +@dataclass(frozen=True, slots=True) +class SceneResult: + scene: str + claim: str + outcome: str + candidate_id: str + proposed_chain: dict[str, Any] + proposal_id: str | None + review_state: str + replay_evidence: dict[str, Any] | None + operator_note: str + error: str | None + corpus_byte_identical: bool + + def as_dict(self) -> dict[str, Any]: + return { + "scene": self.scene, + "claim": self.claim, + "outcome": self.outcome, + "candidate_id": self.candidate_id, + "proposed_chain": self.proposed_chain, + "proposal_id": self.proposal_id, + "review_state": self.review_state, + "replay_evidence": self.replay_evidence, + "operator_note": self.operator_note, + "error": self.error, + "corpus_byte_identical": self.corpus_byte_identical, + } + + +@dataclass(frozen=True, slots=True) +class DemoReport: + scenes: tuple[SceneResult, ...] + all_gates_held: bool + active_corpus_byte_identical: bool + + def as_dict(self) -> dict[str, Any]: + return { + "scenes": [s.as_dict() for s in self.scenes], + "all_gates_held": self.all_gates_held, + "active_corpus_byte_identical": self.active_corpus_byte_identical, + } + + +# --------------------------------------------------------------------------- +# Scenes +# --------------------------------------------------------------------------- + + +def _read_active_corpus_bytes() -> bytes: + from chat.teaching_grounding import _CORPUS_PATH + return _CORPUS_PATH.read_bytes() if _CORPUS_PATH.exists() else b"" + + +def _scene1_eligibility_gate(log_path: Path) -> SceneResult: + _print_header( + "S1. Eligibility predicate refuses ineligible candidates", + "An undetermined-polarity candidate never enters the proposal " + "log. ProposalError raised; no log row; no replay invocation.", + ) + log = ProposalLog(log_path) + candidate = _candidate_undetermined() + bytes_before = _read_active_corpus_bytes() + error: str | None = None + try: + propose_from_candidate(candidate, log=log) + except ProposalError as exc: + error = str(exc) + bytes_after = _read_active_corpus_bytes() + + _say(f" candidate.polarity : {candidate.polarity}") + _say(f" outcome : ProposalError raised") + _say(f" error : {error}") + _say(f" proposal log rows : {len(log.current_state())}") + _say(f" active corpus byte-eq : {bytes_before == bytes_after}") + return SceneResult( + scene="S1_eligibility_gate", + claim=( + "Five mechanical eligibility gates fire before any replay " + "is invoked. Undetermined-polarity candidates never enter " + "the proposal log." + ), + outcome="rejected_pre_replay", + candidate_id=candidate.candidate_id, + proposed_chain=candidate.proposed_chain, + proposal_id=None, + review_state="(not in log)", + replay_evidence=None, + operator_note="", + error=error, + corpus_byte_identical=(bytes_before == bytes_after), + ) + + +def _scene2_replay_auto_reject(log_path: Path) -> SceneResult: + _print_header( + "S2. Replay-equivalence gate auto-rejects a regressing chain", + "An eligible candidate whose append would regress the cognition " + "lane is auto-rejected with the named regressed metrics in the " + "operator note. Active corpus byte-identical pre/post.", + ) + log = ProposalLog(log_path) + candidate = _candidate_for_regression() + bytes_before = _read_active_corpus_bytes() + proposal = propose_from_candidate( + candidate, + log=log, + run_replay=_make_regressed_replay( + regressed_metrics=("surface_groundedness", "term_capture_rate"), + ), + ) + bytes_after = _read_active_corpus_bytes() + rec = log.find(proposal.proposal_id) or {} + ev = rec.get("replay_evidence") or {} + + _say(f" proposal_id : {proposal.proposal_id}") + _say(f" baseline metrics : {ev.get('baseline')}") + _say(f" candidate metrics : {ev.get('candidate')}") + _say(f" regressed_metrics : {ev.get('regressed_metrics')}") + _say(f" replay_equivalent : {ev.get('replay_equivalent')}") + _say(f" state : {rec.get('state')}") + _say(f" operator_note : {rec.get('operator_note')}") + _say(f" active corpus byte-eq : {bytes_before == bytes_after}") + return SceneResult( + scene="S2_replay_auto_reject", + claim=( + "Replay-equivalence gate compares the full cognition lane " + "metrics; any strict-decrease auto-rejects with the regressed " + "metric names in the operator note. Active corpus untouched." + ), + outcome="auto_rejected_on_regression", + candidate_id=candidate.candidate_id, + proposed_chain=candidate.proposed_chain, + proposal_id=proposal.proposal_id, + review_state=str(rec.get("state")), + replay_evidence=ev, + operator_note=str(rec.get("operator_note") or ""), + error=None, + corpus_byte_identical=(bytes_before == bytes_after), + ) + + +def _scene3_real_gate_pass_through(log_path: Path) -> SceneResult: + _print_header( + "S3. Real replay gate runs cognition lane; pass → pending", + "An eligible candidate whose append does not regress reaches " + "'pending' state. Operator --accept is still required to write " + "to the active corpus; the gate is a precondition, not a " + "permission.", + ) + log = ProposalLog(log_path) + candidate = _candidate_pass_through() + bytes_before = _read_active_corpus_bytes() + proposal = propose_from_candidate(candidate, log=log) + bytes_after = _read_active_corpus_bytes() + rec = log.find(proposal.proposal_id) or {} + ev = rec.get("replay_evidence") or {} + + _say(f" proposal_id : {proposal.proposal_id}") + _say(f" baseline metrics : {ev.get('baseline')}") + _say(f" candidate metrics : {ev.get('candidate')}") + _say(f" regressed_metrics : {ev.get('regressed_metrics')}") + _say(f" replay_equivalent : {ev.get('replay_equivalent')}") + _say(f" state : {rec.get('state')}") + _say(f" next step : core teaching review {proposal.proposal_id} " + "--accept --review-date YYYY-MM-DD") + _say(f" active corpus byte-eq : {bytes_before == bytes_after}") + return SceneResult( + scene="S3_real_gate_pass_through", + claim=( + "A replay-equivalent candidate reaches 'pending' but is " + "not auto-applied. Operator --accept is the third gate." + ), + outcome="pending_awaiting_operator", + candidate_id=candidate.candidate_id, + proposed_chain=candidate.proposed_chain, + proposal_id=proposal.proposal_id, + review_state=str(rec.get("state")), + replay_evidence=ev, + operator_note="", + error=None, + corpus_byte_identical=(bytes_before == bytes_after), + ) + + +# --------------------------------------------------------------------------- +# Public entry point +# --------------------------------------------------------------------------- + + +def run_demo(*, emit_json: bool = False) -> dict[str, Any]: + """Run all three scenes and return a structured report.""" + global _VERBOSE + _VERBOSE = not emit_json + + active_bytes_before = _read_active_corpus_bytes() + + with tempfile.TemporaryDirectory() as tmpdir: + log_path = Path(tmpdir) / "demo_proposals.jsonl" + s1 = _scene1_eligibility_gate(log_path) + s2 = _scene2_replay_auto_reject(log_path) + s3 = _scene3_real_gate_pass_through(log_path) + + active_bytes_after = _read_active_corpus_bytes() + + scenes = (s1, s2, s3) + all_gates_held = ( + s1.outcome == "rejected_pre_replay" + and s2.outcome == "auto_rejected_on_regression" + and s3.outcome == "pending_awaiting_operator" + ) + report = DemoReport( + scenes=scenes, + all_gates_held=all_gates_held, + active_corpus_byte_identical=(active_bytes_before == active_bytes_after), + ) + + if _VERBOSE: + _say() + _say("═" * 72) + _say(" RESULT") + _say("═" * 72) + _say(f" all three gates held : {report.all_gates_held}") + _say(f" active corpus byte-eq : {report.active_corpus_byte_identical}") + _say() + _say( + " Each gate is independent and fails closed. Bad proposals " + "stop at the cheapest applicable gate. The active corpus is " + "never written to anywhere in this demo." + ) + _say() + + return report.as_dict() + + +__all__ = ["run_demo"] diff --git a/tests/test_anti_regression_demo.py b/tests/test_anti_regression_demo.py new file mode 100644 index 00000000..8512ed38 --- /dev/null +++ b/tests/test_anti_regression_demo.py @@ -0,0 +1,63 @@ +"""Anti-regression demo — pins each scene's load-bearing claim. + +These are the falsifiable assertions the demo would make to a viewer. +If any assertion fails, the demo's headline claim no longer holds. +""" + +from __future__ import annotations + +from evals.anti_regression.run_demo import run_demo + + +def test_demo_runs_to_completion_with_all_three_gates_holding() -> None: + report = run_demo(emit_json=True) + assert report["all_gates_held"] is True + assert report["active_corpus_byte_identical"] is True + assert len(report["scenes"]) == 3 + + +def test_s1_eligibility_gate_rejects_pre_replay() -> None: + report = run_demo(emit_json=True) + s1 = report["scenes"][0] + assert s1["scene"] == "S1_eligibility_gate" + assert s1["outcome"] == "rejected_pre_replay" + assert s1["proposal_id"] is None + assert s1["replay_evidence"] is None + assert "undetermined" in (s1["error"] or "") + assert s1["corpus_byte_identical"] is True + + +def test_s2_replay_gate_auto_rejects_with_named_metrics() -> None: + report = run_demo(emit_json=True) + s2 = report["scenes"][1] + assert s2["scene"] == "S2_replay_auto_reject" + assert s2["outcome"] == "auto_rejected_on_regression" + assert s2["review_state"] == "rejected" + assert s2["replay_evidence"]["replay_equivalent"] is False + assert "surface_groundedness" in s2["replay_evidence"]["regressed_metrics"] + assert "term_capture_rate" in s2["replay_evidence"]["regressed_metrics"] + # The operator note must name the regressed metrics. + assert "surface_groundedness" in s2["operator_note"] + assert "term_capture_rate" in s2["operator_note"] + assert s2["corpus_byte_identical"] is True + + +def test_s3_real_gate_passes_to_pending_not_accepted() -> None: + report = run_demo(emit_json=True) + s3 = report["scenes"][2] + assert s3["scene"] == "S3_real_gate_pass_through" + assert s3["outcome"] == "pending_awaiting_operator" + assert s3["review_state"] == "pending" + assert s3["replay_evidence"]["replay_equivalent"] is True + assert s3["replay_evidence"]["regressed_metrics"] == [] + assert s3["corpus_byte_identical"] is True + + +def test_active_corpus_never_touched_across_full_demo() -> None: + """Defence-in-depth: even though each scene asserts byte-identity, + re-confirm at the report level — the demo never writes to the + production teaching corpus regardless of scene outcomes.""" + report = run_demo(emit_json=True) + assert report["active_corpus_byte_identical"] is True + for scene in report["scenes"]: + assert scene["corpus_byte_identical"] is True