diff --git a/scripts/ci/current_head_run_coalescer.py b/scripts/ci/current_head_run_coalescer.py index ae40b85ac4..948c80cd01 100644 --- a/scripts/ci/current_head_run_coalescer.py +++ b/scripts/ci/current_head_run_coalescer.py @@ -29,6 +29,7 @@ API_TIMEOUT_SECONDS = 30 CANCELLATION_POLL_ATTEMPTS = 6 CANCELLATION_POLL_INTERVAL_SECONDS = 1.0 +QUEUE_START_RACE_RE = re.compile(r"\bHTTP\s*409\b") class CoalescingRefused(RuntimeError): @@ -373,7 +374,24 @@ def _fetch_run(repo: str, run_id: int) -> dict[str, Any]: def _cancel_run(repo: str, run_id: int) -> None: """Cancel one run and prove GitHub reached its terminal cancelled state.""" - _run_json(["gh", "api", "-X", "POST", f"repos/{repo}/actions/runs/{run_id}/cancel"]) + cancel_args = ["gh", "api", "-X", "POST", f"repos/{repo}/actions/runs/{run_id}/cancel"] + try: + _run_json(cancel_args) + except RuntimeError as exc: + # GitHub can race a queued run into startup between the candidate + # fetch and POST, returning HTTP 409 instead of accepting cancel. + # Re-read the authoritative run state; never turn an unknown + # cancellation error into a successful result or another mutation. + if not QUEUE_START_RACE_RE.search(str(exc)): + raise + current = _fetch_run(repo, run_id) + if current.get("status") == "completed" and current.get("conclusion") == "cancelled": + return + if current.get("status") != "queued": + raise CoalescingRefused(f"workflow run {run_id} is no longer queued after HTTP 409") from exc + raise CoalescingRefused( + f"workflow run {run_id} remained queued after HTTP 409; preserving it" + ) from exc for attempt in range(CANCELLATION_POLL_ATTEMPTS): run_data = _fetch_run(repo, run_id) if run_data.get("status") == "completed" and run_data.get("conclusion") == "cancelled": diff --git a/tests/test_current_head_run_coalescer.py b/tests/test_current_head_run_coalescer.py index 571368677f..136b373539 100644 --- a/tests/test_current_head_run_coalescer.py +++ b/tests/test_current_head_run_coalescer.py @@ -393,6 +393,7 @@ def test_run_json_uses_token_timeout_decodes_success_and_bounds_failure(monkeypa seen: dict[str, object] = {} def success(*args, **kwargs): + """Return bounded JSON while recording the subprocess timeout.""" seen.update(kwargs) return SimpleNamespace(returncode=0, stdout='{"ok":true}', stderr="") @@ -401,6 +402,7 @@ def success(*args, **kwargs): assert seen["timeout"] == module.API_TIMEOUT_SECONDS def timeout(*_args, **_kwargs): + """Raise the subprocess timeout sentinel for transport mapping.""" raise subprocess.TimeoutExpired(cmd="gh", timeout=30) monkeypatch.setattr(module.subprocess, "run", timeout) @@ -434,6 +436,7 @@ def test_fetch_helpers_fail_closed_and_paginate(monkeypatch) -> None: calls: list[list[str]] = [] def pages(args): + """Return two paginated workflow-run pages and then an empty page.""" calls.append(list(args)) status = next(item.split("=", 1)[1] for item in args if item.startswith("status=")) page = int(next(item.split("=", 1)[1] for item in args if item.startswith("page="))) @@ -473,6 +476,191 @@ def test_cancel_run_uses_explicit_transport_and_ordinary_endpoint(monkeypatch) - assert sleeps == [module.CANCELLATION_POLL_INTERVAL_SECONDS] +def test_cancel_run_preserves_started_run_after_cancel_409(monkeypatch) -> None: + """A run that started after the first POST is preserved without a second POST.""" + module = load_module() + cancel_calls = 0 + states = iter( + [ + {"status": "in_progress", "conclusion": None}, + ] + ) + + def run_json(args): + """Raise the queued-start race from the cancellation POST.""" + nonlocal cancel_calls + if args[-1].endswith("/cancel"): + cancel_calls += 1 + raise RuntimeError("gh: Cannot cancel a workflow run that has not been queued yet. (HTTP409)") + raise AssertionError(args) + + monkeypatch.setattr(module, "_run_json", run_json) + monkeypatch.setattr(module, "_fetch_run", lambda _repo, _run_id: next(states)) + with pytest.raises(module.CoalescingRefused, match="no longer queued"): + module._cancel_run("o/r", 123) + assert cancel_calls == 1 + + +def test_cancel_run_preserves_queued_run_after_cancel_409(monkeypatch) -> None: + """A queued run gets no compensating cancellation request after HTTP 409.""" + module = load_module() + cancel_calls = 0 + states = iter( + [ + {"status": "queued", "conclusion": None}, + {"status": "completed", "conclusion": "cancelled"}, + ] + ) + + def run_json(args): + """Raise the queued-start race while preserving the queued state.""" + nonlocal cancel_calls + if args[-1].endswith("/cancel"): + cancel_calls += 1 + raise RuntimeError("Cannot cancel a workflow run that has not been queued yet. (HTTP409)") + raise AssertionError(args) + + monkeypatch.setattr(module, "_run_json", run_json) + monkeypatch.setattr(module, "_fetch_run", lambda _repo, _run_id: next(states)) + with pytest.raises(module.CoalescingRefused, match="remained queued"): + module._cancel_run("o/r", 123) + assert cancel_calls == 1 + + +def test_coalesce_preserves_started_candidate_after_cancel_409(monkeypatch, capsys) -> None: + """The production coalesce path preserves a candidate that starts at POST time.""" + module = load_module() + candidate = run_record(100, 10) + sibling = run_record(101, 10) + candidate_fetches = 0 + cancel_calls = 0 + + monkeypatch.setattr(module, "_fetch_pr", lambda *_args: live_pr()) + monkeypatch.setattr(module, "_active_runs", lambda *_args: [candidate, sibling]) + + def fetch_run(_repo, run_id): + """Return the sibling or transition the candidate to in-progress.""" + nonlocal candidate_fetches + if run_id == 101: + return sibling + candidate_fetches += 1 + return candidate if candidate_fetches == 1 else run_record(100, 10, status="in_progress") + + def run_json(args): + """Raise the queued-start race without permitting unrelated commands.""" + nonlocal cancel_calls + if args[-1].endswith("/cancel"): + cancel_calls += 1 + raise RuntimeError("Cannot cancel a workflow run that has not been queued yet. (HTTP409)") + raise AssertionError(args) + + monkeypatch.setattr(module, "_fetch_run", fetch_run) + monkeypatch.setattr(module, "_run_json", run_json) + + assert module.coalesce( + "ContextualWisdomLab/.github", + 1, + "ContextualWisdomLab/.github", + "feature/current", + "a" * 40, + ) == [] + assert cancel_calls == 1 + assert "Preserving run 100" in capsys.readouterr().out + + +@pytest.mark.parametrize( + ("state", "error", "expected_posts", "expected_gets"), + [ + ({"status": "completed", "conclusion": "cancelled"}, None, 1, 1), + ({"status": "in_progress", "conclusion": None}, "no longer queued", 1, 1), + ({"status": "completed", "conclusion": "success"}, "no longer queued", 1, 1), + ({"status": "mystery", "conclusion": None}, "no longer queued", 1, 1), + ], +) +def test_cancel_run_409_state_gate_never_overclaims( + monkeypatch, state, error, expected_posts, expected_gets +) -> None: + """Only cancelled terminal evidence suppresses the preservation refusal.""" + module = load_module() + calls = {"post": 0, "get": 0} + + def run_json(args): + """Raise the cancellation race for each parameterized state.""" + if args[-1].endswith("/cancel"): + calls["post"] += 1 + raise RuntimeError("Cannot cancel a workflow run that has not been queued yet. (HTTP409)") + raise AssertionError(args) + + def fetch_run(_repo, _run_id): + """Return the parameterized authoritative post-409 state.""" + calls["get"] += 1 + return state + + monkeypatch.setattr(module, "_run_json", run_json) + monkeypatch.setattr(module, "_fetch_run", fetch_run) + if error: + with pytest.raises(module.CoalescingRefused, match=error): + module._cancel_run("o/r", 123) + else: + module._cancel_run("o/r", 123) + assert calls == {"post": expected_posts, "get": expected_gets} + + +def test_cancel_run_ignores_unrelated_error_without_recheck(monkeypatch) -> None: + """A non-409 cancellation error cannot trigger a compensating mutation.""" + module = load_module() + calls: list[str] = [] + + def run_json(args): + """Raise the unrelated cancellation failure without a second request.""" + calls.append("post") + raise RuntimeError("HTTP500 upstream failure") + + monkeypatch.setattr(module, "_run_json", run_json) + monkeypatch.setattr(module, "_fetch_run", lambda *_args: calls.append("get")) + with pytest.raises(RuntimeError, match="HTTP500"): + module._cancel_run("o/r", 123) + assert calls == ["post"] + + +def test_cancel_run_fails_closed_when_queued_after_queue_start_race(monkeypatch) -> None: + """A queued run after a startup race is preserved without a second POST.""" + module = load_module() + calls = {"post": 0} + + def run_json(args): + """Raise the queue-start race while counting cancellation posts.""" + if args[-1].endswith("/cancel"): + calls["post"] += 1 + raise RuntimeError("Cannot cancel a workflow run that has not been queued yet. (HTTP409)") + raise AssertionError(args) + + monkeypatch.setattr(module, "_run_json", run_json) + monkeypatch.setattr(module, "_fetch_run", lambda *_args: {"status": "queued", "conclusion": None}) + with pytest.raises(module.CoalescingRefused, match="remained queued"): + module._cancel_run("o/r", 123) + assert calls == {"post": 1} + + +def test_cancel_run_409_detection_does_not_depend_on_provider_english(monkeypatch) -> None: + """A bare HTTP 409 still preserves a queued run without a second POST.""" + module = load_module() + calls = {"post": 0} + + def run_json(args): + """Raise a bare HTTP 409 to test language-independent detection.""" + if args[-1].endswith("/cancel"): + calls["post"] += 1 + raise RuntimeError("HTTP 409 conflict") + raise AssertionError(args) + + monkeypatch.setattr(module, "_run_json", run_json) + monkeypatch.setattr(module, "_fetch_run", lambda *_args: {"status": "queued"}) + with pytest.raises(module.CoalescingRefused, match="remained queued"): + module._cancel_run("o/r", 123) + assert calls == {"post": 1} + + def test_cancel_run_fails_when_terminal_cancellation_is_unproven(monkeypatch) -> None: """An accepted cancellation is not reported complete while GitHub stays active.""" module = load_module() @@ -577,6 +765,7 @@ def test_coalesce_refetches_candidate_last_and_preserves_started_run(monkeypatch monkeypatch.setattr(module, "_active_runs", lambda *_args: [candidate, sibling]) def fetch_run(_repo: str, run_id: int): + """Return the sibling while showing the candidate started meanwhile.""" return sibling if run_id == 101 else run_record(100, 10, status="in_progress") monkeypatch.setattr(module, "_fetch_run", fetch_run) @@ -673,6 +862,7 @@ def test_main_treats_coalescing_refused_as_a_safe_no_op(monkeypatch, capsys) -> ] def refuse(*_args: object) -> list[int]: + """Raise the safe coalescing refusal handled by the CLI entrypoint.""" raise module.CoalescingRefused("pull request head moved before duplicate classification") monkeypatch.setattr(module, "coalesce", refuse)