ci: park each pull request's build on its root and route its re-push back to it

.github/workflows/ci-macos.yml

@@ -606,6 +606,9 @@ jobs:
         id: owned-state
         if: steps.reuse-products.outputs.hit != 'true' && (github.event_name == 'pull_request' || github.event_name == 'workflow_dispatch' && github.ref == 'refs/heads/main') && startsWith(env.CMUX_PRODUCT_RUNNER, 'glaeda-')
         continue-on-error: true
+        env:
+          # This pull request's build parked on this root comes back first (owned_build_state.py PR slots).
+          CMUX_OWNED_PR: ${{ github.event.pull_request.number }}
         run: |
           set -euo pipefail
           fingerprint="$(scripts/ci/compile-app-host-test-product.sh canonical-fingerprint "$CMUX_COMPILE_ADMISSION_DERIVED_DATA")"

docs/ci-runners.md

@@ -315,6 +315,15 @@ loaded mini, then the name. The picker's candidates, pick and predicted
 seconds go to admission's record (`route.picker`) through the
 `admission_route` output.
 
+When `keep` replaces another pull request's build, it parks that build in
+`pr-builds/pr-<n>` beside the root's store (a rename; at most 2 per root, for
+6 h, and only with 140 GiB free). Admission's
+`check` for that pull request (`CMUX_OWNED_PR`) swaps it back in,
+glaeda's hook ranks the root by it, and `roots` publishes it as `parked`, so
+distance routing sends a re-push to the mini holding its own build. That
+start ranks far even when the pull request changes a package interface,
+where every other start rebuilds the app.
+
 The cost model is `scripts/ci/warm-distance-model.json`, fitted by
 `scripts/ci/warm_distance.py fit` from the line every owned admission appends
 to `/Users/Shared/cmux-build-fleet/ci/admissions.jsonl` on its mini (start,

scripts/ci/owned_build_state.py

@@ -163,6 +163,7 @@
 import shutil
 import subprocess
 import sys
+import time
 
 sys.path.insert(0, str(Path(__file__).resolve().parent))
 import seed_derived_data as seed  # noqa: E402
@@ -268,9 +269,152 @@ def sweep_discarded(store: Path) -> None:
         remove(stale)
 
 
-def check(store: Path, fingerprint: str, workspace: Path, package_store: Path | None = None) -> dict[str, str]:
+# Pull request build slots. A root's kept DerivedData is replaced by the next admission on it, so a
+# pull request's build was usually gone when its next push arrived: of 290 owned admissions on
+# 2026-09-26/27, 69 re-pushes rebuilt the app because of their own package or app changes, 53 of them
+# on another mini than their previous push and 16 on the same one after another pull request's build
+# replaced it. A start from the same pull request's build skipped the app rebuild 11 times in 19.
+# When `keep` replaces the kept build of another pull request, it parks that build in
+# STORE/pr-builds/pr-<n> (a rename, no copy), and `check` for pull request n swaps it back in before
+# anything else reads the kept state. glaeda's hook reads the parked stamp when it ranks roots, and
+# `warm-keys` publishes it (`parked`) for pr_runner_pool.py's distance routing. A root keeps at most
+# PR_SLOTS parked builds, each for PR_SLOT_HOURS, and parks none while the volume has less than
+# MIN_FREE_GIB free, so parked builds never crowd the mini's disk.
+PR_BUILDS = "pr-builds"
+PR_SLOTS = 2
+PR_SLOT_HOURS = 6
+MIN_FREE_GIB = 140
+
+
+def pr_slot(store: Path, number: object) -> Path | None:
+    key = pr_key(number)
+    return store / PR_BUILDS / key if key else None
+
+
+def free_gib(path: Path) -> float:
+    try:
+        return shutil.disk_usage(path).free / 1024**3
+    except OSError:
+        return 0.0
+
+
+def prune_pr_slots(store: Path, now: float | None = None) -> None:
+    """Drop parked builds past PR_SLOT_HOURS, then all but the newest PR_SLOTS."""
+    now = time.time() if now is None else now
+    try:
+        entries = list((store / PR_BUILDS).iterdir())
+    except OSError:
+        return
+    for path in entries:  # a park or clear a killed job left half done
+        if (path.name.startswith(".pr-") and (".incoming-" in path.name or ".discard-" in path.name)
+                and not owned_by_live_process(path)):
+            slot = store / PR_BUILDS / path.name[1:].split(".", 1)[0]
+            with contextlib.suppress(OSError, RuntimeError):
+                # A park killed after its stamp was written is whole: finish it rather than lose the build.
+                if (".incoming-" in path.name and (path / DERIVED).is_dir() and not slot.exists()
+                        and pr_key(read_stamp(path).get("pr")) == slot.name):
+                    path.rename(slot)
+                else:
+                    clear(path)
+    try:
+        entries = list((store / PR_BUILDS).iterdir())
+    except OSError:
+        return
+    slots = [path for path in entries if path.name.startswith("pr-")]
+    dated = []
+    for path in slots:
+        try:
+            dated.append((path.stat().st_mtime, path))
+        except OSError:
+            continue
+    dated.sort(reverse=True)
+    for index, (moment, path) in enumerate(dated):
+        if index >= PR_SLOTS or now - moment > PR_SLOT_HOURS * 3600:
+            with contextlib.suppress(OSError, RuntimeError):
+                clear(path)
+
+
+def park(store: Path) -> str:
+    """Move the kept build of a pull request into its slot; returns the slot name or ""."""
+    stamp = read_stamp(store)
+    slot = pr_slot(store, stamp.get("pr"))
+    if slot is None or not (store / DERIVED).is_dir() or not str(stamp.get("fingerprint") or "").endswith(STATE_VERSION):
+        return ""
+    if free_gib(store) < MIN_FREE_GIB:
+        return ""
+    incoming = slot.with_name(f".{slot.name}.incoming-{os.getpid()}")
+    remove(incoming)
+    incoming.mkdir(parents=True)
+    write_stamp(incoming, stamp)
+    write_stamp(store, {})  # the store no longer holds that build, whatever happens next
+    try:
+        (store / DERIVED).rename(incoming / DERIVED)
+    except OSError:
+        write_stamp(store, stamp)
+        raise
+    clear(slot)
+    incoming.rename(slot)
+    os.utime(slot)
+    return slot.name
+
+
+def unpark(store: Path, number: object, fingerprint: str) -> bool:
+    """Swap pull request NUMBER's parked build in as the kept one, parking the current kept build first."""
+    slot = pr_slot(store, number)
+    if slot is None or not (slot / DERIVED).is_dir():
+        return False
+    try:
+        if time.time() - slot.stat().st_mtime > PR_SLOT_HOURS * 3600:
+            return False  # expired: neither published nor routed to
+    except OSError:
+        return False
+    stamp = read_stamp(slot)
+    if not fingerprint or stamp.get("fingerprint") != stamped(fingerprint):
+        return False
+    current = read_stamp(store)
+    if pr_key(current.get("pr")) == pr_key(number) and (store / DERIVED).is_dir():
+        return False  # the kept build is this pull request's already
+    # A current main build (no pull request) stays, and so does one a park refused for disk: the job
+    # starts from it instead. A kept build no router can read (stale or missing stamp) is replaced.
+    kept_current = str(current.get("fingerprint") or "").endswith(f"-{STATE_VERSION}")
+    if (store / DERIVED).exists() and kept_current and not park(store):
+        return False
+    if (store / DERIVED).exists():
+        clear(store / DERIVED)
+    (slot / DERIVED).rename(store / DERIVED)
+    write_stamp(store, stamp)
+    with contextlib.suppress(OSError, RuntimeError):
+        clear(slot)
+    return True
+
+
+def parked_stamps(store: Path, now: float | None = None) -> list[dict[str, object]]:
+    """The current (STATE_VERSION, under PR_SLOT_HOURS) stamps of the builds parked beside STORE, newest first."""
+    now = time.time() if now is None else now
+    try:
+        dated = sorted(((path.stat().st_mtime, path) for path in (store / PR_BUILDS).iterdir()
+                        if path.name.startswith("pr-") and (path / DERIVED).is_dir()), reverse=True)
+    except OSError:
+        return []
+    slots = [path for moment, path in dated if now - moment <= PR_SLOT_HOURS * 3600]
+    found = []
+    for slot in slots:
+        stamp = read_stamp(slot)
+        if str(stamp.get("fingerprint") or "").endswith(f"-{STATE_VERSION}") and pr_key(stamp.get("pr")) == slot.name:
+            found.append(stamp)
+    return found
+
+
+def check(store: Path, fingerprint: str, workspace: Path, package_store: Path | None = None,
+          pr_number: str = "") -> dict[str, str]:
     store.mkdir(parents=True, exist_ok=True)
     sweep_discarded(store)
+    unparked = False
+    if pr_number:
+        try:
+            unparked = unpark(store, pr_number, fingerprint)
+        except (OSError, RuntimeError):
+            unparked = False
     if fingerprint and os.environ.get("RUNNER_OS") and os.environ.get("RUNNER_ARCH"):
         # Which seeds this root adopts, for seed_derived_data.py `prefetch`
         # to fetch ahead while the Mac is idle. Best effort.
@@ -302,7 +446,7 @@ def check(store: Path, fingerprint: str, workspace: Path, package_store: Path |
             result["reason"] = f"kept DerivedData grew to {size} bytes"
         else:
             result["warm"] = "true"
-            result["reason"] = "kept DerivedData matches"
+            result["reason"] = "this pull request's parked build" if unparked else "kept DerivedData matches"
     packages = (package_store or store) / PACKAGES
     if packages.is_dir():
         destination = workspace / ".ci-source-packages"
@@ -399,6 +543,18 @@ def keep(store: Path, derived: Path, fingerprint: str, merged_onto: str = "", pr
     # A seed's record is never replayed here (adopt reads RECORD only).
     for name in (*UNREAD, seed.MANIFEST):
         remove(incoming / name)
+    # Another pull request's build is parked, not dropped: its next push may come back to it.
+    parked = ""
+    if pr_key(read_stamp(store).get("pr")) != pr_key(pr_number):
+        try:
+            parked = park(store)
+        except (OSError, RuntimeError):
+            parked = ""
+    own_slot = pr_slot(store, pr_number)
+    if own_slot is not None:  # this build supersedes any parked build of the same pull request
+        with contextlib.suppress(OSError, RuntimeError):
+            clear(own_slot)
+    prune_pr_slots(store)
     stamp = read_stamp(store)
     stamp.pop("fingerprint", None)
     stamp.pop("merged_onto", None)
@@ -415,7 +571,7 @@ def keep(store: Path, derived: Path, fingerprint: str, merged_onto: str = "", pr
     if pr_key(pr_number):
         stamp["pr"] = int(pr_number.strip())
     write_stamp(store, stamp)
-    return {"kept": "true"}
+    return {"kept": "true", **({"parked": parked} if parked else {})}
 
 
 def pr_key(number: object) -> str:
@@ -462,14 +618,18 @@ def root_number(store: Path, first: Path) -> int:
 
 
 def root_summary(store: Path, number: int) -> dict[str, object]:
-    """What glaeda's hook reads from root NUMBER's stamp, when it keeps a current build; else the number alone."""
+    """What glaeda's hook reads from root NUMBER's stamp, when it keeps a current build, and from its parked
+    pull request builds (`parked`); else the number alone."""
     stamp = read_stamp(store)
-    if not (store / DERIVED).is_dir() or not str(stamp.get("fingerprint") or "").endswith(f"-{STATE_VERSION}"):
-        return {"root": number}
     summary: dict[str, object] = {"root": number}
-    for field in ROOT_FIELDS:
-        if field in stamp:
-            summary[field] = stamp[field]
+    if (store / DERIVED).is_dir() and str(stamp.get("fingerprint") or "").endswith(f"-{STATE_VERSION}"):
+        for field in ROOT_FIELDS:
+            if field in stamp:
+                summary[field] = stamp[field]
+    # The pull request builds parked beside it (PR slots), which `check` swaps in for their next push.
+    parked = [{field: slot[field] for field in ROOT_FIELDS if field in slot} for slot in parked_stamps(store)]
+    if parked:
+        summary["parked"] = parked[:PR_SLOTS]
     return summary
 
 
@@ -826,7 +986,8 @@ def package_store(argv: list[str]) -> Path | None:
 
 def main(argv: list[str]) -> int:
     if len(argv) in (5, 6) and argv[1] == "check":
-        write_outputs(check(Path(argv[2]), argv[3], Path(argv[4]), package_store(argv)))
+        write_outputs(check(Path(argv[2]), argv[3], Path(argv[4]), package_store(argv),
+                            os.environ.get("CMUX_OWNED_PR", "")))
         return 0
     if len(argv) == 5 and argv[1] == "adopt":
         write_outputs(adopt(Path(argv[2]), Path(argv[3]), Path(argv[4]).resolve()))

scripts/ci/owned_warm_state.py

@@ -92,6 +92,7 @@
 # Roots per mini and files per root kept (owned_build_state.py MAX_PATHS caps the stamp's list).
 MAX_ROOTS = 4
 MAX_ROOT_FILES = 400
+MAX_PARKED = 2  # parked pull request builds per root (owned_build_state.py PR_SLOTS)
 MAX_AGE_HOURS = 24
 MAX_JOB_PAGES = 3
 # Previous snapshots read, newest first, for the last one that has `warm`.
@@ -120,8 +121,31 @@ def keys(document: Any) -> list[str]:
     return found[:MAX_KEYS]
 
 
+def stamp_fields(entry: Mapping[str, Any]) -> dict[str, Any]:
+    """A root's or parked build's stamp fields (owned_build_state.py ROOT_FIELDS), each checked."""
+    clean: dict[str, Any] = {}
+    onto = str(entry.get("merged_onto") or "").lower()
+    if len(onto) == 40 and warm_key(onto):
+        clean["merged_onto"] = onto
+    pr = entry.get("pr")
+    if isinstance(pr, int) and not isinstance(pr, bool) and 0 < pr < 10**9:
+        clean["pr"] = pr
+    files = entry.get("pr_app_swift_files")
+    if isinstance(files, list):
+        clean["pr_app_swift_files"] = [path for path in files[:MAX_ROOT_FILES]
+                                       if isinstance(path, str) and 0 < len(path) <= 512 and "\0" not in path]
+    total = entry.get("pr_app_swift_total")
+    if isinstance(total, int) and not isinstance(total, bool) and 0 <= total < 10**6:
+        clean["pr_app_swift_total"] = total
+    if entry.get("pr_package_interface") in (True, False, None) and "pr_package_interface" in entry:
+        clean["pr_package_interface"] = entry.get("pr_package_interface")
+    return clean
+
+
 def roots(document: Any) -> list[dict[str, Any]]:
-    """The artifact's valid roots (owned_build_state.py `warm-keys`), at most MAX_ROOTS, fields checked."""
+    """The artifact's valid roots (owned_build_state.py `warm-keys`), at most MAX_ROOTS, fields checked.
+
+    A root's `parked` pull request builds (at most MAX_PARKED, each with a pull request) are kept too."""
     raw = document.get("roots") if isinstance(document, Mapping) else None
     found: list[dict[str, Any]] = []
     seen: set[int] = set()
@@ -130,22 +154,13 @@ def roots(document: Any) -> list[dict[str, Any]]:
         if not isinstance(number, int) or isinstance(number, bool) or not 0 < number < 100 or number in seen:
             continue
         seen.add(number)
-        clean: dict[str, Any] = {"root": number}
-        onto = str(entry.get("merged_onto") or "").lower()
-        if len(onto) == 40 and warm_key(onto):
-            clean["merged_onto"] = onto
-        pr = entry.get("pr")
-        if isinstance(pr, int) and not isinstance(pr, bool) and 0 < pr < 10**9:
-            clean["pr"] = pr
-        files = entry.get("pr_app_swift_files")
-        if isinstance(files, list):
-            clean["pr_app_swift_files"] = [path for path in files[:MAX_ROOT_FILES]
-                                           if isinstance(path, str) and 0 < len(path) <= 512 and "\0" not in path]
-        total = entry.get("pr_app_swift_total")
-        if isinstance(total, int) and not isinstance(total, bool) and 0 <= total < 10**6:
-            clean["pr_app_swift_total"] = total
-        if entry.get("pr_package_interface") in (True, False, None) and "pr_package_interface" in entry:
-            clean["pr_package_interface"] = entry.get("pr_package_interface")
+        clean: dict[str, Any] = {"root": number, **stamp_fields(entry)}
+        parked = entry.get("parked")
+        parked = [stamp_fields(item) for item in parked[:MAX_PARKED] if isinstance(item, Mapping)] \
+            if isinstance(parked, list) else []
+        parked = [item for item in parked if "pr" in item]
+        if parked:
+            clean["parked"] = parked
         found.append(clean)
     return sorted(found, key=lambda item: item["root"])[:MAX_ROOTS]