JMR-devandClaude Opus 4.8
2fcee291ce
#350 made ci-trigger.yml dispatch ci.yml with the built-in GITHUB_TOKEN, on the claim that a workflow_dispatch is anti-recursion-exempt so no PAT is needed. In practice a GITHUB_TOKEN-triggered run is held in `action_required` awaiting manual approval and never runs un-attended, so auto-updated PRs' CI never ran (stalled #285). The original #349 design was right: dispatch with a PAT so the run executes as the authorized owner with no approval gate. - ci-trigger.yml: the trigger step's GH_TOKEN is now `${{ secrets.AUTOUPDATE_TOKEN || github.token }}` (was `${{ github.token }}`). AUTOUPDATE_TOKEN (the PAT) is REQUIRED for the scheduler; the `|| github.token` fallback stays fail-open but only starts CI if repo settings don't gate GITHUB_TOKEN-triggered runs. - autoupdate.yml: branch update stays on GITHUB_TOKEN (must NOT retrigger CI -- that would re-introduce the cascade). Clarified that AUTOUPDATE_TOKEN is still required by the repo (by ci-trigger.yml) so the secret isn't deleted. - Corrected the now-wrong "no PAT needed / workflow_dispatch anti-recursion-exempt" comments in ci-trigger.yml and the traffic_control.py docstrings. updates = GITHUB_TOKEN, triggering = PAT. Validation: all three workflow YAMLs parse clean; traffic-control unit tests still pass (59 tests) -- the change is workflow-env only, script logic unchanged. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
806 lines
36 KiB
Python
806 lines
36 KiB
Python
#!/usr/bin/env python3
|
||
# SPDX-License-Identifier: GPL-3.0-or-later
|
||
"""CI traffic-controller: priority-based runner orchestration for LibreMail's
|
||
`ci.yml`. Extracted out of the old inline-bash `traffic-control` step into a
|
||
Python module so the decision logic is developer-legible and, above all, unit-
|
||
testable (see `test_traffic_control.py`).
|
||
|
||
DESIGN: pure decision CORE + thin gh-I/O SHELL
|
||
----------------------------------------------
|
||
The decisions ("who do we cancel?", "do we proceed or wait?") are pure functions
|
||
over plain `PullRequest` snapshots — no network, no clock, no subprocess — so they
|
||
can be exercised exhaustively in unit tests. The SHELL (`run_live`) is the only part
|
||
that touches `gh`: it gathers the snapshot, applies the cancellations, and runs the
|
||
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). The dispatch uses the AUTOUPDATE_TOKEN PAT, NOT the built-in
|
||
GITHUB_TOKEN: a GITHUB_TOKEN-triggered run is held for MANUAL approval (`action_required`) and
|
||
never runs un-attended, whereas a PAT dispatch runs as the authorized owner with no approval gate
|
||
(#350's "no PAT needed" claim was wrong — see ci-trigger.yml + issue #351). 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
|
||
(P0 = highest), default `P5` if none. A `broken` OR `draft` PR is effectively P10
|
||
(bottom, below P9), overriding any P0-P9 label.
|
||
2. PASS 1 - preemption: a strictly-lower OTHER PR's in-progress / queued run is
|
||
cancelled iff THIS PR is P0 (an emergency reclaims ALL lower runners) OR the target
|
||
is broken/draft (a wasted run any higher-priority PR may reclaim). P1-P9 never bump
|
||
a *normal* lower run mid-flight — only a P0 does that.
|
||
3. PASS 2 - bounded hold-back: a non-P0 PR yields (cancels nothing) to any strictly-
|
||
higher-priority OTHER PR that has an active/queued run, and — among its OWN
|
||
priority level — to any PR ordered ahead of it (running-first, then oldest by
|
||
`createdAt`). It proceeds the moment it is at the front, or when the wait budget
|
||
elapses (a PR never blocks itself).
|
||
|
||
SAFETY INVARIANTS (preserved from the original step)
|
||
----------------------------------------------------
|
||
* never cancel a run on `main` / a push event — the shell's `gh run list` query
|
||
filters `--event pull_request` and drops `headBranch == main`, so only PR-event
|
||
runs ever reach the core;
|
||
* never cancel THIS PR's own run — skipped by PR number AND by run id;
|
||
* never cancel an equal-or-higher-priority PR — only strictly-lower (prio > self).
|
||
|
||
This is deliberately NOT a merge gate: every gh call is guarded, the shell always
|
||
exits 0, and the ci.yml step stays `continue-on-error`, so a hiccup (API error,
|
||
missing permission, fork PR) can never fail CI.
|
||
|
||
Pure standard library, cross-platform (the primary dev box is Windows, where the old
|
||
bash + `jq` pipeline had no clean equivalent).
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import os
|
||
import re
|
||
import subprocess
|
||
import sys
|
||
import time
|
||
from dataclasses import dataclass, field
|
||
|
||
# ── Priority model ───────────────────────────────────────────────────────────
|
||
BROKEN_LABEL = "broken"
|
||
DEFAULT_PRIORITY = 5 # a PR with no P0-P9 label
|
||
BOTTOM_PRIORITY = 10 # broken OR draft — below P9
|
||
TOP_PRIORITY = 0 # P0, the only priority that preempts
|
||
_P_LABEL = re.compile(r"^P([0-9])$") # single digit only, matching the old jq `^P[0-9]$`
|
||
|
||
# ── Run-status model (normalised from gh's raw run statuses) ─────────────────
|
||
RUNNING = "running" # gh status in_progress
|
||
QUEUED = "queued" # gh status queued / waiting / requested / pending
|
||
NONE = "none" # no active run (completed or absent)
|
||
ACTIVE = frozenset({RUNNING, QUEUED})
|
||
# For aggregating a branch's overall status from its runs: running beats queued
|
||
# beats none (most-active wins). NB: the same-level ORDER (see _ordering_key) is a
|
||
# coarser two-bucket split — in-flight (running) vs everything-else-by-age.
|
||
_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)
|
||
class PullRequest:
|
||
"""A snapshot of one open PR. The pure decision core consumes only these — no
|
||
network. `run_ids` are the PR's active (non-completed) CI run ids, already
|
||
filtered to pull_request events on a non-main head by the shell that built them."""
|
||
|
||
number: int
|
||
labels: tuple[str, ...] = ()
|
||
is_draft: bool = False
|
||
created_at: str = ""
|
||
run_status: str = NONE
|
||
run_ids: tuple[int, ...] = ()
|
||
|
||
@classmethod
|
||
def from_json(cls, obj: dict) -> "PullRequest":
|
||
"""Build from a snapshot dict. `labels` may be a list of names or of gh's
|
||
label objects (`{"name": ...}`)."""
|
||
raw_labels = obj.get("labels") or []
|
||
names: list[str] = []
|
||
for lab in raw_labels:
|
||
if isinstance(lab, dict):
|
||
name = lab.get("name")
|
||
else:
|
||
name = lab
|
||
if name:
|
||
names.append(str(name))
|
||
status = (obj.get("runStatus") or obj.get("run_status") or NONE).lower()
|
||
if status not in (RUNNING, QUEUED, NONE):
|
||
status = NONE
|
||
raw_ids = obj.get("runIds") or obj.get("run_ids") or ()
|
||
return cls(
|
||
number=int(obj["number"]),
|
||
labels=tuple(names),
|
||
is_draft=bool(obj.get("isDraft") or obj.get("is_draft")
|
||
or obj.get("draft") or False),
|
||
created_at=str(obj.get("createdAt") or obj.get("created_at") or ""),
|
||
run_status=status,
|
||
run_ids=tuple(int(r) for r in raw_ids),
|
||
)
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class Blocker:
|
||
"""A PR that THIS PR must yield to in PASS 2 (purely informational for logging)."""
|
||
|
||
number: int
|
||
priority: int
|
||
kind: str # "higher-priority" | "same-level-ahead"
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class Decision:
|
||
"""The full point-in-time decision for THIS PR (used by --dry-run and tests)."""
|
||
|
||
self_number: int
|
||
self_priority: int
|
||
cancel_run_ids: tuple[int, ...] = ()
|
||
blockers: tuple[Blocker, ...] = ()
|
||
proceed: bool = True
|
||
|
||
|
||
# ── Pure decision core (no network / clock / subprocess) ─────────────────────
|
||
def effective_priority(pr: PullRequest) -> int:
|
||
"""Effective priority: `broken` OR `draft` => 10 (bottom, overriding any P0-P9);
|
||
else the lowest-numbered P0-P9 label present; else the default P5."""
|
||
if pr.is_draft or BROKEN_LABEL in pr.labels:
|
||
return BOTTOM_PRIORITY
|
||
nums = [int(m.group(1)) for name in pr.labels if (m := _P_LABEL.match(name))]
|
||
return min(nums) if nums else DEFAULT_PRIORITY
|
||
|
||
|
||
def priority_label(prio: int) -> str:
|
||
"""Human-readable priority for logs."""
|
||
if prio >= BOTTOM_PRIORITY:
|
||
return f"P{BOTTOM_PRIORITY} (broken/draft — bottom, below P9)"
|
||
return f"P{prio}"
|
||
|
||
|
||
def _ordering_key(pr: PullRequest) -> tuple[int, str, int]:
|
||
"""Same-level ordering (issue #342 rule 3): an in-flight (RUNNING) run keeps its
|
||
place at the front — a same-level peer never reorders it — then, among the PRs
|
||
still waiting to start (QUEUED or no run yet), OLDEST createdAt first (ascending),
|
||
then PR number as a stable final tiebreak so the order is fully deterministic."""
|
||
in_flight = 0 if pr.run_status == RUNNING else 1
|
||
return (in_flight, pr.created_at or _FAR_FUTURE, pr.number)
|
||
|
||
|
||
def runs_to_cancel(
|
||
this_pr: PullRequest,
|
||
all_prs: list[PullRequest],
|
||
*,
|
||
self_run_id: int | None = None,
|
||
) -> list[int]:
|
||
"""PASS 1. Run ids to cancel. A strictly-lower OTHER PR's active (running/queued)
|
||
run is cancelled iff keeping it running is wasteful, i.e. EITHER:
|
||
* THIS PR is P0 — an emergency reclaims every strictly-lower runner now; OR
|
||
* the target is broken/draft (effective priority 10) — its run can't merge /
|
||
isn't merge-ready, so ANY higher-priority PR may reclaim its runner.
|
||
P1-P9 never cancel a *normal* strictly-lower run — they yield in PASS 2 instead.
|
||
Invariants: never cancel self (by number or run id), never cancel an
|
||
equal-or-higher-priority PR (only strictly-lower, prio > self)."""
|
||
self_prio = effective_priority(this_pr)
|
||
to_cancel: list[int] = []
|
||
seen: set[int] = set()
|
||
for pr in all_prs:
|
||
if pr.number == this_pr.number:
|
||
continue # never cancel self
|
||
target_prio = effective_priority(pr)
|
||
if target_prio <= self_prio:
|
||
continue # only strictly-lower (skip equal-or-higher)
|
||
if pr.run_status not in ACTIVE:
|
||
continue # nothing running/queued to cancel
|
||
# Strictly lower: preemptible iff we're P0 OR the target is broken/draft
|
||
# (a bottom, priority-10, wasted run that any higher PR may reclaim).
|
||
if self_prio != TOP_PRIORITY and target_prio < BOTTOM_PRIORITY:
|
||
continue # P1-P9 don't bump a *normal* lower run
|
||
for rid in pr.run_ids:
|
||
if self_run_id is not None and rid == self_run_id:
|
||
continue # never cancel our own run
|
||
if rid in seen:
|
||
continue
|
||
seen.add(rid)
|
||
to_cancel.append(rid)
|
||
return to_cancel
|
||
|
||
|
||
def wait_blockers(this_pr: PullRequest, all_prs: list[PullRequest]) -> list[Blocker]:
|
||
"""PASS 2. The PRs THIS PR must yield to right now (empty => proceed). P0 never
|
||
yields. Otherwise yield to (a) any strictly-higher-priority OTHER PR with an
|
||
active/queued run, and (b) any SAME-priority PR ordered ahead of THIS PR
|
||
(running-first, then oldest createdAt)."""
|
||
self_prio = effective_priority(this_pr)
|
||
if self_prio == TOP_PRIORITY:
|
||
return [] # P0 outranks everything — never wait
|
||
|
||
others = [pr for pr in all_prs if pr.number != this_pr.number]
|
||
blockers: list[Blocker] = []
|
||
|
||
# (a) strictly-higher-priority PRs that actually have an active/queued run.
|
||
for pr in others:
|
||
p = effective_priority(pr)
|
||
if p < self_prio and pr.run_status in ACTIVE:
|
||
blockers.append(Blocker(pr.number, p, "higher-priority"))
|
||
|
||
# (b) same-level ordering: THIS PR proceeds only when it is at the front.
|
||
same_level = [pr for pr in others if effective_priority(pr) == self_prio]
|
||
same_level.append(this_pr) # this_pr appears exactly once
|
||
for pr in sorted(same_level, key=_ordering_key):
|
||
if pr.number == this_pr.number:
|
||
break # reached self => nobody ahead remains
|
||
blockers.append(Blocker(pr.number, self_prio, "same-level-ahead"))
|
||
|
||
return blockers
|
||
|
||
|
||
def decide(
|
||
this_pr: PullRequest,
|
||
all_prs: list[PullRequest],
|
||
*,
|
||
self_run_id: int | None = None,
|
||
) -> Decision:
|
||
"""Convenience: the full point-in-time decision (both passes) for THIS PR."""
|
||
cancels = runs_to_cancel(this_pr, all_prs, self_run_id=self_run_id)
|
||
blockers = wait_blockers(this_pr, all_prs)
|
||
return Decision(
|
||
self_number=this_pr.number,
|
||
self_priority=effective_priority(this_pr),
|
||
cancel_run_ids=tuple(cancels),
|
||
blockers=tuple(blockers),
|
||
proceed=not blockers,
|
||
)
|
||
|
||
|
||
# ── 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)
|
||
|
||
|
||
def _gh_json(args: list[str]) -> list | dict | None:
|
||
"""Run `gh <args> --json ...` and parse stdout as JSON. Returns None (never
|
||
raises) on any failure — the caller fails open."""
|
||
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 None
|
||
if proc.returncode != 0:
|
||
_log(f"::warning::gh exited {proc.returncode} ({' '.join(args[:2])}): "
|
||
f"{proc.stderr.strip()}")
|
||
return None
|
||
try:
|
||
return json.loads(proc.stdout or "null")
|
||
except json.JSONDecodeError as exc:
|
||
_log(f"::warning::could not parse gh JSON ({' '.join(args[:2])}): {exc}")
|
||
return None
|
||
|
||
|
||
def _normalise_status(raw: str) -> str:
|
||
"""Map a gh run status onto our RUNNING / QUEUED / NONE model."""
|
||
if raw == "in_progress":
|
||
return RUNNING
|
||
if raw == "completed":
|
||
return NONE
|
||
return QUEUED # queued / waiting / requested / pending
|
||
|
||
|
||
def _runs_by_head(limit: int = 300) -> dict[str, dict]:
|
||
"""One bulk `gh run list` -> {headBranch: {"status", "ids"}} for active PR-event
|
||
runs. Enforces the 'never cancel main/push' invariant at the source: only
|
||
`event == pull_request`, non-`main`, non-completed runs are kept. Active runs are
|
||
the most recent, so `limit` most-recent runs comfortably covers them."""
|
||
rows = _gh_json([
|
||
"run", "list", "--workflow", "ci.yml", "--event", "pull_request",
|
||
"--limit", str(limit),
|
||
"--json", "databaseId,status,headBranch,event",
|
||
])
|
||
by_head: dict[str, dict] = {}
|
||
for row in rows or []:
|
||
if row.get("event") != "pull_request":
|
||
continue
|
||
head = row.get("headBranch")
|
||
if not head or head == "main":
|
||
continue
|
||
if row.get("status") == "completed":
|
||
continue
|
||
entry = by_head.setdefault(head, {"status": NONE, "ids": []})
|
||
entry["ids"].append(int(row["databaseId"]))
|
||
status = _normalise_status(row.get("status", ""))
|
||
# running beats queued beats none for the branch's aggregate status.
|
||
if _STATUS_RANK[status] < _STATUS_RANK[entry["status"]]:
|
||
entry["status"] = status
|
||
return by_head
|
||
|
||
|
||
def gather_snapshot(self_pr_number: int) -> tuple[PullRequest | None, list[PullRequest]]:
|
||
"""Build (this_pr, all_prs) from live gh data. this_pr is forced to RUNNING —
|
||
by definition our own run is in progress while this job executes."""
|
||
prs = _gh_json([
|
||
"pr", "list", "--state", "open", "--limit", "300",
|
||
"--json", "number,headRefName,labels,isDraft,createdAt",
|
||
])
|
||
if prs is None:
|
||
return None, []
|
||
runs = _runs_by_head()
|
||
all_prs: list[PullRequest] = []
|
||
this_pr: PullRequest | None = None
|
||
for obj in prs:
|
||
head = obj.get("headRefName") or ""
|
||
run_info = runs.get(head, {"status": NONE, "ids": []})
|
||
number = int(obj["number"])
|
||
is_self = number == self_pr_number
|
||
pr = PullRequest.from_json({
|
||
**obj,
|
||
# self is definitionally running (this job is in progress).
|
||
"runStatus": RUNNING if is_self else run_info["status"],
|
||
"runIds": run_info["ids"],
|
||
})
|
||
all_prs.append(pr)
|
||
if is_self:
|
||
this_pr = pr
|
||
return this_pr, all_prs
|
||
|
||
|
||
def _cancel_run(run_id: int) -> bool:
|
||
try:
|
||
proc = subprocess.run(
|
||
["gh", "run", "cancel", str(run_id)],
|
||
capture_output=True, text=True, check=False,
|
||
)
|
||
except (OSError, ValueError) as exc:
|
||
_log(f"::warning::could not cancel run {run_id}: {exc}")
|
||
return False
|
||
if proc.returncode == 0:
|
||
return True
|
||
_log(f"::warning::could not cancel run {run_id} — likely already finished. "
|
||
f"{proc.stderr.strip()}")
|
||
return False
|
||
|
||
|
||
def _positive_int(env_name: str, default: int) -> int:
|
||
raw = os.environ.get(env_name, "")
|
||
return int(raw) if raw.isdigit() and int(raw) > 0 else default
|
||
|
||
|
||
def run_live() -> int:
|
||
"""The gh-driven shell: gather, PASS 1 (cancel), PASS 2 (bounded hold-back). Always
|
||
returns 0 — the traffic-controller must never fail CI."""
|
||
event = os.environ.get("GITHUB_EVENT_NAME", "")
|
||
self_raw = os.environ.get("SELF_PR", "")
|
||
if event != "pull_request" or not self_raw.isdigit():
|
||
_log("Not a pull_request event (or no PR number) — nothing to do.")
|
||
return 0
|
||
self_number = int(self_raw)
|
||
self_run_id = int(os.environ["GITHUB_RUN_ID"]) if os.environ.get(
|
||
"GITHUB_RUN_ID", "").isdigit() else None
|
||
|
||
this_pr, all_prs = gather_snapshot(self_number)
|
||
if this_pr is None:
|
||
_log("::warning::Could not resolve THIS PR from the open-PR list — skipping.")
|
||
return 0
|
||
|
||
self_prio = effective_priority(this_pr)
|
||
_log(f"This PR #{self_number} effective priority: {priority_label(self_prio)} "
|
||
"(P0 = highest/emergency, P9 = lowest, broken/draft = bottom).")
|
||
|
||
# ── PASS 1: PREEMPTION (P0 reclaims all lower; anyone reclaims broken/draft) ──
|
||
to_cancel = runs_to_cancel(this_pr, all_prs, self_run_id=self_run_id)
|
||
if not to_cancel:
|
||
if self_prio == TOP_PRIORITY:
|
||
_log("P0 emergency — no strictly-lower active runs to cancel.")
|
||
else:
|
||
_log("No preemptible runs (P1-P9 only reclaim broken/draft lower runs; "
|
||
"none active).")
|
||
else:
|
||
cancelled = 0
|
||
for rid in to_cancel:
|
||
if _cancel_run(rid):
|
||
_log(f" cancelled run {rid} (freed its runner).")
|
||
cancelled += 1
|
||
_log(f"P0 preemption complete — cancelled {cancelled}/{len(to_cancel)} run(s).")
|
||
|
||
# ── PASS 2: BOUNDED HOLD-BACK (yield to higher / same-level-ahead) ────────
|
||
if self_prio == TOP_PRIORITY:
|
||
_log("P0 emergency — not yielding; proceeding immediately.")
|
||
return 0
|
||
|
||
budget = _positive_int("HOLD_BACK_BUDGET_SECONDS", 180)
|
||
poll = _positive_int("HOLD_BACK_POLL_SECONDS", 15)
|
||
deadline = time.monotonic() + budget
|
||
_log(f"{priority_label(self_prio)} — holding back up to {budget}s for higher / "
|
||
"earlier same-level PRs (no cancellation).")
|
||
|
||
while True:
|
||
remaining = deadline - time.monotonic()
|
||
if remaining <= 0:
|
||
_log("Hold-back budget elapsed — proceeding; higher-priority PRs got their "
|
||
"head start.")
|
||
break
|
||
# Refresh OTHER PRs so newly-opened higher-priority PRs are seen mid-wait;
|
||
# THIS PR's own identity/priority stays fixed (matching the original).
|
||
_, fresh = gather_snapshot(self_number)
|
||
if not fresh:
|
||
_log("::warning::Could not refresh open PRs — proceeding.")
|
||
break
|
||
blockers = wait_blockers(this_pr, fresh)
|
||
if not blockers:
|
||
_log("No higher-priority or earlier same-level PR is ahead — proceeding.")
|
||
break
|
||
tags = " ".join(f"#{b.number}(P{b.priority},{b.kind})" for b in blockers)
|
||
sleep_s = min(poll, int(remaining)) if remaining >= 1 else 0
|
||
_log(f"Yielding to: {tags} — re-checking in {sleep_s}s "
|
||
f"({int(remaining)}s budget left).")
|
||
if sleep_s > 0:
|
||
time.sleep(sleep_s)
|
||
|
||
_log("Hold-back complete — this PR's heavy jobs may now start.")
|
||
return 0
|
||
|
||
|
||
# ── --dry-run: feed the pure core a snapshot JSON, print its decisions ───────
|
||
def _load_snapshot(text: str) -> tuple[PullRequest, list[PullRequest], int | None]:
|
||
data = json.loads(text)
|
||
all_prs = [PullRequest.from_json(o) for o in data.get("prs", [])]
|
||
self_number = int(data["self"])
|
||
self_run_id = data.get("self_run_id")
|
||
self_run_id = int(self_run_id) if self_run_id is not None else None
|
||
this_pr = next((p for p in all_prs if p.number == self_number), None)
|
||
if this_pr is None:
|
||
raise ValueError(f"self #{self_number} not present in prs[]")
|
||
return this_pr, all_prs, self_run_id
|
||
|
||
|
||
def run_dry(text: str) -> int:
|
||
this_pr, all_prs, self_run_id = _load_snapshot(text)
|
||
dec = decide(this_pr, all_prs, self_run_id=self_run_id)
|
||
_log(f"This PR #{dec.self_number} effective priority: "
|
||
f"{priority_label(dec.self_priority)}")
|
||
if dec.cancel_run_ids:
|
||
why = ("P0 emergency (reclaims all strictly-lower)"
|
||
if dec.self_priority == TOP_PRIORITY
|
||
else "reclaiming broken/draft lower runs")
|
||
_log(f"PASS 1 (preemption): {why} — cancel run ids: "
|
||
f"{list(dec.cancel_run_ids)}")
|
||
else:
|
||
_log("PASS 1 (preemption): nothing to cancel.")
|
||
if dec.proceed:
|
||
_log("PASS 2 (hold-back): PROCEED — no blockers.")
|
||
else:
|
||
tags = ", ".join(f"#{b.number}(P{b.priority}, {b.kind})" for b in dec.blockers)
|
||
_log(f"PASS 2 (hold-back): WAIT — yielding to: {tags}")
|
||
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. The
|
||
dispatch runs as GH_TOKEN, which ci-trigger.yml sets to the AUTOUPDATE_TOKEN PAT: a run
|
||
triggered by the built-in GITHUB_TOKEN is held for MANUAL approval (`action_required`) and
|
||
never runs un-attended, so the PAT (authorized owner) is what actually starts the run with no
|
||
approval gate (see issue #351). 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).
|
||
try:
|
||
sys.stdout.reconfigure(encoding="utf-8", errors="replace")
|
||
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.")
|
||
parser.add_argument(
|
||
"--input", help="snapshot JSON file for --dry-run (default: stdin).")
|
||
args = parser.parse_args(argv)
|
||
|
||
if args.dry_run:
|
||
text = (open(args.input, encoding="utf-8").read() if args.input
|
||
else sys.stdin.read())
|
||
return run_trigger_dry(text) if args.mode == "trigger" else run_dry(text)
|
||
return run_trigger() if args.mode == "trigger" else run_live()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
try:
|
||
sys.exit(main())
|
||
except Exception as exc: # never let the controller fail CI
|
||
_log(f"::warning::traffic_control crashed, proceeding fail-open: {exc}")
|
||
sys.exit(0)
|