#31912 ci(runner pools): the overflow watch runs the reviewed linux pool spill

#!/usr/bin/env python3
"""Run main's runner-pool policy, then spill the linux pool off a backed-up size, not only a dead one.

On 2026-11-02 at about 04:40Z blacksmith-4vcpu-ubuntu-2404, which RUNNER_POOLS.linux and LINUX_RUNNER both named
(about 100 runs-on read the pool, about 285 read LINUX_RUNNER), had 428 jobs queued beside 535 running, the oldest
about 45 minutes (721 queued earlier), while blacksmith-16vcpu-ubuntu-2404 had about 11 queued. main's
scripts/ci/runner_pools.py moves a pool only off a member it judges dead, and a size that keeps starting jobs is
never dead, however long its queue, so nothing moved.

The scheduled overflow watch (ci-cloud-overflow-probe.yml) lives on main and keeps main's policy and clients; this
entry point is checked out from feat-cmux-next beside it at a pinned commit, the way runner_pool_reconcile.py is.
Each pass it

- runs main's runner_pools.main() unchanged, with a switch that records its writes instead of making them, so the
  dead-member rule, the probe and janitor evidence and the --check schema stay main's;
- samples the Linux queue itself (sample_load()): the newest MAX_RUNS queued and in-progress runs' jobs, read with
  the job's own GITHUB_TOKEN (actions: read), counted per Blacksmith Linux label as queued, running and `waited`
  (queued WAITED_MINUTES or more, the stall-aware failover's bound). The counts are a lower bound;
- applies spill_choice() to the linux pool: when SPILL_WAITED or more sampled jobs on the pool's live member have
  waited and a later live member has at most SPILL_ROOM_WAITED, the pool moves there, recorded in RUNNER_POOLS'
  `_spill`. It comes back once the member it left shows a drain on a complete sample (nothing waited, at most
  SPILL_RETURN_QUEUED queued, at most SPILL_RETURN_SHARE of what it ran when the pool left) after SPILL_MIN_HOLD,
  or when the target backs up while the member left does not, or after SPILL_MAX_HOLD in any case: if the member
  is still backed up the next pass spills again, so a spill never outlives the backlog by more than about an hour;
- writes RUNNER_POOLS and the LINUX_RUNNER mirror (main's mirror_writes(), so the overflow switch's record still
  wins) only when something changed.

An unreadable sample never moves the pool and keeps a spill as it is until SPILL_MAX_HOLD. The spill target must be
a live member of the declared pool (main's .github/runners.json), so a spill never leaves the runner set; every
linux member is accepted by the bounded `vars.LINUX_RUNNER` selectors (main's tests/test_ci_runner_pools.py).
Nothing here cancels or moves a queued job: jobs already bound to the backed-up size drain there.
"""

from __future__ import annotations

import argparse
import datetime as dt
import json
import os
import re
import sys
from collections.abc import Mapping, Sequence
from pathlib import Path
from typing import Any

sys.path.insert(0, str(Path(__file__).resolve().parent))
import runner_pool_reconcile  # noqa: E402  (load_policy(): main's policy, never this checkout's)

UTC = dt.timezone.utc
VARIABLE = "RUNNER_POOLS"
SPILL_KEY = "_spill"
QUEUE_AWARE_POOLS = ("linux",)
LINUX_LABEL = re.compile(r"^blacksmith-[0-9]+vcpu-ubuntu-[0-9][\w.-]*$")
# A sampled queued job counts as `waited` once it has waited this long.
WAITED_MINUTES = 3
# The sample: the newest runs in flight, created within RUN_WINDOW (older queued runs are ghosts or held).
MAX_RUNS = 60
# The client reads one page of each run's jobs (cloud_overflow_switch.GitHub.jobs(): per_page=100).
JOB_PAGE = 100
RUN_WINDOW = dt.timedelta(hours=6)
SPILL_WAITED = 15
SPILL_ROOM_WAITED = 3
SPILL_MIN_HOLD = dt.timedelta(minutes=20)
SPILL_MAX_HOLD = dt.timedelta(minutes=60)
SPILL_RETURN_QUEUED = 5
SPILL_RETURN_SHARE = 0.7
COMPLETE = "_complete"
ZERO = {"queued": 0, "running": 0, "waited": 0}


