portfolio-optimiser/src/portfolio_optimiser/simulation.py

"""Offline simulation of the full agentic loop — the end-to-end method proof (replaces målbilde
§11.8's real-model run).

**Operator decision (pragmatic, cost-driven):** MAF is NOT run against a real model — neither
Azure/Foundry nor local Ollama — because paying for API runs across both repos (MAF + the
Claude-SDK sibling) is too costly privately. This module is the primary proof instead: it drives
``run_project`` with a **scripted** synthetic chat client (no network, no model) and demonstrates
that the loop's dataflow closes end to end across two runs separated by a promotion:

    context -> hypothesis -> maker/checker debate -> deterministic validator -> persona verdict
    -> PROMOTION into the OKF wiki -> the next run's hypothesis is informed by it.

**What this proves:** the plumbing, the deterministic spine, and that the learning loop closes —
a verdict approved in Run A reaches Run B's hypothesis prompt purely through the file-backed wiki.
**What it does NOT prove (honesty, målbilde §1):** that a live LLM would *produce* the proposal or
the verdict unprompted — those are scripted stand-ins for the swarm and the expert persona. The
genuine model-behaviour comparison lives on the Claude-SDK side (a minimal API run). The scripted
client is MAF-side scaffolding; it is NOT part of the framework-neutral ``shared/`` core.
"""

from __future__ import annotations

import json
import logging
import shutil
from collections.abc import Awaitable, Callable, Mapping, Sequence
from dataclasses import dataclass
from pathlib import Path
from typing import Any, cast

from agent_framework import (
    BaseChatClient,
    ChatResponse,
    ChatResponseUpdate,
    Message,
    ResponseStream,
    UsageDetails,
)
from agent_framework_openai import OpenAIChatCompletionClient

from portfolio_optimiser import okf
from portfolio_optimiser.explore import (
    HYPOTHESIS_MARKER,
    HYPOTHESISER_ROLE,
    MANAGER_ROLE,
    NAVIGATOR_ROLE,
    ExplorationContract,
    ExplorationResult,
    ExplorationTrace,
    explore,
)
from portfolio_optimiser.ir import AffectedItem, CostBaseline, CostBaselineLine, SavingsProposal
from portfolio_optimiser.persona import load_persona_example
from portfolio_optimiser.run import RunResult, run_project
from portfolio_optimiser.shared_root import shared_root
from portfolio_optimiser.tracing import configure_tracing, tracing_notice
from portfolio_optimiser.validator import ValidatedProposal
from portfolio_optimiser.verdicts import (
    Verdict,
    VerdictStore,
    bundle_candidate_features,
    promote_verdict,
    seed_store_from_bundle,
    write_verdict,
)

_PROJECT_ID = "BYGG-KONTOR-NORD"
# The delivered domain bundle (commons `002f000`) — the project the demo runs on stage. Kept beside
# the reserve's id rather than replacing it: the reserve stays reachable as the abort path.
_VEGLYS_PROJECT_ID = "VEGLYS-FV-SOER"

# --- P4 pkt. 2: demo stderr discipline -----------------------------------------------------------
# Measured 2026-08-09, the demo wrote six stderr lines: two ``ExperimentalWarning``s from
# ``agent_framework``, two round-cap notices from the orchestrator, a blank line and the
# ``arbeidskopi:`` path. The last two are ours and stay (the path is the one deliberately
# non-deterministic value, which is exactly why it is on stderr and not on stdout).
#
# **Damped here: the round-cap notices only.** They are emitted DURING the demo's own run, by an
# event the demo deliberately provokes — the maker/checker debate is configured to run to its cap.
#
# **NOT damped: the two ``ExperimentalWarning``s — they are pinned in pkt. 3 instead.** Measured,
# not assumed: they fire while ``portfolio_optimiser/__init__.py`` imports ``run``, which imports
# ``agent_framework`` — always before this module's own import block, under BOTH invocation forms.
# Silencing them would therefore mean putting a warnings filter inside the library package, i.e.
# letting this framework decide what MAF is allowed to tell every consumer that imports it. That is
# a library-behaviour change for stderr nobody projects, days before a freeze. A wrapper that
# muted them only behind the console script was rejected for a second reason: it would make
# ``uv run portfolio-optimiser-demo`` and ``uv run python -m portfolio_optimiser.simulation`` write
# different stderr, and a byte-fasit would then pin the command rather than the program.
#
# The damping that IS here is NARROW by construction, which is the load-bearing part: a different
# orchestration warning still reaches stderr and trips the pkt. 3 pin. A filter that could only
# ever say "drop" would leave a pin that can no longer fail for the reason it exists.

#: The logger that emits the round-cap notice, read out of MAF's source (``logger.warning`` in
#: ``_base_group_chat_orchestrator``), not guessed. Logger filters apply only to the logger the
#: record was logged THROUGH — an ancestor's filters are never consulted — so this must be the
#: emitting module's own name. If MAF moves the call, the notice simply reappears on stderr and the
#: pkt. 3 pin says so; the failure mode is visible, not silent.
ROUND_CAP_LOGGER = "agent_framework_orchestrations._base_group_chat_orchestrator"


class _ExpectedRoundCapFilter(logging.Filter):
    """Drops ONLY the "round limit reached" notice, which the demo reaches by design (the
    maker/checker debate runs to its cap). Keyed on the event, not on the configured number: the
    cap is a demo setting, while the notice is the thing we have decided is expected."""

    def filter(self, record: logging.LogRecord) -> bool:
        message = record.getMessage()
        return not ("reached max_rounds=" in message and "forcing completion" in message)


def quiet_expected_round_cap_notice() -> logging.Filter:
    """Install the round-cap filter on the emitting logger and return it (so a caller — a test —
    can remove it again). Runtime state, installed by ``main()``: importing this module must not
    reconfigure logging for a library consumer."""
    installed = _ExpectedRoundCapFilter()
    logging.getLogger(ROUND_CAP_LOGGER).addFilter(installed)
    return installed


# --- Step 7, the LONG loop: what an expert drops into the inbox between the two runs -------------
# A SECOND marker, deliberately distinct from the persona's (Step 8, promotion). The two mechanisms
# both end in Run B's hypothesis prompt, so a single shared marker would let either path carry it
# alone — and detaching one seam would then go unnoticed by both load-bearing tests. Absence from
# the bundle is asserted by tests/test_step7_demo_inbox_loadbearing.py, not assumed here.
_INBOX_MARKER = "realiseringsgrad=0.66"
# An explicit sentinel id, never a minted one: ``_mint_id`` hashes the candidate FEATURES, so a
# minted id would collide with the promoted verdict's — and ``VerdictStore.add`` is first-write-wins
# per id, which would silently drop whichever arrived second.
_INBOX_VERDICT_ID = "STEG7-EKSPERT-DRIFTSNOTAT"
_INBOX_RATIONALE = (
    "Ettersendt driftsnotat fra energiingeniør, lagt i innboksen etter kjøringen: målingene "
    f"fra fyringssesongen viser at realiseringen faller til {_INBOX_MARKER} når "
    "tilstedeværelsesstyringen står på fabrikkinnstilling. Godkjent, med den korreksjonen."
)
_INBOX_DIR_NAME = "verdict-innboks"


