diff --git a/.github/workflows/pr-title.yml b/.github/workflows/pr-title.yml index 32aa594cf4..0268b9e77c 100644 --- a/.github/workflows/pr-title.yml +++ b/.github/workflows/pr-title.yml @@ -9,7 +9,7 @@ on: types: [opened, edited, synchronize] concurrency: - group: ${{ github.workflow }}-${{ github.head_ref }} + group: ${{ github.workflow }}-${{ github.event.pull_request.number }} cancel-in-progress: true jobs: diff --git a/.github/workflows/static-checks.yml b/.github/workflows/static-checks.yml index a9af7f502d..c6cb0b615b 100644 --- a/.github/workflows/static-checks.yml +++ b/.github/workflows/static-checks.yml @@ -10,7 +10,7 @@ on: branches: [main, develop] concurrency: - group: ${{ github.workflow }}-${{ github.head_ref || github.ref }} + group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }} cancel-in-progress: true env: @@ -41,7 +41,7 @@ jobs: check: needs: scope - if: ${{ always() }} + if: ${{ !cancelled() }} runs-on: ubuntu-latest timeout-minutes: 90 steps: @@ -50,8 +50,8 @@ jobs: persist-credentials: false - name: Validate repository metadata and documentation run: python3 scripts/ci/validate_repository.py - - name: Test CI scope routing - run: python3 scripts/ci/test_ci_scope.py + - name: Test CI contracts + run: python3 -m unittest discover -s scripts/ci -p 'test_*.py' - name: Validate Terminal-Bench harness contracts if: ${{ needs.scope.result != 'success' || needs.scope.outputs.harness == 'true' }} run: | @@ -95,7 +95,7 @@ jobs: sdk: name: "@astra/sdk (typecheck, test+coverage, build)" needs: scope - if: ${{ always() && (needs.scope.result != 'success' || needs.scope.outputs.sdk == 'true') }} + if: ${{ !cancelled() && (needs.scope.result != 'success' || needs.scope.outputs.sdk == 'true') }} runs-on: ubuntu-latest timeout-minutes: 45 steps: @@ -118,7 +118,7 @@ jobs: web: name: "web (typecheck, test, build)" needs: scope - if: ${{ always() && (needs.scope.result != 'success' || needs.scope.outputs.web == 'true') }} + if: ${{ !cancelled() && (needs.scope.result != 'success' || needs.scope.outputs.web == 'true') }} runs-on: ubuntu-latest timeout-minutes: 60 steps: diff --git a/.github/workflows/supersede-pr-runs.yml b/.github/workflows/supersede-pr-runs.yml new file mode 100644 index 0000000000..1b94d67c43 --- /dev/null +++ b/.github/workflows/supersede-pr-runs.yml @@ -0,0 +1,42 @@ +name: Supersede stale PR runs + +on: + pull_request_target: + branches: [main, develop] + types: [synchronize] + +# This controller runs from its trusted workflow revision. It must never check out +# or execute the pull request head while holding permission to cancel runs. +permissions: + actions: write + contents: read + +# Do not add concurrency here. Every synchronize event owns one before -> after +# transition; dropping an intermediate controller could leave its prior head +# running after several rapid pushes. + +jobs: + cancel-stale-runs: + name: Cancel superseded PR runs + if: ${{ github.event.pull_request.head.repo.full_name != github.repository }} + runs-on: ubuntu-latest + timeout-minutes: 5 + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7 + with: + ref: ${{ github.workflow_sha }} + fetch-depth: 1 + persist-credentials: false + - name: Cancel runs for earlier PR heads + env: + GH_TOKEN: ${{ github.token }} + HEAD_REF: ${{ github.event.pull_request.head.ref }} + HEAD_REPOSITORY_ID: ${{ github.event.pull_request.head.repo.id }} + REPOSITORY: ${{ github.repository }} + SUPERSEDED_HEAD_SHA: ${{ github.event.before }} + run: >- + python3 scripts/ci/cancel_stale_pr_runs.py + --repository="$REPOSITORY" + --head-repository-id="$HEAD_REPOSITORY_ID" + --head-ref="$HEAD_REF" + --superseded-head-sha="$SUPERSEDED_HEAD_SHA" diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index f85880dc3c..80b7746995 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -10,7 +10,7 @@ on: branches: [main, develop] concurrency: - group: ${{ github.workflow }}-${{ github.head_ref || github.ref }} + group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }} cancel-in-progress: true env: @@ -63,7 +63,7 @@ jobs: shard-a: name: "Test: astra-cli (${{ matrix.segment }})" needs: scope - if: ${{ always() }} + if: ${{ !cancelled() }} runs-on: ubuntu-latest timeout-minutes: 45 env: @@ -126,7 +126,7 @@ jobs: shard-b: name: "Test: astra-runtime" needs: scope - if: ${{ always() && (needs.scope.result != 'success' || needs.scope.outputs.test_runtime == 'true') }} + if: ${{ !cancelled() && (needs.scope.result != 'success' || needs.scope.outputs.test_runtime == 'true') }} runs-on: ubuntu-latest timeout-minutes: 35 steps: @@ -157,7 +157,7 @@ jobs: shard-c: name: "Test: turn-core + services + plan" needs: scope - if: ${{ always() && (needs.scope.result != 'success' || needs.scope.outputs.test_services == 'true') }} + if: ${{ !cancelled() && (needs.scope.result != 'success' || needs.scope.outputs.test_services == 'true') }} runs-on: ubuntu-latest timeout-minutes: 30 steps: @@ -177,7 +177,7 @@ jobs: shard-d: name: "Test: core crates + bridge hooks" needs: scope - if: ${{ always() && (needs.scope.result != 'success' || needs.scope.outputs.test_core == 'true') }} + if: ${{ !cancelled() && (needs.scope.result != 'success' || needs.scope.outputs.test_core == 'true') }} runs-on: ubuntu-latest timeout-minutes: 35 steps: @@ -209,7 +209,7 @@ jobs: test-online: name: "Test: online (${{ matrix.lane }})" needs: scope - if: ${{ always() }} + if: ${{ !cancelled() }} runs-on: ubuntu-latest timeout-minutes: 45 strategy: diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index e2f70322fd..bfa88f1cce 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -87,6 +87,20 @@ Required check names remain present when their heavy work is skipped, so this routing is compatible with branch protection and the merge queue. The routing contract and its tests live in [`scripts/ci/`](scripts/ci/). +For fork pull requests, an update may require maintainer approval before the +replacement test run can enter the normal concurrency group. A separate +trusted-workflow-revision controller cancels active runs for earlier heads +without waiting for replacement-run approval. The controller never checks out +or executes pull request code; the +test workflows remain low-privilege `pull_request` workflows. Normal concurrency +groups use the pull request number, so identically named branches in different +forks remain isolated. Each synchronize event cancels only its exact `before` +head and is intentionally not coalesced, so rapid pushes cannot make an older +controller cancel a newer generation or skip an intermediate cleanup. Jobs that +must run after scope-classification failure use `!cancelled()` rather than +`always()`, preserving fail-safe coverage without making superseded work resist +cancellation. + ## Open a pull request 1. Rebase the feature branch on the current `main` branch. diff --git a/scripts/ci/cancel_stale_pr_runs.py b/scripts/ci/cancel_stale_pr_runs.py new file mode 100755 index 0000000000..c605a11932 --- /dev/null +++ b/scripts/ci/cancel_stale_pr_runs.py @@ -0,0 +1,251 @@ +#!/usr/bin/env python3 +"""Cancel active GitHub Actions runs for a superseded pull-request head.""" + +from __future__ import annotations + +import argparse +from dataclasses import dataclass +import json +import os +import re +import sys +import time +from typing import Any, Protocol +from urllib.error import HTTPError +from urllib.parse import urlencode, urljoin, urlsplit +from urllib.request import build_opener, HTTPRedirectHandler, Request + + +ACTIVE_RUN_STATUSES = frozenset( + {"queued", "in_progress", "pending", "requested", "waiting"} +) +# These caps keep the worst-case request path within the workflow's five-minute +# timeout while leaving headroom over Astra's three pull-request workflows. +MAX_PAGES = 5 +MAX_SUPERSEDED_RUNS = 5 +REQUEST_TIMEOUT_SECONDS = 10 +TERMINAL_STATUS_POLLS = 3 +TERMINAL_STATUS_POLL_SECONDS = 1 +REPOSITORY_PATTERN = re.compile(r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$") +SHA_PATTERN = re.compile(r"^[0-9a-fA-F]{40}$") + + +@dataclass(frozen=True) +class SupersededHead: + repository_id: int + ref: str + sha: str + + +class ActionsApi(Protocol): + def list_runs(self, repository: str, head_sha: str) -> list[dict[str, Any]]: ... + + def cancel_run(self, repository: str, run_id: int) -> bool: ... + + +class RejectRedirects(HTTPRedirectHandler): + def redirect_request( + self, + req: Request, + fp: Any, + code: int, + msg: str, + headers: Any, + newurl: str, + ) -> Request | None: + raise RuntimeError("refusing to forward the GitHub token through a redirect") + + +class GitHubApi: + """Small GitHub REST client with bounded, origin-checked pagination.""" + + def __init__(self, token: str, api_url: str = "https://api.github.com") -> None: + if not token: + raise ValueError("GH_TOKEN is required") + self._token = token + self._api_url = api_url.rstrip("/") + "/" + parsed_api_url = urlsplit(self._api_url) + if parsed_api_url.scheme.lower() != "https" or not parsed_api_url.netloc: + raise ValueError("GITHUB_API_URL must use HTTPS and include a host") + self._api_origin = (parsed_api_url.scheme.lower(), parsed_api_url.netloc.lower()) + self._opener = build_opener(RejectRedirects()) + + def _request(self, method: str, url: str) -> tuple[bytes, Any]: + target = url if urlsplit(url).scheme else urljoin(self._api_url, url.lstrip("/")) + parsed_target = urlsplit(target) + target_origin = (parsed_target.scheme.lower(), parsed_target.netloc.lower()) + if target_origin != self._api_origin: + raise RuntimeError("refusing to send the GitHub token to another origin") + request = Request( + target, + method=method, + headers={ + "Accept": "application/vnd.github+json", + "Authorization": f"Bearer {self._token}", + "User-Agent": "astra-stale-pr-run-controller", + "X-GitHub-Api-Version": "2022-11-28", + }, + ) + with self._opener.open(request, timeout=REQUEST_TIMEOUT_SECONDS) as response: + return response.read(), response.headers + + def _get_json(self, path: str) -> dict[str, Any]: + body, _ = self._request("GET", path) + value = json.loads(body) + if not isinstance(value, dict): + raise RuntimeError(f"GitHub returned a non-object response for {path}") + return value + + @staticmethod + def _next_link(headers: Any) -> str | None: + for part in headers.get("Link", "").split(","): + match = re.match(r'\s*<([^>]+)>;\s*rel="([^"]+)"\s*$', part) + if match and match.group(2) == "next": + return match.group(1) + return None + + def list_runs(self, repository: str, head_sha: str) -> list[dict[str, Any]]: + query = urlencode( + {"event": "pull_request", "head_sha": head_sha, "per_page": 100} + ) + next_url: str | None = f"repos/{repository}/actions/runs?{query}" + runs: list[dict[str, Any]] = [] + visited: set[str] = set() + while next_url: + if next_url in visited or len(visited) >= MAX_PAGES: + raise RuntimeError("workflow-run pagination exceeded its safe bound") + visited.add(next_url) + body, headers = self._request("GET", next_url) + page = json.loads(body) + page_runs = page.get("workflow_runs") if isinstance(page, dict) else None + if not isinstance(page_runs, list): + raise RuntimeError("GitHub workflow-runs response omitted workflow_runs") + runs.extend(run for run in page_runs if isinstance(run, dict)) + next_url = self._next_link(headers) + return runs + + def cancel_run(self, repository: str, run_id: int) -> bool: + try: + self._request("POST", f"repos/{repository}/actions/runs/{run_id}/cancel") + except HTTPError as error: + # A selected run can finish between the list and cancel requests. + if error.code != 409: + raise + if error.fp is not None: + error.close() + for attempt in range(TERMINAL_STATUS_POLLS): + run = self._get_json(f"repos/{repository}/actions/runs/{run_id}") + if run.get("status") == "completed": + return False + if attempt + 1 < TERMINAL_STATUS_POLLS: + time.sleep(TERMINAL_STATUS_POLL_SECONDS) + raise RuntimeError( + f"GitHub refused to cancel active workflow run {run_id}" + ) from error + return True + + +def select_superseded_runs( + runs: list[dict[str, Any]], superseded_head: SupersededHead +) -> list[dict[str, Any]]: + """Select active runs for exactly one superseded source-head generation.""" + selected: dict[int, dict[str, Any]] = {} + for run in runs: + run_id = run.get("id") + head_repository = run.get("head_repository") + repository_id = head_repository.get("id") if isinstance(head_repository, dict) else None + if not isinstance(run_id, int) or run.get("status") not in ACTIVE_RUN_STATUSES: + continue + if repository_id != superseded_head.repository_id: + continue + if run.get("head_branch") != superseded_head.ref: + continue + if run.get("head_sha") != superseded_head.sha: + continue + selected[run_id] = run + return [selected[run_id] for run_id in sorted(selected)] + + +def cancel_superseded_runs( + api: ActionsApi, repository: str, superseded_head: SupersededHead +) -> tuple[int, int]: + runs = api.list_runs(repository, superseded_head.sha) + superseded_runs = select_superseded_runs(runs, superseded_head) + if len(superseded_runs) > MAX_SUPERSEDED_RUNS: + raise RuntimeError( + "refusing an unexpectedly large cancellation set " + f"({len(superseded_runs)} runs)" + ) + + cancelled = 0 + failures: list[tuple[int, Exception]] = [] + for run in superseded_runs: + run_id = int(run["id"]) + try: + cancellation_requested = api.cancel_run(repository, run_id) + except Exception as error: + failures.append((run_id, error)) + print( + f"Failed to cancel run {run_id}; continuing cleanup.", file=sys.stderr + ) + continue + if cancellation_requested: + cancelled += 1 + print( + f"Cancellation requested for {run.get('name', 'workflow')} " + f"run {run_id} at {superseded_head.sha}." + ) + else: + print(f"Run {run_id} became terminal before cancellation; no action needed.") + + print( + f"Found {len(superseded_runs)} active run(s) for superseded head " + f"repository={superseded_head.repository_id}, ref={superseded_head.ref}, " + f"sha={superseded_head.sha}; requested {cancelled} cancellation(s)." + ) + if failures: + failed_run_ids = ", ".join(str(run_id) for run_id, _ in failures) + raise RuntimeError( + f"failed to cancel {len(failures)} workflow run(s): {failed_run_ids}" + ) from failures[0][1] + return len(superseded_runs), cancelled + + +def parse_args(argv: list[str]) -> argparse.Namespace: + parser = argparse.ArgumentParser() + parser.add_argument("--repository", required=True) + parser.add_argument("--head-repository-id", required=True, type=int) + parser.add_argument("--head-ref", required=True) + parser.add_argument("--superseded-head-sha", required=True) + args = parser.parse_args(argv) + if not REPOSITORY_PATTERN.fullmatch(args.repository): + parser.error("--repository must be an owner/repository pair") + if args.head_repository_id <= 0: + parser.error("--head-repository-id must be positive") + if not args.head_ref: + parser.error("--head-ref must not be empty") + if not SHA_PATTERN.fullmatch(args.superseded_head_sha): + parser.error("--superseded-head-sha must be a 40-character hexadecimal SHA") + return args + + +def main(argv: list[str] | None = None) -> int: + args = parse_args(sys.argv[1:] if argv is None else argv) + api = GitHubApi( + os.environ.get("GH_TOKEN", ""), + os.environ.get("GITHUB_API_URL", "https://api.github.com"), + ) + cancel_superseded_runs( + api, + args.repository, + SupersededHead( + repository_id=args.head_repository_id, + ref=args.head_ref, + sha=args.superseded_head_sha, + ), + ) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/ci/test_cancel_stale_pr_runs.py b/scripts/ci/test_cancel_stale_pr_runs.py new file mode 100755 index 0000000000..ccfca39009 --- /dev/null +++ b/scripts/ci/test_cancel_stale_pr_runs.py @@ -0,0 +1,384 @@ +#!/usr/bin/env python3 + +from __future__ import annotations + +from contextlib import redirect_stderr, redirect_stdout +from io import StringIO +from itertools import permutations +import json +from pathlib import Path +import sys +import unittest +from unittest.mock import Mock, patch +from urllib.error import HTTPError + +sys.path.insert(0, str(Path(__file__).resolve().parent)) + +from cancel_stale_pr_runs import ( # noqa: E402 + ACTIVE_RUN_STATUSES, + GitHubApi, + MAX_SUPERSEDED_RUNS, + RejectRedirects, + SupersededHead, + cancel_superseded_runs, + parse_args, + select_superseded_runs, +) + + +OLD_SHA = "a" * 40 +CURRENT_SHA = "b" * 40 +NEWER_SHA = "c" * 40 +LATEST_SHA = "d" * 40 + + +def workflow_run( + run_id: int, + *, + sha: str = OLD_SHA, + status: str = "in_progress", + repository_id: int = 101, + branch: str = "fix/runtime", +) -> dict[str, object]: + return { + "id": run_id, + "name": "Test Suite", + "status": status, + "head_sha": sha, + "head_branch": branch, + "head_repository": {"id": repository_id, "full_name": "contributor/Astra"}, + } + + +class FakeApi: + def __init__(self, runs: list[dict[str, object]]) -> None: + self.runs = runs + self.listed_heads: list[str] = [] + self.cancelled: list[int] = [] + self.finished_before_cancel: set[int] = set() + + def list_runs(self, repository: str, head_sha: str) -> list[dict[str, object]]: + self.listed_heads.append(head_sha) + return self.runs + + def cancel_run(self, repository: str, run_id: int) -> bool: + if run_id in self.finished_before_cancel: + return False + self.cancelled.append(run_id) + return True + + +class SelectSupersededRunsTests(unittest.TestCase): + def test_selects_only_the_exact_fork_ref_sha_and_active_status(self) -> None: + head = SupersededHead(101, "fix/runtime", OLD_SHA) + runs = [ + workflow_run(1), + workflow_run(2, sha=CURRENT_SHA), + workflow_run(3, repository_id=202), + workflow_run(4, branch="fix/other"), + workflow_run(5, status="completed"), + {"id": 6, "status": "queued", "head_sha": OLD_SHA}, + ] + + self.assertEqual( + [run["id"] for run in select_superseded_runs(runs, head)], [1] + ) + + def test_deduplicates_a_run_observed_during_pagination(self) -> None: + head = SupersededHead(101, "fix/runtime", OLD_SHA) + run = workflow_run(7) + + self.assertEqual( + [item["id"] for item in select_superseded_runs([run, run], head)], [7] + ) + + def test_every_controller_order_cancels_old_generations_only(self) -> None: + runs = [ + workflow_run(1, sha=OLD_SHA), + workflow_run(2, sha=CURRENT_SHA), + workflow_run(3, sha=NEWER_SHA), + workflow_run(4, sha=LATEST_SHA), + ] + + for controller_order in permutations((OLD_SHA, CURRENT_SHA, NEWER_SHA)): + with self.subTest(controller_order=controller_order): + selected = { + int(run["id"]) + for sha in controller_order + for run in select_superseded_runs( + runs, SupersededHead(101, "fix/runtime", sha) + ) + } + + self.assertEqual(selected, {1, 2, 3}) + + +class GitHubApiTests(unittest.TestCase): + def test_list_runs_filters_by_superseded_sha_and_follows_pagination(self) -> None: + api = GitHubApi("test-token") + second_page = "https://api.github.com/repos/matrixorigin/Astra/actions/runs?page=2" + api._request = Mock( # type: ignore[method-assign] + side_effect=[ + ( + json.dumps({"workflow_runs": [workflow_run(1)]}).encode(), + {"Link": f'<{second_page}>; rel="next"'}, + ), + (json.dumps({"workflow_runs": [workflow_run(2)]}).encode(), {}), + ] + ) + + self.assertEqual( + [run["id"] for run in api.list_runs("matrixorigin/Astra", OLD_SHA)], + [1, 2], + ) + first_url = api._request.call_args_list[0].args[1] + self.assertIn(f"head_sha={OLD_SHA}", first_url) + self.assertIn("event=pull_request", first_url) + + def test_request_rejects_a_foreign_origin_before_sending_the_token(self) -> None: + api = GitHubApi("test-token") + + with self.assertRaisesRegex(RuntimeError, "another origin"): + api._request("GET", "https://example.test/steal") + with self.assertRaisesRegex(RuntimeError, "another origin"): + api._request("GET", "http://api.github.com/cleartext") + + def test_client_rejects_a_cleartext_api_root(self) -> None: + with self.assertRaisesRegex(ValueError, "must use HTTPS"): + GitHubApi("test-token", "http://api.github.com") + + def test_redirect_handler_never_forwards_the_token(self) -> None: + handler = RejectRedirects() + + with self.assertRaisesRegex(RuntimeError, "through a redirect"): + handler.redirect_request( + Mock(), + Mock(), + 302, + "Found", + {}, + "https://example.test/steal", + ) + + def test_pagination_cycle_fails_instead_of_waiting_forever(self) -> None: + api = GitHubApi("test-token") + repeated_page = ( + "https://api.github.com/repos/matrixorigin/Astra/actions/runs?page=2" + ) + api._request = Mock( # type: ignore[method-assign] + return_value=( + json.dumps({"workflow_runs": []}).encode(), + {"Link": f'<{repeated_page}>; rel="next"'}, + ) + ) + + with self.assertRaisesRegex(RuntimeError, "pagination"): + api.list_runs("matrixorigin/Astra", OLD_SHA) + + def test_cancel_treats_a_completion_race_as_terminal(self) -> None: + api = GitHubApi("test-token") + api._request = Mock( # type: ignore[method-assign] + side_effect=[ + HTTPError("url", 409, "Conflict", {}, None), + (json.dumps({"status": "completed"}).encode(), {}), + ] + ) + + self.assertFalse(api.cancel_run("matrixorigin/Astra", 42)) + + def test_cancel_does_not_hide_a_conflict_for_an_active_run(self) -> None: + api = GitHubApi("test-token") + api._request = Mock( # type: ignore[method-assign] + side_effect=[ + HTTPError("url", 409, "Conflict", {}, None), + (json.dumps({"status": "in_progress"}).encode(), {}), + (json.dumps({"status": "in_progress"}).encode(), {}), + (json.dumps({"status": "in_progress"}).encode(), {}), + ] + ) + + with ( + patch("cancel_stale_pr_runs.time.sleep") as sleep, + self.assertRaisesRegex(RuntimeError, "refused to cancel active"), + ): + api.cancel_run("matrixorigin/Astra", 42) + self.assertEqual(sleep.call_count, 2) + + def test_cancel_waits_for_an_already_requested_cancellation(self) -> None: + api = GitHubApi("test-token") + api._request = Mock( # type: ignore[method-assign] + side_effect=[ + HTTPError("url", 409, "Conflict", {}, None), + (json.dumps({"status": "in_progress"}).encode(), {}), + (json.dumps({"status": "completed"}).encode(), {}), + ] + ) + + with patch("cancel_stale_pr_runs.time.sleep") as sleep: + self.assertFalse(api.cancel_run("matrixorigin/Astra", 42)) + sleep.assert_called_once() + + +class CancelSupersededRunsTests(unittest.TestCase): + def test_cancels_every_active_status_but_not_completed_runs(self) -> None: + runs = [ + workflow_run(index, status=status) + for index, status in enumerate(sorted(ACTIVE_RUN_STATUSES), start=1) + ] + runs.append(workflow_run(100, status="completed")) + api = FakeApi(runs) + + with redirect_stdout(StringIO()): + found, cancelled = cancel_superseded_runs( + api, + "matrixorigin/Astra", + SupersededHead(101, "fix/runtime", OLD_SHA), + ) + + self.assertEqual( + (found, cancelled), + (len(ACTIVE_RUN_STATUSES), len(ACTIVE_RUN_STATUSES)), + ) + self.assertEqual(api.listed_heads, [OLD_SHA]) + + def test_completion_race_is_not_reported_as_a_failed_cancellation(self) -> None: + api = FakeApi([workflow_run(1), workflow_run(2)]) + api.finished_before_cancel.add(2) + + with redirect_stdout(StringIO()): + found, cancelled = cancel_superseded_runs( + api, + "matrixorigin/Astra", + SupersededHead(101, "fix/runtime", OLD_SHA), + ) + + self.assertEqual((found, cancelled), (2, 1)) + self.assertEqual(api.cancelled, [1]) + + def test_one_failure_does_not_prevent_other_runs_from_being_cancelled(self) -> None: + api = Mock() + api.list_runs.return_value = [workflow_run(1), workflow_run(2)] + api.cancel_run.side_effect = [RuntimeError("transient failure"), True] + + with ( + redirect_stderr(StringIO()), + redirect_stdout(StringIO()), + self.assertRaisesRegex(RuntimeError, "failed to cancel 1 workflow run"), + ): + cancel_superseded_runs( + api, + "matrixorigin/Astra", + SupersededHead(101, "fix/runtime", OLD_SHA), + ) + + self.assertEqual( + [call.args for call in api.cancel_run.call_args_list], + [("matrixorigin/Astra", 1), ("matrixorigin/Astra", 2)], + ) + + def test_unexpected_cancellation_fanout_fails_before_mutation(self) -> None: + api = FakeApi( + [workflow_run(run_id) for run_id in range(1, MAX_SUPERSEDED_RUNS + 2)] + ) + + with self.assertRaisesRegex(RuntimeError, "unexpectedly large"): + cancel_superseded_runs( + api, + "matrixorigin/Astra", + SupersededHead(101, "fix/runtime", OLD_SHA), + ) + + self.assertEqual(api.cancelled, []) + + +class ArgumentContractTests(unittest.TestCase): + def test_option_shaped_head_ref_remains_a_value(self) -> None: + args = parse_args( + [ + "--repository=matrixorigin/Astra", + "--head-repository-id=101", + "--head-ref=--repository", + f"--superseded-head-sha={OLD_SHA}", + ] + ) + + self.assertEqual(args.head_ref, "--repository") + + +class WorkflowSecurityContractTests(unittest.TestCase): + def test_controller_executes_only_the_trusted_base_revision(self) -> None: + workflow = ( + Path(__file__).resolve().parents[2] + / ".github/workflows/supersede-pr-runs.yml" + ).read_text(encoding="utf-8") + + for required in ( + "pull_request_target:", + "actions: write", + "contents: read", + "ref: ${{ github.workflow_sha }}", + "persist-credentials: false", + "github.event.pull_request.head.repo.full_name != github.repository", + "SUPERSEDED_HEAD_SHA: ${{ github.event.before }}", + "python3 scripts/ci/cancel_stale_pr_runs.py", + '--head-ref="$HEAD_REF"', + ): + with self.subTest(required=required): + self.assertIn(required, workflow) + permissions = workflow.partition("permissions:\n")[2].partition("\n\n")[0] + self.assertEqual(permissions, " actions: write\n contents: read") + self.assertEqual(workflow.count("uses: actions/checkout@"), 1) + refs = [ + line.strip() + for line in workflow.splitlines() + if line.strip().startswith("ref:") + ] + self.assertEqual(refs, ["ref: ${{ github.workflow_sha }}"]) + self.assertNotIn("secrets.", workflow) + self.assertNotRegex(workflow, r"(?m)^\s*concurrency:") + + def test_pull_request_concurrency_isolated_by_pr_number(self) -> None: + root = Path(__file__).resolve().parents[2] + expected = "${{ github.event.pull_request.number" + for relative in ( + ".github/workflows/pr-title.yml", + ".github/workflows/static-checks.yml", + ".github/workflows/test.yml", + ): + with self.subTest(workflow=relative): + self.assertIn(expected, (root / relative).read_text(encoding="utf-8")) + + def test_cancellation_bound_covers_every_pull_request_workflow(self) -> None: + workflows = Path(__file__).resolve().parents[2] / ".github/workflows" + pull_request_workflows = [ + path.name + for pattern in ("*.yml", "*.yaml") + for path in workflows.glob(pattern) + if "\n pull_request:\n" in path.read_text(encoding="utf-8") + ] + + self.assertEqual( + sorted(pull_request_workflows), + ["pr-title.yml", "static-checks.yml", "test.yml"], + ) + self.assertLessEqual(len(pull_request_workflows), MAX_SUPERSEDED_RUNS) + + def test_pr_jobs_do_not_resist_workflow_cancellation(self) -> None: + root = Path(__file__).resolve().parents[2] + for relative in ( + ".github/workflows/static-checks.yml", + ".github/workflows/test.yml", + ): + workflow = (root / relative).read_text(encoding="utf-8") + job_level_always = [ + line + for line in workflow.splitlines() + if line.startswith(" if:") and "always()" in line + ] + with self.subTest(workflow=relative): + self.assertEqual(job_level_always, []) + self.assertIn("if: ${{ !cancelled()", workflow) + + +if __name__ == "__main__": + unittest.main()