def iso(moment: dt.datetime) -> str:
    return moment.astimezone(UTC).strftime("%Y-%m-%dT%H:%M:%SZ")


def parse_time(value: object) -> dt.datetime | None:
    if not isinstance(value, str) or not value:
        return None
    try:
        moment = dt.datetime.fromisoformat(value.replace("Z", "+00:00"))
    except ValueError:
        return None
    return moment if moment.tzinfo is not None else None


def sample_load(client: Any, *, now: dt.datetime, max_runs: int = MAX_RUNS) -> dict[str, Any] | None:
    """Blacksmith Linux label -> {queued, running, waited} over the newest in-flight runs; None when unreadable.

    COMPLETE says whether every in-flight run in RUN_WINDOW was read: only then can the sample prove a drain."""
    try:
        listed = {run["id"]: run for status in ("queued", "in_progress") for run in client.runs(status)
                  if isinstance(run, Mapping) and isinstance(run.get("id"), int)}
    except Exception as error:  # noqa: BLE001 - an unread listing is no evidence
        print(f"::warning title=Linux pool spill::could not list runs ({error}); no queue sample")
        return None
    recent = [run for run in listed.values()
              if (created := parse_time(run.get("created_at"))) is not None and now - created <= RUN_WINDOW]
    recent.sort(key=lambda run: (str(run.get("created_at")), run["id"]), reverse=True)
    # A listing page holds 100 runs; a full page may have hidden more.
    complete = len(recent) <= max_runs and len(listed) < 100
    load: dict[str, Any] = {}
    for run in recent[:max_runs]:
        try:
            jobs = client.jobs(run["id"])
        except Exception:  # noqa: BLE001 - one unread run only makes the sample partial
            complete = False
            continue
        if len(jobs) >= JOB_PAGE:
            # One page of jobs; a run with more may hide queued ones (review of cmux#31884).
            complete = False
        for job in jobs:
            status = job.get("status")
            if status not in ("queued", "in_progress"):
                continue
            label = next((str(label) for label in job.get("labels") or () if LINUX_LABEL.match(str(label))), None)
            if label is None:
                continue
            entry = load.setdefault(label, dict(ZERO))
            if status == "in_progress":
                entry["running"] += 1
                continue
            entry["queued"] += 1
            created = parse_time(job.get("created_at"))
            if created is not None and now - created >= dt.timedelta(minutes=WAITED_MINUTES):
                entry["waited"] += 1
    load[COMPLETE] = complete
    return load


def valid_spill(spill: object, listed: Sequence[str]) -> bool:
    if not isinstance(spill, Mapping):
        return False
    running = spill.get("running")
    return (spill.get("from") in listed and spill.get("to") in listed and spill.get("from") != spill.get("to")
            and parse_time(spill.get("since")) is not None
            and isinstance(running, int) and not isinstance(running, bool) and running >= 0)