def _default_bundle_dir() -> Path:
    """The demo bundle under the shared core, resolved at CALL time via ``shared_root()`` (env
    ``PORTFOLIO_SHARED_ROOT`` re-points it — the S4 extraction seam)."""
    return shared_root() / "examples" / "bygg-energi-mikro"


class ScriptedCandidateError(LookupError):
    """No single registered candidate matches the prompt (none, or more than one)."""


@dataclass(frozen=True)
class ScriptedCandidate:
    """One project's scripted proposer script, expressed as DATA.

    Adding a project to the walkthrough is a registry entry — never a second hand-written selector.
    ``overclaimed`` is answered until the deterministic validator's rejection comes back in the
    prompt (recognised by ``flip_key``), after which ``corrected`` is answered. Both are raw reply
    strings, exactly what a model would have returned.

    ``flip_key`` must be ABSENT from the bundle this candidate is demoed against, or attempt 1's
    prompt already contains it and the proposer 'corrects' before anything was falsified.
    """

    project_id: str
    overclaimed: str
    corrected: str
    flip_key: str


def scripted_proposer(
    candidates: Sequence[ScriptedCandidate],
) -> Callable[[str, str], str]:
    """Build the scripted proposer over a candidate registry — a ``reply_selector`` for the canonical
    client, keyed on the PROJECT the prompt names.

    **Why the project id and not the cost code or the measure name** (measured, not assumed — the
    demo-week plan §6 flagged this as unverified): two prompt shapes reach this selector. The debate
    prompt (``run.py``) carries the whole bundle context; the generation prompt
    (``generate._build_messages``) carries ``Project: {id} - {name}`` plus, as its context, the
    *debate output* — which is this selector's own earlier reply. So the cost code and measure name
    are present in the generation prompt only because the script put them there; keying on them
    would key the script on itself. The project id is the one identifier BOTH shapes carry and the
    FRAMEWORK stamps.

    Stateless — no per-turn counter — so the debate turns and the generation attempts share it.

    Anything other than exactly one match raises ``ScriptedCandidateError``. Validation, never
    repair: a default reply would let an unregistered project be answered with another project's
    numbers, which on screen is indistinguishable from a correct run. An ambiguous blob (a bundle
    context that names a sibling project) is a DATA problem, and it must surface at the rehearsal
    rather than be silently decided by registry order.
    """

    def select(prompt: str, _role: str) -> str:
        matches = [c for c in candidates if c.project_id in prompt]
        if len(matches) != 1:
            found = ", ".join(c.project_id for c in matches) or "-"
            raise ScriptedCandidateError(
                f"no scripted candidate uniquely matches the prompt (ambiguous or unknown); "
                f"matched: {found}"
            )
        candidate = matches[0]
        return candidate.corrected if candidate.flip_key in prompt else candidate.overclaimed

    return select


# Two SavingsProposals for BYGG-KONTOR-NORD: total = 300000 x 1.0, so the degenerate Monte Carlo
# P90 = 0.30 x 300000 = 90000 (no `assumptions`). The OVERCLAIMED one asks for 250000 — parseable,
# and internally consistent, but above P90, so the DETERMINISTIC validator falsifies it. The
# corrected one claims 30000 <= 90000 and validates. Together they drive Step 5 (informed
# refinement): the proposer is scripted, but the rejection that turns proposal 1 into proposal 2 is
# genuinely computed by the validator, not scripted.
#
# 250000 — the overclaimed figure the validator's rejection reason carries and
# ``generate._build_messages`` appends to the NEXT attempt's prompt — is verified ABSENT from the
# demo bundle (``test_content_keyed_script_loadbearing``), so the correction is caused by the
# falsification travelling back, never by the proposer simply being asked twice.
_CANDIDATES: tuple[ScriptedCandidate, ...] = (
    ScriptedCandidate(
        project_id=_PROJECT_ID,
        overclaimed=(
            '{"measure":"LED-retrofit av kontorbelysning","affected_items":'
            '[{"code":"ENERGI-TOTAL-EL","quantity":300000,"unit_cost":1.0}],'
            '"claimed_saving_nok":250000}'
        ),
        corrected=(
            '{"measure":"LED-retrofit av kontorbelysning","affected_items":'
            '[{"code":"ENERGI-TOTAL-EL","quantity":300000,"unit_cost":1.0}],'
            '"claimed_saving_nok":30000}'
        ),
        flip_key="250000",
    ),
    # VEGLYS-FV-SOER — the DELIVERED bundle (commons `002f000`), and the one on stage. Every number
    # below is written FROM `shared/examples/veglys-fv-soer/validator-input.json`, never beside it
    # (plan P3 b): this is `baseline_from_scripted_candidate`'s direction REVERSED — the domain team
    # shipped the cost lines, so the register copies them, and the derivation helper is not used on
    # this path (`main` reads the delivered `cost-baseline.json` off the bundle instead).
    #
    # The corrected reply IS the delivered IR projection verbatim: affected_items = the WHOLE
    # portfolio's annual energy cost (their decision — scoping it to the 2 500 touched points would
    # make the RIGHT proposal 38,6 % of its own baseline and the 30 % cap would fell it on stage),
    # claimed_saving_nok = 445500, assumptions = the declared energy-price band. The overclaimed one
    # differs in exactly ONE field.
    #
    # 2100000 is chosen OUTSIDE commons' number inventory and above BOTH thresholds: measured absent
    # from the bundle (so the correction is caused by the falsification travelling back), and above
    # the degenerate cap 0.30 x 4386150 = 1315845 as well as the banded P90. 600000/900000 would have
    # CLEARED the gate — the REJECTED line would never appear and demo-criterion 2 would fail quietly.
    ScriptedCandidate(
        project_id=_VEGLYS_PROJECT_ID,
        overclaimed=(
            '{"measure":"LED-utskifting av 2 500 eldre HPS-armaturer (114 W -> 70 W)",'
            '"affected_items":[{"code":"ENERGI-VEGLYS-EL","quantity":4386150,"unit_cost":1.0}],'
            '"claimed_saving_nok":2100000,'
            '"assumptions":{"ENERGI-VEGLYS-EL":[0.70,1.40]}}'
        ),
        corrected=(
            '{"measure":"LED-utskifting av 2 500 eldre HPS-armaturer (114 W -> 70 W)",'
            '"affected_items":[{"code":"ENERGI-VEGLYS-EL","quantity":4386150,"unit_cost":1.0}],'
            '"claimed_saving_nok":445500,'
            '"assumptions":{"ENERGI-VEGLYS-EL":[0.70,1.40]}}'
        ),
        flip_key="2100000",
    ),
)

