portfolio-optimiser/src/portfolio_optimiser/run.py

"""Vertical-slice orchestrator + single-command entry (two-layer HITL wiring).

``run_project`` composes the whole method for ONE synthetic project on real (or injected)
chat clients:

1. ``load_contracts`` — fail-fast: every config (incl. the verdict-feedback shape) is
   validated BEFORE any chat client is built.
2. load the project + retrieve cited chunks via the Step-7 data source -> first-class
   ``provenance.Citation`` list.
3. a FRESH maker-checker ``GroupChat`` debate (``fresh_workflow``), round-capped
   (``with_max_rounds``); **Layer-1 HITL** = the optional in-run ``with_request_info`` gate.
4. ``generate_via_llm`` -> blocking ``validate_proposal`` -> ``ValidatedProposal | Rejection``
   (the token bound is the ``meter`` checked in the generate loop).
5. attach a first-class ``ProvenanceStamp``.
6. **Layer-2 (out-of-band)**: ``capture_verdict`` mints a stable id from the proposal's
   features; the decision/rationale come from ``verdict_input`` (function arg / CLI / fixture).
   The B11 expert notification is a STUB (``notify``) in Fase 2.
7. persist to the ``VerdictStore`` -> the next run's ``ExpeLContextProvider`` retrieval (the
   learning loop; this run also exercises the two-arg ``extend_instructions`` injection).

The two HITL layers are deliberately distinct: **Layer-1** is the optional synchronous in-run
review gate (no checkpoint — research 01: durable resume is fragile); **Layer-2** is the
durable learned verdict captured out-of-band in the VerdictStore (D7-portable).
"""

from __future__ import annotations

import asyncio
import json
from collections.abc import Awaitable, Callable, Iterable, Sequence
from contextlib import AsyncExitStack
from dataclasses import dataclass, replace
from pathlib import Path
from typing import Any, Literal, cast

from agent_framework import BaseChatClient, SessionContext
from pydantic import ValidationError

from portfolio_optimiser.backends import Profile, get_backend, resolve_model
from portfolio_optimiser.budget import (
    Budget,
    BudgetExceeded,
    BudgetMiddleware,
    BudgetRefused,
    PortfolioMeter,
    TokenMeter,
)
from portfolio_optimiser.contracts import GoalConfig, GoalContract, load_contracts, load_goal_config
from portfolio_optimiser.ledger import SavingsLedger, to_ore
from portfolio_optimiser.datasource import (
    bundle_citations,
    chunk_dict_to_citation,
    make_retrieval_tool,
    retrieve_chunks,
)
from portfolio_optimiser.dimension import Dimension, admits, load_dimension
from portfolio_optimiser.generate import ParseFailure, generate_via_llm
from portfolio_optimiser.ir import SavingsProposal
from portfolio_optimiser.mandate import (
    OWN_PROPOSAL_ID,
    Approach,
    ApproachOutcome,
    Mandate,
    announce,
    load_mandate,
    settle,
)
from portfolio_optimiser.mcp_tools import (
    McpServerConfig,
    ToolCallRecorder,
    build_mcp_tools,
    load_mcp_config,
    service_labels,
    tool_server_index,
)
from portfolio_optimiser.provenance import ProvenanceStamp
from portfolio_optimiser.reference_domain import Project, load_reference_projects
from portfolio_optimiser.validator import Rejection, ValidatedProposal, baseline_from_project
from portfolio_optimiser import okf, outbox
from portfolio_optimiser.semretrieval import (
    SEMANTIC_WEIGHT_DEFAULT,
    Embedder,
    FakeEmbedder,
    HybridRanker,
    build_embedder,
    load_embedder_config,
)
from portfolio_optimiser.verdicts import (
    ExpeLContextProvider,
    ProposalFeatures,
    Verdict,
    VerdictStore,
    bundle_candidate_features,
    capture_verdict,
    load_verdicts_from_dir,
    similarity,
    verdict_key,
)
from portfolio_optimiser.value_report import (
    build_value_report,
    dump_report_json,
    format_report_text,
)
from portfolio_optimiser.workflow import _MAKER_CHECKER_ROLES, fresh_workflow


#: The caps a CLI run actually uses. Named constants rather than repeated literals because the
#: run announcement (Trekk A2) PRINTS them: a second copy could drift and make the announcement
#: describe a run that never happened.
_DEFAULT_MAX_ROUNDS = 3
_DEFAULT_MAX_TOKENS = 100_000


@dataclass(frozen=True)
class RunResult:
    """The outcome of one project run: the validated/rejected proposal, its first-class
    provenance, the captured (Layer-2) verdict, the ExpeL hits surfaced for it, the store, the
    debate's converged output that the candidate was generated from (F1 traceability), and the
    checker's gate decision (Step 3/4: ``"approve" | "reject" | "absent"``). ``checker_verdict``
    records the checker's decision distinctly from ``provenance.validator_decision`` so the two
    falsifiers (reasoning vs numbers) are never conflated."""

    outcome: ValidatedProposal | Rejection
    provenance: ProvenanceStamp
    verdict: Verdict
    retrieved: list[Verdict]
    store: VerdictStore
    debate_output: str
    checker_verdict: str = "absent"
    #: One row per commissioned approach (Trekk A4). EMPTY when the run had no mandate — an empty
    #: report is honest there, because nothing was ordered. It defaults so every existing
    #: constructor call and every frozen aggregate over ``RunResult`` is unaffected.
    coverage: tuple[ApproachOutcome, ...] = ()
    #: Step 5 (målbilde §5/§7): the validator falsifications that informed a LATER generation
    #: attempt, in attempt order — what ``generate_via_llm`` corrected in response to, rather than
    #: only what it ended up with. EMPTY on the common path where the first candidate validates:
    #: nothing was falsified, so there is nothing to show. Honesty limit: with a mandate this is
    #: the run's refinements CONCATENATED across every commissioned approach, not keyed per
    #: approach — ``coverage`` is the per-approach report, and hanging proposals off its rows is
    #: what ``_evaluate_mandate`` deliberately avoids. It defaults, so every existing constructor
    #: call is unaffected (mirrors ``coverage``).
    refinements: tuple[Rejection, ...] = ()
    #: Every cross-link the bundle navigation could not follow. A RUN-level fact, carried here and
    #: NOT on ``provenance``: navigation happens ONCE per run, before any proposal exists, and the
    #: same walk backs every refinement attempt — whereas ``ProvenanceStamp.cost_baseline_anchored``
    #: describes the gate that judged ONE candidate. EMPTY on the road path (no bundle is navigated)
    #: and on any bundle that was read whole; it defaults for the same reason ``coverage`` does.
    skipped_links: tuple[okf.SkippedLink, ...] = ()


@dataclass(frozen=True)
class RunFailure:
    """One project that RAISED during a portfolio pass (S3.3, SC3 collect-and-continue).

    A DISTINCT type from ``RunResult`` rather than an error field on it, and deliberately so: the
    session spec's wording was "a ``RunResult`` slot with an error field", but ``RunResult`` is
    frozen with six required non-defaulted fields (``:89-94``) — a run that raised before producing
    an outcome has no honest value for ``outcome``, ``provenance`` or ``verdict``. Filling them with
    dummies would put FABRICATED provenance into the aggregate, which is the failure mode this
    repo's provenance rules exist to prevent. The exception is recorded as text (``error``) plus its
    class name (``error_type``) rather than the live exception object, so a ``PortfolioResult``
    stays a plain frozen value with no traceback frames held alive."""

    project_id: str
    error: str
    error_type: str


@dataclass(frozen=True)
class DryRunReport:
    """S4.2 offline ``--live-dry-run`` outcome (comparison protocol §4 pkt 3): everything a real run
    would use — profile, the resolved model-id per BUILT role, and the round/token parameters —
    captured WITHOUT a model call. A DISTINCT type from ``RunResult``, whose post-generation fields
    (outcome/provenance/verdict) do not exist yet on a run that stopped before the first model call."""

    profile: str
    resolved_models: dict[str, str]
    max_rounds: int
    max_tokens: int
    top_k: int
    #: Whether a REAL run of this configuration would have its deterministic gate anchored to the
    #: project's own cost lines (see ``ProvenanceStamp.cost_baseline_anchored``). Carried here too
    #: because a dry run stops before any proposal exists, so there is no stamp to read it off —
    #: and this surface is precisely where the un-anchored case was measured to be silent.
    cost_baseline_anchored: bool
    #: Every cross-link the bundle navigation could not follow (``okf.Bundle.skipped``). EMPTY is a
    #: positive statement — "every cross-link was followed" — which is why it DEFAULTS, unlike
    #: ``cost_baseline_anchored`` above: a missing bool would have to claim something about an event
    #: (and both claims would sometimes be false), while a missing trace asserts only that the event
    #: list is empty. The road path navigates no bundle, so empty is literally true there too.
    skipped_links: tuple[okf.SkippedLink, ...] = ()


@dataclass(frozen=True)
class GoalReached:
    """A savings-goal signal VALUE (Step 8, SC6) — NOT an exception. Structured like
    ``BudgetExceeded`` (``budget.py:22-34``) but semantically SUCCESS (the goal was reached), not
    resource exhaustion (H1). Used as a ``stop_reason`` value + a loop ``break``, never ``raise``d.
    ``scope`` is ``"portfolio"`` (the whole pass) or ``"project"`` (one pid); ``limit_ore`` is the
    threshold that was met, ``observed_ore`` the accumulated realized sum that met it (``>=``)."""

    scope: Literal["project", "portfolio"]
    project_id: str | None
    limit_ore: int
    observed_ore: int


@dataclass(frozen=True)
class BudgetStop:
    """A GLOBAL token-cap stop signal VALUE (S3.4/F10) — NOT an exception, and NOT a goal.

    Structured like ``GoalReached`` and carried the same way (a ``stop_reason``-shaped value plus a
    loop ``break``), but it is kept as its OWN field rather than widening ``stop_reason``: the two
    stops mean opposite things. A goal-stop is success (the savings target was met); this is
    resource exhaustion (the pass ran out of tokens). Folding them into one field would let a
    caller read "we stopped" without being able to tell which happened.

    ``required_tokens`` is what one more run would have needed; ``remaining_tokens`` is what was
    actually left. Both are recorded because their DIFFERENCE is the operator's next decision."""

    limit_tokens: int
    spent_tokens: int
    remaining_tokens: int
    required_tokens: int


@dataclass(frozen=True)
class PortfolioResult:
    """The outcome of a sequential fan-out over N projects (SC2).

    ``runs`` is one ``RunResult`` per project in input order; ``store`` is the ONE shared
    ``VerdictStore`` threaded across every run (the cross-project ExpeL learning loop).
    The remaining fields are a thin aggregate over ``runs``: ``validated_count`` /
    ``rejected_count`` partition the outcomes; ``sum_claimed_saving_nok`` totals the claimed
    saving of the validated proposals only; ``sum_token_usage`` totals every run's
    provenance token usage. ``stopped_early`` / ``stop_reason`` record a Step-8 goal-stop,
    ``failures`` records the projects that RAISED (S3.3 collect-and-continue), and ``budget_stop``
    records a S3.4 global-token-cap stop (which also sets ``stopped_early``, but is kept apart from
    ``stop_reason`` because exhaustion is not success): all four default so the frozen aggregate and
    every existing constructor call are unaffected.

    ``runs`` and ``failures`` PARTITION the projects that were actually submitted — a project
    appears in exactly one of them, never both, and the counts do not overlap. ``validated_count`` /
    ``rejected_count`` therefore total ``len(runs)``, not the portfolio size: a failure is neither a
    validation nor a rejection, and folding it into either would misreport the pass."""

    runs: tuple[RunResult, ...]
    store: VerdictStore
    validated_count: int
    rejected_count: int
    sum_claimed_saving_nok: float
    sum_token_usage: int
    stopped_early: bool = False
    stop_reason: GoalReached | None = None
    failures: tuple[RunFailure, ...] = ()
    budget_stop: BudgetStop | None = None


