portfolio-optimiser/spikes/e_magentic.py

"""Spike E — the Magentic exploration loop, measured BEFORE it is built
(order 20260823T162224Z; plan ``docs/plan/2026-08-23-magentic-utforskningssloeyfe.md`` § D.1).

Nothing here is production code and nothing here is wired into ``src/``. Each function is one
measurement whose outcome moves a row of the plan's § F assumption table from "umålt" to a fact.

**The client is the repo's own ``ScriptedChatClient``, not an ad-hoc fake.** The scratch scripts
this spike ports used a bare ``BaseChatClient``, on which ``BudgetMiddleware`` is silently a
no-op (measured, ``simulation.py:373-375``) — so a budget claim proved with one would have proved
nothing. ``ScriptedChatClient`` subclasses the LAYERED ``OpenAIChatCompletionClient``, which is
what makes S2 a real measurement.

**The budget types are the PRODUCTION ones** (``portfolio_optimiser.budget``), deliberately NOT
``spikes/_harness.py``'s private copy. The harness copy is exactly why ``tick_round``'s
``observed`` went four raise-sites without coverage (kø-(y)); S2's whole question is whether the
SHIPPED middleware reaches the manager, and only the shipped object can answer it.

**Routing is on the joined prompt blob, in a fixed priority order, and that order is
load-bearing.** ``ScriptedChatClient``'s selector receives the concatenation of every message in
the call, so one manager call carries two markers (the plan prompt is built on a history that
still holds the pre-survey text — measured: 1 ambiguous call in 5). Checking the later-stage
marker first resolves it; the five-kind call shape each experiment asserts is what proves the
routing stayed correct.
"""

from __future__ import annotations

import json
import statistics
import subprocess
import sys
import time
from collections.abc import Callable, Sequence
from dataclasses import dataclass
from pathlib import Path
from typing import Any

from agent_framework import Agent, FileCheckpointStorage
from agent_framework.orchestrations import (
    AgentRequestInfoResponse,
    MagenticAgentExecutor,
    MagenticBuilder,
    MagenticPlanReviewResponse,
    MagenticResetSignal,
    StandardMagenticManager,
)

# ``AgentApprovalExecutor`` is the ONE name S3b needs that the package does not re-export
# (measured: ``hasattr(agent_framework.orchestrations, "AgentApprovalExecutor")`` is False while
# its response type IS public). Reaching into the private module is therefore part of the S3b
# FINDING, not an oversight: door 3 of § C.6 currently costs a private-API dependency, and that
# is a fact the operator's decision needs to carry.
from agent_framework_orchestrations._orchestration_request_info import AgentApprovalExecutor

import portfolio_optimiser
from portfolio_optimiser.budget import Budget, BudgetExceeded, BudgetMiddleware, TokenMeter
from portfolio_optimiser.ir import SavingsProposal
from portfolio_optimiser.okf import load_ir_projection, load_optional_cost_baseline
from portfolio_optimiser.simulation import ScriptedChatClient
from portfolio_optimiser.validator import validate_proposal

TASK_ALPHA = "TASK-ALPHA: find the saving in the alpha project."
TASK_BETA = "TASK-BETA: find the saving in the beta project."
WORKER_SENTINEL = "WORKER-SAW"


def _ledger(*, satisfied: bool, speaker: str) -> str:
    """A progress ledger naming ``speaker`` as the next talker.

    The name is a PARAMETER because an unknown ``next_speaker`` is a silent footgun: the
    orchestrator does not error on it, it quietly produces a final answer with zero participant
    work (``_magentic.py:1128-1131``). Hard-coding "worker" here made the S3b run — whose
    participant is the ``expert_liaison`` — finish without ever asking anyone (measured).
    """
    return json.dumps(
        {
            "is_request_satisfied": {
                "reason": "the participant replied" if satisfied else "no one has spoken yet",
                "answer": satisfied,
            },
            "is_in_loop": {"reason": "no", "answer": False},
            "is_progress_being_made": {"reason": "yes", "answer": True},
            "next_speaker": {"reason": "it does the work", "answer": speaker},
            "instruction_or_question": {
                "reason": "done" if satisfied else "kick off",
                "answer": "none" if satisfied else "Do the work now.",
            },
        }
    )


@dataclass(frozen=True)
class ExplorationCallRecord:
    """One manager call, reduced to the facts every verdict here reads.

    ``sees_marker`` is the HITL half (S3b): whether a caller-supplied sentinel — an expert's
    answer injected mid-run — had reached this manager prompt. It defaults to ``False`` so the
    contamination experiments, which supply no marker, are unchanged.
    """

    kind: str
    messages: int
    sees_alpha: bool
    sees_beta: bool
    sees_marker: bool = False


def _route(blob: str, speaker: str) -> tuple[str, str]:
    """Map a manager prompt blob to ``(kind, reply)``.

    Order matters: the later-stage marker is tested FIRST because an earlier stage's text is
    still present in the joined blob. Reversing two of these silently changes which prompt a
    kind is attributed to, which is why every experiment asserts the resulting call shape.
    """
    if "provide the final answer" in blob:
        return "final", "FINAL: the worker did it."
    if "pure JSON format" in blob:
        if WORKER_SENTINEL in blob:
            return "ledger_SAT", _ledger(satisfied=True, speaker=speaker)
        return "ledger_UNSAT", _ledger(satisfied=False, speaker=speaker)
    if "went wrong on this last run" in blob:
        return "plan_update", "PLAN-UPDATE: ask the worker again."
    if "rewrite the following fact sheet" in blob:
        return "facts_update", "FACTS-UPDATE: still nothing."
    if "bullet-point plan" in blob:
        return "plan", "PLAN: - ask the worker"
    if "pre-survey" in blob:
        return "facts", "FACTS: nothing given."
    return "unknown", "{}"


def _manager_client(
    records: list[ExplorationCallRecord],
    *,
    marker: str | None = None,
    speaker: str = "worker",
) -> ScriptedChatClient:
    """A manager-shaped scripted client appending one record per call to ``records``.

    ``records`` is CALLER-owned (the parse-failure-capture precedent): the evidence must survive
    however the run ended, including a run cut short by ``BudgetExceeded`` mid-way.
    """

    def _select(blob: str, _role: str) -> str:
        kind, reply = _route(blob, speaker)
        records.append(
            ExplorationCallRecord(
                kind=kind,
                messages=blob.count("\n") + 1,
                sees_alpha="TASK-ALPHA" in blob,
                sees_beta="TASK-BETA" in blob,
                sees_marker=marker is not None and marker in blob,
            )
        )
        return reply

    return ScriptedChatClient(reply_selector=_select, role="manager")


def _worker_client(seen: list[str]) -> ScriptedChatClient:
    def _select(blob: str, _role: str) -> str:
        which = "ALPHA" if "TASK-ALPHA" in blob else ("BETA" if "TASK-BETA" in blob else "NOTHING")
        seen.append(which)
        return f"{WORKER_SENTINEL}-{which}"

    return ScriptedChatClient(reply_selector=_select, role="worker")


def _manager_agent(client: ScriptedChatClient, *, middleware: Sequence[Any] | None = None) -> Agent:
    return Agent(
        client,
        "You are the Magentic manager.",
        name="manager",
        description="plans the work",
        middleware=middleware,
    )


def _worker_agent(client: ScriptedChatClient) -> Agent:
    return Agent(client, "You are the worker.", name="worker", description="does the work")


async def _run(workflow: Any, task: str) -> dict[str, Any]:
    try:
        result = await workflow.run(task)
        outputs = [str(o) for o in (result.get_outputs() or [])]
        return {"ok": True, "outputs": outputs}
    except Exception as exc:  # noqa: BLE001 - the failure mode IS the measurement
        return {"ok": False, "error": type(exc).__name__, "message": str(exc)}


def manager_keeps_persistent_session() -> bool:
    """Does the INSTALLED ``StandardMagenticManager`` hold one ``AgentSession`` for its whole
    life, or mint a throwaway one per call?

    This is the single structural property that decides E2 and E4. Orchestrations 1.0.0 assigns
    ``self._session = self._agent.create_session()`` in ``__init__``; 1.0.1 removed that line and
    creates the session inside the call instead (upstream regression fix #4371). Probing the
    attribute rather than the version string states the CAUSE, and keeps the spike honest across
    a version the plan has not seen.
    """
    manager = StandardMagenticManager(agent=_manager_agent(_manager_client([])))
    return hasattr(manager, "_session")


async def single_use_second_run() -> dict[str, Any]:
    """E1: build one Magentic workflow, run it twice. Measures the second run's outcome and
    what it COST — a refusal that still made model calls would be a different finding."""
    records: list[ExplorationCallRecord] = []
    worker_seen: list[str] = []
    workflow = MagenticBuilder(
        participants=[_worker_agent(_worker_client(worker_seen))],
        manager_agent=_manager_agent(_manager_client(records)),
        max_round_count=6,
    ).build()

    first = await _run(workflow, TASK_ALPHA)
    manager_before, worker_before = len(records), len(worker_seen)
    second = await _run(workflow, TASK_BETA)
    return {
        "first_ok": first["ok"],
        "second_error": second.get("error"),
        "second_message": second.get("message", ""),
        "manager_calls_added": len(records) - manager_before,
        "worker_calls_added": len(worker_seen) - worker_before,
    }


async def _two_runs(
    build_second: Callable[[list[ExplorationCallRecord]], Any],
    *,
    shared_records: bool,
) -> tuple[int, int, list[str]]:
    """Run ALPHA, then BETA, and report how many of run 2's manager calls still see ALPHA.

    ``shared_records`` says whether run 2's manager is the same object as run 1's — when it is,
    run 2's records are the tail of one list; when it is not, they are their own list. Both
    shapes reduce to the same verdict triple so the callers stay comparable.
    """
    records: list[ExplorationCallRecord] = []
    first_worker: list[str] = []
    first = MagenticBuilder(
        participants=[_worker_agent(_worker_client(first_worker))],
        manager_agent=_manager_agent(_manager_client(records)),
        max_round_count=6,
    ).build()
    await _run(first, TASK_ALPHA)
    split = len(records)

    second_records = records if shared_records else []
    second = build_second(second_records)
    await _run(second, TASK_BETA)

    tail = records[split:] if shared_records else second_records
    bled = sum(1 for record in tail if record.sees_alpha)
    return bled, len(tail), [record.kind for record in tail]


async def shared_manager_contamination() -> tuple[int, int, list[str]]:
    """E2: two builders sharing ONE ``StandardMagenticManager`` instance.

    Built directly (not via ``_two_runs``) because run 1 must go through the SAME manager object,
    which ``manager=`` accepts and ``manager_agent=`` does not.
    """
    records: list[ExplorationCallRecord] = []
    shared = StandardMagenticManager(
        agent=_manager_agent(_manager_client(records)), max_round_count=6
    )

    first = MagenticBuilder(
        participants=[_worker_agent(_worker_client([]))], manager=shared
    ).build()
    await _run(first, TASK_ALPHA)
    split = len(records)

    second = MagenticBuilder(
        participants=[_worker_agent(_worker_client([]))], manager=shared
    ).build()
    await _run(second, TASK_BETA)

    tail = records[split:]
    return sum(1 for r in tail if r.sees_alpha), len(tail), [r.kind for r in tail]


async def shared_builder_contamination() -> tuple[int, int, list[str]]:
    """E4: ONE ``MagenticBuilder``, ``.build()`` twice. ``manager_agent=`` constructs the manager
    eagerly and hands the same instance to every build — the accidental route into E2."""
    records: list[ExplorationCallRecord] = []
    builder = MagenticBuilder(
        participants=[_worker_agent(_worker_client([]))],
        manager_agent=_manager_agent(_manager_client(records)),
        max_round_count=6,
    )

    await _run(builder.build(), TASK_ALPHA)
    split = len(records)
    await _run(builder.build(), TASK_BETA)

    tail = records[split:]
    return sum(1 for r in tail if r.sees_alpha), len(tail), [r.kind for r in tail]


async def fresh_manager_contamination() -> tuple[int, int, list[str]]:
    """E3, the control and the mitigation: a fresh builder, agent and client per exploration."""

    def _second(records: list[ExplorationCallRecord]) -> Any:
        return MagenticBuilder(
            participants=[_worker_agent(_worker_client([]))],
            manager_agent=_manager_agent(_manager_client(records)),
            max_round_count=6,
        ).build()

    return await _two_runs(_second, shared_records=False)


async def reset_signal_resets_participant_session() -> dict[str, Any]:
    """E7: does ``MagenticResetSignal`` actually give the participant a clean session?

    Measured no: the fresh session is written to ``_agent_thread``, which nothing reads, while
    the live ``_session`` keeps its identity. The stall-replan path therefore hands the manager a
    clean ledger and the participants their old memory.
    """
    executor = MagenticAgentExecutor(_worker_agent(_worker_client([])))
    session_before = id(executor._session)
    executor._cache.append("sentinel")  # type: ignore[arg-type]
    executor._full_conversation.append("sentinel")  # type: ignore[arg-type]

    await executor.handle_magentic_reset(MagenticResetSignal(), None)  # type: ignore[arg-type]

    return {
        "cache_cleared": len(executor._cache) == 0,
        "conversation_cleared": len(executor._full_conversation) == 0,
        "session_identity_changed": id(executor._session) != session_before,
        "orphan_attribute_written": hasattr(executor, "_agent_thread"),
    }


async def manager_budget_enforced(
    *, max_tokens: int, attach: bool, return_exception: bool = False
) -> dict[str, Any]:
    """S2: put the SHIPPED ``BudgetMiddleware`` on the manager agent and see whether the typed
    refusal leaves ``workflow.run``.

    ``attach=False`` is the detach control: the same one-token budget with no middleware must
    stop nothing. Without that arm the positive test would pass on any implementation in which
    something, anything, raised.
    """
    meter = TokenMeter(Budget(max_tokens=max_tokens, max_rounds=8))
    middleware = [BudgetMiddleware(meter)] if attach else None
    workflow = MagenticBuilder(
        participants=[_worker_agent(_worker_client([]))],
        manager_agent=_manager_agent(_manager_client([]), middleware=middleware),
        max_round_count=6,
    ).build()

    raised: str | None = None
    kind: str | None = None
    exception: BaseException | None = None
    completed = False
    try:
        result = await workflow.run(TASK_ALPHA)
        completed = bool(result.get_outputs())
    except BudgetExceeded as exc:
        raised, kind, exception = type(exc).__name__, exc.kind, exc
    except Exception as exc:  # noqa: BLE001 - a DIFFERENT exception type is itself the finding
        raised, exception = type(exc).__name__, exc

    payload: dict[str, Any] = {
        "raised": raised,
        "kind": kind,
        "completed": completed,
        "meter_tokens": meter.tokens,
    }
    if return_exception:
        payload["exception"] = exception
    return payload


# ---------------------------------------------------------------------------
# S3 / S3b — the two HITL doors (plan § C.5, § C.6)
# ---------------------------------------------------------------------------


def _pending_requests(result: Any) -> list[Any]:
    return [event for event in result if event.type == "request_info"]


async def plan_review_round_trip() -> dict[str, Any]:
    """S3: ``enable_plan_review=True`` → the run stops with a ``MagenticPlanReviewRequest`` and no
    output → ``revise(...)`` replans and asks AGAIN → ``approve()`` lets the loop run.

    The manager-call count PER revise is the number the contract needs: a revise costs model
    calls but is not counted as a round by the orchestration (measured: no ledger call), so an
    uncapped reviser is an unbounded spend the plan's ``max_plan_revisions`` has to bound.
    """
    records: list[ExplorationCallRecord] = []
    workflow = MagenticBuilder(
        participants=[_worker_agent(_worker_client([]))],
        manager_agent=_manager_agent(_manager_client(records)),
        max_round_count=6,
        enable_plan_review=True,
    ).build()

    first = await workflow.run(TASK_ALPHA)
    pending = _pending_requests(first)
    stopped_without_output = not first.get_outputs()
    review_kinds = [r.kind for r in records]

    before_revise = len(records)
    revised = await workflow.run(
        responses={pending[0].request_id: MagenticPlanReviewResponse.revise("Test the LED case.")}
    )
    revise_calls = [r.kind for r in records[before_revise:]]
    pending_after_revise = _pending_requests(revised)

    approve_id = (
        pending_after_revise[0].request_id if pending_after_revise else pending[0].request_id
    )
    approved = await workflow.run(responses={approve_id: MagenticPlanReviewResponse.approve()})

    return {
        "pending_before_review": len(pending),
        "request_type": type(pending[0].data).__name__ if pending else None,
        "is_stalled": bool(pending[0].data.is_stalled) if pending else None,
        "stopped_without_output": stopped_without_output,
        "kinds_before_review": review_kinds,
        "revise_manager_calls": revise_calls,
        "pending_after_revise": len(pending_after_revise),
        "outputs_after_approve": [str(o) for o in (approved.get_outputs() or [])],
    }


async def expert_liaison_answer_round_trip(
    *, answer: str, probe: str | None = None
) -> dict[str, Any]:
    """S3b: an ``AgentApprovalExecutor`` standing in as the ``expert_liaison`` PARTICIPANT — the
    third door of § C.6, the one that lets the manager ask a question MID-run.

    Measured shape, two round-trips per human turn: the manager picks the liaison → its output
    becomes a ``request_info`` → ``from_strings([answer])`` feeds the human's words back INTO the
    liaison, which runs again → a second ``request_info`` → ``approve()`` forwards the liaison's
    (now informed) output to the manager, which resumes. ``from_strings`` alone does NOT resume
    the manager: measured zero manager calls between the two requests.

    ``probe`` is the CONTROL knob: the manager's prompts are scanned for IT instead of for
    ``answer``, so a sentinel the expert never sent must come back absent. Without that arm,
    "the manager saw the answer" could equally mean "the scanner matches anything".
    """
    records: list[ExplorationCallRecord] = []
    liaison_prompts: list[str] = []

    def _liaison_reply(blob: str, _role: str) -> str:
        liaison_prompts.append(blob)
        if answer in blob:
            return f"{WORKER_SENTINEL}-LIAISON heard: {answer}"
        return f"{WORKER_SENTINEL}-LIAISON has no expert input yet."

    liaison = AgentApprovalExecutor(
        Agent(
            ScriptedChatClient(reply_selector=_liaison_reply, role="expert_liaison"),
            "You relay the expert's answers.",
            name="expert_liaison",
            description="asks the human expert",
        )
    )
    workflow = MagenticBuilder(
        participants=[liaison],
        manager_agent=_manager_agent(
            _manager_client(records, marker=probe or answer, speaker="expert_liaison")
        ),
        max_round_count=6,
    ).build()

    first = await workflow.run(TASK_ALPHA)
    asked = _pending_requests(first)

    before_answer = len(records)
    answered = await workflow.run(
        responses={asked[0].request_id: AgentRequestInfoResponse.from_strings([answer])}
    )
    manager_calls_on_answer = [r.kind for r in records[before_answer:]]
    second = _pending_requests(answered)

    approved = (
        await workflow.run(responses={second[0].request_id: AgentRequestInfoResponse.approve()})
        if second
        else answered
    )

    return {
        "reachable": bool(asked),
        "request_type": type(asked[0].data).__name__ if asked else None,
        "manager_calls_between_requests": manager_calls_on_answer,
        "second_request": len(second),
        "liaison_saw_answer": any(answer in prompt for prompt in liaison_prompts),
        "manager_saw_answer": any(r.sees_marker for r in records),
        "outputs": [str(o) for o in (approved.get_outputs() or [])],
    }


# ---------------------------------------------------------------------------
# S4 — resume a pending plan review in a NEW PROCESS (plan U12)
# ---------------------------------------------------------------------------


async def checkpoint_until_plan_review(storage_dir: str) -> dict[str, Any]:
    """Run until the plan review stops the workflow, leaving checkpoints on disk.

    This is the FIRST half of S4 and runs in the parent process; the resume half must run in a
    separate interpreter (``spikes.e_magentic_resume``), because a resume that quietly rode on
    live in-process objects would prove nothing about the asynchronous file inbox U12 needs.
    """
    records: list[ExplorationCallRecord] = []
    workflow = _plan_review_workflow(records, storage_dir)
    result = await workflow.run(TASK_ALPHA)
    pending = _pending_requests(result)
    checkpoints = await _checkpoint_storage(storage_dir).list_checkpoints(
        workflow_name=workflow.name
    )
    return {
        "request_id": pending[0].request_id if pending else None,
        "checkpoint_ids": [c.checkpoint_id for c in checkpoints],
        "outputs": [str(o) for o in (result.get_outputs() or [])],
    }


# Measured, and a real cost of the asynchronous HITL door: ``FileCheckpointStorage`` refuses to
# unpickle a plan-review request unless its type is declared. Without this the checkpoint file is
# written but comes back UNREADABLE ("Checkpoint deserialization blocked for type ..."), and the
# listing is empty — a resume that fails as an absence rather than as an error, which is exactly
# the shape the fourth face of the verification law warns about. Both processes must declare it.
_ALLOWED_CHECKPOINT_TYPES = [
    "agent_framework_orchestrations._magentic:MagenticPlanReviewRequest",
    "agent_framework_orchestrations._magentic:MagenticPlanReviewResponse",
]


def _checkpoint_storage(storage_dir: str) -> FileCheckpointStorage:
    return FileCheckpointStorage(storage_dir, allowed_checkpoint_types=_ALLOWED_CHECKPOINT_TYPES)


def _plan_review_workflow(records: list[ExplorationCallRecord], storage_dir: str) -> Any:
    """The workflow BOTH processes build — identical construction, so the only thing carried
    across the process boundary is the checkpoint on disk."""
    return (
        MagenticBuilder(
            participants=[_worker_agent(_worker_client([]))],
            manager_agent=_manager_agent(_manager_client(records)),
            max_round_count=6,
            enable_plan_review=True,
        )
        .with_checkpointing(_checkpoint_storage(storage_dir))
        .build()
    )


async def resume_from_checkpoint(
    storage_dir: str, *, request_id: str, checkpoint_id: str
) -> dict[str, Any]:
    """The SECOND half of S4, called by ``spikes.e_magentic_resume`` in a fresh interpreter.

    **Measured, and it contradicts the plan's E-table:** ``checkpoint_storage=`` on ``run()`` is
    NOT the load-bearing seam here — removing it leaves the whole suite green (920 passed),
    because ``.with_checkpointing(...)`` on the builder already gave this workflow its storage.
    The two arguments that ARE load-bearing are ``checkpoint_id=`` (drop it → red) and the
    builder's ``.with_checkpointing(...)`` (drop it → red). It is passed anyway, explicitly,
    because an exploration layer that builds its workflow WITHOUT checkpointing and resumes by
    handing storage in at call time is a legitimate second shape — but a criterion that names it
    as the detach point would be a gate that cannot go red.
    """
    records: list[ExplorationCallRecord] = []
    workflow = _plan_review_workflow(records, storage_dir)
    result = await workflow.run(
        responses={request_id: MagenticPlanReviewResponse.approve()},
        checkpoint_id=checkpoint_id,
        checkpoint_storage=_checkpoint_storage(storage_dir),
    )
    return {
        "manager_kinds": [r.kind for r in records],
        "pending_after_resume": len(_pending_requests(result)),
        "outputs": [str(o) for o in (result.get_outputs() or [])],
    }


def run_resume_subprocess(
    storage_dir: str, *, request_id: str, checkpoint_id: str
) -> dict[str, Any]:
    """Launch ``spikes.e_magentic_resume`` in a FRESH interpreter and parse its one JSON line.

    ``sys.executable`` is this venv's Python, and the repo root is the working directory, so the
    child imports the same tree the parent did without any path juggling. A non-zero exit is
    surfaced with the child's stderr attached: a resume that failed must read as a failed resume,
    never as an empty result.
    """
    completed = subprocess.run(
        [sys.executable, "-m", "spikes.e_magentic_resume", storage_dir, request_id, checkpoint_id],
        capture_output=True,
        text=True,
        cwd=str(Path(__file__).resolve().parents[1]),
    )
    if completed.returncode != 0:
        raise RuntimeError(
            f"resume subprocess exited {completed.returncode}: {completed.stderr.strip()}"
        )
    return dict(json.loads(completed.stdout.strip().splitlines()[-1]))


def micro_bundle_dir() -> str:
    """The repo's own anchored micro bundle — the one bundle that ships BOTH a cost baseline and
    an IR projection, so a latency number measured here is measured through the WHOLE gate
    (stage 0 reconciliation + CBC solve + 512-sample Monte Carlo), not a subset of it."""
    return str(
        Path(portfolio_optimiser.__file__).parent
        / "data"
        / "bundles"
        / "bygg-energi-baseline-mikro"
    )


def micro_proposal() -> SavingsProposal:
    """The bundle's IR projection as a validated ``SavingsProposal``, carrying an assumption band.

    The band is not decoration: without one, ``validator._monte_carlo`` falls back to each item's
    own ``unit_cost`` and every draw is identical — a cheaper computation than any real hypothesis
    would trigger, so a latency measured without it would understate the in-loop cost.
    """
    projection = dict(load_ir_projection(micro_bundle_dir()))
    projection.pop("_note", None)
    item = projection["affected_items"][0]
    unit_cost = float(item["unit_cost"])
    projection["assumptions"] = {item["code"]: (unit_cost * 0.9, unit_cost * 1.1)}
    return SavingsProposal.model_validate(projection)


def validator_latency_seconds(*, runs: int) -> tuple[float, int]:
    """S5: median wall-clock of one ``validate_proposal`` against the micro reference bundle —
    the cost the hypothesiser's ``quick_validate`` tool would pay per call, per hypothesis."""
    baseline = load_optional_cost_baseline(micro_bundle_dir())
    proposal = micro_proposal()

    timings: list[float] = []
    for _ in range(runs):
        start = time.perf_counter()
        validate_proposal(proposal, baseline=baseline)
        timings.append(time.perf_counter() - start)
    return statistics.median(timings), len(timings)