feat(ci): publish hosted light candidates in pool artifact

.github/workflows/ci-queue-janitor.yml

@@ -246,6 +246,10 @@ jobs:
           # dashboard (cmuxterm-hq build-fleet/ci-dash); JSON data only, read
           # through env and never expanded into a script. Into the snapshot as `gates`.
           FLEET_GATES: ${{ vars.CI_FLEET_GATES }}
+          # Optional sanitized hosted-light discovery hints from the existing
+          # runner-state variable; the janitor never lists runners or runs for this.
+          CI_RUNNER_STATE: ${{ vars.CI_RUNNER_STATE }}
+          CI_HOSTED_LIGHT_CANDIDATES_DISABLED: ${{ vars.CI_HOSTED_LIGHT_CANDIDATES_DISABLED }}
         run: python3 scripts/ci/queue_janitor.py
 
       # Early warning for the GITHUB_TOKEN budget every workflow shares: when it ran out on

scripts/ci/queue_janitor.py

@@ -162,6 +162,13 @@
 pick a pull request run's pool (and, through it, e2e_runner_pool.py an E2E
 run's) without listing every in-flight run's jobs itself.
 
+When the existing ``CI_RUNNER_STATE`` variable carries the optional
+``hosted_light_candidates`` envelope, a complete sweep copies its strictly
+sanitized projection into ``hosted-light-candidates`` in that same artifact.
+This is discovery only: a consumer still validates the current run, PR head
+and complete jobs before taking any action. Missing or stale runner state is
+omitted, and a truncated janitor sweep never presents candidate completeness.
+
 An owned Mac pool (``glaeda-<class>-xcode-<version>``) is one more pool
 here: stale pull request runs (b) on it are cancelled whatever its queue,
 which frees minis, and the other categories only while it is backed up. With
@@ -198,8 +205,10 @@
 from __future__ import annotations
 
 import argparse
+import base64
 import dataclasses
 import datetime as dt
+import gzip
 import http.client
 import io
 import json
@@ -236,6 +245,22 @@
 from delivery_workflows import dev_publish_published, dev_publish_source, dev_publish_superseded  # noqa: E402
 
 
+# CI_RUNNER_STATE is an optional, already-sanitized source published by the
+# trusted runner-state publisher.  The janitor only copies its hosted-light
+# hints into the public pool artifact after independently checking the source.
+RUNNER_STATE_SCHEMA = "runner-state/v1"
+RUNNER_STATE_REPOSITORY = "manaflow-ai/cmux"
+HOSTED_LIGHT_CANDIDATES_SCHEMA = "hosted-light-candidates/v1"
+HOSTED_LIGHT_CANDIDATES_KEY = "hosted-light-candidates"
+HOSTED_LIGHT_SOURCE_KEY = "hosted_light_candidates"
+RUNNER_STATE_MAX_AGE_S = 300
+RUNNER_STATE_CLOCK_SKEW_S = 5
+RUNNER_STATE_MAX_ENCODED = 48 * 1024
+RUNNER_STATE_MAX_DECODED = 2 * 1024 * 1024
+HOSTED_LIGHT_WORKFLOW = re.compile(r"\.github/workflows/[A-Za-z0-9_.-]+\.ya?ml\Z")
+HOSTED_LIGHT_SHA = re.compile(r"[0-9a-f]{40}\Z")
+
+
 API = "https://api.github.com"
 DEFAULT_THRESHOLD = 6
 DEFAULT_MAX_CANCELS = 10
@@ -871,6 +896,113 @@ def count(value: Any) -> bool:
     return {"version": FLEET_GATES_VERSION, "as_of": as_of, "labels": dict(sorted(out.items()))}, None
 
 
