portfolio-optimiser-claude/src/portfolio_optimiser_claude/run.py

"""The shippable run entrance (C2.0): merge inbox → seed → fold → run (§3, §5).

Where ``run_s10.py`` is the byte-frozen fasit of the programme's ONE live run
(never imported by the suite), this module is the generic, deliverable
entrance the README's inbox claim points at. The composition
(``compose_run_context``) is pure config/file logic and offline-testable: it
ingests the inbox READ-only (role split §3 Step 7), seeds from the bundle, and
folds the retrieved verdicts into the generation context (§3 Step 1). The
orchestration (``execute_run``) drives the loop under the §8 meter and
persists artifacts on BOTH outcomes — a budget stop is a run outcome, not an
absence of one. The model client is injected: the real SDK client is
constructed only by ``default_client_factory`` on the CLI path (wired, never
executed by the suite — honesty rule §1); the navigated docs dir comes from
the validated startup contract, never straight from the raw argument (§10).

K12 makes the whole build drivable from here: ``--bundle`` runs ONE project,
``--portfolio`` runs N from a schema-validated config (with ``--verdict-dir`` as
the portfolio-level expert inbox), ``--goals`` + ``--ledger`` bound ACHIEVEMENT
where the §8 caps bound spend, and ``--value-report`` projects what the run
delivered. A flag the chosen path cannot honour is REFUSED, never silently
ignored (§1).

Run:  uv run python -m portfolio_optimiser_claude.run --bundle <dir> [--inbox <dir>]
      # N projects, sequential, one shared meter + one shared learning store:
      uv run python -m portfolio_optimiser_claude.run --portfolio <file> \\
          [--verdict-dir <dir>]
      # K8 live-run drill (builds all, captures artifacts, STOPS before the first call):
      uv run python -m portfolio_optimiser_claude.run --bundle <dir> \\
          --outbox <dir> --run-id <id> --live-dry-run
"""

from __future__ import annotations

import argparse
import json
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Callable, Sequence

from portfolio_optimiser_claude.artifacts import (
    _dump_json,
    build_citations,
    persist_run_artifacts,
    persist_stop_artifacts,
)
from portfolio_optimiser_claude.outbox import persist_outbox
from portfolio_optimiser_claude.budget import BudgetExceeded, BudgetMeter
from portfolio_optimiser_claude.contracts import (
    Contracts,
    ReferenceProjectsContract,
    load_contracts,
    load_reference_projects,
    resolve_model,
)
from portfolio_optimiser_claude.goals import GoalContract, GoalReached
from portfolio_optimiser_claude.preflight import Refusal, run_preflight
from portfolio_optimiser_claude.experience import (
    CandidateFeatures,
    VerdictStore,
    fold_experience,
    seed_store_from_bundle,
)
from portfolio_optimiser_claude.inbox import merge_inbox_into_store
from portfolio_optimiser_claude.ir import SavingsProposal, load_validator_input
from portfolio_optimiser_claude.okf import bundle_context, navigate_bundle
from portfolio_optimiser_claude.provenance import Citation, Provenance
from portfolio_optimiser_claude.loop import ModelClient, run_project
from portfolio_optimiser_claude.notify import (
    EgressNotPermitted,
    Notification,
    Notifier,
    Transport,
    add_notify_args,
    build_notifiers,
    emit,
    notify_config_from_args,
)
from portfolio_optimiser_claude.validator import Rejection
from portfolio_optimiser_claude.valuereport import build_value_report, load_ledger, report_to_json

_PROPOSER_ROLE = "proposer"
_CHECKER_ROLE = "checker"
# The one backend profile the run resolves against (mirrors SdkModelClient's default).
_DEFAULT_PROFILE = "anthropic"
# Exit codes for the two STRUCTURED stops the entrance can reach. They are
# distinct because the criteria are: 3 bounds SPEND (§8), 4 bounds ACHIEVEMENT
# (K2). Neither is an error — both are outcomes, and both say so in the output.
_BUDGET_STOP_EXIT = 3
_GOAL_STOP_EXIT = 4

# The injected client seam of the entrance: (contracts, max_budget_usd_per_call).
ClientFactory = Callable[[Contracts, float], ModelClient]


