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

feat(adr-0057): anti-regression demo — three-gate defense against learning harm

Browse source 
`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.
This commit is contained in:
Shay 2026-05-18 10:52:23 -07:00
parent 3cad6686cc
commit 6f4b2b7b2c
4 changed files with 506 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

13
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 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 <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 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."
),
)

1
evals/anti_regression/__init__.py Normal file
View file

@ -0,0 +1 @@
"""Anti-regression demo — the replay-equivalence gate in action."""

430
evals/anti_regression/run_demo.py Normal file
View file

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

63
tests/test_anti_regression_demo.py Normal file
View file

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