ci: reconcile fleet Linux pools from live probe receipts

.github/workflows/ci-fast-guards.yml

@@ -47,6 +47,43 @@ jobs:
         with:
           persist-credentials: false
 
+      - name: Checkout main runner-pool policy tooling
+        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
+        with:
+          ref: main
+          path: .runner-pool-tooling
+          sparse-checkout: |
+            .github/runners.json
+            scripts/ci/runner_pools.py
+            scripts/ci/pr_runner_pool.py
+            scripts/ci/simple_pool_picker.py
+            scripts/ci/cloud_overflow_switch.py
+            scripts/ci/runner_label_policy.py
+            scripts/ci/signing_runner_variables.py
+            scripts/ci/python-with-yaml.sh
+          persist-credentials: false
+
+      - name: Checkout feat runner-pool workflow boundary
+        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
+        with:
+          # On this PR, use the exact head so the checked-out helper exists
+          # before the feat branch receives the merge. Once merged, the normal
+          # feat branch ref remains the source for later main wiring.
+          ref: ${{ github.event.pull_request.head.sha || 'feat-cmux-next' }}
+          path: .runner-pool-feat
+          sparse-checkout: |
+            .github/workflows/ci.yml
+            .github/workflows/cmux-next-ios-kit.yml
+            scripts/ci/runner_pool_reconcile.py
+            tests/test_ci_runner_pool_reconcile.py
+          persist-credentials: false
+
+      - name: Validate fleet Linux runner-pool reconciliation
+        env:
+          RUNNER_POOL_POLICY_ROOT: ${{ github.workspace }}/.runner-pool-tooling
+          RUNNER_POOL_WORKFLOW_ROOT: ${{ github.workspace }}/.runner-pool-feat
+        run: bash .runner-pool-tooling/scripts/ci/python-with-yaml.sh .runner-pool-feat/tests/test_ci_runner_pool_reconcile.py
+
       - name: Run the fast guard tests
         env:
           CMUX_GUARDS_BASE_SHA: ${{ github.event.pull_request.base.sha || github.event.merge_group.base_sha || github.event.before || '' }}

.github/workflows/ci.yml

@@ -67,7 +67,7 @@ jobs:
     # since main's required ci-status must not be superseded by a skip.
     if: ${{ !(github.event_name == 'pull_request' && github.event.pull_request.base.ref == 'feat-cmux-next' && (github.event.action == 'labeled' || github.event.action == 'unlabeled') && !contains(fromJSON('["full-ci","dev-build"]'), github.event.label.name)) }}
     environment: ${{ !contains(fromJSON('["pull_request","merge_group"]'), github.event_name) && 'ci-routing-agent' || '' }}
-    runs-on: ${{ github.repository_owner != 'manaflow-ai' && 'ubuntu-24.04' || github.event_name == 'pull_request' && github.event.pull_request.head.repo.full_name != github.repository && 'blacksmith-4vcpu-ubuntu-2404' || fromJSON(vars.RUNNER_POOLS).linux }}
+    runs-on: ${{ github.repository_owner != 'manaflow-ai' && 'ubuntu-24.04' || github.event_name == 'pull_request' && github.event.pull_request.head.repo.full_name != github.repository && 'blacksmith-4vcpu-ubuntu-2404' || contains(fromJSON('["blacksmith-4vcpu-ubuntu-2404","blacksmith-8vcpu-ubuntu-2404","blacksmith-16vcpu-ubuntu-2404"]'), vars.LINUX_RUNNER) && vars.LINUX_RUNNER || 'blacksmith-8vcpu-ubuntu-2404' }}
     timeout-minutes: 5
     outputs:
       # true for a pull request on the cmux NEXT line, stacked ones included

.github/workflows/cmux-next-ios-kit.yml

@@ -81,7 +81,7 @@ jobs:
   # sweeper re-admit a job that stalls on a dead owned label.
   runner:
     environment: ${{ !contains(fromJSON('["pull_request","merge_group"]'), github.event_name) && 'ci-routing-agent' || '' }}