@dataclass(frozen=True)
class ComposedRunContext:
    """The §5 sequence's output: the folded context + what fed it (§9-traceable)."""

    context: str
    citations: list[Citation]
    ir_projection: SavingsProposal
    inbox_merged: int
    seeded: int


def compose_run_context(
    bundle_dir: Path, inbox_dir: Path | None = None, *, k: int, store: VerdictStore | None = None
) -> ComposedRunContext:
    """Compose the run context per §5: merge inbox → seed → fold — read-only.

    Citations are built BEFORE anything else so an uncitable context fails
    fast ahead of any spend (§9). A missing/empty ``inbox_dir`` (or ``None``)
    leaves the composition identical to the no-inbox base. Nothing is ever
    written — the system reads the inbox, the expert writes it (§3 Step 7).

    A passed-in ``store`` is used AS-IS (its existing verdicts survive the
    merge, first-write-wins) — the §5 cross-project threading a portfolio pass
    (K3) relies on: a verdict available at project k reaches project k+1's
    fold. ``None`` (the single-run default) builds a fresh store, so every
    existing caller composes exactly as before.
    """
    citations = build_citations(navigate_bundle(bundle_dir))
    ir_projection = load_validator_input(bundle_dir)
    if store is None:
        store = VerdictStore()
    inbox_merged = merge_inbox_into_store(store, inbox_dir) if inbox_dir is not None else 0
    seeded = seed_store_from_bundle(store, bundle_dir)
    context = fold_experience(
        store,
        CandidateFeatures.from_proposal(ir_projection),
        bundle_context(bundle_dir),
        k,
    )
    return ComposedRunContext(
        context=context,
        citations=citations,
        ir_projection=ir_projection,
        inbox_merged=inbox_merged,
        seeded=seeded,
    )


def default_client_factory(contracts: Contracts, max_budget_usd_per_call: float) -> ModelClient:
    """The CLI's default: the real SDK client (run-path only, §1).

    Imported lazily so composing/executing with an injected client never
    touches the SDK module — the suite drives the same orchestration with the
    scripted stand-in.
    """
    from portfolio_optimiser_claude.sdk_client import SdkModelClient

    return SdkModelClient(contracts.model_map, max_budget_usd_per_call=max_budget_usd_per_call)


def _client_cost_usd(client: ModelClient) -> float | None:
    # Only the SDK client accounts USD; a client without the attribute
    # persists an honest null (the usage artifact allows it), never a 0.0.
    cost = getattr(client, "total_cost_usd", None)
    return None if cost is None else round(float(cost), 6)


def _client_sdk_version(client: ModelClient) -> str | None:
    # The build is read from the PRODUCING CLIENT, never from the environment:
    # a run driven by the scripted stand-in used no SDK at all, and stamping
    # the installed version there would attribute a build to a run that never
    # touched it (§1). Same seam rule as the cost and the model id above.
    version = getattr(client, "sdk_version", None)
    return None if version is None else str(version)


