portfolio-optimiser-claude/tests/test_simulation_loadbearing.py

"""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
        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).
        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).
        assert MARKER not in bundle_context(bundle)