portfolio-optimiser/src/portfolio_optimiser/explore.py

"""U4 + U13-synchronous — the Magentic exploration loop, sitting OVER the normative pipeline.

**One sentence of design, and it is not to be reopened without a measurement.** Magentic is laid
over the eight-step loop, never inside Step 3: ``prompt + knowledge bases -> Mandate ->
run_project(mandate=...)``, with that last call byte-for-byte the one it already was. The manager
is free to choose which base to open and which hypothesis to shape next; what leaves that freedom
is a ``mandate.Mandate`` — approaches worth *testing*. The exploration's own final answer is RAW,
never a proposal, and ``explore()`` writes to neither the outbox nor the wiki. Every number that
survives is gated by ``validate_proposal`` inside ``run_project``, in the same blocking gate as
today, and ``shared/method-spec.md`` §3's maker-checker debate is untouched (it is commons-owned
and normative; replacing it would need an amendment, not a module).

**Three levels of guarantee, stated so nobody reads more into the loop than is there.** The
``quick_validate`` tool the hypothesiser calls in-loop is level 1: the SAME ``validate_proposal``,
against the SAME baseline, but its verdict is advisory — it never becomes provenance. Level 2 is
the pipeline, where each approach is validated and stamped. Level 3 is write access, which only
the pipeline has.

This module imports ``agent_framework.orchestrations`` and therefore may never be imported from
``okf.py`` / ``mandate.py`` / ``hitl.py``, which are held framework-neutral by
``tests/test_okf.py::test_okf_is_maf_free``.
"""

from __future__ import annotations

import json
import sys
from collections.abc import Awaitable, Callable, Mapping, Sequence
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Final, Literal, TextIO

from agent_framework import (
    Agent,
    BaseChatClient,
    FileCheckpointStorage,
    FunctionInvocationContext,
    FunctionMiddleware,
    FunctionTool,
    tool,
)
from agent_framework.orchestrations import (
    MagenticBuilder,
    MagenticOrchestratorEventType,
    MagenticPlanReviewResponse,
)
from pydantic import BaseModel, Field, ValidationError, model_validator

from portfolio_optimiser import okf
from portfolio_optimiser.backends import Profile
from portfolio_optimiser.budget import Budget, BudgetExceeded, BudgetMiddleware, TokenMeter
from portfolio_optimiser.ir import SavingsProposal
from portfolio_optimiser.mandate import OWN_PROPOSAL_ID, Approach, Mandate
from portfolio_optimiser.retrieval import safe_resolve
from portfolio_optimiser.tracing import exploration_tracer
from portfolio_optimiser.validator import Rejection, validate_proposal


class ExplorationContract(BaseModel):
    """The stated bounds of ONE exploration. Every field is required; none has a default.

    Mirrors ``contracts.TerminationContract`` in spirit and goes further in one respect: there is
    nothing to inherit. ``MagenticBuilder`` defaults ``max_round_count`` to ``None`` (unbounded)
    and ``max_reset_count`` to ``None`` (unlimited), so a field left out here would not fall back
    to something conservative — it would fall back to the one shape ``shared/method-spec.md`` §8
    forbids outright, an unbounded loop. A default would also assert an intent nobody stated,
    which is the ground ``ProvenanceStamp.cost_baseline_anchored`` is required on.

    ``max_plan_revisions`` earns its place by measurement, not symmetry (§ F, A3): a plan-review
    ``revise`` costs two manager calls, emits **no** progress ledger and consumes **no** round,
    then asks again. Under the round cap alone an always-revising expert is an unbounded spend
    that the round counter never sees. ``0`` is meaningful and allowed: the plan must be approved
    as first written or the exploration stops.
    """

    #: Hard cap on orchestration rounds (``MagenticBuilder(max_round_count=...)``).
    max_rounds: int = Field(gt=0)
    #: Hard cap on tokens, enforced by ``BudgetMiddleware`` on EVERY agent, manager included.
    max_tokens: int = Field(gt=0)
    #: Consecutive no-progress rounds tolerated before the manager resets and replans. ``0`` is
    #: allowed and means "reset on the first round that reports no progress": the orchestrator's
    #: check is STRICT (``stall_count > max_stall_count``, ``_magentic.py:1118``) and the counter is
    #: incremented ahead of it, so zero is strictness rather than self-defeat.
    max_stall_count: int = Field(ge=0)
    #: Resets tolerated before the exploration is declared stalled. Must be POSITIVE, and that is a
    #: MEASUREMENT: the limit check is ``reset_count >= max_reset_count`` (``:1243``) against a
    #: counter starting at zero, so a cap of zero is already met before the first round. Measured
    #: against the installed stack, ``max_reset_count=0`` produces only the ``facts`` and ``plan``
    #: manager calls, **zero** progress-ledger events, and the canonical "maximum reset count"
    #: termination — an exploration that explored nothing, dressed as a stall that never happened.
    max_reset_count: int = Field(gt=0)
    #: Plan revisions the reviewer may ask for before the exploration stops. See the class note.
    max_plan_revisions: int = Field(ge=0)
    #: Whether a human/persona signs the plan off before the loop runs (the U13 synchronous door).
    enable_plan_review: bool

    @model_validator(mode="after")
    def _a_revision_cap_needs_a_review_to_cap(self) -> "ExplorationContract":
        """A revision cap with the review switched off bounds an event that cannot occur.

        Refused rather than dropped: a caller who wrote ``max_plan_revisions=3`` believes they
        bounded something, and silently ignoring it is the failure mode this repo's flag contract
        exists to prevent (``--embedder-config requires --semantic-retrieval``, same rule). The
        coherent way to say "no reviews" is ``enable_plan_review=False, max_plan_revisions=0``.
        """
        if not self.enable_plan_review and self.max_plan_revisions:
            raise ValueError(
                f"max_plan_revisions={self.max_plan_revisions} bounds plan revisions, but "
                "enable_plan_review is false so no plan review is ever requested and no revision "
                "can occur; set max_plan_revisions=0 or enable the review"
            )
        return self


def exploration_notice(result: "ExplorationResult") -> str:
    """The ONE line an exploration says about itself before the pipeline takes over.

    Always a line, never an omission — unlike ``cost_baseline_notice`` and
    ``skipped_links_notice``, which qualify a run that happened anyway. This one is printed only on
    a surface where ``--explore`` was asked for, so there is no run it could speak for silently.

    ``stop`` is named when there is one, because a mandate shaped by a loop that gave up is a
    smaller mandate than the same loop finishing would have produced, and nothing else on stdout
    would say so.
    """
    ended = "concluded" if result.stop is None else f"stopped ({result.stop})"
    return (
        f"Exploration: {ended} after {len(result.ledger_log)} round(s); "
        f"{len(result.mandate.approaches)} approach(es) to evaluate"
    )


def parked_notice(parked: "ParkedExploration", *, run_id: str) -> str:
    """The ONE thing a parked exploration says at the terminal it stopped on.

    Always a line, for the reason ``exploration_notice`` always is: it is printed only where the
    asynchronous door was asked for. It names the run id, because that is the single coordinate
    ``--resume`` takes, and it is the one an operator will be looking for weeks later.

    It does NOT print the plan. The plan is in the question artefact, where the expert who has to
    read it will be looking — and this terminal belongs to whoever STARTED the run, who is not
    necessarily them.
    """
    return (
        f"Exploration parked: plan review {parked.index} of run {run_id!r} is waiting for a "
        f"human. The question is in {run_id}-plan-review.json; answer it with "
        f"{run_id}-plan-review-answer.json in a review inbox, then --resume {run_id}"
    )


def load_exploration_contract(path: str | Path) -> ExplorationContract:
    """Fail-fast standalone loader for an exploration's bounds (mirrors ``mandate.load_mandate``).

    The bounds are authoritative startup input, and the ONE thing that keeps a Magentic loop from
    being unbounded — so a missing or malformed file refuses the run rather than degrading to
    defaults, which is the shape ``ExplorationContract`` refuses to have in the first place.

    :raises FileNotFoundError: ``path`` does not point at an existing file.
    :raises pydantic.ValidationError: the content is not JSON, or violates the contract.
    """
    p = Path(path)
    if not p.is_file():
        raise FileNotFoundError(f"exploration config not found: {str(path)!r}")
    return ExplorationContract.model_validate_json(p.read_text(encoding="utf-8"))


#: The manager's role name. Not a participant: the manager plans, picks the next speaker and
#: keeps the progress ledger, and ``BudgetMiddleware`` rides on it like on everyone else (A1,
#: measured green — without that, the most talkative agent in the loop would be the one outside
#: the token cap).
MANAGER_ROLE: Final = "manager"
#: Reads the knowledge bases PROGRESSIVELY (§3 Steg 1) — index, then whatever the index links.
NAVIGATOR_ROLE: Final = "navigator"
#: Shapes ONE candidate direction at a time and may have its numbers advisory-checked in-loop.
HYPOTHESISER_ROLE: Final = "hypothesiser"

#: The participants, in build order. The manager's ``next_speaker`` is validated against exactly
#: these names: a ledger naming anyone else makes the orchestrator emit a final answer having
#: asked NOBODY (measured, ``_magentic.py:1128-1131``), which is a plausible answer produced by
#: zero work — the hazard class this repo retired ``E2`` for.
PARTICIPANT_ROLES: Final = (NAVIGATOR_ROLE, HYPOTHESISER_ROLE)

#: The span every exploration records its decisions under. One span per exploration, with the
#: manager's decisions as span EVENTS on it, because those are moments INSIDE one activity rather
#: than activities of their own — and because a reader wants them in order, on one timeline.
EXPLORATION_SPAN: Final = "exploration"

