chore(devex): lint pull request event fanout

.agents/skills/authoring-ci-workflows/SKILL.md

@@ -26,7 +26,7 @@ The linters own the mechanical rules (below); this skill is the **judgment calls
 ## What the linters already enforce
 
 Run `bin/hogli lint:workflows` and `actionlint` before pushing — they gate CI, and they (not this list) are the source of truth for what's enforced.
-Today that's: `timeout-minutes` on every job, the canonical PR concurrency block, `dorny/paths-filter` negation safety, justification for full-depth checkouts, cache-write gating, semgrep service coverage, required-check gate hygiene, and generic GHA correctness (bad `secrets.*` / `needs:` refs, deprecated `::set-output`, unknown runner labels).
+Today that's: `timeout-minutes` on every job, the canonical PR concurrency block, a repo-wide budget for unscoped PR event dispatches, `dorny/paths-filter` negation safety, justification for full-depth checkouts, cache-write gating, semgrep service coverage, required-check gate hygiene, and generic GHA correctness (bad `secrets.*` / `needs:` refs, deprecated `::set-output`, unknown runner labels).
 Third-party action digests are bumped by Renovate.
 
 ## The dispatch budget (500 runs / 10s / repo)

tools/hogli-commands/hogli_commands/tests/test_workflow_lint.py

@@ -14,6 +14,7 @@
 
 import pytest
 