def _coverage_row(
    row_id: str, label: str, outcome: ValidatedProposal | Rejection
) -> ApproachOutcome:
    """One coverage row from one evaluated approach. A rejection carries the validator's reason
    verbatim — a bare status would tell the expert their approach failed without telling them why,
    which is the part they can actually act on."""
    if isinstance(outcome, ValidatedProposal):
        return ApproachOutcome(
            id=row_id,
            label=label,
            status="validated",
            saving_nok=outcome.proposal.claimed_saving_nok,
        )
    return ApproachOutcome(id=row_id, label=label, status="rejected", detail=outcome.reason)


def _select_outcome(
    produced: list[tuple[int, ValidatedProposal | Rejection]],
) -> ValidatedProposal | Rejection:
    """Pick the ONE outcome a ``RunResult`` carries out of everything the mandate produced.

    ``RunResult`` stays single-outcome on purpose: portfolio aggregation, the outbox artefacts and
    the HITL verdict keying all rest on there being exactly one. The choice is deterministic —
    highest validated saving, ties broken by mandate order — so it can never depend on which
    approach happened to run last. When nothing validated, the FIRST rejection stands, which keeps
    a fully-rejected mandate reporting a typed ``Rejection`` rather than a fabricated success.
    """
    validated = [(i, o) for i, o in produced if isinstance(o, ValidatedProposal)]
    if validated:
        return min(validated, key=lambda t: (-t[1].proposal.claimed_saving_nok, t[0]))[1]
    return produced[0][1]


async def _evaluate_mandate(
    mandate: Mandate,
    evaluate: Callable[[Approach | None], Awaitable[ValidatedProposal | Rejection]],
) -> tuple[
    ValidatedProposal | Rejection,
    tuple[ApproachOutcome, ...],
    tuple[tuple[str, ValidatedProposal | Rejection], ...],
]:
    """Evaluate every commissioned approach, then the run's own proposal when allowed, and report
    what became of each (Trekk A3/A4).

    Returns the selected outcome, the coverage report, and — third — every EVALUATED approach's own
    outcome paired with its id (A5), which is what lets each of them be written as a judgeable
    outbox artefact. It is returned alongside rather than folded into ``ApproachOutcome`` on
    purpose: ``mandate.py`` imports only ``pydantic`` + stdlib to stay D7-portable (guarded by
    ``test_okf_is_maf_free``), and hanging a ``ValidatedProposal`` off a coverage row would drag
    ``validator`` — and with it ``pulp`` — into that deliberately thin module. Rows the run never
    reached are absent here by construction: a ``not_evaluated`` approach has no proposal, so there
    is nothing to write and nothing to judge.

    Budget exhaustion mid-list is REPORTED, not swallowed: the approaches that were never reached
    become ``not_evaluated`` rows. But if the very first approach exhausts the budget there is
    nothing honest to return, so ``BudgetExceeded`` propagates exactly as it did before — a run
    that produced nothing must still fail loudly rather than hand back an empty report.
    """
    plan: list[tuple[str, str, Approach | None]] = [(a.id, a.label, a) for a in mandate.approaches]
    if mandate.allow_own_proposals:
        plan.append((OWN_PROPOSAL_ID, "the system's own proposal", None))

    rows: list[ApproachOutcome] = []
    produced: list[tuple[int, ValidatedProposal | Rejection]] = []
    evaluated: list[tuple[str, ValidatedProposal | Rejection]] = []
    for index, (row_id, label, approach) in enumerate(plan):
        try:
            outcome = await evaluate(approach)
        except BudgetExceeded:
            if not produced:
                raise
            rows.extend(
                ApproachOutcome(
                    id=rid,
                    label=lbl,
                    status="not_evaluated",
                    detail="budget exhausted before this approach was evaluated",
                )
                for rid, lbl, _ in plan[index:]
            )
            break
        produced.append((index, outcome))
        evaluated.append((row_id, outcome))
        rows.append(_coverage_row(row_id, label, outcome))

    return _select_outcome(produced), tuple(rows), tuple(evaluated)


def _authored_texts(result: Any, name: str) -> list[str]:
    """The texts of ``get_outputs()`` entries authored by participant ``name`` (proposer/checker),
    in surfaced order. MAF surfaces ``author_name`` on each output's ``messages`` — NOT on the
    ``AgentResponse`` itself (verified against 1.9.0) — so we match through ``messages``. This
    separates the proposer's converged output (fed to generation, F1) from the checker's gate
    verdict (Step 3/4); the orchestrator's termination notice is authored by neither, so it is
    excluded automatically."""
    texts: list[str] = []
    for out in result.get_outputs():
        if not any(getattr(m, "author_name", None) == name for m in getattr(out, "messages", [])):
            continue
        text = out if isinstance(out, str) else getattr(out, "text", None)
        if text:
            texts.append(text)
    return texts


def _debate_text(result: Any) -> str:
    """The PROPOSER's converged output (fed into generation, F1). With ``output_from=agents`` both
    participants surface, so we select proposer-authored outputs specifically — taking the last of
    ALL surfaced outputs would feed the checker's verdict to generation at even round counts.
    Returns ``""`` when the proposer produced no surfaced text."""
    proposer_texts = _authored_texts(result, "proposer")
    return proposer_texts[-1] if proposer_texts else ""


def _checker_verdict(result: Any) -> tuple[str, str]:
    """Parse the checker's gate verdict from its surfaced debate output (Step 3/4, målbilde §2/§6).
    Returns ``(decision, reason)``: ``"reject"`` ONLY on an explicit ``VERDICT: REJECT`` (with the
    trailing reason), ``"approve"`` on an explicit ``VERDICT: APPROVE``, else ``"absent"``. The gate
    is opt-in-reject (fail-open): a missing/unparseable marker never blocks, so the deterministic
    validator stays the sole gate on those runs."""
    checker_texts = _authored_texts(result, "checker")
    text = checker_texts[-1] if checker_texts else ""
    upper = text.upper()
    if "VERDICT: REJECT" in upper:
        reason = text[upper.index("VERDICT: REJECT") + len("VERDICT: REJECT") :]
        return "reject", reason.lstrip(" -:—").strip()
    if "VERDICT: APPROVE" in upper:
        return "approve", ""
    return "absent", ""


def _project_by_id(project_id: str) -> Project:
    for project in load_reference_projects():
        if project.id == project_id:
            return project
    raise ValueError(f"unknown project_id: {project_id!r}")


def _project_from_bundle(
    bundle_dir: str, project_id: str, *, bundle: okf.Bundle | None = None
) -> Project:
    """Derive a minimal ``Project`` from an OKF bundle (so a bundle the loop runs need NOT be a
    road reference-domain project). Only ``id`` + ``name`` reach the generation prompt
    (``generate._build_messages``), so ``cost_items`` is empty and ``verdict_input`` is unused here
    (the Layer-2 decision flows via the ``verdict_input`` argument). Fail-fast: the bundle's IR
    ``project_id`` must match the requested id. ``bundle`` reuses an already-navigated bundle to
    avoid a second navigation."""
    ir = okf.load_ir_projection(bundle_dir)
    if ir["project_id"] != project_id:
        raise ValueError(f"bundle project_id {ir['project_id']!r} != requested {project_id!r}")
    nav = bundle if bundle is not None else okf.navigate_bundle(bundle_dir)
    project_file = next((f for f in nav.files if f.type == "project"), None)
    name = (
        project_file.frontmatter.get("title", project_id).strip('"')
        if project_file is not None
        else project_id
    )
    return Project(
        id=project_id,
        name=name,
        description="",
        currency="NOK",
        cost_items=(),
        docs_dir=bundle_dir,
        verdict_input={},
    )


def _features_of(proposal: SavingsProposal) -> ProposalFeatures:
    return ProposalFeatures(
        affected_codes=frozenset(item.code for item in proposal.affected_items),
        measure_type=proposal.measure,
        claimed_saving_nok=proposal.claimed_saving_nok,
        description=proposal.measure,
    )


def _default_factory(profile: Profile | str) -> Callable[[str], BaseChatClient]:
    def factory(role: str) -> BaseChatClient:
        return get_backend(profile).create_chat_client(model=resolve_model(profile, role))

    return factory


#: The one line a run prints about its own anchoring. Rendered ONLY when the run is un-anchored:
#: an anchored run has nothing to warn about, and ``mandate.announce``'s rule is that a line for
#: something the run does not have is OMITTED rather than rendered empty.
_UNANCHORED_NOTICE = (
    "  Cost baseline: NONE in the bundle — this run is un-anchored: the validator's stage 0 "
    "(reconciling each proposed cost line against the project's own) is SKIPPED"
)


def cost_baseline_notice(anchored: bool) -> str | None:
    """Render the un-anchored notice, or ``None`` when the run IS anchored.

    ONE renderer with N callsites, never N copies of the wording (kø-(p)) — and it takes the
    already-resolved BOOLEAN rather than a bundle path, so the printed line and the machine-readable
    ``ProvenanceStamp.cost_baseline_anchored`` can never disagree: both descend from the single
    ``okf.load_optional_cost_baseline`` call inside ``run_project``. A renderer that re-read the
    bundle would be a second resolution of the same rule, free to drift from the run it describes.

    **Not folded into ``mandate.announce``, and that is a measurement rather than a preference:**
    ``announce`` is printed only when ``--mandate`` is given, so the runs this notice exists for —
    the bare bundle dry-runs that exit 0 in silence — would still say nothing. It also renders
    BEFORE ``run_project`` is called, i.e. before anyone has resolved the baseline; putting the line
    there would have required ``main`` to open the bundle itself.

    English, like every other line this CLI prints; the Norwegian explanation of what an un-anchored
    run means belongs in ``docs/kunnskapsbase-for-en-kjoring.md``, next to the domain expert.

    Tense-neutral on purpose ("is SKIPPED"): the same string serves ``--live-dry-run`` (where the
    run has not happened) and a completed run (where it has)."""
    return None if anchored else _UNANCHORED_NOTICE


def skipped_links_notice(skipped: tuple[okf.SkippedLink, ...]) -> str | None:
    """Render what the run could NOT read, or ``None`` when every cross-link was followed.

    The measured silence this closes: ``okf._walk`` tolerates an unfollowable link exactly as OKF
    SPEC §4 requires (skip, never raise) — correct, and unchanged here — but it left no trace, so a
    knowledge base whose other half was never reached looked identical to one where those documents
    were never written, and ``--live-dry-run`` exited 0 over both.

    ONE renderer with N callsites, never N copies of the wording (kø-(p)), and it takes the
    already-resolved trace rather than a bundle path: a renderer that re-navigated the bundle would
    be a second resolution of the same walk, free to disagree with the run it describes. Both
    callsites read it off the value ``run_project`` returned from its ONE
    ``okf.navigate_bundle`` call.

    ``None`` when the trace is empty — omission, never an empty row (``mandate.announce``'s rule,
    the same one ``cost_baseline_notice`` follows). A run that reached everything has nothing to
    report.

    The per-link line prints the reason TOKEN itself rather than a prose translation of it: a second
    display vocabulary keyed off ``SkipReason`` would be the duplicate free to drift, and the token
    is already the operative word ("missing" vs "outside-bundle"). English, like every other line
    this CLI prints; the Norwegian explanation belongs in
    ``docs/kunnskapsbase-for-en-kjoring.md``, next to the domain expert."""
    if not skipped:
        return None
    lines = [
        f"  Knowledge base: {len(skipped)} cross-link(s) NOT followed — "
        "the agents never read the document(s) behind them:"
    ]
    lines += [f"    - {s.from_file} -> {s.target} ({s.reason})" for s in skipped]
    return "\n".join(lines)


