core/evals/anti_regression/run_demo.py

"""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]:
        # ``all_claims_supported`` is the canonical cross-demo success
        # field — added as an alias so operator tooling (and the CI gate)
        # can rely on one uniform boolean key across every ``core demo``
        # target.  Existing fields are preserved for backwards compat.
        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,
            "all_claims_supported": (
                self.all_gates_held and 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"]