#!/usr/bin/env python3
"""Move a pull request CI run off a busy persistent macOS pool.

pr_runner_pool.py picks one pool per run. When that pool is owned (a
`glaeda-<class>-xcode-<version>` label, pr_runner_pool.persistent), the jobs
it names in `owned_jobs` take it and the rest take retry_runner (Blacksmith).
GitHub never re-routes a queued job: one on the owned pool waits for it
however long the pool stays busy. ci-owned-pool-rescue.yml runs this script
from the default branch, with Actions write, as one sweeper (SWEEP=1,
sweep()): it finds the runs placed on an owned pool by their fixed-name
markers and gives each a thread running follow(), the per-run watch described
below. A dispatch with a run's id (WATCH_RUN_ID) watches that run alone,
checked as a workflow_run event's run would be.

The script waits for ci.yml's `changes` job, which runs the picker. When the
picker chose a persistent pool, that job uploads a marker artifact
(`macos-pool-persistent-<run id>-<attempt>-<jobs>p<placed>-<pool>`, the counts and pool
for the janitor's count); no marker means the run is on an
ephemeral pool and the watch ends. Otherwise it watches the run's jobs until the
run finishes. If a job on the persistent pool is still queued with no runner
after the budget (CI_OWNED_POOL_RESCUE_SECONDS, 90 by default), it confirms the
pull request head has not moved, cancels the run, waits for it to finish, and
re-runs it. The re-run is attempt 2, which runs `changes` again: the picker
places it like attempt 1 without queueing (pr_runner_pool.LAST_OWNED_ATTEMPT),
on the owned machines free now and Blacksmith for the rest. The sweeper finds
attempt 2 among the unfinished CI runs (owned_reruns()) and watches it the way
it watches attempt 1: it waits for `changes` and looks for attempt 2's own
marker (the marker name carries the attempt), because a macOS job gets its
label only after the picker has chosen, then for late-placement's. A job stuck
or refused there has its failed and cancelled jobs re-run on attempt 3, which
always takes Blacksmith.

A run whose pull request is still open at the run's head is never
cancelled, whatever the watch saw (rescue(), OPEN_HEAD_RULE): run
40398277457 (cmux-next, PR #25446) attempt 1 was cancelled and re-run here,
and its attempt 2 was then cancelled by the janitor and left cancelled. So
the cancel-and-re-run in this docstring applies only to runs with no pull
request (main, nightly, E2E and side-lane pushes and dispatches); a stuck
pull request job waits, and the controller's fall-through moves its compile.
A refused job's failed jobs are still re-run once its run has finished (no
cancel), and a superseded or closed pull request's run is handled as before.

An owned runner can also refuse a job it was handed: glaeda's job-started
hook exits 1 when the host is busy (its lock is held), and the job fails
within seconds, before any step of the workflow succeeds. GitHub does not
retry it, so the pull request would stay red until someone re-ran it. A job
on the persistent pool that failed within REFUSAL_SECONDS of starting, with
its runner setup step failed or no workflow step succeeded, counts as refused
(compile admission's `always()` metrics steps still succeed after a refusal).
A failed job that ran no step at all counts too, whatever its length: GitHub
fails a job whose runner went away only after 10 minutes ("The self-hosted
runner lost communication with the server"), and so does one with a step
GitHub cancelled while the job failed: its runner was stopped mid-job ("The
runner has received a shutdown signal"). A failed Select Xcode or Select
helper Xcode step also counts, without a time limit, because a missing pin is
a machine fault even when a helper build succeeded first. The watcher lets the rest of the
run finish, since GitHub re-runs no job of a run in progress and cancelling
it would kill every healthy sibling, then confirms the head has not moved and
re-runs its failed jobs. Only a run still going at the watch's end, or main's
full-suite run (whose failure would open main's red-CI issue), is cancelled
first. GitHub can also hand a job to an owned runner that never takes it, and
after a few minutes it cancels that job alone ("The job was not acquired by
Runner of type self-hosted even after multiple attempts"): cancelled, with no
runner and no step (unacquired()). A run cancelled as a whole leaves its queued
jobs the same way, so such a job counts as refused only once its run has
finished with another conclusion; its failed and cancelled jobs are re-run.
That attempt 2 reuses attempt 1's outputs, so its owned jobs take the owned
labels again (a runner pinned on attempt 1 is not reused) and what already
passed (compile admission, say) is kept. A run on an owned pool is split across pools
anyway (per-job placement, CI_PR_POOL_OWNED_SPLIT), which is
sound only because both sides run the same Xcode: retry_runner is a macOS
26 pool on the lane's pin, the pin the owned label names, and on 2026-09-24
both the minis and Blacksmith's 6vcpu and 12vcpu macOS 26 images reported
Xcode 26.6 build 17F113. If those builds ever differ, re-run the whole run
here instead (rescue with failed_only=False).

glaeda's hook does not refuse a job for the mini's capacity: it waits, with
no limit, inside the runner's setup until the units and tokens the job needs
are free. A job on the persistent pool still in its runner's setup
SETUP_WAIT_SECONDS after it started is stuck like a queued one: once no
sibling is still running (or the watch is about to end), the run is cancelled
and re-run the same way. Until then the job is not accepted, so the watch goes
on.

Attempt 2 goes back to the fleet whoever started it: the failure
attribution's re-run after a machine failure, this rescue's after a refusal
or a stuck queue, or a person's. The job takes the owned label, not the mini
that failed it: a runner that lost communication is offline, and a busy
mini's gui runner stops listening. A person's re-run of a pull request (a code
failure) goes back to the minis on any attempt. Neither has a marker of its
own; the sweeper finds them among the unfinished CI runs (owned_reruns()) and
watches them like attempt 1, and a job stuck or refused there gets the bot's
next re-run, which every runs-on sends to retry_runner (Blacksmith) from
attempt 3 on, so a refusal costs two re-runs at most. cmux-next's bot attempt
3 is the exception: it keeps macos-placement's mini label, so it is watched
for refusals only and a refused one goes to Blacksmith as attempt 4
(REFUSAL_RESCUE_ATTEMPT), three re-runs at most.

E2E runs (test-e2e.yml) are watched the same way. Its `runner` job runs
e2e_runner_pool.py, which may pick an owned pool, and uploads the same marker
(with 1 job). An E2E run is a workflow_dispatch, not a pull request, so there
is no head to re-check, and its build and test jobs are not a split that can
break. A stuck or refused E2E job gets its failed and cancelled jobs re-run,
keeping a build that passed; those jobs keep attempt 1's pick, so they take
the runner job's retry_label, a macOS 26 Blacksmith pool on the same Xcode
build, and the follow-on watch of attempt 2 finds no owned job and stops.
A UI run's retry_label stays on its owned pool, since Blacksmith cannot run UI
tests (e2e_runner_pool.py), so the watch of attempt 2 may re-run it once more;
no attempt past 2 is watched, so it still never loops. A queued UI run moved
that way only rejoins the same owned queue, costing its place in it; the watch
stays for the refusals, which a re-run does clear. When the build itself did
not succeed, every job is re-run instead, so the `sibling` job looks again for
another run compiling the same revision (e2e_build_unfinished), and the runner
job picks again: attempt 2 takes that live pick, which may be an owned Mac,
because a Blacksmith re-run cannot adopt a product an owned Mac compiled (their
Rust toolchains differ) and so compiled it again. That attempt is followed like
a full re-run, by its picker and its own marker, and attempt 3 and later always
take retry_label. A stuck E2E run that finished some other way (a newer dispatch
in its concurrency group cancelled it) is not re-run, since that would cancel
the newer one. Its watch lasts E2E_WATCH_LIMIT_SECONDS, since its test job
queues only after a sibling wait and a build.

Main's full-suite dispatch of ci.yml (ci-main-full-suite.yml, a
workflow_dispatch on main) is watched exactly like a pull request run:
pr_runner_pool.py may put it on an owned pool like a pull request, and its
`changes` job uploads the
same marker. It has no pull request, so in place of the pull request head it
checks main's HEAD: once main has moved past the run's commit, a stuck run is
cancelled but not re-run, because its completion makes
ci-main-full-suite.yml dispatch the newer HEAD, and a re-run would only queue
the older commit behind it in main's CI concurrency group. A refused job's
failed jobs are re-run whether or not main moved, so a fleet refusal never
leaves main's run red.

Dispatches of test-ios.yml and ios-screenshots.yml are watched exactly like an
E2E run (DISPATCH_WORKFLOW_PATHS). Their `runner` job runs ios_runner_pool.py,
which may put the iOS jobs on an owned pool with the glaeda-ios-sim capability
label, and uploads the same marker; from attempt 2 on every macOS job takes
its retry_runs_on, the Blacksmith pool. A job asking for a capability label no
idle mini carries waits like any other queued owned job, so it is moved after
the same budget.

Dispatches of iroh-release-gate.yml, and of cmux-next-dev-daily.yml (its
verify job), are watched the same way. Its `runner`
job runs e2e_runner_pool.py for the Tailscale version-skew job alone, and
only takes an owned pool with a machine free now (no queue rounds), so that
job rarely waits. Its simulator-e2e jobs stay on Blacksmith, but they run in
the same run: a stuck owned job's rescue cancels them with it, and a refused
one's waits for them until the watch ends. The re-run of failed and
cancelled jobs keeps the modes that passed and puts everything on Blacksmith.

Side-lane workflows (SIDE_WORKFLOW_PATHS) have no picker. On attempt 1 of a
trusted run (a same-repository pull request, or a push, schedule or
workflow_dispatch, whose code is this repository's own branch; see
TRUSTED_SIDE_EVENTS), their small macOS jobs take vars.CI_LIGHT_LANE_RUNNER
(the light minis' side label) or vars.CI_SIDE_LANE_RUNNER (the std minis'),
both glaeda-side-* labels that only the minis' non-root runners carry. Most
side lanes take the job's Blacksmith default on attempt 2 and later. cmux-next
keeps trusted attempts 1 and 2 on the owned side label and only overflows a
later bot attempt after its ten-minute queue budget. So the first job on an owned label marks the run as on a persistent
pool (a job behind a Linux gate appears once the gate ends), and the watch
stops once every owned job has been accepted, which a side lane's few short
jobs reach in minutes. A refused side-lane job gets the run's failed jobs
re-run; a stuck one gets the run cancelled and its failed and cancelled jobs
re-run, keeping the jobs that had already finished. That re-run takes the
lane's configured retry route, so the watch ends there or continues on
cmux-next's owned label. A
stuck run that finished some other way (a newer push cancelled it) is not
re-run. Its watch lasts SIDE_WATCH_LIMIT_SECONDS. A nightly-next watch does
not end when its accepted jobs are all that exist: prepare-nightly-variant takes
the side label only after build-nightly-app finishes, so it lasts until the
run finishes, within NIGHTLY_WATCH_LIMIT_SECONDS. Side lanes upload no marker
(cmux-next.yml aside), so the sweeper finds one's run by path once it has
failed (GitHub.failed_runs()), which a refused job makes it: its failed jobs
are re-run then. A failed job of such a run that is not a refusal keeps it
red (a re-run of failed jobs would repeat it), except one that needs a refused
job, directly or through others, with no other failure or cancel upstream of it:
an `if: always()` aggregator such as cmux-tui.yml's hosted-verification fails
whenever a leg it requires was refused, so it fails only because of the refusal
(downstream_of_refusals(), which reads the `needs` graph from the workflow file
at the run's head). GitHub's re-run of failed jobs re-runs it too. A
side-lane job only queued long is moved here only for a run a dispatch names
(WATCH_RUN_ID); queue_janitor.py's unclaimed-leg rule moves the rest, from the
inventory its sweep already lists: an owned leg no runner claimed for
max(45 minutes, the pool's p90 wait), at the head of an open pull request or a
live branch, gets its run cancelled and its failed jobs re-run once, where that
re-run leaves the owned label (bot_rerun_leaves_owned()). A side-lane run that is not
a pull request has no head to move, like a dispatch. cmux-next.yml exists only
on the feat-cmux-next branch; its pull request and dispatch runs upload the
owned-pool-watch marker themselves, so the sweeper adopts them like a picker's
run. Its push runs upload none (they queue each Mac job in a group per branch),
and a cmux-next push run this rescue watches anyway (a dispatch with its run
id) is never cancelled because a newer push run exists; a refused job's failed
jobs are still re-run when no newer push run covers the branch.

Nightly builds (NIGHTLY_WORKFLOW_PATH) are watched like a side lane: there is
no picker, and attempt 1 of a push or schedule run on main (or a push on
nightly-next) puts build-nightly-app on a trusted owned pool
(glaeda-trusted-<class>-xcode-<version>, TRUSTED_LABEL) together with one
runner's own label (vars.CI_NIGHTLY_TRUSTED_RUNNER or
vars.CI_NIGHTLY_NEXT_TRUSTED_RUNNER, `glaeda-runner-<name>`), while every later
attempt takes Blacksmith. That runner is reserved for the build
(reserved_runner()): a wait for it is never stuck, queued or in its setup,
because it ends when that runner finishes what it holds, and a Blacksmith
re-run of a nightly waited an hour or more (runs 39318999424, 39320477722,
39329116254, 2026-10-19). Only a refusal is acted on. A nightly job with no
runner label of its own gets the configured budget (CI_OWNED_POOL_RESCUE_SECONDS)
like a side lane. A refused job waits for the run to finish (the signing job is
skipped behind it) and then gets its failed jobs re-run. The run is not a pull
request, so before any re-run it checks whether its commit is superseded on its
branch (pull_moved()): a newer nightly run that waits pending behind this one in
nightly.yml's concurrency group, which a re-run would join and cancel, or a
newer push run on the branch, not cancelled, once the branch head has moved past
this run's commit. A superseded run is never re-run; the newer run builds the
newer head. A superseded run that has not finished (a refusal at the watch's
end, or a stuck job with no reserved runner) is cancelled, not re-run, so the
newer run starts; a build waiting on its reserved runner never gets that far.

A job stuck with no runner is not rescued while another job of its run is
running on a persistent runner, whatever the workflow: the rescue cancels the
whole run, which would move that job to Blacksmith too (#16463). The stuck
job waits until that job ends, and if the watch ends first, it stays queued
for the mini that frees up. A refused job, or one held in setup at the
watch's end, is still acted on as described above, running siblings or not.

On a pull request run, a stuck or held job is not rescued either once
another job of its attempt has failed on its own (decided_failures(): not a
refusal, not a runner that ran no step): the run is already red, and a
re-run repeats that failure while the stuck job loses its place in the
queue. It stays queued for the minis, and a failure that was really a
Blacksmith machine fault needs a person's re-run. A refusal is still re-run.
Main, nightly and push runs keep the rescue, since a stuck one holds its
concurrency group.

A job's wait is measured from the later of its `created_at` and the first
time the watcher saw it queued, so a job record created before its `needs`
were met can never count as already past the budget.

It stops watching, doing nothing, when:
- owned pools are off (CI_PR_POOL_OWNED is not 1), before any API request;
- the run is not attempt 1 of a same-repository pull request run of ci.yml
  or a side-lane workflow, of a trusted non-PR run of a side-lane workflow, of
  main's full-suite dispatch of ci.yml, or of a dispatch in
  DISPATCH_WORKFLOW_PATHS;
- a side-lane run finished with no job on an owned label, or the fleet
  accepted all of its owned jobs;
- on a re-run of failed jobs, no job runs on an owned label once
  late-placement, if it re-ran, has finished;
- on a full re-run, `changes` and late-placement finished without that
  attempt's markers;
- `changes` finished without a marker: the run is on an ephemeral pool.
  When ci.yml started the watch for late placement (LATE_PLACEMENT=1), it
  first waits for ci-macos.yml's late-placement job and follows the run if
  that job uploaded its marker (it moved jobs onto owned root runners);
- the run finished, or the watch limit passed.

Request budget: the reads use a manaflow-glaeda-route App token when the
workflow could mint one (READ_TOKEN; its own 5000 requests an hour), else
GITHUB_TOKEN, which allows about 1000 requests an hour for the whole
repository. Cancels and re-runs always use GITHUB_TOKEN (GitHub.__doc__). A run on an ephemeral pool costs a jobs listing every
POLL_SECONDS until `changes` finishes (usually two or three) plus one artifact
listing. A run on a persistent pool adds a jobs listing every POLL_SECONDS
while one of its jobs waits for a runner and every IDLE_POLL_SECONDS otherwise,
about 30 in all for an hour-long run. A read that fails is retried
READ_ATTEMPTS times before the watch gives up; a failed cancel or re-run is
never retried.

A CI run's owned jobs may wait on purpose. pr_runner_pool.py puts a run on
an owned pool while its jobs are expected to start there no later than on
Blacksmith, and within CI_PR_POOL_QUEUE_ROUNDS job lengths (default 1, at
most MAX_QUEUE_ROUNDS). No idle machine is held for jobs a run creates later
(its shards): they join the label's queue behind whatever arrived meanwhile,
and the picker keeps that queue within machines x (1 + rounds) by every
run's peak. So any owned job of a CI run may wait up to about that long, and
its budget is the pool's expected wait plus a margin:
CI_OWNED_POOL_RESCUE_SECONDS plus QUEUE_ROUND_SECONDS per round
(queue_seconds(), 900 seconds by default, so 990 in all), under the watch limit so a stuck
job is still moved. With the rounds at 0 the picker takes an owned pool
only with machines free now, and the budget is the configured one. A
test-ios.yml or test-e2e.yml run's picker queues by the same rounds, so it
gets the same allowance, on every attempt: a re-run of failed jobs queues on
the owned labels like attempt 1's jobs. The configured budget alone is an iOS
screenshots or side-lane run's (#14391: no picker; the side lanes share the
runners PR runs queue on, so they are moved to Blacksmith more often).
"""
from __future__ import annotations