async def run_project(
    project_id: str,
    profile: Profile | str = Profile.LOCAL,
    *,
    docs_dir: str,
    verdict_input: dict[str, str],
    bundle_dir: str | None = None,
    dimension: Dimension | None = None,
    store: VerdictStore | None = None,
    verdict_dir: str | None = None,
    outbox_dir: str | None = None,
    run_id: str | None = None,
    client_factory: Callable[[str], BaseChatClient] | None = None,
    max_rounds: int = _DEFAULT_MAX_ROUNDS,
    max_tokens: int = _DEFAULT_MAX_TOKENS,
    top_k: int = 3,
    enable_layer1_hitl: bool = False,
    notify: Callable[[Verdict], None] | None = None,
    meter: TokenMeter | None = None,
    live_dry_run: bool = False,
    semantic_retrieval: bool = False,
    embedder: Embedder | None = None,
    mandate: Mandate | None = None,
    mcp_servers: tuple[McpServerConfig, ...] = (),
) -> RunResult | DryRunReport:
    """Run the vertical slice for ONE project. ``client_factory`` is the test-injection seam
    (defaults to the real backend). ``verdict_input`` carries the expert decision/rationale
    (Layer-2). ``bundle_dir`` (Fase 2a) makes the run OKF-bundle-driven: the project is derived
    from the bundle and, before generation, the candidate's prior verdicts in ``store`` are folded
    into the hypothesis prompt (Step-1 ExpeL wiring, målbilde §5/§7). ``verdict_dir`` (Fase 5,
    Steg 7, målbilde §3/§7) is the async file inbox: a folder of expert/persona-authored verdict
    files (plain JSON, R2 raw layer) MERGED into the store BEFORE the Step-1 fold, so a verdict
    dropped after an earlier run is consumed by this separate, later run — the long feedback loop,
    fully resumable across runs separated in time. The system READS this folder; it does not write
    to it (the expert/persona writes, målbilde §3). ``outbox_dir`` (Fase 2a, Steg 7 output layer,
    målbilde §3, R2) is the RAW OUTBOX: when set, the run's proposal + outcome artefacts are written
    there via ``outbox.write_outbox`` (``run_id`` is then REQUIRED — no wall-clock/uuid default, for
    byte-determinism). The outbox dir should be DISTINCT from any ``verdict_dir`` inbox: writing the
    outbox into a folder later read as an inbox would re-ingest raw agent output and bypass the
    Step-8 promotion gate (self-contamination) — documented here, not enforced. Raises
    ``pydantic.ValidationError`` on a bad contract and ``BudgetExceeded`` when the token/round cap is
    crossed, and ``ValueError`` when ``outbox_dir`` is set without a ``run_id``. ``live_dry_run``
    (S4.2, comparison protocol §4 pkt 2/3) is the offline drill: it walks the whole path up to the
    EAGER client build, writes the run-config artefact (when ``outbox_dir`` is set), and returns a
    ``DryRunReport`` BEFORE the first model call (``debate.run``) — zero chat calls.
    ``semantic_retrieval`` (S3.1) is the opt-in scaling SEAM — the deliverable is the extension
    point, not better retrieval. When true, a ``HybridRanker`` blends a cosine term over the
    embedded feature triple (sorted cost codes, measure type, magnitude bucket) with the structural
    score, which lets a prior verdict on a DIFFERENT cost-code set outrank one that ties
    structurally. The shipped ``FakeEmbedder`` is a deterministic sha256 projection carrying NO
    semantics, so over a structural tie the resulting order is deterministic but arbitrary;
    retrieval *quality* arrives only with an embedder injected via ``embedder=`` or
    ``--embedder-config``. Default false keeps the structural ranking exactly as before."""
    # 0. Fail-fast: an outbox write is byte-deterministic and keyed on run_id — no wall-clock default.
    if outbox_dir is not None and run_id is None:
        raise ValueError(
            "run_id is required when outbox_dir is set (no wall-clock/uuid default — the outbox "
            "artefacts are byte-deterministic and keyed on run_id)"
        )

    # 1. Fail-fast: validate ALL contracts (incl. the verdict-feedback shape) before any client.
    load_contracts(
        {"docs_dir": docs_dir, "top_k": top_k},
        {"max_rounds": max_rounds, "max_tokens": max_tokens},
        verdict_input,
    )

    # 1b. Long loop (Steg 7): ingest the async verdict inbox INTO the store before the Step-1 fold.
    # Merge (not replace) into the passed store so run_portfolio's cross-project threading stays
    # intact; store.add is idempotent on the content-hash id. A verdict that landed after an earlier
    # run thus reaches THIS run's hypothesis via the existing fold below — no change to the fold.
    if verdict_dir is not None:
        store = store if store is not None else VerdictStore(verdicts=[])
        for dropped in load_verdicts_from_dir(verdict_dir):
            store.add(dropped)

    # 2-3. Project + agent read-context + first-class citations. A bundle run derives ALL THREE from
    # the navigated OKF bundle via progressive disclosure (verdict layer EXCLUDED — målbilde §2/§4),
    # NOT keyword chunk-stuffing; the road path keeps the chunk-retrieval data source. ``debate_tools``
    # is the query-time retrieval surface — empty on the bundle path (navigation already placed the
    # curated context in the prompt, and a docs_dir==bundle_dir tool would re-leak the verdict layer).
    # S4.0 (F3): the run path SETS the validator's cost baseline, so the deterministic gate is
    # anchored to the project's real cost lines instead of the ones the proposal asserts.
    #   * road path: the reference project's own ``cost_items`` ARE the baseline -> always anchored.
    #   * bundle path: anchored only when the bundle SHIPS a ``cost-baseline.json``. A bundle written
    #     before the amendment (every commons-owned golden) is legitimately un-anchored -> None =
    #     pre-S4.0 behaviour. A baseline that exists but is malformed still raises (fail-closed).
    if bundle_dir is not None:
        bundle = okf.navigate_bundle(bundle_dir)
        project = _project_from_bundle(bundle_dir, project_id, bundle=bundle)
        baseline = okf.load_optional_cost_baseline(bundle_dir)
        # §4.1a context-scope: agents read ONLY dimension-scoped bundle knowledge (Step-3 filter);
        # dimension=None keeps the full context, byte-identical to before.
        context = okf.bundle_context(bundle, dimension=dimension.id if dimension else None)
        citations = bundle_citations(bundle)
        # What the navigation could NOT reach, taken from the run's ONE walk. The road path below
        # navigates no bundle at all, so its empty tuple is literally true rather than a stand-in.
        skipped_links: tuple[okf.SkippedLink, ...] = bundle.skipped
        debate_tools: list[Any] = []
    else:
        project = _project_by_id(project_id)
        baseline = baseline_from_project(project)
        chunks = retrieve_chunks("cost saving measure", docs_dir, top_k)
        citations = [chunk_dict_to_citation(c) for c in chunks]
        context = "\n".join(c["snippet"] for c in chunks)
        skipped_links = ()
        debate_tools = [make_retrieval_tool(docs_dir, top_k=top_k)]

    # Trekk B2 (krav 3): configured MCP servers become tools the AGENTS can call during the debate.
    # Appended to BOTH paths — on the bundle path they are the first tools that path has ever had.
    # Constructed here but NOT connected: an ``MCPTool`` is an async context manager, so the run
    # enters them around ``debate.run`` below and exits afterwards. Empty tuple -> the tool list is
    # byte-identical to the pre-Trekk-B one, and no network call is possible.
    live_mcp_tools = build_mcp_tools(mcp_servers) if mcp_servers else []
    debate_tools = debate_tools + live_mcp_tools
    if not citations:
        raise ValueError(f"no citable content in docs_dir: {docs_dir!r}")

    # 4. Budget + maker-checker debate (round-capped; Layer-1 HITL optional). The shared meter is
    # driven on the debate's chat calls by the BudgetMiddleware (the brief's named short-circuit
    # mechanism); ``debate_tools`` exposes the citation-bearing data source on the road path.
    meter = (
        meter
        if meter is not None
        else TokenMeter(Budget(max_tokens=max_tokens, max_rounds=max(max_rounds * 4, 4)))
    )
    factory = client_factory if client_factory is not None else _default_factory(profile)
    budget_mw = BudgetMiddleware(meter)
    # Trekk B4: the egress DECLARATION says what a run may contact; this records what it actually
    # called. Attached only when servers are configured — with none there is nothing to attribute a
    # call to, and the middleware list stays exactly what it was before Trekk B.
    call_recorder = ToolCallRecorder(tool_server_index(mcp_servers)) if mcp_servers else None
    debate = fresh_workflow(
        factory,
        max_rounds=max_rounds,
        enable_layer1_hitl=enable_layer1_hitl,
        tools=debate_tools,
        middleware=[budget_mw] if call_recorder is None else [budget_mw, call_recorder],
    )
    # S4.2 cut (comparison protocol §4 pkt 2/3): everything above is offline — contracts, budget, and
    # the EAGER client build (fresh_workflow constructs the proposer+checker clients, workflow.py:64).
    # Capture the run-config (resolved model per BUILT role, profile, params, token cap) and, for a
    # ``--live-dry-run``, STOP HERE — before the first (paid) model call at ``debate.run`` below.
    if outbox_dir is not None or live_dry_run:
        # ``resolved_models`` reflects the configured MAP (the default factory's model-ids for the M2
        # run). Under an injected ``client_factory`` the built clients may differ (e.g. "synthetic");
        # ``provenance.model`` (below) stays the authority on the client actually built.
        resolved_models = {role: resolve_model(profile, role) for role in _MAKER_CHECKER_ROLES}
        if outbox_dir is not None:
            assert run_id is not None  # narrowed by the step-0 guard (no wall-clock default)
            outbox.write_run_config(
                outbox_dir,
                run_id,
                profile=Profile(profile).value,
                resolved_models=resolved_models,
                max_rounds=max_rounds,
                max_tokens=max_tokens,
                top_k=top_k,
            )
        if live_dry_run:
            return DryRunReport(
                profile=Profile(profile).value,
                resolved_models=resolved_models,
                max_rounds=max_rounds,
                max_tokens=max_tokens,
                top_k=top_k,
                cost_baseline_anchored=baseline is not None,
                skipped_links=skipped_links,
            )
    # The MCP lifecycle (Trekk B2): entered HERE, after the dry-run cut above, so a dry run never
    # opens a connection — its promise to stop before the first call covers egress too. Constructed
    # tools that are never entered expose nothing, and ones never exited leave the process hanging,
    # so the stack owns both halves.
    async with AsyncExitStack() as mcp_stack:
        for live_tool in live_mcp_tools:
            await mcp_stack.enter_async_context(live_tool)
        result = await debate.run(
            f"Find a cost-saving measure for {project.id}.\nContext:\n{context}"
        )
    # F1: the candidate must derive from the DEBATE. Feed the proposer's converged output into
    # generation (retrieval context is the last-resort fallback only). The checker's verdict
    # (Step 3/4) is parsed from the SAME debate result and gates the outcome below.
    debate_output = _debate_text(result)
    checker_decision, checker_reason = _checker_verdict(result)
    gen_context = debate_output or context

    # Step-1 ExpeL wiring (Fase 2a, målbilde §5/§7): fold the candidate's prior verdicts INTO the
    # hypothesis context BEFORE generation, keyed on the OKF bundle's candidate features (available
    # pre-hypothesis). THIS is the one missing dataflow — previously ExpeL was computed
    # post-generation into a discarded SessionContext (step 7 below), so a prior verdict could not
    # reach the next hypothesis. Bundle-driven path with a populated store only.
    #
    # Scope of the --semantic-retrieval opt-in, stated precisely (an earlier version of this
    # comment claimed "the road path is untouched", which the flag made false): the ranker built
    # below is passed to ALL THREE retrievals this run performs — this fold, and the post-hoc
    # ExpeLContextProvider + store.retrieve in step 7 — so the flag reaches the road path's
    # proposal-keyed retrieval too. What IS untouched on the road path is the fold itself: it stays
    # bundle-gated, so a --docs-dir-only run remains single-shot either way.
    # S3.1 opt-in: build the hybrid ranker as a LOCAL, then pass it explicitly at each retrieval
    # this run performs. It is deliberately not assigned to ``store.retriever``: the store is
    # caller-owned (``run_portfolio`` threads one store across every project, and a library caller
    # may reuse theirs), so a store-global assignment leaked this run's opt-in into every later use
    # of that object — including a subsequent run with the flag OFF. Flag off => ranker stays None
    # => ``retrieve`` falls through to the StructuralRetriever default.
    ranker = (
        HybridRanker(
            embedder if embedder is not None else FakeEmbedder(),
            similarity,
            SEMANTIC_WEIGHT_DEFAULT,
        )
        if semantic_retrieval
        else None
    )

    if bundle_dir is not None and store is not None and store.verdicts:
        expel_query = bundle_candidate_features(bundle_dir)
        fewshot = ExpeLContextProvider(
            store, expel_query, k=top_k, retriever=ranker
        ).format_fewshot()
        gen_context = f"{fewshot}\n\n{gen_context}"

    # 5. Structured candidate(s) -> blocking validation on the NUMBERS; token bound = the meter.
    # Without a mandate this is the single pre-Trekk-A call, unchanged. With one, every
    # commissioned approach is evaluated in turn (and the run's own proposal last, when allowed),
    # each under the SAME meter — no new unbounded loop; the caps already in force are the bound.
    proposer_client = factory("proposer")

    # Step 5 (målbilde §5/§7): generation now returns its falsification history alongside the
    # outcome. ``_evaluate`` keeps its ``ValidatedProposal | Rejection`` shape so ``_evaluate_mandate``
    # is untouched, and the history is accumulated here in call order — one entry per approach that
    # needed correcting, concatenated (see ``RunResult.refinements`` for that honesty limit).
    refinements: list[Rejection] = []
    # Fase 1b, funn 1: the raw replies that did not parse. Owned HERE, beside ``meter``, and handed
    # down — not read back off a return value. ``generate_via_llm`` raises ``BudgetExceeded`` from
    # inside its own fetch loop when the round ledger runs out on unparseable replies (the measured
    # live failure), and on that path it returns nothing at all; a caller-owned accumulator is the
    # only shape that still holds the evidence afterwards. Concatenated across commissioned
    # approaches rather than keyed per approach, mirroring ``RunResult.refinements``' honesty limit.
    parse_failures: list[ParseFailure] = []

    async def _evaluate(approach: Approach | None) -> ValidatedProposal | Rejection:
        generated = await generate_via_llm(
            proposer_client,
            project,
            gen_context,
            meter,
            baseline=baseline,
            approach=approach,
            parse_failures=parse_failures,
        )
        refinements.extend(generated.refinements)
        return generated.outcome

    coverage: tuple[ApproachOutcome, ...] = ()
    evaluated: tuple[tuple[str, ValidatedProposal | Rejection], ...] = ()
    try:
        if mandate is None:
            validator_outcome = await _evaluate(None)
        else:
            validator_outcome, coverage, evaluated = await _evaluate_mandate(mandate, _evaluate)
    finally:
        # ``finally``, not ``except BudgetExceeded``: the round ledger is today's known way out, but
        # any exception leaving generation destroys the same evidence, and a per-exception-type list
        # is a list that goes stale. Written only when something actually failed to parse, so the
        # file's presence is the signal (a run whose replies all parse leaves the outbox unchanged).
        if outbox_dir is not None and parse_failures:
            assert run_id is not None  # narrowed by the step-0 guard (no wall-clock default)
            outbox.write_parse_failures(
                outbox_dir,
                run_id,
                failures=[{"text": f.text, "error": f.error} for f in parse_failures],
            )
    proposal = validator_outcome.proposal

    # 6. First-class provenance stamp (authoritative; independent of MAF Annotation).
    # F1: an injected client_factory stamps the injected client's REAL model ("unknown" is the
    # neutral fallback for a client that doesn't surface one -- never a fabricated name); the
    # default path keeps the deterministic resolve_model. validator_decision reflects the VALIDATOR
    # (the numbers) ONLY -- stamped from validator_outcome BEFORE the checker override below, so a
    # checker-gated proposal whose numbers passed is never mislabelled as validator-rejected.
    model = (
        (getattr(proposer_client, "model", None) or "unknown")
        if client_factory is not None
        else resolve_model(profile, "proposer")
    )
    stamp = ProvenanceStamp(
        citations=citations,
        model=model,
        role="proposer",
        validator_decision=(
            "validated" if isinstance(validator_outcome, ValidatedProposal) else "rejected"
        ),
        token_usage=meter.tokens,
        # Whether stage 0 of the deterministic gate had a baseline to reconcile against. Read off
        # the SAME ``baseline`` the validator was handed, so the record cannot describe a different
        # anchoring than the one that ran. The road path is anchored by construction (the reference
        # project's own cost_items ARE the baseline); a bundle is anchored iff it ships the file.
        cost_baseline_anchored=baseline is not None,
        # B4: which external service the debate actually called. Read AFTER the debate, so it is a
        # record rather than an intention. The honesty limit lives on ``ExternalCall`` itself: this
        # is the call and its source, not a verified rendering of the service's answer.
        external_calls=call_recorder.calls() if call_recorder is not None else [],
    )

    # 6b. Step 3/4 checker gate (målbilde §2/§6): the validator falsifies the numbers, the checker
    # falsifies the reasoning. An explicit checker REJECT blocks an otherwise-validated proposal; a
    # validator rejection (the numbers) already stands. Fail-open: APPROVE/absent never blocks.
    outcome: ValidatedProposal | Rejection
    if isinstance(validator_outcome, ValidatedProposal) and checker_decision == "reject":
        outcome = Rejection(proposal=proposal, reason=f"checker rejected: {checker_reason}")
    else:
        outcome = validator_outcome

    # 6c. Step 2 dimension scope gate (§4.1b): a candidate whose measure_type/codes fall OUTSIDE the
    # run's dimension is rejected. A scope/type gate placed AFTER the checker override (preserves
    # test_checker_gate_loadbearing) — NOT a new numeric gate: validate_proposal stays the only
    # blocking numeric gate and provenance.validator_decision (the numbers) is untouched. Mirrors the
    # override form: only an otherwise-standing ValidatedProposal can be flipped to a Rejection.
    if dimension is not None and isinstance(outcome, ValidatedProposal):
        feats = _features_of(proposal)
        if not admits(
            measure_type=feats.measure_type, codes=feats.affected_codes, dimension=dimension
        ):
            outcome = Rejection(
                proposal=proposal,
                reason=f"outside dimension {dimension.id!r}: measure_type={feats.measure_type!r}",
            )

    # 7. ExpeL (regression guard + traceability): exercises the two-arg extend_instructions
    #    injection on a REAL SessionContext (the Critical Fase-1 GA-signature guard), and surfaces
    #    the proposal-keyed retrieval for RunResult.retrieved. On the bundle path the load-bearing
    #    ExpeL->prompt dataflow already happened pre-generation (above); this block's SessionContext
    #    is NOT what reaches the prompt.
    store = store if store is not None else VerdictStore(verdicts=[])
    features = _features_of(proposal)
    provider = ExpeLContextProvider(store, features, k=top_k, retriever=ranker)
    sctx = SessionContext(input_messages=[], instructions=[])
    await provider.before_run(agent=None, session=None, context=sctx, state={})
    retrieved = store.retrieve(features, k=top_k, retriever=ranker) if store.verdicts else []

    # 8. Layer-2 (out-of-band): capture the durable verdict + persist; B11 notify is a stub.
    verdict = capture_verdict(features, verdict_input["decision"], verdict_input["rationale"])
    store.add(verdict)
    if notify is not None:
        notify(verdict)

    # S2.1 outbox (RAW output layer, målbilde §3): persist the run's proposal + outcome artefacts
    # when configured. Wired ONLY here — no new consumer (S5.1/S5.2 are Non-Goals this bolk). run_id
    # is guaranteed non-None by the fail-fast guard at the top.
    if outbox_dir is not None:
        assert run_id is not None  # narrowed by the step-0 guard; keeps the type checker honest
        if not evaluated:
            outbox.write_outbox(
                outbox_dir,
                run_id,
                outcome=outcome,
                provenance=stamp,
                checker_verdict=checker_decision,
                verdict_id=verdict.id,
            )
        else:
            # A5: one judgeable artefact PER evaluated approach. Without this the expert can only
            # judge the approach the run happened to select, so every other approach they
            # commissioned teaches the learning loop nothing. The per-approach set REPLACES the
            # single run-level pair rather than joining it — the selected approach is already among
            # these, and writing both would make ``hitl pending`` count it twice.
            for approach_id, approach_outcome in evaluated:
                # The SELECTED approach carries the run's final outcome, so the outbox can never
                # disagree with the ``RunResult``: the checker/dimension overrides above apply to
                # that one. The others carry the validator's verdict, which is the only falsifier
                # that ran on them.
                final = outcome if approach_outcome is validator_outcome else approach_outcome
                outbox.write_outbox(
                    outbox_dir,
                    run_id,
                    outcome=final,
                    # ``validator_decision`` must follow ITS OWN approach — stamping every artefact
                    # with the selected approach's decision would report a rejected candidate as
                    # validated. Everything else (model, citations, token usage) is the run's.
                    provenance=stamp.model_copy(
                        update={
                            "validator_decision": (
                                "validated"
                                if isinstance(approach_outcome, ValidatedProposal)
                                else "rejected"
                            )
                        }
                    ),
                    checker_verdict=checker_decision,
                    # The key an expert verdict on THIS candidate will arrive under (S3.2 content
                    # hash). Reusing the run's single verdict id would let one delivered verdict
                    # clear every approach from the pending queue.
                    verdict_id=verdict_key(_features_of(approach_outcome.proposal)),
                    approach_id=approach_id,
                )

    return RunResult(
        outcome=outcome,
        provenance=stamp,
        verdict=verdict,
        retrieved=retrieved,
        store=store,
        debate_output=debate_output,
        checker_verdict=checker_decision,
        coverage=coverage,
        refinements=tuple(refinements),
        skipped_links=skipped_links,
    )


