From 5af9cb7f9dcec8263e638bfcabddcbb028620c3c Mon Sep 17 00:00:00 2001 From: Leo Li Date: Wed, 4 Nov 2026 10:27:45 +0000 Subject: [PATCH 1/3] feat(ci): publish hosted light candidates in pool artifact Co-authored-by: Codex <199175422+chatgpt-codex-connector[bot]@users.noreply.github.com> Agent-Session: https://cmux.com/deeplink/agent?id=aid-a312e19f20f444de50b7 --- .github/workflows/ci-queue-janitor.yml | 4 + scripts/ci/queue_janitor.py | 136 +++++++++++++++++++++++++ tests/test_ci_queue_janitor.py | 95 +++++++++++++++++ 3 files changed, 235 insertions(+) diff --git a/.github/workflows/ci-queue-janitor.yml b/.github/workflows/ci-queue-janitor.yml index 114ccd390a23..1613cf5482f4 100644 --- a/.github/workflows/ci-queue-janitor.yml +++ b/.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 diff --git a/scripts/ci/queue_janitor.py b/scripts/ci/queue_janitor.py index d6e9a84292a1..40b0ad068f65 100644 --- a/scripts/ci/queue_janitor.py +++ b/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--xcode-``) 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, 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---p-, where @@ -4420,6 +4552,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"]: diff --git a/tests/test_ci_queue_janitor.py b/tests/test_ci_queue_janitor.py index 125668d4be0a..f535cc66a66c 100644 --- a/tests/test_ci_queue_janitor.py +++ b/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 @@ -3986,6 +3988,99 @@ 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_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" From bdd4cd0bafd76e12aa845a3959d75c919f99091c Mon Sep 17 00:00:00 2001 From: Leo Li Date: Wed, 4 Nov 2026 10:45:23 +0000 Subject: [PATCH 2/3] fix(ci): omit malformed runner-state timestamps Co-authored-by: Codex <199175422+chatgpt-codex-connector[bot]@users.noreply.github.com> Agent-Session: https://cmux.com/deeplink/agent?id=aid-a312e19f20f444de50b7 --- scripts/ci/queue_janitor.py | 2 +- tests/test_ci_queue_janitor.py | 9 +++++++++ 2 files changed, 10 insertions(+), 1 deletion(-) diff --git a/scripts/ci/queue_janitor.py b/scripts/ci/queue_janitor.py index 40b0ad068f65..4e06e3ea34a2 100644 --- a/scripts/ci/queue_janitor.py +++ b/scripts/ci/queue_janitor.py @@ -988,7 +988,7 @@ def hosted_light_candidates(raw: str | None, *, now: dt.datetime) -> dict[str, A "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, json.JSONDecodeError, zlib.error): + except (ValueError, TypeError, OSError, EOFError, OverflowError, json.JSONDecodeError, zlib.error): return None diff --git a/tests/test_ci_queue_janitor.py b/tests/test_ci_queue_janitor.py index f535cc66a66c..16cc24177ee9 100644 --- a/tests/test_ci_queue_janitor.py +++ b/tests/test_ci_queue_janitor.py @@ -4072,6 +4072,15 @@ def test_encoded_and_decoded_bounds_omit_source(self): 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") From 1b381fb837b330e1f3e98f7319d7929add37b793 Mon Sep 17 00:00:00 2001 From: Leo Li Date: Wed, 4 Nov 2026 11:34:37 +0000 Subject: [PATCH 3/3] test(ci): classify runner state as discovery metadata Co-authored-by: Codex <199175422+chatgpt-codex-connector[bot]@users.noreply.github.com> Agent-Session: https://cmux.com/deeplink/agent?id=aid-a312e19f20f444de50b7 --- tests/test_runner_label_policy.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/tests/test_runner_label_policy.py b/tests/test_runner_label_policy.py index 70224e737ccd..116e4a1374d6 100644 --- a/tests/test_runner_label_policy.py +++ b/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).; 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]: