diff --git a/scripts/ci/dispatch-focused-test.py b/scripts/ci/dispatch-focused-test.py index 51b7e8077e0a..38c023df8d40 100644 --- a/scripts/ci/dispatch-focused-test.py +++ b/scripts/ci/dispatch-focused-test.py @@ -22,6 +22,9 @@ RUN_DISCOVERY_TIMEOUT_SECONDS = 60.0 PRIOR_ATTEMPT_LIMIT = 100 PRIOR_ATTEMPT_TIMEOUT_SECONDS = 30.0 +# Statuses GitHub reports before a run has a conclusion. Anything else, +# including a missing status, is not treated as occupying a runner. +UNFINISHED = frozenset({"queued", "in_progress", "waiting", "requested", "pending"}) RUNNERS = ( "auto", "blacksmith-6vcpu-macos-15", @@ -122,55 +125,164 @@ def cancel(_signum, _frame): signal.signal(signum, handler) -def prior_attempts(commit: str, selector: str, runner: str | None = None) -> list[dict]: - """Completed runs of this selector/commit, scoped to an explicit runner. +def recent_dispatches() -> list[dict]: + """Recent dispatches of this workflow, or nothing when history is unreadable. - A focused run compiles the tree before it runs anything, so a red result is - often a property of the commit and runner, not of the attempt. Preserve - the existing broad guard for the default/auto runner, but a failure on - macOS 15 must not block an explicitly requested macOS 26 verification. - Re-dispatching the same selector/SHA/runner can reprint the same failure. - The run name carries the dispatch identity -- - " on @ []" -- so earlier attempts - are findable without recording any local state. + One listing answers every pre-dispatch question, for every selector in a + batch. Asking per selector repeated the same request once per entry and + spent shared GitHub API budget to receive the same page back. """ try: payload = output( "gh", "run", "list", "--repo", REPO, "--workflow", WORKFLOW, "--event", "workflow_dispatch", "--limit", str(PRIOR_ATTEMPT_LIMIT), - "--json", "displayTitle,conclusion,status,url", + "--json", "databaseId,displayTitle,conclusion,status,url", timeout=PRIOR_ATTEMPT_TIMEOUT_SECONDS, ) except (subprocess.SubprocessError, OSError, ValueError): - # The guard is an economy measure, never a gate. If the history cannot - # be read, dispatch as before. + # These guards are economy measures, never gates. If the history + # cannot be read, dispatch as before. return [] try: runs = json.loads(payload) except json.JSONDecodeError: return [] - marker = f" @ {commit} [" - - def ran_selector(title: str) -> bool: - # A batched dispatch names several selectors before " on ", so match - # membership rather than a prefix. Otherwise batching would silently - # bypass this guard for every selector it carried. - head, separator, remainder = title.partition(" on ") - if not separator: - return False - if runner not in (None, "auto") and not remainder.startswith(f"{runner} @ "): - return False - return selector in [part.strip() for part in head.split(",")] + if not isinstance(runs, list): + return [] + return [run for run in runs if isinstance(run, dict)] + + +def parse_run_name(title: str) -> tuple[list[str], str, str] | None: + """Split " on @ []" into its parts. + + The dispatch id is optional: a run started from the GitHub UI, or by any + tool that does not pass one, still names its selectors, runner and ref. + Requiring the trailing "[" hid exactly the runs whose compile these guards + exist to protect, because a run with the same ref and filter shares this + workflow's concurrency group whether or not a dispatcher labelled it. + """ + head, separator, remainder = title.partition(" on ") + if not separator: + return None + runner, separator, remainder = remainder.partition(" @ ") + if not separator: + return None + ref = remainder.split(" [", 1)[0].strip() + return [part.strip() for part in head.split(",")], runner.strip(), ref + + +def default_runner() -> str | None: + """The label `runner: auto` resolves to, or None when it cannot be known. + + The workflow reads `vars.MACOS_RUNNER_TESTS` and falls back to a literal + written beside it, so the answer lives half in the repository's variables + and half in the workflow definition. Read both rather than hard-coding + either: the literal moves when the default pool moves, and the variable + overrides it without touching the workflow. + + Returning None means "cannot tell", and every caller treats that as a + reason to dispatch normally rather than to act on a runner it guessed. + """ + try: + payload = output( + "gh", "variable", "list", "--repo", REPO, "--json", "name,value", + timeout=PRIOR_ATTEMPT_TIMEOUT_SECONDS, + ) + variables = json.loads(payload) + except (subprocess.SubprocessError, OSError, ValueError, json.JSONDecodeError): + return None + if not isinstance(variables, list): + return None + for entry in variables: + if isinstance(entry, dict) and entry.get("name") == "MACOS_RUNNER_TESTS": + value = str(entry.get("value", "")).strip() + if value: + return value + break + try: + workflow = (ROOT / ".github/workflows" / WORKFLOW).read_text() + except OSError: + return None + literal = re.search( + r"vars\.MACOS_RUNNER_TESTS \|\| '([^']+)'", workflow + ) + return literal.group(1) if literal else None + + +def attempts( + runs: list[dict], commit: str, selector: str, runner: str | None = None +) -> list[dict]: + """Runs of this selector at this exact commit, newest first. + + `runner` narrows to one pool. None means every pool, which is what the + repeat guard wants: a red result is usually a property of the commit. + """ + found = [] + for run in runs: + parsed = parse_run_name(str(run.get("displayTitle", ""))) + if parsed is None: + continue + selectors, run_runner, ref = parsed + if ref != commit or selector not in selectors: + continue + if runner is not None and run_runner != runner: + continue + found.append(run) + return found + + +def prior_attempts( + runs: list[dict], commit: str, selector: str, runner: str | None = None +) -> list[dict]: + """Completed attempts, whose conclusion is already knowable. + A focused run compiles the tree before it runs anything, so a red result is + often a property of the commit and runner, not of the attempt. Preserve + the existing broad guard for the default/auto runner, but a failure on one + macOS generation must not block a verification explicitly asked of another + -- in either direction, since which generation `auto` means is a + repository variable and has moved before. + Re-dispatching the same selector/SHA/runner can reprint the same failure. + """ return [ - run for run in runs - if isinstance(run, dict) - and ran_selector(str(run.get("displayTitle", ""))) - and marker in str(run.get("displayTitle", "")) - and run.get("status") == "completed" + run for run in attempts(runs, commit, selector, runner) + if run.get("status") == "completed" ] +def live_attempts( + runs: list[dict], commit: str, selector: str, runner: str +) -> list[dict]: + """Attempts GitHub has accepted that have not reported a conclusion yet. + + Dispatching over one of these is worse than wasteful. The workflow's + concurrency group is keyed on runner, ref and the whole test_filter string + with `cancel-in-progress: true`, so an identical dispatch cancels the run + already compiling and starts that compile again from cold. A dispatch that + only overlaps -- a different batch naming one of the same selectors -- does + not collide, and instead pays a second full compile of identical source to + answer a question already in flight. + + `runner` is required and exact. A run on another pool shares neither the + concurrency group nor the question: reusing its result would report macOS + 15's answer to someone who asked about macOS 26. + """ + return [ + run for run in attempts(runs, commit, selector, runner) + if str(run.get("status", "")) in UNFINISHED + ] + + +def watchable(run: dict) -> bool: + """Whether this history entry carries enough to point a caller at the run. + + An entry without an id or a URL cannot be attached to or named, and these + guards never become a gate: a caller that cannot be redirected is dispatched. + """ + return (isinstance(run.get("databaseId"), int) + and bool(str(run.get("url", "")).strip())) + + def find_run( commit: str, selector: str, @@ -243,7 +355,8 @@ def main() -> int: parser.add_argument( "--force", action="store_true", - help="dispatch even if this selector already failed at this commit", + help="dispatch even if this selector already failed at this commit, " + "or is already running there", ) args = parser.parse_args() for entry in args.test_filter: @@ -279,10 +392,60 @@ def main() -> int: raise ValueError("GitHub revision differs from local HEAD; push the intended commit first") if not args.force: + history = recent_dispatches() + # Which pool this dispatch will actually land on. None means the + # answer could not be established, and the in-flight guards below stay + # silent rather than compare against a runner they guessed. + runner = args.runner if args.runner not in (None, "auto") else default_runner() + + if runner is not None: + # An identical dispatch is already answering this exact question on + # this exact pool. Attach to it instead of cancelling it: the + # concurrency group keyed on runner/ref/test_filter would kill the + # run mid-compile and start the same compile again from cold. + requested = set(args.test_filter) + running = [ + run for run in history + if str(run.get("status", "")) in UNFINISHED + and watchable(run) + and (parsed := parse_run_name(str(run.get("displayTitle", "")))) is not None + and parsed[2] == commit + and parsed[1] == runner + and set(parsed[0]) == requested + ] + if running: + live = running[0] + print( + f"{test_filter} is already {live['status']} at {commit} " + f"on {runner}; reusing that run instead of dispatching.", + flush=True, + ) + print(f"Run: {live['url']}", flush=True) + if args.wait: + return subprocess.run([ + "gh", "run", "watch", "--repo", REPO, str(live["databaseId"]), + "--exit-status", + ], cwd=ROOT).returncode + return 0 + # Refuse per entry: one already-red selector makes the whole batch a # reprint of a known failure, and the compile it would pay for is shared. for entry in args.test_filter: - earlier = prior_attempts(commit, entry, args.runner) + live = [run for run in live_attempts(history, commit, entry, runner) + if watchable(run)] if runner is not None else [] + if live: + raise ValueError( + f"{entry} is already {live[0]['status']} at {commit} on " + f"{runner}, in {live[0]['url']}, under a different set of " + "selectors. Dispatching now would compile identical source " + "a second time to answer a question already in flight. Wait " + "for that run, dispatch the remaining selectors on their " + "own, or pass --force." + ) + earlier = prior_attempts( + history, commit, entry, + args.runner if args.runner not in (None, "auto") else None, + ) failures = [run for run in earlier if run.get("conclusion") == "failure"] if failures and not any(run.get("conclusion") == "success" for run in earlier): latest = failures[0] diff --git a/tests/test_run_e2e.py b/tests/test_run_e2e.py index 6db9e4524467..8ae5028553de 100644 --- a/tests/test_run_e2e.py +++ b/tests/test_run_e2e.py @@ -2,6 +2,7 @@ """Exercise the focused-run launcher against a fake GitHub CLI.""" import importlib.util import json +import re import os from pathlib import Path import subprocess @@ -13,6 +14,10 @@ import yaml ROOT = Path(__file__).resolve().parents[1] +DEFAULT_RUNNER = re.search( + r"vars\.MACOS_RUNNER_TESTS \|\| '([^']+)'", + (ROOT / ".github/workflows/test-e2e.yml").read_text(), +).group(1) HEAD = "a" * 40 REMOTE_HEAD = "b" * 40 FAKE_GH = r'''#!/usr/bin/env python3 @@ -28,6 +33,8 @@ elif args[:2] == ["workflow", "run"]: fields = dict(arg.split("=", 1) for arg in args if "=" in arg) (root / "dispatch.json").write_text(json.dumps(fields)) +elif args[:2] == ["variable", "list"]: + print(os.environ.get("LAUNCHER_VARIABLES", "[]")) elif args[:2] == ["run", "list"]: if "conclusion" in " ".join(args): # The pre-dispatch repeat guard asks for conclusions; the post-dispatch @@ -183,6 +190,16 @@ def _prior(self, conclusion, *, selector="cmuxTests/ExampleTests", commit=HEAD, "url": "https://github.com/manaflow-ai/cmux/actions/runs/555", }]) + def _live(self, *, selector="cmuxTests/ExampleTests", commit=HEAD, + runner=DEFAULT_RUNNER, status="in_progress"): + return json.dumps([{ + "databaseId": 777, + "displayTitle": f"{selector} on {runner} @ {commit} [deadbeef]", + "conclusion": None, + "status": status, + "url": "https://github.com/manaflow-ai/cmux/actions/runs/777", + }]) + def test_failure_on_another_runner_allows_explicit_runner_proof(self): result = self.launch( "cmuxTests/ExampleTests", "--runner", "blacksmith-6vcpu-macos-26", @@ -268,6 +285,173 @@ def test_failure_at_a_different_commit_does_not_block(self): ) self.assertEqual(result.returncode, 0, result.stderr) + def test_identical_run_in_flight_is_reused_instead_of_dispatched(self): + # Dispatching here would match the workflow's concurrency group and + # cancel the run already compiling, restarting that compile from cold. + result = self.launch( + "cmuxTests/ExampleTests", LAUNCHER_PRIOR_RUNS=self._live() + ) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertIn("actions/runs/777", result.stdout) + self.assertFalse((self.root / "dispatch.json").exists(), "must not dispatch") + + def test_queued_identical_run_is_reused(self): + result = self.launch( + "cmuxTests/ExampleTests", + LAUNCHER_PRIOR_RUNS=self._live(status="queued"), + ) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertFalse((self.root / "dispatch.json").exists(), "must not dispatch") + + def test_reused_run_is_watched_and_reports_its_result(self): + result = self.launch( + "cmuxTests/ExampleTests", "--wait", + LAUNCHER_PRIOR_RUNS=self._live(), LAUNCHER_WATCH_STATUS="1", + ) + self.assertEqual(result.returncode, 1) + self.assertIn(["run", "watch", "--repo", "manaflow-ai/cmux", "777", "--exit-status"], + self.calls()) + self.assertFalse((self.root / "dispatch.json").exists(), "must not dispatch") + + def test_identical_batch_in_flight_is_reused_regardless_of_order(self): + live = self._live(selector="cmuxTests/ExampleTests,cmuxTests/AlphaTests") + result = self.launch( + "cmuxTests/AlphaTests", "cmuxTests/ExampleTests", + LAUNCHER_PRIOR_RUNS=live, + ) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertFalse((self.root / "dispatch.json").exists(), "must not dispatch") + + def test_overlapping_in_flight_batch_is_refused_rather_than_duplicated(self): + # A different batch does not share the concurrency group, so this would + # pay a second full compile of identical source for an answer already + # in flight. There is no single run to attach to, so refuse instead. + live = self._live(selector="cmuxTests/ExampleTests,cmuxTests/OtherTests") + result = self.launch("cmuxTests/ExampleTests", LAUNCHER_PRIOR_RUNS=live) + self.assertNotEqual(result.returncode, 0) + self.assertIn("already in_progress", result.stderr) + self.assertIn("actions/runs/777", result.stderr) + self.assertFalse((self.root / "dispatch.json").exists(), "must not dispatch") + + def test_force_dispatches_over_an_in_flight_run(self): + result = self.launch( + "cmuxTests/ExampleTests", "--force", LAUNCHER_PRIOR_RUNS=self._live() + ) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertEqual(self.dispatch()["test_filter"], "cmuxTests/ExampleTests") + + def test_in_flight_run_at_a_different_commit_does_not_block(self): + result = self.launch( + "cmuxTests/ExampleTests", + LAUNCHER_PRIOR_RUNS=self._live(commit="c" * 40), + ) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertEqual(self.dispatch()["ref"], HEAD) + + def test_in_flight_run_on_another_runner_does_not_block_explicit_runner(self): + result = self.launch( + "cmuxTests/ExampleTests", "--runner", "blacksmith-6vcpu-macos-26", + LAUNCHER_PRIOR_RUNS=self._live(runner="blacksmith-6vcpu-macos-15"), + ) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertEqual(self.dispatch()["runner"], "blacksmith-6vcpu-macos-26") + + def test_a_run_on_another_runner_is_never_reused_as_the_answer(self): + # The concurrency groups differ, so nothing would have been cancelled, + # and under --wait attaching would report macOS 15's result to someone + # who asked the default pool. Dispatch instead. + result = self.launch( + "cmuxTests/ExampleTests", "--wait", + LAUNCHER_PRIOR_RUNS=self._live(runner="tart-canary"), + LAUNCHER_WATCH_STATUS="0", + ) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertEqual(self.dispatch()["test_filter"], "cmuxTests/ExampleTests") + self.assertNotIn("actions/runs/777", result.stdout) + + def test_the_repository_variable_decides_which_runner_auto_means(self): + variables = json.dumps([{"name": "MACOS_RUNNER_TESTS", "value": "warp-macos-15-arm64-6x"}]) + # The workflow literal is now the wrong answer, so a run named for it + # must not be attached to... + result = self.launch( + "cmuxTests/ExampleTests", + LAUNCHER_PRIOR_RUNS=self._live(), LAUNCHER_VARIABLES=variables, + ) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertTrue((self.root / "dispatch.json").exists()) + # ...while a run named for the variable's value is. + self.setUp() + result = self.launch( + "cmuxTests/ExampleTests", + LAUNCHER_PRIOR_RUNS=self._live(runner="warp-macos-15-arm64-6x"), + LAUNCHER_VARIABLES=variables, + ) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertFalse((self.root / "dispatch.json").exists(), "must not dispatch") + + def test_a_run_without_a_dispatch_id_is_still_seen(self): + # A run started from the GitHub UI shares the concurrency group and its + # compile is just as real. Requiring the trailing "[" hid exactly the + # runs these guards exist to protect. + live = json.dumps([{ + "databaseId": 777, + "displayTitle": f"cmuxTests/ExampleTests on {DEFAULT_RUNNER} @ {HEAD}", + "conclusion": None, "status": "in_progress", + "url": "https://github.com/manaflow-ai/cmux/actions/runs/777", + }]) + result = self.launch("cmuxTests/ExampleTests", LAUNCHER_PRIOR_RUNS=live) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertIn("actions/runs/777", result.stdout) + self.assertFalse((self.root / "dispatch.json").exists(), "must not dispatch") + + def test_an_unattachable_entry_dispatches_rather_than_blocking(self): + # These guards are an economy measure, never a gate. An entry with no + # id cannot be watched, so the caller gets the run they asked for. + live = json.dumps([{ + "displayTitle": f"cmuxTests/ExampleTests on {DEFAULT_RUNNER} @ {HEAD} [deadbeef]", + "conclusion": None, "status": "in_progress", "url": "", + }]) + result = self.launch("cmuxTests/ExampleTests", LAUNCHER_PRIOR_RUNS=live) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertEqual(self.dispatch()["test_filter"], "cmuxTests/ExampleTests") + + def test_an_unknown_status_is_not_treated_as_occupying_a_runner(self): + result = self.launch( + "cmuxTests/ExampleTests", LAUNCHER_PRIOR_RUNS=self._live(status="") + ) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertTrue((self.root / "dispatch.json").exists()) + + def test_unreadable_variables_dispatch_rather_than_guess_a_runner(self): + result = self.launch( + "cmuxTests/ExampleTests", + LAUNCHER_PRIOR_RUNS=self._live(), LAUNCHER_VARIABLES="not json", + ) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertTrue((self.root / "dispatch.json").exists()) + + def test_a_malformed_history_payload_never_blocks_a_dispatch(self): + for payload in ("null", '{"runs": []}', '[null, 3]'): + with self.subTest(payload=payload): + self.setUp() + result = self.launch("cmuxTests/ExampleTests", LAUNCHER_PRIOR_RUNS=payload) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertTrue((self.root / "dispatch.json").exists()) + + def test_one_history_read_serves_every_selector_in_a_batch(self): + # The guards used to re-list runs once per entry, spending shared + # GitHub API budget to receive the same page back. + result = self.launch( + "cmuxTests/AlphaTests", "cmuxTests/ExampleTests", + LAUNCHER_PRIOR_RUNS="[]", + ) + self.assertEqual(result.returncode, 0, result.stderr) + guard_reads = [ + call for call in self.calls() + if call[:2] == ["run", "list"] and any("conclusion" in arg for arg in call) + ] + self.assertEqual(len(guard_reads), 1, guard_reads) +