_proposer_reply = scripted_proposer(_CANDIDATES)

# The filename ``okf``'s loaders default to. Kept local rather than reaching into ``okf``'s private
# constant; the coupling is measured, not assumed — a drifted name makes the materialized bundle
# un-anchored, which ``test_anchored_reserve_loadbearing`` turns red.
_COST_BASELINE_FILE = "cost-baseline.json"
_ANCHORED_DIR_NAME = "forankret-reserve"
_RESERVE_PROVENANCE = "tallene er syntetiske — avledet av demo-manuset, ikke levert av et fagmiljø"
# The GO-day honesty sentence (plan P4 pkt. 4). It claims exactly what the bundle's own `_note`
# documents — the cost line is derived from published sources — and nothing about the realization
# rate, which the seed verdict itself marks as BORROWED from lighting-programme literature.
_VEGLYS_PROVENANCE = (
    "tallene er levert i kunnskapsbasen — utledet av fagkilder (Håndbok V124, NMFV), "
    "ikke av demo-manuset"
)


def _delivered_bundle_dir() -> Path:
    """The delivered VEGLYS bundle, resolved at CALL time via ``shared_root()`` (same seam as
    ``_default_bundle_dir``). It ships its own ``cost-baseline.json``, so the demo is anchored
    WITHOUT ``materialize_anchored_bundle`` — that helper exists for the reserve, whose numbers
    cannot be given a baseline in place (pull-only subtree + byte-unchanged goldens)."""
    return shared_root() / "examples" / "veglys-fv-soer"


def baseline_from_scripted_candidate(candidate: ScriptedCandidate) -> CostBaseline:
    """Derive a project's cost baseline FROM the scripted register's own cost lines (P4 pkt. 0).

    The reserve bundle's numbers are synthetic, so the script is the only ground truth there is;
    deriving in code rather than typing the same numbers into a second file is what keeps the two
    from drifting apart. **On GO day the direction reverses** (plan P3 b): the register is written
    FROM the delivered ``cost-baseline.json``, and this function is not used.

    Both scripted replies must state the SAME cost lines, or ``ValueError``. Validation, never
    repair: were they to differ, hypothesis #1 would be falsified by the reconciliation stage
    instead of by the P90 stage, and the demo's REJECTED line would come from another mechanism
    than the one it narrates — visible on screen as the same line either way.
    """
    lines = {
        item.code: CostBaselineLine(quantity=item.quantity, unit_cost=item.unit_cost)
        for item in (
            AffectedItem.model_validate(raw)
            for raw in json.loads(candidate.corrected)["affected_items"]
        )
    }
    overclaimed = {
        raw["code"]: (raw["quantity"], raw["unit_cost"])
        for raw in json.loads(candidate.overclaimed)["affected_items"]
    }
    if overclaimed != {code: (line.quantity, line.unit_cost) for code, line in lines.items()}:
        raise ValueError(
            f"scripted candidate {candidate.project_id} states different cost lines in its two "
            "replies; a baseline derived from one of them would falsify the other at stage 0"
        )
    return CostBaseline(project_id=candidate.project_id, items=lines)


def _reserve_baseline() -> CostBaseline:
    """The reserve bundle's baseline: the registry entry for the reserve's project, never
    ``_CANDIDATES[0]`` — the registry is a set of DATA entries whose order carries no meaning, and
    an index would silently anchor the demo to another project once a second entry lands."""
    (candidate,) = [c for c in _CANDIDATES if c.project_id == _PROJECT_ID]
    return baseline_from_scripted_candidate(candidate)


def materialize_anchored_bundle(
    dest: str | Path,
    *,
    source: str | Path | None = None,
    baseline: CostBaseline | None = None,
) -> Path:
    """Copy the reserve bundle and ADD the ``cost-baseline.json`` it cannot be given in place — the
    copy-and-extend variant that anchors the deterministic gate (S4.0 stage 0) for the demo.

    ``shared/`` is a pull-only subtree and demo criterion 8 requires the commons-owned goldens
    byte-unchanged, so the reserve can never ship the file itself. That is a PLACEMENT constraint,
    not an impossibility: the run already reads the baseline from whichever bundle directory it is
    handed (``run.py`` -> ``okf.load_optional_cost_baseline``), so an extended copy outside
    ``shared/`` is anchored by exactly the same seam a delivered bundle would use.

    ``baseline`` defaults to the one derived from the scripted register; a caller passes its own to
    model a DELIVERED baseline that disagrees with the script (the 10 % test).
    """
    src = Path(source) if source is not None else _default_bundle_dir()
    out = Path(dest)
    shutil.copytree(src, out)
    resolved = baseline if baseline is not None else _reserve_baseline()
    payload = {
        "_note": (
            "SYNTHETIC cost baseline, materialized for the demo (P4 pkt. 0) — NOT delivered data. "
            "Derived from the scripted register in portfolio_optimiser.simulation so the two "
            "cannot drift. The source bundle is never modified."
        ),
        **resolved.model_dump(),
    }
    (out / _COST_BASELINE_FILE).write_text(
        json.dumps(payload, indent=2, ensure_ascii=False) + "\n", encoding="utf-8"
    )
    return out


# The checker's debate turn ends with the gate marker the run parses (run._checker_verdict).
_CHECKER_APPROVE = "Tallene er innenfor feasibelt område og resonnementet holder. VERDICT: APPROVE"

# The persona's verdict is sourced from the shared expert-reviewer skill (``load_persona_example``),
# NOT inlined here — that de-stubs the persona and makes the shared artifact genuinely consumed. Its
# marker (a realization rate ABSENT from the bundle — the seed is 0.82) is the payload we trace from
# Run A's persona judgement, through promotion, into Run B's hypothesis prompt.