+from click.testing import CliRunner
 from hogli_commands.workflow_lint.check import CheckResult, WorkflowCheck
 from hogli_commands.workflow_lint.checks import CHECKS, _build_lookup, get_check
 from hogli_commands.workflow_lint.checks.cache_writes import (
@@ -27,8 +28,10 @@
 from hogli_commands.workflow_lint.checks.dorny_negation import DornyNegationCheck
 from hogli_commands.workflow_lint.checks.job_timeouts import JobTimeoutsCheck
 from hogli_commands.workflow_lint.checks.pr_concurrency import PrConcurrencyCheck
+from hogli_commands.workflow_lint.checks.pr_event_fanout import PrEventFanoutCheck
 from hogli_commands.workflow_lint.checks.required_gates import RequiredGateCheck
 from hogli_commands.workflow_lint.checks.semgrep_services_coverage import SemgrepServicesCoverageCheck
+from hogli_commands.workflow_lint.cli import cmd_lint_workflows
 from hogli_commands.workflow_lint.model import PR_TRIGGERS, Workflow, WorkflowParseError, read_workflows
 
 
@@ -397,6 +400,97 @@ def test_skips_listed_filenames(self, tmp_path: Path) -> None:
         assert PrConcurrencyCheck().run(_read_all(tmp_path)).issues == []
 
 
+# ---------------------------------------------------------------------------
+# PrEventFanoutCheck
+# ---------------------------------------------------------------------------
+
+
+class TestPrEventFanoutCheck:
+    @pytest.mark.parametrize(
+        "workflow,budget,expected",
+        [
+            (
+                """
+                name: Agent
+                on:
+                  pull_request:
+                    types: [closed]
+                  pull_request_target:
+                    types: [opened, reopened, ready_for_review, edited]
+                jobs: {}
+                """,
+                {"closed": 1},
+                [
+                    "unscoped `edited` PR dispatch fanout is 1; budget is 0",
+                    "unscoped `opened` PR dispatch fanout is 1; budget is 0",
+                    "unscoped `ready_for_review` PR dispatch fanout is 1; budget is 0",
+                    "unscoped `reopened` PR dispatch fanout is 1; budget is 0",
+                ],
+            ),
+            (
+                """
+                name: New workflow
+                on: [pull_request]
+                jobs: {}
+                """,
+                {},
+                [
+                    "unscoped `opened` PR dispatch fanout is 1; budget is 0",
+                    "unscoped `reopened` PR dispatch fanout is 1; budget is 0",
+                    "unscoped `synchronize` PR dispatch fanout is 1; budget is 0",
+                ],
+            ),
+            (
+                """
+                name: Focused
+                on:
+                  pull_request:
+                    paths: [products/example/**]
+                jobs: {}
+                """,
+                {},
+                [],
+            ),
+            # paths-ignore usually excludes a narrow slice, so it still fires on nearly every PR.
+            (
+                """
+                name: Nearly everything
+                on:
+                  pull_request:
+                    paths-ignore: [docs/**]
+                jobs: {}
+                """,
+                {},
+                [
+                    "unscoped `opened` PR dispatch fanout is 1; budget is 0",
+                    "unscoped `reopened` PR dispatch fanout is 1; budget is 0",
+                    "unscoped `synchronize` PR dispatch fanout is 1; budget is 0",
+                ],
+            ),
+            # Label-driven workflows are the intended use, and labels cannot burst.
+            (
+                """
+                name: Opt in by label
+                on:
+                  pull_request:
+                    types: [labeled, unlabeled]
+                jobs: {}
+                """,
+                {},
+                [],
+            ),
+        ],
+    )
+    def test_counts_unscoped_pr_dispatches(
+        self, tmp_path: Path, workflow: str, budget: dict[str, int], expected: list[str]
+    ) -> None:
+        _write(tmp_path, "workflow.yml", workflow)
+
+        result = PrEventFanoutCheck(budget=budget).run(_read_all(tmp_path))
+
+        assert [issue.message for issue in result.issues] == expected
+
+
 # ---------------------------------------------------------------------------
 # DornyNegationCheck
 # ---------------------------------------------------------------------------
@@ -939,6 +1033,34 @@ def test_run_returns_check_result(self, tmp_path: Path) -> None:
             assert isinstance(result, CheckResult)
 
 
+# ---------------------------------------------------------------------------
+# CLI
+# ---------------------------------------------------------------------------
+
+
+class TestCli:
+    def test_fanout_over_budget_fails_run(self, tmp_path: Path) -> None:
+        _write(
+            tmp_path,
+            "assigned.yml",
+            """
+            name: Assigned
+            on:
+              pull_request:
+                types: [assigned]
+            jobs: {}
+            """,
+        )
+
+        result = CliRunner().invoke(
+            cmd_lint_workflows,
+            ["--check", "WF008", "--workflows-dir", str(tmp_path)],
+        )
+
+        assert result.exit_code == 1
+        assert "1 issue(s) across 1 check(s)" in result.output
+
+
 # The shape these fixtures guard against: a `changes` detector cleared with a bare
 # `== "failure"`, then its outputs read to decide "nothing to test". Those outputs
 # are empty on a cancelled job, so the gate exits 0 green with no tests run.

tools/hogli-commands/hogli_commands/workflow_lint/checks/__init__.py

@@ -17,6 +17,7 @@
 from .dorny_negation import DornyNegationCheck
 from .job_timeouts import JobTimeoutsCheck
 from .pr_concurrency import PrConcurrencyCheck
+from .pr_event_fanout import PrEventFanoutCheck
 from .required_gates import RequiredGateCheck
 from .semgrep_services_coverage import SemgrepServicesCoverageCheck
 
@@ -28,6 +29,7 @@
     CheckoutFullDepthCheck(),
     CacheWriteGateCheck(),
     RequiredGateCheck(),
+    PrEventFanoutCheck(),
 ]
 
 

tools/hogli-commands/hogli_commands/workflow_lint/checks/pr_event_fanout.py

