From a71b321a9ab94ba32e486122ec20fec8a2a2e01f Mon Sep 17 00:00:00 2001 From: Shay Date: Mon, 18 May 2026 10:57:41 -0700 Subject: [PATCH] =?UTF-8?q?feat(adr-0055-0057):=20learning-loop=20demo=20?= =?UTF-8?q?=E2=80=94=20cold=20turn=20to=20grounded=20surface,=20end-to-end?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `core demo learning-loop` (+ `--json`) walks a single prompt through the full ADR-0055..0057 inter-session-memory architecture: S1. Cold turn → universal disclosure, grounding_source=none S2. Discovery emission → DiscoveryCandidate to attached sink S3. Operator proposal → real replay-equivalence gate, no regression S4. Operator accept → TRANSIENT corpus only; active untouched S5. Same prompt → teaching-grounded surface with the new chain Before / after on the deterministic prompt "Why does thought exist?": before: [none] I don't know — insufficient grounding for that yet. after: [teaching] thought — teaching-grounded (cognition_chains_v1): cognition.thought; logos.internal. thought reveals meaning (cognition.meaning). No session evidence yet. The active corpus on disk is byte-identical pre/post. The demo writes only to a transient corpus, then swaps `_CORPUS_PATH` for the after turn — the same pattern the replay-equivalence gate uses. - evals/learning_loop/run_demo.py — `run_demo(emit_json=False)` returns a structured `DemoReport` with both surfaces and per-scene detail. - core/cli.py — `core demo learning-loop` target wired. - tests/test_learning_loop_demo.py — 7 tests pin: full loop closes, before is ungrounded, after contains new chain atoms (thought / reveal / meaning), discovery emits ≥1, replay gate reports no regression, S4 byte-identical active + 1 line on transient, same prompt drives both surfaces. Lane state: learning-loop-demo 7 new — green. Demo runs in ~15s end-to-end (cognition lane runs twice via replay gate). No LLM provider has a published equivalent of this loop: per-fact provenance from operator accept to surface, replay-equivalence gate proving non-regression, byte-identical active state regardless of outcome, full audit trail back to the originating cold turn. --- core/cli.py | 13 +- evals/learning_loop/__init__.py | 1 + evals/learning_loop/run_demo.py | 458 +++++++++++++++++++++++++++++++ tests/test_learning_loop_demo.py | 70 +++++ 4 files changed, 541 insertions(+), 1 deletion(-) create mode 100644 evals/learning_loop/__init__.py create mode 100644 evals/learning_loop/run_demo.py create mode 100644 tests/test_learning_loop_demo.py diff --git a/core/cli.py b/core/cli.py index 8635463f..d79e2574 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 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" +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 demo learning-loop\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": ( @@ -1405,6 +1405,14 @@ def cmd_demo(args: argparse.Namespace) -> int: print(json.dumps(report, indent=2, sort_keys=True)) return 0 + if target == "learning-loop": + from evals.learning_loop.run_demo import run_demo as run_loop_demo + + report = run_loop_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, @@ -1795,6 +1803,7 @@ def build_parser() -> argparse.ArgumentParser: "pack-measurements", "long-context-comparison", "anti-regression", + "learning-loop", "list-results", ], help=( @@ -1809,6 +1818,8 @@ def build_parser() -> argparse.ArgumentParser: "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). " + "learning-loop: ADR-0055..0057 — full cold-turn → discovery → " + "propose → accept → same-prompt-now-grounded walkthrough. " "list-results: index every JSON report in the results directory." ), ) diff --git a/evals/learning_loop/__init__.py b/evals/learning_loop/__init__.py new file mode 100644 index 00000000..13445c95 --- /dev/null +++ b/evals/learning_loop/__init__.py @@ -0,0 +1 @@ +"""Learning-loop demo — cold turn → discovery → propose → accept → grounded.""" diff --git a/evals/learning_loop/run_demo.py b/evals/learning_loop/run_demo.py new file mode 100644 index 00000000..6e80df34 --- /dev/null +++ b/evals/learning_loop/run_demo.py @@ -0,0 +1,458 @@ +"""Learning-loop demo — CORE learning a new chain from a cold turn. + +The thesis (the demo's headline claim): + + > CORE, asked a question it cannot ground, emits structured evidence + > that a reviewed chain would have helped. An operator authors a + > proposal from that evidence. The replay-equivalence gate confirms + > the chain would not regress the cognition lane. The operator + > accepts. The **same prompt now produces a deterministic + > teaching-grounded surface** — and CORE will produce that same + > surface for that prompt every time, replayably, with full + > provenance back to the operator's accept. + +No LLM provider has this loop. Continuous pre-training is the +nearest analog and is fundamentally different: opaque gradient +updates over uncurated data without per-fact provenance, without +operator review, without a replay-equivalence gate, without an +audit trail that lets you ask "why did the model say this today +that it would not have said yesterday?" + +Five scenes, each on a real ``ChatRuntime`` against the live active +corpus. The active corpus file bytes are byte-identical pre/post — +the demo writes only to a transient corpus, then swaps ``_CORPUS_PATH`` +to that transient for the "after" turn. The same swap pattern the +replay-equivalence gate uses (``teaching.replay._swap_corpus_path``). + + S1. Cold turn. The runtime cannot ground the prompt. + S2. Discovery emission. A ``DiscoveryCandidate`` is emitted to the + attached sink — structured evidence, not a mutation. + S3. Operator-authored proposal. A complete proposal is built from + the candidate's structure plus operator-provided connective / + object / corpus-evidence pointer. The replay-equivalence gate + runs (real ``teaching.replay.run_replay_equivalence``) and + confirms no regression. + S4. Operator accept against a *transient* corpus. The active corpus + on disk is untouched; the accepted chain is written to a tmp + file. Audit + runtime both honour the transient corpus. + S5. Same prompt, now teaching-grounded. The deterministic + teaching-grounded surface contains the new chain's subject / + connective / object. Identical for any replay of the same + prompt against the same corpus state. +""" + +from __future__ import annotations + +import shutil +import tempfile +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +from chat import teaching_grounding as _tg +from chat.runtime import ChatRuntime +from teaching.discovery import DiscoveryCandidate, EvidencePointer +from teaching.proposals import ( + ProposalLog, + accept_proposal, + propose_from_candidate, +) + + +# The single prompt that drives every scene. CAUSE intent, subject +# "thought", pack-resident lemma but no `(thought, cause)` chain in +# the active corpus today — guarantees the cold-turn path. +_DEMO_PROMPT: str = "Why does thought exist?" + +# Operator-authored proposal payload. The (thought, cause) cell is +# unoccupied; the operator proposes the chain +# thought reveals meaning +# affirming evidence is the existing corpus chain +# cause_creation_reveals_meaning (creation reveals meaning) +# both endpoints are pack-resident. +_OPERATOR_CONNECTIVE: str = "reveals" +_OPERATOR_OBJECT: str = "meaning" +_OPERATOR_EVIDENCE_REF: str = "cause_creation_reveals_meaning" + + +_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() + + +# --------------------------------------------------------------------------- +# Sinks + helpers +# --------------------------------------------------------------------------- + + +class _BufferSink: + """Discovery candidate sink that retains every emitted line.""" + + def __init__(self) -> None: + self.lines: list[str] = [] + + def emit(self, line: str) -> None: + self.lines.append(line) + + +def _active_bytes() -> bytes: + return _tg._CORPUS_PATH.read_bytes() if _tg._CORPUS_PATH.exists() else b"" + + +# --------------------------------------------------------------------------- +# Scene outputs +# --------------------------------------------------------------------------- + + +@dataclass(frozen=True, slots=True) +class SceneResult: + scene: str + claim: str + detail: dict[str, Any] + + def as_dict(self) -> dict[str, Any]: + return {"scene": self.scene, "claim": self.claim, "detail": self.detail} + + +@dataclass(frozen=True, slots=True) +class DemoReport: + prompt: str + before_surface: str + before_grounding_source: str + after_surface: str + after_grounding_source: str + scenes: tuple[SceneResult, ...] + learning_loop_closed: bool + active_corpus_byte_identical: bool + + def as_dict(self) -> dict[str, Any]: + return { + "prompt": self.prompt, + "before": { + "surface": self.before_surface, + "grounding_source": self.before_grounding_source, + }, + "after": { + "surface": self.after_surface, + "grounding_source": self.after_grounding_source, + }, + "scenes": [s.as_dict() for s in self.scenes], + "learning_loop_closed": self.learning_loop_closed, + "active_corpus_byte_identical": self.active_corpus_byte_identical, + } + + +# --------------------------------------------------------------------------- +# Scenes +# --------------------------------------------------------------------------- + + +def _scene1_cold_turn(rt: ChatRuntime, sink: _BufferSink) -> tuple[SceneResult, Any]: + _print_header( + "S1. Cold turn — runtime cannot ground the prompt", + "Active corpus has no (thought, cause) chain. The runtime " + "falls through to the universal insufficient-grounding " + "disclosure. Identity / safety / ethics gates still run.", + ) + response = rt.chat(_DEMO_PROMPT) + _say(f" prompt : {_DEMO_PROMPT}") + _say(f" surface : {response.surface}") + _say(f" grounding_source : {response.grounding_source}") + _say(f" discovery candidates : {len(sink.lines)} (emitted post-turn)") + return SceneResult( + scene="S1_cold_turn", + claim="No teaching chain for (thought, cause) — runtime returns the disclosure.", + detail={ + "prompt": _DEMO_PROMPT, + "surface": response.surface, + "grounding_source": response.grounding_source, + "discovery_candidates_emitted": len(sink.lines), + }, + ), response + + +def _scene2_discovery_emission(sink: _BufferSink) -> tuple[SceneResult, dict[str, Any]]: + _print_header( + "S2. Discovery candidate — structured evidence, not a mutation", + "The runtime emits a DiscoveryCandidate (ADR-0055 Phase B) " + "documenting that a reviewed (thought, cause) chain WOULD have " + "grounded this turn. Contemplation (ADR-0056 Phase C1) " + "enriches with pack/corpus evidence pointers. Active corpus " + "is byte-identical — emission writes to the sink only.", + ) + if not sink.lines: + raise RuntimeError("expected at least one discovery candidate from S1") + import json as _json + payload = _json.loads(sink.lines[0]) + _say(f" candidate_id : {payload['candidate_id'][:16]}…") + _say(f" trigger : {payload['trigger']}") + _say(f" proposed_chain : {payload['proposed_chain']}") + _say(f" polarity : {payload['polarity']}") + _say(f" claim_domain : {payload['claim_domain']}") + _say(f" pack_consistent : {payload['pack_consistent']}") + _say(f" boundary_clean : {payload['boundary_clean']}") + _say(f" evidence (pack-only) : " + f"{[e for e in payload['evidence']]}") + return SceneResult( + scene="S2_discovery_emission", + claim=( + "DiscoveryCandidate is structured evidence: it never mutates " + "the active corpus. Phase C is the only path to mutation." + ), + detail={ + "candidate_id": payload["candidate_id"], + "proposed_chain": payload["proposed_chain"], + "polarity": payload["polarity"], + "evidence": payload["evidence"], + }, + ), payload + + +def _scene3_propose(log_path: Path, candidate_id: str) -> tuple[SceneResult, Any]: + _print_header( + "S3. Operator-authored proposal — replay-equivalence gate runs", + "From the discovery candidate's evidence, the operator authors " + "a complete chain: thought reveals meaning. Affirming evidence " + "is the existing corpus chain cause_creation_reveals_meaning. " + "The real replay gate (teaching.replay.run_replay_equivalence) " + "runs the cognition public split twice — active corpus vs. " + "transient-with-appended-chain — and reports no regression.", + ) + # Construct the operator-augmented candidate. This is the operator + # contribution: connective, object, and an affirming-source evidence + # pointer to a corpus chain that already encodes the relevant + # semantic shape. + augmented = DiscoveryCandidate( + candidate_id=candidate_id, + proposed_chain={ + "subject": "thought", "intent": "cause", + "connective": _OPERATOR_CONNECTIVE, + "object": _OPERATOR_OBJECT, + }, + trigger="would_have_grounded", + source_turn_trace="", + pack_consistent=True, + boundary_clean=True, + polarity="affirms", + claim_domain="factual", + evidence=( + EvidencePointer( + source="corpus", + ref=_OPERATOR_EVIDENCE_REF, + polarity="affirms", + epistemic_status="coherent", + ), + ), + ) + log = ProposalLog(log_path) + proposal = propose_from_candidate(augmented, log=log) + rec = log.find(proposal.proposal_id) or {} + ev = rec.get("replay_evidence") or {} + _say(f" proposal_id : {proposal.proposal_id}") + _say(f" proposed_chain : {proposal.proposed_chain}") + _say(f" evidence (corpus ref) : {_OPERATOR_EVIDENCE_REF}") + _say(f" replay baseline : {ev.get('baseline')}") + _say(f" replay candidate : {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')}") + if rec.get("state") != "pending": + raise RuntimeError( + f"expected pending state but got {rec.get('state')!r}; " + f"regressed metrics: {ev.get('regressed_metrics')}" + ) + return SceneResult( + scene="S3_propose_replay_pass", + claim=( + "Real replay gate confirms no metric regression — the " + "proposal moves to pending. Operator --accept still required." + ), + detail={ + "proposal_id": proposal.proposal_id, + "proposed_chain": proposal.proposed_chain, + "replay_evidence": ev, + "state": rec.get("state"), + }, + ), proposal + + +def _scene4_accept_against_transient( + log_path: Path, + proposal_id: str, +) -> tuple[SceneResult, Path]: + _print_header( + "S4. Operator accept — transient corpus, active corpus untouched", + "accept_proposal writes one JSONL line to a TRANSIENT corpus " + "(copy of active + new chain). The active corpus file bytes " + "are byte-identical pre/post. Provenance on the new entry: " + "adr-0057:discovery_promoted:.", + ) + log = ProposalLog(log_path) + tmp_dir = Path(tempfile.mkdtemp(prefix="learning_loop_demo_")) + transient = tmp_dir / "cognition_chains_v1.jsonl" + if _tg._CORPUS_PATH.exists(): + shutil.copyfile(_tg._CORPUS_PATH, transient) + else: + transient.write_text("", encoding="utf-8") + + active_before = _active_bytes() + transient_lines_before = len(transient.read_text(encoding="utf-8").splitlines()) + + chain_id = accept_proposal( + proposal_id, + log=log, + corpus_path=transient, + review_date="2026-05-18", + operator_note="learning-loop demo (transient corpus only)", + ) + active_after = _active_bytes() + transient_lines_after = len(transient.read_text(encoding="utf-8").splitlines()) + + _say(f" appended chain_id : {chain_id}") + _say(f" transient corpus path : {transient}") + _say(f" transient lines before : {transient_lines_before}") + _say(f" transient lines after : {transient_lines_after}") + _say(f" active corpus byte-eq : {active_before == active_after}") + if active_before != active_after: + raise RuntimeError( + "demo invariant broken: accept_proposal mutated the active corpus" + ) + return SceneResult( + scene="S4_accept_against_transient", + claim=( + "accept_proposal is the sole corpus-write surface. Pointing " + "it at a transient path leaves the active corpus byte-identical." + ), + detail={ + "chain_id": chain_id, + "transient_corpus": str(transient), + "transient_lines_before": transient_lines_before, + "transient_lines_after": transient_lines_after, + "active_corpus_byte_identical": active_before == active_after, + }, + ), transient + + +def _scene5_replay_now_grounded(transient: Path) -> SceneResult: + _print_header( + "S5. Same prompt — now deterministically teaching-grounded", + "With the runtime's corpus path swapped to the transient corpus, " + "the same prompt now returns a teaching-grounded surface " + "containing the operator-accepted chain: " + "thought reveals meaning. Identical bytes for any replay of " + "the same prompt against this corpus state.", + ) + real_path = _tg._CORPUS_PATH + try: + _tg._CORPUS_PATH = transient # type: ignore[assignment] + _tg._corpus_index.cache_clear() + # Fresh runtime to avoid any per-instance state. + rt2 = ChatRuntime() + response = rt2.chat(_DEMO_PROMPT) + finally: + _tg._CORPUS_PATH = real_path # type: ignore[assignment] + _tg._corpus_index.cache_clear() + + surface = response.surface + grounding = response.grounding_source + _say(f" prompt : {_DEMO_PROMPT}") + _say(f" surface : {surface}") + _say(f" grounding_source : {grounding}") + + # Falsifiable assertions for the demo's headline claim. + contains_subject = "thought" in surface.lower() + contains_connective = "reveal" in surface.lower() # humanised + contains_object = "meaning" in surface.lower() + is_teaching_grounded = grounding == "teaching" + + if not (contains_subject and contains_connective and contains_object and is_teaching_grounded): + raise RuntimeError( + f"demo invariant broken: same-prompt surface did not become " + f"teaching-grounded (surface={surface!r}, grounding={grounding!r})" + ) + + return SceneResult( + scene="S5_replay_now_grounded", + claim=( + "The same prompt now produces a deterministic teaching-" + "grounded surface containing the accepted chain's " + "subject / connective / object." + ), + detail={ + "surface": surface, + "grounding_source": grounding, + "contains_subject_thought": contains_subject, + "contains_connective_reveals": contains_connective, + "contains_object_meaning": contains_object, + }, + ) + + +# --------------------------------------------------------------------------- +# Public entry point +# --------------------------------------------------------------------------- + + +def run_demo(*, emit_json: bool = False) -> dict[str, Any]: + """Run all five scenes and return a structured report.""" + global _VERBOSE + _VERBOSE = not emit_json + + active_bytes_before = _active_bytes() + rt = ChatRuntime() + sink = _BufferSink() + rt.attach_discovery_sink(sink) + rt.attach_contemplation(enabled=True) + + with tempfile.TemporaryDirectory() as tmpdir: + log_path = Path(tmpdir) / "demo_proposals.jsonl" + s1, _before_response = _scene1_cold_turn(rt, sink) + s2, candidate_payload = _scene2_discovery_emission(sink) + s3, proposal = _scene3_propose(log_path, candidate_payload["candidate_id"]) + s4, transient = _scene4_accept_against_transient(log_path, proposal.proposal_id) + s5 = _scene5_replay_now_grounded(transient) + + active_bytes_after = _active_bytes() + report = DemoReport( + prompt=_DEMO_PROMPT, + before_surface=s1.detail["surface"], + before_grounding_source=s1.detail["grounding_source"], + after_surface=s5.detail["surface"], + after_grounding_source=s5.detail["grounding_source"], + scenes=(s1, s2, s3, s4, s5), + learning_loop_closed=( + s1.detail["grounding_source"] == "none" + and s5.detail["grounding_source"] == "teaching" + ), + active_corpus_byte_identical=(active_bytes_before == active_bytes_after), + ) + + if _VERBOSE: + _say() + _say("═" * 72) + _say(" BEFORE / AFTER (single deterministic prompt, one accept between)") + _say("═" * 72) + _say(f" prompt : {report.prompt}") + _say(f" before : [{report.before_grounding_source}] {report.before_surface}") + _say(f" after : [{report.after_grounding_source}] {report.after_surface}") + _say() + _say(f" learning_loop_closed : {report.learning_loop_closed}") + _say(f" active corpus byte-identical : {report.active_corpus_byte_identical}") + _say() + + return report.as_dict() + + +__all__ = ["run_demo"] diff --git a/tests/test_learning_loop_demo.py b/tests/test_learning_loop_demo.py new file mode 100644 index 00000000..b047763b --- /dev/null +++ b/tests/test_learning_loop_demo.py @@ -0,0 +1,70 @@ +"""Learning-loop demo — pins the load-bearing before/after claim. + +If any assertion fails, the headline claim ("CORE learned a new chain +from a cold turn and the same prompt is now teaching-grounded with +provenance") no longer holds. +""" + +from __future__ import annotations + +from evals.learning_loop.run_demo import run_demo + + +def test_demo_closes_the_full_loop() -> None: + report = run_demo(emit_json=True) + assert report["learning_loop_closed"] is True + assert report["active_corpus_byte_identical"] is True + assert len(report["scenes"]) == 5 + + +def test_before_is_ungrounded_disclosure() -> None: + report = run_demo(emit_json=True) + assert report["before"]["grounding_source"] == "none" + assert "insufficient grounding" in report["before"]["surface"].lower() + + +def test_after_is_teaching_grounded_with_new_chain_atoms() -> None: + report = run_demo(emit_json=True) + assert report["after"]["grounding_source"] == "teaching" + surface = report["after"]["surface"].lower() + # The accepted chain is (thought, cause, reveals, meaning). + assert "thought" in surface + assert "reveal" in surface # humanised connective + assert "meaning" in surface + assert "teaching-grounded" in surface + + +def test_s1_emits_one_discovery_candidate() -> None: + report = run_demo(emit_json=True) + s1 = report["scenes"][0] + assert s1["scene"] == "S1_cold_turn" + assert s1["detail"]["discovery_candidates_emitted"] >= 1 + + +def test_s3_replay_gate_reports_no_regression() -> None: + report = run_demo(emit_json=True) + s3 = report["scenes"][2] + assert s3["scene"] == "S3_propose_replay_pass" + ev = s3["detail"]["replay_evidence"] + assert ev["replay_equivalent"] is True + assert ev["regressed_metrics"] == [] + assert s3["detail"]["state"] == "pending" + + +def test_s4_active_corpus_byte_identical_after_accept() -> None: + report = run_demo(emit_json=True) + s4 = report["scenes"][3] + assert s4["scene"] == "S4_accept_against_transient" + assert s4["detail"]["active_corpus_byte_identical"] is True + assert s4["detail"]["transient_lines_after"] == s4["detail"]["transient_lines_before"] + 1 + + +def test_same_prompt_drives_before_and_after() -> None: + """The same input string drives both sides of the before/after pair. + Different surfaces emerge from the corpus state change alone, not + from any prompt variation or stochastic sampling.""" + report = run_demo(emit_json=True) + assert report["prompt"] == "Why does thought exist?" + # And the two surfaces are observably different — the loop changed + # the response, not merely the metadata. + assert report["before"]["surface"] != report["after"]["surface"]