cmux-next: the fast tier runs as a build-fleet client when every macOS label is full or dead

.github/runners.json

@@ -247,6 +247,7 @@
     "cmux-next.yml#daemon-test": "apple-swift-tests",
     "cmux-next.yml#dev-landing-compile": "apple-app-build",
     "cmux-next.yml#fast-tier": "apple-app-build",
+    "cmux-next.yml#fast-tier-fleet": "apple-swift-tests",
     "cmux-next.yml#generated-files": "apple-app-build",
     "cmux-next.yml#release-compile": "apple-app-build",
     "cmux-next.yml#release-compile-fast": "apple-app-build",

.github/warm-sources.json

@@ -92,6 +92,7 @@
     "cmux-next.yml:daemon-test": {"class": "pr-swift-test", "warm": ["spm-shared", "actions-cache"]},
     "cmux-next.yml:dev-landing-compile": {"class": "dev-release-compile", "warm": ["deriveddata-reuse", "dev-cas", "product-reuse"]},
     "cmux-next.yml:fast-tier": {"class": "pr-fast-tier", "warm": ["deriveddata-reuse", "spm-shared", "actions-cache", "product-reuse"]},
+    "cmux-next.yml:fast-tier-fleet": {"class": "fleet-dispatch", "warm": ["deriveddata-reuse"]},
     "cmux-next.yml:generated-files": {"class": "mac-test", "warm": ["spm-shared", "actions-cache", "product-reuse"]},
     "cmux-next.yml:release-compile": {"class": "pr-release-compile", "warm": ["deriveddata-reuse", "product-reuse"]},
     "cmux-next.yml:release-compile-fast": {"class": "pr-release-compile", "warm": ["linux-pool-warm"]},

scripts/ci/side_lane_placement.py