import argparse
import base64
import dataclasses
import datetime as dt
import http.client
import json
import os
import re
import sys
import threading
import time
import urllib.error
import urllib.parse
import urllib.request
from collections.abc import Callable, Mapping, Sequence
from pathlib import Path
from typing import Any

sys.path.insert(0, str(Path(__file__).resolve().parent))
from pr_runner_pool import (  # noqa: E402
    MAX_QUEUE_ROUNDS, QUEUE_ROUND_MINUTES, RUNNER_LABEL_PREFIX, parse_queue_rounds, persistent)
import owned_pool_sources  # noqa: E402
import ui_tests_dispatch  # noqa: E402

CI_WORKFLOW_PATH = ".github/workflows/ci.yml"
E2E_WORKFLOW_PATH = ".github/workflows/test-e2e.yml"
IOS_TEST_WORKFLOW_PATH = ".github/workflows/test-ios.yml"
IOS_SCREENSHOTS_WORKFLOW_PATH = ".github/workflows/ios-screenshots.yml"
IROH_RELEASE_GATE_WORKFLOW_PATH = ".github/workflows/iroh-release-gate.yml"
# The cmux NEXT DEV publisher's `runner` job offers its secret-free verify job
# to an owned Mac free now, like iroh-release-gate.yml's.
CMUX_NEXT_DEV_DAILY_WORKFLOW_PATH = ".github/workflows/cmux-next-dev-daily.yml"
# workflow_dispatch runs watched like an E2E run: each has a `runner` job that
# picks the pool and uploads the marker.
DISPATCH_WORKFLOW_PATHS = (E2E_WORKFLOW_PATH, IOS_TEST_WORKFLOW_PATH, IOS_SCREENSHOTS_WORKFLOW_PATH,
                           IROH_RELEASE_GATE_WORKFLOW_PATH, CMUX_NEXT_DEV_DAILY_WORKFLOW_PATH)
# Workflows whose picker may queue a run's jobs on an owned pool within
# CI_PR_POOL_QUEUE_ROUNDS (ios_runner_pool.py and e2e_runner_pool.py read it
# since run 36136190497).
QUEUEING_WORKFLOW_PATHS = (CI_WORKFLOW_PATH, IOS_TEST_WORKFLOW_PATH, E2E_WORKFLOW_PATH)
# Side-lane workflows: no picker job. Their small macOS jobs take
# vars.CI_LIGHT_LANE_RUNNER or vars.CI_SIDE_LANE_RUNNER (glaeda-side-* labels)
# on attempt 1 of a trusted run, and their Blacksmith default from attempt 2 on.
# nightly.yml: no picker job either. build-nightly-app takes the trusted owned
# pool (vars.CI_SEED_TRUSTED_POOL) on attempt 1 of a push or schedule run on
# main, and Blacksmith on every later attempt.
NIGHTLY_WORKFLOW_PATH = ".github/workflows/nightly.yml"
NIGHTLY_EVENTS = frozenset({"push", "schedule"})
# The trusted owned pool's labels: minis with no pull request runners, whose
# job-started hook admits only main's push and schedule jobs. Only nightly.yml's
# app build asks for one through this watch.
TRUSTED_LABEL = re.compile(r"glaeda-(?:root-)?trusted-(?:xl|std|light)-xcode-[0-9]+(?:\.[0-9]+)*")
# feat-cmux-next's session and fast roles (glaeda#1530, pr_runner_pool.OWNED_LABEL there), which main's
# picker never hands out: cmux-next-base-autofix.yml's Mac exports take glaeda-session-std-xcode-26.6.
LANE_LABEL = re.compile(r"glaeda-(?:aws-)?(?:session-|fast-)(?:xl|std|light)-xcode-[0-9]+(?:\.[0-9]+)*")
# Every other workflow that can put a job on an owned pool, on main or on
# feat-cmux-next (whose files are not in main's tree): owned_pool_sources.py
# finds them from their jobs' runs-on, and tests/test_ci_owned_pool_sources.py
# fails when a workflow it finds is missing here, or one here is found in
# neither branch. Most take a glaeda-side-* label on attempt 1 of a trusted
# run and Blacksmith later; some (the base and PR autofix's Mac exports) take
# a session or gui label on every attempt. None uploads a marker, so the
# sweeper finds their failed runs by path (GitHub.failed_runs()); cmux-next.yml
# is marked as well.
SIDE_WORKFLOW_PATHS = frozenset({
    ".github/workflows/agent-brands.yml",  # feat-cmux-next only
    ".github/workflows/app-host-test-rerun.yml",
    ".github/workflows/auth-refresh-tests.yml",
    ".github/workflows/cloud-command-deadlines.yml",
    ".github/workflows/cloud-machine-tests.yml",
    ".github/workflows/cloud-task-local-tests.yml",
    ".github/workflows/cmux-next-base-autofix.yml",  # feat-cmux-next only
    ".github/workflows/cmux-next-batch.yml",  # feat-cmux-next only
    ".github/workflows/cmux-next-frame-pacing.yml",  # feat-cmux-next only
    ".github/workflows/cmux-next-media.yml",  # feat-cmux-next only
    ".github/workflows/cmux-next-pr-autofix.yml",  # feat-cmux-next only
    ".github/workflows/cmux-next-terminal.yml",  # feat-cmux-next only
    # feat-cmux-next only; uploads the owned-pool-watch marker on attempt 1.
    ".github/workflows/cmux-next.yml",
    ".github/workflows/cmux-tui-artifacts.yml",
    ".github/workflows/cmux-tui.yml",
    ".github/workflows/iroh-v2.yml",
    ".github/workflows/relay-tls.yml",
    ".github/workflows/reload-build.yml",
    ".github/workflows/remote-daemon.yml",
    ".github/workflows/terminal-hang-diagnostics.yml",
    ".github/workflows/test-feed-reply-providers.yml",  # feat-cmux-next only
})
# Side workflows whose Mac jobs take a session or gui label on every attempt (feat-cmux-next only), so no
# re-run moves them off the minis (bot_rerun_leaves_owned()).
OWNED_EVERY_ATTEMPT_PATHS = frozenset({
    ".github/workflows/cmux-next-base-autofix.yml",
    ".github/workflows/cmux-next-pr-autofix.yml",
})
# cmux-next is the side lane whose placement and retry route are minis-first;
# other side workflows keep their existing fallback policy.
CMUX_NEXT_WORKFLOW_PATH = ".github/workflows/cmux-next.yml"
# Events whose code is this repository's own: a push or schedule runs a branch
# of it, and a workflow_dispatch needs write access. A side-lane run of one of
# these is owned-eligible like a same-repository pull request. merge_group,
# workflow_run and pull_request_target are not: they can carry fork code.
TRUSTED_SIDE_EVENTS = frozenset({"push", "schedule", "workflow_dispatch"})
# Workflows that can take an owned pool but are never rescued here, each with why. A person decides
# their re-runs. owned_pool_sources.py must find no owned workflow outside these and
# WATCHED_WORKFLOW_PATHS (tests/test_ci_owned_pool_sources.py).
UNRESCUED_WORKFLOW_PATHS = {
    ".github/workflows/app-ffi-release.yml":
        "feat-cmux-next only; builds and publishes the FFI xcframework every app build links on the trusted "
        "promote mini: a release step, so a person re-runs it",
    ".github/workflows/cmux-tui-nightly.yml":
        "feat-cmux-next only; builds (on a side label it passes cmux-tui-build-package.yml) and publishes the "
        "cmux-tui nightly packages: a release step, so a person re-runs it",
    ".github/workflows/ios-internal-nightly.yml":
        "feat-cmux-next only; signs and uploads the internal iOS build: a release step, so a person re-runs it",
    ".github/workflows/main-compile-probe.yml":
        "one optional compile data point for main's compile attribution; its dispatcher already sends a probe "
        "that has not started to Blacksmith, and nothing waits on it",
    ".github/workflows/seed-derived-data.yml":
        "a cache seed on the trusted pool that nothing waits on; the next main push seeds again from the newest "
        "earlier seed",
}
# Every workflow this rescue acts on (target_from_event()); owned_pool_sources.py
# must find no owned workflow outside it and UNRESCUED_WORKFLOW_PATHS.
WATCHED_WORKFLOW_PATHS = frozenset({CI_WORKFLOW_PATH, NIGHTLY_WORKFLOW_PATH, *DISPATCH_WORKFLOW_PATHS,
                                    *SIDE_WORKFLOW_PATHS})
# test-e2e.yml's job that runs e2e_runner_pool.py (and the iOS workflows' job
# that runs ios_runner_pool.py).
E2E_PICKER_JOB = "runner"
# ci.yml's job that runs the pool picker; its jobs-API name (no `name:` override).
PICKER_JOB = "changes"
DEFAULT_BUDGET_SECONDS = 90
# Side lanes deliberately queue on the owned label while minis drain. Their
# overflow is a measured queue decision, not the 90-second CI rescue budget.
SIDE_DEFAULT_BUDGET_SECONDS = 600
MIN_BUDGET_SECONDS = 30
MAX_BUDGET_SECONDS = 600
# A job's budget ends this long before the watch does (job_budget()): two
# looks, so the rescue fires while the watch still runs.
END_MARGIN_SECONDS = 60
# One round of queue on an owned pool: the longest job a queued job commonly
# waits behind, compile admission. Over 80 pull request runs on 2026-09-25 it
# took a median 638 s on the minis (p90 745 s) and a p90 893 s on Blacksmith.
QUEUE_ROUND_SECONDS = QUEUE_ROUND_MINUTES * 60
FIRST_LOOK_SECONDS = 45
POLL_SECONDS = 20
IDLE_POLL_SECONDS = 120
# Long enough for a compile-only pull request run and its consumers to queue.
WATCH_LIMIT_SECONDS = 60 * 60
# An E2E test job queues after a sibling wait (up to 35 min) and a build.
E2E_WATCH_LIMIT_SECONDS = 150 * 60
# A side lane's macOS job is created at once, or after a Linux gate
# (cloud-machine-tests), which can wait in a busy Linux queue; a watch that
# ended before the job existed would leave it on the fleet unwatched.
SIDE_WATCH_LIMIT_SECONDS = WATCH_LIMIT_SECONDS
# A nightly's owned jobs run one after another: the app build (about 20 min
# on a mini, longer cold), then prepare-nightly-variant (up to 40 min).
NIGHTLY_WATCH_LIMIT_SECONDS = E2E_WATCH_LIMIT_SECONDS
READ_ATTEMPTS = 3
READ_RETRY_SECONDS = 10
MARKER_PREFIX = "macos-pool-persistent"
# ci-macos.yml's late-placement moved jobs after compile admission onto idle
# owned root runners (late_placement.py), and started this watch itself.
LATE_MARKER_PREFIX = "macos-pool-late"
LATE_JOB = "macos / late-placement"
# Compile admission: when a re-run of failed jobs runs it again, late-placement runs again after it.
ADMISSION_JOB = "macos / macOS compile admission"
# A cancelled run is only useful re-run: giving up leaves the pull request's
# run cancelled for good. A Mac job mid-compile has taken over 5 minutes to
# settle after a force-cancel (run 36074561333, 2026-09-24), so wait long, and
# force-cancel again while waiting.
CANCEL_WAIT_SECONDS = 20 * 60
FORCE_CANCEL_AFTER_SECONDS = 90
FORCE_CANCEL_AGAIN_SECONDS = 5 * 60
# Room for one watch's waits through GitHub rate limits (RATE_LIMIT_WAIT_SECONDS).
_RATE_LIMIT_ROOM_SECONDS = 15 * 60
# A rescue may run this long past the watch's end, so a refusal found late
# in the watch still gets its cancel settled and its re-run, rate-limit waits
# included (a sweeper's handover joins its rescues for this long).
RESCUE_GRACE_SECONDS = 25 * 60 + _RATE_LIMIT_ROOM_SECONDS
# Kept back from the job timeout for checkout and the summary.
JOB_TIMEOUT_MARGIN_SECONDS = 5 * 60
# ci-owned-pool-rescue.yml's timeout-minutes: the longest watch (an E2E
# run's), its rescue grace, and the margin.
JOB_TIMEOUT_SECONDS = E2E_WATCH_LIMIT_SECONDS + RESCUE_GRACE_SECONDS + JOB_TIMEOUT_MARGIN_SECONDS
# Time kept back after a cancel settles, for the re-run request itself.
RERUN_MARGIN_SECONDS = 60
# A refused job fails within the runner's setup; a real failure of the first
# step after checkout takes longer than this, and one that does not is cheap to
# retry. glaeda's hook once waited up to 240 s inside that setup for the mini's
# capacity before refusing; it now waits with no limit (SETUP_WAIT_SECONDS
# covers that), so a refusal is the hook's other checks, which answer at once.
REFUSAL_SECONDS = 360
# The last attempt of the bot's own re-runs that may run on an owned pool
# (pr_runner_pool.LAST_OWNED_ATTEMPT): attempt 2 is placed like attempt 1. A
# person's re-run of a pull request (a code failure:
# pr_runner_pool.host_fault_retry()) goes back to the minis on any attempt and
# is watched whatever its attempt (owned_rerun()); the rescue's re-run of it is
# the bot's, on Blacksmith.
LAST_OWNED_ATTEMPT = 2
# cmux-next's bot attempt 3 keeps macos-placement's mini label (a re-run reuses
# attempt 1's outputs), so a lend drain can refuse it too, and before 2026-10-07
# nothing re-ran it (#18147's scheme compile, 02:23Z). It is watched for
# refusals only: when every failed job of it was refused, its failed jobs are
# re-run as attempt 4, which cmux-next.yml sends to Blacksmith. Minis-first
# still holds for attempts 1 to 3, and no attempt past 3 is watched.
REFUSAL_RESCUE_ATTEMPT = LAST_OWNED_ATTEMPT + 1
# The bot's re-runs: the workflow token's, and cmux-bot's since ci-owned-pool-rescue.yml mints that App's
# token for its mutations. A re-run by anyone else is a person's (owned_rerun()). Run 40444295785, a DEV
# daily, went through 10 cmux-bot attempts on the minis while only github-actions[bot] counted here.
RESCUE_ACTORS = frozenset({"github-actions[bot]", "cmux-bot[bot]"})
# ci.yml's placement (pr_runner_pool.RESCUE_ACTOR, ci-macos.yml's pr_retry_runner) still moves only
# github-actions[bot]'s attempt 3 off the minis, so a cmux-bot re-run of CI stays a person's here: it
# goes back to the minis and is watched.
CI_RESCUE_ACTOR = "github-actions[bot]"
# Why a pull request's run at its open head is left alone (queue_janitor.py shares it).
OPEN_HEAD_RULE = ("never cancelled: an open pull request's run at its head is never cancelled by anything "
                  "automatic; the controller's fall-through moves its compile instead")


def owned_rerun(run: Mapping[str, Any]) -> bool:
    """A re-run of a CI run whose owned jobs go back to the minis, so it is watched like attempt 1: attempt 2,
    or a later one of a pull request someone other than the bot (CI_RESCUE_ACTOR; cmux-next: RESCUE_ACTORS) started."""
    attempt = int(run.get("run_attempt") or 0)
    if attempt <= 1:
        return False
    actor = str((run.get("triggering_actor") or {}).get("login") or "")
    if run.get("path") == CMUX_NEXT_WORKFLOW_PATH:
        # Manual side-lane reruns stay minis-first at any attempt. The bot's
        # second attempt is still watched so a second long queue can overflow
        # on attempt three; bot attempt three is watched for refusals only.
        return actor not in RESCUE_ACTORS or attempt <= REFUSAL_RESCUE_ATTEMPT
    return run.get("path") == CI_WORKFLOW_PATH and (attempt <= LAST_OWNED_ATTEMPT or (
        run.get("event") == "pull_request" and actor != CI_RESCUE_ACTOR))
# The runner's own steps, which run before glaeda's hook decides.
SETUP_STEPS = frozenset({"Set up job", "Set up runner"})
XCODE_SELECTION_STEPS = frozenset({"Select Xcode", "Select helper Xcode"})
# glaeda's hook no longer refuses a job for the mini's capacity: it waits,
# with no limit, inside the runner's setup ("Set up runner") until the units
# and tokens it needs are free (glaeda CAPACITY_WAIT, 2026-09-28). A job still
# there this long after it started is stuck like a queued one and is moved the
# same way (Look "rescue"). A compile waiting for a canonical root normally
# waits for one running compile, about 7 minutes.
SETUP_WAIT_SECONDS = 15 * 60
MAX_JOB_PAGES = 3
# Main's full-suite dispatch (ci-main-full-suite.yml) runs ci.yml on this branch.
MAIN_BRANCH = "main"
API = "https://api.github.com"


def budget(value: str | None) -> int | None:
    """The queued-seconds budget from the variable, or None when it is invalid."""
    raw = (value or "").strip()
    if not raw:
        return DEFAULT_BUDGET_SECONDS
    try:
        seconds = int(raw)
    except ValueError:
        return None
    return seconds if MIN_BUDGET_SECONDS <= seconds <= MAX_BUDGET_SECONDS else None


def side_budget(value: str | None) -> int | None:
    """The side-lane queue budget, defaulting to ten minutes."""
    if not (value or "").strip():
        return SIDE_DEFAULT_BUDGET_SECONDS
    return budget(value)


def queue_seconds(rounds: str | None) -> int:
    """How long a CI run's owned job may wait on purpose: the pool's expected wait bound (CI_PR_POOL_QUEUE_ROUNDS).

    An invalid value makes the picker keep every run off the owned pools, so
    it adds nothing.
    """
    # parse_queue_rounds() clamps to MAX_QUEUE_ROUNDS, so the longest budget
    # (MAX_BUDGET_SECONDS + 2,700 s) stays under WATCH_LIMIT_SECONDS.
    return min(parse_queue_rounds(rounds) or 0, MAX_QUEUE_ROUNDS) * QUEUE_ROUND_SECONDS


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


def job_pool(job: Mapping[str, Any]) -> str | None:
    """The owned pool a job asked for, if any: a pull request pool or the trusted one."""
    for label in job.get("labels") or []:
        if persistent(str(label)) or TRUSTED_LABEL.fullmatch(str(label)) or LANE_LABEL.fullmatch(str(label)):
            return str(label)
    return None


def reserved_runner(job: Mapping[str, Any]) -> str | None:
    """The one runner a job asked for by its own label (`glaeda-runner-<name>`), if any.

    Only honoured for a nightly build (assess(reserved=True)): nightly.yml names the runner kept for it
    (vars.CI_NIGHTLY_TRUSTED_RUNNER, vars.CI_NIGHTLY_NEXT_TRUSTED_RUNNER) beside the trusted pool, so the
    wait ends when that runner finishes what it holds. A pull request's compile admission also names a warm
    runner, but only as a preference the rescue may override (pr_runner_pool.py's warm affinity).
    """
    for label in job.get("labels") or []:
        if str(label).startswith(RUNNER_LABEL_PREFIX):
            return str(label)
    return None


def waiting_for_runner(job: Mapping[str, Any]) -> bool:
    return job.get("status") == "queued" and not job.get("runner_name")


def in_setup(job: Mapping[str, Any]) -> bool:
    """A job its runner took that is still in the runner's setup, where glaeda's hook waits for capacity."""
    if job.get("status") != "in_progress":
        return False
    steps = [step for step in job.get("steps") or [] if isinstance(step, Mapping)]
    return any(step.get("name") in SETUP_STEPS and step.get("status") == "in_progress" for step in steps)


def setup_seconds(job: Mapping[str, Any], now: dt.datetime) -> float:
    """How long the job has been in its current setup step (glaeda's hook runs in "Set up runner")."""
    steps = [step for step in job.get("steps") or [] if isinstance(step, Mapping)]
    step = next((step for step in steps if step.get("name") in SETUP_STEPS and step.get("status") == "in_progress"),
                None)
    started = parse_time((step or {}).get("started_at")) or parse_time(job.get("started_at"))
    return 0.0 if started is None else max(0.0, (now - started).total_seconds())


def wait_start(job: Mapping[str, Any], first_seen: dt.datetime | None = None) -> dt.datetime | None:
    created = parse_time(job.get("created_at"))
    return max(filter(None, (created, first_seen)), default=None)


def queued_seconds(job: Mapping[str, Any], now: dt.datetime, first_seen: dt.datetime | None = None) -> float:
    since = wait_start(job, first_seen)
    return 0.0 if since is None else max(0.0, (now - since).total_seconds())


def job_budget(job: Mapping[str, Any], budget_seconds: int, *, deadline: dt.datetime | None,
               floor_seconds: int | None, first_seen: dt.datetime | None = None) -> int:
    """A job's budget, cut so a job queued late is still judged before the watch ends.

    A CI run's budget includes the owned wait it may expect (queue_seconds()),
    and its shards appear about 11 minutes in; at 3 rounds a shard that got
    stuck would otherwise outlast the watch and never be moved. So a job
    waiting since `since` is rescued after at most deadline - since -
    END_MARGIN_SECONDS, and never before `floor_seconds` (the configured
    CI_OWNED_POOL_RESCUE_SECONDS).
    """
    since = wait_start(job, first_seen)
    if deadline is None or since is None:
        return budget_seconds
    left = int((deadline - since).total_seconds()) - END_MARGIN_SECONDS
    floor = budget_seconds if floor_seconds is None else min(floor_seconds, budget_seconds)
    return max(floor, min(budget_seconds, left))


def refused(job: Mapping[str, Any]) -> bool:
    """An owned runner's setup or Xcode pin failure, or a lost runner."""
    if not job_pool(job) or job.get("status") != "completed" or job.get("conclusion") != "failure":
        return False
    steps = [step for step in job.get("steps") or [] if isinstance(step, Mapping)]
    if not steps:
        # The runner ran nothing, not even "Set up job": GitHub failed a job whose
        # runner went away ("The self-hosted runner lost communication with the
        # server"), which it reports only after 10 minutes, so no length applies
        # (run 36420353579).
        return True
    if any(step.get("conclusion") == "cancelled" for step in steps) and \
            not any(step.get("conclusion") == "failure" for step in steps):
        # The runner was stopped mid-job ("The runner has received a shutdown signal",
        # then "The operation was canceled."): the step it was in is cancelled while the
        # job fails, whatever its length. Job 121926149188 (run 40424205219, PR #25241) and
        # four other open pull requests' cmux-next jobs failed so at 01:19Z on 2026-10-27,
        # in glaeda's hook and in "Run the fast tier". A failing step concludes failure, a
        # step's timeout-minutes too ("The action has timed out."), and a cancelled or
        # timed-out job concludes cancelled, so only a stopped runner leaves this. A step
        # that failed before the stop (tests, then a cleanup cut short) keeps the job the code's.
        return True
    # A helper build can precede Xcode selection, so neither elapsed time nor
    # earlier successful steps make a missing pin a source failure.
    if any(step.get("name") in XCODE_SELECTION_STEPS and step.get("conclusion") == "failure"
           for step in steps):
        return True
    started, completed = parse_time(job.get("started_at")), parse_time(job.get("completed_at"))
    if started is None or completed is None or (completed - started).total_seconds() > REFUSAL_SECONDS:
        return False
    # The hook runs inside the runner's own setup, so a failed setup step is a
    # refusal even when the job's `always()` steps still ran and succeeded.
    if any(step.get("name") in SETUP_STEPS and step.get("conclusion") == "failure" for step in steps):
        return True
    return not any(step.get("conclusion") == "success" and step.get("name") not in SETUP_STEPS
                   for step in steps)


def unacquired(job: Mapping[str, Any]) -> bool:
    """An owned job GitHub cancelled because no runner took it: cancelled, with no runner and no step.

    Run 39607661629's ios-simulator-build (2026-10-21) waited 4 minutes for glaeda-ios-sim, then GitHub
    cancelled it alone ("The job was not acquired by Runner of type self-hosted even after multiple
    attempts") and the run ended in failure. A run cancelled as a whole leaves its queued jobs the same
    way, so watch() counts this only in a run that finished with another conclusion.
    """
    return bool(job_pool(job)) and job.get("status") == "completed" and job.get("conclusion") == "cancelled" \
        and not job.get("runner_name") and not job.get("steps")


def accepted(job: Mapping[str, Any], now: dt.datetime) -> bool:
    """An owned job its runner took and has not refused: past its setup and started over REFUSAL_SECONDS ago,
    or done."""
    if job.get("status") == "completed":
        return not refused(job)
    started = parse_time(job.get("started_at"))
    return job.get("status") == "in_progress" and not in_setup(job) and started is not None and \
        (now - started).total_seconds() > REFUSAL_SECONDS


def carried(job: Mapping[str, Any], attempt_started: dt.datetime | None = None) -> bool:
    """A job a re-run of failed jobs kept from an earlier attempt.

    GitHub's attempt-jobs response has no documented ``created_at`` field. When
    the workflow run supplies its API-backed attempt start, compare the job's
    start with that boundary; retain the receipt fallback for older callers.
    """
    if attempt_started is not None:
        started = parse_time(job.get("started_at"))
        return started is not None and started < attempt_started
    created, started = parse_time(job.get("created_at")), parse_time(job.get("started_at"))
    return created is not None and started is not None and started < created


def picker_finished(jobs: Sequence[Mapping[str, Any]], picker_job: str = PICKER_JOB) -> bool:
    picker = [job for job in jobs if job.get("name") == picker_job]
    return bool(picker) and all(job.get("status") == "completed" for job in picker)


def run_finished(jobs: Sequence[Mapping[str, Any]]) -> bool:
    return bool(jobs) and all(job.get("status") == "completed" for job in jobs)


@dataclasses.dataclass(frozen=True)
class Look:
    action: str  # "rescue" (cancel, re-run all), "refused" (re-run failed jobs) or "watch"
    reason: str
    waiting: bool = False  # a persistent-pool job has no runner yet


def decided_failures(jobs: Sequence[Mapping[str, Any]], attempt_started: dt.datetime | None = None) -> list[str]:
    """Names of this attempt's jobs that failed on their own: not a fleet refusal (refused()), not a
    job whose runner never ran a step (a lost runner, owned or not) and not one carried from an
    earlier attempt. Such a failure has already made the run red."""
    return sorted(str(job.get("name") or job.get("id")) for job in jobs
                  if job.get("status") == "completed" and job.get("conclusion") == "failure"
                  and job.get("steps") and not refused(job) and not carried(job, attempt_started))


# A `${{ ... }}` expression in a job's `name:`.
NAME_EXPRESSION = re.compile(r"\$\{\{(.*?)\}\}", re.S)
EXPRESSION_LITERAL = re.compile(r"'((?:[^']|'')*)'")
# `operand == operand` or `!=`: a condition whose literal is never the expression's value.
EXPRESSION_COMPARISON = re.compile(r"(?:'(?:[^']|'')*'|[A-Za-z_][\w.\-\[\]*]*)\s*[!=]=\s*"
                                   r"(?:'(?:[^']|'')*'|[A-Za-z_][\w.\-\[\]*]*)")
EXPRESSION_CONTEXT = re.compile(r"\b(?:matrix|inputs|needs|github|vars|env|steps|secrets|strategy|runner|job)\.")


def _name_expression(expression: str) -> str:
    """A regex for the values one `${{ ... }}` of a job name can render to: its string literals (bar
    those a comparison tests against), and anything when a context value can be the result."""
    rest = EXPRESSION_COMPARISON.sub(" ", expression)
    literals = [literal.replace("''", "'") for literal in EXPRESSION_LITERAL.findall(rest)]
    choices = [re.escape(literal) for literal in literals]
    if not literals or EXPRESSION_CONTEXT.search(EXPRESSION_LITERAL.sub(" ", rest)):
        choices.append(".+?")
    return "(?:" + "|".join(choices) + ")"


def _job_name_pattern(key: str, job: Mapping[str, Any]) -> re.Pattern[str]:
    """The jobs API names a workflow job `key` can take: its `name:` (or key) rendered, with a matrix's
    ` (values)` suffix and, for a reusable workflow call, ` / <called job>`."""
    template = str(job.get("name") or key)
    pattern, at = "", 0
    for match in NAME_EXPRESSION.finditer(template):
        pattern += re.escape(template[at:match.start()]) + _name_expression(match.group(1))
        at = match.end()
    pattern += re.escape(template[at:])
    if isinstance(job.get("strategy"), Mapping) and (job.get("strategy") or {}).get("matrix") is not None:
        pattern += r"(?: \(.+\))?"
    if job.get("uses"):
        pattern += r"(?: / .+)?"
    return re.compile(pattern, re.S)