def execute_run(
    client: ModelClient,
    composed: ComposedRunContext,
    *,
    contracts: Contracts,
    out_dir: Path,
    max_debate_rounds: int,
    max_attempts: int,
    outbox_dir: Path | None = None,
    run_id: str | None = None,
    notifiers: Sequence[Notifier] = (),
) -> int:
    """Drive the loop under the §8 meter; persist artifacts on BOTH outcomes.

    Exit 0: the run completed (validated or typed rejection) and its artifacts
    are on disk. Exit 3: a structured budget stop (§8) — the stop event and
    the usage-vs-caps artifact are persisted; a stop is never a silent hang.

    When ``outbox_dir`` is set, the completed run also persists a ``run_id``-named
    proposal/outcome pair to the outbox (S2.1) — the system's own output layer,
    read by K8 (live capture) and K9 (pending tracking). A budget stop has no
    proposal, so it writes no outbox pair.

    ``notifiers`` (K10) receive a structured event on BOTH outcomes — a budget
    stop is a run outcome, not an absence of one, so it notifies too. The list
    is empty unless the operator configured a sink; delivery never gates the run.
    """
    run_label = run_id or out_dir.name
    meter = BudgetMeter(contracts.termination)
    try:
        result = run_project(
            client,
            composed.context,
            meter=meter,
            max_debate_rounds=max_debate_rounds,
            max_attempts=max_attempts,
            default_project_id=composed.ir_projection.project_id,
        )
    except BudgetExceeded as stop:
        print(f"STOPPED by budget: {stop.kind} observed {stop.observed} > limit {stop.limit}")
        stop_paths = persist_stop_artifacts(
            out_dir,
            stop=stop,
            termination=contracts.termination,
            tokens_used=meter.tokens_used,
            rounds_used=meter.rounds_used,
            cost_usd=_client_cost_usd(client),
        )
        for name, path in sorted(stop_paths.items()):
            print(f"artifact: {name} -> {path}")
        emit(
            notifiers,
            Notification(
                event="run.stopped",
                summary=f"run {run_label} stopped by budget: {stop.kind}",
                fields={"kind": stop.kind, "observed": stop.observed, "limit": stop.limit},
            ),
        )
        return _BUDGET_STOP_EXIT

    provenance = Provenance(
        citations=composed.citations,
        model=getattr(client, "last_model", None) or "unknown",  # §9: real id or neutral
        role=_PROPOSER_ROLE,
        validator_decision=result.validator_decision,
        tokens_used=meter.tokens_used,
        sdk_version=_client_sdk_version(client),
    )
    paths = persist_run_artifacts(
        out_dir,
        run=result,
        provenance=provenance,
        termination=contracts.termination,
        tokens_used=meter.tokens_used,
        rounds_used=meter.rounds_used,
        cost_usd=_client_cost_usd(client),
    )
    outcome_kind = "rejected" if isinstance(result.outcome, Rejection) else "validated"
    print(
        f"result: validator={result.validator_decision} checker={result.checker_decision} "
        f"attempts={result.attempts} outcome={outcome_kind}"
    )
    for name, path in sorted(paths.items()):
        print(f"artifact: {name} -> {path}")
    if outbox_dir is not None:
        outbox_paths = persist_outbox(
            outbox_dir, run=result, provenance=provenance, run_id=run_id or ""
        )
        for name, path in sorted(outbox_paths.items()):
            print(f"outbox: {name} -> {path}")
    emit(
        notifiers,
        Notification(
            event="run.completed",
            summary=f"run {run_label}: {outcome_kind}",
            fields={
                "validator_decision": result.validator_decision,
                "checker_decision": result.checker_decision,
                "attempts": result.attempts,
                "outcome": outcome_kind,
            },
        ),
    )
    return 0


def load_goal(path: Path) -> GoalContract:
    """Load + validate the goal contract fail-fast (§10) — a startup contract like the rest.

    A percent goal is D-E-gated and refuses loudly here (``NotImplementedError``
    from ``GoalContract``), before a model client exists — never silent semantics.
    """
    return GoalContract(**json.loads(path.read_text(encoding="utf-8")))


def check_goal_before_spend(
    goal: GoalContract,
    ledger_path: Path | None,
    *,
    notifiers: Sequence[Notifier] = (),
) -> bool:
    """Evaluate the goal against the book BEFORE any model call; True ⇒ stop (K12).

    The §8 caps bound SPEND; the goal bounds ACHIEVEMENT — so the honest place
    to ask "is the target already met?" is ahead of the first call, not after
    paying for one. Observed value is the ledger's dimension-free realized sum
    (K1); an absent ``--ledger`` is an EMPTY book (0 realized), so the goal is
    still evaluated, simply not reached — never a skipped check.

    A HARD goal reached returns True after printing the structured stop and
    notifying the configured sinks (a stop is an outcome, exactly as a budget
    stop is). A SOFT goal reached is a flag: it prints and the run continues.
    """
    realized = load_ledger(ledger_path).total_realized_nok()
    try:
        reached = goal.check(realized)
    except GoalReached as stop:
        print(
            f"GOAL REACHED (hard): realized {stop.observed_nok} NOK >= target "
            f"{stop.target_nok} NOK — stopping BEFORE any model call "
            "(the §8 caps bound spend; the goal bounds achievement)."
        )
        emit(
            notifiers,
            Notification(
                event="run.goal_reached",
                summary=(
                    f"hard goal reached: realized {stop.observed_nok} NOK >= "
                    f"target {stop.target_nok} NOK — run not started"
                ),
                fields={
                    "mode": goal.mode,
                    "target_nok": stop.target_nok,
                    "observed_nok": stop.observed_nok,
                },
            ),
        )
        return True
    if reached:
        print(
            f"GOAL REACHED (soft): realized {realized} NOK >= target {goal.target_nok} NOK "
            "— flagged, the run continues."
        )
    else:
        print(f"goal ({goal.mode}): realized {realized} of {goal.target_nok} NOK — not reached.")
    return False