#: The line prefix that turns a hypothesiser turn into a candidate direction. A marker rather than
#: "parse every turn as JSON" because most turns legitimately are not hypotheses — the agent also
#: reasons out loud. That distinction is what lets an UNPARSEABLE marked line be a hard error
#: instead of a silence: with no marker there is nothing to be silent about.
HYPOTHESIS_MARKER: Final = "HYPOTHESIS:"

_INSTRUCTIONS: Final = {
    NAVIGATOR_ROLE: (
        "You read the project's knowledge bases. Use list_bundles to see what exists, then "
        "read_bundle to open ONE at a time and read_file to follow a specific document. Quote "
        "what you found; never guess at content you have not read."
    ),
    HYPOTHESISER_ROLE: (
        "You shape ONE candidate cost-saving direction at a time from what the navigator found. "
        "You may call quick_validate to sanity-check a candidate's numbers; its verdict is "
        "ADVISORY and is not the project's decision. When you commit to a direction, end your "
        f'turn with a line of the form: {HYPOTHESIS_MARKER} {{"label": "<short name>", '
        '"rationale": "<why this project, in your own words>"}'
    ),
}


class ExplorationError(RuntimeError):
    """The exploration cannot be honoured as configured, or produced something unreadable."""


@dataclass(frozen=True)
class LedgerEntry:
    """One progress-ledger round, reduced to the five fields the manager steers on (C.2).

    ``speaker_known`` is not part of the ledger MAF produces — it is this layer's reading of it,
    and the reason the whole entry is recorded rather than counted. A ``next_speaker`` matching no
    participant does not raise: the orchestrator logs a warning and jumps straight to a final
    answer, so the run returns a plausible result that nobody worked for. Recording the fact is
    what lets ``explore()`` refuse to hand that result onward as if it had been explored.
    """

    round_index: int
    is_request_satisfied: bool
    is_in_loop: bool
    is_progress_being_made: bool
    next_speaker: str
    instruction_or_question: str
    speaker_known: bool


@dataclass(frozen=True)
class PlanReview:
    """One synchronous plan-review round trip (U13, door 2 of § C.6).

    ``is_stalled`` is carried because the same request type serves two very different moments:
    the initial sign-off, and a re-review after the manager reset and replanned. A reviewer that
    cannot tell them apart cannot answer the second one usefully.
    """

    index: int
    plan: str
    is_stalled: bool
    decision: Literal["approve", "revise"]
    feedback: str = ""


@dataclass(frozen=True)
class QuickValidation:
    """One advisory (level 1) verdict, as the tool answered it.

    ``proposal_json`` is kept VERBATIM for the reason the parse-failure capture keeps the raw reply
    verbatim: the operative question after a run is what the hypothesiser actually asked about, and
    a re-serialised form would answer a different one.
    """

    bundle_id: str
    proposal_json: str
    verdict: Mapping[str, Any]


@dataclass(frozen=True)
class ToolCall:
    """One exploration tool invocation: what was asked for, never what came back.

    ``bundle_id`` is the base the call named, and is ``""`` for a tool that takes none
    (``list_bundles``). The RESULT is deliberately absent: it is the base's content, which is the
    very thing that is too big to ride along (MAJOR-3 measured it at 73-93 % of all prompt tokens),
    and a trace carrying it would be a second copy of the context rather than a record of the run.
    """

    name: str
    bundle_id: str


def _bundle_argument(arguments: Any) -> str:
    """The ``bundle_id`` a call named, from either shape ``FunctionInvocationContext`` allows.

    ``arguments`` is typed ``BaseModel | Mapping[str, Any]`` (measured against the installed
    signature), so both are read rather than one being assumed. A tool without the parameter — or
    a value that is not a string — yields ``""``: the recorder describes the call, and inventing a
    label for a base that was never named would be the false-attribution that ``ToolCallRecorder``
    refuses for unconfigured tools.
    """
    if isinstance(arguments, Mapping):
        value: Any = arguments.get("bundle_id")
    else:
        value = getattr(arguments, "bundle_id", None)
    return value if isinstance(value, str) else ""


class ExplorationToolRecorder(FunctionMiddleware):
    """Records WHICH exploration tool an agent actually called, in the order it called them.

    **A sibling of ``mcp_tools.ToolCallRecorder``, never a reuse of it** — the same shape, one
    layer over: that one observes the debate's EXTERNAL tool calls, this one the exploration's
    IN-PROCESS ones. Their invariants are opposites and generalising one to serve both would break
    the other. ``ToolCallRecorder`` filters to configured servers, de-duplicates per
    ``(server, tool)`` and returns SORTED, because its record is an egress claim stamped into a
    byte-deterministic artefact. This one keeps every call in INVOCATION ORDER without dedup,
    because the question it answers is the opposite one: did the navigator open anything, and in
    what sequence. A sorted, de-duplicated set cannot tell a rehearsal that read a base from one
    that only listed them, which is the whole of MAJOR-1.

    It observes only — ``call_next`` is always awaited, and nothing here can block, alter or
    short-circuit an invocation. A trace that changed the run it traces would not be a trace.
    """

    def __init__(self, sink: list[ToolCall]) -> None:
        self._sink = sink

    async def process(
        self, context: FunctionInvocationContext, call_next: Callable[[], Awaitable[None]]
    ) -> None:
        name = getattr(getattr(context, "function", None), "name", None)
        if isinstance(name, str):
            self._sink.append(
                ToolCall(name=name, bundle_id=_bundle_argument(getattr(context, "arguments", None)))
            )
        await call_next()


@dataclass
class ExplorationTrace:
    """The CALLER-owned accumulator for everything one exploration produced along the way.

    **Why a caller-owned sink and not a return value** — the same measurement that shaped the
    parse-failure capture (Fase 1b, funn 1), one layer up. ``explore()`` raises ``BudgetExceeded``
    at its round cap, and a token cap fires from inside ``BudgetMiddleware`` mid-run; on both paths
    no ``ExplorationResult`` is ever constructed. § C.2 requires the exploration artefact to be
    readable "uansett hvilken vakt som fyrte", so the evidence has to live somewhere that survives
    the ending — which a return value, by definition, does not. Step 5's opposite rule ("a returned
    value, never an out-parameter") governs a value that DOES reach the caller; copying it blindly
    here would rebuild the very defect it was written against.

    ``ExplorationResult.ledger_log`` and ``.plan_reviews`` are built FROM these lists rather than
    accumulated beside them: two containers holding one fact drift (kø-(p)), and a drifted pair
    would let the returned result and the written artefact describe different runs.
    """

    ledger: list[LedgerEntry] = field(default_factory=list)
    plan_reviews: list[PlanReview] = field(default_factory=list)
    quick_validations: list[QuickValidation] = field(default_factory=list)
    #: Every exploration tool call, in order (MAJOR-1). It sits BESIDE ``quick_validations``
    #: rather than inside it: that list is the level-1 VERDICTS the hypothesiser saw, this is
    #: whether any base was opened at all. An offline rehearsal that called nothing looks exactly
    #: like a successful one on every other field, which is what made the dress rehearsal vacuous
    #: by construction and unreadable after the fact.
    tool_calls: list[ToolCall] = field(default_factory=list)
    #: Tokens spent so far, refreshed as the loop turns rather than written once at the end. The
    #: meter is internal to ``explore``, so this is the only way the artefact can report a spend —
    #: and updating it per iteration is what makes it readable for a run a cap cut short, which is
    #: the same reason the sink exists at all. Across a park it is what the resumed leg adds to.
    tokens_spent: int = 0


def trace_payload(trace: ExplorationTrace, *, stop: str | None, completed: bool) -> dict[str, Any]:
    """The ONE rendering of a trace into plain data for ``outbox.write_exploration``.

    Plain mappings only, so the RAW output layer stays MAF-free (the ``write_parse_failures``
    precedent — ``outbox.py`` may not import this module).

    ``completed`` is a required field rather than an inference from ``stop``. With no result there
    is no stop, and a ``stop: null`` meaning BOTH "concluded normally" and "we never found out"
    is exactly the silence ``ProvenanceStamp.cost_baseline_anchored`` was made required to close.
    """
    return {
        "completed": completed,
        "stop": stop,
        "tokens_spent": trace.tokens_spent,
        "rounds": [
            {
                "round_index": entry.round_index,
                "is_request_satisfied": entry.is_request_satisfied,
                "is_in_loop": entry.is_in_loop,
                "is_progress_being_made": entry.is_progress_being_made,
                "next_speaker": entry.next_speaker,
                "instruction_or_question": entry.instruction_or_question,
                "speaker_known": entry.speaker_known,
            }
            for entry in trace.ledger
        ],
        "plan_reviews": [
            {
                "index": review.index,
                "plan": review.plan,
                "is_stalled": review.is_stalled,
                "decision": review.decision,
                "feedback": review.feedback,
            }
            for review in trace.plan_reviews
        ],
        "quick_validations": [
            {
                "bundle_id": call.bundle_id,
                "proposal_json": call.proposal_json,
                "verdict": dict(call.verdict),
            }
            for call in trace.quick_validations
        ],
        "tool_calls": [
            {"name": call.name, "bundle_id": call.bundle_id} for call in trace.tool_calls
        ],
    }


#: Why an exploration ended without a satisfied request. ``None`` means it concluded normally.
#: Resource exhaustion is NOT here: tokens and rounds raise ``BudgetExceeded`` (the 429 channel),
#: because "we ran out" and "we finished, unsatisfied" are the two things S3.4 split apart and a
#: single field would fuse back together.
ExplorationStop = Literal["stalled", "plan_revisions_exhausted", "unknown_speaker"]


@dataclass(frozen=True)
class ExplorationResult:
    """What one exploration produced. The mandate is the product; the rest is the evidence.

    ``mandate`` is ALWAYS present, including on a stop: an exploration that was cut short still
    carries the expert's seed approaches forward, because door 1 of § C.6 is a preservation rule
    and not a reward for finishing. ``stop`` is what says the mandate is smaller than it might
    have been, and why.
    """

    mandate: Mandate
    ledger_log: tuple[LedgerEntry, ...]
    stop: ExplorationStop | None
    plan_reviews: tuple[PlanReview, ...]


@dataclass(frozen=True)
class PlanReviewRequest:
    """What the reviewer is shown before the loop is allowed to run."""

    index: int
    plan: str
    current_progress: str
    is_stalled: bool


@dataclass(frozen=True)
class PlanReviewDecision:
    """The reviewer's answer. ``feedback is None`` means approve.

    Two named constructors mirroring ``MagenticPlanReviewResponse.approve()/.revise()``, so a
    persona or operator answering a review never has to import ``agent_framework`` — the adapter
    is one-directional and lives in exactly one place.
    """

    feedback: str | None

    @staticmethod
    def approve() -> PlanReviewDecision:
        return PlanReviewDecision(feedback=None)

    @staticmethod
    def revise(feedback: str) -> PlanReviewDecision:
        if not feedback.strip():
            raise ValueError("a revision must say what to revise; use approve() to sign off")
        return PlanReviewDecision(feedback=feedback)


#: The synchronous HITL seam (U13). Given the request, answer it. Called in-process, so the
#: exploration blocks on it exactly as a human at a terminal would block the loop.
PlanReviewer = Callable[[PlanReviewRequest], PlanReviewDecision]


class PlanReviewInputError(ExplorationError):
    """A terminal plan review was left without an answer: the input ended mid-review.

    A distinct type rather than a distinguishing message, for the reason ``_classify_stop`` reads
    counts instead of the termination prose: a caller deciding what happened should never have to
    match wording. It is an ``ExplorationError`` (a ``RuntimeError``) because the run FAILED — the
    caller's argv was fine and the loop had already started spending; the same channel an
    unreadable marked hypothesis leaves by.
    """


#: The closed answer vocabulary of the terminal door. Two words, matched structurally.
_APPROVE_ANSWER: Final = "approve"
_REVISE_ANSWER: Final = "revise"


def terminal_plan_reviewer(
    *, stream_in: TextIO | None = None, stream_out: TextIO | None = None
) -> PlanReviewer:
    """A ``PlanReviewer`` that asks the operator at a terminal and reads their typed answer.

    This is the door that makes målbilde §3's "still spørsmål, be om svar, bruke svarene" reachable
    without importing the package (F4): ``run.py``'s ``--plan-review`` builds one of these and
    hands it to ``explore()``. Blocking is not an oversight — ``explore()`` calls the reviewer
    synchronously (it is not awaited), so the loop waits on the human exactly as the ``PlanReviewer``
    contract says. That is also the reason the hosted surface keeps refusing the review: there,
    blocking the reviewer would block the event loop that answers ``/readiness``.

    **The streams are resolved at CALL time, not here** (the ``shared_root()`` idiom): a factory
    that captured ``sys.stdin`` at construction could not be driven by a caller — or a test — that
    replaces the stream afterwards, and the only way left to exercise the door would be a
    subprocess.

    **Fail-closed on the operator's own input.** ``approve`` signs off; ``revise <what to change>``
    sends the words back to the manager. Anything else — a blank line, a typo, a bare ``revise`` —
    is asked AGAIN, never taken as a decision. End of input raises ``PlanReviewInputError``:
    reading silence as approval would let an autonomous loop run on a plan no human signed, and do
    it invisibly. Validation, NEVER repair (the ``write_concept_file`` rule).
    """

    def review(request: PlanReviewRequest) -> PlanReviewDecision:
        source = sys.stdin if stream_in is None else stream_in
        sink = sys.stdout if stream_out is None else stream_out
        stalled = " (a RE-PLAN after a stall)" if request.is_stalled else ""
        print(f"\nPLAN REVIEW #{request.index + 1}{stalled}", file=sink)
        print("--- the plan the exploration would run ---", file=sink)
        print(request.plan, file=sink)
        if request.current_progress.strip():
            print("--- progress so far ---", file=sink)
            print(request.current_progress, file=sink)
        while True:
            print(
                f'Answer "{_APPROVE_ANSWER}" to sign it off, '
                f'or "{_REVISE_ANSWER} <what to change>": ',
                file=sink,
            )
            sink.flush()
            line = source.readline()
            if line == "":
                raise PlanReviewInputError(
                    "the plan review reached end of input without an answer. Silence is not a "
                    "sign-off: the exploration will not run a plan nobody approved"
                )
            answer = line.strip()
            if answer == _APPROVE_ANSWER:
                return PlanReviewDecision.approve()
            verb, _, feedback = answer.partition(" ")
            if verb == _REVISE_ANSWER and feedback.strip():
                return PlanReviewDecision.revise(feedback.strip())
            print(f"Not an answer: {answer!r}.", file=sink)

    return review


# ---------------------------------------------------------------------------------------------
# U12 — the ASYNCHRONOUS half of the same door: a review answered days later, in another process.
# ---------------------------------------------------------------------------------------------

#: The two types a plan-review checkpoint carries, in ``"module:qualname"`` form.
#:
#: **This tuple is the whole of the measured trap** (plan § F, A4; re-measured against the
#: installed source, ``_workflows/_checkpoint.py:386-388``): ``FileCheckpointStorage`` runs a
#: restricted unpickler, and ``list_checkpoints`` swallows a blocked type into a ``logger.warning``
#: and returns an EMPTY list. Omit either name and the checkpoint is written but comes back
#: unreadable, so a resume fails as an ABSENCE — "nothing to resume" — rather than as an error.
#: ONE copy, used by every process that touches this storage, because both the writing process and
#: the resuming one must declare them and a second copy is the drift kø-(p) exists to prevent.
_ALLOWED_CHECKPOINT_TYPES: Final[tuple[str, ...]] = (
    "agent_framework_orchestrations._magentic:MagenticPlanReviewRequest",
    "agent_framework_orchestrations._magentic:MagenticPlanReviewResponse",
)


def checkpoint_storage(checkpoint_dir: str | Path) -> FileCheckpointStorage:
    """The ONE construction site for the exploration's checkpoint storage.

    A caller that built its own ``FileCheckpointStorage`` would have to remember the allow-list
    above, and forgetting it is invisible (see ``_ALLOWED_CHECKPOINT_TYPES``). Routing every
    construction through here makes "both processes declare the types" a structural property
    rather than a convention two call sites have to keep.
    """
    return FileCheckpointStorage(
        str(checkpoint_dir), allowed_checkpoint_types=list(_ALLOWED_CHECKPOINT_TYPES)
    )


class CheckpointUnreadable(ExplorationError):
    """The exploration stopped at a review but left nothing a later process could resume from.

    Raised where the framework is silent: an empty listing means the checkpoint could not be read
    back, and parking anyway would hand an expert a question whose answer can never be applied.
    Failing here is the fourth face of the verification law written into our own surface.
    """


class ParkedStateError(ExplorationError):
    """A parked-state file that cannot be read as one.

    Fail-fast, NOT the tolerant RAW-inbox rule: this file is the run's own suspended state, the
    same class as the spend file ``read_spend`` refuses to read loosely. Treating a malformed one
    as "no parked run" would silently drop an exploration a human is waiting to answer.
    """


@dataclass(frozen=True)
class ParkedExploration:
    """Everything needed to resume one suspended exploration in a process that never saw it.

    Two files cross the boundary and they own different halves. MAF's checkpoint holds the
    WORKFLOW's state (the manager's ledgers, the pending request); this holds the EXPLORATION
    LAYER's — what has been spent, what the loop has already found, and which question is open.
    Neither can reconstruct the other, so both are named here rather than one being inferred.

    ``tokens_spent`` and ``ledger`` are carried for a reason that is not bookkeeping. Both budget
    channels live in the process: a resume with a fresh ``TokenMeter`` and an empty ledger would
    get the whole cap AGAIN, once per park — unbounded consumption behind guards that all report
    themselves satisfied, which is the class S3.4 split apart. Carrying them makes the cap span
    the suspension.

    ``hypotheses`` are the marked turns the loop produced BEFORE parking. They are verbatim, and
    they are carried for the same reason: minting the mandate from only what the resuming process
    observed would silently drop everything found before a stalled re-review.
    """

    prompt: str
    request_id: str
    checkpoint_id: str
    index: int
    plan: str
    current_progress: str
    is_stalled: bool
    bundle_dirs: tuple[str, ...]
    contract: ExplorationContract
    ledger: tuple[LedgerEntry, ...]
    plan_reviews: tuple[PlanReview, ...]
    hypotheses: tuple[str, ...]
    tokens_spent: int
    replans: int


class PlanReviewParked(ExplorationError):
    """The exploration is suspended at a plan review, waiting for a human.

    **A raise rather than a return, and the argument is the 429 one** (kø-(y), 14.08). A parked run
    produced NO mandate: the loop is stopped mid-plan and nothing has been explored yet. Handing
    back an ``ExplorationResult`` would let an automated caller book "explored" for a run that
    explored nothing — the same reason an exhausted budget is not a 200 even though it is not a
    crash either. The coordinates travel as STRUCTURE on ``parked``, never as ``str(exc)``.
    """

    def __init__(self, parked: ParkedExploration) -> None:
        super().__init__(
            f"exploration parked at plan review {parked.index} "
            f"(request {parked.request_id}, checkpoint {parked.checkpoint_id})"
        )
        self.parked = parked


