Launch Plan & Preflight

launch_plan

Resolve a config into a launch plan, and stage it — everything before anything launches.

The offline half of a run. plan_run answers "what would this config schedule, and has it run already" without touching a thing; stage_run goes one step further and uploads the artifacts a remote launch needs (the config, and for Spark the code zip), then writes a reproducibility manifest — and still submits nothing. Neither ever starts a job; that is main.run and job_launch.

Two things make the plan trustworthy rather than merely informative:

• Same id as the real run. run_id is a pure digest of the whole config, so a plan resolves the id a live run would use. That is what lets the CLI's --dry-run report an existence verdict (Idempotency) — this exact config has already run, and here is its status — and what makes stage_run's uploads land at the paths the launch will read.
• The source is pinned, but not into the digest. lock_profile_source rewrites profile.source: "auto" to the concrete run it resolves to, so a staged config reproduces its original fleet instead of re-searching. registry.ids then excludes that field from the digest: it is resolved, not authored, and a config whose identity depended on what had been run before it never converged. Two runs a week apart share a run_id and dedupe on read; their manifests and sizing provenance record the different evidence each was sized from.

LaunchPlan is the one return type for both, distinguished by staged: plan_run returns URI templates (where artifacts would go) and stage_run returns the real ones (where they now are), along with commands that are actually runnable. The command strings are built by commands from the resolved deployment envelope, so what is printed is what a launch would execute — the two-tier emit (a portable python -m line, plus the platform-native gcloud form).

Both take an optional Settings; without a reachable SF_* environment plan_run degrades rather than fails (a plan with an unknown verdict and no commands), because resolving a config is useful offline. stage_run requires it — staging touches GCS.

feasibility_report is the deliberate exception to "the plan is offline": it reads one aggregation off the source panel, because how many backtest folds each series actually achieves is a fact about the data and no amount of config inspection will produce it. It is opt-in (--feasibility), reads only, and degrades to a line of text when there is no environment.

Public surface: plan_run, stage_run, lock_profile_source, feasibility_report, and the LaunchPlan / Idempotency result types.

Idempotency dataclass

Idempotency(checked, exists, prior_status)

Whether a run's config has already been submitted (the pre-submit existence check).

run_id is a pure digest of the config, so an unchanged config re-derives the same id — this verdict says whether that id already has a header in the registry. checked is False when the registry couldn't be consulted (no SF_* env, or no reachable registry): the verdict is simply unknown, not "new". prior_status is the existing run's status when exists (e.g. COMPLETED), else None.

LaunchPlan dataclass

LaunchPlan(
    run_id,
    python_runtime,
    python_models,
    bq_models,
    fanout,
    staged,
    config_uri,
    commands,
    idempotency,
    force,
    nodes,
    variant_plan=(),
)

A run resolved to launch-ready form: id, runtime split, fanout, and the launch commands.

Produced offline by plan_run (staged=False — URIs are the templates where artifacts will land; no GCS is touched) or by stage_run (staged=True — artifacts uploaded, URIs real and the commands runnable). commands maps a tier name to its LaunchCommands: "main", "ray", and one "spark:<family>" per Serverless family ("spark:statistical", …) — per-family because each family resolves its own hardware and batch id, so one command cannot stand for all of them. It is None only when the GCP infra identity can't be resolved offline (no SF_* env), in which case the run_id + fanout + runtime split are still returned.

idempotency is the exists-vs-new verdict for this config's id; force records that the caller intends to re-run an already-run config (it only shapes the emitted guidance — a re-run is idempotent regardless, via dedupe-on-read under the shared run_id).

nodes is the run's execution DAG resolved offline (dag.dag_nodes): one node per family job plus the ensemble, each with its deterministic job_key and upstream dependencies — the plan's cross-system identity map, before anything runs.

feasibility_lines

feasibility_lines(cfg, obs_counts)

The feasibility report for a measured panel, as lines to print (pure).

What plan --feasibility prints once it has the counts. Three things, in the order an operator needs them: how much work this is (cells, fits, and the honest backtest multiplier), who actually gets scored (the achieved-fold cohorts, which on a ragged panel are not one number), and what to change if the answer is unwelcome (backtest.suggest_min_train).

Advisory only — it reads nothing but the counts and writes nothing at all. min_train is a config field, so acting on the suggestion moves the run_id, which is the caller's decision to make deliberately rather than something a planning verb should do for them.

Source code in src/scale_forecasting/launch_plan.py

def feasibility_lines(cfg: RunConfig, obs_counts: Sequence[int]) -> list[str]:
    """The feasibility report for a measured panel, as lines to print (pure).

    What ``plan --feasibility`` prints once it has the counts. Three things, in the order an
    operator needs them: how much work this is (cells, fits, and the honest backtest multiplier),
    who actually gets scored (the achieved-fold cohorts, which on a ragged panel are not one
    number), and what to change if the answer is unwelcome (`backtest.suggest_min_train`).

    Advisory only — it reads nothing but the counts and writes nothing at all. ``min_train`` is a
    config field, so acting on the suggestion moves the run_id, which is the caller's decision to
    make deliberately rather than something a planning verb should do for them.
    """
    from .backtest import suggest_min_train
    from .config import estimate_workload

    w = estimate_workload(cfg, obs_counts=obs_counts)
    lines = [
        f"feasibility: {w.n_series} series x {w.n_models} models = {w.n_cells} cells",
        f"  fits: {w.n_fits} ({w.train_rows_total:,} training rows) = "
        f"{w.full_fit_equivalents:.3f} whole-history fits per cell",
    ]
    if not cfg.backtest.enabled:
        lines.append("  backtest: off — one full-history fit per cell, no folds to achieve")
        return lines
    for k, n in sorted(w.fold_histogram.items(), reverse=True):
        share = n / (w.n_series or 1)
        verdict = "all folds" if k == cfg.backtest.n_folds else "UNSCORED" if k == 0 else "reduced"
        lines.append(f"  {k} of {cfg.backtest.n_folds} folds: {n} series ({share:.1%}) — {verdict}")
    if w.n_unscored:
        lines.append(
            f"  {w.n_unscored} series get no folds at all: they are still fit and forecast, but "
            "they contribute nothing to any leaderboard (see v_backtest_coverage)"
        )
    suggested = suggest_min_train(obs_counts, cfg)
    current = cfg.backtest.min_train
    if suggested is None:
        lines.append(
            f"  no min_train reaches 90% of series at all {cfg.backtest.n_folds} folds — the panel "
            f"is short for this geometry; reduce n_folds ({cfg.backtest.n_folds}), step "
            f"({cfg.backtest.step}) or horizon ({cfg.backtest.horizon}) instead"
        )
    elif suggested < current:
        lines.append(
            f"  min_train={current} is above what this panel supports; min_train<={suggested} "
            "brings 90% of series to full folds (changing it moves the run_id)"
        )
    else:
        lines.append(
            f"  min_train={current} is within budget: up to {suggested} still keeps 90% of series "
            "at full folds"
        )
    return lines

feasibility_report

feasibility_report(cfg, *, settings=None)

Read the panel's series lengths and report what the run's fold geometry does to it.

The --feasibility half of the plan verb: the one part of planning that cannot be honest offline, because how many folds a series achieves is a fact about the data, not the config. Best-effort like every other reporting verb — an unreachable environment produces a line saying so rather than an exception, so --dry-run --feasibility still returns a plan with no SF_* env.

Source code in src/scale_forecasting/launch_plan.py

def feasibility_report(cfg: RunConfig, *, settings: Settings | None = None) -> list[str]:
    """Read the panel's series lengths and report what the run's fold geometry does to it.

    The ``--feasibility`` half of the plan verb: the one part of planning that cannot be honest
    offline, because how many folds a series achieves is a fact about the data, not the config.
    Best-effort like every other reporting verb — an unreachable environment produces a line saying
    so rather than an exception, so ``--dry-run --feasibility`` still returns a plan with no
    ``SF_*`` env.
    """
    try:
        obs_counts = read_series_lengths(cfg, settings=settings)
    except Exception as exc:  # noqa: BLE001 - a report that can fail is one nobody runs
        return [f"feasibility: could not read the source panel ({exc}); reporting counts only"]
    if not obs_counts:
        return ["feasibility: the source panel returned no series"]
    return feasibility_lines(cfg, obs_counts)

lock_profile_source

lock_profile_source(cfg, *, settings=None)

Replace compute.profile.source = "auto" with the concrete reference it resolves to.

The lockfile step. Pinning is what makes auto convenient and reproducible: re-running a staged config reproduces the original fleet exactly, because what actually resolved is written into it rather than re-searched.