def execute_portfolio(
    client: ModelClient,
    projects: ReferenceProjectsContract,
    *,
    contracts: Contracts,
    top_k: int,
    max_debate_rounds: int,
    max_attempts: int,
    verdict_dir: Path | None,
    notifiers: Sequence[Notifier] = (),
) -> int:
    """Drive N projects sequentially under ONE §8 meter; print one line per project (K12).

    This is the CLI reach to ``run_portfolio`` — the capability itself is K3's
    and is not extended here: the pass returns typed results and persists
    NOTHING, so nothing is filed and the summary IS the output. A budget stop
    propagates out of the shared meter and is reported as the structured stop it
    is (exit 3); because this path files no artifacts, none are claimed.

    ``run_portfolio`` is imported lazily: ``portfolio.py`` imports this module
    for ``compose_run_context``, so a module-level import would be circular.
    """
    from portfolio_optimiser_claude.portfolio import run_portfolio

    meter = BudgetMeter(contracts.termination)
    try:
        result = run_portfolio(
            projects,
            client,
            meter,
            top_k=top_k,
            max_debate_rounds=max_debate_rounds,
            max_attempts=max_attempts,
            verdict_dir=verdict_dir,
        )
    except BudgetExceeded as stop:
        print(f"STOPPED by budget: {stop.kind} observed {stop.observed} > limit {stop.limit}")
        print("portfolio: no artifacts written — this path persists nothing (§1).")
        emit(
            notifiers,
            Notification(
                event="run.stopped",
                summary=f"portfolio stopped by budget: {stop.kind}",
                fields={"kind": stop.kind, "observed": stop.observed, "limit": stop.limit},
            ),
        )
        return _BUDGET_STOP_EXIT

    for project in result.results:
        outcome_kind = "rejected" if isinstance(project.run.outcome, Rejection) else "validated"
        print(
            f"project: {project.project_id} validator={project.run.validator_decision} "
            f"checker={project.run.checker_decision} attempts={project.run.attempts} "
            f"outcome={outcome_kind}"
        )
    print(
        f"portfolio: {len(result.results)} project(s) completed — results are returned, "
        "not filed (this path persists no artifacts)."
    )
    emit(
        notifiers,
        Notification(
            event="portfolio.completed",
            summary=f"portfolio: {len(result.results)} project(s) completed",
            fields={"projects": [project.project_id for project in result.results]},
        ),
    )
    return 0


def write_value_report(
    *,
    outbox_dir: Path,
    inbox_dir: Path | None,
    ledger_path: Path | None,
    destination: Path,
    goal: GoalContract | None = None,
) -> bool:
    """K11 opt-in: project the persisted layers into a value report AFTER the run.

    The report is a projection over what is already on disk — including the pair
    this run just filed — so it runs after the run, never before, and it can
    never change what the run itself decided. A malformed layer is reported here
    and makes the COMMAND non-zero (the operator asked for a report and did not
    get one), but a budget stop stays a budget stop: reporting never rewrites a
    run's own verdict. Returns whether the report was written.

    A ``goal`` (K12's ``--goals``) is passed straight through, so the operator's
    one declared target drives BOTH the pre-spend stop and the report's goal
    progress — one contract, never two figures that can disagree.
    """
    try:
        report = build_value_report(
            outbox_dir=outbox_dir, inbox_dir=inbox_dir, ledger_path=ledger_path, goal=goal
        )
    except (OSError, TypeError, ValueError) as exc:
        print(f"VALUE REPORT FAILED — refusing to project a malformed layer (§10): {exc}")
        return False
    destination.parent.mkdir(parents=True, exist_ok=True)
    destination.write_text(report_to_json(report), encoding="utf-8", newline="\n")
    print(f"value-report: {destination}")
    return True