def _aggregate(runs: tuple[RunResult, ...], store: VerdictStore) -> PortfolioResult:
    """Thin aggregate over the per-project runs (SC2): partition validated/rejected, total the
    validated claimed saving, and total every run's provenance token usage."""
    validated = [r for r in runs if isinstance(r.outcome, ValidatedProposal)]
    rejected = [r for r in runs if isinstance(r.outcome, Rejection)]
    return PortfolioResult(
        runs=runs,
        store=store,
        validated_count=len(validated),
        rejected_count=len(rejected),
        sum_claimed_saving_nok=sum(r.outcome.proposal.claimed_saving_nok for r in validated),
        sum_token_usage=sum(r.provenance.token_usage for r in runs),
    )


def _baseline_ore(projects: Iterable[Project]) -> int:
    """Addressable baseline in øre, quantized PER COST LINE and summed as integers (Kø-(p)).

    Both sides of the goal comparison must be computed in the same order. ``observed_ore`` is
    ``SavingsLedger``'s sum of per-candidate integer øre; a baseline that summed
    ``Project.total_cost`` floats and quantized the total ONCE put the threshold on a different
    scale — three ``60000.005`` NOK lines are ``18000003`` øre per line but ``18000001`` summed
    first, enough to flip a percent goal. Each cost line is a real amount, so the per-line value
    is the one that exists; integer addition also keeps the total order-independent, which
    ``Project.total_cost``'s float fold is not under the D-D wave model."""
    return sum(to_ore(item.total_cost) for p in projects for item in p.cost_items)


def _goal_limit_if_reached(goal: GoalContract, observed_ore: int, baseline_ore: int) -> int | None:
    """The threshold ``observed_ore`` MET (``>=``), or ``None`` if the goal is not yet reached. An
    absolute-øre target compares directly; a percent target is taken against ``baseline_ore`` (the
    addressable cost in øre). When both are set, reaching EITHER counts as met."""
    if goal.absolute_ore is not None and observed_ore >= goal.absolute_ore:
        return goal.absolute_ore
    if goal.percent is not None:
        if baseline_ore <= 0:
            raise ValueError(
                f"percent goal ({goal.percent}%) is meaningless against a non-positive baseline "
                f"({baseline_ore} øre): int(percent/100 * 0) == 0 would falsely read as 'goal "
                "reached'. Supply an absolute_ore target, or ensure the addressable baseline is > 0."
            )
        threshold = int(goal.percent / 100 * baseline_ore)
        if observed_ore >= threshold:
            return threshold
    return None


