ci: merge_group trigger, cancel-superseded and end-of-queue cache janitor (GEN-022)

.github/workflows/ci.yml

@@ -5,6 +5,10 @@ on:
     types: [opened, reopened, synchronize, labeled, unlabeled]
   push:
     branches: [main]
+  # GEN-022 (forge-ci/ci.yml#405/#432): main merges through a merge queue,
+  # which validates the exact merge commit with the quick lane.
+  merge_group:
+    types: [checks_requested]
   schedule:
     - cron: "0 7 * * *"
   workflow_dispatch:
@@ -27,6 +31,28 @@ concurrency:
   cancel-in-progress: ${{ github.ref != 'refs/heads/main' }}
 
 jobs:
+  # GEN-022 / forge-ci/ci.yml#409: a queue ref carries the base commit
+  # (gh-readonly-queue/main/pr-<N>-<sha>), so the concurrency group above never
+  # pairs a PR's rebuilt queue run with the one main's movement replaced; this
+  # cancels the older one. No `needs:`; only the cache janitor depends on it.
+  cancel-superseded:
+    name: Cancel superseded queue runs
+    if: ${{ !cancelled() && github.event_name == 'merge_group' }}
+    runs-on: ubuntu-24.04
+    timeout-minutes: 5
+    permissions:
+      contents: read
+      actions: write
+    steps:
+      - uses: actions/checkout@d23441a48e516b6c34aea4fa41551a30e30af803 # v6
+        with:
+          sparse-checkout: ci
+          persist-credentials: false
+      - name: Cancel
+        env:
+          GH_TOKEN: ${{ github.token }}
+        run: python3 -m ci.cancel_superseded
+
   # Read the exact PR head and the base's trust policy. Covered quick checks
   # reuse a verified local proof; audit samples retain remote execution.
   ci-mode:
@@ -105,3 +131,40 @@ jobs:
       actions: write
       contents: read
     uses: ./.github/workflows/cache-budget.yml
+
+  # GEN-022 (forge-ci/ci.yml#405): every queue run ends with the ci-lint cache
+  # janitor. It runs after every other job, even failed ones, so the repository
+  # stays under [cache].budget by our LRU order instead of GitHub's, which can
+  # evict a seed a required job restores. The queue serializes default-branch
+  # cache writes; the non-cancelling group keeps one sweep at a time across
+  # runs. Pinned to a ci-lint new enough to run the repo-wide budget LRU pass
+  # (forge-ci/ci.yml#407).
+  queue-cache-janitor:
+    name: End-of-queue cache janitor
+    needs: [cancel-superseded, ci-mode, linux-quick, linux-full, macos, windows, coverage, all-features, cache-budget]
+    if: ${{ always() && github.event_name == 'merge_group' }}
+    runs-on: ubuntu-24.04
+    timeout-minutes: 5
+    concurrency:
+      group: cache-janitor
+      cancel-in-progress: false
+    permissions:
+      contents: read
+      actions: write
+    steps:
+      - uses: actions/checkout@d23441a48e516b6c34aea4fa41551a30e30af803 # v6
+        with:
+          ref: ${{ github.sha }}
+          persist-credentials: false
+      - uses: actions/checkout@d23441a48e516b6c34aea4fa41551a30e30af803 # v6
+        with:
+          repository: zackees/ci.yml
+          ref: 953c4e13691b10c1c4f1f9c39fa259fd732c5789
+          path: .ci-lint
+          sparse-checkout: ci_lint
+          persist-credentials: false
+      - name: Evict down to the cache budget
+        env:
+          GITHUB_TOKEN: ${{ github.token }}
+          PYTHONPATH: .ci-lint
+        run: python3 -m ci_lint cache janitor --repo .

ci.toml

@@ -183,7 +183,9 @@ platform-selector = "crates/running-process-platform-internal/src/lib.rs"
 # PyPI) and ci.yml's / cache-budget.yml's `enforce-cache-budget` (actions:
 # write, to retire entries). Top-level permissions are capped at
 # contents:read + actions:read; any wider per-job grant must be listed here.
-permissions = { "id-token: write" = ["publish-pypi"], "actions: write" = ["enforce-cache-budget"] }
+# The merge queue's cancel-superseded and queue-cache-janitor jobs (GEN-022,
+# forge-ci/ci.yml#409/#405) also take actions: write.
+permissions = { "id-token: write" = ["publish-pypi"], "actions: write" = ["enforce-cache-budget", "cancel-superseded", "queue-cache-janitor"] }
 platform-code     = ["crates/running-process-platform-internal/src/**", "src/running_process/platforms/**"]
 
 # ── Publish ──────────────────────────────────────────────────────────────────

ci/cancel_superseded.py

@@ -0,0 +1,95 @@
+"""Cancel merge-queue runs that a newer queue run for the same PR replaced (#2048).
+
+    python -m ci.cancel_superseded [--dry-run]
+
+Runs as a job of every `merge_group` CI run (GITHUB_RUN_ID, GITHUB_REPOSITORY
+and a GH_TOKEN with actions: write). A queue ref carries the base commit,
+`gh-readonly-queue/main/pr-<N>-<sha>`, so the workflow's concurrency group
+never pairs a PR's rebuilt queue run with its old one. When main moves,
+GitHub starts a new run per queued PR and the old ones keep holding runners.
+The newest queue run for a PR is the live one, so every older in-flight
+queue run for the same PR is cancelled. Ported from obs-rust/obs-studio's
+`ci/cancel_superseded.py` (#986 there).
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import re
+import sys
+from urllib.error import HTTPError
+from urllib.request import Request, urlopen
+
+QUEUE_RE = re.compile(r"(?:^|/)gh-readonly-queue/[^/]+/pr-(\d+)-[0-9a-f]+$")
+LIVE = ("queued", "in_progress", "waiting", "pending", "requested")
+
+
+def queue_pr(ref: str) -> int | None:
+    """The PR number of a merge-queue ref, else None."""
+    match = QUEUE_RE.search(ref)
+    return int(match.group(1)) if match else None
+
+
+def superseded(runs: list[dict], current: dict) -> list[int]:
+    """Ids of in-flight queue runs for current's PR that started before it."""
+    pr = queue_pr(current.get("head_branch") or "")
+    if pr is None:
+        return []
+    return sorted(
+        {
+            run["id"]
+            for run in runs
+            if run["id"] != current["id"]
+            and run.get("status") in LIVE
+            and queue_pr(run.get("head_branch") or "") == pr
+            and run["created_at"] < current["created_at"]
+        }
+    )
+
+
+def api(path: str, method: str = "GET") -> dict:
+    request = Request(
+        f"https://api.github.com/{path}",
+        method=method,
+        headers={
+            "Authorization": f"Bearer {os.environ['GH_TOKEN']}",
+            "Accept": "application/vnd.github+json",
+        },
+    )
+    with urlopen(request, timeout=30) as response:
+        body = response.read()
+    return json.loads(body) if body else {}
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
+    parser.add_argument("--dry-run", action="store_true")
+    args = parser.parse_args(argv)
+    repo = os.environ["GITHUB_REPOSITORY"]
+    current = api(f"repos/{repo}/actions/runs/{os.environ['GITHUB_RUN_ID']}")
+    workflow = current["workflow_id"]
+    runs: list[dict] = []
+    for status in LIVE:
+        listing = api(
+            f"repos/{repo}/actions/workflows/{workflow}/runs"
+            f"?event=merge_group&status={status}&per_page=100"
+        )
+        runs += listing["workflow_runs"]
+    stale = superseded(runs, current)
+    pr = queue_pr(current.get("head_branch") or "")
+    print(f"PR #{pr}: cancelling {len(stale)} superseded queue run(s): {stale}")
+    for run_id in stale:
+        if args.dry_run:
+            continue
+        try:
+            api(f"repos/{repo}/actions/runs/{run_id}/cancel", method="POST")
+        except HTTPError as error:
+            # A run that finished meanwhile can no longer be cancelled (409).
+            print(f"run {run_id}: cancel returned {error.code}; leaving it")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

local-gate.toml

