portfolio-optimiser/tests/test_explore_loadbearing.py

"""U4 + U13-synchronous (økt 56) — the Magentic exploration loop as a MANDATE-FORMER.

**What this loop is, and what it deliberately is not.** ``explore()`` puts a Magentic manager
*over* the normative pipeline, never inside it: the manager is free to choose which knowledge base
to open and which hypothesis to shape next, and what leaves that freedom is a
``mandate.Mandate`` — a list of approaches worth *testing*. It is never a proposal. Every number
that survives is still gated by ``validate_proposal`` inside ``run_project``, in the same blocking
gate as today, and the exploration itself can write to neither the outbox nor the wiki. Step 3's
maker-checker debate is untouched (``shared/method-spec.md`` §3 is commons-owned and normative).

**Everything asserted here was measured before it was built** (plan
``docs/plan/2026-08-23-magentic-utforskningssloeyfe.md`` § F, spikes S0–S6 in økt 54, plus three
probes run at the head of økt 56):

* a plan-review ``revise`` costs two manager calls, **zero** rounds, and asks *again* — so an
  always-revising expert is unbounded spend under a round cap that never ticks. That is the whole
  reason ``max_plan_revisions`` is a required contract field rather than a nicety.
* the round cap and the reset cap **raise nothing**. Both end the run with a canonical assistant
  message and a normal-looking result (measured: ``max_round_count=2`` → two ledger events and
  ``'Workflow terminated due to reaching maximum round count.'``; a stalling ledger with
  ``max_reset_count=1`` → one ``REPLANNED`` event and ``'…maximum reset count.'``). At the
  transport both are indistinguishable from success, so this layer produces the typed stop itself.
* a ``next_speaker`` naming nobody produces a **silent final answer with zero participant work**
  (``_magentic.py:1128-1131``) — a plausible answer produced by no work at all, which is the
  hazard class E2 was retired for. The names are therefore validated, never assumed.

**The client is the repo's own ``ScriptedChatClient``.** A bare ``BaseChatClient`` silently no-ops
``BudgetMiddleware`` (measured, ``simulation.py:373-375``), so a budget claim proved against one
would prove nothing.
"""

from __future__ import annotations

import asyncio
import json
import os
import shutil
import subprocess
import sys
from collections.abc import Callable
from pathlib import Path
from typing import Any

import pytest
from agent_framework import BaseChatClient
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
from pydantic import ValidationError

import portfolio_optimiser
from portfolio_optimiser import explore, okf
from portfolio_optimiser.budget import Budget, BudgetExceeded, TokenMeter
from portfolio_optimiser.explore import ExplorationContract
from portfolio_optimiser.mandate import Approach
from portfolio_optimiser.simulation import ScriptedChatClient


# ---------------------------------------------------------------------------------------------
# C.3 — the contract: an exploration without stated bounds refuses to start
# ---------------------------------------------------------------------------------------------


_FULL_CONTRACT = {
    "max_rounds": 4,
    "max_tokens": 5_000,
    "max_stall_count": 2,
    "max_reset_count": 1,
    "max_plan_revisions": 1,
    "enable_plan_review": True,
}


@pytest.mark.parametrize("omitted", sorted(_FULL_CONTRACT))
def test_every_bound_is_required_with_no_default(omitted: str) -> None:
    """T1: each of the six fields is REQUIRED — dropping any one refuses construction.

    Not a style point. ``MagenticBuilder`` defaults ``max_round_count`` to ``None`` (unbounded)
    and ``max_reset_count`` to ``None`` (unlimited), and inheriting either would give this repo
    the one thing ``shared/method-spec.md`` §8 forbids outright: a loop with no stated end. A
    default here would also be a claim about the operator's intent that nobody made — the same
    ground on which ``ProvenanceStamp.cost_baseline_anchored`` is required without one.
    """
    payload = {k: v for k, v in _FULL_CONTRACT.items() if k != omitted}
    with pytest.raises(ValidationError):
        ExplorationContract(**payload)


def test_full_contract_constructs() -> None:
    """T2: the control for T1 — the complete payload IS valid.

    Without it, T1 would pass on a model that refuses everything, which is the vacuous-gate class
    this repo has paid for six times.
    """
    contract = ExplorationContract(**_FULL_CONTRACT)
    assert contract.max_rounds == 4
    assert contract.enable_plan_review is True


def test_a_revision_cap_without_plan_review_is_refused_not_ignored() -> None:
    """T3: ``max_plan_revisions > 0`` with ``enable_plan_review=False`` refuses.

    A plan revision can only arise from a plan review — with the review off, the cap bounds an
    event that cannot occur, and a caller who set it believes they bounded something. This repo
    refuses a setting that cannot take effect rather than dropping it silently (the same partition
    ``--embedder-config requires --semantic-retrieval`` enforces on the CLI).
    """
    with pytest.raises(ValidationError):
        ExplorationContract(**{**_FULL_CONTRACT, "enable_plan_review": False})


def test_review_off_with_zero_revisions_is_the_coherent_form() -> None:
    """T4: the control for T3 — review off and the cap at ``0`` is a consistent statement, and
    must construct. Without this arm T3 would pass on a model that simply forbade
    ``enable_plan_review=False`` outright, which is a different (and wrong) rule.
    """
    contract = ExplorationContract(
        **{**_FULL_CONTRACT, "enable_plan_review": False, "max_plan_revisions": 0}
    )
    assert contract.enable_plan_review is False
    assert contract.max_plan_revisions == 0


# ---------------------------------------------------------------------------------------------
# The scripted stand-ins. ScriptedChatClient, never a bare BaseChatClient: the latter no-ops
# BudgetMiddleware (measured, simulation.py:373-375), so a budget assertion made against one
# would assert nothing.
# ---------------------------------------------------------------------------------------------

PROMPT = "Find the cheapest saving available in the energy bundle."


def _ledger_json(
    *, satisfied: bool, speaker: str, instruction: str = "Shape one hypothesis."
) -> str:
    """A progress ledger naming ``speaker``.

    The name is a PARAMETER, never a literal, because a ``next_speaker`` matching no participant
    is the measured footgun this module defends against: the orchestrator does not error, it
    quietly emits a final answer having asked nobody (``_magentic.py:1128-1131``).
    """
    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": speaker},
            "instruction_or_question": {"reason": "r", "answer": instruction},
        }
    )


def _stalling_ledger_json(speaker: str) -> str:
    """A ledger reporting NO progress and a loop — the two flags that drive ``stall_count`` up."""
    return json.dumps(
        {
            "is_request_satisfied": {"reason": "r", "answer": False},
            "is_in_loop": {"reason": "circles", "answer": True},
            "is_progress_being_made": {"reason": "none", "answer": False},
            "next_speaker": {"reason": "r", "answer": speaker},
            "instruction_or_question": {"reason": "r", "answer": "Try again."},
        }
    )


def _manager_script(
    ledgers: list[str], calls: list[str] | None = None
) -> Callable[[str, str], str]:
    """Route a manager prompt blob to its scripted reply, consuming ``ledgers`` in order.

    The ORDER of these tests is load-bearing and was measured (§ F, A6): the selector receives the
    CONCATENATION of every message in the call, so a later-stage prompt still carries the earlier
    stage's text — one manager call in five carries two markers. Testing the later stage FIRST is
    what resolves it; reversing two of these silently reattributes a reply to the wrong stage.
    """

    def _select(blob: str, _role: str) -> str:
        if calls is not None:
            calls.append(blob[:40])
        if "provide the final answer" in blob:
            return "FINAL: exploration done."
        if "pure JSON format" in blob:
            return ledgers.pop(0) if ledgers else _ledger_json(satisfied=True, speaker="navigator")
        if "went wrong on this last run" in blob:
            return "PLAN-UPDATE: revised plan."
        if "rewrite the following fact sheet" in blob:
            return "FACTS-UPDATE: revised facts."
        if "bullet-point plan" in blob:
            return "PLAN: - ask the hypothesiser"
        if "pre-survey" in blob:
            return "FACTS: the bundle is anchored."
        return "{}"

    return _select


def _factory(
    *, ledgers: list[str], hypothesiser: list[str], navigator: str = "NAVIGATOR: index read."
) -> Callable[[str], BaseChatClient]:
    """One fresh ``ScriptedChatClient`` per role, exactly as the real factory hands out one per
    role. ``hypothesiser`` is a list consumed in order, so a run can shape several candidates."""

    def factory(role: str) -> BaseChatClient:
        if role == explore.MANAGER_ROLE:
            return ScriptedChatClient(reply_selector=_manager_script(ledgers), role=role)
        if role == explore.HYPOTHESISER_ROLE:
            replies = list(hypothesiser)

            def _hyp(_blob: str, _role: str) -> str:
                return replies.pop(0) if replies else "nothing further."

            return ScriptedChatClient(reply_selector=_hyp, role=role)
        return ScriptedChatClient(navigator, role=role)

    return factory


def _hypothesis_line(label: str, rationale: str) -> str:
    return f"{explore.HYPOTHESIS_MARKER} " + json.dumps({"label": label, "rationale": rationale})


#: The no-review base every stop test derives from. ``enable_plan_review`` and
#: ``max_plan_revisions`` move together — ``ExplorationContract`` refuses them apart — so a test
#: about round or stall behaviour has to say so explicitly rather than inherit ``_FULL_CONTRACT``.
_NO_REVIEW = {**_FULL_CONTRACT, "enable_plan_review": False, "max_plan_revisions": 0}

_CONTRACT = ExplorationContract(
    max_rounds=6,
    max_tokens=100_000,
    max_stall_count=2,
    max_reset_count=1,
    max_plan_revisions=0,
    enable_plan_review=False,
)


# ---------------------------------------------------------------------------------------------
# C.0 / C.6 — the exploration is a MANDATE-FORMER, and a seed never disappears
# ---------------------------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_hypotheses_become_the_mandate_in_the_order_they_were_shaped() -> None:
    """T5: what the hypothesiser MARKED becomes ``Mandate.approaches``, rationale VERBATIM.

    The rationale is the half a model cannot re-derive from cost data — ``mandate.Approach``
    already feeds ``description`` to the proposer verbatim (``generate._build_messages``), so
    paraphrasing it here would drop precisely the part the exploration exists to carry forward.
    """
    ledgers = [
        _ledger_json(satisfied=False, speaker=explore.HYPOTHESISER_ROLE),
        _ledger_json(satisfied=True, speaker=explore.HYPOTHESISER_ROLE),
    ]
    result = await explore.explore(
        PROMPT,
        contract=_CONTRACT,
        bundle_dirs=(),
        client_factory=_factory(
            ledgers=ledgers,
            hypothesiser=[
                "Looking at the bundle.\n"
                + _hypothesis_line("LED retrofit", "the fixtures are 1990s fluorescent")
            ],
        ),
    )

    assert [a.label for a in result.mandate.approaches] == ["LED retrofit"]
    assert result.mandate.approaches[0].description == "the fixtures are 1990s fluorescent"
    assert result.mandate.objective == PROMPT
    assert result.stop is None