@@ -0,0 +1,108 @@
+"""Cap repo-wide fanout from unscoped pull request event subscriptions.
+
+GitHub counts each directly triggered workflow as a separate run. Small jobs
+that listen to every PR should therefore share an existing dispatcher instead
+of adding another top-level ``pull_request`` or ``pull_request_target`` trigger.
+A trigger-level ``paths:`` allowlist is excluded, because it only dispatches for
+a subset of changes. ``paths-ignore`` still counts: it usually excludes a narrow
+slice, so the workflow fires on nearly every PR anyway.
+
+The per-action ceilings make any increase explicit in code review. Raising one
+is allowed when a separate dispatch is justified, but it spends a shared
+repo-wide budget and should not happen as a side effect of adding a small job.
+"""
+
+from __future__ import annotations
+
+from collections import Counter
+from collections.abc import Iterator, Mapping
+
+from ..check import CheckResult, Issue, WorkflowCheck
+from ..model import PR_TRIGGERS, Workflow
+
+DEFAULT_PR_ACTIONS = frozenset({"opened", "reopened", "synchronize"})
+
+# Labels arrive one PR at a time, so they cannot produce the simultaneous burst this
+# budget guards, and every label subscriber left in the tree wants the trigger. Merge
+# gates are a separate rule that a repo-wide sum cannot express; AGENTS.md owns it.
+UNBUDGETED_ACTIONS = frozenset({"labeled", "unlabeled"})
+
+PR_EVENT_FANOUT_BUDGET: Mapping[str, int] = {
+    "closed": 3,
+    "converted_to_draft": 1,
+    "edited": 3,
+    "opened": 28,
+    "ready_for_review": 11,
+    "reopened": 24,
+    "review_requested": 1,
+    "synchronize": 28,
+}
+
+
+def _trigger_configurations(on: object) -> dict[str, object]:
+    if isinstance(on, str):
+        return {on: None}
+    if isinstance(on, list):
+        return dict.fromkeys(str(trigger) for trigger in on)
+    if isinstance(on, dict):
+        return {str(trigger): config for trigger, config in on.items()}
+    return {}
+
+
+def _configured_actions(config: object) -> frozenset[str]:
+    types = config.get("types") if isinstance(config, dict) else None
+    if isinstance(types, str):
+        return frozenset({types})
+    if isinstance(types, list):
+        # An empty list selects no activity type, so the workflow never dispatches.
+        return frozenset(str(action) for action in types)
+    return DEFAULT_PR_ACTIONS
+
+
+def _has_paths_filter(config: object) -> bool:
+    return isinstance(config, dict) and isinstance(config.get("paths"), list) and bool(config["paths"])
+
+
+def _unscoped_pr_actions(workflow: Workflow) -> Iterator[str]:
+    for event, config in _trigger_configurations(workflow.on).items():
+        if event not in PR_TRIGGERS:
+            continue
+        if _has_paths_filter(config):
+            continue
+        yield from _configured_actions(config) - UNBUDGETED_ACTIONS
+
+
+class PrEventFanoutCheck(WorkflowCheck):
+    id = "WF008-pr-event-fanout"
+    label = "PR event fanout"
+    description = "unscoped PR event subscriptions stay within the repo-wide workflow dispatch budget"
+
+    def __init__(self, budget: Mapping[str, int] | None = None) -> None:
+        self._budget = dict(PR_EVENT_FANOUT_BUDGET if budget is None else budget)
+
+    @property
+    def fix_hint(self) -> str | None:
+        return (
+            "Avoid adding another always-fire workflow run. Fold small jobs into an existing dispatcher "
+            "with the same event and security context, or add a trigger-level `paths:` filter when the whole "
+            "workflow is skippable. If another dispatch is necessary, raise the relevant "
+            "`PR_EVENT_FANOUT_BUDGET` ceiling so the cost is explicit in review."
+        )
+
+    def run(self, workflows: list[Workflow]) -> CheckResult:
+        action_counts: Counter[str] = Counter()
+        for workflow in workflows:
+            action_counts.update(_unscoped_pr_actions(workflow))
+
+        result = CheckResult()
+        for action, count in sorted(action_counts.items()):
+            budget = self._budget.get(action, 0)
+            if count <= budget:
+                continue
+            result.issues.append(
+                Issue(
+                    workflow=".github/workflows",
+                    message=f"unscoped `{action}` PR dispatch fanout is {count}; budget is {budget}",
+                )
+            )
+        return result