It deliberately does not move the run_id. registry.ids excludes this field from the digest, because it is resolved rather than authored — see _canonical_config, which explains what went wrong when it was included. Two auto runs a week apart therefore share an id and dedupe on read, while their staged manifests and sizing provenance record the different evidence each was sized from. That is the distinction: a run's identity is what was asked for; its provenance is what answered.

What gets pinned is the pointer, never the numbers. The rows behind a run_id are immutable, so the pointer is as good as the values and keeps the config small enough to stay readable. When discovery finds nothing, the pin is "baseline" rather than "auto": the remainder of the chain (shipped baseline, then static) is deterministic, so pinning it makes the run reproducible instead of leaving it to re-search later.

Degrades to the unlocked config when the registry cannot be reached — a plan produced with no SF_* environment is a preview, and it says so elsewhere too (commands=None, verdict unknown). The run still resolves its source at sizing time; it just is not pinned.

Source code in src/scale_forecasting/launch_plan.py

def lock_profile_source(cfg: RunConfig, *, settings: Settings | None = None) -> RunConfig:
    """Replace ``compute.profile.source = "auto"`` with the concrete reference it resolves to.

    The lockfile step. Pinning is what makes ``auto`` convenient *and* reproducible: re-running a
    staged config reproduces the original fleet exactly, because what actually resolved is written
    into it rather than re-searched.

    It deliberately does **not** move the ``run_id``. ``registry.ids`` excludes this field from the
    digest, because it is resolved rather than authored — see ``_canonical_config``, which explains
    what went wrong when it was included. Two ``auto`` runs a week apart therefore share an id and
    dedupe on read, while their staged manifests and sizing provenance record the different
    evidence each was sized from. That is the distinction: a run's *identity* is what was asked
    for; its *provenance* is what answered.

    What gets pinned is the **pointer**, never the numbers. The rows behind a ``run_id`` are
    immutable, so the pointer is as good as the values and keeps the config small enough to stay
    readable. When discovery finds nothing, the pin is ``"baseline"`` rather than ``"auto"``: the
    remainder of the chain (shipped baseline, then static) is deterministic, so pinning it makes
    the run reproducible instead of leaving it to re-search later.

    Degrades to the unlocked config when the registry cannot be reached — a plan produced with no
    ``SF_*`` environment is a preview, and it says so elsewhere too (``commands=None``, verdict
    unknown). The run still resolves its source at sizing time; it just is not pinned.
    """
    if not cfg.compute.profile.needs_source_resolution:
        return cfg

    from .profiling.signature import signature_from_config
    from .registry.harvest import discover_harvest_run

    want = signature_from_config(cfg)
    try:
        found = discover_harvest_run(
            source_table=want.source_table,
            freq=want.freq,
            target_series=want.n_series,
            target_runtime=cfg.python_runtime,
            settings=settings,
        )
    except Exception as exc:  # noqa: BLE001 - an unpinned plan is a worse plan, not a failed one
        _log.debug("profile source not pinned (registry unreachable): %r", exc)
        return cfg

    locked = found or "baseline"
    _log.info("compute profile source: auto -> %s", locked)
    profile = cfg.compute.profile.model_copy(update={"source": locked})
    return cfg.model_copy(update={"compute": cfg.compute.model_copy(update={"profile": profile})})

plan_run

plan_run(cfg, *, settings=None, infra=None, force=False)

Resolve a run offline to its id, fanout, runtime split, and launch-command templates.

The "plan" verb: pure and GCP-free. Computes _plan and estimate_fanout, then — best-effort — resolves the infra identity, consults the registry for the exists-vs-new verdict (_check_idempotency), and builds the two-tier launch commands with the URIs the artifacts will land at (nothing is uploaded). When the infra can't be resolved offline (no SF_* env), the commands are omitted (commands=None) and the verdict is unknown, but the id/fanout/split are still returned, so a plain --dry-run works with no environment. force shapes only the emitted re-run guidance. Emits the plan to the log and returns it.

Source code in src/scale_forecasting/launch_plan.py

