Runtime Probes & Reconciliation

vocabulary

The closed vocabularies every probe layer speaks — states, verdicts, handles, results.

The lowest layer of this package, and it exists because all three layers above it must agree on exactly these words. runtimes produces them, reconcile fuses them into verdicts, and cancel acts on those verdicts; putting any of this in one of those modules would make the other two import it and cycle.

Two vocabularies live here and they are deliberately distinct even where they share a spelling. The NATIVE_* set is what a runtime says about a job right now. The registry literals (_REGISTRY_RUNNING, _CANCELLED, _TERMINAL) are what a run_jobs row last wrote. Both spell "RUNNING", and conflating them is precisely the bug this package exists to catch: a stale registry row that still says RUNNING for a job whose runtime is long gone.

ProbeHandle is the coordinate blob that makes reconciliation possible at all — built at launch, serialized under run_jobs.job_telemetry.$.probe_handle, parsed back out by a reader. It stays a pure dataclass with dict (de)serialization only, no GCP imports, so importing this module never pulls a cloud extra.

CancelResult dataclass

CancelResult(stopped, already_gone, detail='')

One runtime's answer to a cancel request for a job — the write-boundary evidence.

stopped is True when the runtime accepted the stop (the job was live and is now being torn down). already_gone is True when there was nothing to stop — the batch/job/cluster was already NotFound (a Ray cluster GC'd on completion is the common case), which the caller treats as an effective cancel (the job is not running). Either flag True means the registry row may be finalized to CANCELLED; when both are False the stop genuinely failed (permission denied, timeout) and the caller leaves the registry as-is and reports honestly. detail is a short human reason (a permission hint, a NotFound note, or a runtimes._short_detail of the swallowed exception). Like check, cancel is advisory: it never raises.

ProbeHandle dataclass

ProbeHandle(
    runtime,
    native_id,
    region,
    id_kind="exact",
    spark_mode=None,
    resource_name=None,
    created_at=None,
)

The runtime coordinates for one family's job — the write-blob and read-parse shared type.

native_id is the platform-native id (a Serverless batch_id, a Ray submission_id, a Dataproc cluster job's server-assigned id, or a BigQuery job-id prefix); it is "" when the real id isn't known yet (a cluster job pre-stamp-back). region is where the job runs. id_kind is "prefix" for BigQuery (whose jobs share a deterministic id prefix) and "exact" otherwise; it is carried on the persisted blob so a reader (and the P5 cancel path) can tell a prefix match from an exact id without re-deriving it per runtime. spark_mode ("serverless" | "cluster") is set for Spark only; resource_name (the Ray persistent- resource path) for Ray only. created_at is not a persisted coordinate — it is hydrated from the job row at read time (from_job_row) purely to lower-bound the BigQuery history scan; it is excluded from to_blob and is None for non-BQ handles and pre-feature rows.

from_job_row classmethod

from_job_row(row)

Parse the handle out of a v_run_jobs row, or None when there isn't a usable one.

row["probe_handle"] is the JSON_QUERY string the view projects (or an already-parsed dict); None (a pre-feature run) or a malformed/incomplete blob returns None so a reader degrades to registry-only rather than raising.

Source code in src/scale_forecasting/probes/vocabulary.py

@classmethod
def from_job_row(cls, row: dict[str, Any]) -> ProbeHandle | None:
    """Parse the handle out of a ``v_run_jobs`` row, or ``None`` when there isn't a usable one.

    ``row["probe_handle"]`` is the ``JSON_QUERY`` string the view projects (or an already-parsed
    dict); ``None`` (a pre-feature run) or a malformed/incomplete blob returns ``None`` so a
    reader degrades to registry-only rather than raising.
    """
    raw = row.get("probe_handle")
    if not raw:
        return None
    blob = json.loads(raw) if isinstance(raw, str) else raw
    try:
        return cls(
            runtime=blob["runtime"],
            native_id=blob["native_id"],
            region=blob["region"],
            id_kind=blob.get("id_kind", "exact"),
            spark_mode=blob.get("spark_mode"),
            resource_name=blob.get("resource_name"),
            # Hydrated from the row (not the blob) to bound the BigQuery job-history scan.
            created_at=row.get("started_at") or row.get("created_at"),
        )
    except (KeyError, TypeError):
        return None

to_blob

to_blob()

The compact dict stored under run_jobs.job_telemetry.$.probe_handle.

Omits the runtime-specific fields (spark_mode/resource_name) when unset so a handle carries only the coordinates its runtime actually has.

Source code in src/scale_forecasting/probes/vocabulary.py

def to_blob(self) -> dict[str, Any]:
    """The compact dict stored under ``run_jobs.job_telemetry.$.probe_handle``.

    Omits the runtime-specific fields (``spark_mode``/``resource_name``) when unset so a handle
    carries only the coordinates its runtime actually has.
    """
    blob: dict[str, Any] = {
        "runtime": self.runtime,
        "native_id": self.native_id,
        "region": self.region,
        "id_kind": self.id_kind,
    }
    if self.spark_mode is not None:
        blob["spark_mode"] = self.spark_mode
    if self.resource_name is not None:
        blob["resource_name"] = self.resource_name
    return blob

ProbeResult dataclass

ProbeResult(
    native_state, exists, detail="", telemetry=dict()
)

One runtime's live answer for a job: the normalized state + whether it still exists.

native_state is one of the closed NATIVE_* set. exists is False only for NATIVE_NOT_FOUND (the job / cluster is gone) — a reader uses it to tell "torn down" apart from "failed". detail is a short human reason (a status message, or — when a probe degrades — a one-line Type: message summary of the swallowed exception, via runtimes._short_detail). telemetry carries any extra fields the probe cheaply gathered (Dataproc batch usage, statement counts); it is always a plain JSON-able dict, empty when the probe has nothing to add.

RuntimeProbe

Bases: Protocol

Read (and, on request, stop) one family's runtime job from its ProbeHandle.

name is the runtime key. check maps the platform-native job state into the closed NATIVE_* set and never raises: a probe is advisory, so any error (auth, timeout, a torn-down cluster the native call chokes on) degrades to ProbeResult(NATIVE_UNKNOWN, exists=True, ...) rather than propagating. cancel issues the runtime's stop and returns a CancelResult; it likewise never raises (a failed stop is reported, not thrown).

reconcile

Is this job actually still running? — fuse registry, artifacts and live runtime into verdicts.

The question a reader asks when a progress bar stops moving, and the answer is not any one source's alone. A run_jobs row records a last-written status that a crashed driver never got to update; the landed-cell counts say what work survived; the runtime says whether the job still exists. One FamilyVerdict per family is what you get from fusing all three.

Split pure from I/O, as the rest of this codebase is: _verdict_for_family, _assemble_probe_report and _narrow_to_job are pure and unit-tested per matrix row, and read_and_probe is the one function that reads BigQuery and calls the probes before handing the assembled inputs over.

Escalation is the cost control. A terminal family is never probed — the registry is authoritative once the work is done — so a routine poll of a finished run makes zero native calls. Only non-terminal jobs are escalated, and _is_stale decides whether a vanished one is genuinely LOST or merely still-starting. That startup grace is why a normal launch window does not cry wolf.

FamilyVerdict dataclass

FamilyVerdict(
    family,
    runtime,
    registry_status,
    native_state,
    exists,
    verdict,
    disagreement,
    n_done,
    n_expected,
    detail,
    telemetry=dict(),
)

The reconciled truth for one family: what the registry says vs. what the runtime says.

registry_status is the family's last-written job status; native_state / exists are the live vocabulary.ProbeResult reading (None for a family that wasn't escalated — terminal or never launched). verdict is one of the VERDICT_* set and disagreement is True only when the runtime contradicts the registry (a reader's "look here" flag). n_done / n_expected are the family's landed-vs-expected cell counts (the artifact evidence that splits a vanished job into LIKELY_COMPLETED vs. LOST). detail is a short human reason; telemetry carries anything the probe cheaply gathered.

ProbeReport dataclass

ProbeReport(
    run_id, status, escalated, families, disagreement
)

A run's reconciled snapshot: header status + one FamilyVerdict per family.

escalated is True when at least one family was probed against its runtime (False for a routine poll of an all-terminal run — the terminal short-circuit that keeps polling native- call-free). disagreement is the run-wide roll-up: True when any family's runtime contradicts its registry status.

probe_run

probe_run(
    run_id,
    *,
    job=None,
    settings=None,
    stale_after_s=None,
    abandoned_after_s=None,
)

Reconcile a run's registry state against live runtime state → a ProbeReport.

Reads the run's header + config + job rows + landed-cell counts (registry.reads, registry.jobs), assembles the registry-side progress (review._assemble_progress), then escalates only the non-terminal jobs to their runtime — a routine poll of an already-terminal run touches no runtime (empty to_probe ⇒ escalated=False). job narrows both the escalation and the report to one family (the per-family drill-down; an unknown name raises ConfigError listing the valid ones); settings is the GCP identity (from the SF_* env when None); stale_after_s overrides the startup-grace floor (_DEFAULT_STALE_S) that decides whether a vanished young job reads LOST or still-starting, and abandoned_after_s the far longer window (_DEFAULT_ABANDONED_WAIT_S) past which a capacity wait reads as a walk nobody is walking. A family whose handle can't be parsed (a pre-feature or malformed row) degrades to registry-only via no_handle rather than raising.

Source code in src/scale_forecasting/probes/reconcile.py

def probe_run(
    run_id: str,
    *,
    job: str | None = None,
    settings: Settings | None = None,
    stale_after_s: float | None = None,
    abandoned_after_s: float | None = None,
) -> ProbeReport:  # pragma: no cover - GCP I/O
    """Reconcile a run's registry state against live runtime state → a `ProbeReport`.

    Reads the run's header + config + job rows + landed-cell counts (`registry.reads`,
    `registry.jobs`), assembles the registry-side progress (`review._assemble_progress`), then
    escalates **only** the non-terminal
    jobs to their runtime — a routine poll of an already-terminal run touches no runtime (empty
    ``to_probe`` ⇒ ``escalated=False``). ``job`` narrows *both* the escalation and the report to one
    family (the per-family drill-down; an unknown name raises `ConfigError` listing the valid ones);
    ``settings`` is the GCP identity (from the ``SF_*`` env when ``None``); ``stale_after_s``
    overrides the startup-grace floor (`_DEFAULT_STALE_S`) that decides whether a vanished young job
    reads LOST or still-starting, and ``abandoned_after_s`` the far longer window
    (`_DEFAULT_ABANDONED_WAIT_S`) past which a capacity wait reads as a walk nobody is walking. A
    family whose handle can't be parsed (a pre-feature or malformed row) degrades to registry-only
    via ``no_handle`` rather than raising.
    """
    from ..settings import Settings

    s = settings if settings is not None else Settings.resolve()
    _progress, report, _rows = read_and_probe(
        run_id,
        job=job,
        settings=s,
        stale_after_s=stale_after_s,
        abandoned_after_s=abandoned_after_s,
    )
    return report

read_and_probe

read_and_probe(
    run_id,
    *,
    job,
    settings,
    stale_after_s,
    abandoned_after_s=None,
)

Read the registry, escalate the incomplete jobs, reconcile → (progress, report, rows).

The shared read+probe body behind three callers: probe_run (which keeps just the report), cancel.cancel_run (which also needs the v_run_jobs rows for their job_id + probe_handle to finalize), and review.monitor_run with probe=True (which keeps the progress and attaches the report) — so a monitor that escalates pays for the native calls, not for a second set of registry queries. rows is the run's job rows filtered to job (all families when None); progress and the reconciled report cover the same set. abandoned_after_s overrides the capacity-wait window (_DEFAULT_ABANDONED_WAIT_S).

Source code in src/scale_forecasting/probes/reconcile.py

def read_and_probe(
    run_id: str,
    *,
    job: str | None,
    settings: Settings,
    stale_after_s: float | None,
    abandoned_after_s: float | None = None,
) -> tuple[RunProgress, ProbeReport, list[dict[str, Any]]]:  # pragma: no cover - GCP I/O
    """Read the registry, escalate the incomplete jobs, reconcile → `(progress, report, rows)`.

    The shared read+probe body behind three callers: `probe_run` (which keeps just the report),
    `cancel.cancel_run` (which also needs the ``v_run_jobs`` rows for their ``job_id`` +
    ``probe_handle`` to finalize), and `review.monitor_run` with ``probe=True`` (which keeps the
    progress and attaches the report) — so a monitor that escalates pays for the native calls, not
    for a second set of registry queries. ``rows`` is the run's job rows filtered to ``job`` (all
    families when ``None``); ``progress`` and the reconciled ``report`` cover the same set.
    ``abandoned_after_s`` overrides the capacity-wait window (`_DEFAULT_ABANDONED_WAIT_S`).
    """
    from ..config import RunConfig
    from ..registry.jobs import read_run_jobs
    from ..registry.reads import read_progress, read_run_config, read_run_summary
    from ..review import _assemble_progress

    summary = read_run_summary(run_id, settings=settings)
    raw = read_run_config(run_id, settings=settings)
    cfg = RunConfig.model_validate(raw) if raw else None
    job_rows = read_run_jobs(run_id, settings=settings) if cfg else []
    progress_rows = read_progress(run_id, settings=settings) if cfg else []
    progress = _assemble_progress(
        run_id, summary, cfg, job_rows, progress_rows, now=datetime.now(UTC)
    )

    # --job narrows the report and the rows together, or neither — see `_narrow_to_job`.
    progress, rows = _narrow_to_job(progress, job_rows, job, run_id)
    # Escalate every non-terminal job to its runtime; terminal rows short-circuit to the registry,
    # and so do AWAITING_CAPACITY rows that still look pre-launch — a family between capacity
    # attempts has no runtime job to address, so escalating it would spend a live API call to be
    # told NOT_FOUND, which we already knew. One carrying a runtime job id *is* escalated, because
    # it has provably left the walk. `_probe_targets` is the pure half, checkable with no cloud.
    to_probe = _probe_targets(rows)
    # A RUNNING family quiet longer than the floor is "stale" — past its startup grace, so a
    # vanished runtime job is judged LOST rather than still-starting (see `_verdict_for_family`).
    # Read off the assembled progress, so the monitor's reported age and the probe's escalation
    # decision can never disagree about how quiet a family has been.
    stale = frozenset(f.family for f in progress.families if _is_stale(f, stale_after_s))
    # And an AWAITING_CAPACITY family quiet past any walk's own budget is "abandoned" — the one
    # reading here that comes from the clock with no runtime to ask, because a pre-launch row has
    # no runtime to ask (see `_is_abandoned_wait`). It deliberately does not widen `to_probe`: the
    # clock is the witness of last resort, used only where a runtime reading was unavailable.
    abandoned = frozenset(
        f.family for f in progress.families if _is_abandoned_wait(f, abandoned_after_s)
    )
    native: dict[str, ProbeResult] = {}
    no_handle: set[str] = set()
    for r in to_probe:
        handle = ProbeHandle.from_job_row(r)
        if handle is None:
            no_handle.add(r["family"])
            continue
        native[r["family"]] = get_probe(handle.runtime).check(handle, settings=settings)
    report = _assemble_probe_report(progress, native, frozenset(no_handle), stale, abandoned)
    return progress, report, rows

settle

Write back the verdict the probe already computed — the fourth registry-repair verb.

--probe reads a run, reconciles it against its runtimes, decides that the deep-learning family whose row still says RUNNING actually finished forty minutes ago — and then throws that answer away. Nothing in the system could act on it: close-runs refuses a run with a non-terminal job row (correctly — only a probe can tell a live job from a stale row), --cancel --force stamps CANCELLED over work that succeeded, and drop-run destroys real predictions to fix a status field. settle_run is the missing move: --probe that writes, at the job tier, taking the probe's own per-family verdict rather than a roll-up.

Three properties hold it together, and they are the same three that hold cancel together because they are the properties a repair verb needs:

Preview by default. Without yes, settle_run probes, builds the SettlePlan, and writes nothing. yes= rather than confirm= is deliberate: confirm= gates stopping running work in the cloud, yes= gates repairing a status field in our own registry — the same gate registry.ops.close_runs and registry.ops.sweep_orphans use, and settle belongs with them.

Refusal is the feature. Only five (verdict, native-state) combinations settle. UNKNOWN — the probe degraded, no handle was recorded, the runtime says SUCCEEDED but the cells have not landed — writes nothing, ever. A verb that guessed would be worse than no verb, because its guesses would be indistinguishable from measurements afterwards.

Settle never deletes, and never invents time. One UPDATE per row: the status, a failure_reason when there is a token for it, and an audit blob under job_telemetry.$.settle recording who settled it, from what, on what evidence. It does not stamp ended_at or runtime_seconds — a row settled three days after the fact would report three days of runtime for a ten-minute job, and every consumer already tolerates those being NULL. It does not touch the run header: registry.ops.close_runs owns that tier, and settling the job rows is what unblocks it (the report prints the hint).

The pure pieces — the decision table, the plan, the audit blob, and the plan→rows join — are unit tested per row; settle_run is the I/O orchestrator that sequences them.

SettleDecision dataclass

SettleDecision(status, failure_reason, reason)

What one family's row will be set to, and the one-line reason it can be (pure).

failure_reason is the short token stamped alongside a FAILED status (vocabulary's RUNTIME_FAILED / RUNTIME_LOST); None for a COMPLETED settle, which needs no excuse.

SettleItem dataclass

SettleItem(
    family,
    runtime,
    registry_status,
    verdict,
    native_state,
    n_done,
    n_expected,
    decision,
    note,
)

One family's line in the settle preview: what it says now, what it would say, and why.

decision is None for a family that will be left alone; note is the human one-liner shown in the preview either way, because why a row is being refused is the part an operator most needs — "UNKNOWN: the probe degraded" means go look, and burying it under a count of what was settled is how the one row that mattered gets missed.

SettleOutcome dataclass

SettleOutcome(
    family,
    job_key,
    from_status,
    to_status,
    final_status,
    settled,
    detail,
)

What actually happened to one family's row, read back rather than assumed.

final_status is the row's status re-read after the write, not the status we asked for. A settle UPDATE carries a status guard (_SETTLE_GUARD), and a guarded skip is silent — the statement succeeds and touches nothing — so the only honest way to report the outcome is to look. settled is True when final_status is what the decision asked for.

SettlePlan dataclass

SettlePlan(run_id, header_status, items)

Every family of a run with its settle verdict — one SettleItem each, in report order.

SettleReport dataclass

SettleReport(
    run_id,
    plan,
    executed,
    outcomes,
    header_hint,
    actor,
    reason,
)

The result of a settle call — a preview (executed=False) or the executed outcome.

plan is always present (it is what a no-yes call returns). header_hint is the follow-up line when the run's header is still non-terminal after the job rows were settled: settle deliberately does not close headers, so it says who does.

format_settle_plan

format_settle_plan(plan)

Render a SettlePlan as the operator-facing preview — the exact text a dry run prints.

Source code in src/scale_forecasting/probes/settle.py

def format_settle_plan(plan: SettlePlan) -> str:
    """Render a `SettlePlan` as the operator-facing preview — the exact text a dry run prints."""
    lines = [f"settle run {plan.run_id} (header {plan.header_status or 'unknown'})"]
    for item in plan.items:
        mark = "  ->" if item.settleable else "    "
        lines.append(f"{mark} {item.family:<16} {item.note}")
    lines.append(f"  {plan.n_settleable} of {len(plan.items)} job row(s) would be settled")
    return "\n".join(lines)

settle_run

settle_run(
    run_id,
    *,
    job=None,
    yes=False,
    reason="",
    actor=None,
    settings=None,
    stale_after_s=None,
    abandoned_after_s=None,
)

Repair a run's stale job rows from the probe's own verdicts — preview unless yes.

Probes the run (reconcile.read_and_probe), builds the SettlePlan, and — only when yes — writes each settleable family's row: the decided status, the failure_reason token when there is one, and the audit blob merged under job_telemetry.$.settle. Without yes it returns the plan and writes nothing. job narrows to one family; reason and actor (default: resolved from ADC) are recorded in the audit. stale_after_s and abandoned_after_s pass through to the probe, where they set the two clocks that decide the LOST and ABANDONED_WAIT verdicts respectively.

Every write carries a status guard against _SETTLE_GUARD, so a job that reached a terminal status of its own between the probe read and the UPDATE keeps it — the process that ran the job outranks a repair verb. That skip is silent, so the rows are re-read once afterwards and each outcome reports the status it actually ended up with.

Source code in src/scale_forecasting/probes/settle.py

def settle_run(
    run_id: str,
    *,
    job: str | None = None,
    yes: bool = False,
    reason: str = "",
    actor: str | None = None,
    settings: Settings | None = None,
    stale_after_s: float | None = None,
    abandoned_after_s: float | None = None,
) -> SettleReport:  # pragma: no cover - GCP I/O
    """Repair a run's stale job rows from the probe's own verdicts — preview unless ``yes``.

    Probes the run (`reconcile.read_and_probe`), builds the `SettlePlan`, and — **only when
    ``yes``** — writes each settleable family's row: the decided status, the ``failure_reason``
    token when there is one, and the audit blob merged under ``job_telemetry.$.settle``. Without
    ``yes`` it returns the plan and writes nothing. ``job`` narrows to one family; ``reason`` and
    ``actor`` (default: resolved from ADC) are recorded in the audit. ``stale_after_s`` and
    ``abandoned_after_s`` pass through to the probe, where they set the two clocks that decide the
    ``LOST`` and ``ABANDONED_WAIT`` verdicts respectively.

    Every write carries a status guard against `_SETTLE_GUARD`, so a job that reached a terminal
    status of its own between the probe read and the UPDATE keeps it — the process that ran the
    job outranks a repair verb. That skip is silent, so the rows are re-read once afterwards and
    each outcome reports the status it actually ended up with.
    """
    from ..identity import resolve_principal
    from ..registry.jobs import read_run_jobs, update_job
    from ..settings import Settings

    s = settings if settings is not None else Settings.resolve()
    _progress, report, rows = read_and_probe(
        run_id,
        job=job,
        settings=s,
        stale_after_s=stale_after_s,
        abandoned_after_s=abandoned_after_s,
    )
    plan = _assemble_settle_plan(report)
    if not yes:
        return SettleReport(
            run_id=run_id,
            plan=plan,
            executed=False,
            outcomes=(),
            header_hint="",
            actor=None,
            reason=reason,
        )

    resolved_actor = actor if actor is not None else resolve_principal(s)
    settled_at = datetime.now(UTC)
    written: list[tuple[_SettleTarget, SettleDecision]] = []
    for step in _settle_steps(plan, rows):
        decision = step.item.decision
        if decision is None:  # `_settle_steps` yields settleable items only; belt-and-braces
            continue
        fields: dict[str, Any] = {"status": decision.status}
        if decision.failure_reason is not None:
            fields["failure_reason"] = decision.failure_reason
        update_job(
            step.row["job_id"],
            settings=s,
            merge_telemetry={
                "settle": _build_settle_audit(
                    step.item, actor=resolved_actor, settled_at=settled_at, reason=reason
                )
            },
            unless_status_in=_SETTLE_GUARD,
            **fields,
        )
        written.append((step, decision))

    # Re-read rather than assume: a guarded skip is silent, so the plan's intent is not evidence.
    after = {r["family"]: r.get("status") for r in read_run_jobs(run_id, settings=s)}
    outcomes = tuple(
        SettleOutcome(
            family=step.item.family,
            job_key=step.row["job_id"],
            from_status=step.item.registry_status,
            to_status=decision.status,
            final_status=after.get(step.item.family),
            settled=after.get(step.item.family) == decision.status,
            detail=decision.reason,
        )
        for step, decision in written
    )
    hint = _header_hint(plan, after)
    return SettleReport(
        run_id=run_id,
        plan=plan,
        executed=True,
        outcomes=outcomes,
        header_hint=hint,
        actor=resolved_actor,
        reason=reason,
    )

cancel

Stop a run's jobs — the blast-radius plan, the confirmation gate, and the audit trail.

The only destructive, outward-facing path in this package, so it is built around honesty about what stops and what survives (§7 of the design). Three properties hold it together:

Preview by default. Without confirm, cancel_run probes, builds the CancelPlan, and touches no runtime and no registry. You see the blast radius before anything happens to it.

Cancel never deletes. Partial results that already landed are retained; they are labelled by the CANCELLED status and the cell counts, never removed. A cancelled run stays readable.

Every stop is auditable. Each finalized row carries who cancelled it, when, why, and what the runtime state was at the moment of the stop, merged into job_telemetry.$.cancel alongside the handle that addressed the job — so a cancelled attempt stays reconcilable afterwards.

The pure pieces — plan, audit blob, header roll-up, and the plan→rows join (_cancel_steps) that decides which families a confirmed cancel actually addresses — are unit-tested per row; cancel_run is the I/O orchestrator that sequences them.

CancelOutcome dataclass

CancelOutcome(
    family,
    job_key,
    requested,
    cancelled,
    stopped,
    already_gone,
    detail,
)

What actually happened to one family's job on a confirmed cancel.

requested is True when a stop was issued to the runtime (False when we couldn't even address the job — no handle). cancelled is True when the registry row was finalized to CANCELLED (the runtime confirmed the stop, or the job was already gone). stopped / already_gone are the raw vocabulary.CancelResult flags; detail is the short reason.

CancelPlan dataclass

CancelPlan(run_id, items, ensemble_suppressed)

The full blast radius of a cancel: one CancelPlanItem per family + the ensemble impact.

ensemble_suppressed is True when cancelling a base family will cause the run's ensemble node to be skipped (§7.4 — a cancelled base family suppresses the ensemble rather than producing a lopsided one). n_cancellable is how many jobs a confirmed cancel would actually stop.

CancelPlanItem dataclass

CancelPlanItem(
    family,
    runtime,
    registry_status,
    native_state,
    cancellable,
    n_done,
    n_expected,
    note,
)

One family's line in the blast-radius preview: what will happen and what data is retained.

cancellable is keyed on the registry status (non-terminal ⇒ will cancel) so the plan is robust even when the live probe degraded; native_state enriches the note when we have it. n_done/n_expected are the landed-vs-expected cells — the partial results a cancel retains (never deletes). note is the human one-liner shown in the preview.

CancelReport dataclass

CancelReport(
    run_id,
    plan,
    executed,
    outcomes,
    header_status,
    actor,
    reason,
)

The result of a cancel call — a preview (executed=False) or the executed outcome.

plan is the blast radius (always present — it is what a no-confirm call returns). outcomes is empty for a preview and one CancelOutcome per attempted family otherwise. header_status is the run's rolled-up status after the cancel (None when the header was not changed — e.g. a preview, or a cancel whose stops all failed). actor / reason are the audit identity recorded on the cancelled rows.

cancel_run

cancel_run(
    run_id,
    *,
    job=None,
    confirm=False,
    reason="",
    actor=None,
    settings=None,
    stale_after_s=None,
)

Cancel a run (or one family) — preview unless confirm; finalize registry to CANCELLED.

Probes the run first (reconcile.read_and_probe) to reconcile live state, builds the blast- radius CancelPlan, and — only when confirm — stops each cancellable family's runtime job and finalizes its row to CANCELLED with an audit blob (_finalize_cancelled), then rolls the run header up (_roll_header_after_cancel). Without confirm it returns the plan, touches no runtime and no registry (the CLI prints the preview and exits). job narrows to one family; reason and actor (default: resolved from ADC) are recorded for audit. Cancel never deletes — landed partial results are retained and labelled by the CANCELLED status + counts.

Source code in src/scale_forecasting/probes/cancel.py

def cancel_run(
    run_id: str,
    *,
    job: str | None = None,
    confirm: bool = False,
    reason: str = "",
    actor: str | None = None,
    settings: Settings | None = None,
    stale_after_s: float | None = None,
) -> CancelReport:  # pragma: no cover - GCP I/O
    """Cancel a run (or one family) — preview unless ``confirm``; finalize registry to CANCELLED.

    Probes the run first (`reconcile.read_and_probe`) to reconcile live state, builds the blast-
    radius `CancelPlan`, and — **only when ``confirm``** — stops each cancellable family's runtime
    job and finalizes its row to ``CANCELLED`` with an audit blob (`_finalize_cancelled`), then
    rolls the run header up (`_roll_header_after_cancel`). Without ``confirm`` it returns the plan,
    touches no runtime and no registry (the CLI prints the preview and exits). ``job`` narrows to
    one family; ``reason`` and ``actor`` (default: resolved from ADC) are recorded for audit. Cancel
    **never deletes** — landed partial results are retained and labelled by the CANCELLED status +
    counts.
    """
    from ..identity import resolve_principal
    from ..settings import Settings

    s = settings if settings is not None else Settings.resolve()
    _progress, report, rows = read_and_probe(
        run_id, job=job, settings=s, stale_after_s=stale_after_s
    )
    plan = _assemble_cancel_plan(report)
    if not confirm:
        return CancelReport(
            run_id=run_id,
            plan=plan,
            executed=False,
            outcomes=(),
            header_status=None,
            actor=None,
            reason=reason,
        )

    resolved_actor = actor if actor is not None else resolve_principal(s)
    cancelled_at = datetime.now(UTC)
    outcomes: list[CancelOutcome] = []
    # Which families this actually addresses is decided out here, in the open — see `_cancel_steps`.
    for step in _cancel_steps(plan, rows):
        if isinstance(step, CancelOutcome):  # unaddressable; nothing to stop, but say so
            outcomes.append(step)
            continue
        result = get_probe(step.handle.runtime).cancel(step.handle, settings=s)
        finalized = result.stopped or result.already_gone
        if finalized:
            _finalize_cancelled(
                step.row,
                step.handle,
                step.item,
                actor=resolved_actor,
                cancelled_at=cancelled_at,
                reason=reason,
                settings=s,
            )
        outcomes.append(
            CancelOutcome(
                family=step.item.family,
                job_key=step.row["job_id"],
                requested=True,
                cancelled=finalized,
                stopped=result.stopped,
                already_gone=result.already_gone,
                detail=result.detail,
            )
        )

    # Re-read every family (not just the --job subset) so the header reflects the true post-cancel
    # state — a per-family cancel makes a multi-family run PARTIAL, a whole-run cancel CANCELLED.
    from ..registry.header import update_header
    from ..registry.jobs import read_run_jobs

    all_statuses = [r.get("status") for r in read_run_jobs(run_id, settings=s)]
    header_status = _roll_header_after_cancel(all_statuses)
    if header_status is not None:
        update_header(run_id, settings=s, status=header_status)
    return CancelReport(
        run_id=run_id,
        plan=plan,
        executed=True,
        outcomes=tuple(outcomes),
        header_status=header_status,
        actor=resolved_actor,
        reason=reason,
    )