@@ -6,6 +6,9 @@
 [gate]
 run = ["uv", "run", "--no-project", "--python", "3.11", "python", "ci/local_gate.py"]
 verify = "ci.yml:ci-mode"
+# cancel-superseded (GEN-022, ci.yml#409) runs only on merge_group and must
+# run first, with no `needs:`.
+verify-exempt = ["cancel-superseded"]
 mode = "enforce"
 
 [gate.lanes.linux-quick]

tests/test_ci_lint.py

@@ -47,9 +47,7 @@ def test_aggregate_check_fails_when_a_requested_dylint_did_not_pass() -> None:
 
 
 def test_default_pr_path_does_not_run_macos_or_windows_dylint() -> None:
-    main = (ci_lint.ROOT / ".github" / "workflows" / "ci.yml").read_text(
-        encoding="utf-8"
-    )
+    main = (ci_lint.ROOT / ".github" / "workflows" / "ci.yml").read_text(encoding="utf-8")
     jobs = main.split("\n\n")
     for name in ("ci-macos.yml", "ci-windows.yml"):
         job = next(j for j in jobs if f"uses: ./.github/workflows/{name}" in j)
@@ -75,9 +73,7 @@ def test_local_lint_runs_both_dylint_libraries_in_ci_order() -> None:
     assert f'cargo-dylint-version: "{dylint_gate.DYLINT_VERSION}"' in reusable
 
 
-def test_dylint_gate_skips_without_toolchain_and_fails_when_required(
-    monkeypatch, capsys
-) -> None:
+def test_dylint_gate_skips_without_toolchain_and_fails_when_required(monkeypatch, capsys) -> None:
     from ci import dylint_gate
 
     monkeypatch.setattr(dylint_gate, "missing_tools", lambda: ["install-it"])
@@ -109,8 +105,7 @@ def fake_run(command, **kwargs):
     monkeypatch.setattr(
         dylint_gate.subprocess,
         "run",
-        lambda command, **kwargs: seen.append(command)
-        or type("Result", (), {"returncode": 0})(),
+        lambda command, **kwargs: seen.append(command) or type("Result", (), {"returncode": 0})(),
     )
     assert dylint_gate.main() == 0
     assert seen == dylint_gate.commands()
@@ -423,3 +418,30 @@ def test_main_runs_lint_commands_through_running_process_cli(monkeypatch) -> Non
         ],
         [python, "-m", "ci.dylint_gate"],
     ]
+
+
+def test_merge_queue_runs_cancel_superseded_first_and_ends_with_janitor() -> None:
+    """GEN-022 (forge-ci/ci.yml#405/#409): the queue's housekeeping shape."""
+    main = (ci_lint.ROOT / ".github" / "workflows" / "ci.yml").read_text(encoding="utf-8")
+    assert "  merge_group:\n    types: [checks_requested]\n" in main
+    jobs = main.split("\n\n")
+    cancel = next(j for j in jobs if "\n  cancel-superseded:\n" in j)
+    assert "github.event_name == 'merge_group'" in cancel
+    assert "\n    needs:" not in cancel
+    assert "actions: write" in cancel
+    assert "run: python3 -m ci.cancel_superseded" in cancel
+    janitor = next(j for j in jobs if "\n  queue-cache-janitor:\n" in j)
+    assert "always() && github.event_name == 'merge_group'" in janitor
+    assert "group: cache-janitor\n      cancel-in-progress: false" in janitor
+    assert "run: python3 -m ci_lint cache janitor --repo ." in janitor
+    ids = {
+        line.strip().rstrip(":")
+        for line in main.split("\njobs:\n", 1)[1].splitlines()
+        if line.startswith("  ") and not line.startswith("   ") and line.endswith(":")
+    }
+    needs = janitor.split("needs: [", 1)[1].split("]", 1)[0].split(", ")
+    assert set(needs) == ids - {"queue-cache-janitor"}
+    # Nothing else may wait on cancel-superseded.
+    assert main.count("cancel-superseded") == janitor.count("cancel-superseded") + (
+        cancel.count("cancel-superseded")
+    )