def plan_run(
    cfg: RunConfig,
    *,
    settings: Settings | None = None,
    infra: object | None = None,
    force: bool = False,
) -> LaunchPlan:
    """Resolve a run offline to its id, fanout, runtime split, and launch-command *templates*.

    The "plan" verb: pure and GCP-free. Computes `_plan` and `estimate_fanout`, then — best-effort —
    resolves the infra identity, consults the registry for the exists-vs-new verdict
    (`_check_idempotency`), and
    builds the two-tier launch commands with the URIs the artifacts *will* land at (nothing is
    uploaded). When the infra can't be resolved offline (no ``SF_*`` env), the commands are omitted
    (``commands=None``) and the verdict is *unknown*, but the id/fanout/split are still returned, so
    a plain ``--dry-run`` works with no environment. ``force`` shapes only the emitted re-run
    guidance. Emits the plan to the log and returns it.
    """
    from .config import estimate_fanout
    from .dag import dag_nodes, plan_dag
    from .profiling.source import check_pinned_source

    # Before the lock, so this only ever judges what a *person* pinned (see `check_pinned_source`),
    # and outside the try, because a rotted pin is the one thing here that is *not* best-effort.
    # It degrades on its own when the registry is unreachable — no drift is visible, so none is
    # reported.
    check_pinned_source(cfg, settings=settings, force=force)

    # The infra identity is resolved first and separately, because `lock_profile_source` needs it
    # and its result is part of the digest — so it has to land before `_plan` computes the run_id.
    # Best-effort, exactly like everything else in this try: no environment means an unpinned
    # preview, not a failure.
    with suppress(Exception):
        settings = settings or resolve_settings()
        cfg = lock_profile_source(cfg, settings=settings)

    plan = _plan(cfg)
    fanout = estimate_fanout(cfg)
    nodes = dag_nodes(plan_dag(cfg))
    config_uri: str | None = None
    commands: dict[str, LaunchCommands] | None = None
    idempotency = Idempotency(checked=False, exists=False, prior_status=None)
    try:
        settings = settings or resolve_settings()
        # The verdict is a registry read. A core-only install (no ``gcp`` extra) gets the same
        # answer a missing environment does — the plan without the verdict — not a stack trace.
        require_extra("gcp", purpose="The plan's exists-vs-new registry lookup")
        idempotency = _check_idempotency(plan.run_id, settings)
        # Only now that a registry is reachable — see `_nodes_with_submit_attempts`. Rebinding
        # `nodes` rather than using a second name is deliberate: the LaunchPlan below carries these
        # too, so what a reader sees in the plan is what a paste of the command would submit.
        nodes = _nodes_with_submit_attempts(nodes, plan.run_id, force=force, settings=settings)
        resolved_infra = _resolve_infra(cfg, infra)
        code_bucket = resolved_infra.code_bucket  # type: ignore[attr-defined]
        config_uri, package_uri, launcher_uri = _template_uris(cfg, plan, code_bucket)
        commands = _assemble_commands(
            cfg,
            plan,
            settings,
            resolved_infra,
            nodes,
            config_uri=config_uri,
            package_uri=package_uri,
            launcher_uri=launcher_uri,
        )
    except ConfigError:
        # No SF_* env (or no injected settings/infra): return the plan without commands.
        config_uri = None
        commands = None
    except MissingExtraError as exc:
        _log.info("plan: registry verdict and launch commands skipped — %s", exc)
        config_uri = None
        commands = None
    from .features import variant_diagnostics_for

    variant_plan = tuple({"model_type": m, **variant_diagnostics_for(m, cfg)} for m in cfg.models)
    result = LaunchPlan(
        run_id=plan.run_id,
        python_runtime=cfg.python_runtime,
        python_models=plan.python_models,
        bq_models=plan.bq_models,
        fanout=fanout,
        staged=False,
        config_uri=config_uri,
        commands=commands,
        idempotency=idempotency,
        force=force,
        nodes=nodes,
        variant_plan=variant_plan,
    )
    _emit_plan(result)
    return result

preflight_short_series

preflight_short_series(cfg, *, settings=None)

Enforce backtest.short_series="error" at submit time, where the operator is watching.

main.run calls this once, before it launches anything. It is the useful place to refuse a run under that policy: the message reaches the terminal the operator typed into, no cluster has been provisioned, and no registry rows exist to clean up. Each Python engine's driver holds a second check (spark_io.assert_source_supports_folds and the Ray twin) for the paths that do not come through here — a staged config launched from an emitted command, a Composer task, or an engine driven directly.

Costs one aggregation, and only when the policy is armed; every other value of short_series returns before touching BigQuery, because those policies have already decided what a short series gets.

A failed read warns rather than refuses. short_series="error" means "stop if the panel is too short", not "stop if I could not tell" — turning a transient BigQuery hiccup into a refused run would be a worse failure than the one this guards against. The driver backstop still runs, on a panel the engine has by then definitely read, so nothing is skipped by letting this one go.

Source code in src/scale_forecasting/launch_plan.py