@pytest.mark.asyncio
async def test_a_seed_approach_survives_whatever_the_manager_found() -> None:
    """T6: an expert's own hypothesis is in the output mandate, FIRST, untouched.

    Door 1 of § C.6, and the ``not_evaluated`` rule applied one stage earlier: a direction the
    domain expert asked for may never vanish because an autonomous loop preferred its own. Seeds
    lead so the pipeline reaches them before spending its budget on discovered ones.
    """
    seed = Approach(id="fagperson-1", label="Night setback", description="the expert's own words")
    ledgers = [
        _ledger_json(satisfied=False, speaker=explore.HYPOTHESISER_ROLE),
        _ledger_json(satisfied=True, speaker=explore.HYPOTHESISER_ROLE),
    ]
    result = await explore.explore(
        PROMPT,
        contract=_CONTRACT,
        bundle_dirs=(),
        seed_approaches=(seed,),
        client_factory=_factory(
            ledgers=ledgers,
            hypothesiser=[_hypothesis_line("LED retrofit", "fluorescent fixtures")],
        ),
    )

    assert [a.id for a in result.mandate.approaches] == ["fagperson-1", "hypothesis-1"]
    assert result.mandate.approaches[0] == seed


@pytest.mark.asyncio
async def test_an_exploration_that_shaped_nothing_still_returns_the_seeds() -> None:
    """T7: the control for T6 — with the hypothesiser silent, the seed is still the mandate.

    This is what makes T6 a statement about PRESERVATION rather than about ordering: a test that
    only ever saw seeds alongside discoveries could not tell "seeds are kept" from "seeds sort
    first".
    """
    seed = Approach(id="fagperson-1", label="Night setback", description="the expert's own words")
    result = await explore.explore(
        PROMPT,
        contract=_CONTRACT,
        bundle_dirs=(),
        seed_approaches=(seed,),
        client_factory=_factory(
            ledgers=[_ledger_json(satisfied=True, speaker=explore.NAVIGATOR_ROLE)],
            hypothesiser=[],
        ),
    )

    assert result.mandate.approaches == (seed,)
    assert result.mandate.allow_own_proposals is True


def test_zero_resets_is_refused_because_it_silently_explores_nothing() -> None:
    """T8: ``max_reset_count=0`` refuses — MEASURED, not reasoned.

    The orchestrator's limit check is ``reset_count >= max_reset_count`` (``_magentic.py:1243``)
    and ``reset_count`` starts at zero, so a cap of zero is already met before the first round.
    Measured against the installed stack: the run makes only the ``facts`` and ``plan`` manager
    calls, emits **zero** progress-ledger events, and returns
    ``'Workflow terminated due to reaching maximum reset count.'`` — an exploration that explored
    nothing, reported as a stall that never happened. An operator writing "allow no resets" would
    get "do no work", quietly. So it is refused at construction, where the reason can be said.
    """
    with pytest.raises(ValidationError):
        ExplorationContract(**{**_FULL_CONTRACT, "max_reset_count": 0})


def test_zero_stalls_is_allowed_because_it_means_something() -> None:
    """T9: the control for T8 — ``max_stall_count=0`` is a real setting and must construct.

    The stall check is STRICT (``stall_count > max_stall_count``, ``:1118``) and the counter is
    incremented before it, so zero means "reset on the first round that reports no progress".
    That is strictness, not self-defeat, and refusing both zeroes on symmetry would have banned a
    usable configuration on the strength of a measurement about a different field.
    """
    contract = ExplorationContract(**{**_FULL_CONTRACT, "max_stall_count": 0})
    assert contract.max_stall_count == 0


# ---------------------------------------------------------------------------------------------
# C.2 / C.3 — three endings the orchestration reports as if they were success
# ---------------------------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_the_round_cap_leaves_as_a_typed_budget_stop() -> None:
    """T10: the round cap becomes ``BudgetExceeded(kind="exploration_rounds")``.

    Measured (§ F, E5, re-measured at the head of this økt): ``max_round_count`` raises NOTHING.
    The run ends with the assistant message ``'Workflow terminated due to reaching maximum round
    count.'`` and a result that ``get_outputs()`` answers like any other — at the transport it is
    indistinguishable from a finished exploration. Left alone, a caller would read a run that
    explored two rounds of a six-round question as a completed answer. The triple is the one
    kø-(y) defends: WHICH cap bound, what it was, and how far the run actually got.
    """
    contract = ExplorationContract(**{**_NO_REVIEW, "max_rounds": 2})
    with pytest.raises(BudgetExceeded) as excinfo:
        await explore.explore(
            PROMPT,
            contract=contract,
            bundle_dirs=(),
            client_factory=_factory(
                ledgers=[_ledger_json(satisfied=False, speaker=explore.HYPOTHESISER_ROLE)] * 4,
                hypothesiser=[_hypothesis_line("LED", "worth a look")] * 4,
            ),
        )

    assert excinfo.value.kind == "exploration_rounds"
    assert excinfo.value.limit == 2
    assert excinfo.value.observed == 2


@pytest.mark.asyncio
async def test_a_request_satisfied_on_the_last_allowed_round_is_success() -> None:
    """T11: the discriminator for T10 — reaching the cap is not the same as being cut off by it.

    Both runs end with exactly ``max_rounds`` progress-ledger events, so a check written on the
    count alone would raise on this one too and turn a completed exploration into a budget error.
    What separates them is the LAST ledger's ``is_request_satisfied``, which is also what the
    orchestrator itself branches on (``:1106``). Without this arm, T10 would pass on an
    implementation that refuses every exploration that uses its whole allowance.
    """
    contract = ExplorationContract(**{**_NO_REVIEW, "max_rounds": 2})
    result = await explore.explore(
        PROMPT,
        contract=contract,
        bundle_dirs=(),
        client_factory=_factory(
            ledgers=[
                _ledger_json(satisfied=False, speaker=explore.HYPOTHESISER_ROLE),
                _ledger_json(satisfied=True, speaker=explore.HYPOTHESISER_ROLE),
            ],
            hypothesiser=[_hypothesis_line("LED", "worth a look")],
        ),
    )

    assert len(result.ledger_log) == contract.max_rounds
    assert result.stop is None
    assert [a.label for a in result.mandate.approaches] == ["LED"]


@pytest.mark.asyncio
async def test_stalling_out_is_a_typed_value_never_an_exception() -> None:
    """T12: stall → reset → out of resets is ``stop="stalled"``, and the run still returns.

    Kept as a VALUE while the round cap RAISES, and the split is S3.4's, not a preference: a
    stalled exploration is an outcome (the manager tried and got nowhere), whereas an exhausted
    round or token cap is resource exhaustion. Fusing them would leave a caller unable to tell
    "there was nothing here" from "we could not afford to look".
    """
    contract = ExplorationContract(
        **{**_NO_REVIEW, "max_rounds": 6, "max_stall_count": 1, "max_reset_count": 1}
    )
    result = await explore.explore(
        PROMPT,
        contract=contract,
        bundle_dirs=(),
        client_factory=_factory(
            ledgers=[_stalling_ledger_json(explore.HYPOTHESISER_ROLE)] * 6,
            hypothesiser=["still nothing."] * 6,
        ),
    )

    assert result.stop == "stalled"
    assert len(result.ledger_log) < contract.max_rounds
    assert all(entry.is_in_loop for entry in result.ledger_log)


@pytest.mark.asyncio
async def test_a_ledger_naming_nobody_withholds_what_the_run_produced() -> None:
    """T13: a ``next_speaker`` matching no participant stops the exploration and drops its finds.

    The measured footgun (``_magentic.py:1128-1131``): the orchestrator neither raises nor retries
    on an unknown speaker — it logs a warning and jumps to ``_prepare_final_answer``. The run
    therefore returns a plausible answer that no participant was asked for. This is the shape E2
    was retired over ("a plausible verdict produced by zero work"), so the mandate is NOT built
    from what such a run said it found.

    The scripted run reaches the bad ledger on round TWO, after a good round in which the
    hypothesiser really did commit to a direction. That ordering is what makes the assertion
    sharp: with the bad ledger first, nobody would ever have spoken and "nothing was carried
    forward" would be true of any implementation at all.
    """
    seed = Approach(id="fagperson-1", label="Night setback", description="expert's own")
    result = await explore.explore(
        PROMPT,
        contract=_CONTRACT,
        bundle_dirs=(),
        seed_approaches=(seed,),
        client_factory=_factory(
            ledgers=[
                _ledger_json(satisfied=False, speaker=explore.HYPOTHESISER_ROLE),
                _ledger_json(satisfied=False, speaker="a-name-nobody-answers-to"),
            ],
            hypothesiser=[_hypothesis_line("LED retrofit", "fluorescent fixtures")],
        ),
    )

    assert result.stop == "unknown_speaker"
    assert result.ledger_log[0].speaker_known is True
    assert result.ledger_log[-1].speaker_known is False
    # The seed survives — preservation is unconditional (§ C.6 door 1) — while the loop's own
    # find does not, because nothing stands behind the turn that ended the run.
    assert result.mandate.approaches == (seed,)


# ---------------------------------------------------------------------------------------------
# C.2 — the token cap covers the MANAGER, which is the loop's most talkative agent
# ---------------------------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_the_token_cap_binds_the_manager_before_any_participant_speaks() -> None:
    """T14: a one-token budget stops the exploration on the MANAGER's own first call.

    Agent-level ``ChatMiddleware`` does fire on the manager's calls (§ F, A1, measured green), and
    the manager talks more than anyone else in a Magentic loop — it extracts facts, writes the
    plan, and writes a progress ledger every single round. A cap fastened only to the participants
    would be a cap in name.

    The assertion is deliberately not "something raised". ``kind == "tokens"`` separates it from
    the round-cap stop, ``meter.tokens == 8`` shows the charge came from a call that was actually
    made and metered, and the EMPTY ledger log shows it landed before the loop had run a single
    round — which is exactly what a manager-attached middleware does and a participant-only one
    cannot.
    """
    meter = TokenMeter(Budget(max_tokens=1, max_rounds=6))
    contract = ExplorationContract(**{**_NO_REVIEW, "max_tokens": 1})

    with pytest.raises(BudgetExceeded) as excinfo:
        await explore.explore(
            PROMPT,
            contract=contract,
            bundle_dirs=(),
            meter=meter,
            client_factory=_factory(
                ledgers=[_ledger_json(satisfied=True, speaker=explore.NAVIGATOR_ROLE)],
                hypothesiser=[],
            ),
        )

    assert excinfo.value.kind == "tokens"
    assert meter.tokens == 8, "the manager's own call must have been charged to the meter"


# ---------------------------------------------------------------------------------------------
# C.5 / U13 — the synchronous plan review, and the cap the measurement forced
# ---------------------------------------------------------------------------------------------


def _reviewer(script: list[explore.PlanReviewDecision], seen: list[explore.PlanReviewRequest]):
    def review(request: explore.PlanReviewRequest) -> explore.PlanReviewDecision:
        seen.append(request)
        return script.pop(0) if script else explore.PlanReviewDecision.approve()

    return review


@pytest.mark.asyncio
async def test_a_revision_reaches_the_manager_and_the_review_is_asked_again() -> None:
    """T15: revise → replan → asked AGAIN → approve → the loop runs.

    This is målbilde's "ask the question, use the answer, carry on" on the installed stack: the
    expert's words go into the manager's history, the manager replans, and the human is asked to
    sign off on the NEW plan rather than the old one. Both round trips are recorded, in order,
    with the feedback verbatim — an audit of what a human actually told an autonomous loop is
    worth nothing paraphrased.
    """
    seen: list[explore.PlanReviewRequest] = []
    contract = ExplorationContract(
        **{**_FULL_CONTRACT, "enable_plan_review": True, "max_plan_revisions": 2}
    )
    result = await explore.explore(
        PROMPT,
        contract=contract,
        bundle_dirs=(),
        plan_reviewer=_reviewer(
            [explore.PlanReviewDecision.revise("Also test night setback.")], seen
        ),
        client_factory=_factory(
            ledgers=[
                _ledger_json(satisfied=False, speaker=explore.HYPOTHESISER_ROLE),
                _ledger_json(satisfied=True, speaker=explore.HYPOTHESISER_ROLE),
            ],
            hypothesiser=[_hypothesis_line("Night setback", "the expert asked for it")],
        ),
    )

    assert [r.decision for r in result.plan_reviews] == ["revise", "approve"]
    assert result.plan_reviews[0].feedback == "Also test night setback."
    assert len(seen) == 2, "a revision must produce a SECOND review, not resume silently"
    assert seen[1].plan != "", "the second review must show the revised plan"
    assert result.stop is None
    assert [a.label for a in result.mandate.approaches] == ["Night setback"]


@pytest.mark.asyncio
async def test_an_always_revising_reviewer_is_stopped_by_the_cap() -> None:
    """T16: the cap terminates a reviewer that never signs off — the reason it exists.

    Measured (§ F, A3): a revise costs two manager calls, emits NO progress ledger and consumes
    NO round, then asks again. The round cap therefore never ticks, and without
    ``max_plan_revisions`` this is an unbounded spend under caps that all look satisfied —
    precisely what ``shared/method-spec.md`` §8 forbids. The stop is typed and the exploration
    still returns; the reviewer's last (refused) revision is recorded, because the record is of
    what the human decided and ``stop`` is what says it was not applied.
    """
    seen: list[explore.PlanReviewRequest] = []
    always_revise = [explore.PlanReviewDecision.revise(f"Again #{n}.") for n in range(10)]
    contract = ExplorationContract(
        **{**_FULL_CONTRACT, "enable_plan_review": True, "max_plan_revisions": 1}
    )
    result = await explore.explore(
        PROMPT,
        contract=contract,
        bundle_dirs=(),
        plan_reviewer=_reviewer(always_revise, seen),
        client_factory=_factory(
            ledgers=[_ledger_json(satisfied=True, speaker=explore.NAVIGATOR_ROLE)],
            hypothesiser=[],
        ),
    )

    assert result.stop == "plan_revisions_exhausted"
    assert [r.decision for r in result.plan_reviews] == ["revise", "revise"]
    assert result.ledger_log == (), "the loop must never have run: the plan was never approved"


@pytest.mark.asyncio
async def test_a_reviewer_that_signs_off_at_once_is_not_capped() -> None:
    """T17: the control for T16 — the same cap, a reviewer that approves, and no stop.

    Without it, T16 would pass on an implementation that refuses every plan review it is given,
    which would stop the runaway loop and every legitimate one with it.
    """
    seen: list[explore.PlanReviewRequest] = []
    contract = ExplorationContract(
        **{**_FULL_CONTRACT, "enable_plan_review": True, "max_plan_revisions": 1}
    )
    result = await explore.explore(
        PROMPT,
        contract=contract,
        bundle_dirs=(),
        plan_reviewer=_reviewer([], seen),
        client_factory=_factory(
            ledgers=[_ledger_json(satisfied=True, speaker=explore.NAVIGATOR_ROLE)],
            hypothesiser=[],
        ),
    )

    assert result.stop is None
    assert [r.decision for r in result.plan_reviews] == ["approve"]
    assert len(result.ledger_log) >= 1, "an approved plan must let the loop actually run"


# ---------------------------------------------------------------------------------------------
# C.0 level 3 — the exploration has no write access, and level 1 is advisory
# ---------------------------------------------------------------------------------------------


def _tree(root: Path) -> dict[str, bytes]:
    return {
        str(p.relative_to(root)): p.read_bytes() for p in sorted(root.rglob("*")) if p.is_file()
    }


