"""Closed-loop two-run simulation — LOAD-BEARING (K4/R-1; method-spec §11 «Closed loop»). The seam this file keeps alive: the persona's marker crosses from run A to run B's hypothesis prompt VIA the §6 promotion gate — and NEVER crosses without it. Both directions are bound: detach the promotion step from the driver → the marker is absent from run B → red; render the verdict layer into the read-context → the marker leaks around the gated fold → red (the verdict-exclusion invariant, re-proven at simulation level). """ from __future__ import annotations import json from pathlib import Path import pytest from _scripted import ScriptedClient, reply from portfolio_optimiser_claude.ir import load_validator_input from portfolio_optimiser_claude.okf import bundle_context from portfolio_optimiser_claude.persona import load_persona_example from portfolio_optimiser_claude.simulation import ClosedLoopResult, simulate_closed_loop SHARED_BUNDLE = Path(__file__).resolve().parents[1] / "shared" / "examples" / "bygg-energi-mikro" SHARED_ARTIFACT = ( Path(__file__).resolve().parents[1] / "shared" / "skills" / "expert-reviewer" / "references" / "example-verdict.json" ) # The shared artifact's own marker (realiseringsgrad=0.79) — distinct from the # bundle seed's 0.82, so its presence in run B proves the PROMOTION crossing, # not the pre-existing seeding path. MARKER = load_persona_example(SHARED_ARTIFACT).marker TIMESTAMP = "2026-07-17T04:00:00Z" EXPERIMENT = "K4-closed-loop-sim" @pytest.fixture() def bundle(tmp_path: Path) -> Path: import shutil target = tmp_path / "bundle" shutil.copytree(SHARED_BUNDLE, target) return target def _proposal_json(bundle: Path, *, measure_suffix: str = "") -> str: proposal = load_validator_input(bundle).model_dump() if measure_suffix: proposal["measure"] = f"{proposal['measure']}{measure_suffix}" return json.dumps(proposal) def _scripted_client(proposal_json: str) -> ScriptedClient: # One full run: debate round (proposer + approving checker) + generation. return ScriptedClient( replies=[reply("debate reasoning"), reply("VERDICT: APPROVE"), reply(proposal_json)] ) def _simulate( bundle: Path, artifact: Path, inbox: Path ) -> tuple[ClosedLoopResult, ScriptedClient, ScriptedClient]: # Run A proposes a DISTINCT candidate (measure variant): the persona's # verdict then gets its own §4.2 id. Judging the bundle seed's own # candidate would mint the seed's id, and the store's first-write-wins # would silently shadow the promoted rationale behind the seed (the known # C-F5 feature-keying limitation — deferred to C3.2, not worked around # in src). client_a = _scripted_client(_proposal_json(bundle, measure_suffix="-simulert-variant")) client_b = _scripted_client(_proposal_json(bundle)) result = simulate_closed_loop( bundle, artifact, inbox, client_a=client_a, client_b=client_b, timestamp=TIMESTAMP, experiment=EXPERIMENT, ) return result, client_a, client_b def _rejecting_artifact(tmp_path: Path, marker: str) -> Path: path = tmp_path / "rejecting-example-verdict.json" path.write_text( json.dumps( { "decision": "rejected", "marker": marker, "rationale": f"Avvist: driftstimene er overestimert ({marker}).", } ), encoding="utf-8", ) return path class TestMarkerCrossesViaPromotion: """§11 «Closed loop», direction 1: WITH promotion the marker MUST cross.""" def test_the_persona_marker_reaches_run_b_via_promotion( self, bundle: Path, tmp_path: Path ) -> None: result, _, client_b = _simulate(bundle, SHARED_ARTIFACT, tmp_path / "inbox") # The marker did not exist anywhere before the persona judged run A. assert MARKER not in result.run_a.composed.context # The gate promoted the approved verdict into the bundle... assert result.promoted_path is not None and result.promoted_path.is_file() assert result.promotion_refusal is None # ...and run B's FRESH store picked it up through seeding into the fold. assert result.run_b.composed.seeded == result.run_a.composed.seeded + 1 assert MARKER in result.run_b.composed.context # Detach proof 1 anchor: the marker reaches run B's actual prompts. assert any(MARKER in prompt for prompt in client_b.prompts("proposer")) def test_both_runs_complete_against_the_scripted_clients( self, bundle: Path, tmp_path: Path ) -> None: result, _, _ = _simulate(bundle, SHARED_ARTIFACT, tmp_path / "inbox") assert result.run_a.result.validator_decision == "validated" assert result.run_b.result.validator_decision == "validated" # The persona judged run A's candidate — the verdict carries its rationale. assert MARKER in result.verdict.rationale class TestMarkerNeverCrossesWithoutPromotion: """§11 «Closed loop», direction 2: WITHOUT promotion the marker NEVER crosses. The rejected persona verdict IS authored into the inbox — if run B read the inbox (or any channel other than the promoted wiki layer), the marker would cross anyway and this control goes red. """ def test_a_rejected_verdict_is_refused_and_the_marker_stays_out( self, bundle: Path, tmp_path: Path ) -> None: marker = "realiseringsgrad=0.11" artifact = _rejecting_artifact(tmp_path, marker) result, _, client_b = _simulate(bundle, artifact, tmp_path / "inbox") assert result.promoted_path is None assert result.promotion_refusal is not None assert "rejected" in result.promotion_refusal # Positive controls: the docstring's claim that the marker IS authored into the # artifact the run reads, asserted rather than stated — and prompts really were # issued, since `all()` over an empty list is vacuously True. Without both, this # control would pass for a marker that was never written anywhere. assert marker in artifact.read_text(encoding="utf-8") assert client_b.prompts("proposer") assert marker not in result.run_b.composed.context assert all(marker not in prompt for prompt in client_b.prompts("proposer")) def test_a_refused_promotion_leaves_the_bundle_untouched( self, bundle: Path, tmp_path: Path ) -> None: before = {p.name: p.read_bytes() for p in bundle.iterdir()} _simulate( bundle, _rejecting_artifact(tmp_path, "realiseringsgrad=0.11"), tmp_path / "inbox" ) after = {p.name: p.read_bytes() for p in bundle.iterdir()} assert after == before class TestGatedFoldIsTheOnlyChannel: """LOAD-BEARING (§11): the marker crosses ONLY via the gated experience fold. Detach proof 2 anchor: render the verdict layer into the read-context and the marker leaks into the base context around the fold — red. """ def test_the_verdict_layer_is_never_rendered_into_either_context( self, bundle: Path, tmp_path: Path ) -> None: result, _, _ = _simulate(bundle, SHARED_ARTIFACT, tmp_path / "inbox") assert MARKER in result.run_b.composed.context # crossing happened... # ...but NO verdict-typed concept is rendered into the read-context — # a `## verdict` section in either run's context means the seed's (or # a promoted file's) body leaked around the gated fold (§3 Step 1). # Positive control: `## `-typed sections ARE rendered into both contexts, so the # two negatives measure a GATED type and not a context without sections at all. assert "## project:" in result.run_a.composed.context assert "## project:" in result.run_b.composed.context assert "## verdict" not in result.run_a.composed.context assert "## verdict" not in result.run_b.composed.context # Belt and braces: the promoted marker itself never surfaces via # rendering (promoted files are frontmatter-only by design, §6). # Positive control: the marker IS live on disk in the promoted file, so the # absence below measures the RENDERING boundary, not a marker never written. assert result.promoted_path is not None assert MARKER in result.promoted_path.read_text(encoding="utf-8") assert MARKER not in bundle_context(bundle)