class ScriptedChatClient(OpenAIChatCompletionClient):
    """The ONE canonical network-free scripted chat client (S2.5 consolidation): a single
    ``_inner_get_response`` body shared by the simulation's fixed-reply client AND conftest's three
    test doubles (which subclass it). Parametrized by a ``reply_selector`` over
    ``(prompt_blob, role)`` plus an optional ``sink`` recording every prompt — the four
    previously-divergent ``_inner_get_response`` bodies collapse to this one.

    Subclasses the LAYERED ``OpenAIChatCompletionClient`` (not the minimal ``BaseChatClient``) so the
    always-attached ``BudgetMiddleware`` is not silently no-op'd (verified). Construction is offline
    (loopback ``base_url`` + dummy key); ``_inner_get_response`` intercepts before any HTTP.

    Back-compat constructors are preserved (divergent PUBLIC surfaces the external test call-sites
    depend on): ``ScriptedChatClient(reply, sink)`` (POSITIONAL — used by ``scripted_factory``) is
    sugar for a constant selector; the ``call_count`` attribute + ``model``/OTEL ``"synthetic"`` are
    always present; subclasses pass ``reply_selector=`` / ``default_reply=`` for scripted-list,
    prompt-scan, or record-only behaviour."""

    OTEL_PROVIDER_NAME = "synthetic"

    def __init__(
        self,
        reply: str | None = None,
        sink: list[str] | None = None,
        *,
        reply_selector: Callable[[str, str], str] | None = None,
        role: str = "",
        default_reply: str = "ok",
        tokens_per_reply: int = 8,
    ) -> None:
        super().__init__(model="synthetic", api_key="synthetic", base_url="http://127.0.0.1:9/v1")
        self._sink = sink
        self._role = role
        self._default = default_reply
        # The reply-selector over (prompt_blob, role). A positional ``reply`` is sugar for a constant
        # selector (scripted_factory back-compat); with neither, the constant is ``default_reply``.
        if reply_selector is not None:
            self._select: Callable[[str, str], str] = reply_selector
        elif reply is not None:
            self._select = lambda _prompt, _role: reply
        else:
            self._select = lambda _prompt, _role: self._default
        self._tokens = tokens_per_reply
        self.call_count = 0

    def _inner_get_response(
        self,
        *,
        messages: Sequence[Message],
        options: Mapping[str, Any],
        stream: bool = False,
        **kwargs: Any,
    ) -> Awaitable[ChatResponse] | ResponseStream[ChatResponseUpdate, ChatResponse]:
        blob = " ".join(getattr(m, "text", "") or "" for m in messages)
        if self._sink is not None:
            self._sink.append(blob)
        self.call_count += 1
        reply = self._select(blob, self._role)
        usage = UsageDetails(total_token_count=self._tokens)
        if stream:

            async def _agen() -> Any:
                # The framework accepts a {"type": "text", ...} dict here (its types under-specify it).
                yield ChatResponseUpdate(
                    role="assistant",
                    contents=[{"type": "text", "text": reply}],  # type: ignore[list-item]
                )

            return self._build_response_stream(_agen())

        async def _coro() -> ChatResponse:
            return ChatResponse(
                messages=[Message(role="assistant", contents=[reply])],
                response_id="synthetic",
                usage_details=usage,
            )

        return _coro()


def scripted_factory(
    replies: Mapping[str, str | Callable[[str, str], str]], sink: list[str]
) -> Callable[[str], BaseChatClient]:
    """A role-keyed client factory: ``factory("proposer")`` and ``factory("checker")`` each return a
    fresh ``ScriptedChatClient`` with that role's reply, all sharing ONE ``sink``. MAF stamps the
    proposer/checker identity from the agent name, so role-keyed stateless replies suffice (no
    per-turn counter); the shared ``sink`` spans the debate turns and the generation call.

    A role's value may be a constant reply OR a ``reply_selector`` over ``(prompt_blob, role)`` —
    the canonical client's existing seam, passed straight through. That is what lets a role answer
    DIFFERENTLY on a later attempt (Step 5: the proposer corrects once the validator's rejection
    comes back in the prompt) without a per-turn counter and without a second scripted body."""

    def factory(role: str) -> BaseChatClient:
        reply = replies[role]
        if callable(reply):
            return ScriptedChatClient(sink=sink, role=role, reply_selector=reply)
        return ScriptedChatClient(reply, sink, role=role)

    return factory


@dataclass(frozen=True)
class LearningSimulationResult:
    """The trace of one two-run learning simulation, carrying BOTH feedback timescales.

    ``marker_in_run_b_prompt`` true while ``marker_in_run_a_prompt`` false is the Step-8 loop: the
    persona knowledge approved in Run A reached Run B's hypothesis via promotion into the wiki.
    ``inbox_marker_*`` is the same shape for the Step-7 LONG loop: a verdict file an expert dropped
    into a folder after Run A, merged into Run B's store before the fold. The two are kept apart on
    purpose — see ``tests/test_step7_demo_inbox_loadbearing.py``."""

    run_a: RunResult
    run_b: RunResult
    promoted_path: Path
    marker: str
    marker_in_run_a_prompt: bool
    marker_in_run_b_prompt: bool
    inbox_path: Path
    inbox_marker: str
    inbox_marker_in_run_a_prompt: bool
    inbox_marker_in_run_b_prompt: bool
    run_a_generation_prompts: list[str]
    run_b_generation_prompts: list[str]


def _generation_prompts(sink: list[str]) -> list[str]:
    """The generation-call prompts (``generate._build_messages`` embeds 'SavingsProposal'), isolated
    from the debate-round prompts also captured in the shared sink."""
    return [p for p in sink if "SavingsProposal" in p]