def spill_choice(name: str, candidates: Sequence[str], dead: Mapping[str, Any], base: str,
                 spill: Mapping[str, Any] | None, load: Mapping[str, Any] | None, *,
                 now: dt.datetime) -> tuple[str, dict[str, Any] | None, list[str]]:
    """(member, spill record or None, lines) for a pool whose first live member (main's choice) is `base`."""
    if name not in QUEUE_AWARE_POOLS:
        return base, None, []
    later = candidates[candidates.index(base) + 1:] if base in candidates else []
    if spill is not None:
        target = str(spill.get("to"))
        if spill.get("from") != base or target not in later or target in dead:
            return base, None, [f"pools.{name}: spill {spill.get('from')} -> {target} ended; "
                                f"{base} is the first live member"]
        since = parse_time(spill.get("since"))
        held = now - since if since is not None else SPILL_MAX_HOLD
        if held >= SPILL_MAX_HOLD:
            return base, None, [f"pools.{name}: spill to {target} reached {SPILL_MAX_HOLD}; back to {base} "
                                "(the next pass spills again if it is still backed up)"]
        if load is None:
            return target, dict(spill), []
        left, took = load.get(base, ZERO), load.get(target, ZERO)
        if took["waited"] >= SPILL_WAITED and left["waited"] < SPILL_WAITED and load.get(COMPLETE) is True:
            return base, None, [f"pools.{name}: {target} is backed up too ({took['waited']} sampled jobs waited) "
                                f"and {base} is not; back to {base}"]
        if held >= SPILL_MIN_HOLD and load.get(COMPLETE) is True and left["waited"] == 0 \
                and left["queued"] <= SPILL_RETURN_QUEUED \
                and left["running"] <= SPILL_RETURN_SHARE * int(spill.get("running") or 0):
            return base, None, [f"pools.{name}: {base} drained ({left['queued']} queued, {left['running']} running, "
                                f"{spill.get('running')} when the pool left); back to {base}"]
        return target, dict(spill), []
    if load is None:
        return base, None, []
    left = load.get(base, ZERO)
    if left["waited"] < SPILL_WAITED:
        return base, None, []
    for member in later:
        if member in dead:
            continue
        room = load.get(member, ZERO)
        if room["waited"] <= SPILL_ROOM_WAITED:
            record = {"from": base, "to": member, "since": iso(now), "running": left["running"]}
            return member, record, [f"pools.{name}: {base} backed up ({left['waited']} of {left['queued']} sampled "
                                    f"queued jobs waited {WAITED_MINUTES}+ min, {left['running']} running); "
                                    f"spill to {member} ({room['waited']} waited)"]
    return base, None, [f"pools.{name}: {base} backed up ({left['waited']} sampled jobs waited) and no later live "
                        "member has room; staying"]


class Recorder:
    """The switch main's policy writes through: reads go to the real variables, writes are only recorded."""

    def __init__(self, real: Any, env: Mapping[str, str]) -> None:
        self.real, self.env, self.writes = real, env, {}

    def get_variable(self, name: str) -> str | None:
        if name in self.writes:
            return self.writes[name]
        if self.real is not None:
            return self.real.get_variable(name)
        return self.env.get(name)

    def set_variable(self, name: str, value: str) -> None:
        self.writes[name] = value


def apply(policy: Any, raw: str | None, policy_value: str | None, pools: Mapping[str, Any],
          load: Mapping[str, Any] | None, *, now: dt.datetime) -> tuple[dict[str, Any], list[str]]:
    """The RUNNER_POOLS value after the spill step: main's next value (`policy_value`, else the current one) with the
    linux pool spilled or brought back, and `_spill` carried from the current raw value."""
    chosen, _ = policy.parse_value(policy_value if policy_value is not None else raw, pools)
    try:
        previous = json.loads(raw) if (raw or "").strip() else {}
    except ValueError:
        previous = {}
    spills = previous.get(SPILL_KEY) if isinstance(previous, Mapping) else None
    spills = spills if isinstance(spills, Mapping) else {}
    value = dict(chosen)
    dead = value.get(policy.DEAD_KEY) or {}
    lines: list[str] = []
    kept: dict[str, Any] = {}
    for name in QUEUE_AWARE_POOLS:
        if name not in pools or name not in value:
            continue
        listed = policy.members(pools[name])
        spill = spills.get(name)
        if spill is not None and not valid_spill(spill, listed):
            lines.append(f"{VARIABLE}.{SPILL_KEY}.{name} is not a spill between two of its members; dropped")
            spill = None
        member, record, said = spill_choice(name, listed, dead, value[name], spill, load, now=now)
        lines += said
        if member != value[name]:
            lines.append(f"pools.{name}: {value[name]} -> {member}")
        value[name] = member
        if record is not None:
            kept[name] = record
    if kept:
        value[SPILL_KEY] = kept
    else:
        value.pop(SPILL_KEY, None)
    return value, lines