def build_dry_run_config(
    contracts: Contracts,
    *,
    client: ModelClient,
    profile: str,
    bundle_name: str,
    run_id: str,
    max_rounds: int,
    max_tokens: int,
    max_budget_usd_per_call: float,
    max_debate_rounds: int,
    max_attempts: int,
    top_k: int,
) -> dict[str, Any]:
    """The run-config log (comparison protocol §4 pt 3): model-id, parameters, caps.

    Records the model id each role the loop calls resolves to — THROUGH
    ``resolve_model`` (the run's own resolution path), never a raw dict read — the
    profile, and every cap/parameter a live run would carry. Deliberately carries
    NO wall-clock date: the outbox promises byte-determinism (same input + run_id
    ⇒ identical file), and the run's date is stamped at report time (§4 pt 3),
    never into the deterministic log.

    ``client`` is the client the drill CONSTRUCTED (never called): the rig's SDK
    build is read from it, the same seam rule the provenance stamp follows. A
    drill driven by the scripted stand-in used no SDK and stamps null — reading
    the installed version from the environment would describe a rig that never
    existed (§1).
    """
    return {
        "run_id": run_id,
        "profile": profile,
        "bundle": bundle_name,
        "sdk_version": _client_sdk_version(client),
        "models": {
            role: resolve_model(contracts.model_map, role, profile=profile)
            for role in (_PROPOSER_ROLE, _CHECKER_ROLE)
        },
        "caps": {
            "max_rounds": max_rounds,
            "max_tokens": max_tokens,
            "max_budget_usd_per_call": max_budget_usd_per_call,
            "max_debate_rounds": max_debate_rounds,
            "max_attempts": max_attempts,
            "top_k": top_k,
        },
    }


def execute_dry_run(
    *,
    outbox_dir: Path,
    run_id: str,
    profile: str,
    run_config: dict[str, Any],
    refusals: list[Refusal],
) -> int:
    """Capture the run-config + preflight artifacts; STOP before any model call (K8).

    Writes the run_id-named PAIR — ``{run_id}-runconfig.json`` and
    ``{run_id}-preflight.json`` — to the outbox as deterministic house JSON, then
    returns WITHOUT ever driving the loop: the drill rehearses the whole build and
    artifact capture offline, so a future operator-gated live run (the M2-analog)
    is fully rigged. Exit 0 when the preflight is clear (rig go-live-ready); exit 1
    when it refused — the artifacts are captured EITHER way (the refusal is itself
    one of them), and no model call is made in either case.
    """
    outbox_dir.mkdir(parents=True, exist_ok=True)
    paths = {
        "runconfig": outbox_dir / f"{run_id}-runconfig.json",
        "preflight": outbox_dir / f"{run_id}-preflight.json",
    }
    _dump_json(paths["runconfig"], run_config)
    _dump_json(
        paths["preflight"],
        {
            "run_id": run_id,
            "profile": profile,
            "clear": not refusals,
            "refusals": [{"check": r.check, "detail": r.detail} for r in refusals],
        },
    )
    for name, path in sorted(paths.items()):
        print(f"artifact: {name} -> {path}")
    if refusals:
        print(
            f"DRILL: preflight REFUSED ({len(refusals)}) — rig NOT clear to go live "
            "(artifacts captured, no model call was made):"
        )
        for refusal in refusals:
            print(f"  [{refusal.check}] {refusal.detail}")
        return 1
    print(
        "DRILL OK — built all, captured artifacts, stopped before the first model call "
        "(0 model calls); rig clear to go live."
    )
    return 0


# The dest names the PORTFOLIO pass honours — an ALLOWLIST, deliberately.
# Classification is required: a flag added to the parser later and named here
# nowhere is REFUSED in portfolio mode rather than accepted and then ignored.
# The blocklist this replaces failed OPEN — a forgotten flag no-opped in
# silence, which is a claim the run does not back (§1).
_PORTFOLIO_SUPPORTED_DESTS: frozenset[str] = frozenset(
    {
        "portfolio",  # the mode itself
        "verdict_dir",  # portfolio-level expert inbox, read before each fold (K3)
        "ledger",  # read by the pre-spend goal check
        "goals",  # ditto — both are consulted BEFORE the portfolio branch
        "max_rounds",  # §8 caps, shared by every project under one meter
        "max_tokens",
        "max_budget_usd_per_call",
        "max_debate_rounds",
        "max_attempts",
        "top_k",
        "notify_console",  # K10 sinks: the portfolio pass emits events too
        "notify_file",
        "notify_webhook",
        "allow_webhook_egress",
    }
)


def unsupported_flags_given(
    parser: argparse.ArgumentParser,
    args: argparse.Namespace,
    *,
    supported: frozenset[str],
) -> list[str]:
    """Flags the operator actually GAVE that ``supported`` does not name.

    "Given" is measured against the parser's own default, so this needs no
    knowledge of which flags exist — which is the point: the answer stays
    correct for flags added after it was written. Returns the flag spellings,
    sorted, for a deterministic error message.
    """
    return sorted(
        "--" + dest.replace("_", "-")
        for dest, value in vars(args).items()
        if dest not in supported and value != parser.get_default(dest)
    )


def main(
    argv: list[str] | None = None,
    *,
    client_factory: ClientFactory | None = None,
    notifier_transport: Transport | None = None,
) -> int:
    """The thin CLI: contracts fail-fast (§10) → compose (§5) → execute (§3, §8).

    ``notifier_transport`` is the injected webhook seam (K10): ``None`` uses the
    real ``default_webhook_transport`` on the CLI path; the suite injects a
    canned transport so no socket is opened.
    """
    parser = argparse.ArgumentParser(
        description=(
            "Run one project (--bundle) or a whole portfolio (--portfolio) through the "
            "loop (merge inbox → seed → fold → run), under the §8 caps and an optional "
            "savings goal."
        )
    )
    parser.add_argument(
        "--bundle", type=Path, default=None, help="OKF bundle dir of ONE project to run."
    )
    parser.add_argument(
        "--portfolio",
        type=Path,
        default=None,
        help="reference-projects config (JSON): run N projects sequentially under one "
        "shared §8 meter and one shared learning store (K3).",
    )
    parser.add_argument(
        "--verdict-dir",
        type=Path,
        default=None,
        help="portfolio-level expert inbox, READ before each project's fold (requires "
        "--portfolio; a single run uses --inbox).",
    )
    parser.add_argument("--inbox", type=Path, default=None)
    parser.add_argument(
        "--out", type=Path, default=None, help="run-artifact dir (default runs/run)."
    )
    parser.add_argument("--outbox", type=Path, default=None)
    parser.add_argument("--run-id", type=str, default=None)
    parser.add_argument("--max-rounds", type=int, default=12)
    parser.add_argument("--max-tokens", type=int, default=150_000)
    parser.add_argument("--max-budget-usd-per-call", type=float, default=0.25)
    parser.add_argument("--max-debate-rounds", type=int, default=3)
    parser.add_argument("--max-attempts", type=int, default=3)
    parser.add_argument("--top-k", type=int, default=3)
    parser.add_argument(
        "--value-report",
        type=Path,
        default=None,
        help="after the run, project the outbox (+ --inbox, + --ledger) into a "
        "deterministic value report written as JSON to this path (K11; no model call).",
    )
    parser.add_argument(
        "--ledger",
        type=Path,
        default=None,
        help="realized-savings ledger (K1) read by --goals and --value-report; "
        "absent = an empty book.",
    )
    parser.add_argument(
        "--goals",
        type=Path,
        default=None,
        help="goal contract (JSON: target_nok + hard/soft) checked against --ledger "
        "BEFORE any model call; a hard target already reached stops the run (exit 4).",
    )
    parser.add_argument(
        "--live-dry-run",
        action="store_true",
        help="build all, capture run-config + preflight to the outbox, STOP before "
        "the first model call (K8 live-run drill; no spend, no model call).",
    )
    add_notify_args(parser)
    args = parser.parse_args(argv)

    # K12: exactly ONE run shape. Refusing both the empty and the ambiguous call
    # keeps the entrance honest — neither flag can silently win over the other.
    if (args.bundle is None) == (args.portfolio is None):
        parser.error(
            "give exactly one of --bundle (one project) or --portfolio (a config of N projects)"
        )
    if args.verdict_dir is not None and args.portfolio is None:
        parser.error(
            "--verdict-dir is the PORTFOLIO-level expert inbox and requires --portfolio "
            "(a single run reads its inbox with --inbox)"
        )
    if args.portfolio is not None:
        # The portfolio pass returns typed results and persists NOTHING (K3), and
        # the outbox names its pairs by run_id, which a portfolio pass has none of.
        # Every flag that would therefore do nothing is refused rather than
        # silently ignored (§1) — a flag that quietly no-ops is a false claim.
        # ALLOWLIST (never a blocklist): the refusal covers flags nobody has
        # classified yet, so a new one fails closed instead of failing quiet.
        unsupported = unsupported_flags_given(parser, args, supported=_PORTFOLIO_SUPPORTED_DESTS)
        if unsupported:
            parser.error(
                f"--portfolio does not support {', '.join(unsupported)}: the portfolio "
                "pass persists nothing and has no run_id of its own, and it honours only "
                "the flags it is known to act on — anything else is refused rather than "
                "silently ignored. Run the projects individually with --bundle to file "
                "per-run artifacts."
            )

    # fail-fast (§10 spirit): a run persisted to the outbox MUST carry an explicit
    # run_id — reject BEFORE composing or constructing a client, so no spend rides
    # on a run that cannot be filed (no wall-clock default fills the gap). The dry
    # run's artifacts are ALSO run_id-named, so it requires both an outbox and an id.
    if args.live_dry_run:
        if args.outbox is None or not (args.run_id or "").strip():
            parser.error(
                "--live-dry-run requires --outbox and --run-id "
                "(the drill's artifacts are run_id-named in the outbox)"
            )
    elif args.outbox is not None and not (args.run_id or "").strip():
        parser.error("--outbox requires --run-id (no wall-clock default)")
    # K11: the value report PROJECTS the outbox — without one there is nothing to
    # project. Rejected here, before any spend, so a run never completes only to
    # find it cannot produce the report the operator asked for.
    if args.value_report is not None and args.outbox is None:
        parser.error("--value-report requires --outbox (the report projects the outbox pairs)")

    # K10 (§8): build the notify sinks BEFORE any spend — a --notify-webhook
    # without --allow-webhook-egress is refused fail-fast here, so no run rides
    # on a misconfigured egress. The opt-in is a run argument, never a config
    # field (the config cannot grant itself network access).
    try:
        notifiers = build_notifiers(
            notify_config_from_args(args), webhook_transport=notifier_transport
        )
    except EgressNotPermitted as exc:
        parser.error(str(exc))

    # §10: ALL startup contracts schema-validated BEFORE any model client exists —
    # the portfolio config and the goal contract included.
    projects: ReferenceProjectsContract | None = None
    if args.portfolio is not None:
        try:
            projects = load_reference_projects(
                json.loads(args.portfolio.read_text(encoding="utf-8"))
            )
        except (OSError, TypeError, ValueError) as exc:
            parser.error(f"--portfolio config is not a valid reference-projects file: {exc}")
    goal: GoalContract | None = None
    if args.goals is not None:
        try:
            goal = load_goal(args.goals)
        except (OSError, TypeError, ValueError) as exc:
            parser.error(f"--goals is not a valid goal contract: {exc}")
        except NotImplementedError as exc:  # the D-E-gated percent goal, refused loudly
            parser.error(f"--goals: {exc}")

    # The §10 contract carries ONE data source. A portfolio pass carries one per
    # project, so the contract records the first and every project still navigates
    # its OWN bundle_dir inside run_portfolio — never the contract's.
    docs_dir = str(args.bundle) if projects is None else projects.projects[0].bundle_dir
    contracts = load_contracts(
        data_source={"docs_dir": docs_dir, "top_k": args.top_k},
        termination={"max_rounds": args.max_rounds, "max_tokens": args.max_tokens},
        feedback={"decision": "approved", "rationale": "startup shape check (§10)"},
    )

    # K12: achievement is checked BEFORE spend. A hard target already met by the
    # book means the run has nothing left to buy — it stops here, structured,
    # with no client ever constructed.
    if goal is not None and check_goal_before_spend(goal, args.ledger, notifiers=notifiers):
        return _GOAL_STOP_EXIT

    factory = default_client_factory if client_factory is None else client_factory

    if projects is not None:
        print(
            f"portfolio: {len(projects.projects)} project(s) caps: "
            f"max_rounds={args.max_rounds} max_tokens={args.max_tokens} "
            f"max_budget_usd_per_call={args.max_budget_usd_per_call}"
        )
        return execute_portfolio(
            factory(contracts, args.max_budget_usd_per_call),
            projects,
            contracts=contracts,
            top_k=contracts.data_source.top_k,
            max_debate_rounds=args.max_debate_rounds,
            max_attempts=args.max_attempts,
            verdict_dir=args.verdict_dir,
            notifiers=notifiers,
        )

    assert args.bundle is not None  # narrowed by the exactly-one check above
    # The navigated dir is the CONTRACT's, so the validated config is load-bearing.
    composed = compose_run_context(
        Path(contracts.data_source.docs_dir), args.inbox, k=contracts.data_source.top_k
    )
    print(
        f"run: bundle={args.bundle.name} inbox_merged={composed.inbox_merged} "
        f"seeded={composed.seeded} caps: max_rounds={args.max_rounds} "
        f"max_tokens={args.max_tokens} "
        f"max_budget_usd_per_call={args.max_budget_usd_per_call}"
    )
    client = factory(contracts, args.max_budget_usd_per_call)

    # K8: the live-run drill builds the client (the key-free SDK construction
    # premise) but never calls it — it captures the run-config + preflight
    # artifacts and STOPS before the first model call. A future operator-gated
    # live run is thus rigged and rehearsed offline, with zero spend.
    if args.live_dry_run:
        assert args.outbox is not None  # narrowed by the fail-fast above
        run_id = args.run_id or ""
        refusals = run_preflight(
            profile=_DEFAULT_PROFILE,
            max_rounds=args.max_rounds,
            max_tokens=args.max_tokens,
            max_budget_usd_per_call=args.max_budget_usd_per_call,
        )
        run_config = build_dry_run_config(
            contracts,
            client=client,
            profile=_DEFAULT_PROFILE,
            bundle_name=args.bundle.name,
            run_id=run_id,
            max_rounds=args.max_rounds,
            max_tokens=args.max_tokens,
            max_budget_usd_per_call=args.max_budget_usd_per_call,
            max_debate_rounds=args.max_debate_rounds,
            max_attempts=args.max_attempts,
            top_k=args.top_k,
        )
        return execute_dry_run(
            outbox_dir=args.outbox,
            run_id=run_id,
            profile=_DEFAULT_PROFILE,
            run_config=run_config,
            refusals=refusals,
        )

    code = execute_run(
        client,
        composed,
        contracts=contracts,
        out_dir=args.out if args.out is not None else Path("runs") / "run",
        max_debate_rounds=args.max_debate_rounds,
        max_attempts=args.max_attempts,
        outbox_dir=args.outbox,
        run_id=args.run_id,
        notifiers=notifiers,
    )
    # K11: the value report is produced on BOTH outcomes — a budget-stopped run
    # still has a value picture worth reporting (the accumulated outbox, inbox
    # and ledger are what it projects, not this one run's success).
    if args.value_report is not None:
        assert args.outbox is not None  # narrowed by the fail-fast above
        written = write_value_report(
            outbox_dir=args.outbox,
            inbox_dir=args.inbox,
            ledger_path=args.ledger,
            destination=args.value_report,
            goal=goal,
        )
        if not written and code == 0:
            code = 1
    return code


if __name__ == "__main__":
    raise SystemExit(main())