def preflight_short_series(cfg: RunConfig, *, settings: Settings | None = None) -> None:
    """Enforce ``backtest.short_series="error"`` at submit time, where the operator is watching.

    `main.run` calls this once, before it launches anything. It is the *useful* place to refuse a
    run under that policy: the message reaches the terminal the operator typed into, no cluster has
    been provisioned, and no registry rows exist to clean up. Each Python engine's driver holds a
    second check (`spark_io.assert_source_supports_folds` and the Ray twin) for the paths that do
    not come through here — a staged config launched from an emitted command, a Composer task, or
    an engine driven directly.

    Costs one aggregation, and only when the policy is armed; every other value of
    ``short_series`` returns before touching BigQuery, because those policies have already decided
    what a short series gets.

    **A failed read warns rather than refuses.** ``short_series="error"`` means "stop if the panel
    is too short", not "stop if I could not tell" — turning a transient BigQuery hiccup into a
    refused run would be a worse failure than the one this guards against. The driver backstop
    still runs, on a panel the engine has by then definitely read, so nothing is skipped by
    letting this one go.
    """
    bt = cfg.backtest
    if not (bt.enabled and bt.short_series == "error"):
        return
    from .backtest import assert_panel_supports_folds

    try:
        obs_counts = read_series_lengths(cfg, settings=settings)
    except Exception as exc:  # noqa: BLE001 - see the docstring: unreadable is not too short
        _log.warning(
            "short_series='error': could not read the source panel (%s); deferring the check to "
            "the engine driver",
            exc,
        )
        return
    if obs_counts:
        assert_panel_supports_folds(obs_counts, cfg)

read_series_lengths

read_series_lengths(cfg, *, settings=None)

Observation count of every series the run will forecast — one BigQuery aggregation.

SELECT ts_id, COUNT(*) … GROUP BY ts_id, honouring data.series_limit the same way the engines do. One scan of two columns with a single shuffle: cheap enough to run before a launch, which is the whole point of feasibility_report.

The counts are raw rows, and raw rows are the exact input a fit sees — features.build_features creates lag columns but does no dropna, so nothing shrinks the panel between here and the model. No lag-warmup correction is needed and applying one would understate every series.

Source code in src/scale_forecasting/launch_plan.py

def read_series_lengths(
    cfg: RunConfig, *, settings: Settings | None = None
) -> list[int]:  # pragma: no cover - GCP I/O, covered by the @gcp smokes
    """Observation count of every series the run will forecast — one BigQuery aggregation.

    ``SELECT ts_id, COUNT(*) … GROUP BY ts_id``, honouring ``data.series_limit`` the same way the
    engines do. One scan of two columns with a single shuffle: cheap enough to run before a launch,
    which is the whole point of `feasibility_report`.

    The counts are raw rows, and raw rows are the exact input a fit sees — `features.build_features`
    creates lag columns but does no ``dropna``, so nothing shrinks the panel between here and the
    model. No lag-warmup correction is needed and applying one would understate every series.
    """

    from .engines.bigquery_names import source_ref
    from .settings import Settings as _Settings

    settings = settings or _Settings.resolve()
    table = source_ref(cfg, settings.dataset_ref)
    limit = "" if cfg.data.series_limit is None else f"\nLIMIT {int(cfg.data.series_limit)}"
    sql = (
        f"SELECT `{cfg.data.ts_id_col}` AS ts_id, COUNT(*) AS n_obs\n"
        f"FROM `{table}`\n"
        f"GROUP BY ts_id\n"
        f"ORDER BY ts_id{limit}"
    )
    rows = bigquery_client(settings.project_id).query(sql).result()
    return [int(r["n_obs"]) for r in rows]

resolve_settings

resolve_settings()

Resolve Settings from the SF_* env (raises ConfigError when unset).

Source code in src/scale_forecasting/launch_plan.py

def resolve_settings() -> Settings:
    """Resolve `Settings` from the ``SF_*`` env (raises `ConfigError` when unset)."""
    from .settings import Settings

    return Settings.resolve()

stage_run

stage_run(cfg, *, settings=None, infra=None, force=False)

Stage a run's artifacts to GCS and return the runnable launch commands — no submit.

The "stage" verb: uploads the config (and, for Spark, the code zip + launcher shim) to the code bucket, builds the launch commands against those real URIs (so they run as-is from any ADC box), and writes the reproducibility manifest runs/<run_id>.plan.json next to the config. Unlike plan_run, the infra identity is required — staging touches GCS — so a missing SF_* env raises rather than degrading; the exists-vs-new verdict (_check_idempotency) is therefore always resolved. force shapes only the emitted re-run guidance. Returns the LaunchPlan with staged=True.