def main(argv: Sequence[str] | None = None, env: Mapping[str, str] | None = None, *, policy: Any = None,
         switch: Any = None, client: Any = None, api: Any = None, now: dt.datetime | None = None) -> int:
    env = os.environ if env is None else env
    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    parser.add_argument("--policy-root", type=Path, default=Path.cwd(),
                        help="the main checkout whose scripts/ci/runner_pools.py and .github/runners.json decide")
    parser.add_argument("--dry-run", action="store_true",
                        default=(env.get("DRY_RUN", "").strip().lower() in {"1", "true", "yes"}))
    args = parser.parse_args(argv)
    now = now or dt.datetime.now(UTC)
    policy = policy or runner_pool_reconcile.load_policy(args.policy_root)
    repo = env.get("GH_REPO") or env.get("GITHUB_REPOSITORY") or ""
    if switch is None and (env.get("SWITCH_TOKEN") or "").strip() and repo:
        switch = policy.Variables(str(env["SWITCH_TOKEN"]).strip(), repo)
    if client is None and (env.get("GH_TOKEN") or "").strip() and repo:
        import cloud_overflow_switch  # noqa: PLC0415 - main's client, loaded beside its policy
        client = cloud_overflow_switch.GitHub(str(env["GH_TOKEN"]).strip(), repo)

    # The current value, read live: the job-start value may predate the switch step's writes.
    raw = env.get(VARIABLE)
    if switch is not None:
        try:
            live = switch.get_variable(VARIABLE)
            raw = live if live is not None else raw
        except Exception as error:  # noqa: BLE001 - fall back to the job-start value
            print(f"::warning title=Linux pool spill::could not read {VARIABLE} live ({error}); using the job's value")
    recorder = Recorder(switch, env)
    policy_env = {**env, VARIABLE: raw or "", "DRY_RUN": "false"}
    code = policy.main([], policy_env, api=api, switch=recorder, now=now)
    if code != 0:
        return code
    pools = policy.load_pools(args.policy_root / ".github" / "runners.json")
    load = sample_load(client, now=now) if client is not None else None
    if load is not None:
        print("linux queue sample: " + json.dumps(load, sort_keys=True))
    value, lines = apply(policy, raw, recorder.writes.get(VARIABLE), pools, load, now=now)
    for line in lines:
        print(line)
    try:
        before = json.loads(raw) if (raw or "").strip() else {}
    except ValueError:
        before = {}
    before = before if isinstance(before, dict) else {}
    strip = lambda mapping: {k: v for k, v in mapping.items() if k != policy.UPDATED_KEY}  # noqa: E731
    if strip(value) != strip(before) or policy.UPDATED_KEY not in before:
        value[policy.UPDATED_KEY] = iso(now)
    else:
        value[policy.UPDATED_KEY] = before[policy.UPDATED_KEY]
    writes: list[tuple[str, str]] = []
    rendered = policy.render(value)
    if rendered != (raw or "").strip():
        writes.append((VARIABLE, rendered))
    live_mirrors, record, readable = policy._live(switch, env)
    if readable:
        writes += policy.mirror_writes(value, live_mirrors, record=record)
    summary = env.get("GITHUB_STEP_SUMMARY")
    if summary:
        with open(summary, "a", encoding="utf-8") as handle:
            handle.write("### Linux pool spill\n\n" + ("\n".join(f"- {line}" for line in lines) or "- no change")
                         + "\n\n")
    if not writes:
        print(f"{VARIABLE} and its mirrors unchanged")
        return 0
    if switch is not None and VARIABLE in dict(writes):
        # Another writer may have changed RUNNER_POOLS (a dead mark, say) while this pass sampled: never
        # overwrite it with a value planned from the old one; the next pass plans from the new one.
        try:
            again = switch.get_variable(VARIABLE)
        except Exception as error:  # noqa: BLE001 - unread means unknown: do not write over it
            print(f"::warning title=Linux pool spill::could not re-read {VARIABLE} ({error}); writing nothing")
            return 0
        if (again or "").strip() != (raw or "").strip():
            print(f"::warning title=Linux pool spill::{VARIABLE} changed during this pass; writing nothing")
            return 0
    if args.dry_run:
        for name, text in writes:
            print(f"{name} would be {text} (dry run)")
        return 0
    if switch is None:
        for name, text in writes:
            print(f"::error title=Linux pool spill::no switch token (GLAEDA_ROUTE_APP_ID); {name} should be {text}")
        return 1
    for name, text in writes:
        switch.set_variable(name, text)
        print(f"{name} = {text}")
    return 0