@pytest.mark.asyncio
async def test_an_exploration_leaves_the_knowledge_base_byte_identical(tmp_path: Path) -> None:
    """T18: ``explore()`` writes NOTHING — not to the base, not anywhere under it.

    Level 3 of the guarantee table: only the pipeline may write an outbox artefact, and only the
    gated ``promote_verdict`` may write to the wiki. An exploration that could write would be a
    route around the gate that makes an answer checkable — and, promoting into the base it reads,
    the self-contamination loop the Step-8 gate exists to prevent.

    Compared BYTE for byte over the whole subtree rather than by listing names, so a rewritten
    ``index.md`` of the same length would still fail.

    **The tools are exercised DIRECTLY, and that is a correction, not thoroughness.** A first
    version of this test drove only ``explore()`` — and a mutation that made ``read_bundle`` write
    a file into the base it reads left the WHOLE suite green (measured: 974 passed). A
    ``ScriptedChatClient`` returns text and never emits a tool call, so no scripted run reaches a
    tool body: the read surface, which is the only place a write could plausibly come from, was
    outside the gate entirely.
    """
    base = tmp_path / "bygg-energi-baseline-mikro"
    shutil.copytree(
        Path(portfolio_optimiser.__file__).parent / "data" / "bundles" / base.name, base
    )
    before = _tree(tmp_path)

    result = await explore.explore(
        PROMPT,
        contract=_CONTRACT,
        bundle_dirs=(str(base),),
        client_factory=_factory(
            # Three ledgers, and the third is what makes the second one matter: the orchestrator
            # tests ``is_request_satisfied`` BEFORE it reads ``next_speaker`` (``:1106``), so a
            # satisfied ledger naming the hypothesiser never actually asks it anything.
            ledgers=[
                _ledger_json(satisfied=False, speaker=explore.NAVIGATOR_ROLE),
                _ledger_json(satisfied=False, speaker=explore.HYPOTHESISER_ROLE),
                _ledger_json(satisfied=True, speaker=explore.HYPOTHESISER_ROLE),
            ],
            hypothesiser=[_hypothesis_line("LED retrofit", "fluorescent fixtures")],
        ),
    )

    assert result.mandate.approaches[0].label == "LED retrofit"

    # Every read tool, called on the same base, with model-shaped arguments.
    tools = {t.name: t for t in explore.navigator_tools((str(base),))}
    assert tools["list_bundles"].func()[0]["id"] == base.name
    assert tools["read_bundle"].func(bundle_id=base.name) != ""
    assert tools["read_file"].func(bundle_id=base.name, path="index.md") != ""
    explore.quick_validate_tool((str(base),)).func(
        bundle_id=base.name, proposal_json=json.dumps(_micro_projection())
    )

    assert _tree(tmp_path) == before


def _micro_bundle_dir() -> str:
    return str(
        Path(portfolio_optimiser.__file__).parent
        / "data"
        / "bundles"
        / "bygg-energi-baseline-mikro"
    )


def _micro_projection() -> dict[str, Any]:
    projection = dict(okf.load_ir_projection(_micro_bundle_dir()))
    projection.pop("_note", None)
    return projection


def test_quick_validate_reports_the_real_verdict_and_says_whether_it_was_anchored() -> None:
    """T19: the in-loop check is the SAME validator, and it declares its own anchoring.

    Level 1 is advisory but never fake: it runs ``validate_proposal`` against the base's own
    ``cost-baseline.json``, so stage 0 reconciliation is live and a fabricated cost line is caught
    in the loop rather than three steps later. ``anchored`` rides along for the reason
    ``ProvenanceStamp.cost_baseline_anchored`` is a required field — a verdict reached without the
    project's real cost lines is a weaker claim, and one that does not say so is a silence.
    """
    base = _micro_bundle_dir()
    projection = _micro_projection()
    validate = explore.quick_validate_tool((base,))

    honest = validate.func(
        bundle_id="bygg-energi-baseline-mikro", proposal_json=json.dumps(projection)
    )
    assert honest["decision"] == "validated"
    assert honest["anchored"] is True
    assert honest["p90"] >= honest["p50"] >= honest["p10"]

    # A cost code the project does not have is refused by stage 0 — the one stage that can tell a
    # fabricated line from a real one, and the reason `anchored` is worth reporting at all.
    invented = dict(projection)
    invented["affected_items"] = [
        {**dict(projection["affected_items"][0]), "code": "CODE-THAT-DOES-NOT-EXIST"}
    ]
    fabricated = validate.func(
        bundle_id="bygg-energi-baseline-mikro", proposal_json=json.dumps(invented)
    )
    assert fabricated["decision"] == "rejected"
    assert "CODE-THAT-DOES-NOT-EXIST" in fabricated["reason"]


def test_an_unknown_knowledge_base_is_refused_by_name() -> None:
    """T20: a tool call naming a base nobody configured refuses, and says what IS configured.

    Model-chosen arguments are untrusted input. Answering an unknown id with an empty result would
    let the manager conclude the base is empty rather than absent — the fourth face of the
    verification law, arrived at through a tool rather than a query.
    """
    validate = explore.quick_validate_tool(("/tmp/base-a",))
    with pytest.raises(explore.ExplorationError) as excinfo:
        validate.func(bundle_id="base-b", proposal_json="{}")
    assert "base-a" in str(excinfo.value)


def test_two_bases_with_the_same_name_are_refused() -> None:
    """T21: duplicate ids refuse at construction — the S3.2 key-collision class, one layer up.

    The id is how the manager names a base. Two bases answering to one name would let it read A
    while believing it read B, and every quotation it produced afterwards would be attributed to
    the wrong project.
    """
    with pytest.raises(explore.ExplorationError):
        explore.navigator_tools(("/tmp/one/shared-name", "/tmp/two/shared-name"))


# ---------------------------------------------------------------------------------------------
# U14 — the three events the tracing seam was landed for, now that they have a call site
# ---------------------------------------------------------------------------------------------


def _recording_tracer() -> tuple[Any, InMemorySpanExporter]:
    """A REAL OpenTelemetry tracer over an in-memory exporter — not a spy.

    A recorder standing in for ``add_event`` would prove that this module calls something shaped
    like OTel; this proves the events survive the actual SDK, with the attribute types it will
    accept. The provider is LOCAL and is never installed globally, so the pytest process keeps
    whatever tracing configuration it had (the same restraint U14's own tests exercise).
    """
    provider = TracerProvider()
    exporter = InMemorySpanExporter()
    provider.add_span_processor(SimpleSpanProcessor(exporter))
    return provider.get_tracer("test"), exporter


@pytest.mark.asyncio
async def test_the_three_orchestrator_events_reach_the_trace(monkeypatch: Any) -> None:
    """T22: ``plan_created``, ``replanned`` and ``progress_ledger_updated`` are recorded.

    These are the events U14 deliberately did NOT build in økt 55 — "an emitter with no call site
    is a shape guessed instead of measured". This is the call site. A Magentic manager decides
    which base to open and who speaks next; without these, the only trace of that reasoning is
    MAF's own ``invoke_agent`` spans, which say a call happened and nothing about what it decided.

    The ledger event carries the decision fields rather than a rendered sentence, for the reason
    ``SkippedLink`` is structured and ``BudgetExceeded`` carries three fields: "who was asked" and
    "was the request satisfied" are separate operative questions, and a reader who has to re-parse
    prose to tell them apart has a trace they cannot query.
    """
    tracer, exporter = _recording_tracer()
    # Patched where the name is BOUND (the ``hosting.run_project`` precedent): ``explore``
    # imports it by name, so patching ``tracing`` would leave that binding untouched and this
    # test would quietly measure nothing.
    monkeypatch.setattr(explore, "exploration_tracer", lambda: tracer)

    seen: list[explore.PlanReviewRequest] = []
    contract = ExplorationContract(
        **{**_FULL_CONTRACT, "enable_plan_review": True, "max_plan_revisions": 2}
    )
    await explore.explore(
        PROMPT,
        contract=contract,
        bundle_dirs=(),
        plan_reviewer=_reviewer([explore.PlanReviewDecision.revise("Test night setback.")], seen),
        client_factory=_factory(
            ledgers=[
                _ledger_json(satisfied=False, speaker=explore.HYPOTHESISER_ROLE),
                _ledger_json(satisfied=True, speaker=explore.HYPOTHESISER_ROLE),
            ],
            hypothesiser=[_hypothesis_line("Night setback", "the expert asked")],
        ),
    )

    spans = exporter.get_finished_spans()
    assert [s.name for s in spans] == [explore.EXPLORATION_SPAN]
    events = [(e.name, dict(e.attributes or {})) for e in spans[0].events]
    names = [name for name, _ in events]
    assert names.count("plan_created") == 1
    assert names.count("replanned") == 1, "the human's revision must be visible in the trace"
    assert names.count("progress_ledger_updated") == 2

    ledger_events = [attrs for name, attrs in events if name == "progress_ledger_updated"]
    assert [a["round_index"] for a in ledger_events] == [1, 2]
    assert [a["next_speaker"] for a in ledger_events] == [explore.HYPOTHESISER_ROLE] * 2
    assert [a["is_request_satisfied"] for a in ledger_events] == [False, True]
    assert all(a["speaker_known"] for a in ledger_events)


#: A complete exploration in a CHILD interpreter. The stdout/stderr question cannot be answered
#: in-process: ``ConsoleSpanExporter``'s ``out`` default is bound when
#: ``opentelemetry.sdk.trace.export`` is first imported, so under pytest it is whatever stdout was
#: at COLLECTION time — and ``capsys``, which replaces ``sys.stdout`` later, never sees it. That is
#: not a testing quirk to work around; it is precisely the fact U14 exists for, and the reason
#: ``configure_tracing`` passes ``out=`` explicitly instead of trusting the default. Measured: a
#: mutation routing exploration spans to that default left an in-process ``capsys`` assertion
#: GREEN while the spans really were on stdout.
_CHILD_EXPLORATION = """
import asyncio, json, sys
from portfolio_optimiser import explore
from portfolio_optimiser.simulation import ScriptedChatClient
from portfolio_optimiser.tracing import configure_tracing

configure_tracing()

LEDGER = json.dumps({
    "is_request_satisfied": {"reason": "r", "answer": True},
    "is_in_loop": {"reason": "r", "answer": False},
    "is_progress_being_made": {"reason": "r", "answer": True},
    "next_speaker": {"reason": "r", "answer": "navigator"},
    "instruction_or_question": {"reason": "r", "answer": "none"},
})

def _select(blob, _role):
    if "provide the final answer" in blob:
        return "FINAL: done."
    if "pure JSON format" in blob:
        return LEDGER
    if "bullet-point plan" in blob:
        return "PLAN: - ask the navigator"
    if "pre-survey" in blob:
        return "FACTS: none."
    return "{}"

def factory(role):
    if role == explore.MANAGER_ROLE:
        return ScriptedChatClient(reply_selector=_select, role=role)
    return ScriptedChatClient("ok", role=role)

contract = explore.ExplorationContract(
    max_rounds=4, max_tokens=100000, max_stall_count=2,
    max_reset_count=1, max_plan_revisions=0, enable_plan_review=False,
)
result = asyncio.run(
    explore.explore("probe", contract=contract, bundle_dirs=(), client_factory=factory)
)
assert result.stop is None, result.stop
print("EXPLORATION-OK", file=sys.stderr)
"""


def _run_child(**env: str) -> subprocess.CompletedProcess[str]:
    return subprocess.run(
        [sys.executable, "-c", _CHILD_EXPLORATION],
        capture_output=True,
        text=True,
        cwd=str(Path(__file__).resolve().parent.parent),
        env={**os.environ, **env},
    )


def test_an_untraced_exploration_writes_nothing_to_stdout_or_stderr() -> None:
    """T23: in a real process, with tracing off, an exploration prints NOTHING.

    A subprocess and not ``capsys``, for the reason recorded above ``_CHILD_EXPLORATION`` — and the
    stakes are the pinned artefacts: ``tests/golden/demo-transcript.stdout`` is byte-fixed and the
    demo's stderr is fixed at four lines, so one stray span dump would break both.

    ``EXPLORATION-OK`` on stderr is the control. Without it, "stdout was empty" would be equally
    true of a child that crashed on import, which is the fourth face of the verification law: an
    absence is only evidence once you have shown the measurement could have found something.
    """
    proc = _run_child(PORTFOLIO_OTEL="")
    assert proc.returncode == 0, proc.stderr
    assert "EXPLORATION-OK" in proc.stderr, "the child must really have run an exploration"
    assert proc.stdout == ""
    # Not an exact-equality assertion on stderr: MAF emits two ``ExperimentalWarning`` lines while
    # importing, under every run form, and they are the same pair the demo's pinned stderr already
    # carries. What must be absent is TRACE data, so that is what is asserted.
    assert '"name": "exploration"' not in proc.stderr
    assert "progress_ledger_updated" not in proc.stderr


def test_a_traced_exploration_puts_its_span_on_stderr_and_leaves_stdout_clean() -> None:
    """T24: the positive arm — ``PORTFOLIO_OTEL=console`` and the exploration span is on STDERR.

    This is what the whole U14 seam was landed for, now carrying the events U4 gave it a call site
    for. Both halves are asserted: the span and its ``progress_ledger_updated`` event ARE exported
    (so tracing is real), and stdout is STILL empty (so the byte-pinned transcript survives a
    traced run). Asserting only the first would pass on an exporter writing to stdout — which is
    OpenTelemetry's own default, and therefore the mistake actually available to make.
    """
    proc = _run_child(PORTFOLIO_OTEL="console")
    assert proc.returncode == 0, proc.stderr
    assert "EXPLORATION-OK" in proc.stderr
    assert proc.stdout == "", "a traced run must not put one byte on stdout"
    assert '"name": "exploration"' in proc.stderr
    assert "progress_ledger_updated" in proc.stderr


def test_a_marked_line_that_will_not_parse_is_a_hard_error() -> None:
    """T24: the marker is what makes fail-closed affordable here.

    Most hypothesiser turns legitimately are not hypotheses — the agent reasons out loud — so
    "parse every turn or fail" would refuse a normal exploration. The marker separates a turn that
    is not a claim from a claim that cannot be read. The second is the run's own product coming
    back unreadable, so it raises (``write_concept_file``'s rule: validation, never repair) rather
    than following the tolerant RAW-inbox rule, which belongs to folders anyone may drop files in.
    """
    with pytest.raises(explore.HypothesisParseError):
        explore._parse_hypotheses([f"{explore.HYPOTHESIS_MARKER} not json at all"])
    with pytest.raises(explore.HypothesisParseError):
        explore._parse_hypotheses([f'{explore.HYPOTHESIS_MARKER} {{"label": "no rationale"}}'])


def test_unmarked_prose_is_not_a_failure() -> None:
    """T25: the control for T24 — ordinary reasoning yields no hypothesis and no error.

    Without it, T24 would pass on an implementation that refused every hypothesiser turn that was
    not a hypothesis, which would make the loop unusable and the strictness meaningless.
    """
    assert explore._parse_hypotheses(["I looked at the index and nothing stands out yet."]) == []


def test_a_review_nobody_can_answer_is_refused_before_the_first_model_call() -> None:
    """T26: plan review without a reviewer refuses; a reviewer without plan review refuses too.

    The first would hang: the workflow stops at a ``request_info`` and nothing ever answers it, and
    a hang is the one failure mode that reports nothing at all. The second is the silent-ignore the
    repo's flag partition forbids — a caller who supplied a reviewer believes a human is in the
    loop. Both are refused BEFORE anything is built, so neither costs a model call.
    """
    with pytest.raises(explore.ExplorationError):
        asyncio.run(
            explore.explore(
                PROMPT,
                contract=ExplorationContract(**_FULL_CONTRACT),
                bundle_dirs=(),
                client_factory=_factory(ledgers=[], hypothesiser=[]),
            )
        )
    with pytest.raises(explore.ExplorationError):
        asyncio.run(
            explore.explore(
                PROMPT,
                contract=ExplorationContract(**_NO_REVIEW),
                bundle_dirs=(),
                plan_reviewer=lambda _r: explore.PlanReviewDecision.approve(),
                client_factory=_factory(ledgers=[], hypothesiser=[]),
            )
        )