def downstream_of_refusals(workflow_text: str, jobs: Sequence[Mapping[str, Any]],
                           attempt_started: dt.datetime | None = None) -> set[str]:
    """Names of this attempt's failed jobs that failed only because a job they need was refused.

    A job counts when the workflow file's `needs` graph puts a refused job (refused(), or one no runner
    acquired, unacquired()) upstream of it, directly or through other jobs, and every job upstream of it
    that failed, timed out or was cancelled is a refusal or counts itself. That is an `if: always()`
    aggregator failing on a refused leg, such as cmux-tui.yml's hosted-verification (`require_success
    "Rust tests"` after the macOS test leg was refused). A job's own failure cannot be told apart from
    its refused upstream's, so such a job is re-run with it: GitHub's re-run of failed jobs re-runs
    every failed job and everything that needs one. Names map to workflow jobs by their rendered
    `name:`; a name that could be several jobs counts only when every one of them qualifies, and a
    name that matches none never counts.
    """
    document = owned_pool_sources._load(workflow_text)
    spec = {str(key): job for key, job in (document.get("jobs") or {}).items() if isinstance(job, Mapping)}
    needs: dict[str, set[str]] = {}
    for key, job in spec.items():
        wanted = job.get("needs") or []
        needs[key] = {str(item) for item in ([wanted] if isinstance(wanted, str) else wanted)} & set(spec)
    upstream: dict[str, set[str]] = {}

    def ancestors(key: str) -> set[str]:
        if key not in upstream:
            upstream[key] = set()  # a cycle is invalid YAML for GitHub; never loop on one
            found: set[str] = set()
            for parent in needs.get(key, ()):
                found |= {parent} | ancestors(parent)
            upstream[key] = found
        return upstream[key]

    patterns = {key: _job_name_pattern(key, job) for key, job in spec.items()}
    keys_of: dict[int, set[str]] = {}
    for job in jobs:
        name = str(job.get("name") or "")
        keys_of[id(job)] = {key for key, pattern in patterns.items() if pattern.fullmatch(name)}
    current = [job for job in jobs if job.get("status") == "completed" and not carried(job, attempt_started)]
    causes = [job for job in current if refused(job) or unacquired(job)]
    broken = [job for job in current if job.get("conclusion") in ("failure", "timed_out", "cancelled")
              and not any(job is cause for cause in causes)]
    excused: list[Mapping[str, Any]] = []
    changed = True
    while changed:
        changed = False
        for job in broken:
            keys = keys_of[id(job)]
            if job.get("conclusion") == "cancelled" or not keys or any(job is other for other in excused):
                continue
            above = set().union(*(ancestors(key) for key in keys))
            # Every job this name could be needs a refused job, whichever job each refused name is.
            if not all(any(keys_of[id(cause)] and keys_of[id(cause)] <= ancestors(key) for cause in causes)
                       for key in keys):
                continue
            # No other failure or cancel upstream of it that is not itself excused.
            if any(other is not job and keys_of[id(other)] & above and not any(other is ok for ok in excused)
                   for other in broken):
                continue
            excused.append(job)
            changed = True
    return {str(job.get("name") or job.get("id")) for job in excused}


def setup_budget(job: Mapping[str, Any], now: dt.datetime, budget_seconds: int,
                 deadline: dt.datetime | None) -> float:
    """SETUP_WAIT_SECONDS, cut so a job that entered setup late is still judged before the watch ends (as
    job_budget() cuts a queued job's), and never below the queued budget."""
    if deadline is None:
        return SETUP_WAIT_SECONDS
    left = setup_seconds(job, now) + (deadline - now).total_seconds() - END_MARGIN_SECONDS
    return min(SETUP_WAIT_SECONDS, max(budget_seconds, left))


def assess(jobs: Sequence[Mapping[str, Any]], *, now: dt.datetime, budget_seconds: int,
           first_seen: Mapping[Any, dt.datetime] | None = None, deadline: dt.datetime | None = None,
           floor_seconds: int | None = None, allow_running_rescue: bool = False,
           leave_decided: bool = False, attempt_started_at: dt.datetime | None = None,
           reserved: bool = False, stranded: bool = False) -> Look:
    """One look at the jobs of a run on a persistent pool (each job's budget: job_budget()).

    With `leave_decided` (a pull request run), a stuck or held job is left queued once another job of
    the attempt has failed on its own (decided_failures()). With `reserved` (a nightly build), a job
    waiting for the one runner it names (reserved_runner()) is never stuck or held: only a refusal of
    it is acted on. With `stranded` (the run finished, not cancelled), a job no runner acquired
    (unacquired()) counts as refused.
    """
    seen = first_seen or {}
    waiting = [job for job in jobs if job_pool(job) and waiting_for_runner(job)]
    budgets = {id(job): job_budget(job, budget_seconds, deadline=deadline, floor_seconds=floor_seconds,
                                   first_seen=seen.get(job.get("id"))) for job in waiting}
    kept = [job for job in jobs if reserved and job_pool(job) and reserved_runner(job)
            and (waiting_for_runner(job) or in_setup(job))]
    stuck = [job for job in waiting if queued_seconds(job, now, seen.get(job.get("id"))) >= budgets[id(job)]
             and not any(job is other for other in kept)]
    names = ", ".join(sorted(str(job.get("name") or job.get("id")) for job in stuck))
    # Rescuing normally cancels the whole run, so a job already running on a
    # persistent runner would die with the stuck one and move to Blacksmith
    # too (#16463). cmux-next's explicit rescue window is the overflow
    # boundary: move the run as the bot once a side job waits past budget even
    # when another side job is still running. Manual reruns stay minis-first
    # through attempt 2; the bot's next rerun is the overflow.
    on_mini = [job for job in jobs if job_pool(job) and job.get("status") == "in_progress" and not in_setup(job)]
    # A job that failed on its own (not refused by the fleet) has already made the run red. Cancelling
    # and re-running it repeats that failure and puts the stuck job at the back of the same queue
    # (feat-cmux-next, 2026-10-20: 33 of 44 cmux-next pull request runs reached attempt 2 or 3, each
    # failing `checks` again), so the stuck job stays queued for the minis instead.
    decided = decided_failures(jobs, attempt_started_at) if leave_decided else []
    if stuck and not decided and (allow_running_rescue or not on_mini):
        suffix = (" while another owned job is running; overflowing the queued side jobs"
                  if on_mini else " with no runner")
        return Look("rescue", f"{names} queued on {job_pool(stuck[0])} for at least "
                              f"{min(budgets[id(job)] for job in stuck)}s{suffix}")
    settling = [job for job in jobs if job_pool(job) and in_setup(job)]
    held = [job for job in settling if setup_seconds(job, now) >= setup_budget(job, now, budget_seconds, deadline)
            and not any(job is other for other in kept)]
    # Cancelling the run would kill siblings still running (run 36198335113 lost five
    # shards that way to a refusal), so a job held in setup waits for them, as a
    # refusal does, until the watch is about to end.
    running = [job for job in jobs if job.get("status") == "in_progress" and not in_setup(job)]
    closing = deadline is not None and now >= deadline - dt.timedelta(seconds=END_MARGIN_SECONDS)
    if held and not decided and (not running or closing):
        names = ", ".join(sorted(str(job.get("name") or job.get("id")) for job in held))
        return Look("rescue", f"{names} waited in {job_pool(held[0])}'s runner setup for capacity for at least "
                              f"{SETUP_WAIT_SECONDS}s")
    turned_away = [job for job in jobs if refused(job)]
    if turned_away:
        names = ", ".join(sorted(str(job.get("name") or job.get("id")) for job in turned_away))
        return Look("refused", f"{names} refused by {job_pool(turned_away[0])} at job start or Xcode selection")
    lost = [job for job in jobs if stranded and unacquired(job)]
    if lost:
        names = ", ".join(sorted(str(job.get("name") or job.get("id")) for job in lost))
        return Look("refused", f"{names} cancelled by GitHub: no {job_pool(lost[0])} runner acquired it")
    if (stuck or held) and decided:
        waits = ", ".join(sorted(str(job.get("name") or job.get("id")) for job in stuck + held))
        return Look("watch", f"{waits} waiting on {job_pool((stuck or held)[0])}, but {', '.join(decided)} "
                             "already failed, so a re-run would repeat that failure; leaving it queued",
                    waiting=True)
    if kept:
        held_for = ", ".join(sorted(f"{job.get('name') or job.get('id')} ({reserved_runner(job)})" for job in kept))
        return Look("watch", f"{held_for} waiting for the runner reserved for it; never moved", waiting=True)
    if stuck:
        return Look("watch", f"{names} queued on {job_pool(stuck[0])} with no runner, but "
                             f"{len(on_mini)} job(s) of the run are running on a persistent runner", waiting=True)
    if waiting or settling:
        return Look("watch", f"{len(waiting)} job(s) waiting for a persistent runner, {len(settling)} in its setup",
                    waiting=True)
    return Look("watch", "no job is waiting for a persistent runner")


class Aborted(Exception):
    pass


def _headers(token: str) -> dict[str, str]:
    return {
        "Accept": "application/vnd.github+json",
        "Authorization": f"Bearer {token}",
        "X-GitHub-Api-Version": "2022-11-28",
        "User-Agent": "cmux-ci-owned-pool-rescue",
    }


# A request GitHub refuses for a rate limit waits and is tried again instead of ending the watch.
# A rescue that cancelled a run and then lost its re-run to a refusal left the run cancelled for
# good (2026-10-26: nightly-next run 40340830337 at e22f9384 was cancelled at 14:36:06Z and the
# watch "gave up: HTTP Error 403" at 14:37:38Z, so the tip never built). The repository's
# GITHUB_TOKENs share one installation budget and a secondary limit that sends no reset, so a
# refusal waits Retry-After, or until x-ratelimit-reset when the budget reads spent, or else
# 60, 120, 240, then 300 s; one job waits at most RATE_LIMIT_WAIT_SECONDS in all, which the
# job timeout keeps room for (JOB_TIMEOUT_SECONDS, ci-owned-pool-rescue.yml's timeout-minutes).
RATE_LIMIT_WAIT_SECONDS = _RATE_LIMIT_ROOM_SECONDS
RATE_LIMIT_FIRST_WAIT_SECONDS = 60
RATE_LIMIT_MAX_WAIT_SECONDS = 300


def rate_limit_wait(code: int, headers: Mapping[str, str], body: str, now: float, tries: int) -> float | None:
    """Seconds to wait before retrying a refused request, or None when it is not a rate limit.

    A permission 403 has no Retry-After, no spent budget and no "rate limit" in its message."""
    if code not in (403, 429):
        return None
    get = {str(key).lower(): str(value) for key, value in (headers or {}).items()}.get
    retry_after = (get("retry-after") or "").strip()
    remaining = (get("x-ratelimit-remaining") or "").strip()
    said = body.lower()
    # The same messages scripts/ci/gh_budget_wait.sh treats as a rate limit.
    limited = code == 429 or bool(retry_after) or remaining == "0" or any(
        phrase in said for phrase in ("rate limit", "abuse detection", "too many requests"))
    if not limited:
        return None
    if retry_after.isdigit():
        return max(float(retry_after), 1.0)
    reset = (get("x-ratelimit-reset") or "").strip()
    if remaining == "0" and reset.isdigit() and int(reset) > now:
        return int(reset) - now + 5
    return float(min(RATE_LIMIT_FIRST_WAIT_SECONDS << min(tries, 8), RATE_LIMIT_MAX_WAIT_SECONDS))


class GitHub:
    """The Actions API. Reads may use `read_token`, writes always use `token`.

    `read_token` is a manaflow-glaeda-route App installation token, so the
    watch's polling draws on the App's own 5000 requests an hour instead of
    the repository's GITHUB_TOKEN budget. Cancels and re-runs keep
    GITHUB_TOKEN: a re-run's triggering actor must stay github-actions[bot],
    which ci-macos.yml's attempt-2 routing checks. An installation token
    lasts an hour and a watch may outlive it, so a 401 on a read drops back to
    `token` for the rest of the watch. A 403 is a read the App may not make
    (branch_head needs contents, which it lacks): that one read uses `token`.
    """

    def __init__(self, token: str, repo: str, read_token: str = "") -> None:
        self.repo = repo
        self.headers = _headers(token)
        self.read_headers = _headers(read_token) if read_token else self.headers
        self.sleep: Callable[[float], None] = time.sleep
        # Each watch (the sweeper runs one per thread on this one client) has its own
        # RATE_LIMIT_WAIT_SECONDS, so one run's waits never spend another's.
        self._limits = threading.local()

    @property
    def rate_limit_waited(self) -> float:
        """This thread's waits for rate limits so far."""
        return getattr(self._limits, "waited", 0.0)

    @property
    def rate_limit_tries(self) -> int:
        return getattr(self._limits, "tries", 0)

    def no_rate_limit_waits(self) -> None:
        """This thread's requests raise on a rate limit instead of waiting (the sweeper's own
        listings, which run on its schedule and must not push its handover past the job's timeout)."""
        self._limits.waited = float(RATE_LIMIT_WAIT_SECONDS)

    def _wait_for_rate_limit(self, method: str, path: str, error: urllib.error.HTTPError) -> bool:
        """Wait out a GITHUB_TOKEN request's rate-limit refusal; False when it is not one or the budget is spent."""
        try:
            said = error.read().decode("utf-8", "replace")
        except Exception:  # noqa: BLE001 - the body only explains the refusal
            said = ""
        headers = dict(error.headers.items()) if error.headers is not None else {}
        wait = rate_limit_wait(error.code, headers, said, time.time(), self.rate_limit_tries)
        if wait is None:
            return False
        left = RATE_LIMIT_WAIT_SECONDS - self.rate_limit_waited
        if left < 1:
            return False
        wait = min(wait, left)
        self._limits.tries = self.rate_limit_tries + 1
        self._limits.waited = self.rate_limit_waited + wait
        shown = {key: value for key, value in headers.items()
                 if key.lower().startswith(("x-ratelimit", "retry-after", "x-github-request-id"))}
        print(f"::notice title=GitHub API budget::{method} {path} was refused for a rate limit "
              f"(HTTP {error.code}, {shown or 'no rate-limit headers'}: {said.strip()[:200]}); "
              f"waiting {wait:.0f}s ({self.rate_limit_waited - wait:.0f}s waited so far)",
              file=sys.stderr, flush=True)
        self.sleep(wait)
        return True

    def request(self, method: str, path: str, *, own_token: bool = False, body: Mapping | None = None,
                before_retry: Callable[[], None] | None = None) -> Any:
        headers = self.read_headers if method == "GET" and not own_token else self.headers
        data = json.dumps(body).encode() if body is not None else None
        request = urllib.request.Request(f"{API}/repos/{self.repo}{path}", method=method, headers=headers, data=data)
        try:
            with urllib.request.urlopen(request, timeout=20) as response:
                body = response.read()
                seen = getattr(response, "headers", None) or {}
                self.remaining = seen.get("X-RateLimit-Remaining") or self.remaining
                self.limit = seen.get("X-RateLimit-Limit") or self.limit
            self._limits.tries = 0
        except urllib.error.HTTPError as error:
            if headers is self.headers and error.code in (403, 429) and self._wait_for_rate_limit(method, path, error):
                # The same request again, after the wait (a refused request did nothing on GitHub's side),
                # unless what it acts on changed meanwhile (before_retry raises).
                if before_retry is not None:
                    before_retry()
                return self.request(method, path, own_token=own_token,
                                    body=json.loads(data) if data is not None else None, before_retry=before_retry)
            if error.code not in (401, 403, 429) or headers is self.headers:
                raise
            if error.code in (403, 429):
                # The installation lacks this read's permission, or its budget is spent: this one
                # read goes on GITHUB_TOKEN (which waits through a rate limit).
                return self.request(method, path, own_token=True)
            self.read_headers = self.headers
            return self.request(method, path)
        return json.loads(body) if body else None

    remaining = ""  # the token's requests left this hour, from the last response
    limit = ""  # and its hourly limit

    def marked_runs(self, name: str, count: int, oldest: dt.datetime | None = None, pages: int = 1,
                    log: Callable[[str], None] = lambda message: None) -> list[tuple[int, dt.datetime | None]]:
        """Runs with an artifact named `name`, by artifact id, newest first, with when each was uploaded (sweep()).

        Reads up to `pages` pages of `count`, stopping at a short page or at
        one whose last marker is older than `oldest` by MARKER_ORDER_SKEW:
        the listing is ordered by id, which trails upload time by up to that
        much. A page that cannot be read ends the listing with what came
        before it, unless it is the first.
        """
        found = []
        for page in range(1, pages + 1):
            try:
                data = self.request("GET", f"/actions/artifacts?name={name}&per_page={count}&page={page}")
            except READ_ERRORS:
                if page == 1:
                    raise
                log(f"could not read page {page} of the {name} markers; using the first {page - 1}")
                return found
            items = (data or {}).get("artifacts") or []
            for item in items:
                run_id = int(((item or {}).get("workflow_run") or {}).get("id") or 0)
                if run_id:
                    found.append((run_id, parse_time(item.get("created_at"))))
            last = parse_time(items[-1].get("created_at")) if items else None
            if len(items) < count or oldest is None or last is None or last < oldest - MARKER_ORDER_SKEW:
                return found
        if oldest is not None:
            log(f"read {pages} pages of {name} markers without reaching {oldest:%H:%M}; older ones wait for a later tick")
        return found

    def run(self, run_id: int) -> Mapping[str, Any]:
        return self.request("GET", f"/actions/runs/{run_id}")

    def owned_reruns(self, count: int) -> list[tuple[int, int]]:
        """(run id, attempt) of unfinished CI and cmux-next re-runs whose owned jobs go back to the minis
        (owned_rerun()): a re-run of failed jobs uploads no marker. GitHub lists a run as `queued` while a job of it waits for a
        runner, so both statuses are read."""
        found: list[tuple[int, int]] = []
        for workflow in (CI_WORKFLOW_PATH, CMUX_NEXT_WORKFLOW_PATH):
            name = workflow.rsplit("/", 1)[-1]
            for status in ("queued", "in_progress"):
                data = self.request("GET", f"/actions/workflows/{name}/runs?status={status}&per_page={count}")
                found += [(int(run["id"]), int(run["run_attempt"])) for run in (data or {}).get("workflow_runs") or []
                          if isinstance(run, Mapping) and run.get("id") and owned_rerun(run)]
        return found

    def failed_runs(self, count: int, oldest: dt.datetime, since: dt.datetime | None = None,
                    pages: int = 1) -> list[tuple[int, int]]:
        """(run id, attempt) of the failed runs of SIDE_WORKFLOW_PATHS created since `oldest` and, when
        `since` is given, finished after it: newest first, `count` a page, at most `pages` pages.

        Those workflows upload no marker (sweep()), and a job the fleet refused fails its run, so a
        refusal shows here once the run ends, which is when its failed jobs can be re-run.
        """
        created = oldest.astimezone(dt.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
        found: list[tuple[int, int]] = []
        for page in range(1, pages + 1):
            data = self.request("GET", f"/actions/runs?status=failure&exclude_pull_requests=true&per_page={count}"
                                       f"&page={page}&created=%3E%3D{created}")
            runs = [run for run in (data or {}).get("workflow_runs") or [] if isinstance(run, Mapping)]
            for run in runs:
                finished = parse_time(run.get("updated_at"))
                entry = (int(run.get("id") or 0), int(run.get("run_attempt") or 0))
                if entry[0] and entry not in found and run.get("path") in SIDE_WORKFLOW_PATHS and not (
                        since is not None and finished is not None and finished < since):
                    found.append(entry)
            # A status-filtered page can come back short before the last one (queue_janitor.GitHub.
            # in_flight_runs()), so only an empty page ends the listing before `pages`.
            if not runs:
                break
        return found

    def jobs(self, run_id: int, attempt: int) -> list[Mapping[str, Any]]:
        found: list[Mapping[str, Any]] = []
        for page in range(1, MAX_JOB_PAGES + 1):
            data = self.request("GET", f"/actions/runs/{run_id}/attempts/{attempt}/jobs?per_page=100&page={page}")
            batch = [job for job in (data or {}).get("jobs") or [] if isinstance(job, Mapping)]
            found.extend(batch)
            if len(batch) < 100:
                break
        return found

    def workflow_file(self, path: str, ref: str) -> str | None:
        """The text of `path` at commit `ref`, or None when it is not there."""
        try:
            data = self.request("GET", f"/contents/{urllib.parse.quote(path)}?ref={urllib.parse.quote(ref)}")
        except urllib.error.HTTPError as error:
            if error.code == 404:
                return None
            raise
        if not isinstance(data, Mapping) or data.get("encoding") != "base64":
            return None
        return base64.b64decode(str(data.get("content") or "")).decode("utf-8")

    def has_artifact(self, run_id: int, prefix: str, pages: int = 5) -> bool:
        """Whether the run uploaded an artifact whose name starts with `prefix`."""
        for page in range(1, pages + 1):
            data = self.request("GET", f"/actions/runs/{run_id}/artifacts?per_page=100&page={page}")
            names = [str(item.get("name") or "") for item in (data or {}).get("artifacts") or []]
            if any(name.startswith(prefix) for name in names):
                return True
            if len(names) < 100:
                return False
        return False

    def pull(self, number: int) -> Mapping[str, Any]:
        return self.request("GET", f"/pulls/{number}")

    def newer_unfinished_runs(self, path: str, run_id: int, branch: str) -> list[int]:
        """Ids of `path`'s runs on `branch` newer than `run_id` that wait behind it (one request).

        Pending runs only: nightly.yml's `full` group never cancels in
        progress, so a newer push, daily-schedule or full dispatch run waits
        there as `pending` while this run holds the group, and a re-run of this
        run would cancel it. The six-hourly cache seed runs in its own
        cancel-in-progress group and is never pending, so it does not count.
        (A seed-only or fast dispatch pending in its own group behind another
        of its kind still counts: a missed rescue, not a cancelled build.)
        """
        workflow = path.rsplit("/", 1)[-1]
        data = self.request("GET", f"/actions/workflows/{workflow}/runs?branch={branch}&per_page=20")
        return sorted(int(run.get("id") or 0) for run in (data or {}).get("workflow_runs") or []
                      if isinstance(run, Mapping) and int(run.get("id") or 0) > run_id
                      and run.get("status") == "pending")

    def newer_push_runs(self, path: str, run_id: int, branch: str) -> list[int]:
        """Ids of `path`'s push runs on `branch` newer than `run_id` and not cancelled (one request)."""
        workflow = path.rsplit("/", 1)[-1]
        data = self.request("GET", f"/actions/workflows/{workflow}/runs?branch={branch}&event=push&per_page=20")
        return sorted(int(run.get("id") or 0) for run in (data or {}).get("workflow_runs") or []
                      if isinstance(run, Mapping) and int(run.get("id") or 0) > run_id
                      and run.get("event") == "push" and run.get("conclusion") != "cancelled")

    def branch_head(self, branch: str) -> str:
        return str(((self.request("GET", f"/branches/{branch}") or {}).get("commit") or {}).get("sha") or "")

    def cancel(self, run_id: int) -> None:
        self.request("POST", f"/actions/runs/{run_id}/cancel")

    def force_cancel(self, run_id: int) -> None:
        self.request("POST", f"/actions/runs/{run_id}/force-cancel")

    def rerun(self, run_id: int, next_attempt: int) -> None:
        self.request("POST", f"/actions/runs/{run_id}/rerun", before_retry=self._still_finished(run_id, next_attempt))
        self.request_ui_tests(run_id, next_attempt)

    def rerun_failed(self, run_id: int, next_attempt: int) -> None:
        self.request("POST", f"/actions/runs/{run_id}/rerun-failed-jobs",
                     before_retry=self._still_finished(run_id, next_attempt))
        self.request_ui_tests(run_id, next_attempt)

    def _still_finished(self, run_id: int, next_attempt: int) -> Callable[[], None]:
        """A re-run sent again after a rate-limit wait only while the run is still the finished attempt it re-runs.

        Two watches can own one failed attempt (a cmux-next `completed` event's and the sweeper's): the
        other's re-run may have started, and even finished, during the wait, and sending this one then
        would start one attempt more than either meant (rescue() checked the attempt only before the wait).
        """
        def check() -> None:
            run = self.run(run_id) or {}
            attempt = int(run.get("run_attempt") or 0)
            if attempt != next_attempt - 1 or run.get("status") != "completed":
                raise Aborted(f"run {run_id} is at attempt {attempt} ({run.get('status') or 'unknown'}) after a "
                              f"rate-limit wait; someone else already re-ran it, so it is not re-run again")
        return check

    def request_ui_tests(self, run_id: int, attempt: int) -> None:
        """Start ci-ui-tests.yml for the attempt a re-run of a pull request's CI began.

        This token's re-run may emit no workflow_run event, and that attempt's
        ui-tests job waits for ci-ui-tests.yml (ui_tests_dispatch.rerun_dispatch()).
        Best effort: a failure here never stops the rescue.
        """
        try:
            run = self.request("GET", f"/actions/runs/{run_id}") or {}
            if run.get("path") != CI_WORKFLOW_PATH or run.get("event") != "pull_request":
                return
            # The caller's attempt: a read right after the re-run may still show the old one.
            path, body = ui_tests_dispatch.rerun_dispatch(run_id, attempt)
            self.request("POST", f"/{path}", body=body)
        except (urllib.error.URLError, OSError, ValueError) as error:
            print(f"::warning::could not start {ui_tests_dispatch.DISPATCH_WORKFLOW_FILE} for run {run_id}: {error}",
                  flush=True)


@dataclasses.dataclass
class Target:
    run_id: int
    attempt: int
    head_sha: str
    pr_number: int  # 0 for an E2E dispatch, which has no pull request
    e2e: bool = False  # a dispatch of DISPATCH_WORKFLOW_PATHS, watched as an E2E run
    path: str = CI_WORKFLOW_PATH
    # This attempt is a full re-run: `changes` runs again and picks a pool,
    # so it is watched the attempt-1 way (picker, then marker).
    full_rerun: bool = False
    side: bool = False  # a side-lane workflow (SIDE_WORKFLOW_PATHS): no picker job
    main: bool = False  # main's full-suite dispatch of ci.yml: no pull request, main's HEAD instead
    # A push or schedule run of nightly.yml on main (watched as a side lane,
    # against main's HEAD).
    nightly: bool = False
    # ci.yml started this watch because late-placement may move jobs onto owned
    # root runners after compile admission (LATE_PLACEMENT=1); the picker placed none.
    late: bool = False
    attempt_started_at: dt.datetime | None = None
    # A side lane's push run: its branch. Push runs of a side lane may each hold
    # their own concurrency group, so a newer push does not cancel this one.
    push_branch: str = ""
    # cmux-next's bot attempt 3 (REFUSAL_RESCUE_ATTEMPT): still on the minis, so a
    # refusal there goes to Blacksmith as attempt 4; a long queue is not moved.
    refusals_only: bool = False

    @property
    def picker_job(self) -> str:
        return E2E_PICKER_JOB if self.e2e else PICKER_JOB

    @property
    def late_owned_jobs(self) -> bool:
        """nightly-next's nightly.yml puts prepare-nightly-variant on the side label after the app build."""
        return self.nightly and self.push_branch == "nightly-next"

    @property
    def watch_limit(self) -> int:
        if self.late_owned_jobs:
            return NIGHTLY_WATCH_LIMIT_SECONDS
        if self.side:
            return SIDE_WATCH_LIMIT_SECONDS
        return E2E_WATCH_LIMIT_SECONDS if self.e2e else WATCH_LIMIT_SECONDS


def target_from_event(event: Mapping[str, Any], repository: str) -> Target | str:
    """The CI, E2E, iOS or nightly run to watch, or why this event is not one."""
    run = event.get("workflow_run") or {}
    path = run.get("path")
    if path == NIGHTLY_WORKFLOW_PATH:
        return nightly_target(run, repository)
    side = path in SIDE_WORKFLOW_PATHS
    if path != CI_WORKFLOW_PATH and path not in DISPATCH_WORKFLOW_PATHS and not side:
        return (f"started by {path or 'an unknown workflow'}, not {CI_WORKFLOW_PATH}, {NIGHTLY_WORKFLOW_PATH}, "
                f"a side-lane workflow or one of {', '.join(DISPATCH_WORKFLOW_PATHS)}")
    e2e = path in DISPATCH_WORKFLOW_PATHS
    attempt_started = parse_time(run.get("run_started_at") or run.get("created_at"))
    on_main = (path == CI_WORKFLOW_PATH and run.get("event") == "workflow_dispatch"
            and run.get("head_branch") == MAIN_BRANCH)
    # test-ios.yml also runs for pull requests: watched as an E2E run, but
    # against its pull request's head like a CI run.
    ios_pull = path == IOS_TEST_WORKFLOW_PATH and run.get("event") == "pull_request"
    expected = "workflow_dispatch" if e2e else "pull_request"
    # A side lane's trusted non-PR run (push, schedule, dispatch) is watched too.
    side_trusted = side and run.get("event") in TRUSTED_SIDE_EVENTS
    if run.get("event") != expected and not on_main and not ios_pull and not side_trusted:
        what = f"{expected} or a dispatch on {MAIN_BRANCH}" if path == CI_WORKFLOW_PATH else expected
        return f"a {run.get('event') or 'unknown'} run of {path}, not a {what}"
    head = (run.get("head_repository") or {}).get("full_name") or ""
    if head.casefold() != repository.casefold():
        return "a fork head; forks never take a persistent pool"
    attempt = int(run.get("run_attempt") or 0)
    if attempt != 1:
        return f"attempt {attempt}; its first attempt's watch follows it"
    if e2e and not ios_pull:
        return Target(int(run["id"]), attempt, str(run.get("head_sha") or ""), 0, e2e=True,
                      path=str(path), attempt_started_at=attempt_started)
    if on_main:
        return Target(int(run["id"]), attempt, str(run.get("head_sha") or ""), 0, path=str(path), main=True,
                      attempt_started_at=attempt_started)
    if side_trusted:
        # No pull request: no head to re-check (pull_moved), like a dispatch.
        push_branch = str(run.get("head_branch") or "") if run.get("event") == "push" else ""
        return Target(int(run["id"]), attempt, str(run.get("head_sha") or ""), 0, side=True, path=str(path),
                      attempt_started_at=attempt_started, push_branch=push_branch)
    pulls = [pr for pr in run.get("pull_requests") or [] if isinstance(pr, Mapping) and pr.get("number")]
    if len(pulls) != 1:
        return "the run does not name exactly one pull request"
    return Target(int(run["id"]), attempt, str(run.get("head_sha") or ""), int(pulls[0]["number"]),
                  e2e=e2e, side=side, path=str(path), attempt_started_at=attempt_started)


def completed_target(event: Mapping[str, Any], repository: str) -> Target | str:
    """A cmux-next run that finished red (workflow_run `completed`): the attempt that failed, any attempt.

    ci-owned-pool-rescue.yml wakes on a failed DEV daily dispatch or feat-cmux-next push of cmux-next.yml,
    so a refused DEV compile gets its own watch instead of waiting for the sweeper, whose listings run
    out of the hour's API budget. The attempt is resolved the sweeper's way (sweep_target()): a bot attempt
    3 is refusals only, and a bot attempt past it is on Blacksmith and left alone.
    """
    run = event.get("workflow_run") or {}
    if run.get("path") != CMUX_NEXT_WORKFLOW_PATH:
        return (f"a finished run of {run.get('path') or 'an unknown workflow'}; only {CMUX_NEXT_WORKFLOW_PATH} "
                "wakes a rescue when it finishes")
    return sweep_target(run, repository, late=False)


def nightly_target(run: Mapping[str, Any], repository: str) -> Target | str:
    """Attempt 1 of main push/schedule or nightly-next push, from this repository."""
    if run.get("event") not in NIGHTLY_EVENTS:
        return f"a {run.get('event') or 'unknown'} run of {NIGHTLY_WORKFLOW_PATH}, not a push or schedule"
    if run.get("head_branch") not in (MAIN_BRANCH, "nightly-next") or (run.get("head_branch") == "nightly-next" and run.get("event") != "push"):
        return f"a run of {NIGHTLY_WORKFLOW_PATH} on {run.get('head_branch') or 'an unknown branch'}, not a trusted nightly event"
    head = (run.get("head_repository") or {}).get("full_name") or ""
    if head.casefold() != repository.casefold():
        return "a fork head; forks never take a persistent pool"
    attempt = int(run.get("run_attempt") or 0)
    if attempt != 1:
        return f"attempt {attempt}; its first attempt's watch follows it"
    return Target(int(run["id"]), attempt, str(run.get("head_sha") or ""), 0, path=NIGHTLY_WORKFLOW_PATH,
                  side=True, nightly=True, push_branch=str(run["head_branch"]))


def marker_name(target: Target) -> str:
    """The marker's name up to its jobs and pool, which only the janitor reads."""
    return f"{MARKER_PREFIX}-{target.run_id}-{target.attempt}-"


def late_marker_name(target: Target) -> str:
    """The marker late-placement uploads when it moved jobs onto owned root runners."""
    return f"{LATE_MARKER_PREFIX}-{target.run_id}-{target.attempt}"


READ_ERRORS = (urllib.error.URLError, http.client.HTTPException, OSError, ValueError)


def read(call: Callable[[], Any], sleep: Callable[[float], None], log: Callable[[str], None]) -> Any:
    """A GET, retried: one transient error must not end the watch it exists for."""
    for attempt in range(1, READ_ATTEMPTS + 1):
        try:
            return call()
        except READ_ERRORS as error:
            if attempt == READ_ATTEMPTS:
                raise
            log(f"read failed ({error}); retrying")
            sleep(READ_RETRY_SECONDS * attempt)
    raise AssertionError("unreachable")


def watch(api: GitHub, target: Target, *, budget_seconds: int,
          now: Callable[[], dt.datetime], sleep: Callable[[float], None],
          log: Callable[[str], None], deadline: dt.datetime | None = None,
          floor_seconds: int | None = None) -> tuple[str, str]:
    """Watch until a stop, a rescue or `deadline`. Returns (outcome, reason).

    A job's budget is cut to end before `deadline`, never below
    `floor_seconds` (job_budget()).

    One deadline covers every attempt a job watches (main()), so attempt 2
    cannot stretch the job past its timeout.
    """
    if deadline is None:
        deadline = now() + dt.timedelta(seconds=target.watch_limit)
    sleep(FIRST_LOOK_SECONDS)
    looks = 0
    on_persistent = False
    picker_marker: bool | None = None
    first_seen: dict[Any, dt.datetime] = {}
    while True:
        looks += 1
        jobs = read(lambda: api.jobs(target.run_id, target.attempt), sleep, log)
        if not on_persistent and target.side:
            # No picker: a job that asks for an owned label is the choice. A
            # gated job (cloud-machine-tests) appears once its Linux gate ends.
            if any(job_pool(job) for job in jobs):
                on_persistent = True
                log("a side-lane job asked for a persistent pool")
            elif run_finished(jobs):
                return "stop", "no job of the run asked for a persistent pool"
            elif not jobs and read(lambda: api.run(target.run_id), sleep, log).get("status") == "completed":
                # A pending run cancelled before any job (a newer nightly took its place).
                return "stop", "the run finished with no job"
        elif not on_persistent and target.attempt > 1 and not target.full_rerun:
            # A re-run of failed jobs: no `changes` job, no marker. Follow it
            # if a job asks for an owned pool. When it runs compile admission
            # again, late-placement runs after it and may move the jobs after
            # admission onto one, so its marker decides; otherwise the first
            # look that lists jobs does.
            late = next((job for job in jobs if job.get("name") == LATE_JOB
                         and not carried(job, target.attempt_started_at)), None)
            if any(job_pool(job) for job in jobs):
                on_persistent = True
                log("a re-run job asked for a persistent pool")
            elif late is not None and late.get("status") == "completed":
                if late.get("conclusion") != "success" or not read(
                        lambda: api.has_artifact(target.run_id, late_marker_name(target)), sleep, log):
                    return "stop", "late placement moved no job onto a persistent pool"
                log("late placement moved jobs onto a persistent pool")
                on_persistent = True
            elif jobs and (run_finished(jobs) or late is None and not any(
                    job.get("name") == ADMISSION_JOB and not carried(job, target.attempt_started_at) for job in jobs)):
                return "stop", "no job of this attempt asked for a persistent pool"
        elif not on_persistent:
            if picker_finished(jobs, target.picker_job):
                if picker_marker is None:
                    picker_marker = bool(read(lambda: api.has_artifact(target.run_id, marker_name(target)),
                                              sleep, log))
                if picker_marker:
                    log("the picker chose a persistent pool")
                    on_persistent = True
                elif not target.late:
                    return "stop", "the run is on an ephemeral pool"
                elif picker_finished(jobs, LATE_JOB):
                    if not read(lambda: api.has_artifact(target.run_id, late_marker_name(target)), sleep, log):
                        return "stop", "late placement moved no job onto a persistent pool"
                    log("late placement moved jobs onto a persistent pool")
                    on_persistent = True
                elif run_finished(jobs):
                    return "stop", "the run finished on an ephemeral pool"
            elif run_finished(jobs):
                return "stop", "the run finished before the pool choice"
        # Waiting for compile admission and late placement: nothing can be stuck yet.
        interval = POLL_SECONDS if on_persistent or picker_marker is None else IDLE_POLL_SECONDS
        if on_persistent:
            run = read(lambda: api.run(target.run_id), sleep, log) if run_finished(jobs) else {}
            finished = run.get("status") == "completed"
            # A job no runner acquired looks like one a cancelled run left queued (unacquired()).
            stranded = finished and run.get("conclusion") not in (None, "cancelled")
            if finished and not any(refused(job) or stranded and unacquired(job) for job in jobs):
                return "stop", "the run finished"
            seen_at = now()
            for job in jobs:
                if job_pool(job) and waiting_for_runner(job):
                    first_seen.setdefault(job.get("id"), seen_at)
            look = assess(jobs, now=seen_at, budget_seconds=budget_seconds, first_seen=first_seen,
                          deadline=deadline, floor_seconds=floor_seconds,
                          allow_running_rescue=target.path == CMUX_NEXT_WORKFLOW_PATH,
                          # Main, nightly and push runs keep the rescue: a stuck one holds its
                          # concurrency group, and rescue() cancels it once its head has moved.
                          leave_decided=bool(target.pr_number) and not (
                              target.main or target.nightly or target.push_branch),
                          attempt_started_at=target.attempt_started_at,
                          reserved=target.nightly, stranded=stranded)
            if look.action == "refused" and not finished and seen_at < deadline and not target.main:
                # GitHub re-runs no job of a run still in progress (403 "already
                # running", for one job or the failed ones), and cancelling
                # the run to re-run it killed every healthy sibling (run
                # 36198335113: two refused app-host shards cost five running
                # shards and the CLI product tests, all re-run on Blacksmith).
                # Let the siblings finish; at the deadline, cancel as before.
                # Main's run still cancels at once: a run that ends in failure
                # makes ci-main-full-suite.yml open the red-CI issue before
                # the re-run starts, and a cancelled one does not.
                log(f"look {looks}: {look.reason}; waiting for the rest of the run to finish")
                sleep(IDLE_POLL_SECONDS)
                continue
            log(f"look {looks}: {look.reason}")
            if look.action in ("rescue", "refused"):
                return look.action, look.reason
            if not look.waiting:
                interval = IDLE_POLL_SECONDS
                owned = [job for job in jobs if job_pool(job)]
                # A re-run that runs its own admission (a full one, or a re-run
                # of a failed admission) makes its shards only after that
                # admission, so its watch goes on until the run finishes.
                own_admission = any(job.get("name") == ADMISSION_JOB
                                    and not carried(job, target.attempt_started_at) for job in jobs)
                if (target.attempt > 1 and not target.full_rerun and not own_admission and owned
                        and all(accepted(job, seen_at) for job in owned)):
                    # The fleet took the retry; later attempts never come back to it.
                    return "stop", "the fleet accepted the retry"
                if target.side and not target.late_owned_jobs and owned \
                        and all(accepted(job, seen_at) for job in owned):
                    # A side lane's jobs are all created by now, and none can be refused any more.
                    # Not nightly-next's: prepare-nightly-variant takes the side label only once
                    # build-nightly-app has finished, and its runner can refuse it then
                    # (run 40144317744), so that watch lasts until the run finishes.
                    return "stop", "the fleet accepted the side-lane jobs"
        if now() >= deadline:
            if target.nightly:
                # Never moved (assess(reserved=True)); say so, without claiming why it did not start.
                for job in jobs:
                    runner = reserved_runner(job)
                    if runner and job_pool(job) and (waiting_for_runner(job) or in_setup(job)):
                        log(f"::warning::{job.get('name') or job.get('id')} was still waiting for its reserved "
                            f"runner {runner} at the watch limit")
            return "stop", "watch limit reached"
        sleep(interval)


def bot_rerun_leaves_owned(path: str, attempt: int) -> bool:
    """Whether a bot's re-run of failed jobs of `attempt` sends a leg queued on an owned label elsewhere.

    queue_janitor.py's unclaimed-leg rule cancels a run whose owned leg no runner claimed within its bound
    and later re-runs its failed jobs (rerun-failed-jobs), so it acts only where that re-run is a rescue,
    as next_attempt() describes the routes:
    - a side lane other than cmux-next takes its Blacksmith default from attempt 2 on (cmux-tui.yml's
      macOS legs read `github.run_attempt == 1 && vars.CI_SIDE_LANE_RUNNER || 'blacksmith-...-macos-...'`);
    - cmux-next keeps the minis through REFUSAL_RESCUE_ATTEMPT and takes Blacksmith after it;
    - ci.yml's re-run of failed jobs reuses attempt 1's placement through LAST_OWNED_ATTEMPT;
    - test-ios.yml, ios-screenshots.yml, iroh-release-gate.yml and cmux-next-dev-daily.yml take their retry
      route from attempt 2;
      test-e2e.yml does not count, since a UI run's retry_label stays on its owned pool and the path alone
      cannot tell one apart;
    - the base and PR autofix's Mac exports (OWNED_EVERY_ATTEMPT_PATHS) keep their owned label on every
      attempt, so a re-run only rejoins the queue;
    - nightly, the UNRESCUED_WORKFLOW_PATHS and any workflow not watched here are a person's call.
    """
    following = attempt + 1
    if path in OWNED_EVERY_ATTEMPT_PATHS:
        return False
    if path == CMUX_NEXT_WORKFLOW_PATH:
        return following > REFUSAL_RESCUE_ATTEMPT
    if path == CI_WORKFLOW_PATH:
        return following > LAST_OWNED_ATTEMPT
    if path in DISPATCH_WORKFLOW_PATHS:
        return path != E2E_WORKFLOW_PATH and following > 1
    return path in SIDE_WORKFLOW_PATHS and following > 1


def next_attempt(target: Target) -> str:
    """Where a re-run of failed jobs goes next."""
    following = target.attempt + 1
    if target.path == CMUX_NEXT_WORKFLOW_PATH:
        if following <= LAST_OWNED_ATTEMPT:
            return f"attempt {following} stays on the side lane's owned label"
        if following > REFUSAL_RESCUE_ATTEMPT:
            return f"attempt {following} takes Blacksmith: a mini refused attempt {target.attempt}"
        return f"attempt {following} takes the side lane's Blacksmith overflow after the long mini queue"
    if target.side:
        return f"attempt {following} takes the side lane's Blacksmith default"
    if following <= LAST_OWNED_ATTEMPT and not target.e2e:
        return f"attempt {following} goes back to the owned labels, where the sweeper watches it"
    return f"attempt {following} takes retry_runner on Blacksmith"


def e2e_build_unfinished(api: GitHub, target: Target, sleep: Callable[[float], None],
                         log: Callable[[str], None]) -> bool:
    """An E2E run whose build job did not succeed, so its re-run compiles.

    A re-run of failed jobs keeps the `sibling` job's attempt-1 answer, taken
    before the refusal, so it never waits for a sibling that started compiling
    the same revision since: run 36168890047's attempt 2 compiled product
    8c48a10e beside run 36168944875. Re-running every job runs the Linux
    jobs and that wait again, which costs seconds. A build that passed is
    kept, as always.
    """
    if target.path != E2E_WORKFLOW_PATH:
        return False
    jobs = read(lambda: api.jobs(target.run_id, target.attempt), sleep, log)
    build = next((job for job in jobs if job.get("name") == "build"), None)
    return build is None or build.get("conclusion") != "success"


def pull_moved(api: GitHub, target: Target, sleep: Callable[[float], None],
               log: Callable[[str], None]) -> str:
    """Why the pull request (or main) no longer wants this run, or "" when it still does."""
    if target.nightly and target.push_branch == "nightly-next":
        # Each nightly-next run builds and signs in groups of its own (nightly.yml
        # scopes them to github.run_id), so a newer tip builds alongside this one
        # and never makes it moot: a newer tip is never a reason to cancel a
        # nightly (Nightly freshness). publish-nightly's order check refuses a
        # build no newer than the feed, so re-running an older one never downgrades.
        return ""
    if target.nightly:
        # nightly.yml's concurrency group never cancels in progress: a newer
        # run waits behind this one, and a re-run of this one would join the
        # group and cancel that pending run. Leave the revision to it.
        branch = target.push_branch or MAIN_BRANCH
        newer = read(lambda: api.newer_unfinished_runs(target.path, target.run_id, branch), sleep, log)
        if newer:
            return f"a newer nightly run on {branch} ({newer[0]}) has not finished and builds instead"
        # Not pending does not mean not superseded: a nightly-next push run is
        # never pending at the workflow level (its builds share a job-level
        # slot), and a newer main push may already be running or done. Once the
        # branch has moved past this commit and a newer push run covers it, a
        # re-run would only build (and publish) an older head on Blacksmith.
        pushes = read(lambda: api.newer_push_runs(target.path, target.run_id, branch), sleep, log)
        if pushes:
            head = read(lambda: api.branch_head(branch), sleep, log)
            if head and head != target.head_sha:
                return (f"{branch} has moved on to {head[:12]}, and its newer push run ({pushes[-1]}) "
                        "builds it instead")
        return ""
    if target.push_branch:
        # Not the branch head: a head commit outside the workflow's paths has no
        # run, and the newest push run that does cover the branch must finish.
        newer = read(lambda: api.newer_push_runs(target.path, target.run_id, target.push_branch), sleep, log)
        if newer:
            return f"a newer push run on {target.push_branch} ({newer[-1]}) covers the branch instead"
        return ""
    if (target.e2e or target.side) and not target.pr_number:
        return ""  # a dispatch or schedule has no head to move
    if target.main:
        head = read(lambda: api.branch_head(MAIN_BRANCH), sleep, log)
        if head != target.head_sha:
            return (f"{MAIN_BRANCH} has moved on, and ci-main-full-suite.yml dispatches its new HEAD "
                    "once this run completes")
        return ""
    pull = read(lambda: api.pull(target.pr_number), sleep, log)
    if pull.get("state") != "open":
        return "the pull request is closed"
    if (pull.get("head") or {}).get("sha") != target.head_sha:
        return "the pull request has a newer head, whose own run replaces this one"
    return ""


def rescue(api: GitHub, target: Target, *, now: Callable[[], dt.datetime], sleep: Callable[[float], None],
           log: Callable[[str], None], failed_only: bool = False,
           deadline: dt.datetime | None = None, refused: bool | None = None, refusal: bool = False) -> str:
    """Cancel and re-run, unless the pull request has moved on. Returns what happened.

    `failed_only` (a refused job) re-runs only the failed and cancelled jobs,
    keeping what passed, and needs no cancel when the run already finished.
    `refused` (default `failed_only`) is whether a run that already finished
    may be re-run: an E2E run stuck in the queue that then finished was
    likely cancelled by a newer dispatch, which re-running it would cancel.
    `refusal` is whether the fleet refused a job (not merely left it queued).
    """
    # Main's run is re-run after a refusal whether or not main moved: the
    # refusal is the fleet's, and a red run would open main's red-CI issue.
    # A stuck later attempt re-runs its failed jobs too, but is no refusal.
    keep_main = target.main and refusal
    moved = "" if keep_main else pull_moved(api, target, sleep, log)
    if moved and target.push_branch and target.path == CMUX_NEXT_WORKFLOW_PATH:
        # cmux-next.yml queues each push Mac job in a group per branch (one
        # runs, one waits, a newer push replaces only the waiting one), so a
        # newer push never makes this run cancel. Cancelling here killed every
        # tip run's running jobs at one push a minute (2026-10-07: no tip swift
        # test completed from 00:20Z to 01:05Z). Not re-run either: a re-run
        # would join the branch group and replace the newer run's waiting job.
        return f"not rescued: {moved}; a cmux-next push run is never cancelled for a newer push"
    if moved and (target.main or target.nightly or target.push_branch):
        # Main's stuck run holds its concurrency group, so nothing newer can
        # start until it finishes: cancel it, and its completion dispatches
        # the new HEAD. A superseded push run only holds a slot the newer run needs.
        run = read(lambda: api.run(target.run_id), sleep, log)
        if not run:
            return f"not rescued: the run could not be read ({moved})"
        if int(run.get("run_attempt") or 0) != target.attempt:
            return "not rescued: someone else already re-ran the run"
        if run.get("status") == "completed":
            return f"not rescued: {moved}"
        api.cancel(target.run_id)
        return f"cancelled run {target.run_id}, not re-run: {moved}"
    if moved:
        return f"not rescued: {moved}"
    run = read(lambda: api.run(target.run_id), sleep, log)
    if int(run.get("run_attempt") or 0) != target.attempt:
        return "not rescued: someone else already re-ran the run"
    if run.get("status") != "completed" and target.pr_number:
        # pull_moved() just read the pull request open at this head. Run 40398277457 (PR #25446)
        # attempt 1 was cancelled here and re-run in-process; nothing automatic cancels such a run now.
        return f"not rescued: PR #{target.pr_number} is open at the run's head; {OPEN_HEAD_RULE}"
    if run.get("status") != "completed" and deadline is not None and \
            (deadline - now()).total_seconds() < CANCEL_WAIT_SECONDS + RERUN_MARGIN_SECONDS + max(
                0.0, RATE_LIMIT_WAIT_SECONDS - getattr(api, "rate_limit_waited", RATE_LIMIT_WAIT_SECONDS)):
        # (The rate-limit waits this watch may still make count too: they do not shorten the settle.)
        # A job killed between the cancel and the re-run would leave the
        # pull request's run cancelled for good; leave it as GitHub has it.
        return "not rescued: too little of the job left to cancel and re-run"
    if run.get("status") == "completed":
        should_rerun = failed_only if refused is None else refused
        if not should_rerun:
            return "not rescued: the run already finished"
        if not failed_only:
            api.rerun(target.run_id, target.attempt + 1)
            return f"re-ran every job of run {target.run_id}; {next_attempt(target)}"
        if e2e_build_unfinished(api, target, sleep, log):
            api.rerun(target.run_id, target.attempt + 1)
            return f"re-ran every job of run {target.run_id}, so its sibling wait runs again; {next_attempt(target)}"
        api.rerun_failed(target.run_id, target.attempt + 1)
        return f"re-ran the failed jobs of run {target.run_id}; {next_attempt(target)}"
    api.cancel(target.run_id)
    log(f"cancelled run {target.run_id}")
    started = now()
    # Time spent waiting through rate limits (GitHub.request) is not the run's settling time:
    # counting it could end the settle window before the re-run (the job timeout has room for it).
    limited_at_cancel = getattr(api, "rate_limit_waited", 0.0)
    forced_at: float | None = None
    while True:
        sleep(10)
        run = read(lambda: api.run(target.run_id), sleep, log)
        if int(run.get("run_attempt") or 0) != target.attempt:
            return "not rescued: someone else already re-ran the run"
        if run.get("status") == "completed":
            break
        waited = (now() - started).total_seconds() - (getattr(api, "rate_limit_waited", 0.0) - limited_at_cancel)
        if (forced_at is None and waited >= FORCE_CANCEL_AFTER_SECONDS) or \
                (forced_at is not None and waited - forced_at >= FORCE_CANCEL_AGAIN_SECONDS):
            forced_at = waited
            try:
                api.force_cancel(target.run_id)
                log(f"force-cancelled run {target.run_id} ({round(waited)}s after cancel)")
            except urllib.error.HTTPError as error:
                # Most likely the run settled since the read; the next read
                # sees it. Aborting here would leave it cancelled for good.
                log(f"force-cancel of run {target.run_id} refused ({error.code}); still waiting")
        if waited >= CANCEL_WAIT_SECONDS:
            raise Aborted(f"run {target.run_id} did not finish {CANCEL_WAIT_SECONDS}s after cancel; not re-run")
    # A push during the cancel starts the new head's run; re-running the old
    # head now would join its concurrency group and cancel it.
    moved = "" if keep_main else pull_moved(api, target, sleep, log)
    if moved:
        return f"cancelled but not re-run: {moved}"
    if failed_only:
        if e2e_build_unfinished(api, target, sleep, log):
            api.rerun(target.run_id, target.attempt + 1)
            return f"re-ran every job of run {target.run_id}, so its sibling wait runs again; {next_attempt(target)}"
        api.rerun_failed(target.run_id, target.attempt + 1)
        return f"re-ran the failed jobs of run {target.run_id}; {next_attempt(target)}"
    api.rerun(target.run_id, target.attempt + 1)
    if target.attempt + 1 <= LAST_OWNED_ATTEMPT and not target.e2e:
        return (f"re-ran run {target.run_id}; attempt {target.attempt + 1} picks again, the owned machines "
                "free now first, and the sweeper watches it")
    return f"re-ran run {target.run_id}; attempt {target.attempt + 1} takes an ephemeral pool"


def gave_up(target: Target, error: BaseException) -> str:
    """The give-up line; for a nightly an error annotation, since its run may now sit cancelled.

    nightly-next-tick.yml re-runs a nightly-next tip whose newest run is cancelled, so the
    tip still builds; the annotation says why it needed to."""
    text = f"gave up: {error}"
    if target.nightly:
        text = (f"::error title=nightly rescue::run {target.run_id} {text}; if it was cancelled, "
                "the nightly-next tick re-runs it (nightly-next-tick.yml)")
    return text


def main(argv: Sequence[str] | None = None, env: Mapping[str, str] | None = None, *,
         api: GitHub | None = None, now: Callable[[], dt.datetime] | None = None,
         sleep: Callable[[float], None] = time.sleep) -> int:
    env = os.environ if env is None else env
    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    parser.parse_args(argv)
    clock = now or (lambda: dt.datetime.now(dt.timezone.utc))
    lines: list[str] = []

    def log(text: str) -> None:
        print(text, flush=True)
        lines.append(text)

    def finish(outcome: str) -> int:
        log(outcome)
        if env.get("GITHUB_STEP_SUMMARY"):
            with open(env["GITHUB_STEP_SUMMARY"], "a", encoding="utf-8") as handle:
                handle.write("### Persistent-pool rescue\n\n" + "\n".join(f"- {line}" for line in lines) + "\n")
        return 0

    if (env.get("POOL_OWNED") or "").strip() != "1":
        return finish("owned pools are off (CI_PR_POOL_OWNED is not 1); nothing to watch")
    seconds = budget(env.get("RESCUE_SECONDS"))
    if seconds is None:
        return finish(f"CI_OWNED_POOL_RESCUE_SECONDS must be {MIN_BUDGET_SECONDS} to {MAX_BUDGET_SECONDS}; "
                      "nothing to watch")
    side_seconds = side_budget(env.get("SIDE_RESCUE_SECONDS"))
    if side_seconds is None:
        return finish(f"SIDE_RESCUE_SECONDS must be {MIN_BUDGET_SECONDS} to {MAX_BUDGET_SECONDS}; "
                      "nothing to watch")
    repository = env.get("GITHUB_REPOSITORY") or ""
    client = api or GitHub(env.get("GH_TOKEN") or env.get("GITHUB_TOKEN") or "", repository,
                           read_token=env.get("READ_TOKEN") or "")
    if (env.get("SWEEP") or "").strip() == "1":
        # One job for every marked run (sweep()); its per-run lines go to the log only.
        outcomes = sweep(client, repository, seconds=seconds, queue_rounds=env.get("QUEUE_ROUNDS"),
                         now=clock, log=lambda text: print(text, flush=True), side_seconds=side_seconds)
        return finish("swept: " + (", ".join(f"{count} {outcome}" for outcome, count in sorted(outcomes.items()))
                                   or "no run needed a watch"))
    run_id = (env.get("WATCH_RUN_ID") or "").strip()
    if run_id:
        # Dispatched by the picker's job: read the run it names and check it
        # exactly as a workflow_run event's run would be.
        if not run_id.isdigit():
            return finish(f"not watched: run id {run_id!r} is not a number")
        try:
            event = {"workflow_run": read(lambda: client.run(int(run_id)), sleep, log)}
        except READ_ERRORS as error:
            finish(f"gave up: could not read run {run_id}: {error}")
            return 1
    else:
        with open(env["GITHUB_EVENT_PATH"], encoding="utf-8") as handle:
            event = json.load(handle)
    target = completed_target(event, repository) if event.get("action") == "completed" else \
        target_from_event(event, repository)
    if isinstance(target, str):
        return finish(f"not watched: {target}")
    if ((env.get("LATE_PLACEMENT") or "").strip() == "1" and target.attempt == 1
            and not (target.e2e or target.main or target.side)):
        target = dataclasses.replace(target, late=True)
    try:
        return finish(follow(client, target, seconds=seconds, queue_rounds=env.get("QUEUE_ROUNDS"),
                             now=clock, sleep=sleep, log=log, side_seconds=side_seconds))
    except (*READ_ERRORS, Aborted) as error:
        # A failed watch leaves the run exactly as GitHub scheduled it.
        finish(gave_up(target, error))
        return 1


def follow(client: GitHub, target: Target, *, seconds: int, queue_rounds: str | None,
           now: Callable[[], dt.datetime], sleep: Callable[[float], None], log: Callable[[str], None],
           rescue_sleep: Callable[[float], None] | None = None, latest: dt.datetime | None = None,
           light_retry: bool = False, side_seconds: int | None = None) -> str:
    """Watch one run and rescue it when it needs it. Returns the outcome; raises READ_ERRORS or Aborted.

    The sweeper stops a watch by making `sleep` raise; a rescue paces itself
    with `rescue_sleep` (default `sleep`) so a cancel it started is always
    followed by its re-run, and `latest` caps the rescue deadline at the
    sweeper's end plus its grace.
    """
    rescue_sleep = rescue_sleep or sleep
    clock = now
    effective_seconds = side_seconds if target.path == CMUX_NEXT_WORKFLOW_PATH and side_seconds is not None else seconds

    def capped(deadline: dt.datetime) -> dt.datetime:
        return min(deadline, latest) if latest is not None else deadline

    subject = (f"pull request #{target.pr_number}'s {target.path}" if target.pr_number else
               "an E2E dispatch" if target.path == E2E_WORKFLOW_PATH else f"a dispatch of {target.path}") \
        if target.e2e else f"main's full-suite dispatch at {target.head_sha[:12]}" if target.main \
        else f"{target.push_branch or MAIN_BRANCH}'s nightly build at {target.head_sha[:12]}" if target.nightly \
        else f"pull request #{target.pr_number}" if target.pr_number else f"a run of {target.path}"
    if target.side and not target.nightly:
        subject += " (side lane)"
    # ci.yml's, test-ios.yml's and test-e2e.yml's pickers queue on purpose, within the queue
    # rounds: their owned jobs may wait up to the pool's expected wait (see the docstring).
    # A nightly build's wait for its reserved runner is never stuck (reserved_runner()), so it needs no
    # allowance here; a nightly job without one gets the configured budget like any side lane.
    queue_extra = queue_seconds(queue_rounds) if target.path in QUEUEING_WORKFLOW_PATHS else 0
    log(f"watching run {target.run_id} of {subject} (budget {effective_seconds + queue_extra}s"
        + (f": {effective_seconds}s past the {queue_extra}s an owned job may expect to wait)" if queue_extra else ")"))
    # A rescue may run past the watch deadline, within the job's own timeout,
    # so a cancel is never started without the time to settle and re-run.
    started = clock()
    deadline = started + dt.timedelta(seconds=target.watch_limit)
    rescue_deadline = capped(deadline + dt.timedelta(seconds=RESCUE_GRACE_SECONDS))
    # Every attempt gets the queue allowance: a re-run's owned jobs queue on the owned labels like attempt 1's.
    first_budget = effective_seconds + queue_extra
    outcome, reason = watch(client, target, budget_seconds=first_budget, now=clock, sleep=sleep,
                            log=log, deadline=deadline, floor_seconds=effective_seconds)
    if outcome not in ("rescue", "refused"):
        return f"stopped: {reason}"
    if target.refusals_only and outcome == "rescue":
        # A refusals-only cmux-next attempt deliberately stays on its owned
        # label while it waits for the hook to refuse the job. A queue alarm
        # is not the refusal: keep the watch alive until the job fails or the
        # run finishes, otherwise the later refusal has no rescue owner.
        remaining = max(0, int((deadline - clock()).total_seconds()))
        if remaining == 0:
            return f"stopped: {reason}; attempt {target.attempt} stays on the minis (refusals only)"
        log(f"queue alarm while refusals-only; continuing until refusal or run finish ({remaining}s left)")
        outcome, reason = watch(client, target, budget_seconds=max(effective_seconds, remaining),
                                now=clock, sleep=sleep, log=log, deadline=deadline,
                                floor_seconds=effective_seconds)
    if target.refusals_only and outcome != "refused":
        return f"stopped: {reason}; attempt {target.attempt} stays on the minis (refusals only)"
    refusal_jobs: list[Mapping[str, Any]] | None = None
    dev_compile = False
    if outcome == "refused" and target.path == CMUX_NEXT_WORKFLOW_PATH:
        refusal_jobs = read(lambda: client.jobs(target.run_id, target.attempt), sleep, log)
        dev_compile = any(str(job.get("name") or "") in {
            "cmux NEXT DEV Release compile", "cmux-scheme-compile"
        } for job in refusal_jobs)
    # A refusal re-runs the run's failed jobs. A side lane (found by the failed-run listing, sweep()) or a
    # refusals-only run that also failed on its own would re-run that failure on the minis too, so it is
    # left red; a nightly keeps its re-run (its build is the product).
    if outcome == "refused" and (target.refusals_only or target.side and not target.nightly):
        jobs = refusal_jobs if refusal_jobs is not None else read(
            lambda: client.jobs(target.run_id, target.attempt), sleep, log)
        other = [str(job.get("name")) for job in jobs if job.get("conclusion") in ("failure", "timed_out")
                 and not carried(job, target.attempt_started_at) and not refused(job)]
        if other:
            # An `if: always()` aggregator that failed only because a job it needs was refused is
            # re-run with that job (downstream_of_refusals()); its `needs` come from the run's own file.
            excused: set[str] = set()
            try:
                text = read(lambda: client.workflow_file(target.path, target.head_sha), sleep, log)
                if text:
                    excused = downstream_of_refusals(text, jobs, target.attempt_started_at)
                else:
                    log(f"{target.path} is not at {target.head_sha[:12]}; no failure is put down to a refusal")
            except Aborted:
                raise  # the sweeper is handing over (Stopping)
            except Exception as error:  # noqa: BLE001 - an unreadable or invalid file only excuses nothing
                log(f"could not read {target.path}'s needs at {target.head_sha[:12]} "
                    f"({type(error).__name__}: {error}); no failure is put down to a refusal")
            if excused:
                log(f"{', '.join(sorted(excused))} failed only because a job it needs was refused; "
                    "the re-run of failed jobs re-runs it too")
            other = [name for name in other if name not in excused]
        if other:
            # A re-run of failed jobs would re-run these too: a real failure is never retried here.
            return (f"not rescued: attempt {target.attempt} also failed for a reason that is not a refusal "
                    f"({', '.join(sorted(other))})")
    log(f"{'rescue' if outcome == 'rescue' else 'refused'}: {reason}")
    while True:
        # From attempt 2 on, keep what passed: only the owned jobs are moved.
        # An E2E run always keeps what passed (see the module docstring).
        failed_only = (outcome == "refused" or target.attempt > 1 or target.e2e or target.side) and not dev_compile
        result = rescue(client, target, now=clock, sleep=rescue_sleep, log=log, failed_only=failed_only,
                        deadline=rescue_deadline,
                        refused=(outcome == "refused") if target.e2e or target.side else None,
                        refusal=outcome == "refused")
        log(result)
        # Only an E2E full re-run picks a new runner in this watch. Other CI
        # re-runs are resumed by the sweeper through owned_reruns().
        if not target.e2e or not result.startswith("re-ran every job") or target.attempt + 1 > LAST_OWNED_ATTEMPT:
            return "done"
        full_rerun = True
        target = dataclasses.replace(target, attempt=target.attempt + 1, full_rerun=full_rerun, late=False)
        deadline = min(clock() + dt.timedelta(seconds=target.watch_limit), started + dt.timedelta(
            seconds=JOB_TIMEOUT_SECONDS - RESCUE_GRACE_SECONDS - JOB_TIMEOUT_MARGIN_SECONDS))
        rescue_deadline = capped(deadline + dt.timedelta(seconds=RESCUE_GRACE_SECONDS))
        outcome, reason = watch(client, target, budget_seconds=effective_seconds, now=clock, sleep=sleep, log=log,
                                deadline=deadline)
        if outcome not in ("rescue", "refused"):
            return f"stopped watching attempt {target.attempt}: {reason}"
        log(f"attempt {target.attempt}: {'rescue' if outcome == 'rescue' else 'refused'}: {reason}")


# The sweeper (SWEEP=1). The pickers of ci.yml, test-e2e.yml and test-ios.yml
# upload an artifact named WATCH_MARKER when they place attempt 1 on an owned
# pool, and ci-macos.yml's late-placement uploads LATE_WATCH_MARKER when it
# moves jobs onto one. Listing each name repository-wide is one request that
# names every such run, finished or not, so one job watches them all: each
# run gets a thread running follow(), the per-run watch. The side lanes,
# which upload no marker, are found by one listing of failed runs a tick
# (GitHub.failed_runs()).
WATCH_MARKER = "owned-pool-watch"
LATE_WATCH_MARKER = "owned-pool-watch-late"
# cmux-next.yml's placement uploads this instead of WATCH_MARKER when it found
# no free Blacksmith slot: a re-run would only rejoin the same owned queue, so
# the run (and every re-run of it) is watched for refusals only (run
# 39615562861 was cancelled and re-run twice while it waited, 2026-10-21).
REFUSALS_WATCH_MARKER = "owned-pool-watch-refusals"
SWEEP_TICK_SECONDS = 60
# A sweeper adopts runs this long, then stops its watches and gives any rescue
# it started RESCUE_GRACE_SECONDS. ci-owned-pool-rescue.yml's two-hourly cron
# queues the next one in the concurrency group, so it starts as this one
# ends, and one dropped cron still leaves one queued.
SWEEP_SECONDS = 5 * 60 * 60
# Older runs are left alone: their watch would be past its limit. A run that
# finished (a refusal) during a handover is still inside it, so the next
# sweeper re-runs it.
SWEEP_MAX_AGE_SECONDS = E2E_WATCH_LIMIT_SECONDS
# A marker listing page covers this many runs, newest first by artifact id.
# One page held about two hours of owned placements on 2026-09-25 but only one
# on 2026-09-27 (87 an hour), and an artifact's id trails its upload time by
# up to 78 minutes (measured over 400 markers that day): six markers uploaded
# 11:51 to 12:15 took ids among markers an hour older, never reached page
# one, and two of their runs failed unrescued. So the sweeper pages back until
# a page ends MARKER_ORDER_SKEW past SWEEP_MAX_AGE_SECONDS, at most
# SWEEP_LISTING_PAGES.
SWEEP_LISTING = 100
SWEEP_LISTING_PAGES = 5
# The failed side-lane runs are listed every this many ticks: a refused job is re-run within about
# that many minutes, for at most SWEEP_LISTING_PAGES requests each time.
SIDE_FAILURE_TICKS = 5
MARKER_ORDER_SKEW = dt.timedelta(minutes=90)


class Stopping(Aborted):
    pass


# A run that finished this long before a sweeper started is left alone: the
# sweeper before it was running then and has already acted on it.
SWEEP_FINISHED_SECONDS = 30 * 60


def sweep_target(run: Mapping[str, Any], repository: str, *, late: bool, full_rerun: bool = False,
                 since: dt.datetime | None = None) -> Target | str:
    """The target for a marked run, resuming the attempt its rescue re-ran (LAST_OWNED_ATTEMPT at most).

    A finished run is only worth a look when it failed after `since`: a
    refusal nobody re-ran yet. `full_rerun` says the re-run ran the picker
    again (a stuck run's re-run, or a person's full one), so its own markers
    decide, the picker's and then late-placement's.
    """
    attempt = int(run.get("run_attempt") or 0)
    if attempt < 1 or attempt > LAST_OWNED_ATTEMPT and not owned_rerun(run):
        return f"attempt {attempt}"
    if run.get("status") == "completed":
        if run.get("conclusion") != "failure":
            return f"finished ({run.get('conclusion')})"
        finished = parse_time(run.get("updated_at"))
        if since is not None and finished is not None and finished < since:
            return "finished before this sweeper's predecessor stopped"
    target = target_from_event({"workflow_run": {**run, "run_attempt": 1}}, repository)
    if isinstance(target, str):
        return target
    if attempt > 1:
        # A re-run (a rescue, the failure attribution, or a person): follow its owned jobs, if any.
        actor = str((run.get("triggering_actor") or {}).get("login") or "")
        return dataclasses.replace(target, attempt=attempt, full_rerun=full_rerun,
                                   late=full_rerun and not (target.e2e or target.main or target.side),
                                   refusals_only=(run.get("path") == CMUX_NEXT_WORKFLOW_PATH and actor in RESCUE_ACTORS
                                                  and attempt == REFUSAL_RESCUE_ATTEMPT))
    if late and not (target.e2e or target.main or target.side):
        return dataclasses.replace(target, late=True)
    return target


def sweep(client: GitHub, repository: str, *, seconds: int, queue_rounds: str | None,
          now: Callable[[], dt.datetime], log: Callable[[str], None],
          sweep_seconds: int = SWEEP_SECONDS, tick_seconds: float = SWEEP_TICK_SECONDS,
          wait: Callable[[float], None] = time.sleep, light_retry: bool = False,
          side_seconds: int | None = None) -> dict[str, int]:
    """Watch every marked run until `sweep_seconds` pass. Returns outcome counts."""
    # The sweeper's own listings raise on a rate limit as before (a failed read is tried next
    # tick); only the watch threads, which cancel and re-run, wait (RESCUE_GRACE_SECONDS has room).
    if hasattr(client, "no_rate_limit_waits"):
        client.no_rate_limit_waits()
    stopping = threading.Event()
    lock = threading.Lock()
    outcomes: dict[str, int] = {}
    seen: set[int] = set()
    rerun_seen: set[tuple[int, int]] = set()  # re-runs, once per attempt (owned_reruns())
    failed_seen: set[tuple[int, int]] = set()  # failed side-lane runs, once per attempt (failed_runs())
    ticks = 0
    refusals_runs: set[int] = set()  # runs with REFUSALS_WATCH_MARKER, re-runs included
    threads: list[threading.Thread] = []
    # The watch thread of each adopted run: a side lane's watch stops once the fleet accepted its jobs,
    # and a runner stopped after that fails the run later, so the failed-run listing takes it up again.
    watching: dict[int, threading.Thread] = {}
    started = now()
    latest = started + dt.timedelta(seconds=sweep_seconds + RESCUE_GRACE_SECONDS)
    since = started - dt.timedelta(seconds=SWEEP_FINISHED_SECONDS)

    def watch_sleep(delay: float) -> None:
        if stopping.wait(delay):
            raise Stopping("the sweeper is handing over")

    def one(target: Target) -> None:
        def say(text: str) -> None:
            log(f"[run {target.run_id}] {text}")
        try:
            outcome = follow(client, target, seconds=seconds, queue_rounds=queue_rounds,
                             now=now, sleep=watch_sleep, log=say, rescue_sleep=wait, latest=latest,
                             light_retry=light_retry, side_seconds=side_seconds)
        except Stopping:
            outcome = "handed over"
        except (*READ_ERRORS, Aborted) as error:
            outcome = "gave up"
            say(f"gave up: {error}")
            if target.nightly:
                log(gave_up(target, error))  # unprefixed, so the runner reads the annotation
        except Exception as error:  # noqa: BLE001 - one run's bug must not end every other watch
            outcome = "error"
            say(f"error: {type(error).__name__}: {error}")
        else:
            say(outcome)
        with lock:
            key = outcome.split(":")[0]
            outcomes[key] = outcomes.get(key, 0) + 1

    def adopt(run_id: int, late: bool, *, refusals_known: bool = True) -> bool:
        """Watch the run if it needs it. False when it could not be judged yet (a read failed): try later."""
        was_seen = run_id in seen
        seen.add(run_id)
        try:
            run = read(lambda: client.run(run_id), wait, log)
        except READ_ERRORS as error:
            seen.discard(run_id)  # the next tick tries again
            log(f"[run {run_id}] could not read the run ({error})")
            return False
        if not refusals_known and run.get("path") == CMUX_NEXT_WORKFLOW_PATH:
            # Without this tick's refusals listing a cmux-next re-run could lose its mark.
            if not was_seen:
                seen.discard(run_id)
            log(f"[run {run_id}] not adopted yet: the refusals markers were not listed; next tick")
            return False
        rerun_seen.add((run_id, int(run.get("run_attempt") or 0)))
        full_rerun = False
        if int(run.get("run_attempt") or 0) > 1:
            # A full re-run ran the picker again; a re-run of failed jobs kept attempt 1's.
            try:
                jobs = read(lambda: client.jobs(run_id, int(run["run_attempt"])), wait, log)
            except READ_ERRORS as error:
                seen.discard(run_id)
                log(f"[run {run_id}] could not read attempt {run['run_attempt']} ({error})")
                return False
            picker = E2E_PICKER_JOB if run.get("path") in DISPATCH_WORKFLOW_PATHS else PICKER_JOB
            attempt_started = parse_time(run.get("run_started_at") or run.get("created_at"))
            full_rerun = any(job.get("name") == picker and int(job.get("run_attempt") or 0) > 1
                             and not carried(job, attempt_started) for job in jobs)
        target = sweep_target(run, repository, late=late, full_rerun=full_rerun, since=since)
        if isinstance(target, str):
            log(f"[run {run_id}] not watched: {target}")
            return True
        if run_id in refusals_runs:
            target = dataclasses.replace(target, refusals_only=True)
        thread = threading.Thread(target=one, args=(target,), name=f"run-{run_id}", daemon=True)
        thread.start()
        threads.append(thread)
        watching[run_id] = thread
        return True

    while (now() - started).total_seconds() < sweep_seconds:
        oldest = now() - dt.timedelta(seconds=SWEEP_MAX_AGE_SECONDS)
        # The picker's marker first: a run with both is watched the ordinary way.
        # Refusals-only runs are listed before the re-runs, which inherit the mark.
        refusals_listed = False
        for name, late in ((REFUSALS_WATCH_MARKER, False), (WATCH_MARKER, False), (LATE_WATCH_MARKER, True)):
            try:
                marked = client.marked_runs(name, SWEEP_LISTING, oldest, SWEEP_LISTING_PAGES, log=log)
            except READ_ERRORS as error:
                log(f"could not list {name} markers ({error}); next tick")
                continue
            refusals_listed = refusals_listed or name == REFUSALS_WATCH_MARKER
            for run_id, created in marked:
                if name == REFUSALS_WATCH_MARKER:
                    refusals_runs.add(run_id)  # at any age: a later re-run inherits it
                if run_id in seen or (created is not None and created < oldest):
                    continue
                adopt(run_id, late)
        # A re-run of failed jobs goes back to the minis with no marker of its own; two listings a tick.
        try:
            reruns = client.owned_reruns(SWEEP_LISTING)
        except READ_ERRORS as error:
            log(f"could not list re-runs ({error}); next tick")
            reruns = []
        for run_id, attempt in reruns:
            if (run_id, attempt) not in rerun_seen:
                adopt(run_id, False, refusals_known=refusals_listed)
        # A side lane uploads no marker: its run is found once a refused job has failed it, every
        # SIDE_FAILURE_TICKS ticks (paging through every workflow's failures costs a few requests).
        failed: list[tuple[int, int]] = []
        if ticks % SIDE_FAILURE_TICKS == 0:
            try:
                failed = client.failed_runs(SWEEP_LISTING, oldest, since, SWEEP_LISTING_PAGES)
            except READ_ERRORS as error:
                log(f"could not list failed runs ({error}); next tick")
                ticks -= 1  # try again on the next tick
        ticks += 1
        for run_id, attempt in failed:
            # A marked run's own watch (or the re-run listing) already follows it. One whose watch has
            # ended (it stopped once the fleet accepted the jobs, before a runner was stopped mid-job)
            # is taken up again, once per attempt.
            ended = run_id in watching and not watching[run_id].is_alive()
            followed = (run_id, attempt) in rerun_seen or attempt <= 1 and run_id in seen
            if (run_id, attempt) in failed_seen or followed and not ended:
                continue
            if adopt(run_id, False, refusals_known=refusals_listed):
                failed_seen.add((run_id, attempt))  # not before: a read that failed is tried again
        threads = [thread for thread in threads if thread.is_alive()]
        log(f"tick: {len(threads)} run(s) watched, {client.remaining or '?'} of "
            f"{client.limit or '?'} API requests left this hour")
        wait(tick_seconds)
    log("handing over: stopping watches; rescues under way finish")
    stopping.set()
    # One grace for all of them: a rescue started before the stop settles
    # within it (CANCEL_WAIT_SECONDS plus the re-run; rescue() checks `latest`).
    end = time.monotonic() + RESCUE_GRACE_SECONDS
    for thread in threads:
        thread.join(max(0.0, end - time.monotonic()))
    return outcomes

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