+def _runner_state_moment(value: Any) -> dt.datetime:
+    if not isinstance(value, str):
+        raise ValueError("missing source timestamp")
+    result = dt.datetime.fromisoformat(value.replace("Z", "+00:00"))
+    if result.tzinfo is None:
+        raise ValueError("undated source timestamp")
+    return result.astimezone(UTC)
+
+
+def _runner_state_fresh(value: Any, now: dt.datetime, *, max_age: int = RUNNER_STATE_MAX_AGE_S) -> str:
+    stamp = _runner_state_moment(value)
+    age = (now - stamp).total_seconds()
+    if not -RUNNER_STATE_CLOCK_SKEW_S <= age <= max_age:
+        raise ValueError("source timestamp is stale or future-dated")
+    return stamp.isoformat().replace("+00:00", "Z")
+
+
+def hosted_light_candidates(raw: str | None, *, now: dt.datetime) -> dict[str, Any] | None:
+    """Decode and sanitize the optional runner-state candidate envelope.
+
+    This is deliberately a one-way projection: no runner-state or controller
+    payload is forwarded to the public artifact. Invalid optional input is
+    unknown and leaves the existing pool snapshot usable.
+    """
+    if not raw or len(raw) > RUNNER_STATE_MAX_ENCODED:
+        return None
+    try:
+        encoded = base64.b64decode(raw, validate=True)
+        with gzip.GzipFile(fileobj=io.BytesIO(encoded)) as source:
+            body = source.read(RUNNER_STATE_MAX_DECODED + 1)
+        if len(body) > RUNNER_STATE_MAX_DECODED:
+            return None
+        state = json.loads(body)
+        if (not isinstance(state, dict) or state.get("schema") != RUNNER_STATE_SCHEMA
+                or state.get("repository") != RUNNER_STATE_REPOSITORY or state.get("complete") is not True):
+            return None
+        _runner_state_fresh(state.get("generated_at"), now)
+        source = state.get(HOSTED_LIGHT_SOURCE_KEY)
+        if (not isinstance(source, dict) or source.get("schema") != HOSTED_LIGHT_CANDIDATES_SCHEMA
+                or source.get("repository") != RUNNER_STATE_REPOSITORY
+                or source.get("coverage") != "partial" or source.get("complete") is not False
+                or source.get("truncated") is not True):
+            return None
+        generated_at = _runner_state_fresh(source.get("generated_at"), now)
+        controller_generated_at = _runner_state_fresh(source.get("controller_generated_at"), now)
+        receiver_last_delivery = _runner_state_fresh(source.get("receiver_last_delivery"), now)
+        jobs_since = _runner_state_moment(source.get("jobs_since"))
+        controller_clock = _runner_state_moment(source["controller_generated_at"])
+        if (jobs_since > controller_clock + dt.timedelta(seconds=5)
+                or _runner_state_moment(source["receiver_last_delivery"]) > controller_clock + dt.timedelta(seconds=5)):
+            return None
+        source_rows = source.get("runs")
+        total_runs = source.get("total_runs")
+        truncated = source.get("truncated")
+        if (not isinstance(source_rows, list) or type(total_runs) is not int or total_runs < 0
+                or type(truncated) is not bool
+                or total_runs < len(source_rows)):
+            return None
+        rows: list[dict[str, Any]] = []
+        seen: set[int] = set()
+        for row in source_rows:
+            if not isinstance(row, dict):
+                continue
+            run_id = row.get("id")
+            path = row.get("path")
+            event = row.get("event")
+            attempt = row.get("run_attempt")
+            status = row.get("status")
+            head_sha = row.get("head_sha")
+            created_at = row.get("created_at")
+            try:
+                created = _runner_state_moment(created_at)
+                age = (now - created).total_seconds()
+            except (TypeError, ValueError, OverflowError):
+                continue
+            if (type(run_id) is not int or run_id <= 0 or run_id in seen
+                    or not isinstance(path, str) or not HOSTED_LIGHT_WORKFLOW.fullmatch(path)
+                    or event != "pull_request" or attempt != 1
+                    or status not in ("queued", "in_progress")
+                    or not isinstance(head_sha, str) or not HOSTED_LIGHT_SHA.fullmatch(head_sha)
+                    or not -RUNNER_STATE_CLOCK_SKEW_S <= age <= 90 * 60):
+                continue
+            seen.add(run_id)
+            rows.append({"id": run_id, "path": path, "event": event, "run_attempt": attempt,
+                         "status": status, "head_sha": head_sha, "created_at": created_at})
+        rows.sort(key=lambda row: (_runner_state_moment(row["created_at"]), row["id"]))
+        return {"schema": HOSTED_LIGHT_CANDIDATES_SCHEMA, "repository": RUNNER_STATE_REPOSITORY,
+                "generated_at": generated_at, "controller_generated_at": controller_generated_at,
+                "receiver_last_delivery": receiver_last_delivery,
+                "jobs_since": jobs_since.isoformat().replace("+00:00", "Z"),
+                "coverage": "partial", "complete": False, "runs": rows,
+                "truncated": True, "total_runs": total_runs}
+    except (ValueError, TypeError, OSError, EOFError, OverflowError, json.JSONDecodeError, zlib.error):
+        return None
+
+
+def attach_hosted_light_candidates(snapshot: dict[str, Any], raw: str | None, *, now: dt.datetime) -> None:
+    """Attach candidates only to a complete outer janitor sweep."""
+    if snapshot.get("truncated") or snapshot.get("starved_incomplete"):
+        return
+    if os.environ.get("CI_HOSTED_LIGHT_CANDIDATES_DISABLED") == "1":
+        return
+    envelope = hosted_light_candidates(raw, now=now)
+    if envelope is not None:
+        snapshot[HOSTED_LIGHT_CANDIDATES_KEY] = envelope
+
+
 MAX_ARTIFACT_PAGES = 5
 # ci.yml's `changes` job uploads this marker when the picker chose an owned
 # pool: macos-pool-persistent-<run>-<attempt>-<jobs>p<placed>-<pool>, where
@@ -4434,6 +4566,10 @@ def read_heads(branches_wanted: Sequence[str]) -> None:
         else:
             snapshot["gates_error"] = gates_error
             print(f"queue-janitor: fleet gates: {gates_error}", file=sys.stderr)
+        # This optional hint is copied only after the ordinary listing has
+        # proved complete. A partial janitor sweep must retain its existing
+        # unknown/unsafe semantics, even when runner-state has candidates.
+        attach_hosted_light_candidates(snapshot, os.environ.get("CI_RUNNER_STATE"), now=now)
         args.pool_load.write_text(json.dumps(snapshot, indent=2) + "\n", encoding="utf-8")
 
     if unread or listing["truncated"]:

tests/test_ci_queue_janitor.py

@@ -4,7 +4,9 @@
 from __future__ import annotations
 
 import datetime as dt
+import base64
 import email.message
+import gzip
 import http.client
 import importlib.util
 import io
@@ -4060,6 +4062,108 @@ def test_the_workflow_passes_the_variable_through_env_only(self):
         self.assertEqual(text.count("vars.CI_FLEET_GATES"), 1)  # never expanded into a script
 
 
+class HostedLightCandidatesTests(unittest.TestCase):
+    def _raw(self, *, rows=None, state_overrides=None, candidate_overrides=None):
+        created = (NOW - dt.timedelta(minutes=5)).strftime("%Y-%m-%dT%H:%M:%SZ")
+        rows = rows if rows is not None else [{
+            "id": 123, "path": ".github/workflows/ci.yml", "event": "pull_request",
+            "run_attempt": 1, "status": "queued", "head_sha": "a" * 40,
+            "created_at": created, "private": "drop-me",
+        }]
+        source = {
+            "schema": "hosted-light-candidates/v1", "repository": "manaflow-ai/cmux",
+            "generated_at": (NOW - dt.timedelta(seconds=20)).strftime("%Y-%m-%dT%H:%M:%SZ"),
+            "controller_generated_at": (NOW - dt.timedelta(seconds=10)).strftime("%Y-%m-%dT%H:%M:%SZ"),
+            "receiver_last_delivery": (NOW - dt.timedelta(seconds=20)).strftime("%Y-%m-%dT%H:%M:%SZ"),
+            "jobs_since": (NOW - dt.timedelta(hours=1)).strftime("%Y-%m-%dT%H:%M:%SZ"),
+            "coverage": "partial", "complete": False, "truncated": True,
+            "total_runs": len(rows), "runs": rows,
+        }
+        source.update(candidate_overrides or {})
+        state = {"schema": "runner-state/v1", "repository": "manaflow-ai/cmux", "complete": True,
+                 "generated_at": (NOW - dt.timedelta(seconds=30)).strftime("%Y-%m-%dT%H:%M:%SZ"),
+                 "runner_groups": {}, "repository_runners": [], "scale_sets": {},
+                 "hosted_light_candidates": source}
+        state.update(state_overrides or {})
+        return base64.b64encode(gzip.compress(json.dumps(state).encode(), mtime=0)).decode()
+
+    def test_valid_source_uses_receiver_clock_and_strips_payload_fields(self):
+        snapshot = {"truncated": False}
+        janitor.attach_hosted_light_candidates(snapshot, self._raw(), now=NOW)
+        value = snapshot["hosted-light-candidates"]
+        self.assertEqual(value["generated_at"], "2026-09-22T16:59:40Z")
+        self.assertEqual(value["controller_generated_at"], "2026-09-22T16:59:50Z")
+        self.assertEqual(value["receiver_last_delivery"], value["generated_at"])
+        self.assertEqual(set(value["runs"][0]), {
+            "id", "path", "event", "run_attempt", "status", "head_sha", "created_at",
+        })
+        self.assertNotIn("drop-me", json.dumps(snapshot))
+
+    def test_invalid_source_and_kill_switch_omit_optional_field(self):
+        for overrides in (
+            {"schema": "runner-state/v2"},
+            {"repository": "other/repo"},
+            {"generated_at": "2026-09-22T16:54:59Z"},
+            {"generated_at": "2026-09-22T17:00:06Z"},
+        ):
+            with self.subTest(overrides=overrides):
+                snapshot = {"truncated": False}
+                janitor.attach_hosted_light_candidates(snapshot, self._raw(state_overrides=overrides), now=NOW)
+                self.assertNotIn("hosted-light-candidates", snapshot)
+        snapshot = {"truncated": False}
+        janitor.attach_hosted_light_candidates(
+            snapshot, self._raw(candidate_overrides={"receiver_last_delivery": "2026-09-22T17:00:00Z"}), now=NOW)
+        self.assertNotIn("hosted-light-candidates", snapshot)
+        with mock.patch.dict("os.environ", {"CI_HOSTED_LIGHT_CANDIDATES_DISABLED": "1"}):
+            snapshot = {"truncated": False}
+            janitor.attach_hosted_light_candidates(snapshot, self._raw(), now=NOW)
+            self.assertNotIn("hosted-light-candidates", snapshot)
+
+    def test_outer_truncation_never_presents_candidate_field(self):
+        for snapshot in ({"truncated": True}, {"truncated": False, "starved_incomplete": True}):
+            janitor.attach_hosted_light_candidates(snapshot, self._raw(), now=NOW)
+            self.assertNotIn("hosted-light-candidates", snapshot)
+
+    def test_malformed_rows_are_dropped_and_candidate_remains_partial(self):
+        rows = [{"id": 1, "path": ".github/workflows/ci.yml", "event": "pull_request", "run_attempt": 1,
+                 "status": "queued", "head_sha": "b" * 40, "created_at": "2026-09-22T16:55:00Z"},
+                {"id": "bad", "path": "private.txt"}]
+        snapshot = {"truncated": False}
+        janitor.attach_hosted_light_candidates(snapshot, self._raw(rows=rows), now=NOW)
+        value = snapshot["hosted-light-candidates"]
+        self.assertEqual([row["id"] for row in value["runs"]], [1])
+        self.assertTrue(value["truncated"])
+        self.assertEqual(value["total_runs"], 2)
+
+    def test_encoded_and_decoded_bounds_omit_source(self):
+        snapshot = {"truncated": False}
+        janitor.attach_hosted_light_candidates(snapshot, "A" * (48 * 1024 + 1), now=NOW)
+        self.assertNotIn("hosted-light-candidates", snapshot)
+        oversized = {"x": "x" * (2 * 1024 * 1024 + 1)}
+        state = {"schema": "runner-state/v1", "repository": "manaflow-ai/cmux", "complete": True,
+                 "generated_at": "2026-09-22T16:59:30Z", "hosted_light_candidates": oversized}
+        raw = base64.b64encode(gzip.compress(json.dumps(state).encode(), mtime=0)).decode()
+        janitor.attach_hosted_light_candidates(snapshot, raw, now=NOW)
+        self.assertNotIn("hosted-light-candidates", snapshot)
+
+    def test_timestamp_overflow_omits_optional_source(self):
+        snapshot = {"truncated": False}
+        janitor.attach_hosted_light_candidates(
+            snapshot,
+            self._raw(candidate_overrides={"receiver_last_delivery": "0001-01-01T00:00:00+23:59"}),
+            now=NOW,
+        )
+        self.assertNotIn("hosted-light-candidates", snapshot)
+
+    def test_workflow_passes_runner_state_only_through_env(self):
+        steps = yaml.safe_load(WORKFLOW.read_text(encoding="utf-8"))["jobs"]["sweep"]["steps"]
+        step = next(item for item in steps if item.get("name") == "Cancel wasted macOS runs")
+        self.assertEqual(step["env"]["CI_RUNNER_STATE"], "${{ vars.CI_RUNNER_STATE }}")
+        self.assertEqual(step["env"]["CI_HOSTED_LIGHT_CANDIDATES_DISABLED"],
+                         "${{ vars.CI_HOSTED_LIGHT_CANDIDATES_DISABLED }}")
+        self.assertEqual(WORKFLOW.read_text(encoding="utf-8").count("vars.CI_RUNNER_STATE"), 1)
+
+
 NEXT_PATH = ".github/workflows/cmux-next.yml"
 NEXT_BRANCH = "feat-cmux-" + "next"
 LARGE_MAC = "blacksmith-12vcpu-macos-26"

tests/test_runner_label_policy.py

@@ -194,7 +194,8 @@ def test_a_non_string_value_is_ignored_rather_than_crashing(self) -> None:
 HEALTH_REPORT_WORKFLOW = ROOT / ".github" / "workflows" / "ci-health-report.yml"
 # RUNNER_POOLS is a JSON map of pool -> label, read as fromJSON(vars.RUNNER_POOLS).<pool>; its labels are
 # checked by scripts/ci/runner_pools.py --check (ci-repo-variables.yml), not by the label policy.
-NON_LABEL_RUNNER_VARIABLES = {"CI_SEED_KEEP_LOCAL_RUNNERS", "RUNNER_POOLS"}
+# CI_RUNNER_STATE is compressed discovery metadata, not a runs-on label.
+NON_LABEL_RUNNER_VARIABLES = {"CI_RUNNER_STATE", "CI_SEED_KEEP_LOCAL_RUNNERS", "RUNNER_POOLS"}
 
 
 def reported_runner_variables() -> set[str]: