fix(scheduler): recheck cancellation races

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":

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)