-    runs-on: ${{ github.repository_owner != 'manaflow-ai' && 'ubuntu-24.04' || github.event_name == 'pull_request' && github.event.pull_request.head.repo.full_name != github.repository && 'blacksmith-4vcpu-ubuntu-2404' || !startsWith(vars.LINUX_RUNNER, 'blacksmith-') && vars.LINUX_RUNNER || fromJSON(vars.RUNNER_POOLS || '{}').linux || vars.LINUX_RUNNER || 'blacksmith-4vcpu-ubuntu-2404' }}
+    runs-on: ${{ github.repository_owner != 'manaflow-ai' && 'ubuntu-24.04' || github.event_name == 'pull_request' && github.event.pull_request.head.repo.full_name != github.repository && 'blacksmith-4vcpu-ubuntu-2404' || contains(fromJSON('["blacksmith-4vcpu-ubuntu-2404","blacksmith-8vcpu-ubuntu-2404","blacksmith-16vcpu-ubuntu-2404"]'), vars.LINUX_RUNNER) && vars.LINUX_RUNNER || 'blacksmith-8vcpu-ubuntu-2404' }}
     timeout-minutes: 5
     permissions:
       contents: read

scripts/ci/cloud_probe_rescue.py

@@ -0,0 +1,146 @@
+#!/usr/bin/env python3
+"""Bounded rescue for the credentialed cloud-overflow probe watcher.
+
+This is deliberately separate from the owned-pool/nightly rescue policy.  It
+touches one workflow and one unstarted control job only.  A queued watcher is
+safe to retry because it has not received the App key; started watchers are
+left alone and execute their normal probe/switch path.
+"""
+from __future__ import annotations
+
+import dataclasses
+import datetime as dt
+import time
+from collections.abc import Callable, Mapping
+
+WORKFLOW_PATH = ".github/workflows/ci-cloud-overflow-probe.yml"
+WATCH_JOB = "Watch the probe and switch overflow"
+QUEUED = frozenset({"queued", "waiting", "pending", "requested"})
+NO_START_SECONDS = 180
+MAX_ATTEMPT = 3
+CANCEL_WAIT_SECONDS = 20 * 60
+FORCE_CANCEL_AFTER_SECONDS = 5 * 60
+
+# The probe matrix occupies the measured Linux labels. Keep its first rescue on
+# the measured 8-vCPU Linux size, then use the existing Mac and 16-vCPU sizes.
+WATCH_LABELS = (
+    "blacksmith-8vcpu-ubuntu-2404",
+    "blacksmith-6vcpu-macos-26",
+    "blacksmith-16vcpu-ubuntu-2404",
+)
+
+
+@dataclasses.dataclass(frozen=True)
+class Decision:
+    action: str  # "rescue" or "noop"
+    reason: str
+    next_label: str = ""
+
+
+def _time(value: object) -> dt.datetime | None:
+    if not isinstance(value, str) or not value:
+        return None
+    try:
+        parsed = dt.datetime.fromisoformat(value.replace("Z", "+00:00"))
+    except ValueError:
+        return None
+    if parsed.tzinfo is None:
+        return None
+    return parsed.astimezone(dt.timezone.utc)
+
+
+def _run_id(run: Mapping[str, object]) -> int:
+    try:
+        value = int(run.get("id") or 0)
+    except (TypeError, ValueError):
+        return 0
+    return value
+
+
+def decide(run: Mapping[str, object], jobs: list[Mapping[str, object]],
+           *, now: dt.datetime, no_start_seconds: int = NO_START_SECONDS) -> Decision:
+    """Return a retry decision only for an exact queued, unassigned watcher."""
+    if run.get("path") != WORKFLOW_PATH:
+        return Decision("noop", "different workflow")
+    try:
+        attempt = int(run.get("run_attempt") or 0)
+    except (TypeError, ValueError):
+        return Decision("noop", "invalid run attempt")
+    if attempt < 1 or attempt >= MAX_ATTEMPT:
+        return Decision("noop", "attempt is at rescue limit")
+    watch = [job for job in jobs if job.get("name") == WATCH_JOB]
+    if len(watch) != 1:
+        return Decision("noop", "watch job is absent or ambiguous")
+    job = watch[0]
+    if job.get("status") not in QUEUED or job.get("runner_name") or job.get("runner_id"):
+        return Decision("noop", "watch job has started or is not queued")
+    created = _time(job.get("created_at")) or _time(run.get("created_at"))
+    if created is None or (now.astimezone(dt.timezone.utc) - created).total_seconds() < no_start_seconds:
+        return Decision("noop", "watch job has not reached the no-start budget")
+    return Decision("rescue", f"watch job queued without a runner for at least {no_start_seconds}s",
+                    WATCH_LABELS[attempt])
+
+
+def rescue_control(api: object, run: Mapping[str, object], *, now: Callable[[], dt.datetime] = lambda: dt.datetime.now(dt.timezone.utc),
+                   sleep: Callable[[float], None] = time.sleep) -> Decision:
+    """Cancel-settle-rerun one stalled probe run; return the next label.
+
+    ``api`` is the existing rescue client: ``jobs(id, attempt)``,
+    ``run(id)``, ``cancel(id)``, ``force_cancel(id)`` and ``rerun(id, next)``.
+    No caller should invoke this for a nightly run; ``decide`` rejects every
+    other workflow path before a mutation.
+    """
+    run_id = _run_id(run)
+    if not run_id:
+        return Decision("noop", "invalid run id")
+    try:
+        attempt = int(run.get("run_attempt") or 0)
+    except (TypeError, ValueError):
+        return Decision("noop", "invalid run attempt")
+    # A workflow_run event can be delayed until a later attempt exists. Read
+    # the run and its current-attempt jobs immediately before mutating it.
+    current = api.run(run_id)
+    try:
+        current_attempt = int(current.get("run_attempt") or 0)
+    except (TypeError, ValueError):
+        return Decision("noop", "current run attempt is invalid")
+    if current.get("path") != WORKFLOW_PATH or current_attempt != attempt:
+        return Decision("noop", "run attempt changed before rescue")
+    jobs = list(api.jobs(run_id, current_attempt))
+    decision = decide(current, jobs, now=now())
+    if decision.action != "rescue":
+        return decision
+    # Close the last read-to-cancel gap: a runner may be assigned while the
+    # first job listing is in flight.
+    confirm = api.run(run_id)
+    try:
+        confirmed_attempt = int(confirm.get("run_attempt") or 0)
+    except (TypeError, ValueError):
+        return Decision("noop", "current run attempt is invalid")
+    if confirm.get("path") != WORKFLOW_PATH or confirmed_attempt != attempt:
+        return Decision("noop", "run attempt changed before cancel")
+    confirmed = decide(confirm, list(api.jobs(run_id, attempt)), now=now())
+    if confirmed.action != "rescue":
+        return confirmed
+    api.cancel(run_id)
+    started = now()
+    forced = False
+    while True:
+        current = api.run(run_id)
+        try:
+            settling_attempt = int(current.get("run_attempt") or 0)
+        except (TypeError, ValueError):
+            return Decision("noop", "current run attempt became invalid while cancelling")
+        if current.get("path") != WORKFLOW_PATH or settling_attempt != attempt:
+            return Decision("noop", "run attempt changed while cancelling")
+        if current.get("status") in {"completed", "cancelled"}:
+            break
+        elapsed = (now() - started).total_seconds()
+        if not forced and elapsed >= FORCE_CANCEL_AFTER_SECONDS:
+            api.force_cancel(run_id)
+            forced = True
+        if elapsed >= CANCEL_WAIT_SECONDS:
+            return Decision("noop", "cancel did not settle before rescue deadline")
+        sleep(min(10, max(1, CANCEL_WAIT_SECONDS - elapsed)))
+    api.rerun(run_id, attempt + 1)
+    return decision