def parked_payload(parked: ParkedExploration) -> dict[str, Any]:
    """The ONE rendering of a parked exploration into plain data for ``outbox.write_plan_review``.

    Plain mappings only, so the RAW output layer stays MAF-free — the ``trace_payload`` precedent,
    for the same reason: ``outbox.py`` may not import this module.
    """
    return {
        "prompt": parked.prompt,
        "request_id": parked.request_id,
        "checkpoint_id": parked.checkpoint_id,
        "index": parked.index,
        "plan": parked.plan,
        "current_progress": parked.current_progress,
        "is_stalled": parked.is_stalled,
        "bundle_dirs": list(parked.bundle_dirs),
        "contract": parked.contract.model_dump(),
        "ledger": [
            {
                "round_index": entry.round_index,
                "is_request_satisfied": entry.is_request_satisfied,
                "is_in_loop": entry.is_in_loop,
                "is_progress_being_made": entry.is_progress_being_made,
                "next_speaker": entry.next_speaker,
                "instruction_or_question": entry.instruction_or_question,
                "speaker_known": entry.speaker_known,
            }
            for entry in parked.ledger
        ],
        "plan_reviews": [
            {
                "index": review.index,
                "plan": review.plan,
                "is_stalled": review.is_stalled,
                "decision": review.decision,
                "feedback": review.feedback,
            }
            for review in parked.plan_reviews
        ],
        "hypotheses": list(parked.hypotheses),
        "tokens_spent": parked.tokens_spent,
        "replans": parked.replans,
    }


def load_parked(payload: Mapping[str, Any]) -> ParkedExploration:
    """Read a parked-state payload back, fail-fast (``ParkedStateError`` on anything missing).

    Validation, never repair (the ``write_concept_file`` rule): a payload that has lost, say, its
    ``checkpoint_id`` describes a suspension nobody can lift, and defaulting it would produce a
    resume that looks like one and is not.
    """
    try:
        return ParkedExploration(
            prompt=str(payload["prompt"]),
            request_id=str(payload["request_id"]),
            checkpoint_id=str(payload["checkpoint_id"]),
            index=int(payload["index"]),
            plan=str(payload["plan"]),
            current_progress=str(payload["current_progress"]),
            is_stalled=bool(payload["is_stalled"]),
            bundle_dirs=tuple(str(d) for d in payload["bundle_dirs"]),
            contract=ExplorationContract.model_validate(payload["contract"]),
            ledger=tuple(
                LedgerEntry(
                    round_index=int(row["round_index"]),
                    is_request_satisfied=bool(row["is_request_satisfied"]),
                    is_in_loop=bool(row["is_in_loop"]),
                    is_progress_being_made=bool(row["is_progress_being_made"]),
                    next_speaker=str(row["next_speaker"]),
                    instruction_or_question=str(row["instruction_or_question"]),
                    speaker_known=bool(row["speaker_known"]),
                )
                for row in payload["ledger"]
            ),
            plan_reviews=tuple(
                PlanReview(
                    index=int(row["index"]),
                    plan=str(row["plan"]),
                    is_stalled=bool(row["is_stalled"]),
                    decision="revise" if row["decision"] == "revise" else "approve",
                    feedback=str(row["feedback"]),
                )
                for row in payload["plan_reviews"]
            ),
            hypotheses=tuple(str(h) for h in payload["hypotheses"]),
            tokens_spent=int(payload["tokens_spent"]),
            replans=int(payload["replans"]),
        )
    except (KeyError, TypeError, ValueError, ValidationError) as exc:
        raise ParkedStateError(f"parked exploration state is unreadable: {exc}") from exc


# ---------------------------------------------------------------------------------------------
# The tools. Level 1 of the three-guarantee table: real computation, ADVISORY verdicts.
# ---------------------------------------------------------------------------------------------


def _bundle_index(bundle_dirs: Sequence[str]) -> dict[str, str]:
    """Map each knowledge base's id to its directory. The id is the directory's BASENAME.

    A duplicate basename is REFUSED rather than resolved by order: the id is what the manager
    names a base by, and two bases answering to one name would let it read A while believing it
    read B — the S3.2 key-collision class, one layer up.
    """
    index: dict[str, str] = {}
    for raw in bundle_dirs:
        bundle_id = Path(raw).name
        if bundle_id in index:
            raise ExplorationError(
                f"two knowledge bases share the id {bundle_id!r} ({index[bundle_id]!r} and "
                f"{raw!r}); the manager names a base by that id, so it must be unique"
            )
        index[bundle_id] = raw
    return index


def _resolve_bundle(index: Mapping[str, str], bundle_id: str) -> str:
    if bundle_id not in index:
        known = ", ".join(sorted(index)) or "(none configured)"
        raise ExplorationError(f"unknown knowledge base {bundle_id!r}; configured: {known}")
    return index[bundle_id]


#: Characters of the root index body one catalogue entry may carry. The catalogue's job is to let a
#: manager pick a base, not to read one, so the excerpt is a fixed-size window rather than a share
#: of the base: cost then scales with how many bases are configured, which the operator chose, and
#: not with how much they contain, which they did not. The number is deliberately small — a headline
#: and a first line or two — and the ceiling that guards it lives in the TEST, not here, because
#: raising this constant is the regression the gate exists to catch.
_CATALOGUE_EXCERPT_CHARS: Final = 200


def _index_excerpt(body: str) -> tuple[str, bool]:
    """A bounded VERBATIM prefix of an index body, plus whether anything was cut.

    Cut at the last line break inside the budget, never mid-line: half a markdown link is a target a
    model may well try to follow, and a path that never existed is worse than no path. A first line
    longer than the whole budget has no break to cut at, and is cut hard — the bound is the promise.

    Returns the body UNTOUCHED and ``False`` when it already fits: an index that was never cut must
    not be reported as cut (omission, never a lie in either direction — ``cost_baseline_notice``'s
    rule applied to a value instead of a line).
    """
    if len(body) <= _CATALOGUE_EXCERPT_CHARS:
        return body, False
    cut = body.rfind("\n", 0, _CATALOGUE_EXCERPT_CHARS + 1)
    return (body[:cut] if cut > 0 else body[:_CATALOGUE_EXCERPT_CHARS]), True


def navigator_tools(bundle_dirs: Sequence[str]) -> list[FunctionTool]:
    """The navigator's three tools: survey the catalogue, open one base, read one document.

    Progressive disclosure, not stuffing (målbilde §2/§4): ``list_bundles`` never returns content,
    only what each base IS — and crucially whether it ships a ``cost-baseline.json``, because a base
    without one cannot have its hypotheses reconciled against the project's real cost lines, and a
    manager that does not know which bases are anchored cannot plan around it (§ C.7).

    **The catalogue costs O(bases), never O(corpus)**, and that is the rung's whole point. It used
    to return the WHOLE root index body per base plus one object per unfollowed cross-link — both
    corpus-sized — so the cheapest rung of the ladder was the most expensive call in the loop:
    112 116 o200k_base tokens over three flat Vegnormal bases, 124 942 over the 171 branch bases
    that replaced them (measured 25.-26.08; ``tests/test_catalogue_cost_loadbearing.py``). Each
    entry is now bounded by construction: a ``_CATALOGUE_EXCERPT_CHARS``-long VERBATIM prefix of the
    index body plus counts. The full index stays exactly one ``read_file(id, "index.md")`` away, so
    this is a disclosure level, not data loss.

    **Truncation is ANNOUNCED, never silent** — ``index_truncated`` is a field beside the excerpt,
    not a marker glued into it, for the reason ``BudgetExceeded`` carries its triple as fields
    (kø-(y)): a consumer that has to re-parse prose to learn whether it is holding the whole thing
    has been handed a diagnostic it cannot act on. And an unfollowed cross-link keeps a COUNT here
    (session 51's "a skip is tolerated but no longer silent") while its per-link detail stays where
    it is actionable, on ``RunResult.skipped_links`` / ``DryRunReport.skipped_links``.

    ``read_bundle`` returns ``okf.bundle_context``, which EXCLUDES the ``type: verdict`` layer by
    construction — prior verdicts reach a hypothesis only through the gated ExpeL fold inside
    ``run_project``, never by being read as context here.
    """
    index = _bundle_index(bundle_dirs)

    @tool(
        name="list_bundles",
        description=(
            "List the knowledge bases available to this exploration: id, the OPENING of what the "
            "index says, how many documents and prior expert verdicts the base holds, whether it "
            "ships a cost baseline (without one, numbers cannot be reconciled against the "
            "project's own), and how many cross-links could not be followed. The excerpt is cut "
            "when index_truncated is true; read_file(id, 'index.md') gives the whole index."
        ),
    )
    def list_bundles() -> list[dict[str, Any]]:
        catalogue: list[dict[str, Any]] = []
        for bundle_id, bundle_dir in index.items():
            bundle = okf.navigate_bundle(bundle_dir)
            excerpt, truncated = _index_excerpt(bundle.index_summary)
            catalogue.append(
                {
                    "id": bundle_id,
                    "index_excerpt": excerpt,
                    "index_truncated": truncated,
                    "documents": len(bundle.context_files),
                    "verdict_count": len(bundle.verdicts),
                    # Tolerant on CONTENT, fail-fast on the PATH: an operator's bad directory is
                    # refused by navigate_bundle above, while a navigable base that simply has no
                    # baseline is legitimate (load_optional_cost_baseline's own contract).
                    "cost_baseline": okf.load_optional_cost_baseline(bundle_dir) is not None,
                    "unreachable_links": len(bundle.skipped),
                }
            )
        return catalogue

    @tool(
        name="read_bundle",
        description="Open ONE knowledge base by id and read its navigated context.",
    )
    def read_bundle(bundle_id: str) -> str:
        return okf.bundle_context(okf.navigate_bundle(_resolve_bundle(index, bundle_id)))

    @tool(
        name="read_file",
        description="Read ONE document inside a knowledge base, by base id and relative path.",
    )
    def read_file(bundle_id: str, path: str) -> str:
        bundle_dir = _resolve_bundle(index, bundle_id)
        # safe_resolve is the ONE in-/out-of-bundle test in this repo, and it is fail-closed. A
        # model-chosen path is untrusted input by definition, so it goes through the same gate the
        # navigation walk uses rather than a second, laxer check.
        return Path(safe_resolve(bundle_dir, path)).read_text(encoding="utf-8")

    return [list_bundles, read_bundle, read_file]