Source code in src/scale_forecasting/launch_plan.py

def stage_run(
    cfg: RunConfig,
    *,
    settings: Settings | None = None,
    infra: object | None = None,
    force: bool = False,
) -> LaunchPlan:
    """Stage a run's artifacts to GCS and return the *runnable* launch commands — no submit.

    The "stage" verb: uploads the config (and, for Spark, the code zip + launcher shim) to the code
    bucket, builds the launch commands against those **real** URIs (so they run as-is from any ADC
    box), and writes the reproducibility manifest ``runs/<run_id>.plan.json`` next to the config.
    Unlike `plan_run`, the infra identity is required — staging touches GCS — so a missing ``SF_*``
    env raises rather than degrading; the exists-vs-new verdict (`_check_idempotency`) is therefore
    always resolved. ``force`` shapes only the emitted re-run guidance. Returns the `LaunchPlan`
    with ``staged=True``.
    """
    from datetime import UTC, datetime

    from .config import estimate_fanout
    from .dag import dag_nodes, plan_dag, preflight
    from .profiling.source import check_pinned_source
    from .staging import stage_config, stage_manifest

    # Refuse an unhonourable config, or an incoherent hardware plan, before anything is staged.
    preflight(cfg)
    # Pin `source: "auto"` to what it resolves to *now*, before the digest — the staged config is
    # the reproducibility artifact, so what actually sized this run has to be written into it.
    settings = settings or resolve_settings()
    # Before the lock, so this only ever judges what a *person* pinned (see `check_pinned_source`).
    check_pinned_source(cfg, settings=settings, force=force)
    cfg = lock_profile_source(cfg, settings=settings)

    plan = _plan(cfg)
    fanout = estimate_fanout(cfg)
    nodes = dag_nodes(plan_dag(cfg))
    idempotency = _check_idempotency(plan.run_id, settings)
    # Staging is the path whose commands a human actually pastes, so the attempt has to be the one
    # a submit would use rather than the planner's attempt 1 — see `_nodes_with_submit_attempts`.
    nodes = _nodes_with_submit_attempts(nodes, plan.run_id, force=force, settings=settings)
    resolved_infra = _resolve_infra(cfg, infra)
    code_bucket: str = resolved_infra.code_bucket  # type: ignore[attr-defined]

    config_uri = stage_config(cfg, plan.run_id, code_bucket)
    package_uri: str | None = None
    launcher_uri: str | None = None
    if _needs_batch_infra(cfg) and plan.python_models:
        from .batch_infra import BatchInfra
        from .staging import stage_code

        assert isinstance(resolved_infra, BatchInfra)  # spark / vertex runtime → BatchInfra
        package_uri, launcher_uri = stage_code(resolved_infra.code_bucket)

    commands = _assemble_commands(
        cfg,
        plan,
        settings,
        resolved_infra,
        nodes,
        config_uri=config_uri,
        package_uri=package_uri,
        launcher_uri=launcher_uri,
    )
    from .features import variant_diagnostics_for

    variant_plan = tuple({"model_type": m, **variant_diagnostics_for(m, cfg)} for m in cfg.models)
    result = LaunchPlan(
        run_id=plan.run_id,
        python_runtime=cfg.python_runtime,
        python_models=plan.python_models,
        bq_models=plan.bq_models,
        fanout=fanout,
        staged=True,
        config_uri=config_uri,
        commands=commands,
        idempotency=idempotency,
        force=force,
        nodes=nodes,
        variant_plan=variant_plan,
    )
    manifest_uri = stage_manifest(
        _manifest_dict(result, created_at=datetime.now(UTC).isoformat()), plan.run_id, code_bucket
    )
    _log.info("wrote run manifest: %s", manifest_uri)
    # Last, once the commands are real and the artifacts are up: record the attempt numbers those
    # commands just spent, so the counter can see a launch this process will never make, and give
    # the run a header if it has none so the repair verbs can reach those rows. `plan_run`
    # deliberately does not do this — a dry preview hands out nothing, and filing rows for it would
    # burn an attempt every time somebody looked at a plan.
    _file_emitted_rows(cfg, nodes, plan.run_id, settings, idempotency=idempotency)
    _emit_plan(result)
    return result