Merge pull request #350 from JMR-dev/ci-349-traffic-owns-triggering

ci: traffic-controller owns CI triggering (GITHUB_TOKEN updates, priority-ordered dispatch)

.github/scripts/test_traffic_control.py

@@ -7,7 +7,13 @@ preemption (P0 reclaims all strictly-lower; ANY higher PR reclaims a broken/draf
 lower run; P1-P9 never bump a *normal* lower run; self / main / equal-or-higher
 never cancelled), PASS 2 hold-back (yield to strictly-higher with an active run;
 same-level running-first then oldest-first), and a few end-to-end decision
-scenarios."""
+scenarios.
+
+Also covers the --mode trigger scheduler core (issue #349): head-SHA run
+classification (absent/cancelled => needy; success/failure => not needy),
+select_triggers (priority order, oldest-first fairness, inflight cap, fork skip,
+P0 bypasses-cap-and-preempts), and a liveness/anti-starvation simulation proving
+every eligible PR is triggered within a bounded number of passes."""
 
 from __future__ import annotations
 
@@ -283,5 +289,221 @@ class SnapshotParsingTests(unittest.TestCase):
         self.assertEqual(self_run_id, 222)
 
 
+# ── --mode trigger scheduler core (issue #349) ───────────────────────────────
+def npr(number, labels=("P5",), *, draft=False, created_at="", status=tc.NONE, run_ids=()):
+    """Terse builder defaulting to a P5 PR (for the trigger tests)."""
+    return pr(number, labels, draft=draft, created_at=created_at, status=status,
+              run_ids=run_ids)
+
+
+class ClassifyShaRunsTests(unittest.TestCase):
+    def test_no_runs_is_needy(self):
+        status, ids, needy = tc.classify_sha_runs([])
+        self.assertEqual(status, tc.NONE)
+        self.assertEqual(ids, ())
+        self.assertTrue(needy)                     # absent checks => must be triggered
+
+    def test_success_verdict_not_needy(self):
+        _, ids, needy = tc.classify_sha_runs(
+            [{"status": "completed", "conclusion": "success", "databaseId": 1}])
+        self.assertEqual(ids, ())
+        self.assertFalse(needy)
+
+    def test_failure_verdict_not_needy(self):
+        # A real failure is the author's to fix — never auto-retriggered (no fail loop).
+        _, _, needy = tc.classify_sha_runs(
+            [{"status": "completed", "conclusion": "failure", "databaseId": 1}])
+        self.assertFalse(needy)
+
+    def test_only_cancelled_is_needy(self):
+        # Cancelled leaves no verdict → re-trigger so the PR can reach a mergeable state.
+        _, ids, needy = tc.classify_sha_runs(
+            [{"status": "completed", "conclusion": "cancelled", "databaseId": 1}])
+        self.assertEqual(ids, ())
+        self.assertTrue(needy)
+
+    def test_in_progress_is_active_not_needy(self):
+        status, ids, needy = tc.classify_sha_runs(
+            [{"status": "in_progress", "conclusion": None, "databaseId": 9}])
+        self.assertEqual(status, tc.RUNNING)
+        self.assertEqual(ids, (9,))
+        self.assertFalse(needy)
+
+    def test_queued_is_active_not_needy(self):
+        status, ids, needy = tc.classify_sha_runs(
+            [{"status": "queued", "conclusion": None, "databaseId": 8}])
+        self.assertEqual(status, tc.QUEUED)
+        self.assertEqual(ids, (8,))
+        self.assertFalse(needy)
+
+    def test_running_beats_queued_in_status(self):
+        status, ids, _ = tc.classify_sha_runs([
+            {"status": "queued", "databaseId": 1},
+            {"status": "in_progress", "databaseId": 2},
+        ])
+        self.assertEqual(status, tc.RUNNING)
+        self.assertEqual(sorted(ids), [1, 2])
+
+    def test_cancelled_plus_active_not_needy(self):
+        # An active run already covers the SHA — cancelled siblings don't make it needy.
+        _, ids, needy = tc.classify_sha_runs([
+            {"status": "completed", "conclusion": "cancelled", "databaseId": 1},
+            {"status": "in_progress", "databaseId": 2},
+        ])
+        self.assertEqual(ids, (2,))
+        self.assertFalse(needy)
+
+
+class SelectTriggersTests(unittest.TestCase):
+    def test_empty_needy_triggers_nothing(self):
+        dec = tc.select_triggers([npr(1), npr(2)], set(), max_inflight=3)
+        self.assertEqual(dec.trigger_numbers, ())
+
+    def test_single_needy_triggered(self):
+        dec = tc.select_triggers([npr(1)], {1}, max_inflight=3)
+        self.assertEqual(dec.trigger_numbers, (1,))
+        self.assertEqual(dec.cancel_run_ids, ())
+
+    def test_priority_order(self):
+        prs = [npr(1, ["P5"]), npr(2, ["P2"]), npr(3, ["P8"])]
+        dec = tc.select_triggers(prs, {1, 2, 3}, max_inflight=3)
+        self.assertEqual(dec.trigger_numbers, (2, 1, 3))      # P2, P5, P8
+
+    def test_same_level_oldest_first(self):
+        prs = [
+            npr(1, ["P5"], created_at="2026-07-03T00:00:00Z"),
+            npr(2, ["P5"], created_at="2026-07-01T00:00:00Z"),
+            npr(3, ["P5"], created_at="2026-07-02T00:00:00Z"),
+        ]
+        dec = tc.select_triggers(prs, {1, 2, 3}, max_inflight=3)
+        self.assertEqual(dec.trigger_numbers, (2, 3, 1))      # oldest createdAt first
+
+    def test_cap_limits_triggers(self):
+        prs = [npr(1, ["P2"]), npr(2, ["P3"]), npr(3, ["P4"])]
+        dec = tc.select_triggers(prs, {1, 2, 3}, max_inflight=2)
+        self.assertEqual(dec.trigger_numbers, (1, 2))         # only 2 free slots
+        self.assertEqual(dec.slots, 2)
+
+    def test_inflight_consumes_slots(self):
+        prs = [
+            npr(1, ["P5"], status=tc.RUNNING, run_ids=[100]),  # inflight — occupies a slot
+            npr(2, ["P2"]),
+            npr(3, ["P3"]),
+        ]
+        dec = tc.select_triggers(prs, {2, 3}, max_inflight=2)
+        self.assertEqual(dec.slots, 1)                        # 2 cap - 1 inflight
+        self.assertEqual(dec.trigger_numbers, (2,))           # highest-priority needy only
+        self.assertEqual(dec.inflight_numbers, (1,))
+
+    def test_running_needy_is_not_retriggered(self):
+        # Defensive: a PR flagged needy but already running is never a candidate.
+        prs = [npr(1, ["P5"], status=tc.RUNNING, run_ids=[100])]
+        dec = tc.select_triggers(prs, {1}, max_inflight=3)
+        self.assertEqual(dec.trigger_numbers, ())
+
+    def test_fork_pr_skipped(self):
+        dec = tc.select_triggers([npr(1, ["P2"]), npr(2, ["P1"])],
+                                 {1, 2}, max_inflight=3, forks={2})
+        self.assertEqual(dec.trigger_numbers, (1,))           # fork #2 not token-triggerable
+        self.assertEqual(dec.skipped_fork_numbers, (2,))
+
+    def test_p0_bypasses_cap_and_preempts_lower(self):
+        # Cap full (a P5 running), but a needy P0 still triggers AND preempts the strictly-
+        # lower running run to free a runner immediately.
+        prs = [
+            npr(1, ["P5"], status=tc.RUNNING, run_ids=[500]),  # inflight, strictly-lower
+            npr(2, ["P0"]),                                     # needy emergency
+        ]
+        dec = tc.select_triggers(prs, {2}, max_inflight=1)
+        self.assertEqual(dec.slots, 0)                        # cap is full
+        self.assertEqual(dec.trigger_numbers, (2,))           # P0 bypasses the cap
+        self.assertEqual(dec.cancel_run_ids, (500,))          # preempts the lower run
+
+    def test_p0_does_not_preempt_equal_priority(self):
+        prs = [
+            npr(1, ["P0"], status=tc.RUNNING, run_ids=[500]),  # equal P0 — spared
+            npr(2, ["P0"]),                                     # needy emergency
+        ]
+        dec = tc.select_triggers(prs, {2}, max_inflight=1)
+        self.assertEqual(dec.trigger_numbers, (2,))
+        self.assertEqual(dec.cancel_run_ids, ())              # never preempts an equal P0
+
+    def test_needy_numbers_reports_all_candidates_in_order(self):
+        dec = tc.select_triggers([npr(1, ["P5"]), npr(2, ["P2"])], {1, 2}, max_inflight=1)
+        self.assertEqual(dec.needy_numbers, (2, 1))           # priority order, cap-independent
+        self.assertEqual(dec.trigger_numbers, (2,))           # but only 1 slot triggered
+
+    def test_zero_cap_is_clamped_to_one(self):
+        # A misconfigured cap must never stall everything: clamp to >= 1 so at least the
+        # top-priority needy PR still gets a slot (fail-safe forward progress).
+        dec = tc.select_triggers([npr(1, ["P5"])], {1}, max_inflight=0)
+        self.assertEqual(dec.max_inflight, 1)
+        self.assertEqual(dec.trigger_numbers, (1,))
+
+
+class TriggerStarvationTests(unittest.TestCase):
+    def test_every_needy_pr_is_triggered_within_bounded_passes(self):
+        # Liveness / anti-starvation: with a fixed needy set and cap=2, simulate scheduler
+        # passes where a triggered PR gains a run (leaves the needy set) and its run finishes
+        # one pass later (freeing its slot). Mixed priorities prove lower-priority PRs are
+        # served too — never starved — while higher-priority PRs still go first.
+        labels = {1: ["P1"], 2: ["P1"], 3: ["P5"], 4: ["P5"],
+                  5: ["P8"], 6: ["P8"], 7: ["P5"]}
+        created = {n: f"2026-07-{n:02d}T00:00:00Z" for n in labels}
+        needy = set(labels)
+        running: dict[int, int] = {}                # number -> passes left running
+        triggered_ever: set[int] = set()
+        first_pass: dict[int, int] = {}
+        cap, max_passes = 2, 12
+        for pass_no in range(1, max_passes + 1):
+            snap = [
+                pr(n, labels[n], created_at=created[n],
+                   status=tc.RUNNING if n in running else tc.NONE,
+                   run_ids=[1000 + n] if n in running else [])
+                for n in labels
+            ]
+            dec = tc.select_triggers(snap, needy, max_inflight=cap)
+            for n in dec.trigger_numbers:
+                triggered_ever.add(n)
+                first_pass.setdefault(n, pass_no)
+                needy.discard(n)                    # gained a run -> no longer needy
+                running[n] = 1                      # occupies a slot for one pass
+            for n in list(running):                 # running PRs finish after one pass
+                running[n] -= 1
+                if running[n] <= 0:
+                    del running[n]
+            if not needy and not running:
+                break
+        self.assertEqual(triggered_ever, set(labels),
+                         "a PR was starved (never triggered)")
+        # Priority respected: the two P1s go in the very first pass; everything else later.
+        self.assertTrue(all(first_pass[n] == 1 for n in (1, 2)))
+        self.assertTrue(all(first_pass[n] >= 2 for n in (3, 4, 5, 6, 7)))
+
+
+class TriggerSnapshotParsingTests(unittest.TestCase):
+    def test_load_trigger_snapshot_roundtrip(self):
+        text = json.dumps({
+            "max_inflight": 2,
+            "needy": [1, 3],
+            "forks": [3],
+            "prs": [
+                {"number": 1, "labels": ["P2"], "createdAt": "2026-07-01T00:00:00Z"},
+                {"number": 2, "labels": ["P5"], "runStatus": "running", "runIds": [22]},
+                {"number": 3, "labels": ["P1"], "createdAt": "2026-07-02T00:00:00Z"},
+            ],
+        })
+        all_prs, needy, forks, cap = tc._load_trigger_snapshot(text)
+        self.assertEqual(len(all_prs), 3)
+        self.assertEqual(needy, {1, 3})
+        self.assertEqual(forks, {3})
+        self.assertEqual(cap, 2)
+        dec = tc.select_triggers(all_prs, needy, max_inflight=cap, forks=forks)
+        # #2 is inflight (uses a slot); #3 is a fork (skipped); only #1 fits the 1 free slot.
+        self.assertEqual(dec.inflight_numbers, (2,))
+        self.assertEqual(dec.skipped_fork_numbers, (3,))
+        self.assertEqual(dec.trigger_numbers, (1,))
+
+
 if __name__ == "__main__":
     unittest.main()

.github/scripts/traffic_control.py

@@ -14,6 +14,30 @@ that touches `gh`: it gathers the snapshot, applies the cancellations, and runs
 bounded hold-back poll loop. Feed the core a snapshot JSON (`--dry-run`) to see its
 decisions with zero network.
 
+TWO MODES
+---------
+* ``--mode orchestrate`` (default; unchanged behaviour): the in-run `traffic-control`
+  job of `ci.yml`. Orders runner ACCESS for the PR whose run is already executing —
+  PASS 1 preemption + PASS 2 hold-back (below). This is `run_live`.
+* ``--mode trigger`` (issue #349): the *scheduler* (companion `ci-trigger.yml`, run
+  after each auto-update and on a cron backstop). It OWNS CI *triggering*: it
+  (re-)triggers CI for the highest-priority PR(s) whose head SHA has absent/stale
+  required checks — a few at a time (an inflight cap), in the SAME priority order —
+  via a `workflow_dispatch`. This is `run_trigger` / the pure `select_triggers`.
+
+WHY --mode trigger EXISTS (issue #349): `autoupdate.yml` now updates PR branches with the
+built-in GITHUB_TOKEN instead of a PAT, so an update push no longer auto-retriggers CI
+(GitHub's anti-recursion rule) — killing the merge-cascade that cancelled every open PR's
+run on every merge. The cost is that a freshly-updated PR's required checks go stale/absent
+on its NEW head SHA, so this scheduler deliberately (re-)triggers them in priority order (a
+poor-man's merge queue). A `workflow_dispatch` is EXEMPT from the anti-recursion rule, so
+even the GITHUB_TOKEN's dispatch DOES start the run — no PAT needed (the workflow grants its
+token `actions: write`). FAIL-OPEN, structurally: `ci.yml` KEEPS its `on: pull_request`
+trigger, so any human push — and a brand-new PR — always gets CI regardless of this
+scheduler; the scheduler only fills the gap left by GITHUB_TOKEN auto-updates and can never
+leave a PR un-triggerable. Fork PRs (no token/secret access) are skipped by the scheduler and
+left to `on: pull_request`, so they are never wedged either.
+
 ORDER OF OPERATIONS (issue #342)
 --------------------------------
 1. Effective priority orders everything: the lowest-numbered `P0`-`P9` label present
@@ -75,6 +99,15 @@ _STATUS_RANK = {RUNNING: 0, QUEUED: 1, NONE: 2}
 # createdAt sentinel so a PR with an unknown timestamp sorts LAST (never wrongly
 # "oldest"/front, so it yields rather than preempts another PR's front slot).
 _FAR_FUTURE = "9999-12-31T23:59:59Z"
+# Run conclusions that count as a FINAL VERDICT on a head SHA (issue #349, --mode trigger).
+# A SHA with one of these is NOT re-triggered: success = green, failure/timeout/etc. = the
+# author's to fix — auto-retriggering a real failure would waste runners and could loop.
+# Everything else a completed run can report (cancelled / skipped / stale / startup_failure /
+# null) is treated as "no verdict", so a SHA whose only runs are those — or that has no run at
+# all (absent checks after a GITHUB_TOKEN auto-update) — is NEEDY and gets (re-)triggered.
+VERDICT_CONCLUSIONS = frozenset(
+    {"success", "failure", "timed_out", "action_required", "neutral"}
+)
 
 
 @dataclass(frozen=True)
@@ -250,6 +283,116 @@ def decide(
     )
 
 
+# ── Pure TRIGGER-decision core (issue #349, --mode trigger) ──────────────────
+def classify_sha_runs(runs: list[dict]) -> tuple[str, tuple[int, ...], bool]:
+    """PURE. Summarise the CI runs on ONE head SHA. Returns (run_status, active_run_ids,
+    needy):
+      * run_status: RUNNING if any run is in progress, else QUEUED if any is queued/pending,
+        else NONE;
+      * active_run_ids: databaseIds of the non-completed (running/queued) runs;
+      * needy: True iff the SHA has NO active run AND NO run with a final VERDICT — i.e. its
+        required checks are absent/stale (a fresh SHA after a GITHUB_TOKEN auto-update) or
+        only cancelled/infra-aborted, so the PR cannot merge until CI is (re-)triggered on
+        that SHA. A success/failure/timeout verdict is NOT needy (green, or the author's to
+        fix — never auto-retried)."""
+    status = NONE
+    active_ids: list[int] = []
+    has_verdict = False
+    for r in runs:
+        raw = (r.get("status") or "").lower()
+        if raw == "completed":
+            if (r.get("conclusion") or "").lower() in VERDICT_CONCLUSIONS:
+                has_verdict = True
+            continue
+        norm = _normalise_status(raw)          # in_progress -> running; else queued
+        if _STATUS_RANK[norm] < _STATUS_RANK[status]:
+            status = norm
+        rid = r.get("databaseId")
+        if rid is not None:
+            active_ids.append(int(rid))
+    needy = not active_ids and not has_verdict
+    return status, tuple(active_ids), needy
+
+
+def _trigger_order_key(pr: PullRequest) -> tuple[int, str, int]:
+    """Trigger ordering: highest priority first (lowest effective-priority number), then
+    OLDEST createdAt first (the longest-waiting PR at a level goes first — the same-level
+    fairness / anti-starvation rule), then PR number as a stable final tiebreak."""
+    return (effective_priority(pr), pr.created_at or _FAR_FUTURE, pr.number)
+
+
+@dataclass(frozen=True)
+class TriggerDecision:
+    """PURE output of `select_triggers`: which PR(s) the scheduler should (re-)trigger CI
+    for right now, in order, plus any strictly-lower runs a P0 emergency preempts to free a
+    runner. Exercised by `--mode trigger --dry-run` and the unit tests."""
+
+    trigger_numbers: tuple[int, ...] = ()
+    cancel_run_ids: tuple[int, ...] = ()
+    inflight_numbers: tuple[int, ...] = ()
+    needy_numbers: tuple[int, ...] = ()
+    skipped_fork_numbers: tuple[int, ...] = ()
+    slots: int = 0
+    max_inflight: int = 0
+
+
+def select_triggers(
+    all_prs: list[PullRequest],
+    needy: "set[int] | frozenset[int]",
+    *,
+    max_inflight: int,
+    forks: "set[int] | frozenset[int]" = frozenset(),
+) -> TriggerDecision:
+    """PURE. Choose the PR(s) to (re-)trigger CI for now — a poor-man's merge queue over the
+    existing priority model. No network / clock / subprocess, so it is exhaustively unit-
+    tested (see TestSelectTriggers / TestTriggerStarvation).
+
+    * inflight = PRs already running/queued on their head SHA — they occupy the cap.
+    * candidates = NEEDY PRs (absent/stale checks on their head SHA) that are not already
+      running and are not forks (forks have no token/secret access — see `run_trigger`).
+    * order = effective priority, then oldest createdAt, then number (`_trigger_order_key`).
+    * P0 = EMERGENCY: always triggered, BYPASSING the cap, and it PREEMPTS its strictly-lower
+      OTHER runs (reusing `runs_to_cancel`) so a runner frees for it immediately.
+    * P1–P10 fill only the remaining ``slots = max_inflight - len(inflight)``; the rest wait
+      for a later pass.
+
+    STARVATION is bounded, not by aging but structurally: triggering a PR gives its head SHA
+    a run, so it LEAVES the needy set; between merges the needy set only shrinks, and the
+    scheduler re-runs on every auto-update plus a cron backstop, so every eligible PR is
+    triggered within a bounded number of passes (proved by TestTriggerStarvation). Ordering
+    is still by priority, so higher-priority PRs are simply served first, never exclusively
+    forever (a served PR stops being needy until its next push/auto-update)."""
+    cap = max(1, max_inflight)
+    inflight = [p for p in all_prs if p.run_status in ACTIVE]
+    candidates = [
+        p for p in all_prs
+        if p.number in needy and p.run_status not in ACTIVE and p.number not in forks
+    ]
+    ordered = sorted(candidates, key=_trigger_order_key)
+    emergencies = [p for p in ordered if effective_priority(p) == TOP_PRIORITY]
+    normal = [p for p in ordered if effective_priority(p) != TOP_PRIORITY]
+    slots = max(0, cap - len(inflight))
+    chosen = emergencies + normal[:slots]      # P0 bypasses the cap; P1–P10 fill free slots
+
+    cancel_ids: list[int] = []
+    seen: set[int] = set()
+    for emergency in emergencies:              # P0 preempts its strictly-lower active runs
+        for rid in runs_to_cancel(emergency, all_prs):
+            if rid not in seen:
+                seen.add(rid)
+                cancel_ids.append(rid)
+
+    return TriggerDecision(
+        trigger_numbers=tuple(p.number for p in chosen),
+        cancel_run_ids=tuple(cancel_ids),
+        inflight_numbers=tuple(sorted(p.number for p in inflight)),
+        needy_numbers=tuple(p.number for p in ordered),
+        skipped_fork_numbers=tuple(sorted(n for n in needy if n in forks)),
+        slots=slots,
+        max_inflight=cap,
+    )
+
+
 # ── gh I/O shell (the only part that touches the network) ────────────────────
 def _log(msg: str) -> None:
     print(msg, flush=True)
@@ -475,6 +618,155 @@ def run_dry(text: str) -> int:
     return 0
 
 
+# ── --mode trigger: the PAT-free scheduler shell (issue #349) ────────────────
+def _gh_ok(args: list[str]) -> bool:
+    """Run `gh <args>` for its side effect (no JSON parse). Returns True on exit 0; never
+    raises — the scheduler fails open on any I/O error."""
+    try:
+        proc = subprocess.run(["gh", *args], capture_output=True, text=True, check=False)
+    except (OSError, ValueError) as exc:
+        _log(f"::warning::gh invocation failed ({' '.join(args[:2])}): {exc}")
+        return False
+    if proc.returncode != 0:
+        _log(f"::warning::gh exited {proc.returncode} ({' '.join(args[:3])}): "
+             f"{proc.stderr.strip()}")
+        return False
+    return True
+
+
+def _ci_workflow_file() -> str:
+    return os.environ.get("CI_WORKFLOW_FILE", "ci.yml")
+
+
+def gather_trigger_snapshot() -> tuple[list[PullRequest], set[int], set[int], dict[int, dict]]:
+    """Build (all_prs, needy, forks, meta) from live gh data for the trigger scheduler.
+      * all_prs: PullRequest snapshots whose run_status / run_ids reflect the runs on each
+        PR's CURRENT head SHA (so 'inflight' means a live run on the mergeable SHA, never a
+        stale one on a superseded SHA);
+      * needy: PR numbers whose head SHA has absent/stale checks (must be (re-)triggered);
+      * forks: cross-repository PR numbers — no token/secret access, so NOT token-triggerable;
+      * meta: number -> {headRefName, headRefOid} for the dispatch I/O.
+    Returns empty structures (never raises) if gh can't be reached — the caller fails open."""
+    prs = _gh_json([
+        "pr", "list", "--state", "open", "--limit", "300",
+        "--json", "number,headRefName,headRefOid,isCrossRepository,labels,isDraft,createdAt",
+    ])
+    if prs is None:
+        return [], set(), set(), {}
+    runs = _gh_json([
+        "run", "list", "--workflow", _ci_workflow_file(), "--limit", "300",
+        "--json", "databaseId,status,conclusion,headSha,headBranch,event",
+    ]) or []
+    by_sha: dict[str, list[dict]] = {}
+    for row in runs:
+        sha = row.get("headSha")
+        if sha:
+            by_sha.setdefault(sha, []).append(row)
+
+    all_prs: list[PullRequest] = []
+    needy: set[int] = set()
+    forks: set[int] = set()
+    meta: dict[int, dict] = {}
+    for obj in prs:
+        number = int(obj["number"])
+        sha = obj.get("headRefOid") or ""
+        status, run_ids, is_needy = classify_sha_runs(by_sha.get(sha, []))
+        all_prs.append(PullRequest.from_json(
+            {**obj, "runStatus": status, "runIds": list(run_ids)}))
+        meta[number] = {
+            "headRefName": obj.get("headRefName") or "",
+            "headRefOid": sha,
+        }
+        if obj.get("isCrossRepository"):
+            forks.add(number)
+        if is_needy:
+            needy.add(number)
+    return all_prs, needy, forks, meta
+
+
+def _dispatch_ci(pr_number: int, head_ref: str, head_sha: str) -> bool:
+    """Trigger `ci.yml` for one PR via a `workflow_dispatch` on the PR's head branch. A
+    workflow_dispatch is exempt from GitHub's anti-recursion rule, so even the built-in
+    GITHUB_TOKEN's dispatch DOES start a run — no PAT required (the workflow grants its token
+    `actions: write`). Running on the head branch puts the run's checks on the PR head SHA, so
+    they satisfy branch protection's required checks."""
+    if not head_ref:
+        _log(f"::warning::PR #{pr_number} has no head branch — cannot dispatch; skipping.")
+        return False
+    ok = _gh_ok([
+        "workflow", "run", _ci_workflow_file(), "--ref", head_ref,
+        "-f", f"pr={pr_number}",
+        "-f", f"head_sha={head_sha}",
+        "-f", "reason=traffic-controller",
+    ])
+    if ok:
+        short = head_sha[:8] if head_sha else "?"
+        _log(f"    triggered CI for #{pr_number} on {head_ref} (head {short}).")
+    return ok
+
+
+def run_trigger() -> int:
+    """The scheduler shell (companion `ci-trigger.yml`): pick the highest-priority needy
+    PR(s) within the inflight cap and (re-)trigger their CI via workflow_dispatch; a P0
+    emergency additionally preempts its strictly-lower runs. ALWAYS returns 0 — the scheduler
+    must never wedge CI, and structurally it cannot: `ci.yml` keeps `on: pull_request`, so any
+    human push (and a brand-new PR) still gets CI independently of this scheduler."""
+    max_inflight = _positive_int("MAX_INFLIGHT_RUNS", 2)
+    all_prs, needy, forks, meta = gather_trigger_snapshot()
+    if not all_prs:
+        _log("No open PRs (or could not list them) — nothing to trigger.")
+        return 0
+
+    dec = select_triggers(all_prs, needy, max_inflight=max_inflight, forks=forks)
+    _log(f"Open PRs: {len(all_prs)} | needy (absent/stale checks): {list(dec.needy_numbers)} "
+         f"| inflight: {list(dec.inflight_numbers)} | cap {dec.max_inflight}, "
+         f"free slots {dec.slots}.")
+    if dec.skipped_fork_numbers:
+        _log(f"Fork PR(s) needing CI left to `on: pull_request` (no token access — not "
+             f"wedged): {list(dec.skipped_fork_numbers)}.")
+
+    for rid in dec.cancel_run_ids:                  # P0 emergency preemption
+        if _cancel_run(rid):
+            _log(f"    P0 preemption: cancelled lower run {rid} (freed its runner).")
+
+    if not dec.trigger_numbers:
+        _log("Nothing to trigger this pass (no needy PR fits a free slot).")
+        return 0
+    triggered = 0
+    for number in dec.trigger_numbers:
+        info = meta.get(number, {})
+        if _dispatch_ci(number, info.get("headRefName", ""), info.get("headRefOid", "")):
+            triggered += 1
+    _log(f"Trigger pass complete — dispatched {triggered}/{len(dec.trigger_numbers)} "
+         "run(s) in priority order.")
+    return 0
+
+
+def _load_trigger_snapshot(
+    text: str,
+) -> tuple[list[PullRequest], set[int], set[int], int]:
+    data = json.loads(text)
+    all_prs = [PullRequest.from_json(o) for o in data.get("prs", [])]
+    needy = {int(n) for n in data.get("needy", [])}
+    forks = {int(n) for n in data.get("forks", [])}
+    max_inflight = int(data.get("max_inflight", 2))
+    return all_prs, needy, forks, max_inflight
+
+
+def run_trigger_dry(text: str) -> int:
+    all_prs, needy, forks, max_inflight = _load_trigger_snapshot(text)
+    dec = select_triggers(all_prs, needy, max_inflight=max_inflight, forks=forks)
+    _log(f"Trigger decision (cap {dec.max_inflight}, free slots {dec.slots}):")
+    _log(f"  inflight (occupying the cap): {list(dec.inflight_numbers)}")
+    _log(f"  needy candidates (priority order): {list(dec.needy_numbers)}")
+    if dec.skipped_fork_numbers:
+        _log(f"  skipped forks (no token access): {list(dec.skipped_fork_numbers)}")
+    if dec.cancel_run_ids:
+        _log(f"  P0 preemption — cancel run ids: {list(dec.cancel_run_ids)}")
+    _log(f"  => TRIGGER (in priority order): {list(dec.trigger_numbers)}")
+    return 0
+
+
 def main(argv: list[str] | None = None) -> int:
     # Emit UTF-8 regardless of the host console so the log typography is stable on
     # the UTF-8 CI runners (and never raises on a legacy Windows code page).
@@ -483,6 +775,11 @@ def main(argv: list[str] | None = None) -> int:
     except (AttributeError, ValueError):
         pass
     parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument(
+        "--mode", choices=("orchestrate", "trigger"), default="orchestrate",
+        help="orchestrate (default): in-run runner-priority for the executing PR "
+             "(unchanged). trigger: the scheduler that (re-)triggers CI by priority "
+             "(issue #349).")
     parser.add_argument(
         "--dry-run", action="store_true",
         help="read a snapshot JSON (from --input or stdin), print decisions, no network.")
@@ -493,8 +790,8 @@ def main(argv: list[str] | None = None) -> int:
     if args.dry_run:
         text = (open(args.input, encoding="utf-8").read() if args.input
                 else sys.stdin.read())
-        return run_dry(text)
+        return run_trigger_dry(text) if args.mode == "trigger" else run_dry(text)
-    return run_live()
+    return run_trigger() if args.mode == "trigger" else run_live()
 
 
 if __name__ == "__main__":

.github/workflows/autoupdate.yml

@@ -6,13 +6,17 @@ name: Auto-update PR branches
 # touched (PR_FILTER: all) — this is no longer limited to PRs with GitHub auto-merge
 # enabled.
 #
-# IMPORTANT: for the branch update to RE-TRIGGER the PR's CI (so it can pass and merge),
+# IMPORTANT (issue #349): the branch update runs with the default GITHUB_TOKEN — ON PURPOSE.
-# this must run with a PAT, not the default GITHUB_TOKEN — pushes made by GITHUB_TOKEN do
+# A GITHUB_TOKEN push does NOT start new workflow runs (GitHub's anti-recursion rule), so
-# not start new workflow runs (GitHub's anti-recursion rule), so the updated PR would sit
+# updating every behind PR here NO LONGER re-triggers every PR's CI. That deliberately breaks
-# with stale checks. Create a fine-grained PAT scoped to this repo with
+# the old merge-cascade (every merge -> autoupdate rebases all PRs with a PAT -> all re-run ->
-# contents:read/write + pull-requests:read/write and add it as the AUTOUPDATE_TOKEN secret.
+# ci.yml's cancel-in-progress kills each in-flight run -> PRs thrash and can't converge).
-# Without it this falls back to GITHUB_TOKEN, which updates the branch but will NOT re-run
+# Branches still go up to date (satisfying "require branches up to date"); they just don't
-# the PR's checks.
+# auto-run CI on the new head SHA. Re-triggering that SHA's CI is now OWNED by the traffic-
+# controller scheduler (`.github/workflows/ci-trigger.yml` -> `traffic_control.py --mode
+# trigger`), which triggers the updated PRs deliberately, in priority order, a few at a time.
+# So this workflow must NOT use the PAT for the update push (that would re-introduce the
+# cascade). The AUTOUPDATE_TOKEN secret is no longer needed by this workflow.
 
 on:
   push:
@@ -38,7 +42,9 @@ jobs:
       - name: Update all behind PRs
         uses: chinthakagodawita/autoupdate@0707656cd062a3b0cf8fa9b2cda1d1404d74437e # v1.7.0
         env:
-          GITHUB_TOKEN: ${{ secrets.AUTOUPDATE_TOKEN || secrets.GITHUB_TOKEN }}
+          # Default GITHUB_TOKEN — NOT a PAT — so this update push does not auto-retrigger CI
+          # (anti-recursion). See the header: re-triggering is owned by ci-trigger.yml.
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
           PR_FILTER: "all"
           PR_READY_STATE: "all"
           MERGE_CONFLICT_ACTION: "ignore"

.github/workflows/ci-trigger.yml

@@ -0,0 +1,85 @@
+# SPDX-License-Identifier: GPL-3.0-or-later
+name: CI trigger (traffic-controller)
+
+# The traffic-controller SCHEDULER (issue #349). It OWNS CI *triggering*. After main advances,
+# autoupdate.yml updates every behind PR's branch with the built-in GITHUB_TOKEN which, by
+# GitHub's anti-recursion rule, does NOT start CI — so those PRs sit with absent/stale required
+# checks on their new head SHA and cannot merge. This workflow then (re-)triggers CI for the
+# highest-priority such PR(s), a few at a time (an inflight cap), in the existing P0–P9 /
+# broken-draft priority order — a poor-man's merge queue that replaces the old "every merge
+# re-runs every PR" thundering herd (the cascade; see the ci-merge-cascade note + issue #349).
+#
+# HOW IT TRIGGERS: `traffic_control.py --mode trigger` runs `gh workflow run ci.yml --ref
+# <pr-head-branch>`. A workflow_dispatch is EXEMPT from the anti-recursion rule, so even the
+# built-in GITHUB_TOKEN's dispatch DOES start the run — no PAT is required (this job grants its
+# token `actions: write`). The dispatched run executes on the PR's head branch, so its checks
+# land on the PR head SHA and satisfy branch protection's required checks.
+#
+# WHEN IT RUNS:
+#   • workflow_run, after "Auto-update PR branches" completes — the race-free moment: autoupdate
+#     has finished moving branches to their new (checkless) head SHAs, so this pass sees exactly
+#     the PRs that now need a run. (A bare `push: main` trigger would race autoupdate and often
+#     read the pre-update SHAs, missing them until the next pass.)
+#   • schedule (cron) — a backstop so no PR is ever permanently un-triggered even if a
+#     workflow_run is missed/skipped (part of the fail-open guarantee), and so a brand-new PR
+#     whose first `on: pull_request` run got cancelled is still picked up.
+#   • workflow_dispatch — manual kick.
+#
+# FAIL-OPEN: the script guards every gh call and always exits 0; and structurally, ci.yml keeps
+# its `on: pull_request` trigger, so a human push (and a brand-new PR) always triggers CI
+# regardless of this scheduler — CI can never become permanently un-triggerable. Fork PRs (no
+# token/secret access) are skipped here and left to `on: pull_request`, so they are never wedged.
+
+on:
+  workflow_run:
+    workflows: ["Auto-update PR branches"]
+    types: [completed]
+  schedule:
+    # Backstop cadence (UTC). GitHub may delay scheduled runs under load; that is fine — this
+    # is only a safety net behind the immediate workflow_run trigger above.
+    - cron: "*/15 * * * *"
+  workflow_dispatch:
+
+# Trigger-only; this workflow never gates a merge. `actions: write` lets the built-in
+# GITHUB_TOKEN dispatch ci.yml (workflow_dispatch) and cancel strictly-lower runs when a P0
+# emergency preempts. `pull-requests: read` + `contents: read` cover the PR/label enumeration.
+permissions:
+  contents: read
+  pull-requests: read
+  actions: write
+
+# One trigger pass at a time. Do NOT cancel an in-flight pass (cancel-in-progress: false):
+# a half-finished pass could leave some needy PRs un-triggered until the next pass.
+concurrency:
+  group: ci-trigger
+  cancel-in-progress: false
+
+jobs:
+  trigger:
+    name: Trigger CI by priority
+    runs-on: ubuntu-latest
+    timeout-minutes: 10 # generous backstop; the script only enumerates + dispatches, no waits
+    steps:
+      - name: Check out source
+        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+
+      - name: Set up Python
+        uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 # v6.3.0
+        with:
+          python-version: "3.x"
+
+      # gh is auto-configured from GH_TOKEN / GH_REPO. The script guards every gh call and
+      # always exits 0, so a hiccup (API error, missing permission, fork PR) can never wedge CI
+      # — and even a total failure here leaves ci.yml's `on: pull_request` path intact.
+      - name: Trigger CI for the highest-priority PR(s) needing a run
+        env:
+          GH_TOKEN: ${{ github.token }}
+          GH_REPO: ${{ github.repository }}
+          # Poor-man's merge-queue width: at most this many PRs run CI concurrently under the
+          # scheduler (a P0 emergency bypasses this cap). Kept conservative because each PR
+          # fans out to the whole E2E matrix (~8 API levels + preview); this is the main knob
+          # to raise for throughput vs runner budget. The coordinator drives runner allocation.
+          MAX_INFLIGHT_RUNS: "2"
+          # The workflow file the scheduler enumerates runs for and dispatches.
+          CI_WORKFLOW_FILE: "ci.yml"
+        run: python3 .github/scripts/traffic_control.py --mode trigger

.github/workflows/ci.yml

@@ -4,10 +4,33 @@ name: CI
 on:
   pull_request:
     branches: [main]
+  # The traffic-controller SCHEDULER (ci-trigger.yml, issue #349) (re-)triggers CI for a
+  # specific PR via this workflow_dispatch after a GITHUB_TOKEN auto-update has left the PR's
+  # head SHA with absent/stale checks. Dispatched on the PR's head BRANCH, so the run's checks
+  # land on the PR head SHA and satisfy branch protection. `on: pull_request` above is KEPT so
+  # brand-new PRs, human pushes, and fork PRs still get CI directly — this is the fail-open
+  # guarantee: CI is always triggerable even if the scheduler is broken or absent.
+  workflow_dispatch:
+    inputs:
+      pr:
+        description: "PR number this run is for (set by the traffic-controller scheduler)."
+        required: false
+        type: string
+      head_sha:
+        description: "Expected head SHA (informational, for traceability in the run log)."
+        required: false
+        type: string
+      reason:
+        description: "Why this run was dispatched (informational)."
+        required: false
+        type: string
 
-# A new push to a PR cancels any in-flight run for that PR.
+# A new trigger for a PR cancels that PR's own in-flight run (a newer head SHA supersedes).
+# The group is keyed to the PR NUMBER so a `pull_request` run and a scheduler
+# `workflow_dispatch` run for the SAME PR share one concurrency group (either supersedes a
+# stale run of the other); it falls back to the ref when no PR number is in context.
 concurrency:
-  group: ci-${{ github.ref }}
+  group: ci-pr-${{ github.event.pull_request.number || inputs.pr || github.ref }}
   cancel-in-progress: true
 
 permissions:
@@ -86,7 +109,13 @@ jobs:
     env:
       GH_TOKEN: ${{ github.token }}
       GH_REPO: ${{ github.repository }}
-      SELF_PR: ${{ github.event.pull_request.number }}
+      # On a `pull_request` run this is the PR number and the script does its full in-run
+      # runner-priority orchestration. On a scheduler `workflow_dispatch` run (issue #349)
+      # the event is not `pull_request`, so the script no-ops here (`--mode orchestrate`
+      # only acts on pull_request events) — priority was ALREADY applied at trigger time by
+      # ci-trigger.yml, so re-doing the in-run hold-back would just waste runner time. The
+      # `|| inputs.pr` keeps the number in the log for a dispatched run.
+      SELF_PR: ${{ github.event.pull_request.number || inputs.pr }}
       # P1–P9 bounded hold-back knobs, read by traffic_control.py. BUDGET must stay
       # comfortably below timeout-minutes so the poll loop always exits 0 before the
       # hard job timeout fires — a timed-out job would skip the heavy jobs and fail