def quick_validate_tool(
    bundle_dirs: Sequence[str], *, sink: list[QuickValidation] | None = None
) -> FunctionTool:
    """The hypothesiser's in-loop deterministic check — level 1, and advisory by construction.

    It is the SAME ``validate_proposal`` against the SAME baseline the pipeline will use, so the
    numbers it reports are real; what it never becomes is provenance. ``ProvenanceStamp`` is
    written by ``run_project`` alone, and nothing here reaches it.

    Measured cheap (S5: 13.6 ms median through stage 0 + the CBC solve + a 512-sample Monte Carlo,
    on an anchored base with assumption bands), so it may be called freely in the loop and the
    contract carries no latency budget.

    ``anchored`` is reported alongside the verdict for the same reason
    ``ProvenanceStamp.cost_baseline_anchored`` is a required field: a verdict reached without the
    project's own cost lines is a weaker claim, and one that does not say so is the silence S4.0's
    visibility work closed.

    ``sink`` is the caller's ``ExplorationTrace.quick_validations``. These verdicts are the one
    thing ``ExplorationResult`` deliberately does not carry — they are level-1 advisory, and their
    home is the ``{run_id}-exploration.json`` artefact — so without a sink they would leave no
    trace of having been asked for at all.
    """
    index = _bundle_index(bundle_dirs)

    @tool(
        name="quick_validate",
        description=(
            "Run the deterministic validator over a candidate, as an ADVISORY check. Pass the "
            "knowledge base id and the candidate as IR JSON. The verdict is not the project's "
            "decision — the pipeline re-validates and stamps."
        ),
    )
    def quick_validate(bundle_id: str, proposal_json: str) -> dict[str, Any]:
        bundle_dir = _resolve_bundle(index, bundle_id)
        baseline = okf.load_optional_cost_baseline(bundle_dir)
        verdict: dict[str, Any]
        try:
            proposal = SavingsProposal.model_validate_json(proposal_json)
        except ValidationError as exc:
            verdict = {
                "decision": "unparseable",
                "reason": str(exc),
                "anchored": baseline is not None,
            }
        else:
            outcome = validate_proposal(proposal, baseline=baseline)
            if isinstance(outcome, Rejection):
                verdict = {
                    "decision": "rejected",
                    "reason": outcome.reason,
                    "anchored": baseline is not None,
                }
            else:
                verdict = {
                    "decision": "validated",
                    "reason": "",
                    "anchored": baseline is not None,
                    "p10": outcome.p10,
                    "p50": outcome.p50,
                    "p90": outcome.p90,
                }
        # Recorded AFTER the verdict is decided and on EVERY branch — an unparseable candidate is
        # as much a thing the hypothesiser asked about as a validated one. A refused bundle id
        # raises above and is deliberately not recorded: nothing was validated.
        if sink is not None:
            sink.append(
                QuickValidation(bundle_id=bundle_id, proposal_json=proposal_json, verdict=verdict)
            )
        return verdict

    return quick_validate


# ---------------------------------------------------------------------------------------------
# The workflow. One builder, one build, ONE run per exploration (B7 / § C.4).
# ---------------------------------------------------------------------------------------------


def fresh_exploration_workflow(
    client_factory: Callable[[str], BaseChatClient],
    *,
    contract: ExplorationContract,
    bundle_dirs: Sequence[str] = (),
    middleware: Sequence[Any] | None = None,
    quick_validate_sink: list[QuickValidation] | None = None,
    checkpoint_dir: str | None = None,
) -> Any:
    """Build a FRESH Magentic workflow with fresh agents and fresh clients (mirrors
    ``workflow.fresh_workflow``).

    **Fresh builder per exploration, never a reused one.** A built Magentic workflow is
    single-use: a second ``run()`` on it raises ``RuntimeError`` having made zero model calls
    (measured, E1) — which is a safe failure, unlike GroupChat 1.9.0's silently empty second run.
    Reuse is prevented here rather than relied upon to fail.

    ``manager_agent_factory=`` rather than ``manager_agent=``: the eager form constructs the
    manager once and hands the same instance to every ``build()`` (``:1683``, ``:1729-1730``),
    which on orchestrations 1.0.0 leaked one exploration's task into the next. Orchestrations
    1.0.1 removed the manager's persistent session, so that leak is GONE and **no test here can
    tell the two forms apart today** (measured, § F A8: 4/4 → 0/5). The factory form is used
    anyway because it costs nothing and does not depend on an upstream regression fix staying
    fixed — but it is stated plainly rather than gated, because a gate that cannot go red proves
    nothing.

    Every agent carries the SAME ``middleware`` list, manager included. That is the whole of the
    token guarantee: agent-level ``ChatMiddleware`` does fire on the manager's own calls (measured
    A1), and the manager is the most talkative participant in the loop.
    """
    hypothesiser_tools: list[Any] = [quick_validate_tool(bundle_dirs, sink=quick_validate_sink)]
    tools_by_role: dict[str, list[Any]] = {
        NAVIGATOR_ROLE: list(navigator_tools(bundle_dirs)),
        HYPOTHESISER_ROLE: hypothesiser_tools,
    }
    participants = [
        Agent(
            client_factory(role),
            _INSTRUCTIONS[role],
            name=role,
            description=_INSTRUCTIONS[role],
            tools=tools_by_role[role],
            middleware=middleware,
        )
        for role in PARTICIPANT_ROLES
    ]

    def _manager_agent() -> Agent:
        return Agent(
            client_factory(MANAGER_ROLE),
            "You plan and coordinate an exploration of a project's knowledge bases to find "
            "cost-saving directions worth testing. You never state a saving figure yourself.",
            name=MANAGER_ROLE,
            description="plans the exploration",
            middleware=middleware,
        )

    builder = MagenticBuilder(
        participants=participants,
        manager_agent_factory=_manager_agent,
        max_round_count=contract.max_rounds,
        max_stall_count=contract.max_stall_count,
        max_reset_count=contract.max_reset_count,
        enable_plan_review=contract.enable_plan_review,
    )
    if checkpoint_dir is not None:
        # Measured (spike S4, and re-measured in the plan's own E-table correction): it is the
        # BUILDER's ``.with_checkpointing`` that is load-bearing, not ``checkpoint_storage=`` on
        # ``run()`` — dropping the latter leaves the whole suite green. Both processes call THIS
        # function, so the graph they build is identical, which is what lets a checkpoint written
        # by one be restored by the other (``_runner.py:275-279`` matches on the graph signature).
        builder = builder.with_checkpointing(checkpoint_storage(checkpoint_dir))
    return builder.build()


def _truthy(answer: Any) -> bool:
    """Read a ledger answer exactly as the orchestrator does.

    ``MagenticProgressLedgerItem.answer`` is ``str | bool`` and is NOT normalised per field
    (``:292``); the orchestrator then tests it with plain Python truthiness (``:1106``, ``:1112``).
    A smarter coercion here — treating the string ``"false"`` as false, say — would describe a run
    the orchestrator did not have. This layer reports what the loop DID, so it copies the loop's
    own rule rather than improving on it.
    """
    return bool(answer)


def _plan_text(content: Any) -> str:
    """The task-ledger plan as text. ``PLAN_CREATED``/``REPLANNED`` carry a ``Message``."""
    return str(getattr(content, "text", "") or "")