if __name__ == "__main__":
    sys.exit(main())

#!/usr/bin/env python3
"""Reconcile live runner variables using the declared policy in a trusted checkout.

The scheduled workflow lives on main. Its policy and clients remain in that
checkout; this entry point is checked out from feat-cmux-next beside it.
"""

from __future__ import annotations

import argparse
import datetime as dt
import importlib
import json
import os
import re
from pathlib import Path
import sys
import time
from typing import Any, Mapping

FLEET_LINUX = "cmux-fleet-linux-x64"
ROUTE_LABEL = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*$")
ROUTE_VARIABLE = re.compile(r"^(?:[A-Z][A-Z0-9_]*|RUNNER_POOLS\.[a-z][a-z0-9_]*)$")


def spec_problems(policy: Any, pools: Mapping[str, Any]) -> list[str]:
    """Only the audited Linux pool may add the existing secretless JIT route."""
    ephemeral = {name: dict(spec) for name, spec in pools.items()}
    linux = ephemeral.get("linux")
    if linux and FLEET_LINUX in linux["members"]:
        linux["members"] = [label for label in linux["members"] if label != FLEET_LINUX]
        if "linux_ephemeral" not in pools:
            return ["fleet Linux requires a separate linux_ephemeral pool for legacy secret-bearing jobs"]
    return policy.spec_problems(ephemeral)


def mirror_pool(pools: Mapping[str, Any], name: str, fallback: str) -> str:
    return "linux_ephemeral" if name == "LINUX_RUNNER" and "linux_ephemeral" in pools else fallback


def record_moves(policy: Any, previous: Mapping[str, Any], value: dict[str, Any],
                 mirrors: list[tuple[str, str]], live: Mapping[str, Any], pools: Mapping[str, Any],
                 *, now: dt.datetime) -> None:
    """The bounded _moved wire format from #30390, consumed by HQ's queued-only rescue."""
    changes = [(f"RUNNER_POOLS.{name}", previous.get(name), value.get(name)) for name in pools]
    changes += [(name, live.get(name), label) for name, label in mirrors]
    current = {f"RUNNER_POOLS.{name}": value.get(name) for name in pools}
    current.update(live)
    current.update(mirrors)
    kept = []
    for entry in previous.get("_moved", []) if isinstance(previous.get("_moved"), list) else []:
        if not isinstance(entry, Mapping) or not all(isinstance(entry.get(k), str)
                                                    for k in ("variable", "from", "to", "at")):
            continue
        stamp = policy.parse_time(entry["at"])
        if (ROUTE_VARIABLE.fullmatch(entry["variable"]) and ROUTE_LABEL.fullmatch(entry["from"])
                and ROUTE_LABEL.fullmatch(entry["to"]) and entry["from"] != entry["to"]
                and stamp is not None and dt.timedelta(minutes=-5) <= now - stamp <= dt.timedelta(hours=6)
                and current.get(entry["variable"]) != entry["from"]):
            kept.append({key: entry[key] for key in ("variable", "from", "to", "at")})
    for variable, old, new in changes:
        if (not isinstance(old, str) or not isinstance(new, str) or old == new
                or not ROUTE_LABEL.fullmatch(old) or not ROUTE_LABEL.fullmatch(new)):
            continue
        kept = [entry for entry in kept if (entry["variable"], entry["from"]) != (variable, old)]
        kept.append({"variable": variable, "from": old, "to": new, "at": policy.iso(now)})
    if kept or "_moved" in previous:
        value["_moved"] = kept[-16:]


def load_policy(source_root: Path) -> Any:
    """Load main's policy and its clients together, never this entry point's deps."""
    source = source_root.resolve() / "scripts" / "ci"
    if not (source / "runner_pools.py").is_file():
        raise ValueError(f"no runner-pool policy in {source}")
    sys.path.insert(0, str(source))
    policy = importlib.import_module("runner_pools")
    if Path(policy.__file__).resolve().parent != source:
        raise ValueError("runner-pool policy was imported from a different checkout")
    return policy


def plan(policy: Any, raw: str | None, pools: Mapping[str, Any], health: Any,
         snapshot: Any, *, now: dt.datetime) -> tuple[dict[str, Any], list[str]]:
    """Keep negative probe evidence until a later start, and never select a dead label."""
    current, notes = policy.parse_value(raw, pools)
    dead, started, evidence_notes = policy.label_evidence(health, snapshot, now=now)
    notes += evidence_notes
    fleet = FLEET_LINUX in pools.get("linux", {}).get("members", [])
    # A stalled probe does not become a successful probe merely by aging. A
    # fresh positive receipt (from either source) still clears its dead mark.
    if isinstance(health, Mapping):
        checked = policy.parse_time(health.get("checked"))
        labels = health.get("labels")
        if checked is not None and checked <= now + dt.timedelta(minutes=1) and isinstance(labels, Mapping):
            dead |= {label for label, state in labels.items() if state == "stalled"} - started
    if fleet:
        # A probe really requests the JIT label; merely listing an idle runner
        # is not a lease or evidence that a job started. Paid overflow requires
        # that probe's recorded three-minute no-start, never a missing read.
        labels = health.get("labels", {}) if isinstance(health, Mapping) else {}
        checked = policy.parse_time(health.get("checked")) if isinstance(health, Mapping) else None
        try:
            minutes = float(health.get("minutes", 0)) if isinstance(health, Mapping) else 0
        except (TypeError, ValueError):
            minutes = 0
        stalled_receipt = (checked is not None and checked <= now + dt.timedelta(minutes=1)
                           and labels.get(FLEET_LINUX) == "stalled" and minutes >= 3)
        if FLEET_LINUX not in started and not stalled_receipt and current.get("linux") != FLEET_LINUX:
            raise ValueError("no fleet start or recorded three-minute no-start; do not publish a guessed route")
        if FLEET_LINUX in dead and not stalled_receipt:
            raise ValueError("fleet appears stalled but has no three-minute probe receipt; no overflow write")
    value, changes = policy.reconcile(
        current, pools, dead_now=dead, served_now=started, now=now,
        covered=policy.evidence_covers(health, snapshot, pools, now=now))
    # The older policy chooses candidates[0] when all are dead. Refuse the
    # write rather than advertise a known stalled member as a live route.
    for name in pools:
        if value[name] in value.get(policy.DEAD_KEY, {}):
            if name == "linux_ephemeral" and value.get("linux") == FLEET_LINUX:
                # Blacksmith being entirely unavailable must not stop audited
                # secretless work on a fleet that just started its probe.
                value[name] = None
                notes.append("linux_ephemeral unavailable: all measured labels stalled; legacy mirror stays put")
                continue
            raise ValueError(f"pools.{name}: every declared member is stalled; no live route to publish")
    # A newly deployed consumer may add a pool before this checkout sees its
    # declaration. The writer owns declared keys only; retain the rest.
    try:
        previous = json.loads(raw or "{}")
    except ValueError:
        previous = {}
    if isinstance(previous, dict):
        old_marks = previous.get(policy.DEAD_KEY)
        known = {label for spec in pools.values() for label in policy.members(spec)}
        if isinstance(old_marks, dict):
            value[policy.DEAD_KEY].update({label: since for label, since in old_marks.items()
                                          if label not in known and policy.parse_time(since) is not None})
        value = {**previous, **value}
    return value, notes + changes


def run(policy: Any, env: Mapping[str, str], *, check: bool = False,
        dry_run: bool = False, switch: Any = None, api: Any = None,
        now: dt.datetime | None = None) -> int:
    now = now or dt.datetime.now(dt.timezone.utc)
    pools = policy.load_pools()
    problems = spec_problems(policy, pools)
    if problems:
        raise ValueError("; ".join(problems))
    if switch is None:
        if check:
            raw = env.get(policy.VARIABLE)
            live = {name: env.get(name) for name in policy.MIRRORS}
            record = env.get(policy.OVERFLOW_RECORD)
        else:
            raise ValueError("no variable client; no routing writes")
    else:
        # Job-start values cannot protect against another completed writer.
        # Any failed live read aborts before the first mutation.
        raw = switch.get_variable(policy.VARIABLE)
        live = {name: switch.get_variable(name) for name in policy.MIRRORS}
        record = switch.get_variable(policy.OVERFLOW_RECORD)
    health = policy._health(switch, env)
    snapshot = None
    if api is not None:
        try:
            snapshot = api.snapshot(now=now)
        except Exception as error:
            print(f"::warning::No fresh runner snapshot: {error}")
    value, notes = plan(policy, raw, pools, health, snapshot, now=now)
    for note in notes:
        print(note)
    current, dropped = policy.parse_value(raw, pools)
    if check:
        available_pools = dict(pools)
        unavailable = "linux_ephemeral" in value and value["linux_ephemeral"] is None
        if unavailable:
            available_pools.pop("linux_ephemeral", None)
        problems = dropped + policy.value_problems(current, available_pools)
        if unavailable:
            problems.append("linux_ephemeral is unavailable: all trusted overflow labels stalled")
        owned = policy.switch_owned(record)
        for name, original_pool in policy.MIRRORS.items():
            pool = mirror_pool(pools, name, original_pool)
            if owned is not None and name not in owned and live.get(name) not in policy.members(pools[pool]):
                problems.append(f"{name} is not a member of {pool}")
        for name in pools:
            if current.get(name) in value.get(policy.DEAD_KEY, {}):
                problems.append(f"{policy.VARIABLE}.{name} selects stalled {current[name]}")
        for name, pool in policy.MIRRORS.items():
            if live.get(name) in value.get(policy.DEAD_KEY, {}):
                problems.append(f"{name} selects stalled {live[name]}")
        for problem in problems:
            print(f"::error::{problem}")
        return bool(problems)
    # Keep global legacy variables on their explicitly ephemeral pool. Their
    # existing consumers include jobs holding secrets on feat-cmux-next.
    mirror_value = dict(value)
    for name, original_pool in policy.MIRRORS.items():
        pool = mirror_pool(pools, name, original_pool)
        mirror_value[original_pool] = value[pool]
    recover = dict(live)
    # Recover a previously misrouted fleet value too, without adopting any
    # unrelated manually selected owned runner.
    if recover.get("LINUX_RUNNER") == FLEET_LINUX:
        recover["LINUX_RUNNER"] = ""
    mirrors = policy.mirror_writes(mirror_value, recover, record=record)
    try:
        previous = json.loads(raw or "{}")
    except ValueError:
        previous = {}
    record_moves(policy, previous if isinstance(previous, dict) else {}, value, mirrors, live, pools, now=now)
    while len(policy.render(value).encode()) > 48 * 1024 and value.get("_moved"):
        value["_moved"].pop(0)
    if len(policy.render(value).encode()) > 48 * 1024:
        raise ValueError("RUNNER_POOLS exceeds GitHub's variable limit; no routing writes")
    writes = []
    if policy.render(value) != (raw or "").strip():
        writes.append((policy.VARIABLE, policy.render(value)))
    writes += mirrors
    if writes and not dry_run and switch.get_variable(policy.VARIABLE) != raw:
        raise ValueError("RUNNER_POOLS changed during planning; leave the newer writer's value intact")
    for name, val in writes:
        print(f"{name}{' would be' if dry_run else ' ='} {val}")
        if not dry_run:
            switch.set_variable(name, val)
    if "linux_ephemeral" in value and value["linux_ephemeral"] is None:
        print("::error::Fleet routing published, but no trusted ephemeral capacity exists for LINUX_RUNNER")
        return 1
    return 0


def probe(policy: Any, env: Mapping[str, str]) -> int:
    """Keep the existing outage switch scoped to Blacksmith; observe fleet separately."""
    cloud = importlib.import_module("cloud_overflow_switch")
    pools = policy.load_pools()
    problems = spec_problems(policy, pools)
    if problems:
        raise ValueError("; ".join(problems))
    labels = tuple(policy.members(pools.get("linux_ephemeral", pools["linux"])))
    if not all(policy.EPHEMERAL.fullmatch(label) for label in labels):
        raise ValueError("the outage switch may probe only declared ephemeral Blacksmith labels")
    # This is configuration for the existing switch, not a replacement for its
    # secret/restore/cancellation policy. In particular a fleet start must not
    # count as Blacksmith recovering, or prevent a genuine Blacksmith outage.
    cloud.PROBE_LABELS = labels
    scoped_env = dict(env, PROBE_MINUTES="3")
    code = cloud.main([], scoped_env)
    if FLEET_LINUX not in policy.members(pools["linux"]) or env.get("PROBE_LABEL_DRILL"):
        return code
    repo = env.get("GH_REPO") or env.get("GITHUB_REPOSITORY") or ""
    api = cloud.GitHub(env["GH_TOKEN"], repo)
    run_id, attempt = int(env["GITHUB_RUN_ID"]), int(env.get("GITHUB_RUN_ATTEMPT") or 1)
    all_labels = (FLEET_LINUX, *labels)
    outcome, stalled = cloud.wait_for_probe(api, run_id, attempt, 3, all_labels,
                                           now=lambda: dt.datetime.now(dt.timezone.utc), sleep=time.sleep,
                                           log=lambda line: print(line, flush=True))
    if outcome != "unknown" and env.get("DRY_RUN", "").lower() not in {"1", "true", "yes"}:
        if not env.get("SWITCH_TOKEN"):
            raise ValueError("no switch token to publish observed probe health")
        health = cloud.label_health(all_labels, stalled, checked=policy.iso(dt.datetime.now(dt.timezone.utc)),
                                    run=f"https://github.com/{repo}/actions/runs/{run_id}", minutes=3)
        cloud.GitHub(env["SWITCH_TOKEN"], repo).set_variable(policy.HEALTH_VARIABLE, json.dumps(health))
    # A queued fleet probe must not retain the workflow's concurrency group
    # after all Blacksmith probes started. Existing final cleanup owns it.
    if outcome != "started" and env.get("GITHUB_OUTPUT"):
        with open(env["GITHUB_OUTPUT"], "a", encoding="utf-8") as handle:
            handle.write(f"outcome={outcome}\n")
    return code


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--source-root", type=Path, required=True)
    parser.add_argument("--check", action="store_true")
    parser.add_argument("--probe", action="store_true")
    parser.add_argument("--dry-run", action="store_true")
    args = parser.parse_args()
    policy = load_policy(args.source_root)
    env = os.environ
    if args.probe:
        return probe(policy, env)
    repo = env.get("GH_REPO") or env.get("GITHUB_REPOSITORY") or ""
    switch = policy.Variables(env["SWITCH_TOKEN"], repo) if env.get("SWITCH_TOKEN") and repo else None
    api = policy.pr_runner_pool.GitHub(env["GH_TOKEN"], repo) if env.get("GH_TOKEN") and repo else None
    try:
        return run(policy, env, check=args.check,
                   dry_run=args.dry_run or env.get("DRY_RUN", "").lower() in {"1", "true", "yes"},
                   switch=switch, api=api)
    except (ValueError, RuntimeError) as error:
        print(f"::error::Runner pools: {error}")
        return 1


if __name__ == "__main__":
    sys.exit(main())