ci(actions): scope superseded PR cancellation

.github/workflows/code-quality.yml

@@ -4,13 +4,22 @@ on:
   push:
     branches: ["master", "main"]
   pull_request:
+    types: [opened, synchronize, reopened, ready_for_review, converted_to_draft, closed]
     branches: ["master", "main"]
 
 permissions:
   contents: read
 
+concurrency:
+  group: >-
+    ${{ github.workflow }}-${{ github.repository }}-${{
+    github.event_name == 'pull_request' && github.run_attempt == 1 &&
+    github.event.pull_request.number || github.run_id }}
+  cancel-in-progress: ${{ github.event_name == 'pull_request' }}
+
 jobs:
   quality:
+    if: ${{ github.event_name != 'pull_request' || (github.event.action != 'closed' && github.event.pull_request.draft == false) }}
     runs-on: ubuntu-latest
 
     steps:
@@ -27,6 +36,11 @@ jobs:
           python3 -m pip install --user yamllint
           python3 -m yamllint .yamllint.yml .github/dependabot.yml .github/workflows/*.yml
 
+      - name: Validate workflow concurrency contract
+        run: |
+          python3 scripts/ci/test_workflow_concurrency_contract_unit.py
+          python3 scripts/ci/test_workflow_concurrency_contract.py
+
       - name: Lint markdown docs
         run: |
           npm install -g markdownlint-cli2@0.18.1

.github/workflows/r.yml

@@ -4,18 +4,23 @@ on:
   push:
     branches: ["master", "main"]
   pull_request:
+    types: [opened, synchronize, reopened, ready_for_review, converted_to_draft, closed]
     branches: ["master", "main"]
   workflow_dispatch:
 
 permissions:
   contents: read
 
 concurrency:
-  group: r-cmd-check-${{ github.ref }}
-  cancel-in-progress: true
+  group: >-
+    ${{ github.workflow }}-${{ github.repository }}-${{
+    github.event_name == 'pull_request' && github.run_attempt == 1 &&
+    github.event.pull_request.number || github.run_id }}
+  cancel-in-progress: ${{ github.event_name == 'pull_request' }}
 
 jobs:
   check:
+    if: ${{ github.event_name != 'pull_request' || (github.event.action != 'closed' && github.event.pull_request.draft == false) }}
     runs-on: ubuntu-latest
     env:
       R_PROFILE_USER: /dev/null

.github/workflows/security-audit.yml

@@ -4,13 +4,22 @@ on:
   push:
     branches: ["master", "main"]
   pull_request:
+    types: [opened, synchronize, reopened, ready_for_review, converted_to_draft, closed]
     branches: ["master", "main"]
 
 permissions:
   contents: read
 
+concurrency:
+  group: >-
+    ${{ github.workflow }}-${{ github.repository }}-${{
+    github.event_name == 'pull_request' && github.run_attempt == 1 &&
+    github.event.pull_request.number || github.run_id }}
+  cancel-in-progress: ${{ github.event_name == 'pull_request' }}
+
 jobs:
   secret-and-workflow-audit:
+    if: ${{ github.event_name != 'pull_request' || (github.event.action != 'closed' && github.event.pull_request.draft == false) }}
     runs-on: ubuntu-latest
 
     steps:

scripts/ci/test_workflow_concurrency_contract.py

@@ -0,0 +1,165 @@
+"""Validate repository-owned workflow concurrency without parsing lookalike text."""
+
+from pathlib import Path
+
+
+WORKFLOWS = Path(".github/workflows")
+EXPECTED_GROUP = (
+    "${{ github.workflow }}-${{ github.repository }}-"
+    "${{ github.event_name == 'pull_request' && github.run_attempt == 1 && "
+    "github.event.pull_request.number || github.run_id }}"
+)
+EXPECTED_CANCEL = "${{ github.event_name == 'pull_request' }}"
+EXPECTED_PR_TYPES = (
+    "types: [opened, synchronize, reopened, ready_for_review, converted_to_draft, closed]"
+)
+EXPECTED_PR_ADMISSION = (
+    "${{ github.event_name != 'pull_request' || "
+    "(github.event.action != 'closed' && github.event.pull_request.draft == false) }}"
+)
+
+
+def discover_workflows(root: Path = WORKFLOWS) -> list[Path]:
+    """Return every YAML workflow file under the repository workflow directory."""
+    return sorted({*root.glob("*.yml"), *root.glob("*.yaml")})
+
+
+def _top_level_concurrency_entries(path: Path, text: str) -> dict[str, list[str]]:
+    """Return direct key/value entries from the sole top-level concurrency block."""
+    lines = text.splitlines()
+    starts = [index for index, line in enumerate(lines) if line == "concurrency:"]
+    assert len(starts) == 1, f"{path}: expected exactly one top-level concurrency block"
+
+    start = starts[0] + 1
+    end = len(lines)
+    for index in range(start, len(lines)):
+        stripped = lines[index].strip()
+        if not stripped or stripped.startswith("#"):
+            continue
+        if lines[index][0] not in " \t":
+            end = index
+            break
+
+    entries: dict[str, list[str]] = {}
+    index = start
+    while index < end:
+        line = lines[index]
+        if not line.startswith("  ") or line.startswith("    "):
+            index += 1
+            continue
+
+        key, separator, raw_value = line[2:].partition(":")
+        if not separator:
+            index += 1
+            continue
+
+        value = raw_value.strip()
+        if value in {">", ">-"}:
+            continuation: list[str] = []
+            cursor = index + 1
+            while cursor < end and lines[cursor].startswith("    "):
+                assert not lines[cursor].startswith(
+                    "      "
+                ), f"{path}: folded concurrency scalar uses indentation that preserves newlines"
+                stripped = lines[cursor].strip()
+                if stripped and not stripped.startswith("#"):
+                    continuation.append(stripped)
+                cursor += 1
+            value = " ".join(continuation)
+            index = cursor
+        else:
+            index += 1
+
+        entries.setdefault(key, []).append(value)
+
+    return entries
+
+
+def _has_pull_request_trigger(text: str) -> bool:
+    """Return whether the workflow has a top-level pull-request trigger block."""
+    lines = text.splitlines()
+    try:
+        start = lines.index("on:") + 1
+    except ValueError:
+        return False
+    for line in lines[start:]:
+        if line and not line[0].isspace():
+            break
+        if line == "  pull_request:":
+            return True
+    return False
+
+
+def _pull_request_types(text: str) -> list[str]:
+    """Return direct entries from the top-level pull-request trigger."""
+    lines = text.splitlines()
+    start = lines.index("  pull_request:") + 1
+    entries: list[str] = []
+    for line in lines[start:]:
+        if line and (
+            not line[0].isspace()
+            or (line.startswith("  ") and not line.startswith("    "))
+        ):
+            break
+        if line.startswith("    ") and not line.startswith("      "):
+            entries.append(line[4:])
+    return entries
+
+
+def _job_admissions(text: str) -> list[str]:
+    """Return direct ``if`` values for every top-level job."""
+    lines = text.splitlines()
+    start = lines.index("jobs:") + 1
+    admissions: list[str] = []
+    for index in range(start, len(lines)):
+        line = lines[index]
+        if line and not line[0].isspace():
+            break
+        if line.startswith("  ") and not line.startswith("    ") and line.endswith(":"):
+            job_end = next(
+                (
+                    candidate
+                    for candidate in range(index + 1, len(lines))
+                    if lines[candidate].startswith("  ")
+                    and not lines[candidate].startswith("    ")
+                ),
+                len(lines),
+            )
+            direct_if = [
+                entry[8:]
+                for entry in lines[index + 1 : job_end]
+                if entry.startswith("    if: ")
+            ]
+            admissions.extend(direct_if or [""])
+    return admissions
+
+
+def validate_workflow_text(path: Path, text: str) -> None:
+    """Require exact PR lifecycle, admission, grouping, and cancellation semantics."""
+    entries = _top_level_concurrency_entries(path, text)
+
+    assert entries.get("group") == [EXPECTED_GROUP], f"{path}: unsafe concurrency group"
+    assert entries.get("cancel-in-progress") == [
+        EXPECTED_CANCEL
+    ], f"{path}: unsafe cancellation policy"
+    assert _has_pull_request_trigger(
+        text
+    ), f"{path}: missing structured pull-request trigger"
+    assert EXPECTED_PR_TYPES in _pull_request_types(
+        text
+    ), f"{path}: incomplete pull-request lifecycle"
+    assert _job_admissions(text) and all(
+        admission == EXPECTED_PR_ADMISSION for admission in _job_admissions(text)
+    ), f"{path}: draft or closed pull requests occupy a runner"
+
+
+def main() -> None:
+    """Validate every source-backed workflow in the repository."""
+    files = discover_workflows()
+    assert files, "no workflows found"
+    for path in files:
+        validate_workflow_text(path, path.read_text())
+
+
+if __name__ == "__main__":
+    main()

scripts/ci/test_workflow_concurrency_contract_unit.py

@@ -0,0 +1,125 @@
+import tempfile
+import unittest
+from pathlib import Path
+
+from test_workflow_concurrency_contract import discover_workflows, validate_workflow_text
+
+
+VALID = """name: Example
+on:
+  pull_request:
+    types: [opened, synchronize, reopened, ready_for_review, converted_to_draft, closed]
+concurrency:
+  group: >-
+    ${{ github.workflow }}-${{ github.repository }}-${{
+    github.event_name == 'pull_request' && github.run_attempt == 1 &&
+    github.event.pull_request.number || github.run_id }}
+  cancel-in-progress: ${{ github.event_name == 'pull_request' }}
+jobs:
+  check:
+    if: ${{ github.event_name != 'pull_request' || (github.event.action != 'closed' && github.event.pull_request.draft == false) }}
+    runs-on: ubuntu-latest
+"""
+
+
+class WorkflowConcurrencyContractTest(unittest.TestCase):
+    """Exercise discovery and top-level concurrency parsing edge cases."""
+
+    def test_discovers_yml_and_yaml(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            (root / "a.yml").write_text("name: a\n")
+            (root / "b.yaml").write_text("name: b\n")
+            (root / "ignored.txt").write_text("name: ignored\n")
+            self.assertEqual([path.name for path in discover_workflows(root)], ["a.yml", "b.yaml"])
+
+    def test_accepts_exact_top_level_contract(self) -> None:
+        validate_workflow_text(Path("valid.yml"), VALID)
+
+    def test_requires_cleanup_pull_request_lifecycle_events(self) -> None:
+        malformed = VALID.replace(
+            "types: [opened, synchronize, reopened, ready_for_review, converted_to_draft, closed]",
+            "types: [opened, synchronize, reopened, ready_for_review]",
+        )
+        with self.assertRaisesRegex(AssertionError, "pull-request lifecycle"):
+            validate_workflow_text(Path("missing-cleanup-events.yml"), malformed)
+
+    def test_rejects_folded_group_that_preserves_newlines(self) -> None:
+        malformed = VALID.replace(
+            "\n    github.event_name == 'pull_request'",
+            "\n      github.event_name == 'pull_request'",
+        )
+        with self.assertRaisesRegex(AssertionError, "folded concurrency scalar"):
+            validate_workflow_text(Path("newline-group.yml"), malformed)
+
+    def test_rejects_nested_lookalike(self) -> None:
+        malformed = """name: Example
+on: pull_request
+jobs:
+  check:
+    concurrency:
+      group: ${{ github.workflow }}-${{ github.repository }}-${{ github.event.pull_request.number || github.run_id }}
+      cancel-in-progress: ${{ github.event_name == 'pull_request' }}
+    runs-on: ubuntu-latest
+"""
+        with self.assertRaisesRegex(AssertionError, "top-level concurrency"):
+            validate_workflow_text(Path("nested.yml"), malformed)
+
+    def test_rejects_nested_concurrency_entries(self) -> None:
+        malformed = VALID.replace("  group:", "  policy:\n    group:").replace(
+            "  cancel-in-progress:", "    cancel-in-progress:"
+        )
+        with self.assertRaisesRegex(AssertionError, "unsafe concurrency group"):
+            validate_workflow_text(Path("nested-entries.yml"), malformed)
+
+    def test_rejects_duplicate_top_level_concurrency(self) -> None:
+        with self.assertRaisesRegex(AssertionError, "exactly one top-level concurrency"):
+            validate_workflow_text(Path("duplicate.yml"), VALID + "\nconcurrency:\n  group: duplicate\n")
+
+    def test_rejects_wrong_group(self) -> None:
+        malformed = VALID.replace("github.repository", "github.ref")
+        with self.assertRaisesRegex(AssertionError, "unsafe concurrency group"):
+            validate_workflow_text(Path("wrong-group.yml"), malformed)
+
+    def test_rejects_unconditional_cancellation(self) -> None:
+        malformed = VALID.replace(
+            "cancel-in-progress: ${{ github.event_name == 'pull_request' }}",
+            "cancel-in-progress: true",
+        )
+        with self.assertRaisesRegex(AssertionError, "unsafe cancellation policy"):
+            validate_workflow_text(Path("wrong-cancel.yml"), malformed)
+
+    def test_rejects_draft_runner_admission(self) -> None:
+        malformed = VALID.replace(
+            "github.event.pull_request.draft == false",
+            "github.event.pull_request.draft == true",
+        )
+        with self.assertRaisesRegex(AssertionError, "draft or closed pull requests"):
+            validate_workflow_text(Path("draft.yml"), malformed)
+
+    def test_rejects_missing_closed_runner_admission(self) -> None:
+        malformed = VALID.replace("github.event.action != 'closed' && ", "")
+        with self.assertRaisesRegex(AssertionError, "draft or closed pull requests"):
+            validate_workflow_text(Path("closed.yml"), malformed)
+
+    def test_rejects_rerun_that_shares_the_pull_request_group(self) -> None:
+        malformed = VALID.replace(" && github.run_attempt == 1", "")
+        with self.assertRaisesRegex(AssertionError, "unsafe concurrency group"):
+            validate_workflow_text(Path("rerun.yml"), malformed)
+
+    def test_rejects_flow_style_pull_request_trigger(self) -> None:
+        malformed = VALID.replace(
+            "on:\n  pull_request:\n    types: [opened, synchronize, reopened, ready_for_review, converted_to_draft, closed]",
+            "on: [pull_request]",
+        )
+        with self.assertRaisesRegex(AssertionError, "pull-request trigger"):
+            validate_workflow_text(Path("flow.yml"), malformed)
+
+    def test_ignores_comment_lookalikes(self) -> None:
+        malformed = VALID.replace("  pull_request:", "  # pull_request:")
+        with self.assertRaisesRegex(AssertionError, "pull-request trigger"):
+            validate_workflow_text(Path("comment.yml"), malformed)
+
+
+if __name__ == "__main__":
+    unittest.main()