def _wave_snapshot(store: VerdictStore) -> VerdictStore:
    """A per-project copy of the shared store's CURRENT verdicts (D-D wave model).

    Two coroutines appending to one list would interleave by completion order, which no barrier
    could then undo. Giving every project in a wave its own copy removes the race at its source
    rather than serializing it away with a lock — a lock would order the appends by whoever won,
    which is exactly the nondeterminism being eliminated.

    ``retriever`` is carried across deliberately: it is the S3.1 opt-in seam, and a snapshot that
    dropped it would silently downgrade a caller-owned store's semantic retrieval to the
    structural default mid-pass.

    **Field-complete by construction.** ``dataclasses.replace`` carries over every field
    ``VerdictStore`` declares and overrides only ``verdicts``, so a field added later is copied
    without this function being touched. The hand-enumerated version this replaced could silently
    drop one — the exact defect it was first written with (the ``retriever`` omission), which its
    own docstring then recorded as a standing hazard. Deriving the copy from the dataclass removes
    the hazard instead of documenting it, and does so without the field-count assertion that idea
    was rejected for: there is nothing left to keep in sync.

    That docstring credited a Step-2 contract test with catching the omission. MEASURED while this
    change was made: no such coverage remained — reinstating the hand-enumerated form left the
    whole suite green, so the S3.1 retriever seam could be downgraded mid-pass in silence. The gate
    is now ``test_wave_snapshot_carries_every_field_except_the_copied_verdicts``.

    ``verdicts`` is still listed explicitly, and must be: ``replace`` copies field REFERENCES, so
    omitting it would hand back a store sharing the caller's list — the very race this snapshot
    exists to remove, reintroduced by the call that looks tidiest."""
    return replace(store, verdicts=list(store.verdicts))


def _merge_wave(store: VerdictStore, wave: Sequence[tuple[str, VerdictStore]]) -> None:
    """The deterministic merge barrier — the seam S3.3 rests on.

    Each project ran against its own snapshot, so its NEW verdicts are the ones absent from the
    wave-start store. They are merged back in **wave-submission order** — the order the caller
    listed the projects in — so the shared store's contents depend on the portfolio's membership
    and never on which project's model round-trips happened to finish first.

    **Submission order, NOT lexicographic project_id.** The plan specified a sort on ``project_id``;
    the Step-2 contract test measured that it produces a deterministic order which is nevertheless
    the WRONG one. On the shipped fixture, lexicographic order is BRU/FV42/RV13 while the sequential
    pass yields FV42/RV13/BRU, so a ``project_id`` sort satisfies "deterministic" while breaking
    "identical to ``concurrency=1``" — and the second is the actual contract. ``wave`` arrives in
    submission order, so preserving it is the fix.

    **Detach point: this function's ordering discipline is only half the seam — see
    ``_wave_snapshot``, which is the half that carries the load.** Because each project writes to
    its own copy, ``wave`` is never reordered by completion, so iterating it in order is already
    deterministic. Removing the SNAPSHOT is what turns the shared list back into a race and takes
    ``test_concurrent_pass_is_byte_identical_to_sequential`` RED. This is recorded plainly rather
    than dressing the loop in a ``sorted(...)`` that would re-sort an already-ordered list and read
    as a guard while guarding nothing.

    Safe to apply as-is because ``VerdictStore.add`` is first-write-wins per content-hash id
    (``verdicts.py:303-304``) — the wave-start verdicts every snapshot carries are re-offered and
    dropped, so only the new ones land."""
    seen = {v.id for v in store.verdicts}
    for _pid, snapshot in wave:
        for verdict in snapshot.verdicts:
            if verdict.id not in seen:
                store.add(verdict)
                seen.add(verdict.id)


def _waves(ids: list[str], k: int) -> list[list[str]]:
    """Partition ``ids`` into consecutive waves of at most ``k``, preserving caller order (D-D).

    Order preservation is the whole point: the wave model may change WHEN a project runs, never
    WHICH projects run nor in what sequence they are submitted. At ``k=1`` every wave holds a
    single project, so the wave loop degenerates to the pre-S3.3 sequential order by
    construction — that is what makes the parameter behaviour-preserving at its default."""
    return [ids[i : i + k] for i in range(0, len(ids), k)]


def _run_meter(
    meter_factory: Callable[[], TokenMeter] | None,
    portfolio_meter: PortfolioMeter | None,
    max_rounds: int,
) -> TokenMeter | None:
    """The per-project meter ``run_portfolio`` hands to one run: the injected test seam, a meter
    BOUND to the portfolio ledger, or ``None`` (letting ``run_project`` build its own, unchanged).

    The bound meter mirrors ``run_project``'s own construction (``max(max_rounds * 4, 4)``) so the
    only difference the portfolio cap introduces is the token ceiling and the shared ledger — the
    round budget is not quietly redefined along the way."""
    if meter_factory is not None:
        return meter_factory()
    if portfolio_meter is None:
        return None
    return TokenMeter(
        Budget(
            max_tokens=portfolio_meter.budget.max_tokens_per_run,
            max_rounds=max(max_rounds * 4, 4),
        ),
        portfolio=portfolio_meter,
    )


async def run_portfolio(
    project_ids: Sequence[str] | None = None,
    profile: Profile | str = Profile.LOCAL,
    *,
    dimension: Dimension | None = None,
    ledger: SavingsLedger | None = None,
    goals: GoalConfig | None = None,
    store: VerdictStore | None = None,
    client_factory: Callable[[str], BaseChatClient] | None = None,
    max_rounds: int = _DEFAULT_MAX_ROUNDS,
    max_tokens: int = _DEFAULT_MAX_TOKENS,
    top_k: int = 3,
    concurrency: int = 1,
    meter_factory: Callable[[], TokenMeter] | None = None,
    portfolio_meter: PortfolioMeter | None = None,
    semantic_retrieval: bool = False,
    embedder: Embedder | None = None,
    mandate: Mandate | None = None,
    mcp_servers: tuple[McpServerConfig, ...] = (),
) -> PortfolioResult:
    """Fan out over a portfolio of independent projects SEQUENTIALLY, composing ``run_project``
    as-is (every project's execution state — meter, debate, retrieval context — is built fresh
    per call, so sequential reuse is inherently isolated). ONE ``VerdictStore`` is threaded
    across every run, AND each project's ``bundle_dir``/``verdict_dir`` are threaded into
    ``run_project`` (Fase 2a S2.0): a bundle-backed project's ``bundle_dir``-gated Step-1 ExpeL fold
    then folds the store's prior verdicts into its hypothesis prompt, so a verdict captured on
    project k materially reaches project k+1's hypothesis — the cross-project learning loop is
    delivered, not merely asserted. ``verdict_dir`` also lets each project merge its async file inbox
    (Steg 7) on the portfolio path. ``project_ids`` defaults to every loaded project; an unknown id
    raises ``ValueError``. ``meter_factory`` (test seam) supplies a per-project meter — inject a
    shared meter to make the isolation guard go red (SC3).

    Step 8 goal-stop (SC6): ``ledger`` carries EARLIER, out-of-band HITL realizations (the long file
    loop / Steg-7 role split — ``run_project`` never realizes mid-pass, C3), so the check reads an
    ACCUMULATED sum, it does not build one during the pass. BEFORE running each pid, the accumulated
    realized sum is compared to ``goals`` with a ``>=`` boundary (reached, not exceeded — H1): a HARD
    portfolio goal ``break``s the pass (``stopped_early``); a HARD per-project goal SKIPS that pid
    (its further runs are future passes, not more runs here); a SOFT goal flags ``stop_reason`` but
    continues. Because the ledger is static during the pass, a reached goal is observed on the first
    iteration. ``stop_reason`` surfaces the first goal event; a per-project skip is also observable
    as the pid's absence from ``runs``.

    ``concurrency`` (S3.3, D-D wave model) caps how many projects run at once. It is validated
    BEFORE anything loads — an unusable cap must surface as an error, never as a silently-empty
    pass. At the default ``1`` the pass is the sequential one described above.

    Error policy (S3.3, SC3) is COLLECT-AND-CONTINUE: a project that raises does not cancel its
    siblings and does not abort the pass. Its exception is recorded as a ``RunFailure`` in
    ``failures`` while every project that completed keeps its full ``RunResult``, so a portfolio of
    independent projects degrades one project at a time rather than all at once. This holds at
    every ``k``, including ``k=1``. The policy is carried by ``return_exceptions=True`` on the
    wave's ``gather``; ``asyncio.TaskGroup`` would cancel the siblings and is rejected for that
    reason. Note the pass still raises for errors that are NOT one project's failure — an unknown
    ``project_id`` and a non-positive ``concurrency`` are caller mistakes and fail fast.

    Goal-stop under waves (S3.3 reading of the Step-8 semantics above — a reading, NOT a redesign):
    the checks run at WAVE ASSEMBLY, per member, before the wave starts. A HARD per-project goal
    removes that pid from its wave, so an excluded project is never STARTED — it costs no model
    round-trip and leaves no verdict behind, which a post-hoc filter over completed runs could not
    achieve. A HARD portfolio goal stops the pass, and the wave already assembled still crosses its
    merge barrier before the loop exits, so a stop never strands verdicts outside the store. A SOFT
    goal flags ``stop_reason`` and continues. Membership is identical at every ``concurrency`` — but
    the reason is that the ledger is STATIC during a pass (C3: no realization happens on the run
    path, which ``test_concurrent_pass_does_not_write_on_the_run_path`` pins), so every check reads
    the same accumulated sum regardless of when it runs. Wave-assembly placement buys the
    never-started property, not determinism; determinism is the one-writer rule's.

    **The one semantic difference ``k > 1`` introduces, stated plainly (OQ1).** Every project in a
    wave reads the WAVE-START store, so a verdict captured by project A does NOT reach project B's
    hypothesis prompt when A and B share a wave — sequentially it would. This is a deliberate trade,
    not an oversight: the snapshot is what removes the append race, and restoring intra-wave
    visibility would restore exactly the completion-order dependence the barrier exists to
    eliminate (what B saw would depend on whether A happened to finish first). Learning therefore
    flows ACROSS wave boundaries, not within them, and ``concurrency`` is the knob that trades
    learning granularity for wall-clock — at ``k=1`` nothing changes, and at ``k=len(portfolio)``
    the pass learns nothing from itself.

    On the SHIPPED reference fixture this difference is invisible in outcomes, because the Step-1
    ExpeL fold is ``bundle_dir``-gated and no reference project sets ``bundle_dir`` — the chain is
    live for store CONTENT but inert for OUTCOMES. That is a property of the fixture, never a
    property of the design, so it is pinned on a bundle-backed pair where the fold does fire
    (``test_intra_wave_visibility_is_the_documented_semantic_difference``): same fixture, same
    sentinel, only the wave boundary moves. Store content stays identical across ``k`` — the
    difference is confined to what each project READ, never to what the pass produced or persisted.

    ``portfolio_meter`` (S3.4, F10) installs the GLOBAL token cap. Without it nothing changes: each
    run is bounded only by its own ``max_tokens``, so N projects can cost N times that with no
    ceiling over the pass. With it, one ``PortfolioMeter`` is shared by every run (each run's meter
    is BOUND to it, which is why ``meter_factory`` — whose meters are unbound — is refused
    alongside it: accepting both would run a pass that looks capped and is not), and the cap is
    enforced in three places that are deliberately different:

    - **At startup**, a remainder that cannot fund one run raises ``BudgetRefused`` — a pass that
      can afford zero projects is a caller mistake, not a result.
    - **At wave assembly**, a project that cannot be funded is NEVER STARTED, and the pass stops
      with ``budget_stop`` + ``stopped_early``, every completed run preserved. Never-started is the
      property that matters: an unfunded project that is merely interrupted mid-run has already
      cost model round-trips. Because every member of a wave is checked against the SAME pre-wave
      remainder, admission RESERVES each member's requirement as it goes — otherwise a wave of k
      would over-commit the cap by up to k runs. This makes membership identical at every
      ``concurrency``, matching the goal-stop's guarantee above.
    - **Before each chat call** (``BudgetMiddleware``'s pre-call guard), a call the remainder cannot
      pay for is refused rather than made. This is what bounds a run that was funded at admission
      but whose siblings drained the pass while it was in flight; such a run surfaces as a
      ``RunFailure``, not as overspend.

    Spend is carried ACROSS passes by seeding the meter from ``budget.read_spend`` and writing
    ``budget.write_spend`` afterwards. Those are the caller's calls, not this function's — the pass
    reads its ledger, it does not own the file (mirroring the Steg-7 role split)."""
    if concurrency < 1:
        raise ValueError(
            f"concurrency must be >= 1, got {concurrency}: a non-positive wave size would run no "
            "projects at all and read as an empty portfolio. Pass 1 for the sequential pass."
        )
    if portfolio_meter is not None and meter_factory is not None:
        raise ValueError(
            "portfolio_meter and meter_factory are mutually exclusive: a meter_factory meter is "
            "not bound to the portfolio ledger, so the global cap would be silently unenforced"
        )
    # Startup refusal (fail-fast, before anything loads): a pass that cannot fund its first run
    # must not begin. Raised, not returned — an empty PortfolioResult would read as "nothing to do".
    if portfolio_meter is not None and not portfolio_meter.can_fund_run():
        raise BudgetRefused(portfolio_meter.remaining(), portfolio_meter.required_per_run)
    projects = {p.id: p for p in load_reference_projects()}
    ids = list(project_ids) if project_ids is not None else list(projects)
    store = store if store is not None else VerdictStore(verdicts=[])
    # S3.1: the flag is FORWARDED per project rather than installed on the shared store here. The
    # store is threaded across every project in the pass (and may be caller-owned), so a pre-loop
    # assignment outlived the pass; forwarding keeps the opt-in scoped to each run's own retrievals.
    ledger = ledger if ledger is not None else SavingsLedger(entries=[])
    goals = goals if goals is not None else GoalConfig()
    portfolio_baseline_ore = _baseline_ore(projects[p] for p in ids if p in projects)

    runs: list[RunResult] = []
    failures: list[RunFailure] = []
    stopped_early = False
    stop_reason: GoalReached | None = None
    budget_stop: BudgetStop | None = None
    for wave_ids in _waves(ids, concurrency):
        members: list[str] = []
        # Tokens committed to members already admitted to THIS wave but that have not spent yet.
        # Every member reads the same pre-wave remainder, so without this a wave of k would admit
        # k projects off one project's worth of budget.
        reserved = 0
        for pid in wave_ids:
            if pid not in projects:
                raise ValueError(f"unknown project_id: {pid!r}")
            project = projects[pid]

            if goals.portfolio is not None:
                observed = ledger.portfolio_total()
                limit = _goal_limit_if_reached(goals.portfolio, observed, portfolio_baseline_ore)
                if limit is not None:
                    if goals.portfolio.mode == "hard":
                        stopped_early = True
                        stop_reason = GoalReached("portfolio", None, limit, observed)
                        break
                    if stop_reason is None:
                        stop_reason = GoalReached("portfolio", None, limit, observed)  # soft flag

            per_project_goal = goals.per_project.get(pid)
            if per_project_goal is not None:
                observed = ledger.per_project_total(pid)
                limit = _goal_limit_if_reached(
                    per_project_goal, observed, _baseline_ore((project,))
                )
                if limit is not None:
                    if stop_reason is None:
                        stop_reason = GoalReached("project", pid, limit, observed)
                    if per_project_goal.mode == "hard":
                        continue  # skip THIS pid; the rest of the pass proceeds

            # S3.4 funding check, placed AFTER the goal checks: a reached goal is success and owns
            # the stop when both apply, and a pid a hard per-project goal already skipped costs no
            # budget, so it must not consume a reservation.
            if portfolio_meter is not None and not portfolio_meter.can_fund_run(reserved=reserved):
                stopped_early = True
                budget_stop = BudgetStop(
                    limit_tokens=portfolio_meter.budget.max_total_tokens,
                    spent_tokens=portfolio_meter.spent,
                    remaining_tokens=portfolio_meter.remaining(),
                    required_tokens=portfolio_meter.required_per_run,
                )
                break

            if portfolio_meter is not None:
                reserved += portfolio_meter.required_per_run
            members.append(pid)

        # Every project in the wave reads the SAME wave-start state and writes only its own copy,
        # so no two coroutines touch one list. The snapshot is what makes the barrier sufficient.
        snapshots = [(pid, _wave_snapshot(store)) for pid in members]

        # run_portfolio only drives full runs (never dry-run), so the return narrows to RunResult;
        # the cast keeps the widened run_project signature honest without an @overload duplication.
        # ``return_exceptions=True`` is the collect-and-continue seam (SC3): one project's
        # exception must not cancel its siblings. ``asyncio.TaskGroup`` is deliberately NOT used —
        # it cancels the remaining tasks on first exception, which is this policy's exact negation.
        wave_results = await asyncio.gather(
            *(
                run_project(
                    pid,
                    profile,
                    docs_dir=projects[pid].docs_dir,
                    verdict_input=projects[pid].verdict_input,
                    bundle_dir=projects[pid].bundle_dir,
                    verdict_dir=projects[pid].verdict_dir,
                    dimension=dimension,
                    store=snapshot,
                    client_factory=client_factory,
                    max_rounds=max_rounds,
                    max_tokens=max_tokens,
                    top_k=top_k,
                    semantic_retrieval=semantic_retrieval,
                    embedder=embedder,
                    mandate=mandate,
                    mcp_servers=mcp_servers,
                    meter=_run_meter(meter_factory, portfolio_meter, max_rounds),
                )
                for pid, snapshot in snapshots
            ),
            return_exceptions=True,
        )
        # ``gather`` resolves in ARGUMENT order, not completion order, and waves follow
        # ``project_ids`` — so ``runs`` stays in caller order however the schedule interleaved, and
        # position is a sound key for pairing each result back to the pid that produced it.
        #
        # ``strict=True`` is FUTURE-PROOFING and is deliberately untested — measured, not assumed:
        # dropping it leaves the whole suite green, because ``gather`` is constructed from exactly
        # ``snapshots``, so the two lengths cannot diverge today. A test could only go red by
        # manufacturing a mismatch, which would exercise ``zip`` rather than this pass. It earns its
        # place against a later edit that filters or extends one sequence without the other — then a
        # silent truncation would mis-attribute every result after the gap, and this fails instead.
        for (pid, _snapshot), outcome in zip(snapshots, wave_results, strict=True):
            # ``BaseException``, NOT ``Exception``, and the width is load-bearing: a member raising
            # ``asyncio.CancelledError`` (a BaseException since 3.8) is COLLECTED by
            # ``return_exceptions=True`` and must land in ``failures``. Narrowed to ``Exception`` it
            # would fall to the ``cast`` below and put a live exception object into ``runs``, which
            # dies later in ``_aggregate`` — a crash pointing away from its cause. Gated by
            # ``tests/test_portfolio_failure_accounting_loadbearing.py``.
            if isinstance(outcome, BaseException):
                failures.append(
                    RunFailure(
                        project_id=pid, error=str(outcome), error_type=type(outcome).__name__
                    )
                )
            else:
                runs.append(cast(RunResult, outcome))
        # ``snapshots`` is handed over UNFILTERED. ``_merge_wave`` consumes the ORDER this sequence
        # arrives in and nothing else, so that order is the contract. Measured honestly: filtering
        # the failed members out here would in fact be harmless, because filtering preserves
        # relative order — the variant is green. What is NOT harmless is anything that REORDERS
        # (or re-pairs) the sequence, and keeping the list whole is what leaves no place for that
        # mistake to be written. A failed project's snapshot holds no new verdicts, so it costs
        # nothing to pass through.
        _merge_wave(store, snapshots)

        if stopped_early:
            break

    base = _aggregate(tuple(runs), store)
    if stopped_early or stop_reason is not None or failures or budget_stop is not None:
        return replace(
            base,
            stopped_early=stopped_early,
            stop_reason=stop_reason,
            failures=tuple(failures),
            budget_stop=budget_stop,
        )
    return base


# The roles ``debate``/``generate`` ask the factory for. Fixed here so a malformed replies file is
# caught at the door instead of mid-run.
_SCRIPTED_ROLES = ("proposer", "checker")

# The honesty banner for the scripted door. It is a REQUIREMENT, not decoration (målbilde §1):
# a scripted run that reads like a model run is worse than having no offline mode at all, so this
# prints on every scripted invocation and mirrors ``simulation.main``'s banner.
_SCRIPTED_BANNER = (
    "=" * 78
    + "\nSCRIPTED OFFLINE RUN — every agent reply is read from your --scripted-replies file."
    + "\nNO MODEL WAS CALLED (ingen modellkall gjort). The context navigation, the debate"
    + "\nplumbing, the deterministic validator and the verdict are real; the agents' answers"
    + "\nare yours, not a model's. This proves the loop closes — not that an LLM would say this."
    + "\n"
    + "=" * 78
)


def _load_scripted_replies(path: str) -> dict[str, str]:
    """Load the caller's scripted answers, fail-fast. Every role the debate can ask for must be
    present AND a string: a missing role would otherwise surface as a ``KeyError`` deep inside
    ``scripted_factory``'s lookup, mid-run, long after the run appeared to start cleanly."""
    try:
        raw = json.loads(Path(path).read_text(encoding="utf-8"))
    except FileNotFoundError as exc:
        raise ValueError(f"--scripted-replies file not found: {path}") from exc
    except json.JSONDecodeError as exc:
        raise ValueError(f"--scripted-replies is not valid JSON ({path}): {exc}") from exc
    if not isinstance(raw, dict):
        raise ValueError(f"--scripted-replies must be a JSON object of role -> reply ({path})")
    missing = [r for r in _SCRIPTED_ROLES if not isinstance(raw.get(r), str)]
    if missing:
        raise ValueError(
            f"--scripted-replies needs a string reply for each of {', '.join(_SCRIPTED_ROLES)}; "
            f"missing or non-string: {', '.join(missing)} ({path})"
        )
    return {role: raw[role] for role in _SCRIPTED_ROLES}


def main(argv: list[str] | None = None) -> int:
    """Single-command console entry: run the slice for one project against a docs folder."""
    import argparse
    import asyncio
    import sys

    parser = argparse.ArgumentParser(description="portfolio-optimiser vertical slice")
    # project_id + --docs-dir are relaxed from required to a mode-conditional refusal (below): the
    # single-project path still requires both, but portfolio mode takes neither. The compensating
    # guard keeps the legacy contract failing loudly (rc 1 refusal) instead of via argparse exit 2.
    parser.add_argument("project_id", nargs="?", default=None)
    parser.add_argument("--profile", default="local")
    parser.add_argument("--docs-dir", default=None)
    parser.add_argument(
        "--bundle-dir", default=None, help="OKF bundle dir (enables the Step-1 fold)"
    )
    parser.add_argument(
        "--verdict-dir",
        default=None,
        help="async verdict inbox: a folder of dropped expert verdicts, ingested before generation "
        "(the long loop — a verdict that landed after an earlier run is consumed by this run)",
    )
    parser.add_argument(
        "--dimension-config",
        default=None,
        help="fail-fast dimension scope config (JSON): scopes the run to one cost axis; a "
        "missing or malformed file refuses the run (authoritative startup config, not a RAW inbox)",
    )
    parser.add_argument(
        "--mandate",
        default=None,
        metavar="FILE",
        help="run mandate (JSON, fail-fast): what a domain expert commissions this run to "
        "evaluate — named approaches and/or the system's own — plus the objective and success "
        "criteria. The run ANNOUNCES it before the first model call and SETTLES against it "
        "afterwards, one row per approach. Valid in both modes; in portfolio mode it applies to "
        "every project in the pass",
    )
    parser.add_argument(
        "--mcp-config",
        default=None,
        metavar="FILE",
        help="external MCP servers this run may contact (JSON, fail-fast): name, transport "
        "(stdio|http), coordinates, the ALLOWED tool names, a timeout, and optionally the NAME of "
        "an env var holding the credential (never the credential itself). Every server and tool is "
        "named in the run announcement BEFORE the first call; without this flag no network call is "
        "made at all",
    )
    parser.add_argument(
        "--embedder-config",
        default=None,
        help='fail-fast embedder config (JSON, e.g. {"type": "fake"}): selects the embedder '
        "used by --semantic-retrieval from a CLOSED registry. Not an import path — a config file "
        "can never name arbitrary code to load (a new embedder is added as a registry branch)",
    )
    parser.add_argument(
        "--outbox-dir",
        default=None,
        help="RAW outbox dir for the run's proposal/outcome artefacts (REQUIRES --run-id). Should "
        "differ from --verdict-dir: the two folders have OPPOSITE ownership (the system writes the "
        "outbox, the expert writes the inbox). Measured: sharing one folder is inert TODAY, because "
        "the outbox artefacts are named {run_id}-*.json and carry none of the verdict keys, so the "
        "tolerant inbox loader skips them — the hazard is a future verdict-shaped artefact in the "
        "outbox being re-ingested past the Step-8 promotion gate (self-contamination). Not "
        "CLI-enforced: there is no reachable contamination to refuse",
    )
    parser.add_argument(
        "--run-id",
        default=None,
        help="stable run id for --outbox-dir artefacts (required when --outbox-dir is set; no "
        "wall-clock/uuid default — the outbox artefacts are byte-deterministic)",
    )
    parser.add_argument(
        "--portfolio",
        action="store_true",
        help="portfolio mode: dispatch to run_portfolio over all reference projects (or the single "
        "given PROJECT_ID). Takes --goals/--ledger/--dimension-config; the single-project-only flags "
        "are refused in this mode (the two CLI modes are a documented partition)",
    )
    parser.add_argument(
        "--goals",
        default=None,
        help="portfolio mode: goal config JSON (fail-fast) — the GoalReached stop is checked against "
        "the ledger before each project",
    )
    parser.add_argument(
        "--ledger",
        default=None,
        help="portfolio mode: accumulated savings ledger JSON (fail-fast) read for the goal-stop "
        "(earlier out-of-band HITL realizations — never built during the pass)",
    )
    parser.add_argument(
        "--semantic-retrieval",
        action="store_true",
        help="S3.1 opt-in scaling SEAM: blend a cosine term over the embedded feature triple with "
        "the structural score, so a prior verdict on a DIFFERENT cost-code set can outrank one that "
        "ties structurally. The shipped embedder is a semantics-free sha256 projection — this buys "
        "the extension point, not better retrieval; inject a real one with --embedder-config. "
        "Accepted in both modes, but in single-project mode it REQUIRES --bundle-dir and "
        "--verdict-dir (without them it cannot take effect, and is refused rather than ignored). "
        "OFF by default, and off means the structural ranking is unchanged",
    )
    parser.add_argument("--decision", default="approved", choices=["approved", "rejected"])
    parser.add_argument("--rationale", default="reviewed by expert")
    parser.add_argument(
        "--live-dry-run",
        action="store_true",
        help="offline drill: build contracts/clients/budget, STOP before the first model call",
    )
    parser.add_argument(
        "--scripted-replies",
        default=None,
        metavar="FILE",
        help="offline WHOLE-LOOP run over your own bundle with ZERO model calls: FILE is JSON "
        '{"proposer": "<reply>", "checker": "<reply>"} and those fixed strings stand in for every '
        "model answer. Unlike --live-dry-run (which stops before the first call) the complete loop "
        "runs — hypothesis, debate, deterministic validator, verdict. The answers are yours, not a "
        "model's, and the run says so on every invocation",
    )
    parser.add_argument(
        "--report",
        action="store_true",
        help="S5.4 read-only value report: roll up the --ledger's realized savings (per-project + "
        "portfolio totals, flagged cross-dimension overlaps, per-entry provenance) to stdout. "
        "Mode-exclusive: only --ledger/--json are permitted alongside it; makes NO model calls",
    )
    parser.add_argument(
        "--json",
        action="store_true",
        help="value report output form (requires --report): emit the roll-up as deterministic JSON "
        "instead of the human table",
    )
    args = parser.parse_args(argv)

    # S5.4: read-only value-report dispatch — placed FIRST (right after parse_args, BEFORE the
    # mode-exclusivity block below) so it returns before any model/portfolio path can start and no
    # later branch can shadow it (the bare `--ledger`-outside-portfolio refusal at the elif below is
    # left UNCHANGED — a bare --ledger with no --report still flows there and refuses as before).
    if args.json and not args.report:
        # A stray --json is never silently ignored (honors S5.3's "refused, never ignored" partition).
        print("run refused: --json requires --report", file=sys.stderr)
        return 1
    if args.report:
        # Mode-exclusivity as an ALLOWLIST (not a short blocklist): report mode permits ONLY --ledger
        # and --json; ANY other distinguishable mode/config flag is refused — else --report --goals
        # would silently drop --goals, whereas bare --goals is refused below (adding --report must not
        # suppress an existing refusal). --decision/--rationale are excluded: their non-None argparse
        # defaults are indistinguishable from an explicit value (exactly as the block below excludes
        # them); they are inert in report mode.
        report_forbidden = {
            "--portfolio": args.portfolio,
            "--live-dry-run": args.live_dry_run,
            "PROJECT_ID": args.project_id is not None,
            "--goals": args.goals is not None,
            "--docs-dir": args.docs_dir is not None,
            "--bundle-dir": args.bundle_dir is not None,
            "--verdict-dir": args.verdict_dir is not None,
            "--outbox-dir": args.outbox_dir is not None,
            "--run-id": args.run_id is not None,
            "--dimension-config": args.dimension_config is not None,
            "--semantic-retrieval": args.semantic_retrieval,
            "--embedder-config": args.embedder_config is not None,
            "--scripted-replies": args.scripted_replies is not None,
        }
        if any(report_forbidden.values()):
            print(
                "run report refused: mode-exclusive (only --ledger/--json permitted with --report)",
                file=sys.stderr,
            )
            return 1
        if not args.ledger:
            # Guards SavingsLedger.load(None) -> Path(None) TypeError (NOT in the load except tuple).
            print("run report refused: --report requires --ledger <file>", file=sys.stderr)
            return 1
        try:
            # `report_ledger`, not `ledger`: the portfolio branch below binds `ledger` as
            # `SavingsLedger | None`, so reusing that name here (type `SavingsLedger`) collides on
            # mypy's function-scoped declared type.
            report_ledger = SavingsLedger.load(args.ledger)
        except (FileNotFoundError, ValidationError, ValueError) as exc:
            # A load failure must never masquerade as a real zero-savings result (SC5): stderr + rc 1,
            # no table. Only a successfully-loaded (possibly empty) ledger prints.
            print(f"run report refused: {exc}", file=sys.stderr)
            return 1
        rep = build_value_report(report_ledger)  # NB: `rep`, not `report` (`report` is bound below)
        print(dump_report_json(rep) if args.json else format_report_text(rep))
        return 0

    # Step 4: mode-exclusivity validation (structured refusal, NOT argparse.error — keeps the rc 1
    # refusal contract). The two CLI modes are a documented partition: single-project-only flags are
    # refused in portfolio mode, and --goals/--ledger are refused outside it — never silently ignored.
    # --decision/--rationale are excluded: their non-None argparse defaults make an explicit value
    # indistinguishable from the default, so an honest refusal is unimplementable (they are inert in
    # portfolio mode; the README documents that). --dimension-config is valid in BOTH modes.
    if args.portfolio:
        single_only = {
            "--docs-dir": args.docs_dir,
            "--bundle-dir": args.bundle_dir,
            "--verdict-dir": args.verdict_dir,
            "--outbox-dir": args.outbox_dir,
            "--run-id": args.run_id,
            "--live-dry-run": args.live_dry_run,
        }
        offending = [name for name, value in single_only.items() if value]
        if offending:
            print(
                f"portfolio run refused: {', '.join(offending)} belong to single-project mode, "
                "not --portfolio (the two CLI modes are a documented partition)",
                file=sys.stderr,
            )
            return 1
    elif args.goals is not None or args.ledger is not None:
        print(
            "run refused: --goals/--ledger require --portfolio mode",
            file=sys.stderr,
        )
        return 1

    # Single-project mode requires PROJECT_ID + --docs-dir (compensating for the relaxed argparse
    # required/positional so the legacy contract keeps failing loudly via the refusal surface).
    # HOISTED above the scripted door (below) so an incomplete argv is refused BEFORE the honesty
    # banner could claim a scripted run happened; the refusal ORDER within single-project mode
    # (required args -> semantic-retrieval -> scripted) is unchanged.
    if not args.portfolio and (args.project_id is None or args.docs_dir is None):
        print(
            "run refused: single-project mode requires PROJECT_ID and --docs-dir "
            "(use --portfolio for portfolio mode)",
            file=sys.stderr,
        )
        return 1

    # --semantic-retrieval is refused, never silently ignored (the repo's flag contract). In
    # single-project mode it can only do observable work with BOTH of these: the Step-1 fold is
    # gated on ``bundle_dir``, and ``--verdict-dir`` is the only route by which ``main()`` can hand
    # ``run_project`` a non-empty store (``main()`` never passes ``store=``, and ``run_project``
    # never seeds one). Without them the flag would rank nothing that reaches a prompt, and
    # ``RunResult.retrieved`` never leaves the process — ``main()`` prints one outcome line only.
    #
    # DELIBERATELY STATIC. There is no runtime "refuse if the store ends up empty" check: a
    # missing, empty or partially-skipped inbox is the Steg-7 tolerant-load contract, so refusing
    # there would fire on a legitimate first run. The refusal is therefore necessary, not
    # sufficient — it catches the configuration that CANNOT work, not every run that finds nothing.
    #
    # main() only. As a library API, ``run_project(semantic_retrieval=True, store=…)`` with a
    # caller-supplied store stays legitimate — that is the path the tests drive. Portfolio mode is
    # unaffected: ``run_portfolio`` always resolves a store and populates it by cross-project capture.
    if not args.portfolio and args.semantic_retrieval:
        required = {"--bundle-dir": args.bundle_dir, "--verdict-dir": args.verdict_dir}
        missing = [name for name, value in required.items() if not value]
        if missing:
            print(
                f"run refused: --semantic-retrieval requires {' and '.join(missing)} in "
                "single-project mode (the Step-1 fold is bundle-path-only, and --verdict-dir is "
                "the only route to a non-empty store)",
                file=sys.stderr,
            )
            return 1

    # --embedder-config selects the embedder for the HybridRanker, and that ranker is built ONLY
    # when semantic_retrieval is on; the default StructuralRetriever takes no embedder at all. So
    # without --semantic-retrieval the config is loaded fail-fast and then dropped on the floor —
    # MEASURED, not inferred: an injected embedder is consulted ZERO times with the flag off and
    # once with it on (tests/test_run_cli.py::test_injected_embedder_is_never_consulted_with_the
    # _flag_off + its control). That is the silent-ignore this CLI's flag contract exists to
    # prevent, and the same ground on which --semantic-retrieval itself is refused above when it
    # cannot take effect.
    #
    # REFUSED, not wired — the opposite call from --scripted-replies in portfolio mode, and for a
    # stated reason: there the seam already existed (run_portfolio takes the same client_factory),
    # so refusing would have left a whole mode without an offline door. Here there is nothing to
    # wire to; an embedder has no job outside the hybrid ranker.
    #
    # MODE-INDEPENDENT (hence above the portfolio dispatch, not inside either branch): both modes
    # gate the embedder on the same flag, since run_portfolio forwards it to run_project unchanged.
    # Placed ABOVE the scripted door for the reason the required-args guard was hoisted there — a
    # refused run must not first print a banner claiming a scripted loop closed.
    if args.embedder_config is not None and not args.semantic_retrieval:
        print(
            "run refused: --embedder-config requires --semantic-retrieval (the embedder is only "
            "consulted by the hybrid ranker that flag builds — without it the config would be "
            "loaded and then ignored)",
            file=sys.stderr,
        )
        return 1

    # The commission, loaded fail-fast BEFORE anything runs: a missing or malformed mandate is
    # REFUSED rather than degraded to "no mandate", because the settlement would then describe work
    # nobody ordered. Placed with the other refusals and ABOVE the scripted banner, for the same
    # reason the required-args guard was hoisted there — a refused run must not first print a
    # banner claiming a scripted loop closed.
    mandate: Mandate | None = None
    if args.mandate is not None:
        try:
            mandate = load_mandate(args.mandate)
        except (FileNotFoundError, ValidationError, ValueError) as exc:
            print(f"run refused: {exc}", file=sys.stderr)
            return 1

    # The egress config, loaded fail-fast alongside the commission. Degrading a broken one to "no
    # external services" would make the announcement describe a run nobody configured, and a
    # partially-parsed one could contact a subset nobody chose.
    mcp_servers: tuple[McpServerConfig, ...] = ()
    if args.mcp_config is not None:
        try:
            mcp_servers = load_mcp_config(args.mcp_config)
        except (FileNotFoundError, ValidationError, ValueError) as exc:
            print(f"run refused: {exc}", file=sys.stderr)
            return 1

    # The scripted door (offline WHOLE-loop run over the caller's own data). Resolved BEFORE the
    # dry-run branch so the two offline modes cannot both be honoured — and BEFORE the portfolio
    # dispatch, because the door serves BOTH modes. It originally sat below that dispatch, which
    # made ``--portfolio --scripted-replies`` silently drop the flag: no banner, and four real
    # model calls attempted (measured). That is the failure mode the "refused, never ignored"
    # partition exists to prevent, and here the honest resolution is to WIRE it — ``run_portfolio``
    # already exposes the same ``client_factory`` seam ``run_project`` does, so refusing would have
    # left portfolio mode with no offline door at all for an adopter without a model budget.
    scripted_client_factory: Callable[[str], BaseChatClient] | None = None
    if args.scripted_replies is not None:
        if args.live_dry_run:
            # Both are offline, and they contradict: --live-dry-run stops before the first model
            # call while --scripted-replies answers every one of them. Refuse rather than let one
            # silently win (S5.3's "refused, never ignored" partition). Unreachable in portfolio
            # mode, where --live-dry-run is already refused by the single_only partition above.
            print(
                "run refused: --scripted-replies and --live-dry-run are both offline modes and "
                "contradict each other (dry-run stops before the first model call; scripted "
                "answers all of them) — pick one",
                file=sys.stderr,
            )
            return 1
        try:
            replies = _load_scripted_replies(args.scripted_replies)
        except (OSError, ValueError) as exc:
            print(f"run refused: {exc}", file=sys.stderr)
            return 1
        # Imported HERE rather than at module scope: ``simulation`` imports ``run``, so a top-level
        # import would be circular. The scripted client already exists as MAF-side scaffolding —
        # this flag is a DOOR onto that one seam, never a second implementation of it.
        from portfolio_optimiser.simulation import scripted_factory

        scripted_client_factory = scripted_factory(replies, [])
        print(_SCRIPTED_BANNER)

    if mandate is not None:
        # The scope line reads the dimension config only to NAME it. A config that fails to load is
        # left unnamed here and refused a moment later by the dispatch below, which stays the single
        # owner of that refusal (and of its mode-specific wording) — announcing must never change
        # which error an operator sees.
        dimension_label: str | None = None
        if args.dimension_config is not None:
            try:
                _dim = load_dimension(args.dimension_config)
            except (FileNotFoundError, ValidationError, ValueError):
                dimension_label = None
            else:
                dimension_label = f"{_dim.id} ({_dim.label})"
        print(
            announce(
                mandate,
                project_id=args.project_id or "the portfolio",
                max_rounds=_DEFAULT_MAX_ROUNDS,
                max_tokens=_DEFAULT_MAX_TOKENS,
                dimension_label=dimension_label,
                external_services=service_labels(mcp_servers),
            )
        )
    elif mcp_servers:
        # The egress declaration must NOT depend on a commission being present. Without this
        # branch, configuring servers and omitting --mandate would contact third parties with
        # nothing printed at all — silent egress, which this repo forbids outright.
        print("Contacts:    " + ", ".join(service_labels(mcp_servers)))

    if args.portfolio:
        # Portfolio mode (Step 3): dispatch to the EXISTING run_portfolio via the fail-fast loaders
        # (run_portfolio itself is unchanged). Loader/ValueError failures surface through the same
        # structured-refusal contract as the single-project path (stderr + rc 1, no traceback).
        try:
            goals = load_goal_config(args.goals) if args.goals else None
            ledger = SavingsLedger.load(args.ledger) if args.ledger else None
            dimension = load_dimension(args.dimension_config) if args.dimension_config else None
            embedder = (
                build_embedder(load_embedder_config(args.embedder_config))
                if args.embedder_config
                else None
            )
            project_ids = (args.project_id,) if args.project_id is not None else None
            portfolio_result = asyncio.run(
                run_portfolio(
                    project_ids,
                    args.profile,
                    dimension=dimension,
                    embedder=embedder,
                    ledger=ledger,
                    goals=goals,
                    semantic_retrieval=args.semantic_retrieval,
                    client_factory=scripted_client_factory,
                    mandate=mandate,
                    mcp_servers=mcp_servers,
                )
            )
        except (ValueError, FileNotFoundError, ValidationError) as exc:
            print(f"portfolio run refused: {exc}", file=sys.stderr)
            return 1
        for r in portfolio_result.runs:
            print(f"{type(r.outcome).__name__}: verdict id={r.verdict.id}")
            # Per project, because anchoring is a per-project fact. DEFENSIVE and currently
            # unreachable from this branch — measured, and said out loud for the same reason the
            # ``budget_stop`` arm below is: no reference project sets ``bundle_dir``, so every
            # portfolio run today takes the road path and is anchored by construction. The test
            # that covers it drives a crafted ``PortfolioResult``, and says so.
            run_notice = cost_baseline_notice(r.provenance.cost_baseline_anchored)
            if run_notice is not None:
                print(run_notice)
            # One settlement per project: the mandate applies to each project in the pass, so
            # each project answers for it separately. Empty without a mandate.
            project_settlement = settle(r.coverage)
            if project_settlement:
                print(project_settlement)
        if portfolio_result.stop_reason is not None:
            sr = portfolio_result.stop_reason
            print(
                f"goal reached: scope={sr.scope} project={sr.project_id or '-'} "
                f"observed_ore={sr.observed_ore} limit_ore={sr.limit_ore} "
                f"stopped_early={portfolio_result.stopped_early}"
            )
        # A ``PortfolioResult`` has FOUR outcome channels and this branch reported one of them:
        # ``failures`` and ``budget_stop`` never reached the operator, and rc was unconditionally 0
        # — so a pass in which every project died printed nothing and exited 0 (measured: four
        # projects, four APIConnectionError, silent success). ``BudgetStop`` is kept apart from
        # ``stop_reason`` precisely so a caller can tell exhaustion from success; showing neither
        # collapsed the distinction the dataclass was split to preserve.
        # The budget-stop arm is DEFENSIVE and currently UNREACHABLE from here — measured, not
        # assumed, and said out loud for the same reason ``strict=True`` below is: ``main()`` never
        # constructs a ``PortfolioMeter``, and every write to ``budget_stop`` is gated on one, so
        # only a LIBRARY caller passing ``portfolio_meter=`` can produce this field today. It is
        # printed anyway because the field exists and a CLI door onto the global cap is a natural
        # next step; the test that covers it drives a crafted ``PortfolioResult``, and says so.
        # TRAP for whoever wires that door: ``BudgetRefused`` is a ``RuntimeError``, so the
        # ``except`` above would NOT catch the startup refusal — it needs adding explicitly.
        if portfolio_result.budget_stop is not None:
            bs = portfolio_result.budget_stop
            print(
                f"budget stop: limit_tokens={bs.limit_tokens} spent_tokens={bs.spent_tokens} "
                f"remaining_tokens={bs.remaining_tokens} required_tokens={bs.required_tokens} "
                f"stopped_early={portfolio_result.stopped_early}"
            )
        for failure in portfolio_result.failures:
            print(
                f"project failed: {failure.project_id} [{failure.error_type}] {failure.error}",
                file=sys.stderr,
            )
        # rc 1 iff something RAISED. Collect-and-continue (S3.3) exists so a partial pass does not
        # LOSE the work that completed — every finished run still printed above — not so a pass with
        # dead projects can report success to a scripted caller. A ``budget_stop`` alone stays rc 0:
        # exhaustion is a structured stop the operator asked for by setting a cap, not a crash.
        return 1 if portfolio_result.failures else 0

    if args.live_dry_run:
        # S4.2 drill (comparison protocol §4 pkt 2/3): walk the offline path, STOP before the first
        # model call, print the run-config. A misconfigured profile (e.g. AZURE with a
        # REPLACE-WITH-* placeholder) makes resolve_model raise inside the eager factory build —
        # refuse cleanly (mirror preflight.main) instead of tracebacking; S4.1 is the config gate.
        try:
            report = asyncio.run(
                run_project(
                    args.project_id,
                    args.profile,
                    docs_dir=args.docs_dir,
                    bundle_dir=args.bundle_dir,
                    verdict_dir=args.verdict_dir,
                    dimension=(
                        load_dimension(args.dimension_config) if args.dimension_config else None
                    ),
                    embedder=(
                        build_embedder(load_embedder_config(args.embedder_config))
                        if args.embedder_config
                        else None
                    ),
                    outbox_dir=args.outbox_dir,
                    run_id=args.run_id,
                    verdict_input={"decision": args.decision, "rationale": args.rationale},
                    mcp_servers=mcp_servers,
                    live_dry_run=True,
                )
            )
        except (ValueError, FileNotFoundError, ValidationError) as exc:
            # Structured refusal (rc 1, no traceback) for ANY offline-path ValueError. The
            # azure-preflight remediation is only meaningful for the AZURE config gate (S4.1), so
            # scope it to that profile — a LOCAL-profile ValueError (unknown project_id, empty
            # docs_dir, bundle mismatch) must not carry an irrelevant azure hint.
            msg = f"live-dry-run refused: {exc}"
            if args.profile == "azure":
                msg += (
                    "\nkjør 'python -m portfolio_optimiser.preflight --profile azure' først "
                    "(S4.1 offline config-gate)"
                )
            print(msg, file=sys.stderr)
            return 1
        assert isinstance(report, DryRunReport)  # live_dry_run=True always returns a DryRunReport
        print(
            f"{args.project_id}: LIVE-DRY-RUN OK (profile={report.profile}, "
            f"models={report.resolved_models}, max_rounds={report.max_rounds}, "
            f"max_tokens={report.max_tokens}, top_k={report.top_k}) — "
            "ingen modellkall gjort (stoppet før første debate.run)"
        )
        # The measured silence this closes: a bundle without ``cost-baseline.json`` used to dry-run
        # to rc 0 with nothing said about the gate's stage 0 being skipped. Printed AFTER the line
        # it qualifies, and only when there is something to say.
        notice = cost_baseline_notice(report.cost_baseline_anchored)
        if notice is not None:
            print(notice)
        # The second measured silence on this surface: a bundle with an unfollowable cross-link
        # dry-ran to rc 0 with nothing said, so a half-read base looked exactly like a small one.
        nav_notice = skipped_links_notice(report.skipped_links)
        if nav_notice is not None:
            print(nav_notice)
        return 0

    try:
        result = cast(
            RunResult,
            asyncio.run(
                run_project(
                    args.project_id,
                    args.profile,
                    docs_dir=args.docs_dir,
                    bundle_dir=args.bundle_dir,
                    verdict_dir=args.verdict_dir,
                    dimension=(
                        load_dimension(args.dimension_config) if args.dimension_config else None
                    ),
                    embedder=(
                        build_embedder(load_embedder_config(args.embedder_config))
                        if args.embedder_config
                        else None
                    ),
                    outbox_dir=args.outbox_dir,
                    run_id=args.run_id,
                    verdict_input={"decision": args.decision, "rationale": args.rationale},
                    semantic_retrieval=args.semantic_retrieval,
                    client_factory=scripted_client_factory,
                    mandate=mandate,
                    mcp_servers=mcp_servers,
                )
            ),
        )
    except (ValueError, FileNotFoundError, ValidationError) as exc:
        # Structured refusal (rc 1, no traceback) for the full-run path: run_project's fail-fast
        # loaders (contracts, load_dimension, outbox run_id guard) surface here as one clean line.
        print(f"run refused: {exc}", file=sys.stderr)
        return 1
    kind = type(result.outcome).__name__
    print(f"{args.project_id}: {kind} (verdict id={result.verdict.id}, decision={args.decision})")
    # Same notice, same renderer, read off the run's OWN stamp — so stdout and the outbox artefact
    # cannot disagree about whether the gate was anchored.
    notice = cost_baseline_notice(result.provenance.cost_baseline_anchored)
    if notice is not None:
        print(notice)
    # Same renderer on the full run, and deliberately so: a run that PRODUCED a proposal from a
    # half-read base is where the silence cost the most — the dry run at least produced nothing.
    nav_notice = skipped_links_notice(result.skipped_links)
    if nav_notice is not None:
        print(nav_notice)
    # The settlement against the commission (Trekk A4). Empty without a mandate, so an
    # un-commissioned run prints exactly what it printed before.
    settlement = settle(result.coverage)
    if settlement:
        print(settlement)
    return 0


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