def _absorb(
    result: Any,
    *,
    ledger_log: list[LedgerEntry],
    hypotheses: list[str],
    seen: set[int],
    span: Any,
) -> int:
    """Fold one ``run()``'s events into the running log and onto the trace. Returns this call's
    replans.

    ``seen`` de-duplicates by object identity because MAF may surface the same payload under more
    than one event type; the alternative — pinning one event-type name — would silently under- or
    over-count if that changed. Participant output is matched on ``executor_id``, which is the
    agent's name, exactly as ``run._authored_texts`` matches the debate's ``author_name``.

    **The three span events land here** (U14's deferred half, now that it has a call site). They
    carry the manager's DECISIONS — which plan, who was asked, whether the request was satisfied —
    as separate typed attributes rather than a rendered sentence, so a collector can query them.
    Stated honesty limit: they are recorded as the event stream is folded, i.e. after ``run()``
    returns, so their ORDER is faithful and their timestamps are not the moments the manager acted.
    Emitting live would mean driving the loop through ``run_stream``, which is a different
    measurement from the one the plan-review round trip was proved against.
    """
    replans = 0
    for event in result:
        data = getattr(event, "data", None)
        if data is None or id(data) in seen:
            continue
        event_type = getattr(data, "event_type", None)
        if event_type is not None:
            seen.add(id(data))
            if event_type == MagenticOrchestratorEventType.PLAN_CREATED:
                span.add_event("plan_created", {"plan": _plan_text(data.content)})
            elif event_type == MagenticOrchestratorEventType.REPLANNED:
                replans += 1
                span.add_event("replanned", {"plan": _plan_text(data.content)})
            elif event_type == MagenticOrchestratorEventType.PROGRESS_LEDGER_UPDATED:
                ledger = data.content
                speaker = str(ledger.next_speaker.answer)
                entry = LedgerEntry(
                    round_index=len(ledger_log) + 1,
                    is_request_satisfied=_truthy(ledger.is_request_satisfied.answer),
                    is_in_loop=_truthy(ledger.is_in_loop.answer),
                    is_progress_being_made=_truthy(ledger.is_progress_being_made.answer),
                    next_speaker=speaker,
                    instruction_or_question=str(ledger.instruction_or_question.answer),
                    speaker_known=speaker in PARTICIPANT_ROLES,
                )
                ledger_log.append(entry)
                span.add_event(
                    "progress_ledger_updated",
                    {
                        "round_index": entry.round_index,
                        "is_request_satisfied": entry.is_request_satisfied,
                        "is_in_loop": entry.is_in_loop,
                        "is_progress_being_made": entry.is_progress_being_made,
                        "next_speaker": entry.next_speaker,
                        "speaker_known": entry.speaker_known,
                        "instruction_or_question": entry.instruction_or_question,
                    },
                )
            continue
        if getattr(data, "executor_id", None) == HYPOTHESISER_ROLE:
            response = getattr(data, "agent_response", None)
            text = getattr(response, "text", None)
            if text:
                seen.add(id(data))
                hypotheses.append(text)
    return replans


class HypothesisParseError(ExplorationError):
    """A hypothesiser turn carried the marker and then something unreadable.

    Fail-closed, and the marker is what makes that affordable: most turns legitimately are not
    hypotheses, so an unmarked turn is not a failure and there is nothing to be silent about. A
    MARKED line that will not parse is a claim the loop tried to make and could not — raising it
    is the ``write_concept_file`` rule (validation, never repair) rather than the tolerant RAW
    inbox rule, because this is the product of the run, not a folder anyone may drop things in.
    """


def _resolve_hypothesis_bundle(raw: str, bundle_ids: Sequence[str]) -> str:
    """Which knowledge base a marked hypothesis belongs to (§ C.7), fail-closed.

    Three rules, and the middle one is the whole reason the field can default at all:

    * a stated id must be one that was CONFIGURED — refused by name, never resolved by position,
      exactly as ``_resolve_bundle`` refuses an unknown base for the read tools. A hypothesis that
      names a base must not be treated more leniently than a read of one.
    * a silent marker with at most one base configured resolves to that base (or to ``""`` with
      none). With one configured base there is no other value the field could take, so this is the
      only answer rather than a guess.
    * a silent marker with SEVERAL configured refuses. A marked line is a claim the loop committed
      to; one that cannot be routed is a claim it could not finish making, and the marker is
      precisely what makes that affordable to refuse — an unmarked turn is not a hypothesis, so
      there is no silence being turned into an error.
    """
    if raw:
        if raw not in bundle_ids:
            known = ", ".join(bundle_ids) or "(none configured)"
            raise ExplorationError(
                f"hypothesis names knowledge base {raw!r}, which is not configured for this "
                f"exploration; configured: {known}"
            )
        return raw
    if len(bundle_ids) <= 1:
        return bundle_ids[0] if bundle_ids else ""
    raise HypothesisParseError(
        f"a marked hypothesis must name its knowledge base when several are configured "
        f'({", ".join(bundle_ids)}): add "bundle_id" to the marker payload. Routing it here '
        "would evaluate a direction against a project nobody asked about"
    )


def _parse_hypotheses(
    texts: Sequence[str], bundle_ids: Sequence[str]
) -> list[tuple[str, str, str]]:
    """Every marked ``(label, rationale, bundle_id)`` triple the hypothesiser committed to, in turn
    order. The base is resolved HERE rather than at minting time so an unroutable claim is refused
    while the line that made it is still in hand for the error message."""
    found: list[tuple[str, str, str]] = []
    for text in texts:
        for line in text.splitlines():
            stripped = line.strip()
            if not stripped.startswith(HYPOTHESIS_MARKER):
                continue
            payload = stripped[len(HYPOTHESIS_MARKER) :].strip()
            try:
                data = json.loads(payload)
            except json.JSONDecodeError as exc:
                raise HypothesisParseError(
                    f"hypothesiser marked a line as a hypothesis and it is not JSON: {exc}; "
                    f"line was: {stripped}"
                ) from exc
            if not isinstance(data, dict) or not data.get("label") or not data.get("rationale"):
                raise HypothesisParseError(
                    "a marked hypothesis needs a non-empty 'label' and 'rationale'; got: "
                    f"{stripped}"
                )
            found.append(
                (
                    str(data["label"]),
                    str(data["rationale"]),
                    _resolve_hypothesis_bundle(str(data.get("bundle_id") or ""), bundle_ids),
                )
            )
    return found


def refuse_unroutable_seeds(seeds: Sequence[Approach], bundle_ids: Sequence[str]) -> None:
    """Refuse an expert's seed that could not be dispatched — BEFORE the first model call.

    Validation, ALDRI reparasjon, and here the second half is the load-bearing one: a seed is the
    expert's own words and § C.6 door 1 is a PRESERVATION rule, so filling in a missing
    ``bundle_id`` on their behalf would put their name on a routing decision they did not make.
    The single-base default belongs to ``route_by_bundle``, at consumption, where it is unambiguous
    by construction — never to a rewrite of the input.

    Called before the loop starts rather than after it returns, and that placement is the økt-57
    outbox-hoist precedent: at the exception alone, a refusal after the exploration has spent its
    whole budget looks identical to one before it spent anything. The mandate is undispatchable
    either way; what is at stake is whether the expert pays to find that out.
    """
    for seed in seeds:
        if seed.bundle_id:
            if seed.bundle_id not in bundle_ids:
                known = ", ".join(bundle_ids) or "(none configured)"
                raise ExplorationError(
                    f"seed approach {seed.id!r} names knowledge base {seed.bundle_id!r}, which is "
                    f"not configured for this exploration; configured: {known}"
                )
        elif len(bundle_ids) > 1:
            raise ExplorationError(
                f"seed approach {seed.id!r} names no knowledge base and {len(bundle_ids)} are "
                f"configured ({', '.join(bundle_ids)}); set Approach.bundle_id so the mandate can "
                "be dispatched to the base the hypothesis is actually about"
            )


def _mint_approaches(
    seeds: Sequence[Approach], discovered: Sequence[tuple[str, str, str]]
) -> tuple[Approach, ...]:
    """Seeds FIRST, untouched, then one approach per discovered direction.

    Seeds lead because the pipeline evaluates approaches in order under one shared budget: a
    direction the domain expert asked for must be reached before the loop's own findings spend
    what is left. The minted ids skip anything a seed already claims, so a seed literally named
    ``hypothesis-1`` cannot collide — ``Mandate`` refuses duplicate ids at construction, and
    turning an expert's naming choice into a hard failure would be a refusal that teaches nothing.
    """
    taken = {approach.id for approach in seeds}
    minted: list[Approach] = list(seeds)
    counter = 0
    for label, rationale, bundle_id in discovered:
        counter += 1
        candidate = f"hypothesis-{counter}"
        while candidate in taken or candidate == OWN_PROPOSAL_ID:
            counter += 1
            candidate = f"hypothesis-{counter}"
        taken.add(candidate)
        # description is the hypothesiser's own words, VERBATIM: ``generate._build_messages`` feeds
        # an Approach.description to the proposer unchanged, and the reason a direction is worth
        # trying is exactly the half a model cannot re-derive from the cost table.
        #
        # ``bundle_id`` is stamped on a MINTED approach because there is nothing here to preserve —
        # the opposite call from the seeds above, which pass through untouched (§ C.6 door 1).
        minted.append(
            Approach(id=candidate, label=label, description=rationale, bundle_id=bundle_id)
        )
    return tuple(minted)


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


async def explore(
    prompt: str,
    *,
    contract: ExplorationContract,
    bundle_dirs: Sequence[str] = (),
    profile: Profile | str = Profile.LOCAL,
    client_factory: Callable[[str], BaseChatClient] | None = None,
    seed_approaches: Sequence[Approach] = (),
    plan_reviewer: PlanReviewer | None = None,
    meter: TokenMeter | None = None,
    success_criteria: str = "",
    trace: ExplorationTrace | None = None,
    checkpoint_dir: str | None = None,
) -> ExplorationResult:
    """Explore the knowledge bases and return the ``Mandate`` the pipeline should evaluate.

    Writes NOTHING. No outbox artefact, no wiki promotion, no verdict — level 3 of the guarantee
    table belongs to ``run_project`` alone, and an exploration that could write would be a route
    around the gate that makes an answer checkable.

    **Three stops, three channels, deliberately not one.** Tokens raise ``BudgetExceeded`` exactly
    as the debate does, so the hosted surface's 429 arm needs no new case. Rounds raise it too,
    with ``kind="exploration_rounds"`` — the orchestration itself raises nothing at its round cap
    (measured, § F): it returns a canonical assistant message that is indistinguishable from
    success at the transport, so this layer produces the typed stop. Everything semantic —
    stalling out, an exhausted revision cap, a ledger naming nobody — is a VALUE in ``stop``,
    because those are outcomes of the exploration rather than exhaustion of a resource, and S3.4
    split those two apart for a reason.

    ``seed_approaches`` are the domain expert's own hypotheses (door 1 of § C.6). They are in the
    returned mandate whatever the loop found, including when the loop found nothing and including
    when it stopped early.

    **Multi-base (§ C.7).** Every approach in the returned mandate carries ``bundle_id``, and
    ``mandate.route_by_bundle`` partitions the commission by it so ``run.run_mandate_across_bundles``
    can run the pipeline once per base. ``run_project`` itself still takes ONE ``bundle_dir``, and
    that is the measured shape rather than a step not yet taken: it derives the project, the
    validator's cost baseline, the read context and the ExpeL query key from THE bundle and returns
    one stamped ``RunResult``, so a second directory on that signature would force a silent
    pick-one for all four.

    Honesty limits, stated rather than implied. (1) The ``quick_validate`` verdicts the
    hypothesiser saw are not in
    ``ExplorationResult``: they are level-1 advisory, and their home is the
    ``{run_id}-exploration.json`` artefact the CLI wiring writes — pass an ``ExplorationTrace`` to
    collect them. (2) The exploration roles resolve through ``resolve_model``'s ``default``
    fallback unless an operator maps them explicitly.

    ``trace`` is the caller's accumulator and is the ONLY way to see what a run that RAISED
    produced: both budget channels destroy the ``ExplorationResult`` before it exists. When it is
    omitted a private one is used, so the returned result is unchanged for every existing caller.
    """
    if plan_reviewer is not None and checkpoint_dir is not None:
        raise ExplorationError(
            "a plan_reviewer and a checkpoint_dir are two doors onto one review: the first answers "
            "it in this process, the second parks it for another one. Refused rather than ranked, "
            "because silently preferring either would block a caller that asked for the other"
        )
    if contract.enable_plan_review and plan_reviewer is None and checkpoint_dir is None:
        raise ExplorationError(
            "enable_plan_review is set but no plan_reviewer was given and no checkpoint_dir was "
            "offered to park it: the exploration would stop at a review nobody can answer, which "
            "is a hang rather than a result"
        )
    if checkpoint_dir is not None and not contract.enable_plan_review:
        raise ExplorationError(
            "a checkpoint_dir was given but enable_plan_review is false, so nothing would ever "
            "park and the storage would be written and never read (a cap on an event that cannot "
            "happen, refused for the reason max_plan_revisions is)"
        )
    if plan_reviewer is not None and not contract.enable_plan_review:
        raise ExplorationError(
            "a plan_reviewer was given but enable_plan_review is false, so no review is ever "
            "requested and the reviewer would never be called (refused, never silently ignored)"
        )

    # The base ids, resolved ONCE and before anything is built: they are what a seed and a marked
    # hypothesis are both routed against, and ``_bundle_index`` is the one place that decides them
    # (a duplicate basename refuses here rather than letting the manager read A believing it read B).
    bundle_ids = tuple(_bundle_index(bundle_dirs))
    refuse_unroutable_seeds(seed_approaches, bundle_ids)

    if meter is None:
        meter = TokenMeter(Budget(max_tokens=contract.max_tokens, max_rounds=contract.max_rounds))
    if client_factory is None:
        # Imported HERE, not at module scope: ``run`` imports this module for its ``--explore``
        # door, so a top-level import would be circular. One copy of the backend factory, never a
        # second (the (p) rule) — the same move ``run.main`` makes for ``scripted_factory``.
        from portfolio_optimiser.run import _default_factory

        client_factory = _default_factory(profile)

    if trace is None:
        trace = ExplorationTrace()

    workflow = fresh_exploration_workflow(
        client_factory,
        contract=contract,
        bundle_dirs=bundle_dirs,
        middleware=[BudgetMiddleware(meter), ExplorationToolRecorder(trace.tool_calls)],
        quick_validate_sink=trace.quick_validations,
        checkpoint_dir=checkpoint_dir,
    )

    hypothesis_texts: list[str] = []

    # ONE span for the whole exploration, opened before the first model call and closed however
    # the loop ends — including on a BudgetExceeded, which the span records rather than swallows.
    # With no provider installed this is a no-op tracer and every event is discarded, which is
    # exactly what "tracing is off" has meant since U14.
    with exploration_tracer().start_as_current_span(EXPLORATION_SPAN) as span:
        result = await workflow.run(prompt)
        stop, _ = await _drive(
            workflow,
            result,
            contract=contract,
            trace=trace,
            hypotheses=hypothesis_texts,
            seen=set(),
            replans=0,
            span=span,
            plan_reviewer=plan_reviewer,
            checkpoint_dir=checkpoint_dir,
            prompt=prompt,
            bundle_dirs=bundle_dirs,
            meter=meter,
        )

    return _finish(
        prompt=prompt,
        stop=stop,
        trace=trace,
        hypotheses=hypothesis_texts,
        bundle_ids=bundle_ids,
        seed_approaches=seed_approaches,
        success_criteria=success_criteria,
    )


async def _drive(
    workflow: Any,
    result: Any,
    *,
    contract: ExplorationContract,
    trace: ExplorationTrace,
    hypotheses: list[str],
    seen: set[int],
    replans: int,
    span: Any,
    plan_reviewer: PlanReviewer | None,
    checkpoint_dir: str | None,
    prompt: str,
    bundle_dirs: Sequence[str],
    meter: TokenMeter,
) -> tuple[ExplorationStop | None, int]:
    """Drive a built workflow from one ``run()`` result to an ending, answering plan reviews.

    **ONE copy of this loop, shared by ``explore`` and ``resume_exploration``.** The asynchronous
    door is not a second loop that happens to look like the first: a resumed exploration answers
    reviews, absorbs ledgers, mints nothing, and classifies its stop by exactly the same rules, and
    two copies of that would drift the moment one of them was corrected (kø-(p)).

    Raises ``PlanReviewParked`` when the asynchronous door is armed — the loop stops mid-plan and
    the caller decides where to write the question.
    """
    # ONE accumulator per fact, held by the caller (see ``ExplorationTrace``). The local names are
    # aliases, never copies — a second list here is the kø-(p) drift this shape exists to prevent.
    ledger_log = trace.ledger
    plan_reviews = trace.plan_reviews
    stop: ExplorationStop | None = None

    while True:
        trace.tokens_spent = meter.tokens
        replans += _absorb(
            result,
            ledger_log=ledger_log,
            hypotheses=hypotheses,
            seen=seen,
            span=span,
        )
        pending = _pending_plan_reviews(result)
        if not pending:
            break
        request = pending[0]
        review = request.data
        if plan_reviewer is None:
            # The asynchronous door (U12). ``checkpoint_dir`` is what armed it, and the guard in
            # ``explore`` refused every other way of arriving here with no reviewer.
            assert checkpoint_dir is not None
            raise PlanReviewParked(
                await _park(
                    workflow,
                    request,
                    checkpoint_dir=checkpoint_dir,
                    prompt=prompt,
                    bundle_dirs=bundle_dirs,
                    contract=contract,
                    trace=trace,
                    hypotheses=hypotheses,
                    meter=meter,
                    replans=replans,
                )
            )
        decision = plan_reviewer(
            PlanReviewRequest(
                index=len(plan_reviews),
                plan=_plan_text(review.plan),
                current_progress=str(review.current_progress),
                is_stalled=_truthy(review.is_stalled),
            )
        )
        if decision.feedback is None:
            plan_reviews.append(
                PlanReview(
                    index=len(plan_reviews),
                    plan=_plan_text(review.plan),
                    is_stalled=_truthy(review.is_stalled),
                    decision="approve",
                )
            )
            response = MagenticPlanReviewResponse.approve()
        else:
            # The revision is recorded whether or not it is APPLIED: ``plan_reviews`` is the
            # record of what the reviewer DECIDED, and ``stop`` is what says the last one was
            # refused.
            applied = sum(1 for entry in plan_reviews if entry.decision == "revise")
            plan_reviews.append(
                PlanReview(
                    index=len(plan_reviews),
                    plan=_plan_text(review.plan),
                    is_stalled=_truthy(review.is_stalled),
                    decision="revise",
                    feedback=decision.feedback,
                )
            )
            if applied >= contract.max_plan_revisions:
                # Measured (§ F, A3): a revise costs two manager calls, emits no progress
                # ledger and consumes no round, then asks AGAIN. Under the round cap alone this
                # loop never terminates. Stopping is the honest move — forcing an approve the
                # reviewer did not give would be repair, and repair of a human's decision most
                # of all.
                stop = "plan_revisions_exhausted"
                break
            response = MagenticPlanReviewResponse.revise(decision.feedback)
        result = await workflow.run(responses={request.request_id: response})

    trace.tokens_spent = meter.tokens
    if stop is None:
        stop = _classify_stop(ledger_log, replans=replans, contract=contract)
    return stop, replans


async def _park(
    workflow: Any,
    request: Any,
    *,
    checkpoint_dir: str,
    prompt: str,
    bundle_dirs: Sequence[str],
    contract: ExplorationContract,
    trace: ExplorationTrace,
    hypotheses: Sequence[str],
    meter: TokenMeter,
    replans: int,
) -> ParkedExploration:
    """Freeze the suspended exploration and name the checkpoint a later process resumes from.

    ``get_latest`` rather than the last entry of a listing: it picks by timestamp
    (``_checkpoint.py:424``), while the listing's order is whatever ``Path.glob`` returned.

    The checkpoint named here must be the one that carries THIS question, and that is asserted
    against the declared ``pending_request_info_events`` field rather than inferred from the
    listing being non-empty. **The weaker form went inert under core 1.16.0 (F15, measured):** a
    park now writes TWO checkpoints, and only one of them carries a plan-review type. A
    mis-declared ``_ALLOWED_CHECKPOINT_TYPES`` therefore no longer empties the listing — it drops
    exactly the checkpoint that matters, ``get_latest`` returns the OTHER one, and the old
    ``latest is None`` guard passed while the question file named a checkpoint that can never bear
    the answer. That is the silent dead question this layer exists to refuse, so the guard now
    checks the property it always meant rather than the symptom that used to imply it.

    Refusing is where this layer is louder than the framework: a blocked deserialisation is
    swallowed into a warning upstream, so the alternative to raising here is a question whose
    answer can never be applied — an exploration that fails as an absence, days later, to somebody
    who has already written their answer.
    """
    latest = await checkpoint_storage(checkpoint_dir).get_latest(workflow_name=workflow.name)
    if latest is None or str(request.request_id) not in latest.pending_request_info_events:
        raise CheckpointUnreadable(
            f"the exploration reached a plan review but no checkpoint carrying it could be read "
            f"back from {checkpoint_dir!r}: without one the review can never be resumed, so it is "
            f"refused here rather than written as a question nobody can answer"
        )
    review = request.data
    return ParkedExploration(
        prompt=prompt,
        request_id=str(request.request_id),
        checkpoint_id=str(latest.checkpoint_id),
        index=len(trace.plan_reviews),
        plan=_plan_text(review.plan),
        current_progress=str(review.current_progress),
        is_stalled=_truthy(review.is_stalled),
        bundle_dirs=tuple(bundle_dirs),
        contract=contract,
        ledger=tuple(trace.ledger),
        plan_reviews=tuple(trace.plan_reviews),
        hypotheses=tuple(hypotheses),
        tokens_spent=meter.tokens,
        replans=replans,
    )


def _finish(
    *,
    prompt: str,
    stop: ExplorationStop | None,
    trace: ExplorationTrace,
    hypotheses: Sequence[str],
    bundle_ids: Sequence[str],
    seed_approaches: Sequence[Approach],
    success_criteria: str,
) -> ExplorationResult:
    """Mint the mandate from a finished drive. Shared, for the reason ``_drive`` is."""
    discovered = _parse_hypotheses(hypotheses, bundle_ids) if stop != "unknown_speaker" else []
    return ExplorationResult(
        mandate=Mandate(
            objective=prompt,
            approaches=_mint_approaches(seed_approaches, discovered),
            allow_own_proposals=True,
            success_criteria=success_criteria,
        ),
        ledger_log=tuple(trace.ledger),
        stop=stop,
        plan_reviews=tuple(trace.plan_reviews),
    )


async def resume_exploration(
    parked: ParkedExploration,
    decision: PlanReviewDecision,
    *,
    checkpoint_dir: str,
    profile: Profile | str = Profile.LOCAL,
    client_factory: Callable[[str], BaseChatClient] | None = None,
    seed_approaches: Sequence[Approach] = (),
    success_criteria: str = "",
    trace: ExplorationTrace | None = None,
) -> ExplorationResult:
    """Answer a parked plan review and drive the exploration onward, in a process that never ran it.

    Everything about the workflow is rebuilt from ``parked`` rather than from argv: the prompt, the
    bounds and the bases are what the suspended run used, and the graph must match the checkpoint's
    signature (``_runner.py:275-279``) for the restore to be accepted at all. A caller that had to
    re-supply them could get one of them wrong and would find out as a restore failure days later.

    **The budget spans the suspension.** The meter starts at ``parked.tokens_spent`` and the ledger
    at ``parked.ledger``, so the round cap and the token cap measure the whole exploration rather
    than this leg of it. Without that a park would hand back a full budget every time it happened.

    Parking AGAIN is a normal outcome, not a failure: a revision makes the manager replan and ask
    about the NEW plan, which is the second half of "be om svar, BRUKE svarene" on this time-scale.
    It leaves by ``PlanReviewParked`` exactly as the first park did.
    """
    if trace is None:
        trace = ExplorationTrace()
    # The carried state is put back BEFORE anything runs: ``_drive`` reads these as its own running
    # log, and ``_classify_stop`` counts the ledger to decide whether the ROUND cap bound.
    trace.ledger.extend(parked.ledger)
    trace.plan_reviews.extend(parked.plan_reviews)
    hypothesis_texts = list(parked.hypotheses)

    bundle_ids = tuple(_bundle_index(parked.bundle_dirs))
    refuse_unroutable_seeds(seed_approaches, bundle_ids)

    meter = TokenMeter(
        Budget(max_tokens=parked.contract.max_tokens, max_rounds=parked.contract.max_rounds)
    )
    # Through ``charge``, not by assigning ``tokens``: charging re-tests the cap, so a suspension
    # that already spent everything refuses HERE instead of buying one more leg of the loop.
    meter.charge(parked.tokens_spent)
    if client_factory is None:
        from portfolio_optimiser.run import _default_factory

        client_factory = _default_factory(profile)

    workflow = fresh_exploration_workflow(
        client_factory,
        contract=parked.contract,
        bundle_dirs=parked.bundle_dirs,
        middleware=[BudgetMiddleware(meter), ExplorationToolRecorder(trace.tool_calls)],
        quick_validate_sink=trace.quick_validations,
        checkpoint_dir=checkpoint_dir,
    )

    # The revision cap counted across the SUSPENSION, over the reviews carried in ``parked``.
    # Measured, and it is why the carry-over is load-bearing rather than tidy: a revise costs two
    # manager calls, emits no progress ledger and consumes no round (§ F, A3), so a cap that reset
    # at every park would leave the asynchronous door with no bound at all — an expert could revise
    # forever, one process at a time, under guards that all report themselves satisfied. The same
    # arithmetic ``_drive`` does for the synchronous door, and the same refusal to repair: the
    # decision is RECORDED and the loop stops, never forced into an approve nobody gave.
    applied = sum(1 for entry in parked.plan_reviews if entry.decision == "revise")
    trace.plan_reviews.append(
        PlanReview(
            index=parked.index,
            plan=parked.plan,
            is_stalled=parked.is_stalled,
            decision="approve" if decision.feedback is None else "revise",
            feedback=decision.feedback or "",
        )
    )
    if decision.feedback is not None and applied >= parked.contract.max_plan_revisions:
        return _finish(
            prompt=parked.prompt,
            stop="plan_revisions_exhausted",
            trace=trace,
            hypotheses=hypothesis_texts,
            bundle_ids=bundle_ids,
            seed_approaches=seed_approaches,
            success_criteria=success_criteria,
        )

    response = (
        MagenticPlanReviewResponse.approve()
        if decision.feedback is None
        else MagenticPlanReviewResponse.revise(decision.feedback)
    )

    with exploration_tracer().start_as_current_span(EXPLORATION_SPAN) as span:
        result = await workflow.run(
            responses={parked.request_id: response},
            checkpoint_id=parked.checkpoint_id,
            checkpoint_storage=checkpoint_storage(checkpoint_dir),
        )
        stop, _ = await _drive(
            workflow,
            result,
            contract=parked.contract,
            trace=trace,
            hypotheses=hypothesis_texts,
            seen=set(),
            replans=parked.replans,
            span=span,
            plan_reviewer=None,
            checkpoint_dir=checkpoint_dir,
            prompt=parked.prompt,
            bundle_dirs=parked.bundle_dirs,
            meter=meter,
        )

    return _finish(
        prompt=parked.prompt,
        stop=stop,
        trace=trace,
        hypotheses=hypothesis_texts,
        bundle_ids=bundle_ids,
        seed_approaches=seed_approaches,
        success_criteria=success_criteria,
    )


def _classify_stop(
    ledger_log: Sequence[LedgerEntry], *, replans: int, contract: ExplorationContract
) -> ExplorationStop | None:
    """Decide what a finished ``run()`` actually was. Structural, never prose-matched.

    The orchestration ends four ways and reports three of them with an ordinary-looking result:
    a satisfied request, the round cap, the reset cap, and a ``next_speaker`` matching nobody.
    Only the first is success. The discriminators used here are counts and ledger flags — the
    termination MESSAGE is deliberately not matched, because it is an f-string built inline
    (``:1253``) with no constant to pin against, and a model's own final answer could contain the
    same words.

    A ledger naming nobody is checked FIRST and outranks everything else. It is the only ending
    in which the orchestrator produced an answer having asked no participant at all (``:1128``),
    so the run has a result that no work stands behind — the same class as the retired E2 finding,
    and not something a later, softer verdict should be allowed to paper over.
    """
    if any(not entry.speaker_known for entry in ledger_log):
        return "unknown_speaker"
    if ledger_log and ledger_log[-1].is_request_satisfied:
        return None
    if len(ledger_log) >= contract.max_rounds:
        raise BudgetExceeded("exploration_rounds", contract.max_rounds, len(ledger_log))
    # Whatever is left ended without satisfying the request and without exhausting the round cap:
    # the manager reset until it ran out of resets, or terminated with nothing to show. Both are
    # the same fact for a caller — the exploration gave up — so they share one token rather than
    # inventing a distinction the ledger cannot support.
    return "stalled"