@@ -107,6 +107,7 @@
 """
 from __future__ import annotations
 
+import dataclasses
 import datetime as dt
 import importlib.util
 import math
@@ -588,6 +589,31 @@ def gateway_taken(env: Mapping[str, str], runners: Sequence[Mapping[str, Any]] |
     return taken
 
 
+def fleet_gateway(placed: Any, env: Mapping[str, str], runners: Sequence[Mapping[str, Any]] | None,
+                  gateway: str, taken: Mapping[str, int] | None = None) -> Any:
+    """`placed` with its fleet_jobs kept only while `gateway` has an idle runner for each of them.
+
+    A listed job's build-fleet client (cmux-next.yml's <job>-fleet) runs on gateway_runner beside
+    GATEWAY_JOBS, so it needs an idle runner there beyond theirs and beyond the run's own jobs (`taken`,
+    as gateway_route() counts them). Without one the client would queue behind another gateway job
+    until its timeout and report no verdict, so placement lists nothing and the job keeps its route
+    (it waits on the owned label as before). No gateway (none idle, unread runners, a re-run) lists
+    nothing either.
+    """
+    if not placed.fleet_jobs:
+        return placed
+    want = len((env.get("GATEWAY_JOBS") or "").split()) + len(placed.fleet_jobs)
+    room = (pool.live_owned_free(runners, [gateway])[gateway] - (taken or {}).get(gateway, 0)
+            if gateway and runners is not None else 0)
+    if room >= want:
+        return placed
+    clients = ", ".join(f"{job}-fleet" for job in placed.fleet_jobs)
+    where = f"`{gateway}`" if gateway else "the gateway"
+    return dataclasses.replace(placed, fleet_jobs=(), why=(
+        f"{placed.why}; {where} has no idle runner left for {clients} ({max(room, 0)} for {want} gateway jobs), "
+        f"so {', '.join(placed.fleet_jobs)} keep their route"))
+
+
 def without_cascade(env: Mapping[str, str], job: str) -> dict[str, str]:
     """`env` with `job` dropped from CASCADE_RUNS and FIRST_JOBS."""
     trimmed = dict(env)
@@ -664,6 +690,8 @@ def rerun_main(env: Mapping[str, str], runners: Sequence[Mapping[str, Any]] | No
     # Runners read: the jobs past `owned` have no idle mini, so the cascade's overflow may take them.
     placed = cascade.place(env, owned, live, waiting, seen, saturated=runners is not None,
                            route_wait=route_wait(label, runners, seen))
+    # A re-run emits no gateway_runner, so no build-fleet client runs (fleet_gateway()).
+    placed = fleet_gateway(placed, env, runners, "")
     overflow = "" if free_gui else gui_overflow(env, live, waiting, seen, placed)
     gui_why = (f"the GUI jobs take idle `{free_gui}`" if free_gui
                else f"no idle GUI runner, so the GUI jobs take `{overflow}`")
@@ -735,7 +763,8 @@ def _main(env: Mapping[str, str] = os.environ) -> int:
                      "like the other jobs")
     gateway_label = (env.get("GATEWAY_LABEL") or "").strip()
     attempt = (env.get("GITHUB_RUN_ATTEMPT") or "").strip() or "1"
-    gateway = gateway_route(env, runners, gateway_taken(env, runners, fast))
+    taken = gateway_taken(env, runners, fast)
+    gateway = gateway_route(env, runners, taken)
     if gateway_label:
         if gateway:
             why += f"; the build-fleet compiles take idle `{gateway}`"
@@ -769,6 +798,8 @@ def _main(env: Mapping[str, str] = os.environ) -> int:
     # for longer than 6 vCPU's oldest job has waited, or one host alone serves it), they take 6 vCPU.
     placed = cascade.place(env, owned, live, waiting, seen, promote_busy=promotion_busy,
                            saturated=runners is not None, route_wait=route_wait(route, runners, seen))
+    # A job listed for the build fleet runs its client on the gateway only with an idle runner of its own there.
+    placed = fleet_gateway(placed, env, runners, gateway, taken)
     # With no GUI route that has room, the GUI jobs take the largest Blacksmith macOS size that starts them now,
     # after the slots this run's cascade just took (gui_overflow()). An unread fleet vouches for no owned GUI
     # queue either, so it overflows too, as it did before (the step's synthesized `runner` sent it to 6 vCPU).

tests/test_ci_fork_runner_routing.py

@@ -1058,11 +1058,21 @@ def test_every_fork_exercised_runner_has_a_hosted_path(self) -> None:
                     or (matrix_hosted and FORK_MACOS_MATRIX_BRANCH in line)
                 )
 
+                # A mixed macOS/Linux selector (cmux-next.yml's fast tier, whose
+                # fleet-client fallback runs on vars.LINUX_RUNNER) is fork-safe when
+                # its first clause is a hosted fork branch: a fork's own repository
+                # stops there and never reads the Linux label, and a fork PR into
+                # manaflow-ai still has to pass the macOS checks below.
+                fork_short_circuits = any(
+                    re.search(r"runs-on:\s*\$\{\{\s*" + re.escape(branch) + r"\s*\|\|", line)
+                    for branch in FORK_BRANCHES
+                )
+
                 with self.subTest(workflow=path.name, line=number):
                     if "vars.LINUX_RUNNER" in line:
                         saw_linux += 1
                         self.assertTrue(
-                            hosted_linux,
+                            hosted_linux or (fork_short_circuits and hosted_macos),
                             f"{path.name}:{number} has no GitHub-hosted Linux fork branch",
                         )
                     if "vars.MACOS_RUNNER" in line:

tests/test_ci_macos_xcode_selection.py

@@ -105,6 +105,10 @@
         "build-fleet gateway: checks nothing out and only runs cmux-ci "
         "compile-check; the compile runs on a fleet worker, never with this "
         "runner's Xcode",
+    ("cmux-next.yml", "fast-tier-fleet"):
+        "build-fleet gateway: checks nothing out and only runs cmux-ci "
+        "run (package-test-lane.sh suites); the suites build on a fleet "
+        "worker, never with this runner's Xcode",
     ("cmux-next.yml", "release-compile-fast"):
         "build-fleet gateway: checks nothing out and only runs cmux-ci "
         "compile-check --release; the compile runs on a fleet worker, never "

tests/test_ci_next_macos_cascade.py

@@ -372,13 +372,21 @@ def test_all_full_routes_reserve_only_requested_fleet_jobs(self):
                                {LARGE: 5, SMALL: 10}, {LARGE: 1, SMALL: 1}, cascade.seen(shot, NOW), saturated=True)
         self.assertEqual(result.fleet_jobs, ("fast-tier",))
 
-    def test_gateway_output_is_reserved_without_wiring_raw_mac_jobs(self):
+    def test_no_tier_runs_on_the_gateway_itself(self):
+        """fleet_jobs never sends a whole Mac job (checkout, build) to gateway_runner: a listed tier's
+        <tier>-fleet client runs there (the checks-workflow guard GatewayRunsOnlyTheFleetClient)
+        and the tier only reports its verdict."""
         jobs = yaml.safe_load(WORKFLOW.read_text())["jobs"]
         self.assertEqual(jobs["macos-placement"]["outputs"]["fleet_jobs"], "${{ steps.place.outputs.fleet_jobs }}")
+        wiring = Wiring()
         for tier in (*CASCADED, *GUI_JOBS):
             self.assertNotIn("outputs.fleet_jobs", jobs[tier]["runs-on"])
-        place = next(step for step in jobs["macos-placement"]["steps"] if step.get("id") == "place")
-        self.assertNotIn("FLEET_CLIENT_JOBS", place["env"])
+            with self.subTest(tier=tier):
+                self.assertNotIn("gateway_runner", str(jobs[tier]["runs-on"]))
+                context = wiring.context()
+                context["needs"]["macos-placement"]["outputs"].update(
+                    fleet_jobs=f" {tier} ", gateway_runner="gateway-test", placed_attempt="1")
+                self.assertNotEqual(evaluate(jobs[tier]["runs-on"], context), "gateway-test")
 
     def test_gateway_reservation_is_empty_without_requested_jobs(self):
         shot = snapshot(minutes_old=1)
@@ -928,7 +936,9 @@ def test_no_job_reads_a_hosted_route(self):
             self.assertNotIn(f"outputs.{key}", text)
         jobs = yaml.safe_load(text)["jobs"]
         # The fast tier and the canary keep their own budget: no cold hosted build needs 45 minutes.
-        self.assertEqual(jobs["fast-tier"]["timeout-minutes"], 20)
+        # (Five while it only reports its build-fleet client's verdict: GatewayRunsOnlyTheFleetClient.)
+        fast = jobs["fast-tier"]["timeout-minutes"]
+        self.assertEqual(evaluate(fast, {"needs": {"fast-tier-fleet": {"result": "skipped"}}}), 20)
         canary = jobs["swift-canary"]["timeout-minutes"]
         self.assertEqual(evaluate(canary, {"needs": {"swift-canary-fleet": {"outputs": {"verdict": ""}}}}), 20)
         self.assertEqual(jobs["cmux-scheme-compile"]["timeout-minutes"], 60)
@@ -2148,6 +2158,64 @@ def test_the_sign_runner_reads_the_same_rule(self):
         self.assertEqual(sign.dead(snap, NOW), set(cascade.dead(snap, NOW)))
 
 
+class FleetClientJobs(unittest.TestCase):
+    """With every owned and Blacksmith macOS label full or dead, only the jobs that have a build-fleet client
+    (FLEET_CLIENT_JOBS: cmux-next.yml's <job>-fleet) are listed in fleet_jobs; nothing is while any label
+    can take the job, and the workflow lists only jobs that have their client."""
+
+    LIVE, WAITING = {LARGE: 5, SMALL: 10}, {LARGE: 1, SMALL: 1}
+
+    def shot(self, **side: object) -> dict:
+        shot = snapshot(minutes_old=1, large=1, small=1)
+        for label in (LARGE, SMALL, SIDE):
+            shot["pools"][label] = {"cap": 1, "running": 1, "queued": 1,
+                                    "listening": 0, "held_offline": {"units": 1, "cpu": 0, "livelock": 0},
+                                    "oldest_queued_s": 240, "as_of": "2026-10-19T21:59:00Z"}
+        shot["pools"][SIDE].update(side)
+        return shot
+
+    def place(self, owned=(), shot=None, live=None, **overrides):
+        return cascade.place(env(**overrides), owned, live or self.LIVE, self.WAITING,
+                             cascade.seen(shot or self.shot(), NOW), saturated=True)
+
+    def test_only_listed_jobs_take_the_fleet(self):
+        self.assertEqual(self.place(FLEET_CLIENT_JOBS="fast-tier").fleet_jobs, ("fast-tier",))
+        self.assertEqual(self.place(FLEET_CLIENT_JOBS="fast-tier swift-canary").fleet_jobs,
+                         ("fast-tier", "swift-canary"))
+        self.assertEqual(self.place().fleet_jobs, ())
+        self.assertEqual(self.place(FLEET_CLIENT_JOBS="").fleet_jobs, ())
+
+    def test_nothing_is_listed_while_a_label_can_take_the_job(self):
+        # A Blacksmith label under its declared cap.
+        shot = self.shot()
+        shot["pools"][LARGE].update(cap=5, running=1)
+        self.assertEqual(self.place(shot=shot, live={LARGE: 1, SMALL: 10}, FLEET_CLIENT_JOBS="fast-tier").fleet_jobs,
+                         ())
+        # The side label has a listener, by a fresh health record (a stale record or snapshot reads full).
+        fresh = dt.datetime.now(dt.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
+        listening = self.shot(listening=1, held_offline={"units": 0, "cpu": 0, "livelock": 0}, as_of=fresh)
+        listening["generated_at"] = fresh
+        self.assertEqual(self.place(shot=listening, FLEET_CLIENT_JOBS="fast-tier").fleet_jobs, ())
+        # No snapshot: nothing is known full.
+        self.assertEqual(cascade.place(env(FLEET_CLIENT_JOBS="fast-tier"), (), self.LIVE, self.WAITING, None,
+                                       saturated=True).fleet_jobs, ())
+
+    def test_the_output_is_space_delimited_like_the_blacksmith_lists(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            out = Path(tmp) / "out"
+            cascade.write(self.place(FLEET_CLIENT_JOBS="fast-tier"), {"GITHUB_OUTPUT": str(out)})
+            self.assertIn("fleet_jobs= fast-tier \n", out.read_text())
+
+    def test_the_workflow_lists_only_jobs_with_a_client(self):
+        jobs = yaml.safe_load(WORKFLOW.read_text(encoding="utf-8"))["jobs"]
+        place = next(step for step in jobs["macos-placement"]["steps"] if step.get("id") == "place")
+        listed = str(place["env"]["FLEET_CLIENT_JOBS"]).split()
+        self.assertEqual(listed, ["fast-tier"])
+        for job in listed:
+            with self.subTest(job=job):
+                self.assertIn(job, cascade.CASCADE_JOBS, "only a cascade job is ever left unplaced")
+                self.assertIn(f"{job}-fleet", jobs, "a listed job needs its <job>-fleet client")
+
 
 class GuiOverflow(unittest.TestCase):
     """The GUI jobs' Blacksmith leg (swift-test, daemon-test, generated-files): the largest size that starts them now.

tests/test_ci_side_lane_placement.py

@@ -922,6 +922,42 @@ def test_without_the_variable_nothing_is_judged(self):
         lines = self.run_main(self.light(False))
         self.assertEqual((lines["gateway_runner"], lines["gateway_attempt"]), ("", ""))
 
+    def fleet_placed(self, *jobs: str):
+        return placement.cascade.Placement({placement.cascade.LARGE: (), placement.cascade.SMALL: ()}, "", "placed",
+                                           fleet_jobs=jobs)
+
+    def test_a_fleet_client_needs_its_own_idle_gateway_runner(self):
+        # fast-tier-fleet runs on gateway_runner beside GATEWAY_JOBS: with one idle runner for two jobs one
+        # would queue behind the other until its timeout, so placement lists nothing and fast-tier keeps its route.
+        values = {"GATEWAY_LABEL": LIGHT_SIDE, "GATEWAY_JOBS": "swift-canary-fleet"}
+        placed = self.fleet_placed("fast-tier")
+        two, one = self.light(False, False), self.light(False, True)
+        self.assertEqual(placement.fleet_gateway(placed, values, two, LIGHT_SIDE).fleet_jobs, ("fast-tier",))
+        self.assertEqual(placement.fleet_gateway(placed, values, one, LIGHT_SIDE).fleet_jobs, ())
+        # The run's own jobs racing for the gateway's runners count too, as gateway_route() counts them.
+        self.assertEqual(placement.fleet_gateway(placed, values, two, LIGHT_SIDE, {LIGHT_SIDE: 1}).fleet_jobs, ())
+        self.assertEqual(placement.fleet_gateway(placed, {"GATEWAY_LABEL": LIGHT_SIDE}, one, LIGHT_SIDE).fleet_jobs,
+                         ("fast-tier",))
+        # No gateway, unread runners: no client runs, so nothing is listed.
+        self.assertEqual(placement.fleet_gateway(placed, values, two, "").fleet_jobs, ())
+        self.assertEqual(placement.fleet_gateway(placed, values, None, LIGHT_SIDE).fleet_jobs, ())
+        dropped = placement.fleet_gateway(placed, values, one, LIGHT_SIDE)
+        self.assertIn("no idle runner left for fast-tier-fleet", dropped.why)
+        self.assertIs(placement.fleet_gateway(self.fleet_placed(), values, one, LIGHT_SIDE).fleet_jobs, ())
+
+    def test_main_lists_a_fleet_job_only_with_a_gateway_runner_for_its_client(self):
+        values = {"GATEWAY_LABEL": LIGHT_SIDE, "GATEWAY_JOBS": "swift-canary-fleet", "CASCADE_RUNS": ""}
+        for runners, listed in ((self.light(False, False), " fast-tier "), (self.light(False, True), "")):
+            with self.subTest(idle=len([r for r in runners if not r.get("busy")])), \
+                    mock.patch.object(placement.cascade, "place", return_value=self.fleet_placed("fast-tier")):
+                lines = self.run_main(runners, **values)
+                self.assertEqual(lines["gateway_runner"], LIGHT_SIDE)
+                self.assertEqual(lines["fleet_jobs"], listed)
+
+    def test_a_rerun_lists_no_fleet_job(self):
+        self.assertEqual(placement.fleet_gateway(self.fleet_placed("fast-tier"), {}, self.light(False, False), "")
+                         .fleet_jobs, ())
+
 
 class SpillToCompatiblePool(unittest.TestCase):
     def setUp(self):