async def simulate_learning_loop(
    bundle_dir: str,
    work_dir: str,
    *,
    project_id: str = _PROJECT_ID,
    persona_rationale: str | None = None,
    marker: str | None = None,
    inbox_rationale: str = _INBOX_RATIONALE,
    inbox_marker: str = _INBOX_MARKER,
    timestamp: str = "2026-06-30",
    max_rounds: int = 3,
) -> LearningSimulationResult:
    """Run the loop twice on a throwaway COPY of the bundle (the shared fixture is never mutated),
    with a promotion in between, and trace whether the persona's approved knowledge crosses runs.

    ``bundle_dir`` and ``project_id`` are BOTH arguments, so pointing the walkthrough at new content
    is a call-site change plus a ``_CANDIDATES`` entry — no edit to this module's logic. The id must
    match the bundle's own IR projection (``run._project_from_bundle`` raises on a mismatch) and must
    have a registered scripted candidate.

    The persona's verdict (decision + rationale + traced ``marker``) defaults to the shared
    expert-reviewer skill's canonical example (``load_persona_example``), read at CALL time — so the
    simulation is genuinely artifact-driven, not inlined. Callers may override ``marker`` /
    ``persona_rationale`` for a control.

    Run A: a fresh (empty) wiki -> an uninformed hypothesis; the persona approves with NEW realization
    knowledge (``marker`` in ``persona_rationale``). ``promote_verdict`` lifts that verdict into the
    wiki; ``seed_store_from_bundle`` re-reads the wiki; Run B's Step-1 ExpeL fold then carries the
    marker into its hypothesis prompt. The two runs use SEPARATE sinks so each prompt set is
    inspected independently."""
    example = load_persona_example()
    if marker is None:
        marker = example.marker
    if persona_rationale is None:
        persona_rationale = example.rationale
    if marker not in persona_rationale:
        raise ValueError("marker must be a substring of persona_rationale (the carried payload)")
    if inbox_marker not in inbox_rationale:
        raise ValueError(
            "inbox_marker must be a substring of inbox_rationale (the carried payload)"
        )
    if inbox_marker == marker:
        raise ValueError(
            "inbox_marker must differ from marker: the Step-7 inbox and the Step-8 promotion are "
            "two mechanisms that both end in Run B's prompt, so a shared marker would let either "
            "path carry it alone and make both load-bearing tests vacuous"
        )

    copy = Path(work_dir) / "bundle"
    shutil.copytree(bundle_dir, copy)
    copy_s = str(copy)
    replies: dict[str, str | Callable[[str, str], str]] = {
        "proposer": _proposer_reply,
        "checker": _CHECKER_APPROVE,
    }
    verdict_input = {"decision": example.decision, "rationale": persona_rationale}

    # Run A — empty wiki isolates the persona's NEW knowledge.
    sink_a: list[str] = []
    run_a = cast(
        RunResult,  # the sim only drives full runs; never dry-run (S4.2 widened run_project)
        await run_project(
            project_id,
            "local",
            docs_dir=copy_s,
            bundle_dir=copy_s,
            verdict_input=verdict_input,
            store=VerdictStore(verdicts=[]),
            client_factory=scripted_factory(replies, sink_a),
            max_rounds=max_rounds,
        ),
    )

    # Gate-promote the persona verdict from the raw output layer into the OKF wiki (Steg 8).
    promoted_path = promote_verdict(
        copy_s,
        run_a.verdict,
        approver="ekspert-persona (sim)",
        experiment="sim-run-A",
        timestamp=timestamp,
    )

    # Steg 7, the LONG loop: an expert drops a verdict FILE into an inbox AFTER Run A. This is the
    # OTHER timescale — no promotion gate, no wiki, just a folder the next run reads. The inbox sits
    # beside the bundle copy, never inside it: a verdict file within the bundle would reach Run B as
    # navigable context, which is a different mechanism wearing this one's label. ``write_verdict``
    # is the same public primitive a human expert would use (målbilde §3: the system READS, the
    # expert WRITES) — which is why ``run_project`` is never handed the writing job.
    inbox = Path(work_dir) / _INBOX_DIR_NAME
    inbox_path = write_verdict(
        str(inbox),
        Verdict(
            id=_INBOX_VERDICT_ID,
            proposal_features=bundle_candidate_features(copy_s),
            decision="approved",
            rationale=inbox_rationale,
        ),
    )

    # Re-seed the wiki: the promoted verdict is now navigable and folds into the next run.
    store_b = seed_store_from_bundle(copy_s)

    # Run B — a separate, later run reads the updated wiki.
    sink_b: list[str] = []
    run_b = cast(
        RunResult,  # the sim only drives full runs; never dry-run (S4.2 widened run_project)
        await run_project(
            project_id,
            "local",
            docs_dir=copy_s,
            bundle_dir=copy_s,
            verdict_input=verdict_input,
            store=store_b,
            verdict_dir=str(inbox),
            client_factory=scripted_factory(replies, sink_b),
            max_rounds=max_rounds,
        ),
    )

    gen_a = _generation_prompts(sink_a)
    gen_b = _generation_prompts(sink_b)
    return LearningSimulationResult(
        run_a=run_a,
        run_b=run_b,
        promoted_path=promoted_path,
        marker=marker,
        marker_in_run_a_prompt=any(marker in p for p in gen_a),
        marker_in_run_b_prompt=any(marker in p for p in gen_b),
        inbox_path=inbox_path,
        inbox_marker=inbox_marker,
        inbox_marker_in_run_a_prompt=any(inbox_marker in p for p in gen_a),
        inbox_marker_in_run_b_prompt=any(inbox_marker in p for p in gen_b),
        run_a_generation_prompts=gen_a,
        run_b_generation_prompts=gen_b,
    )


#: The U4 walkthrough's prompt and the direction its scripted hypothesiser commits to. The LABEL
#: must be absent from the knowledge base — checked at call time, refused rather than assumed —
#: because a label the base already states would reach the hypothesis prompt as ordinary navigated
#: context, and the scenario would demonstrate nothing (``simulate_learning_loop``'s two-marker
#: guard, in the one form this scenario can go vacuous).
_EXPLORE_PROMPT = "Finn den rimeligste besparelsen som kan testes i denne kunnskapsbasen."
_EXPLORE_LABEL = "styring etter tilstedevaerelse"
_EXPLORE_RATIONALE = (
    "kunnskapsbasen beskriver armaturer som staar paa hele driftsdoegnet, saa en styringsgevinst "
    "kan testes mot de samme kostlinjene som selve retrofitten"
)
#: The walkthrough's stated bounds. Written out rather than defaulted for the reason
#: ``ExplorationContract`` has no defaults at all: the demo is the one place a reader looks to see
#: what bounding an exploration actually looks like.
_EXPLORE_CONTRACT = ExplorationContract(
    max_rounds=4,
    max_tokens=100_000,
    max_stall_count=1,
    max_reset_count=1,
    max_plan_revisions=0,
    enable_plan_review=False,
)


def _exploration_manager_reply(ledgers: Sequence[str]) -> Callable[[str, str], str]:
    """Route a manager prompt to its scripted reply by STAGE, consuming ``ledgers`` in order.

    Keyed on the orchestrator's own prompt text rather than on the project id (the
    ``scripted_proposer`` rule), and that is not an exception to it: the manager is asked five
    DIFFERENT questions — facts, plan, progress ledger, replan, final answer — and which one it is
    being asked is the only thing a reply can be keyed on. The project id is constant across all
    five and would select nothing."""
    remaining = list(ledgers)

    def _select(blob: str, _role: str) -> str:
        if "provide the final answer" in blob:
            return "FINAL: utforskningen er ferdig."
        if "pure JSON format" in blob:
            return remaining.pop(0) if remaining else _exploration_ledger(satisfied=True)
        if "went wrong on this last run" in blob:
            return "PLAN-UPDATE: revidert plan."
        if "rewrite the following fact sheet" in blob:
            return "FACTS-UPDATE: reviderte fakta."
        if "bullet-point plan" in blob:
            return "PLAN: - la hypotesiseren forme en retning"
        if "pre-survey" in blob:
            return "FACTS: kunnskapsbasen er forankret."
        return "{}"

    return _select


def _exploration_ledger(*, satisfied: bool) -> str:
    """One progress ledger naming a REAL participant. The name is what makes the run non-vacuous:
    a ``next_speaker`` matching nobody makes the orchestrator answer having asked no one."""
    return json.dumps(
        {
            "is_request_satisfied": {"reason": "r", "answer": satisfied},
            "is_in_loop": {"reason": "r", "answer": False},
            "is_progress_being_made": {"reason": "r", "answer": True},
            "next_speaker": {"reason": "r", "answer": HYPOTHESISER_ROLE},
            "instruction_or_question": {"reason": "r", "answer": "Form én retning."},
        }
    )


def scripted_exploration_factory(
    sink: list[str], *, label: str, rationale: str, ledgers: Sequence[str]
) -> Callable[[str], BaseChatClient]:
    """A role-keyed client factory covering BOTH the exploration's roles and the pipeline's.

    One factory, because the walkthrough is one continuous story: the same call that shapes the
    mandate hands it to the run that evaluates it. The pipeline's roles fall through to the
    existing ``_proposer_reply`` / ``_CHECKER_APPROVE`` scaffolding, so nothing about the debate
    changes."""
    pipeline = scripted_factory({"proposer": _proposer_reply, "checker": _CHECKER_APPROVE}, sink)
    hypothesis_line = f"{HYPOTHESIS_MARKER} " + json.dumps({"label": label, "rationale": rationale})

    def factory(role: str) -> BaseChatClient:
        if role == MANAGER_ROLE:
            return ScriptedChatClient(
                sink=sink, role=role, reply_selector=_exploration_manager_reply(ledgers)
            )
        if role == HYPOTHESISER_ROLE:
            return ScriptedChatClient(hypothesis_line, sink, role=role)
        if role == NAVIGATOR_ROLE:
            return ScriptedChatClient("NAVIGATOR: leste indeksen.", sink, role=role)
        return pipeline(role)

    return factory


@dataclass(frozen=True)
class ExplorationSimulationResult:
    """The trace of one U4 walkthrough: what the loop shaped, and whether the pipeline used it.

    ``label_in_bundle`` is the causality control carried in the result, exactly as
    ``marker_in_run_a_prompt`` is: without it, a label the base already states would look like a
    shaped direction and the whole scenario would prove nothing."""

    exploration: ExplorationResult
    trace: ExplorationTrace
    run: RunResult
    label: str
    label_in_bundle: bool
    label_in_generation_prompt: bool
    generation_prompts: list[str]


async def simulate_exploration(
    bundle_dir: str,
    work_dir: str,
    *,
    project_id: str = _PROJECT_ID,
    prompt: str = _EXPLORE_PROMPT,
    label: str = _EXPLORE_LABEL,
    rationale: str = _EXPLORE_RATIONALE,
    contract: ExplorationContract | None = None,
    max_rounds: int = 3,
) -> ExplorationSimulationResult:
    """Walk U4 offline: prompt + knowledge base -> ``Mandate`` -> the pipeline that evaluates it.

    A THIRD scenario beside ``simulate_learning_loop``, and deliberately not part of it. The
    demo's stdout and stderr are both byte-pinned fixtures (``tests/golden/demo-transcript.*``),
    and the learning walkthrough's claim — that a verdict crosses two runs — is a different claim
    from this one. Folding them together would make each harder to read and would move a pinned
    transcript for a reason unrelated to what it pins. This scenario is reachable by NAME only;
    ``main()`` does not call it.

    Honesty limit, the same one §1 states for the rest of the demo: every reply is scripted, so
    what is shown is that the plumbing closes — that a shaped direction really becomes an
    ``Approach`` the proposer is asked about — never that a live model would shape a good one.
    """
    copy = Path(work_dir) / "explore-bundle"
    shutil.copytree(bundle_dir, copy)
    copy_s = str(copy)

    # The vacuity guard. A label the base already states would reach the hypothesis prompt as
    # navigated context whether or not the exploration ran, so the scenario's own assertion would
    # hold against an implementation that never wired the mandate at all.
    context = okf.bundle_context(okf.navigate_bundle(copy_s))
    if label in context:
        raise ValueError(
            f"label {label!r} already appears in the knowledge base, so it would reach the "
            "hypothesis prompt as ordinary context and the walkthrough would demonstrate nothing; "
            "choose a direction the base does not already state"
        )

    sink: list[str] = []
    trace = ExplorationTrace()
    exploration = await explore(
        prompt,
        contract=contract or _EXPLORE_CONTRACT,
        bundle_dirs=(copy_s,),
        client_factory=scripted_exploration_factory(
            sink,
            label=label,
            rationale=rationale,
            ledgers=[_exploration_ledger(satisfied=False), _exploration_ledger(satisfied=True)],
        ),
        trace=trace,
    )

    run = cast(
        RunResult,
        await run_project(
            project_id,
            "local",
            docs_dir=copy_s,
            bundle_dir=copy_s,
            verdict_input={"decision": "approved", "rationale": "ekspert-persona (sim)"},
            store=VerdictStore(verdicts=[]),
            client_factory=scripted_exploration_factory(
                sink, label=label, rationale=rationale, ledgers=[]
            ),
            mandate=exploration.mandate,
            max_rounds=max_rounds,
        ),
    )

    prompts = _generation_prompts(sink)
    return ExplorationSimulationResult(
        exploration=exploration,
        trace=trace,
        run=run,
        label=label,
        label_in_bundle=label in context,
        label_in_generation_prompt=any(label in p for p in prompts),
        generation_prompts=prompts,
    )


def _outcome_line(result: RunResult) -> str:
    o = result.outcome
    if isinstance(o, ValidatedProposal):
        return (
            f"VALIDATED  (påstått {o.proposal.claimed_saving_nok:.0f} <= P90 {o.p90:.0f} NOK; "
            f"tiltak: {o.proposal.measure})"
        )
    return f"REJECTED   ({o.reason})"


def _clip(text: str, limit: int = 92) -> str:
    """One-line, length-capped rendering of a captured agent text. The trace is a walkthrough a
    listener can follow, not a transcript dump."""
    flat = " ".join(text.split())
    return flat if len(flat) <= limit else flat[: limit - 1] + "…"


def _num(value: float) -> str:
    """Render a cost magnitude without exponent notation.

    ``:g`` — what these lines used while the demo ran the reserve — switches to scientific notation
    at the 7th significant digit, so the DELIVERED baseline printed as ``4.38615e+06``. The reserve's
    ``300000`` has six digits and never reached the switch: a formatting defect the synthetic numbers
    hid and delivered content exposed on the first run. A demo whose headline cost line is unreadable
    cannot claim the gate is anchored in real cost lines.
    """
    if value == int(value):
        return str(int(value))
    return f"{value:f}".rstrip("0").rstrip(".")


def _items_line(proposal: SavingsProposal) -> str:
    return ", ".join(
        f"{i.code} {_num(i.quantity)} x {_num(i.unit_cost)}" for i in proposal.affected_items
    )


def _first_hypothesis(result: RunResult) -> SavingsProposal:
    """Step 2 made visible: the FIRST candidate the run produced. When Step 5 corrected the run,
    that first candidate is the one the validator falsified (``refinements[0]``); with no
    refinement, the candidate that left the run IS the first one."""
    if result.refinements:
        return result.refinements[0].proposal
    return result.outcome.proposal


def _step5_lines(result: RunResult) -> list[str]:
    """Step 5 made visible: every falsification that was fed back into a further hypothesis, then
    what the bounded loop ended on. No refinement means the first hypothesis validated — saying so
    is the honest output there, not printing nothing."""
    if not result.refinements:
        return ["      (ingen — første hypotese validerte; ingen forbedring var nødvendig)"]
    lines = [
        f"      #{n}: grunnen fra {rejected.proposal.claimed_saving_nok:.0f} NOK-hypotesen mates "
        "tilbake i neste forsøk (bundet av max_attempts)"
        for n, rejected in enumerate(result.refinements, start=1)
    ]
    lines.append(f"      etter forbedring: {_outcome_line(result)}")
    return lines


def _decision_line(result: RunResult) -> str:
    """Step 6 made visible: the TYPED outcome that leaves the run, with BOTH falsifiers named —
    the deterministic validator gated the numbers, the checker gated the reasoning."""
    o = result.outcome
    if isinstance(o, ValidatedProposal):
        return (
            f"FORESLÅTT — {o.proposal.measure}: {o.proposal.claimed_saving_nok:.0f} NOK  "
            f"(validator={result.provenance.validator_decision}, checker={result.checker_verdict})"
        )
    return f"FORKASTET — {o.reason}"


def _verdict_origin_line(verdicts: Sequence[Verdict], *, marker: str, inbox_marker: str) -> str:
    """P4 pkt. 4, the seed sentence: of Run B's previous verdicts, how many came WITH the knowledge
    base and how many the demo learned in this session.

    Said out loud on stage, so it has to be true of the content actually being run — and the split
    is therefore DERIVED, never written down. The line above it already prints the retrieved count;
    a hand-written "one of three" would be a second copy of that number, and would go on being said
    unchanged after a bundle shipped a second seeded verdict.

    The classifier is the two markers the demo has traced all along: a retrieved verdict carrying one
    of them is one this session produced (the Step-8 promotion, the Step-7 inbox note). That rests on
    the seeded verdict carrying NEITHER, which is measured on the delivered bundle in
    ``tests/test_p4_honesty_sentences_loadbearing.py`` rather than assumed here.
    """
    learned = sum(1 for v in verdicts if marker in v.rationale or inbox_marker in v.rationale)
    seeded = len(verdicts) - learned
    return (
        f"      av disse fulgte {seeded} av {len(verdicts)} med kunnskapsbasen; "
        f"de øvrige {learned} er dem demoen lærte i denne økten"
    )


def _run_trace_lines(result: RunResult, *, marker: str, marker_in_prompt: bool) -> list[str]:
    """Steps 1-7 of one run, one labelled line per step (``method-spec`` §3), for the walkthrough.

    PRESENTATION ONLY: every value is read off the ``RunResult`` the run already returned — nothing
    is recomputed against the bundle and nothing is inferred. Two honesty limits are visible in what
    is printed rather than papered over:

    * ``retrieved`` is the POST-hoc, proposal-keyed retrieval (``run.py`` step 7), not the Step-1
      fold itself. The marker line is the evidence that a prior verdict reached the hypothesis
      PROMPT, which is the property Step 1 actually claims.
    * the run carries the checker's DECISION, not its prose (``_checker_verdict`` parses the gate
      marker and keeps the decision). The decision is what gates, so it is what is shown.
    """
    hypothesis = _first_hypothesis(result)
    files = [c.file for c in result.provenance.citations]
    lines = [
        "  Steg 1 — FORSTÅ KONTEKSTEN  (navigert kunnskapsbase + tidligere dommer)",
        f"      navigerte konseptfiler ({len(files)}): {', '.join(files)}",
        f"      tidligere dommer hentet for kandidaten: {len(result.retrieved)}",
        f"      markør '{marker}' i hypotese-prompten: {marker_in_prompt}",
        "  Steg 2 — HYPOTESE  (kandidat med parametere)",
        f"      tiltak: {hypothesis.measure}",
        f"      kostlinjer: {_items_line(hypothesis)}",
        f"      påstått besparelse: {hypothesis.claimed_saving_nok:.0f} NOK",
        "  Steg 3 — DEBATT  (maker-checker, Group Chat)",
        f"      proposer (konvergert): {_clip(result.debate_output)}",
        f"      checker (gate på resonnementet): VERDICT={result.checker_verdict.upper()}",
        "  Steg 4 — VALIDER / FALSIFISER  (deterministisk, blokkerende)",
    ]
    if result.refinements:
        # The status tokens stay VALIDATED/REJECTED (English) on purpose: they are the same
        # vocabulary as ``provenance.validator_decision``, which the Step-6 line prints verbatim.
        lines.append(f"      hypotese #1: REJECTED   ({result.refinements[0].reason})")
    else:
        lines.append(f"      {_outcome_line(result)}")
    lines.append("  Steg 5 — FORBEDRE, INFORMERT OG BUNDET")
    lines.extend(_step5_lines(result))
    lines.append("  Steg 6 — FORKAST ELLER FORESLÅ  (typet utfall forlater kjøringen)")
    lines.append(f"      {_decision_line(result)}")
    # Step 7 has TWO timescales, and only the short one happens INSIDE a run: the expert verdict
    # captured while the run is live. The long one — a verdict file dropped into a folder after the
    # run — is printed by ``main`` between the two runs, because that is when it happens.
    lines.append("  Steg 7 — SVAR PÅ TILBAKEMELDING  (ekspert-persona, kort løkke i kjøringen)")
    lines.append(f"      dom: {result.verdict.decision}")
    lines.append(f"      begrunnelse: {_clip(result.verdict.rationale, 300)}")
    return lines


def _baseline_lines(bundle_dir: Path, provenance: str) -> list[str]:
    """What the knowledge base DECLARES about the project's own cost lines.

    Read off the BUNDLE, not off a ``RunResult`` — which is why it is printed by ``main`` and not by
    ``_run_trace_lines`` (whose contract is that every value comes from the run). Whether the run
    then USES the baseline is not something a screen can show: every other line of the demo is
    byte-identical anchored or not, so that property is measured by
    ``tests/test_anchored_reserve_loadbearing.py`` instead.

    ``provenance`` is a required argument, not a default: the caller who chooses the bundle is the
    only one who knows where its numbers came from, and saying so is the honesty claim itself.

    The two branches share no wording (measured: an "ingen kostbaseline erklært" phrasing CONTAINS
    "kostbaseline erklært", which made the entry-point test pass with the anchoring detached)."""
    baseline = okf.load_optional_cost_baseline(str(bundle_dir))
    if baseline is None:
        return [
            f"KUNNSKAPSBASE: {bundle_dir.name} — uten kostbaseline",
            "      validatoren regner kun på tallene forslaget selv oppgir",
        ]
    items = ", ".join(
        f"{code} {_num(line.quantity)} x {_num(line.unit_cost)}"
        for code, line in sorted(baseline.items.items())
    )
    return [
        f"KUNNSKAPSBASE: {bundle_dir.name} — kostbaseline erklært ({items})",
        "      validatorens stage 0 avstemmer forslagets kostlinjer mot disse, FØR løseren",
        f"      {provenance}",
    ]


def main(argv: list[str] | None = None) -> int:  # pragma: no cover - console trace
    """Run the simulation against the energi bundle in a throwaway temp dir and print an honest,
    readable trace. Invoke: ``uv run python -m portfolio_optimiser.simulation``."""
    import asyncio
    import sys
    import tempfile

    # P4 pkt. 2: runtime half of the stderr discipline (the import-time half wrapped the
    # ``agent_framework`` import above). Installed here rather than at import time so that a
    # library consumer of this module keeps its own logging configuration.
    quiet_expected_round_cap_notice()

    # U14: install OpenTelemetry providers if — and only if — PORTFOLIO_OTEL asks for it, BEFORE
    # any workflow runs (MAF's own "call once at startup, before telemetry is captured"). Absent
    # the variable this configures nothing at all, which is what keeps the pinned stdout AND the
    # four-line pinned stderr byte-identical; console mode writes spans to stderr, never stdout.
    # The announcement goes to stderr for the same reason.
    setup = configure_tracing()
    notice = tracing_notice(setup)
    if notice is not None:
        print(notice, file=sys.stderr)

    work = tempfile.mkdtemp(prefix="po-sim-")
    # THE call site (P3, GO): the demo runs the DELIVERED bundle, which ships its own
    # `cost-baseline.json` — so the gate is anchored on numbers a domain team wrote, not on numbers
    # the script derived. The abort path is these three lines reverted to the anchored reserve
    # (`materialize_anchored_bundle` + `_RESERVE_PROVENANCE` + `_PROJECT_ID`); the run path's seam
    # is identical either way, which is what made pointing at delivered content a call-site change.
    bundle = _delivered_bundle_dir()
    provenance = _VEGLYS_PROVENANCE
    project_id = _VEGLYS_PROJECT_ID
    result = asyncio.run(simulate_learning_loop(str(bundle), work, project_id=project_id))

    print("=" * 78)
    print("OFFLINE SIMULERING — skriptede agent-svar, INGEN ekte modell.")
    print("Beviser dataflyten, den deterministiske ryggraden og at læringssløyfa lukkes.")
    print("Beviser IKKE at en levende modell ville produsert dette — forslag og dom er skriptet.")
    print("=" * 78)

    print()
    for line in _baseline_lines(bundle, provenance):
        print(line)

    # Run A walks steps 1-7 of the method; the promotion between the runs IS step 8. Run B is not
    # re-numbered — it re-runs the same eight steps, and what the demo needs from it is the ONE
    # thing that changed: the marker now reaches the hypothesis prompt.
    print(f"\nKJØRING A  ({project_id} — fersk kunnskapsbase, ingen tidligere dommer)")
    for line in _run_trace_lines(
        result.run_a, marker=result.marker, marker_in_prompt=result.marker_in_run_a_prompt
    ):
        print(line)
    print("      (forventet: markøren er FRAVÆRENDE her — dommen finnes ikke i wikien ennå)")

    print("\n  MELLOM KJØRINGENE — to uavhengige tilbakemeldings-veier tas i bruk")

    print("\n  Steg 7 (lang løkke) — EN EKSPERT LEGGER EN DOM I INNBOKSEN, ETTER KJØRINGEN")
    print(f"      fil: {result.inbox_path.parent.name}/{result.inbox_path.name}")
    print(f"      bærer: {result.inbox_marker}  (ettersendt driftsmåling — ny kunnskap)")
    print("      rollene byttes aldri: systemet LESER denne mappa, eksperten SKRIVER den")
    print("      neste kjøring merger fila inn i minnet FØR hypotesen formes — dager kan gå")

    print("\n  Steg 8 — PROMOTER GODKJENT KUNNSKAP  (gatet wiki-promotering)")
    print(f"      skrev: {result.promoted_path.name}  (lenket i index.md, nøytral etikett)")
    print(f"      bærer: {result.marker}  (personaens dom fra kjøring A)")
    print("      gaten er fail-closed: kun en godkjent dom promoteres — rå agent-output aldri")

    print("\nKJØRING B  (re-seedet kunnskapsbase + innboksen lest)")
    print("  samme åtte steg kjøres igjen; her vises kun det som ENDRET seg:")
    print(f"      tidligere dommer hentet for kandidaten: {len(result.run_b.retrieved)}")
    print(
        _verdict_origin_line(
            result.run_b.retrieved, marker=result.marker, inbox_marker=result.inbox_marker
        )
    )
    print(
        f"      markør '{result.marker}' (Steg 8, wiki) i hypotese-prompten: "
        f"{result.marker_in_run_b_prompt}  (forventet True)"
    )
    print(
        f"      markør '{result.inbox_marker}' (Steg 7, innboks) i hypotese-prompten: "
        f"{result.inbox_marker_in_run_b_prompt}  (forventet True)"
    )
    print(f"      utfall: {_decision_line(result.run_b)}")

    # Both paths must close, and neither may have been present in Run A — two markers means the two
    # mechanisms are checked separately here, exactly as the load-bearing tests check them.
    closed = (
        result.marker_in_run_b_prompt
        and not result.marker_in_run_a_prompt
        and result.inbox_marker_in_run_b_prompt
        and not result.inbox_marker_in_run_a_prompt
    )
    print("\n" + "-" * 78)
    if closed:
        print("LÆRINGSSLØYFA ER LUKKET, PÅ BEGGE TIDSSKALAER: kunnskapen eksperten godkjente i")
        print("kjøring A nådde kjøring B's hypotese via den fil-baserte wikien (promoter ->")
        print("re-seed -> fold), OG driftsnotatet som ble lagt i innboksen etterpå nådde den via")
        print("fil-innboksen (skriv fil -> merge -> fold). Ingen av dem gikk gjennom minnet.")
    else:
        print("LÆRINGSSLØYFA ER IKKE LUKKET — en av markørene krysset ikke kjøringene som ventet.")
    print("-" * 78)
    # The throwaway copy's path is the ONE non-deterministic value here (``mkdtemp``), so it goes to
    # stderr: stdout is then byte-identical across runs, which is what the dress rehearsal's
    # ``diff <(run1) <(run2)`` check compares. It stays visible on a terminal either way.
    print(f"\n(arbeidskopi: {work})", file=sys.stderr)
    return 0 if closed else 1


if __name__ == "__main__":  # pragma: no cover - console entry
    raise SystemExit(main())