diff --git a/.github/actions/noema-review/two_phase.py b/.github/actions/noema-review/two_phase.py old mode 100755 new mode 100644 diff --git a/.github/workflows/actions-queue-health.yml b/.github/workflows/actions-queue-health.yml new file mode 100644 index 0000000000..2084765946 --- /dev/null +++ b/.github/workflows/actions-queue-health.yml @@ -0,0 +1,55 @@ +name: GitHub Actions queue health + +on: + schedule: + - cron: "7 * * * *" + +concurrency: + group: github-actions-queue-health + cancel-in-progress: false + +permissions: + contents: read + actions: read + +jobs: + collect: + name: Collect exact-head queue evidence + runs-on: ubuntu-24.04 + timeout-minutes: 30 + permissions: + contents: read + actions: read + steps: + - name: Harden the runner (Audit all outbound calls) + uses: step-security/harden-runner@b09bb98e06d4d774595224525879c09bc6e98c40 # v2.20.1 + with: + egress-policy: audit + + - name: Checkout trusted queue-health source + uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0 + with: + persist-credentials: false + + - name: Collect read-only repository and runner evidence + env: + GH_TOKEN: ${{ secrets.PR_REVIEW_MERGE_TOKEN || secrets.OPENCODE_APPROVE_TOKEN }} + run: | + if [ -z "${GH_TOKEN:-}" ]; then + echo "::error::PR_REVIEW_MERGE_TOKEN or OPENCODE_APPROVE_TOKEN is required for cross-repository queue reads." + exit 1 + fi + echo "::add-mask::$GH_TOKEN" + python3 scripts/ci/actions_queue_health.py \ + --allowlist config/actions_queue_health_repositories.json \ + --output-json "$RUNNER_TEMP/actions-queue-health.json" \ + --output-html "$RUNNER_TEMP/actions-queue-health.html" + + - name: Upload queue-health evidence + uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1 + with: + name: github-actions-queue-health-${{ github.run_id }} + path: | + ${{ runner.temp }}/actions-queue-health.json + ${{ runner.temp }}/actions-queue-health.html + if-no-files-found: error diff --git a/config/actions_queue_health_repositories.json b/config/actions_queue_health_repositories.json new file mode 100644 index 0000000000..e8c9374394 --- /dev/null +++ b/config/actions_queue_health_repositories.json @@ -0,0 +1,11 @@ +{ + "repositories": [ + "ContextualWisdomLab/.github", + "ContextualWisdomLab/ConceptWeave", + "ContextualWisdomLab/ELUNVERA", + "ContextualWisdomLab/TEPP", + "ContextualWisdomLab/contextual-orchestrator", + "ContextualWisdomLab/fast-mlsirm", + "ContextualWisdomLab/naruon" + ] +} \ No newline at end of file diff --git a/docs/doctoring/actions-queue-cancelled-before-runner.md b/docs/doctoring/actions-queue-cancelled-before-runner.md new file mode 100644 index 0000000000..983f36549a --- /dev/null +++ b/docs/doctoring/actions-queue-cancelled-before-runner.md @@ -0,0 +1,56 @@ +# Actions queue cancellation before runner assignment + +## Status + +Proposed owner-side diagnostic extension for `ContextualWisdomLab/.github#1150` and the organization Actions incident tracked by `ContextualWisdomLab/.github#712`. + +## Problem + +A current pull-request head can produce a terminal GitHub Actions run whose job was cancelled before any runner was assigned or any step executed. Treating that evidence as a generic terminal job loses the first non-executed boundary and can mislead incident triage even though it must never count as passing evidence. + +Observed organization evidence on 2026-09-02 includes `.github#1653`, where a current-head `Repository Metadata Reconcile` job terminated `cancelled` after previously showing `runner_id=0`, empty runner identity, and `steps=[]`. Separate ContextualWisdomLab repositories also reproduce zero-job `startup_failure` and long-lived unassigned queue states, so these states must remain distinct rather than being collapsed into a product-source failure. + +A second adapter-boundary case is `pull_request_target`: GitHub records the workflow run against the base commit while the linked pull-request object carries the exact pull-request head. A terminal diagnostic collector that searches only by the current pull-request `head_sha` therefore cannot see a target-triggered cancellation even when its linked pull-request identity is current. Conversely, once target evidence is collected, a concurrent push or close can make the identity snapshot used for classification stale unless the collector revalidates the PR view after all terminal/job reads. + +## Decision + +The queue-health collector keeps external GitHub conclusion values unchanged at the adapter boundary and adds a semantic internal/report classification. For exact current heads it now: + +- retains `startup_failure` and `cancelled` terminal diagnostics from the bounded exact-`head_sha` `status=completed` query for ordinary pull-request/head-bound runs, filtering the returned conclusion locally; +- performs a bounded `status=cancelled&event=pull_request_target` candidate read for the target-triggered cancellation case and retains a candidate only after the existing linked pull-request number/head identity resolver proves it belongs to an exact current head; +- does **not** send `status=startup_failure` to GitHub's workflow-run list endpoint because that value is not in the endpoint's documented `status`/conclusion filter enumeration; target-triggered zero-job startup-failure discovery therefore remains a separate unresolved diagnostic gap rather than being implemented through an invalid REST request; +- fetches job evidence only for retained current-head terminal diagnostics; +- re-reads the bounded open-PR identity view after terminal and job evidence collection and fails the repository snapshot if PR number/state/head identity differs from the view used for classification; +- classifies a job as `cancelled_before_runner_assignment` only when both the run and that job conclude `cancelled`, the job has no runner assignment, and it has zero executed/materialized steps; +- never reclassifies sibling jobs that concluded `skipped`, `success`, or another non-cancelled state merely because their parent run concluded `cancelled`; +- keeps ordinary exact-head zero-job startup failures as `startup_failure_before_job_materialization`; +- reports an additive `admission_state` and a summary count without changing any GitHub check conclusion or synthesizing success; +- recommends inspection of Actions runner admission, billing/usage, runner-group policy, scheduler capacity, concurrency, and cancellation provenance rather than leaf-source churn or gate weakening. + +## TDD lineage + +RED commit `af72a26e0d1d845a7b447a63c7d4de4867815a87` added the first deterministic regression whose current-head cancelled run has one job with `runner_id=0`, an empty runner name, and `steps=[]`. GREEN commit `79e0758d0583474934327039b065956976c64453` introduced the initial cancellation classification. + +RED commit `b4f95bc290e625649b8ce7ae59e157c3869466f2` then captured two successor defects found on the live writer: a `pull_request_target` cancellation whose run-level SHA is the base commit but whose linked pull-request head is current, and a skipped sibling job inside a cancelled run that must not be counted as a pre-runner cancellation incident. GREEN commit `5a4950bb996f80f7be2519432a3f5b74bea02d58` added target-event candidate collection with linked-head verification and required the matched job itself to conclude `CANCELLED` before applying the semantic incident classification. + +Primary-source verification then found that GitHub's documented repository workflow-run `status` filter accepts `completed`, `action_required`, `cancelled`, `failure`, `neutral`, `skipped`, `stale`, `success`, `timed_out`, `in_progress`, `queued`, `requested`, `waiting`, and `pending`, but not `startup_failure`. RED `b99839bdcffecccc88b364a9813676b3964535b9` rejects any attempted `status=startup_failure` request in the deterministic target-cancellation fixture. GREEN `f567b1182308e4b45e22bf2f13b214998f59f5d0` narrows target-event filtering to the supported `cancelled` conclusion while leaving ordinary exact-head `status=completed` collection and local `startup_failure` conclusion classification intact. + +Review of that successor exposed a final consistency-window defect: target cancellation/job reads occurred after the collector's prior `final_pull_requests` read, so a later push or close could allow stale target evidence to survive. RED `d3a11383ce717217ec4c80a5d65c84aa947570e3` changes the PR head only after target and job evidence has been read and requires fail-closed rejection. GREEN `7683d2219c8007f9e7fa6001c98d0944290fa756` adds the post-evidence identity read and rejects any number/state/head divergence before a repository snapshot is emitted. + +## Compatibility and risk + +This is an additive diagnostic-contract change. It does not mutate repository branches outside the canonical PR, cancel/rerun Actions, alter branch protection, change database state, or modify an external GitHub schema. `status`, `conclusion`, `runner_id`, and related GitHub payload keys remain vendor-owned adapter fields; organization-owned report vocabulary uses semantic multiword names. + +Exact-head completed-run searches retain the existing twenty-page / 1,000-result fail-closed ceiling. Because GitHub's repository workflow-run API does not expose a pull-request-number filter for `pull_request_target`, cancelled target-event candidates are read by supported `cancelled` status and event under the same bounded ceiling, then filtered by linked current-head identity before retention. If that bounded candidate set is exceeded, or if the post-evidence PR identity view changes, the repository becomes explicit incomplete collection evidence rather than silently truncating or preserving stale evidence. This is an availability trade-off, not permission to synthesize success or churn leaf repositories. + +A cancelled run with a runner-assigned, step-executing, or non-cancelled matched job remains ordinary terminal evidence and is not reclassified as a pre-runner admission failure. A `pull_request_target` startup failure that cannot be discovered through the supported target-cancellation query also remains incomplete evidence; it is not silently treated as healthy. + +## Primary-source traceability + +GitHub, Inc. (2026). *REST API endpoints for workflow runs*. GitHub Docs. Retrieved September 2, 2026, from https://docs.github.com/en/rest/actions/workflow-runs + +The documented endpoint contract is treated as the authority for request-filter vocabulary; live GitHub run payloads remain the authority for observed run conclusions. The distinction prevents an undocumented observed conclusion such as `startup_failure` from being incorrectly assumed to be a valid REST query-filter value. + +## Verification + +Only checks produced from the unchanged final `ContextualWisdomLab/.github#1150` head qualify. Queued, pending, cancelled, zero-job startup failures, predecessor checks, or stale reviews are incomplete evidence and must not be transferred to a newer head. The RED/GREEN lineage above documents source intent; hosted 100% statement/branch/docstring and required-workflow evidence must be re-established on the final exact head before ordinary merge. \ No newline at end of file diff --git a/docs/doctoring/actions-queue-health.md b/docs/doctoring/actions-queue-health.md new file mode 100644 index 0000000000..e7d08007a2 --- /dev/null +++ b/docs/doctoring/actions-queue-health.md @@ -0,0 +1,106 @@ +# GitHub Actions queue-health evidence + +The scheduled `actions-queue-health.yml` workflow reads a fixed allowlist of +CWL repositories once per hour and publishes a JSON report plus a keyboard- +readable HTML report as an artifact. The collector uses only `gh api` reads +through the configured cross-repository `PR_REVIEW_MERGE_TOKEN` or +`OPENCODE_APPROVE_TOKEN`; it fails visibly when neither credential is present. +It does not cancel runs, mutate branches, dispatch workflows, or alter merge +gates, and it never relies on the central repository's scoped `GITHUB_TOKEN` +for sibling-repository reads. + +The report schema is `actions.queue_health.v1`. Each observed run records its +repository, pull-request number, head SHA, event, run attempt, concurrency +group (or an explicit unavailable marker), stable workflow identity, queue age, +job state, and runner assignment. When GitHub supplies a positive +`workflow_id`, the report exposes `workflow_identity` as `workflow_id:` and +uses that value for duplicate-lane grouping; `workflow_name` remains +presentation data. Older/offline v1 snapshots that lack `workflow_id` retain a +compatibility fallback of `workflow_name:`. A malformed present +`workflow_id` fails closed instead of being coerced. + +A run is `current_head` only when its linked open pull request and head SHA +match. The match compares the open pull request's head SHA against the *linked* +pull-request entry's head SHA carried on the run (`run.pull_requests[].head.sha`), +never against the run-level `head_sha`. `pull_request_target`-triggered runs +report the base-branch commit that was checked out as their run-level +`head_sha`, so comparing against that value would misclassify a genuinely +active, current required-workflow run as obsolete and skip its job evidence. +Stale linked runs are `obsolete`; runs without a pull-request link are +`unlinked`. Queued evidence remains incomplete even when a report is +successfully produced. GitHub's `waiting` job status (paused on an environment +or deployment approval) is also treated as pending evidence, distinct from a +runner-capacity blocker. + +Pull-request identity is sampled before and after the bounded active-run +sweeps. The repository snapshot is accepted only when the open pull-request +number/state/head view is unchanged. A push, closure, or other identity change +between those samples becomes repository-scoped incomplete evidence instead of +being allowed to invert current/obsolete classification. A pull-request +response with incomplete head/base identity retains one bounded retry after a +one-second delay. + +Queue age for a fetched job is measured from that job's own `created_at`, not +the parent run's, so a later job in an already in-progress run (for example one +gated by `needs:`) that only just became eligible is not measured against the +whole run's age and does not trigger a false capacity-breach alert. Every row +exports both `queue_age_started_at` and `queue_age_source` (`job_created_at` or +`run_created_at`) so consumers can reproduce the reported `queue_age_seconds`. +Requested, pending, and queued runs intentionally use run-level evidence when +GitHub has not supplied job detail. + +Two bounded active-status sweeps run in opposite orders and must agree before +the snapshot is accepted. This prevents historical completed runs from +exhausting the bound while rejecting evidence that changes between partitioned +reads. Each status read is capped at one 50-run page, limiting collection to ten +run-list calls per repository; exceeding the cap is reported as incomplete +evidence. Current-head `in_progress` and `waiting` runs make the additional jobs +API read needed to distinguish concrete runner assignment from an environment +or deployment approval wait. + +List endpoints use collector-controlled GitHub API pagination with at most 20 +explicit page reads; the collector never asks GitHub CLI to download an +unbounded page set and never requests page 21. Pull-request and job lists use +pages of 100 records; workflow-run lists use pages of 50 so a large Actions +queue does not require one oversized response. An incomplete, malformed, or +larger response is recorded as repository-scoped incomplete evidence and the +collector continues with the remaining allowlisted repositories; it never +silently claims that the visible page is the whole queue. The JSON and HTML +reports expose each collection error explicitly. + +Every external `gh api` read has a 30-second subprocess timeout, and the +collector job has a 30-minute execution ceiling. A timeout is typed as +incomplete queue evidence rather than success. Repository names reject `.` and +`..` path segments. Offline snapshots also reject duplicate repository entries +before counting runs so repeated input cannot inflate the reported queue. + +The default queue-age SLO is 900 seconds. A current-head job that remains +unassigned beyond that limit produces a warning and an explicit manual action +to inspect runner capacity, billing, runner-group policy, environment approval, +and concurrency saturation. The workflow intentionally remains read-only and +fail-closed when GitHub API or runner evidence is unavailable. Paged API reads +are not atomic; changing totals are retained only when the collected records +cover the largest observed total, and the report remains explicitly an +observation rather than a merge decision. + +Implementation ownership is intentionally split without duplicate collector +copies: `actions_queue_health_core.py` owns the shared bounded parsing and +reporting primitives, while the executable `actions_queue_health.py` entrypoint +owns stable pull/workflow identity and audit-provenance reconciliation. Tests +load the executable boundary used by the scheduled workflow. + +The allowlist is deliberately explicit in +`config/actions_queue_health_repositories.json`; adding a repository requires +review of its governance and data boundary. This first slice does not claim +that a queued run is obsolete or safe to cancel. + +## References + +GitHub. (n.d.). *REST API endpoints for workflow runs*. Retrieved August 20, +2026, from https://docs.github.com/en/rest/actions/workflow-runs + +Internet Engineering Task Force. (2022). *HTTP semantics* (RFC 9110). +https://www.rfc-editor.org/rfc/rfc9110 + +OWASP Foundation. (n.d.). *Path traversal*. Retrieved August 20, 2026, from +https://owasp.org/www-community/attacks/Path_Traversal diff --git a/scripts/ci/actions_queue_health.py b/scripts/ci/actions_queue_health.py new file mode 100644 index 0000000000..77a0d33906 --- /dev/null +++ b/scripts/ci/actions_queue_health.py @@ -0,0 +1,559 @@ +#!/usr/bin/env python3 +"""Queue-health CLI with stable identity and audit-provenance guarantees. + +The shared collector implementation lives in ``actions_queue_health_core.py``. +This entrypoint owns the consistency boundary that binds active-run evidence to +a stable pull-request view, carries stable workflow identity, and exports the +exact timestamp used for queue-age calculations. +""" + +from __future__ import annotations + +from datetime import datetime, timezone +import importlib.util +from pathlib import Path +import sys +from urllib.parse import quote + +_CORE_MODULE_PATH = Path(__file__).with_name("actions_queue_health_core.py") +_CORE_MODULE_SPEC = importlib.util.spec_from_file_location( + "actions_queue_health_core", _CORE_MODULE_PATH +) +if _CORE_MODULE_SPEC is None or _CORE_MODULE_SPEC.loader is None: # pragma: no cover + raise RuntimeError("unable to load queue-health core module") +_core_module = importlib.util.module_from_spec(_CORE_MODULE_SPEC) +sys.modules.setdefault("actions_queue_health_core", _core_module) +_CORE_MODULE_SPEC.loader.exec_module(_core_module) + +for core_symbol_name, core_symbol in vars(_core_module).items(): + if not core_symbol_name.startswith("__"): + globals()[core_symbol_name] = core_symbol + +_CORE_NORMALISE_RUN = _core_module._normalise_run +_CORE_BUILD_REPORT = _core_module.build_report +TERMINAL_DIAGNOSTIC_STATUSES = ("startup_failure", "cancelled") +TARGET_TERMINAL_DIAGNOSTIC_STATUSES = ("cancelled",) +TERMINAL_DIAGNOSTIC_MAX_API_PAGES = MAX_API_PAGES + + +def _normalise_run( + repository_name: str, + workflow_run: dict[str, Any], + workflow_jobs: list[dict[str, Any]], +) -> dict[str, Any]: + """Normalize one run while preserving stable GitHub workflow identity.""" + if not isinstance(workflow_run, dict): + raise QueueHealthError("workflow run must be an object") + workflow_id = workflow_run.get("workflow_id") + if workflow_id is not None and ( + isinstance(workflow_id, bool) + or not isinstance(workflow_id, int) + or workflow_id <= 0 + ): + raise QueueHealthError("workflow id must be a positive integer") + + normalized_run = _CORE_NORMALISE_RUN( + repository_name, workflow_run, workflow_jobs + ) + workflow_name = normalized_run["workflow_name"] + normalized_run["workflow_id"] = workflow_id + normalized_run["workflow_identity"] = ( + f"workflow_id:{workflow_id}" + if workflow_id is not None + else f"workflow_name:{workflow_name}" + ) + return normalized_run + + +_core_module._normalise_run = _normalise_run + + +def _read_pull_request_snapshot( + pulls_endpoint: str, *, runner: Runner +) -> list[dict[str, Any]]: + """Read and normalize one bounded open-pull-request identity snapshot.""" + pull_request_entries = _list_payload( + github_json(pulls_endpoint, paginate=True, runner=runner), + "pulls", + max_items=MAX_API_PAGE_SIZE * MAX_API_PAGES, + ) + return sorted( + (_normalise_pull_request(pull_request) for pull_request in pull_request_entries), + key=lambda pull_request: pull_request["number"], + ) + + +def _pull_request_identity_view( + pull_requests: list[dict[str, Any]], +) -> dict[int, tuple[str, str]]: + """Return the number/state/head view that must stay stable during collection.""" + return { + pull_request["number"]: ( + str(pull_request.get("state") or ""), + str(pull_request.get("head_sha") or ""), + ) + for pull_request in pull_requests + } + + +def collect_snapshot( + repositories: Sequence[str], + *, + runner: Runner = subprocess.run, + generated_at: str | None = None, +) -> dict[str, Any]: + """Collect active and pre-job terminal evidence bound to stable PR identities.""" + validated_repositories = sorted( + {_repository_name(repository_name) for repository_name in repositories} + ) + if len(validated_repositories) != len(repositories): + raise QueueHealthError("collection repository list contains duplicates") + + snapshot_timestamp = generated_at or datetime.now(timezone.utc).isoformat().replace( + "+00:00", "Z" + ) + parse_timestamp(snapshot_timestamp) + collected_repositories: list[dict[str, Any]] = [] + collection_errors: list[dict[str, str]] = [] + active_statuses = ("in_progress", "pending", "queued", "requested", "waiting") + + for repository_name in validated_repositories: + try: + repository_metadata = github_json( + f"repos/{repository_name}", runner=runner + ) + if not isinstance(repository_metadata, dict): + raise QueueHealthError( + f"repository metadata for {repository_name} is not an object" + ) + pulls_endpoint = ( + f"repos/{repository_name}/pulls?state=open&per_page={MAX_API_PAGE_SIZE}" + ) + try: + initial_pull_requests = _read_pull_request_snapshot( + pulls_endpoint, runner=runner + ) + except IncompletePullRequestIdentity: + time.sleep(PULL_REQUEST_RETRY_DELAY_SECONDS) + initial_pull_requests = _read_pull_request_snapshot( + pulls_endpoint, runner=runner + ) + except QueueHealthError as collection_error: + collection_errors.append( + {"repository": repository_name, "error": str(collection_error)} + ) + continue + + try: + active_snapshots: list[dict[int, dict[str, Any]]] = [] + for status_order in (active_statuses, tuple(reversed(active_statuses))): + active_snapshot: dict[int, dict[str, Any]] = {} + for workflow_status in status_order: + workflow_runs = _list_payload( + github_json( + f"repos/{repository_name}/actions/runs?status={workflow_status}" + f"&per_page={WORKFLOW_RUN_PAGE_SIZE}", + paginate=True, + max_pages=ACTIVE_RUN_MAX_API_PAGES, + runner=runner, + ), + "workflow_runs", + max_items=( + WORKFLOW_RUN_PAGE_SIZE * ACTIVE_RUN_MAX_API_PAGES + ), + ) + for workflow_run in workflow_runs: + workflow_run_id = workflow_run.get("id") + if ( + isinstance(workflow_run_id, bool) + or not isinstance(workflow_run_id, int) + or workflow_run_id <= 0 + ): + raise QueueHealthError( + "workflow run id must be a positive integer" + ) + active_snapshot[workflow_run_id] = workflow_run + active_snapshots.append(active_snapshot) + + first_snapshot, second_snapshot = active_snapshots + first_run_states = { + workflow_run_id: str(workflow_run.get("status") or "").upper() + for workflow_run_id, workflow_run in first_snapshot.items() + } + second_run_states = { + workflow_run_id: str(workflow_run.get("status") or "").upper() + for workflow_run_id, workflow_run in second_snapshot.items() + } + if first_run_states != second_run_states: + raise QueueHealthError( + "active workflow run snapshot changed during collection" + ) + + try: + final_pull_requests = _read_pull_request_snapshot( + pulls_endpoint, runner=runner + ) + except IncompletePullRequestIdentity: + time.sleep(PULL_REQUEST_RETRY_DELAY_SECONDS) + try: + final_pull_requests = _read_pull_request_snapshot( + pulls_endpoint, runner=runner + ) + except QueueHealthError as retry_error: + raise QueueHealthError( + "pull-request identity validation failed: " + f"{retry_error}" + ) from retry_error + + if ( + _pull_request_identity_view(initial_pull_requests) + != _pull_request_identity_view(final_pull_requests) + ): + raise QueueHealthError( + "pull-request identity snapshot changed during collection" + ) + + pull_requests_by_number = { + pull_request["number"]: pull_request + for pull_request in final_pull_requests + } + terminal_diagnostic_snapshot: dict[int, dict[str, Any]] = {} + current_head_shas = sorted( + {pull_request["head_sha"] for pull_request in final_pull_requests} + ) + for current_head_sha in current_head_shas: + encoded_head_sha = quote(current_head_sha, safe="") + workflow_runs = _list_payload( + github_json( + f"repos/{repository_name}/actions/runs?status=completed" + f"&head_sha={encoded_head_sha}" + f"&per_page={WORKFLOW_RUN_PAGE_SIZE}", + paginate=True, + max_pages=TERMINAL_DIAGNOSTIC_MAX_API_PAGES, + runner=runner, + ), + "workflow_runs", + max_items=( + WORKFLOW_RUN_PAGE_SIZE * TERMINAL_DIAGNOSTIC_MAX_API_PAGES + ), + ) + for workflow_run in workflow_runs: + if str(workflow_run.get("conclusion") or "").lower() not in ( + TERMINAL_DIAGNOSTIC_STATUSES + ): + continue + workflow_run_id = workflow_run.get("id") + if ( + isinstance(workflow_run_id, bool) + or not isinstance(workflow_run_id, int) + or workflow_run_id <= 0 + ): + raise QueueHealthError( + "workflow run id must be a positive integer" + ) + terminal_diagnostic_snapshot[workflow_run_id] = workflow_run + + for terminal_status in TARGET_TERMINAL_DIAGNOSTIC_STATUSES: + target_workflow_runs = _list_payload( + github_json( + f"repos/{repository_name}/actions/runs?status={terminal_status}" + "&event=pull_request_target" + f"&per_page={WORKFLOW_RUN_PAGE_SIZE}", + paginate=True, + max_pages=TERMINAL_DIAGNOSTIC_MAX_API_PAGES, + runner=runner, + ), + "workflow_runs", + max_items=( + WORKFLOW_RUN_PAGE_SIZE * TERMINAL_DIAGNOSTIC_MAX_API_PAGES + ), + ) + for workflow_run in target_workflow_runs: + normalized_candidate = _normalise_run( + repository_name, workflow_run, [] + ) + identity_state, _ = _run_identity( + normalized_candidate, pull_requests_by_number + ) + if identity_state != "current_head": + continue + terminal_diagnostic_snapshot[normalized_candidate["id"]] = ( + workflow_run + ) + + observed_snapshot = dict(second_snapshot) + observed_snapshot.update(terminal_diagnostic_snapshot) + runs_by_id: dict[int, dict[str, Any]] = {} + for workflow_run_id, workflow_run in observed_snapshot.items(): + normalized_run = _normalise_run( + repository_name, workflow_run, [] + ) + identity_state, _ = _run_identity( + normalized_run, pull_requests_by_number + ) + needs_job_evidence = ( + identity_state == "current_head" + and ( + normalized_run["status"] in {"IN_PROGRESS", "WAITING"} + or normalized_run["conclusion"] + in {status.upper() for status in TERMINAL_DIAGNOSTIC_STATUSES} + ) + ) + if not needs_job_evidence: + runs_by_id[workflow_run_id] = normalized_run + continue + + jobs_payload = github_json( + f"repos/{repository_name}/actions/runs/{workflow_run_id}/jobs" + f"?per_page={MAX_API_PAGE_SIZE}", + paginate=True, + runner=runner, + ) + workflow_jobs = _list_payload( + jobs_payload, + "jobs", + max_items=MAX_API_PAGE_SIZE * MAX_API_PAGES, + ) + runs_by_id[workflow_run_id] = _normalise_run( + repository_name, workflow_run, workflow_jobs + ) + + try: + post_evidence_pull_requests = _read_pull_request_snapshot( + pulls_endpoint, runner=runner + ) + except IncompletePullRequestIdentity: + time.sleep(PULL_REQUEST_RETRY_DELAY_SECONDS) + try: + post_evidence_pull_requests = _read_pull_request_snapshot( + pulls_endpoint, runner=runner + ) + except QueueHealthError as retry_error: + raise QueueHealthError( + "pull-request identity validation failed: " + f"{retry_error}" + ) from retry_error + if ( + _pull_request_identity_view(final_pull_requests) + != _pull_request_identity_view(post_evidence_pull_requests) + ): + raise QueueHealthError( + "pull-request identity snapshot changed during evidence collection" + ) + except QueueHealthError as collection_error: + collection_errors.append( + {"repository": repository_name, "error": str(collection_error)} + ) + continue + + collected_repositories.append( + { + "full_name": repository_name, + "default_branch": str( + repository_metadata.get("default_branch") or "" + ), + "pull_requests": final_pull_requests, + "runs": sorted( + runs_by_id.values(), key=lambda workflow_run: workflow_run["id"] + ), + } + ) + + return { + "generated_at": snapshot_timestamp, + "repositories": collected_repositories, + "collection_errors": collection_errors, + } + + +def _normalized_snapshot_runs( + snapshot: dict[str, Any], +) -> dict[tuple[str, int], dict[str, Any]]: + """Index normalized run metadata for additive report provenance fields.""" + normalized_runs: dict[tuple[str, int], dict[str, Any]] = {} + snapshot_repositories = snapshot.get("repositories") + if not isinstance(snapshot_repositories, list): + return normalized_runs + for repository_entry in snapshot_repositories: + if not isinstance(repository_entry, dict): + continue + repository_name = repository_entry.get("full_name") + workflow_runs = repository_entry.get("runs") or [] + if not isinstance(repository_name, str) or not isinstance(workflow_runs, list): + continue + for workflow_run in workflow_runs: + if not isinstance(workflow_run, dict): + continue + workflow_jobs = workflow_run.get("jobs") or [] + if not isinstance(workflow_jobs, list): + workflow_jobs = [] + normalized_run = _normalise_run( + repository_name, workflow_run, workflow_jobs + ) + normalized_runs[(repository_name, normalized_run["id"])] = normalized_run + return normalized_runs + + +def build_report( + snapshot: dict[str, Any], + *, + now: datetime | None = None, + queue_age_slo_seconds: int = DEFAULT_QUEUE_AGE_SLO_SECONDS, +) -> dict[str, Any]: + """Build the v1 report with stable workflow identity and age provenance.""" + normalized_runs = _normalized_snapshot_runs(snapshot) + report = _CORE_BUILD_REPORT( + snapshot, + now=now, + queue_age_slo_seconds=queue_age_slo_seconds, + ) + + for report_row in report["runs"]: + run_metadata = normalized_runs.get( + (report_row["repository"], report_row["run_id"]) + ) + if run_metadata is None: # pragma: no cover - core report guarantees the row. + continue + report_row["workflow_id"] = run_metadata["workflow_id"] + report_row["workflow_identity"] = run_metadata["workflow_identity"] + report_row["run_conclusion"] = run_metadata.get("conclusion", "") + report_row["jobs_materialized"] = bool(run_metadata["jobs"]) + matching_job = next( + ( + workflow_job + for workflow_job in run_metadata["jobs"] + if workflow_job["id"] == report_row["job_id"] + ), + None, + ) + report_row["admission_state"] = ( + "runner_assigned" if report_row["runner_assigned"] else "runner_not_assigned" + ) + if matching_job and matching_job.get("created_at"): + report_row["queue_age_started_at"] = matching_job["created_at"] + report_row["queue_age_source"] = "job_created_at" + else: + report_row["queue_age_started_at"] = run_metadata.get("created_at", "") + report_row["queue_age_source"] = "run_created_at" + if ( + report_row["identity_state"] == "current_head" + and report_row["run_conclusion"] == "STARTUP_FAILURE" + and not report_row["jobs_materialized"] + ): + report_row["admission_state"] = "startup_failure_before_job_materialization" + report_row["blocker"] = "startup_failure_before_job_materialization" + report_row["recommended_action"] = ( + "inspect_actions_control_plane_without_leaf_bypass" + ) + elif ( + report_row["identity_state"] == "current_head" + and report_row["run_conclusion"] == "CANCELLED" + and matching_job is not None + and matching_job.get("conclusion") == "CANCELLED" + and not report_row["runner_assigned"] + and matching_job.get("steps_count") == 0 + ): + report_row["admission_state"] = "cancelled_before_runner_assignment" + report_row["blocker"] = "cancelled_before_runner_assignment" + report_row["recommended_action"] = ( + "inspect_actions_control_plane_without_leaf_bypass" + ) + + current_pending_rows = [ + report_row + for report_row in report["runs"] + if report_row["is_pending"] + and report_row["identity_state"] == "current_head" + ] + lane_run_ids: dict[tuple[str, int, str], set[int]] = {} + lane_workflow_names: dict[tuple[str, int, str], str] = {} + for report_row in current_pending_rows: + lane_identity = ( + report_row["repository"], + report_row["pull_request_number"], + report_row["workflow_identity"], + ) + lane_run_ids.setdefault(lane_identity, set()).add(report_row["run_id"]) + lane_workflow_names.setdefault( + lane_identity, report_row["workflow_name"] + ) + + duplicate_pending_lanes = [ + { + "repository": lane_identity[0], + "pull_request_number": lane_identity[1], + "workflow_identity": lane_identity[2], + "workflow_name": lane_workflow_names[lane_identity], + "count": len(workflow_run_ids), + } + for lane_identity, workflow_run_ids in sorted(lane_run_ids.items()) + if len(workflow_run_ids) > 1 + ] + report["duplicate_pending_lanes"] = duplicate_pending_lanes + report["summary"]["duplicate_pending_lane_count"] = len( + duplicate_pending_lanes + ) + cancelled_before_runner_assignment_count = sum( + report_row.get("admission_state") == "cancelled_before_runner_assignment" + for report_row in report["runs"] + ) + report["summary"]["cancelled_before_runner_assignment_count"] = ( + cancelled_before_runner_assignment_count + ) + if cancelled_before_runner_assignment_count: + external_action = ( + "Inspect Actions runner admission, billing/usage, runner-group policy, " + "scheduler capacity, and cancellation provenance; cancelled pre-runner " + "evidence remains incomplete." + ) + if external_action not in report["summary"]["external_actions"]: + report["summary"]["external_actions"].append(external_action) + report["summary"]["external_actions"].sort() + return report + + +def main( + argv: Sequence[str] | None = None, *, stderr: TextIO = sys.stderr +) -> int: + """Collect or load a snapshot, write reports, and return a stable CLI status.""" + cli_arguments = parse_args(argv) + try: + queue_snapshot = ( + load_snapshot(cli_arguments.snapshot) + if cli_arguments.snapshot + else collect_snapshot(load_allowlist(cli_arguments.allowlist)) + ) + evaluation_time = ( + parse_timestamp(cli_arguments.now) + if cli_arguments.now + else datetime.now(timezone.utc) + ) + queue_report = build_report( + queue_snapshot, + now=evaluation_time, + queue_age_slo_seconds=cli_arguments.queue_age_slo_seconds, + ) + write_reports( + queue_report, cli_arguments.output_json, cli_arguments.output_html + ) + except (OSError, QueueHealthError, ValueError) as report_error: + print(f"ERROR: queue-health report failed: {report_error}", file=stderr) + return 2 + + breach_count = queue_report["summary"]["unassigned_slo_breached_count"] + if breach_count: + print( + "::warning::Actions queue-health found " + f"{breach_count} unassigned current-head SLO breach(es)." + ) + print( + "QUEUE_HEALTH_RESULT=" + f"observed={queue_report['summary']['observed_job_count']} " + f"pending={queue_report['summary']['pending_job_count']} " + f"slo_breaches={breach_count}" + ) + return 0 + + +if __name__ == "__main__": # pragma: no cover - exercised through CLI tests. + raise SystemExit(main()) diff --git a/scripts/ci/actions_queue_health_core.py b/scripts/ci/actions_queue_health_core.py new file mode 100644 index 0000000000..db3e5570ba --- /dev/null +++ b/scripts/ci/actions_queue_health_core.py @@ -0,0 +1,855 @@ +#!/usr/bin/env python3 +"""Produce a read-only, exact-head GitHub Actions queue-health report. + +The collector intentionally treats queued, cancelled, skipped, missing, and +unlinked evidence as incomplete. It never cancels runs, changes branches, or +turns an unavailable runner into a successful check. +""" + +from __future__ import annotations + +import argparse +from datetime import datetime, timezone +import html +import json +from pathlib import Path +import re +import subprocess +import sys +import time +from typing import Any, Callable, Sequence, TextIO + + +REPOSITORY_PATTERN = re.compile(r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$") +QUEUE_STATES = {"QUEUED", "IN_PROGRESS", "PENDING", "REQUESTED"} +TERMINAL_STATES = {"COMPLETED"} +DEFAULT_QUEUE_AGE_SLO_SECONDS = 900 +SCHEMA_VERSION = "actions.queue_health.v1" +MAX_API_PAGE_SIZE = 100 +WORKFLOW_RUN_PAGE_SIZE = 50 +MAX_API_PAGES = 20 +ACTIVE_RUN_MAX_API_PAGES = 1 +GITHUB_API_TIMEOUT_SECONDS = 30 +PULL_REQUEST_RETRY_DELAY_SECONDS = 1 +PAGINATED_PAGES_KEY = "_queue_health_pages" +Runner = Callable[..., subprocess.CompletedProcess[str]] + + +class QueueHealthError(ValueError): + """Raised when a queue-health input or trusted read is invalid.""" + + +class IncompletePullRequestIdentity(QueueHealthError): + """Raised when a pull-request read omits exact head or base identity.""" + + +def parse_timestamp(value: str) -> datetime: + """Parse an explicit UTC timestamp and reject ambiguous local time.""" + if not isinstance(value, str) or not value.strip(): + raise QueueHealthError("timestamp must be a non-empty string") + try: + parsed = datetime.fromisoformat(value.strip().replace("Z", "+00:00")) + except ValueError as exc: + raise QueueHealthError(f"invalid timestamp: {value!r}") from exc + if parsed.tzinfo is None: + raise QueueHealthError("timestamp must include a timezone") + return parsed.astimezone(timezone.utc) + + +def _repository_name(value: Any) -> str: + """Validate and return one owner/repository identifier.""" + if not isinstance(value, str) or not REPOSITORY_PATTERN.fullmatch(value): + raise QueueHealthError(f"invalid repository identifier: {value!r}") + if any(segment in {".", ".."} for segment in value.split("/")): + raise QueueHealthError(f"invalid repository identifier: {value!r}") + return value + + +def load_allowlist(path: Path) -> list[str]: + """Load a unique, sorted repository allowlist from a JSON array/object.""" + try: + payload = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + raise QueueHealthError(f"unable to load repository allowlist: {exc}") from exc + values = payload.get("repositories") if isinstance(payload, dict) else payload + if not isinstance(values, list) or not values: + raise QueueHealthError("repository allowlist must be a non-empty JSON array") + repositories = sorted({_repository_name(value) for value in values}) + if len(repositories) != len(values): + raise QueueHealthError("repository allowlist contains duplicates") + return repositories + + +def _list_payload( + payload: Any, + key: str, + *, + max_items: int = MAX_API_PAGE_SIZE * MAX_API_PAGES, +) -> list[dict[str, Any]]: + """Extract one bounded GitHub list response without accepting under-collection.""" + declared_total_counts: list[Any] = [] + if isinstance(payload, dict) and PAGINATED_PAGES_KEY in payload: + pages = payload[PAGINATED_PAGES_KEY] + if not isinstance(pages, list) or not pages or len(pages) > MAX_API_PAGES: + raise QueueHealthError(f"GitHub response field {key!r} exceeds the bounded page count") + page_values = [] + for page in pages: + if isinstance(page, list): + page_values.extend(page) + elif isinstance(page, dict): + page_items = page.get(key) + if not isinstance(page_items, list): + raise QueueHealthError(f"GitHub response field {key!r} page must contain an array") + page_values.extend(page_items) + if "total_count" in page: + declared_total_counts.append(page["total_count"]) + else: + raise QueueHealthError(f"GitHub response field {key!r} page must be an array or object") + values = page_values + else: + values = payload if isinstance(payload, list) else payload.get(key) if isinstance(payload, dict) else None + if isinstance(payload, dict) and "total_count" in payload: + declared_total_counts.append(payload["total_count"]) + if not isinstance(values, list) or not all(isinstance(value, dict) for value in values): + raise QueueHealthError(f"GitHub response field {key!r} must be an array of objects") + if isinstance(payload, dict) and PAGINATED_PAGES_KEY in payload: + record_identities: list[tuple[str, int]] = [] + for value in values: + record_id = value.get("id") + if not isinstance(record_id, bool) and isinstance(record_id, int) and record_id > 0: + record_identities.append(("id", record_id)) + continue + record_number = value.get("number") + if ( + not isinstance(record_number, bool) + and isinstance(record_number, int) + and record_number > 0 + ): + record_identities.append(("number", record_number)) + continue + else: + raise QueueHealthError( + f"GitHub response field {key!r} paginated records must have a positive integer id or number" + ) + if len(record_identities) != len(set(record_identities)): + raise QueueHealthError( + f"GitHub response field {key!r} contains a duplicate record identity across pages" + ) + if declared_total_counts: + if any(isinstance(total_count, bool) or not isinstance(total_count, int) for total_count in declared_total_counts): + raise QueueHealthError(f"GitHub response field {key!r} has invalid total counts") + total_count = max(declared_total_counts) + if total_count < len(values) or total_count > max_items: + raise QueueHealthError(f"GitHub response field {key!r} exceeds the bounded page size") + if PAGINATED_PAGES_KEY in payload and total_count != len(values): + raise QueueHealthError(f"GitHub response field {key!r} is incompletely paginated") + return values + + +def github_json( + path: str, + *, + paginate: bool = False, + max_pages: int = MAX_API_PAGES, + runner: Runner = subprocess.run, +) -> Any: + """Read one GitHub REST endpoint through ``gh`` without shell evaluation.""" + if not path.startswith("repos/"): + raise QueueHealthError(f"GitHub endpoint is outside repository scope: {path}") + pages: list[Any] = [] + page_size_match = re.search(r"(?:[?&])per_page=(\d+)(?:&|$)", path) + page_size = int(page_size_match.group(1)) if page_size_match else MAX_API_PAGE_SIZE + page_numbers = range(1, max_pages + 1) if paginate else range(1, 2) + for page_number in page_numbers: + page_path = path + if paginate and page_number > 1: + page_path = f"{path}{'&' if '?' in path else '?'}page={page_number}" + try: + result = runner( + ["gh", "api", page_path], + capture_output=True, + text=True, + check=False, + timeout=GITHUB_API_TIMEOUT_SECONDS, + ) + except subprocess.TimeoutExpired as exc: + raise QueueHealthError( + f"GitHub API read timed out after {GITHUB_API_TIMEOUT_SECONDS} seconds for {page_path}" + ) from exc + if result.returncode != 0: + detail = (result.stderr or result.stdout or "GitHub API read failed").strip() + raise QueueHealthError(f"GitHub API read failed for {page_path}: {detail[:400]}") + try: + payload = json.loads(result.stdout) + except json.JSONDecodeError as exc: + raise QueueHealthError(f"GitHub API returned invalid JSON for {page_path}") from exc + if not paginate: + return payload + pages.append(payload) + values = payload if isinstance(payload, list) else None + total_count = payload.get("total_count") if isinstance(payload, dict) else None + if isinstance(payload, dict): + values = next((value for value in payload.values() if isinstance(value, list)), None) + if not isinstance(values, list): + raise QueueHealthError(f"GitHub API page has no bounded array for {page_path}") + collected = sum( + len(page) if isinstance(page, list) else len(next((value for value in page.values() if isinstance(value, list)), [])) + for page in pages + ) + if (type(total_count) is int and total_count <= collected) or len(values) < page_size: + return {PAGINATED_PAGES_KEY: pages} + raise QueueHealthError( + f"GitHub API pagination exceeds {max_pages} pages for {path}" + ) + + +def _normalise_pull_request( + pull_request: dict[str, Any], *, allow_normalized: bool = False +) -> dict[str, Any]: + """Keep only exact-head identity fields needed for queue classification. + + Empty or missing ``head_sha``, ``base_ref``, ``base_repository``, or + ``updated_at`` values are treated as an incomplete identity — the same + as a missing ``head``/``base`` object — so a transient, partially + populated GitHub API response triggers the caller's bounded retry + instead of being silently accepted and later misclassifying an active + run as obsolete. + """ + if not isinstance(pull_request, dict): + raise QueueHealthError("pull request entry must be an object") + number = pull_request.get("number") + if allow_normalized and "head" not in pull_request and "base" not in pull_request: + if not all( + isinstance(pull_request.get(field), str) and pull_request.get(field) + for field in ("base_ref", "base_repository", "head_sha", "updated_at") + ): + raise IncompletePullRequestIdentity( + "normalized pull request identity fields must be non-empty strings" + ) + if isinstance(number, bool) or not isinstance(number, int) or number <= 0: + raise QueueHealthError("pull request number must be a positive integer") + return { + "number": number, + "state": pull_request.get("state", "open"), + "base_ref": pull_request["base_ref"], + "base_repository": pull_request["base_repository"], + "head_sha": pull_request["head_sha"], + "updated_at": pull_request["updated_at"], + } + head = pull_request.get("head") + base = pull_request.get("base") + if not isinstance(head, dict) or not isinstance(base, dict): + raise IncompletePullRequestIdentity("pull request head and base must be objects") + if isinstance(number, bool) or not isinstance(number, int) or number <= 0: + raise QueueHealthError("pull request number must be a positive integer") + head_sha = head.get("sha", "") + base_ref = base.get("ref", "") + base_repository = ( + (base.get("repo") or {}).get("full_name", "") if isinstance(base.get("repo"), dict) else "" + ) + updated_at = pull_request.get("updated_at", "") + if not all( + isinstance(value, str) and value for value in (head_sha, base_ref, base_repository, updated_at) + ): + raise IncompletePullRequestIdentity( + "pull request head, base, and updated_at identity fields must be non-empty" + ) + return { + "number": number, + "state": pull_request.get("state", "open"), + "base_ref": base_ref, + "base_repository": base_repository, + "head_sha": head_sha, + "updated_at": updated_at, + } + + +def _normalise_job(job: dict[str, Any]) -> dict[str, Any]: + """Keep job state and runner assignment evidence without log contents. + + Preserves the job's own ``created_at`` (when GitHub scheduled that + specific job) separately from the parent run's ``created_at``, so a + job that only became eligible after an earlier stage in the same + in-progress run finished is not measured against the whole run's age. + """ + if not isinstance(job, dict): + raise QueueHealthError("workflow job entry must be an object") + job_id = job.get("id") + if isinstance(job_id, bool) or not isinstance(job_id, int) or job_id <= 0: + raise QueueHealthError("job id must be a positive integer") + runner_id = job.get("runner_id") + if isinstance(runner_id, bool) or not isinstance(runner_id, int): + runner_id = 0 + if "steps" in job: + workflow_steps = job["steps"] + if workflow_steps is not None and not isinstance(workflow_steps, list): + raise QueueHealthError("workflow job steps must be an array or null") + steps_count = len(workflow_steps) if isinstance(workflow_steps, list) else None + else: + steps_count = job.get("steps_count") + if steps_count is not None and ( + isinstance(steps_count, bool) + or not isinstance(steps_count, int) + or steps_count < 0 + ): + raise QueueHealthError( + "normalized workflow job steps_count must be a non-negative integer or null" + ) + return { + "id": job_id, + "name": str(job.get("name") or "unnamed job"), + "status": str(job.get("status") or "").upper(), + "conclusion": str(job.get("conclusion") or "").upper(), + "runner_id": runner_id, + "runner_name": str(job.get("runner_name") or ""), + "created_at": str(job.get("created_at") or ""), + "steps_count": steps_count, + } + + +def _normalise_run(repository: str, run: dict[str, Any], jobs: list[dict[str, Any]]) -> dict[str, Any]: + """Keep run identity and job state required for deterministic reporting. + + Accepts a pull-request link either in GitHub's raw shape + (``{"number": ..., "head": {"sha": ...}}``) or in the flattened shape + this function itself emits (``{"number": ..., "head_sha": ...}``), so + re-normalising an already-normalised run loaded back from a collected + snapshot (as ``build_report`` does) does not silently zero out the + linked head SHA that exact-head identity resolution depends on. + """ + if not isinstance(run, dict): + raise QueueHealthError("workflow run entry must be an object") + if not isinstance(jobs, list) or not all(isinstance(job, dict) for job in jobs): + raise QueueHealthError("workflow run jobs must be an array of objects") + run_id = run.get("id") + if isinstance(run_id, bool) or not isinstance(run_id, int) or run_id <= 0: + raise QueueHealthError("workflow run id must be a positive integer") + pull_requests = run.get("pull_requests", []) + if pull_requests is None: + pull_requests = [] + if not isinstance(pull_requests, list) or not all(isinstance(item, dict) for item in pull_requests): + raise QueueHealthError("workflow run pull_requests must be an array of objects") + links = [] + for item in pull_requests: + number = item.get("number") + if isinstance(number, bool) or not isinstance(number, int) or number <= 0: + raise QueueHealthError("workflow run pull request number must be positive") + if "head" not in item and isinstance(item.get("head_sha"), str): + links.append({"number": number, "head_sha": item["head_sha"]}) + continue + head = item.get("head", {}) + if head is None: + head = {} + if not isinstance(head, dict): + raise QueueHealthError("workflow run pull request head must be an object") + links.append({"number": number, "head_sha": str(head.get("sha") or "")}) + return { + "repository": repository, + "id": run_id, + "workflow_name": str(run.get("name") or run.get("workflow_name") or "unnamed workflow"), + "event": str(run.get("event") or "unknown"), + "status": str(run.get("status") or "").upper(), + "conclusion": str(run.get("conclusion") or "").upper(), + "head_sha": str(run.get("head_sha") or ""), + "created_at": str(run.get("created_at") or ""), + "updated_at": str(run.get("updated_at") or ""), + "run_attempt": run.get("run_attempt", 1), + "concurrency_group": str(run.get("concurrency_group") or "unavailable_from_actions_api"), + "pull_requests": sorted(links, key=lambda item: item["number"]), + "jobs": sorted((_normalise_job(job) for job in jobs), key=lambda item: item["id"]), + } + + +def collect_snapshot( + repositories: Sequence[str], + *, + runner: Runner = subprocess.run, + generated_at: str | None = None, +) -> dict[str, Any]: + """Collect bounded queued/in-progress run and job data using read-only API calls.""" + validated = sorted({_repository_name(repository) for repository in repositories}) + if len(validated) != len(repositories): + raise QueueHealthError("collection repository list contains duplicates") + collected_repositories: list[dict[str, Any]] = [] + collection_errors: list[dict[str, str]] = [] + for repository in validated: + try: + metadata = github_json(f"repos/{repository}", runner=runner) + if not isinstance(metadata, dict): + raise QueueHealthError(f"repository metadata for {repository} is not an object") + pulls_endpoint = f"repos/{repository}/pulls?state=open&per_page={MAX_API_PAGE_SIZE}" + pull_requests = _list_payload( + github_json(pulls_endpoint, paginate=True, runner=runner), + "pulls", + max_items=MAX_API_PAGE_SIZE * MAX_API_PAGES, + ) + normalized_pull_requests = sorted( + (_normalise_pull_request(item) for item in pull_requests), + key=lambda item: item["number"], + ) + except IncompletePullRequestIdentity: + time.sleep(PULL_REQUEST_RETRY_DELAY_SECONDS) + try: + retry_pull_requests = _list_payload( + github_json(pulls_endpoint, paginate=True, runner=runner), + "pulls", + max_items=MAX_API_PAGE_SIZE * MAX_API_PAGES, + ) + normalized_pull_requests = sorted( + (_normalise_pull_request(item) for item in retry_pull_requests), + key=lambda item: item["number"], + ) + except QueueHealthError as retry_exc: + collection_errors.append( + { + "repository": repository, + "error": f"pull-request identity validation failed: {retry_exc}", + } + ) + continue + except QueueHealthError as exc: + collection_errors.append({"repository": repository, "error": str(exc)}) + continue + pull_requests_by_number = {item["number"]: item for item in normalized_pull_requests} + runs_by_id: dict[int, dict[str, Any]] = {} + try: + active_statuses = ("in_progress", "pending", "queued", "requested", "waiting") + snapshots: list[dict[int, dict[str, Any]]] = [] + for status_order in (active_statuses, tuple(reversed(active_statuses))): + snapshot: dict[int, dict[str, Any]] = {} + for status in status_order: + runs = _list_payload( + github_json( + f"repos/{repository}/actions/runs?status={status}" + f"&per_page={WORKFLOW_RUN_PAGE_SIZE}", + paginate=True, + max_pages=ACTIVE_RUN_MAX_API_PAGES, + runner=runner, + ), + "workflow_runs", + max_items=WORKFLOW_RUN_PAGE_SIZE * ACTIVE_RUN_MAX_API_PAGES, + ) + for run in runs: + run_id = run.get("id") + if isinstance(run_id, bool) or not isinstance(run_id, int) or run_id <= 0: + raise QueueHealthError("workflow run id must be a positive integer") + snapshot[run_id] = run + snapshots.append(snapshot) + first_snapshot, second_snapshot = snapshots + first_states = { + run_id: str(run.get("status") or "").upper() + for run_id, run in first_snapshot.items() + } + second_states = { + run_id: str(run.get("status") or "").upper() + for run_id, run in second_snapshot.items() + } + if first_states != second_states: + raise QueueHealthError("active workflow run snapshot changed during collection") + for run_id, run in second_snapshot.items(): + run_id = run.get("id") + candidate = _normalise_run(repository, run, []) + identity, _ = _run_identity(candidate, pull_requests_by_number) + if identity != "current_head" or candidate["status"] not in { + "IN_PROGRESS", + "WAITING", + }: + runs_by_id[run_id] = candidate + continue + jobs_payload = github_json( + f"repos/{repository}/actions/runs/{run_id}/jobs?per_page={MAX_API_PAGE_SIZE}", + paginate=True, + runner=runner, + ) + jobs = _list_payload( + jobs_payload, + "jobs", + max_items=MAX_API_PAGE_SIZE * MAX_API_PAGES, + ) + runs_by_id[run_id] = _normalise_run(repository, run, jobs) + except QueueHealthError as exc: + collection_errors.append({"repository": repository, "error": str(exc)}) + continue + collected_repositories.append( + { + "full_name": repository, + "default_branch": str(metadata.get("default_branch") or ""), + "pull_requests": normalized_pull_requests, + "runs": sorted(runs_by_id.values(), key=lambda item: item["id"]), + } + ) + timestamp = generated_at or datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") + parse_timestamp(timestamp) + return { + "generated_at": timestamp, + "repositories": collected_repositories, + "collection_errors": collection_errors, + } + + +def load_snapshot(path: Path) -> dict[str, Any]: + """Load a JSON snapshot for offline, deterministic report generation.""" + try: + payload = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + raise QueueHealthError(f"unable to load queue-health snapshot: {exc}") from exc + if not isinstance(payload, dict): + raise QueueHealthError("queue-health snapshot root must be an object") + return payload + + +def _run_identity(run: dict[str, Any], pull_requests: dict[int, dict[str, Any]]) -> tuple[str, int | None]: + """Resolve one run to current-head, obsolete, or unlinked identity. + + Compares the open pull request's head SHA against the *linked* + pull-request head SHA carried on the run (``run["pull_requests"][*] + ["head_sha"]``), never against the run-level ``head_sha``. For + ``pull_request_target``-triggered runs, GitHub reports the run-level + ``head_sha`` as the base-branch commit that was checked out, not the + pull request's head commit; only the linked pull-request entry carries + the real head SHA that was reviewed. Using the run-level value there + would misclassify a genuinely current, active required-workflow run as + ``obsolete`` and skip fetching its job evidence. + """ + links = run.get("pull_requests") or [] + for link in links: + number = link.get("number") + pull_request = pull_requests.get(number) + if pull_request and pull_request.get("head_sha") == link.get("head_sha"): + return "current_head", number + if links: + return "obsolete", links[0].get("number") + return "unlinked", None + + +def _job_state(job: dict[str, Any]) -> tuple[str, bool, bool]: + """Return normalized execution state, pending flag, and runner assignment. + + GitHub's ``waiting`` job status (a job paused on an environment or + deployment approval) is incomplete pending evidence just like + ``queued``/``in_progress`` — it must remain visible with its own + blocker and action rather than silently dropping out of the pending + count as unclassified ``unknown`` evidence. + """ + status = str(job.get("status") or "").upper() + conclusion = str(job.get("conclusion") or "").upper() + assigned = bool(job.get("runner_name")) or (isinstance(job.get("runner_id"), int) and job.get("runner_id", 0) > 0) + if status == "WAITING": + return "waiting_approval", True, assigned + if status in QUEUE_STATES: + return ("queued_assigned" if assigned else "queued_unassigned"), True, assigned + if status in TERMINAL_STATES or conclusion: + return "terminal", False, assigned + return "unknown", False, assigned + + +def _format_age(created_at: str, now: datetime) -> int: + """Return non-negative queue age seconds from an explicit timestamp.""" + created = parse_timestamp(created_at) + return max(0, int((now - created).total_seconds())) + + +def build_report( + snapshot: dict[str, Any], + *, + now: datetime | None = None, + queue_age_slo_seconds: int = DEFAULT_QUEUE_AGE_SLO_SECONDS, +) -> dict[str, Any]: + """Classify every observed job without treating incomplete evidence as success.""" + if queue_age_slo_seconds < 0: + raise QueueHealthError("queue age SLO must not be negative") + generated_at = parse_timestamp(snapshot.get("generated_at")) + if now is not None and (not isinstance(now, datetime) or now.tzinfo is None): + raise QueueHealthError("evaluation time must include a timezone") + report_now = (now or datetime.now(timezone.utc)).astimezone(timezone.utc) + repositories = snapshot.get("repositories") + if not isinstance(repositories, list): + raise QueueHealthError("queue-health snapshot repositories must be an array") + raw_collection_errors = snapshot.get("collection_errors", []) + if raw_collection_errors is None: + raw_collection_errors = [] + if not isinstance(raw_collection_errors, list): + raise QueueHealthError("queue-health collection_errors must be an array") + collection_errors: list[dict[str, str]] = [] + for item in raw_collection_errors: + if not isinstance(item, dict): + raise QueueHealthError("queue-health collection error must be an object") + repository_name = _repository_name(item.get("repository")) + error = item.get("error") + if not isinstance(error, str) or not error: + raise QueueHealthError("queue-health collection error must contain text") + collection_errors.append({"repository": repository_name, "error": error}) + + rows: list[dict[str, Any]] = [] + seen_repositories: set[str] = set() + for repository in repositories: + if not isinstance(repository, dict): + raise QueueHealthError("queue-health repository entry must be an object") + full_name = _repository_name(repository.get("full_name")) + if full_name in seen_repositories: + raise QueueHealthError(f"duplicate repository entry {full_name}") + seen_repositories.add(full_name) + pull_request_entries = repository.get("pull_requests", []) + if pull_request_entries is None: + pull_request_entries = [] + if not isinstance(pull_request_entries, list): + raise QueueHealthError(f"pull requests for {full_name} must be an array") + pull_requests: dict[int, dict[str, Any]] = {} + for pull_request in pull_request_entries: + normalized = _normalise_pull_request(pull_request, allow_normalized=True) + if normalized["number"] in pull_requests: + raise QueueHealthError(f"duplicate pull request {normalized['number']} for {full_name}") + pull_requests[normalized["number"]] = normalized + runs = repository.get("runs", []) + if runs is None: + runs = [] + if not isinstance(runs, list): + raise QueueHealthError(f"runs for {full_name} must be an array") + run_ids: set[int] = set() + for raw_run in runs: + if not isinstance(raw_run, dict): + raise QueueHealthError("workflow run entry must be an object") + raw_jobs = raw_run.get("jobs", []) + if raw_jobs is None: + raw_jobs = [] + if not isinstance(raw_jobs, list): + raise QueueHealthError("workflow run jobs must be an array") + run = _normalise_run(full_name, raw_run, raw_jobs) + if run["id"] in run_ids: + raise QueueHealthError(f"duplicate workflow run {run['id']} for {full_name}") + run_ids.add(run["id"]) + identity, pull_request_number = _run_identity(run, pull_requests) + jobs = run["jobs"] + for job in jobs or [{"id": run["id"], "name": "run", "status": run.get("status")}]: + state, pending, assigned = _job_state(job) + # Prefer the job's own created_at: for a job with `needs:` + # dependencies inside an already in-progress run, GitHub + # sets it when the job became eligible, which can be long + # after the run itself started. Falling back to the run's + # created_at only applies to the synthetic run-level job + # used when no job evidence was fetched. + age_created_at = job.get("created_at") or run.get("created_at") + age_seconds = _format_age(age_created_at, report_now) + slo_breached = pending and age_seconds > queue_age_slo_seconds + if identity == "obsolete": + blocker = "obsolete_run_requires_identity_confirmed_cleanup" + action = "owner_cleanup_after_exact_identity_confirmation" + elif identity == "unlinked": + blocker = "run_not_linked_to_pull_request" + action = "reconcile_run_identity_before_cleanup" + elif state == "waiting_approval": + blocker = "environment_or_deployment_approval_required" + action = "reviewer_or_owner_approve_pending_environment_deployment" + elif pending and not assigned and slo_breached: + blocker = "external_runner_assignment_or_capacity" + action = "owner_check_runner_billing_policy_and_concurrency" + elif pending: + blocker = "current_head_required_evidence_incomplete" + action = "wait_for_runner_or_escalate_after_slo" + else: + blocker = None + action = "none" + rows.append( + { + "repository": full_name, + "workflow_name": run.get("workflow_name", "unnamed workflow"), + "run_id": run.get("id"), + "run_attempt": run.get("run_attempt", 1), + "job_id": job.get("id"), + "job_name": job.get("name", "unnamed job"), + "event": run.get("event", "unknown"), + "head_sha": run.get("head_sha", ""), + "pull_request_number": pull_request_number, + "identity_state": identity, + "status": job.get("status", ""), + "conclusion": job.get("conclusion", ""), + "execution_state": state, + "is_pending": pending, + "runner_assigned": assigned, + "created_at": run.get("created_at", ""), + "updated_at": run.get("updated_at", ""), + "queue_age_seconds": age_seconds, + "slo_breached": slo_breached, + "concurrency_group": run.get("concurrency_group", "unavailable_from_actions_api"), + "obsolete": identity == "obsolete", + "blocker": blocker, + "recommended_action": action, + } + ) + + rows.sort(key=lambda row: (row["repository"], row["run_id"], row["job_id"])) + pending = [row for row in rows if row["is_pending"]] + current_pending = [row for row in pending if row["identity_state"] == "current_head"] + lane_run_ids: dict[tuple[str, int, str], set[int]] = {} + for row in current_pending: + lane = ( + row["repository"], + row["pull_request_number"], + row["workflow_name"], + ) + lane_run_ids.setdefault(lane, set()).add(row["run_id"]) + duplicate_lanes = [ + { + "repository": key[0], + "pull_request_number": key[1], + "workflow_name": key[2], + "count": len(run_ids), + } + for key, run_ids in sorted(lane_run_ids.items()) + if len(run_ids) > 1 + ] + external_actions = sorted( + { + "Inspect GitHub-hosted runner assignment, Actions billing/usage, runner-group policy, environment approval, and concurrency saturation; queued evidence remains incomplete." + for row in rows + if row["blocker"] == "external_runner_assignment_or_capacity" + } + ) + summary = { + "observed_job_count": len(rows), + "pending_job_count": len(pending), + "current_head_pending_count": len(current_pending), + "unassigned_slo_breached_count": sum( + row["identity_state"] == "current_head" + and row["execution_state"] == "queued_unassigned" + and row["slo_breached"] + for row in rows + ), + "obsolete_job_count": sum(row["obsolete"] for row in rows), + "unlinked_job_count": sum(row["identity_state"] == "unlinked" for row in rows), + "duplicate_pending_lane_count": len(duplicate_lanes), + "terminal_job_count": sum(row["execution_state"] == "terminal" for row in rows), + "collection_error_count": len(collection_errors), + "external_actions": external_actions, + } + return { + "schema_version": SCHEMA_VERSION, + "generated_at": generated_at.isoformat().replace("+00:00", "Z"), + "evaluated_at": report_now.isoformat().replace("+00:00", "Z"), + "queue_age_slo_seconds": queue_age_slo_seconds, + "repositories": sorted(_repository_name(repository["full_name"]) for repository in repositories), + "collection_errors": collection_errors, + "summary": summary, + "duplicate_pending_lanes": duplicate_lanes, + "runs": rows, + "limitations": [ + "The Actions REST API does not expose the evaluated concurrency group for every run; unavailable values are reported explicitly.", + "This read-only slice never cancels runs or changes branch/check state.", + ], + } + + +def render_html(report: dict[str, Any]) -> str: + """Render a keyboard-readable HTML report with escaped untrusted fields.""" + summary = report["summary"] + rows = report["runs"] + table_rows = [] + for row in rows: + table_rows.append( + "" + + f'{html.escape(str(row["repository"]))}' + + "".join( + f"{html.escape(str(row[field]))}" + for field in ( + "workflow_name", + "run_id", + "job_name", + "identity_state", + "execution_state", + "head_sha", + "queue_age_seconds", + "blocker", + ) + ) + + "" + ) + body = "".join(table_rows) or 'No queued or in-progress jobs observed.' + collection_error_section = "" + if report.get("collection_errors"): + collection_error_section = ( + "

" + "Incomplete collection evidence

    " + + "".join( + "
  • " + + html.escape(str(item["repository"])) + + ": " + + html.escape(str(item["error"])) + + "
    • " + for item in report["collection_errors"] + ) + + "
" + ) + return ( + "\n" + '' + "GitHub Actions queue health" + "" + '
' + "

GitHub Actions queue health

" + + collection_error_section + + f"

Evaluated at ; queue-age SLO: {report['queue_age_slo_seconds']} seconds.

" + f"

Observed jobs: {summary['observed_job_count']}; current-head pending: {summary['current_head_pending_count']}; SLO breaches: {summary['unassigned_slo_breached_count']}.

" + '' + "" + + "".join(f"" for field in ( + "repository", "workflow_name", "run_id", "job_name", "identity_state", "execution_state", "head_sha", "queue_age_seconds", "blocker" + )) + + f"{body}
Run and job evidence; queued evidence is not a passing check.
{field.replace('_', ' ').title()}
\n" + ) + + +def write_reports(report: dict[str, Any], json_path: Path, html_path: Path) -> None: + """Write deterministic JSON and accessible HTML reports.""" + json_path.parent.mkdir(parents=True, exist_ok=True) + html_path.parent.mkdir(parents=True, exist_ok=True) + json_path.write_text( + json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True) + "\n", + encoding="utf-8", + ) + html_path.write_text(render_html(report), encoding="utf-8") + + +def parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace: + """Parse live-collection or offline-report CLI arguments.""" + parser = argparse.ArgumentParser(description=__doc__) + source = parser.add_mutually_exclusive_group(required=True) + source.add_argument("--snapshot", type=Path) + source.add_argument("--allowlist", type=Path) + parser.add_argument("--output-json", type=Path, required=True) + parser.add_argument("--output-html", type=Path, required=True) + parser.add_argument("--queue-age-slo-seconds", type=int, default=DEFAULT_QUEUE_AGE_SLO_SECONDS) + parser.add_argument("--now", help="Explicit timezone-aware evaluation time for deterministic reports") + return parser.parse_args(argv) + + +def main(argv: Sequence[str] | None = None, *, stderr: TextIO = sys.stderr) -> int: + """Collect or load a snapshot, write reports, and return a stable CLI status.""" + args = parse_args(argv) + try: + snapshot = load_snapshot(args.snapshot) if args.snapshot else collect_snapshot(load_allowlist(args.allowlist)) + now = parse_timestamp(args.now) if args.now else datetime.now(timezone.utc) + report = build_report( + snapshot, + now=now, + queue_age_slo_seconds=args.queue_age_slo_seconds, + ) + write_reports(report, args.output_json, args.output_html) + except (OSError, QueueHealthError, ValueError) as exc: + print(f"ERROR: queue-health report failed: {exc}", file=stderr) + return 2 + breaches = report["summary"]["unassigned_slo_breached_count"] + if breaches: + print(f"::warning::Actions queue-health found {breaches} unassigned current-head SLO breach(es).") + print( + "QUEUE_HEALTH_RESULT=" + f"observed={report['summary']['observed_job_count']} " + f"pending={report['summary']['pending_job_count']} " + f"slo_breaches={breaches}" + ) + return 0 + + +if __name__ == "__main__": # pragma: no cover - exercised through the CLI tests. + raise SystemExit(main()) diff --git a/tests/test_actions_queue_health.py b/tests/test_actions_queue_health.py new file mode 100644 index 0000000000..d32346fc62 --- /dev/null +++ b/tests/test_actions_queue_health.py @@ -0,0 +1,1220 @@ +"""Contract and behavior tests for the read-only Actions queue collector.""" + +import importlib.util +from datetime import datetime, timezone +import io +import json +from pathlib import Path +from subprocess import CompletedProcess, TimeoutExpired + +import pytest + + +ROOT = Path(__file__).resolve().parents[1] +MODULE_PATH = ROOT / "scripts/ci/actions_queue_health.py" +SPEC = importlib.util.spec_from_file_location("actions_queue_health", MODULE_PATH) +assert SPEC and SPEC.loader +queue_health = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(queue_health) + + +NOW = datetime(2026, 8, 19, 12, 0, tzinfo=timezone.utc) + + +def test_queue_health_module_path_is_independent_of_working_directory() -> None: + """Load production code from the repository root, not the caller's cwd.""" + expected = Path(__file__).resolve().parents[1] / "scripts/ci/actions_queue_health.py" + assert MODULE_PATH == expected + assert MODULE_PATH.is_file() + + +def pull_request(number: int = 1, head_sha: str = "head") -> dict: + """Return a compact open pull-request fixture.""" + return { + "number": number, + "state": "open", + "base": {"ref": "main", "repo": {"full_name": "owner/repo"}}, + "head": {"sha": head_sha}, + "updated_at": "2026-08-19T11:00:00Z", + } + + +def workflow_run( + run_id: int, + *, + head_sha: str = "head", + pull_requests: list[dict] | None = None, + status: str = "queued", + jobs: list[dict] | None = None, + workflow_name: str = "required-check", + created_at: str = "2026-08-19T10:00:00Z", +) -> dict: + """Return one raw workflow-run fixture.""" + return { + "id": run_id, + "name": workflow_name, + "event": "pull_request", + "status": status, + "conclusion": "", + "head_sha": head_sha, + "created_at": created_at, + "updated_at": created_at, + "run_attempt": 1, + "pull_requests": pull_requests or [], + "jobs": jobs or [], + } + + +def job( + job_id: int, + *, + status: str = "queued", + conclusion: str | None = None, + runner_id: int | None = None, + runner_name: str | None = None, + name: str = "required-check", +) -> dict: + """Return one raw workflow-job fixture.""" + return { + "id": job_id, + "name": name, + "status": status, + "conclusion": conclusion, + "runner_id": runner_id, + "runner_name": runner_name, + "steps": [], + } + + +def report_snapshot() -> dict: + """Return a fixture covering current, obsolete, unlinked, and terminal jobs.""" + return { + "generated_at": "2026-08-19T11:00:00Z", + "repositories": [ + { + "full_name": "owner/repo", + "pull_requests": [pull_request()], + "runs": [ + workflow_run( + 10, + pull_requests=[{"number": 1, "head": {"sha": "head"}}], + jobs=[ + job(100), + job(101, runner_id=7, runner_name="runner-7"), + job(102, status="waiting"), + ], + ), + workflow_run( + 11, + head_sha="old", + pull_requests=[{"number": 1, "head": {"sha": "old"}}], + jobs=[job(110)], + ), + workflow_run(12, jobs=[job(120)]), + workflow_run( + 13, + pull_requests=[{"number": 1, "head": {"sha": "head"}}], + jobs=[], + workflow_name="required-check", + ), + workflow_run( + 14, + pull_requests=[{"number": 1, "head": {"sha": "head"}}], + status="completed", + jobs=[job(140, status="completed", conclusion="success")], + ), + ], + } + ], + } + + +@pytest.mark.parametrize("value", [None, "", " ", "not-a-time", "2026-08-19T12:00:00"]) +def test_parse_timestamp_rejects_ambiguous_or_invalid_values(value: object) -> None: + """Reject missing, malformed, and timezone-free timestamps.""" + with pytest.raises(queue_health.QueueHealthError): + queue_health.parse_timestamp(value) # type: ignore[arg-type] + + +def test_parse_timestamp_normalises_z_and_offsets() -> None: + """Normalize UTC and offset timestamps to the same instant.""" + assert queue_health.parse_timestamp("2026-08-19T12:00:00Z") == NOW + assert queue_health.parse_timestamp("2026-08-19T21:00:00+09:00") == NOW + + +@pytest.mark.parametrize("value", ["owner", "owner/repo/extra", "../..", "./repo", "owner/.", 1]) +def test_repository_name_rejects_non_repository_identifiers(value: object) -> None: + """Reject malformed and traversal-like repository identifiers.""" + with pytest.raises(queue_health.QueueHealthError): + queue_health._repository_name(value) + + +def test_load_allowlist_accepts_array_and_object_and_rejects_bad_inputs(tmp_path: Path) -> None: + """Load both supported allowlist shapes and reject unsafe input files.""" + array_path = tmp_path / "array.json" + array_path.write_text(json.dumps(["z/repo", "a/repo"]), encoding="utf-8") + assert queue_health.load_allowlist(array_path) == ["a/repo", "z/repo"] + + object_path = tmp_path / "object.json" + object_path.write_text(json.dumps({"repositories": ["a/repo"]}), encoding="utf-8") + assert queue_health.load_allowlist(object_path) == ["a/repo"] + + for name, payload in ( + ("empty.json", []), + ("missing-key.json", {}), + ("duplicate.json", ["a/repo", "a/repo"]), + ("invalid-repository.json", ["a repo"]), + ): + path = tmp_path / name + path.write_text(json.dumps(payload), encoding="utf-8") + with pytest.raises(queue_health.QueueHealthError): + queue_health.load_allowlist(path) + + (tmp_path / "invalid.json").write_text("{", encoding="utf-8") + with pytest.raises(queue_health.QueueHealthError): + queue_health.load_allowlist(tmp_path / "invalid.json") + with pytest.raises(queue_health.QueueHealthError): + queue_health.load_allowlist(tmp_path / "missing.json") + + +@pytest.mark.parametrize( + "payload, key, expected", + [ + ([{"id": 1}], "items", [{"id": 1}]), + ({"items": [{"id": 2}]}, "items", [{"id": 2}]), + ({"items": [{"id": 3}], "total_count": 1}, "items", [{"id": 3}]), + ], +) +def test_list_payload_accepts_api_list_shapes(payload: object, key: str, expected: list[dict]) -> None: + """Accept the bounded list response shapes emitted by GitHub APIs.""" + assert queue_health._list_payload(payload, key) == expected + + +@pytest.mark.parametrize( + "payload", + [ + None, + {"items": "bad"}, + [{"id": 1}, "bad"], + {"items": [{"id": 1}], "total_count": 2001}, + {"items": [{"id": 1}], "total_count": 0}, + {"items": [{"id": 1}], "total_count": True}, + {"items": [{"id": 1}], "total_count": "1"}, + {"items": [{"id": 1}], "total_count": []}, + ], +) +def test_list_payload_rejects_untrusted_shapes(payload: object) -> None: + """Reject malformed, oversized, and dishonest list responses.""" + with pytest.raises(queue_health.QueueHealthError): + queue_health._list_payload(payload, "items") + + +def test_list_payload_flattens_bounded_paginated_responses() -> None: + """Flatten bounded pages while validating every page and total count.""" + assert queue_health._list_payload( + {"_queue_health_pages": [[{"id": 1}], [{"id": 2}]]}, "items" + ) == [{"id": 1}, {"id": 2}] + assert queue_health._list_payload( + {"_queue_health_pages": [{"items": [{"id": 3}], "total_count": 2}, {"items": [{"id": 4}], "total_count": 2}]}, + "items", + ) == [{"id": 3}, {"id": 4}] + for payload in ( + {"_queue_health_pages": []}, + {"_queue_health_pages": [[]] * (queue_health.MAX_API_PAGES + 1)}, + {"_queue_health_pages": [None]}, + {"_queue_health_pages": [{"items": "bad"}]}, + {"_queue_health_pages": [{"items": [{"id": 1}], "total_count": 3}, {"items": [{"id": 2}], "total_count": "2"}]}, + ): + with pytest.raises(queue_health.QueueHealthError): + queue_health._list_payload(payload, "items") + assert queue_health._list_payload( + {"_queue_health_pages": [{"items": [{"id": 1}], "total_count": 1}, {"items": [{"id": 2}], "total_count": 2}]}, + "items", + ) == [{"id": 1}, {"id": 2}] + + +def test_list_payload_rejects_incompletely_paginated_total_count() -> None: + """A declared total larger than collected pages cannot look healthy.""" + with pytest.raises(queue_health.QueueHealthError, match="incompletely paginated"): + queue_health._list_payload( + {"_queue_health_pages": [{"items": [{"id": 1}], "total_count": 2}]}, + "items", + ) + + +def test_list_payload_rejects_duplicate_identities_across_pages() -> None: + """Moving records between pages cannot conceal omitted queue evidence.""" + with pytest.raises(queue_health.QueueHealthError, match="duplicate record identity"): + queue_health._list_payload( + { + "_queue_health_pages": [ + {"items": [{"id": 1}], "total_count": 2}, + {"items": [{"id": 1}], "total_count": 2}, + ] + }, + "items", + ) + with pytest.raises(queue_health.QueueHealthError, match="positive integer id or number"): + queue_health._list_payload( + {"_queue_health_pages": [{"items": [{}], "total_count": 1}]}, + "items", + ) + + +def test_github_json_is_read_only_and_rejects_failures() -> None: + """Use safe read-only CLI arguments and fail closed on transport errors.""" + def success_runner(*args: object, **kwargs: object) -> CompletedProcess[str]: + """Return one successful non-paginated API response.""" + assert args[0] == ["gh", "api", "repos/a/repo"] + assert kwargs == { + "capture_output": True, + "text": True, + "check": False, + "timeout": 30, + } + return CompletedProcess([], 0, "[{\"id\": 1}]", "") + + assert queue_health.github_json("repos/a/repo", runner=success_runner) == [{"id": 1}] + + def paginated_runner(*args: object, **kwargs: object) -> CompletedProcess[str]: + """Return one successful paginated API response.""" + assert args[0] == ["gh", "api", "repos/a/repo"] + return CompletedProcess([], 0, "[{\"id\": 1}]", "") + + assert queue_health.github_json("repos/a/repo", paginate=True, runner=paginated_runner) == { + "_queue_health_pages": [[{"id": 1}]] + } + with pytest.raises(queue_health.QueueHealthError, match="no bounded array"): + queue_health.github_json( + "repos/a/repo", + paginate=True, + runner=lambda *args, **kwargs: CompletedProcess([], 0, '{}', ''), + ) + requested_pages: list[str] = [] + + def full_page_runner(args: list[str], **kwargs: object) -> CompletedProcess[str]: + requested_pages.append(args[-1]) + return CompletedProcess(args, 0, json.dumps([{}] * 100), "") + + with pytest.raises(queue_health.QueueHealthError, match="exceeds 20 pages"): + queue_health.github_json("repos/a/repo?per_page=100", paginate=True, runner=full_page_runner) + assert requested_pages[-1].endswith("page=20") + assert not any(path.endswith("page=21") for path in requested_pages) + with pytest.raises(queue_health.QueueHealthError): + queue_health.github_json("orgs/a/repos", runner=success_runner) + + def failed_runner(*args: object, **kwargs: object) -> CompletedProcess[str]: + """Return a failed API response with stderr evidence.""" + return CompletedProcess([], 1, "fallback", "api failed") + + with pytest.raises(queue_health.QueueHealthError, match="api failed"): + queue_health.github_json("repos/a/repo", runner=failed_runner) + + def stdout_failure_runner(*args: object, **kwargs: object) -> CompletedProcess[str]: + """Return a failed API response with stdout-only evidence.""" + return CompletedProcess([], 1, "stdout failure", "") + + with pytest.raises(queue_health.QueueHealthError, match="stdout failure"): + queue_health.github_json("repos/a/repo", runner=stdout_failure_runner) + + def empty_failure_runner(*args: object, **kwargs: object) -> CompletedProcess[str]: + """Return a failed API response without diagnostic text.""" + return CompletedProcess([], 1, "", "") + + with pytest.raises(queue_health.QueueHealthError, match="GitHub API read failed"): + queue_health.github_json("repos/a/repo", runner=empty_failure_runner) + + def invalid_json_runner(*args: object, **kwargs: object) -> CompletedProcess[str]: + """Return a successful process containing invalid JSON.""" + return CompletedProcess([], 0, "not json", "") + + with pytest.raises(queue_health.QueueHealthError, match="invalid JSON"): + queue_health.github_json("repos/a/repo", runner=invalid_json_runner) + + +def test_github_json_fails_closed_when_external_read_times_out() -> None: + """A stalled GitHub CLI read must not occupy the workflow indefinitely.""" + def timeout_runner(*args: object, **kwargs: object) -> CompletedProcess[str]: + """Raise the subprocess timeout seen by the production boundary.""" + raise TimeoutExpired(args[0], timeout=30) + + with pytest.raises(queue_health.QueueHealthError, match="timed out after 30 seconds"): + queue_health.github_json("repos/a/repo", runner=timeout_runner) + + +def test_normalise_pull_request_preserves_exact_head_identity() -> None: + """Retain exact pull-request identity and reject incomplete records.""" + normalized = queue_health._normalise_pull_request(pull_request()) + assert normalized["number"] == 1 + assert normalized["head_sha"] == "head" + assert normalized["base_repository"] == "owner/repo" + normalized_snapshot = queue_health._normalise_pull_request( + { + "number": 1, + "state": "open", + "base_ref": "main", + "base_repository": "owner/repo", + "head_sha": "head", + "updated_at": "2026-08-19T11:00:00Z", + }, + allow_normalized=True, + ) + assert normalized_snapshot == { + "number": 1, + "state": "open", + "base_ref": "main", + "base_repository": "owner/repo", + "head_sha": "head", + "updated_at": "2026-08-19T11:00:00Z", + } + with pytest.raises(queue_health.IncompletePullRequestIdentity, match="normalized"): + queue_health._normalise_pull_request( + {"number": 1, "base_ref": "main"}, allow_normalized=True + ) + with pytest.raises(queue_health.QueueHealthError, match="positive integer"): + queue_health._normalise_pull_request( + { + "number": 0, + "base_ref": "main", + "base_repository": "owner/repo", + "head_sha": "head", + "updated_at": "2026-08-19T11:00:00Z", + }, + allow_normalized=True, + ) + for invalid in ({"number": True}, {"number": 0}, {"number": "1"}, "bad"): + with pytest.raises(queue_health.QueueHealthError): + queue_health._normalise_pull_request(invalid) # type: ignore[arg-type] + with pytest.raises(queue_health.IncompletePullRequestIdentity, match="head and base"): + queue_health._normalise_pull_request({"number": 1, "head": {}, "base": "bad"}) + with pytest.raises(queue_health.QueueHealthError, match="positive integer"): + queue_health._normalise_pull_request({"number": 0, "head": {}, "base": {}}) + + +def test_normalise_pull_request_rejects_empty_identity_fields_instead_of_retrying_later() -> None: + """An empty API-provided identity field must retry, not silently pass through.""" + empty_head_sha = pull_request() + empty_head_sha["head"] = {"sha": ""} + with pytest.raises(queue_health.IncompletePullRequestIdentity, match="non-empty"): + queue_health._normalise_pull_request(empty_head_sha) + + empty_base_ref = pull_request() + empty_base_ref["base"]["ref"] = "" + with pytest.raises(queue_health.IncompletePullRequestIdentity, match="non-empty"): + queue_health._normalise_pull_request(empty_base_ref) + + empty_base_repository = pull_request() + empty_base_repository["base"]["repo"]["full_name"] = "" + with pytest.raises(queue_health.IncompletePullRequestIdentity, match="non-empty"): + queue_health._normalise_pull_request(empty_base_repository) + + missing_base_repo = pull_request() + del missing_base_repo["base"]["repo"] + with pytest.raises(queue_health.IncompletePullRequestIdentity, match="non-empty"): + queue_health._normalise_pull_request(missing_base_repo) + + empty_updated_at = pull_request() + empty_updated_at["updated_at"] = "" + with pytest.raises(queue_health.IncompletePullRequestIdentity, match="non-empty"): + queue_health._normalise_pull_request(empty_updated_at) + + empty_normalized = { + "number": 1, + "base_ref": "main", + "base_repository": "owner/repo", + "head_sha": "", + "updated_at": "2026-08-19T11:00:00Z", + } + with pytest.raises(queue_health.IncompletePullRequestIdentity, match="non-empty"): + queue_health._normalise_pull_request(empty_normalized, allow_normalized=True) + + +def test_normalise_job_preserves_runner_assignment_and_fails_closed() -> None: + """Retain runner evidence while rejecting malformed job identities.""" + normalized = queue_health._normalise_job(job(1, runner_id=3, runner_name="runner")) + assert normalized["runner_id"] == 3 + assert normalized["steps_count"] == 0 + assert queue_health._normalise_job( + {"id": 2, "status": "queued", "runner_id": "bad", "steps": []} + )["runner_id"] == 0 + assert queue_health._normalise_job({"id": 3, "runner_id": True})[ + "steps_count" + ] is None + assert queue_health._normalise_job({"id": 4, "steps": None})[ + "steps_count" + ] is None + assert queue_health._normalise_job({"id": 5, "steps_count": 2})[ + "steps_count" + ] == 2 + with pytest.raises(queue_health.QueueHealthError, match="steps must be an array"): + queue_health._normalise_job({"id": 6, "steps": "bad"}) + for invalid_steps_count in (True, -1, "2"): + with pytest.raises(queue_health.QueueHealthError, match="steps_count"): + queue_health._normalise_job( + {"id": 7, "steps_count": invalid_steps_count} + ) + for invalid in ({"id": True}, {"id": 0}, {"id": "1"}, "bad"): + with pytest.raises(queue_health.QueueHealthError): + queue_health._normalise_job(invalid) # type: ignore[arg-type] + + +def test_normalise_run_validates_links_jobs_and_fallback_names() -> None: + """Normalize workflow links and jobs with bounded fallback values.""" + normalized = queue_health._normalise_run( + "owner/repo", + { + "id": 1, + "workflow_name": "fallback-name", + "pull_requests": [{"number": 2, "head": {"sha": "sha"}}], + }, + [job(2)], + ) + assert normalized["workflow_name"] == "fallback-name" + assert normalized["pull_requests"] == [{"number": 2, "head_sha": "sha"}] + assert queue_health._normalise_run("owner/repo", {"id": 2, "pull_requests": None}, [])["pull_requests"] == [] + assert queue_health._normalise_run( + "owner/repo", {"id": 3, "pull_requests": [{"number": 1, "head": None}]}, [] + )["pull_requests"] == [{"number": 1, "head_sha": ""}] + # Re-normalising an already-normalised link (as build_report does when + # loading a snapshot collect_snapshot produced) must preserve head_sha + # rather than treating the flattened shape as missing "head". + assert queue_health._normalise_run( + "owner/repo", {"id": 4, "pull_requests": [{"number": 5, "head_sha": "flat-sha"}]}, [] + )["pull_requests"] == [{"number": 5, "head_sha": "flat-sha"}] + + for invalid_run, invalid_jobs in ( + ("bad", []), + ({"id": True}, []), + ({"id": 0}, []), + ({"id": 1}, "bad"), + ({"id": 1, "pull_requests": "bad"}, []), + ({"id": 1, "pull_requests": [{"number": 0}]}, []), + ({"id": 1, "pull_requests": [{"number": 1, "head": "bad"}]}, []), + ({"id": 1}, ["bad"]), + ): + with pytest.raises(queue_health.QueueHealthError): + queue_health._normalise_run("owner/repo", invalid_run, invalid_jobs) # type: ignore[arg-type] + + +def test_collect_snapshot_deduplicates_status_views_and_preserves_order( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Deduplicate status views and isolate malformed repository evidence.""" + queued_current = workflow_run(10, pull_requests=[{"number": 1, "head": {"sha": "head"}}]) + current = workflow_run( + 12, + status="in_progress", + pull_requests=[{"number": 1, "head": {"sha": "head"}}], + jobs=[job(100)], + ) + unlinked = workflow_run(11, jobs=[]) + responses = { + "repos/owner/repo": {"default_branch": "main"}, + "repos/owner/repo/pulls?state=open&per_page=100": [pull_request()], + "repos/owner/repo/actions/runs?per_page=50": [queued_current, current, unlinked], + "repos/owner/repo/actions/runs/12/jobs?per_page=100": {"jobs": [job(100)]}, + } + for status in ("in_progress", "pending", "queued", "requested", "waiting"): + responses[f"repos/owner/repo/actions/runs?status={status}&per_page=50"] = [ + run + for run in responses["repos/owner/repo/actions/runs?per_page=50"] + if run["status"] == status + ] + responses["repos/owner/repo/actions/runs?status=completed&head_sha=head&per_page=50"] = [] + responses["repos/owner/repo/actions/runs?status=cancelled&event=pull_request_target&per_page=50"] = [] + + def runner(args: list[str], **kwargs: object) -> CompletedProcess[str]: + """Return the deterministic API response for each requested endpoint.""" + payload = responses[args[-1]] + if "--paginate" in args: + payload = [payload] + return CompletedProcess(args, 0, json.dumps(payload), "") + + snapshot = queue_health.collect_snapshot(["owner/repo"], runner=runner, generated_at="2026-08-19T11:00:00Z") + assert snapshot["repositories"][0]["runs"][0]["id"] == 10 + assert [run["id"] for run in snapshot["repositories"][0]["runs"]] == [10, 11, 12] + assert snapshot["repositories"][0]["default_branch"] == "main" + report = queue_health.build_report(snapshot, now=NOW) + assert report["summary"]["observed_job_count"] == 3 + assert report["summary"]["current_head_pending_count"] == 2 + + with pytest.raises(queue_health.QueueHealthError): + queue_health.collect_snapshot(["owner/repo", "owner/repo"], runner=runner) + with pytest.raises(queue_health.QueueHealthError): + queue_health.collect_snapshot(["owner/repo"], runner=runner, generated_at="bad") + + bad_responses = dict(responses) + bad_responses["repos/owner/repo"] = [] + + def bad_metadata_runner(args: list[str], **kwargs: object) -> CompletedProcess[str]: + """Return malformed repository metadata for the isolation case.""" + payload = bad_responses[args[-1]] + if "--paginate" in args: + payload = [payload] + return CompletedProcess(args, 0, json.dumps(payload), "") + + bad_snapshot = queue_health.collect_snapshot(["owner/repo"], runner=bad_metadata_runner) + assert bad_snapshot["repositories"] == [] + assert bad_snapshot["collection_errors"][0]["repository"] == "owner/repo" + + invalid_run_responses = dict(responses) + invalid_run_responses["repos/owner/repo/actions/runs?status=queued&per_page=50"] = [ + {"id": 0, "status": "queued"} + ] + + def invalid_run_runner(args: list[str], **kwargs: object) -> CompletedProcess[str]: + """Return an invalid workflow-run identity for the isolation case.""" + payload = invalid_run_responses[args[-1]] + if "--paginate" in args: + payload = [payload] + return CompletedProcess(args, 0, json.dumps(payload), "") + + invalid_run_snapshot = queue_health.collect_snapshot(["owner/repo"], runner=invalid_run_runner) + assert invalid_run_snapshot["repositories"] == [] + assert invalid_run_snapshot["collection_errors"][0]["repository"] == "owner/repo" + + bad_pull = pull_request() + bad_pull["base"] = "temporarily incomplete" + retry_calls = 0 + requested_paths: list[str] = [] + sleep_calls: list[float] = [] + monkeypatch.setattr(queue_health.time, "sleep", sleep_calls.append) + + def retry_runner(args: list[str], **kwargs: object) -> CompletedProcess[str]: + """Return one incomplete pull response followed by a valid response.""" + nonlocal retry_calls + requested_paths.append(args[-1]) + payload = responses[args[-1]] + if args[-1] == "repos/owner/repo/pulls?state=open&per_page=100": + retry_calls += 1 + payload = [bad_pull] if retry_calls == 1 else payload + if "--paginate" in args: + payload = [payload] + return CompletedProcess(args, 0, json.dumps(payload), "") + + queue_health.collect_snapshot(["owner/repo"], runner=retry_runner) + assert retry_calls == 4 + pull_snapshot_indices = [ + request_index + for request_index, request_path in enumerate(requested_paths) + if request_path == "repos/owner/repo/pulls?state=open&per_page=100" + ] + first_run_request_index = next( + request_index + for request_index, request_path in enumerate(requested_paths) + if "/actions/runs?" in request_path + ) + assert pull_snapshot_indices[1] < first_run_request_index + assert sleep_calls == [queue_health.PULL_REQUEST_RETRY_DELAY_SECONDS] + + def persistent_bad_runner(args: list[str], **kwargs: object) -> CompletedProcess[str]: + """Return an incomplete pull response on every retry.""" + payload = ( + [bad_pull] + if args[-1] == "repos/owner/repo/pulls?state=open&per_page=100" + else responses[args[-1]] + ) + if "--paginate" in args: + payload = [payload] + return CompletedProcess(args, 0, json.dumps(payload), "") + + persistent_bad_snapshot = queue_health.collect_snapshot( + ["owner/repo"], runner=persistent_bad_runner + ) + assert persistent_bad_snapshot["repositories"] == [] + assert persistent_bad_snapshot["collection_errors"][0]["repository"] == "owner/repo" + + bad_number = pull_request(number=0) + + def invalid_pull_runner(args: list[str], **kwargs: object) -> CompletedProcess[str]: + """Return a pull request with an invalid number.""" + payload = ( + [bad_number] + if args[-1] == "repos/owner/repo/pulls?state=open&per_page=100" + else responses[args[-1]] + ) + if "--paginate" in args: + payload = [payload] + return CompletedProcess(args, 0, json.dumps(payload), "") + + invalid_pull_snapshot = queue_health.collect_snapshot( + ["owner/repo"], runner=invalid_pull_runner + ) + assert invalid_pull_snapshot["repositories"] == [] + assert invalid_pull_snapshot["collection_errors"][0]["repository"] == "owner/repo" + + +def test_collect_snapshot_retries_pull_request_with_empty_identity_fields( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """An empty head_sha in one API response must retry like a missing head/base.""" + queued_current = workflow_run(20, pull_requests=[{"number": 1, "head": {"sha": "head"}}]) + responses = { + "repos/owner/repo": {"default_branch": "main"}, + "repos/owner/repo/pulls?state=open&per_page=100": [pull_request()], + "repos/owner/repo/actions/runs?per_page=50": [queued_current], + } + for status in ("in_progress", "pending", "queued", "requested", "waiting"): + responses[f"repos/owner/repo/actions/runs?status={status}&per_page=50"] = ( + [queued_current] if status == "queued" else [] + ) + responses["repos/owner/repo/actions/runs?status=completed&head_sha=head&per_page=50"] = [] + responses["repos/owner/repo/actions/runs?status=cancelled&event=pull_request_target&per_page=50"] = [] + empty_identity_pull = pull_request() + empty_identity_pull["head"] = {"sha": ""} + retry_calls = 0 + sleep_calls: list[float] = [] + monkeypatch.setattr(queue_health.time, "sleep", sleep_calls.append) + + def runner(args: list[str], **kwargs: object) -> CompletedProcess[str]: + """Return one empty-identity pull response followed by a complete one.""" + nonlocal retry_calls + payload = responses[args[-1]] + if args[-1] == "repos/owner/repo/pulls?state=open&per_page=100": + retry_calls += 1 + payload = [empty_identity_pull] if retry_calls == 1 else payload + if "--paginate" in args: + payload = [payload] + return CompletedProcess(args, 0, json.dumps(payload), "") + + snapshot = queue_health.collect_snapshot(["owner/repo"], runner=runner) + assert retry_calls == 4 + assert sleep_calls == [queue_health.PULL_REQUEST_RETRY_DELAY_SECONDS] + assert snapshot["collection_errors"] == [] + assert snapshot["repositories"][0]["pull_requests"][0]["head_sha"] == "head" + + def persistent_empty_runner(args: list[str], **kwargs: object) -> CompletedProcess[str]: + """Return an empty-identity pull response on every attempt.""" + payload = ( + [empty_identity_pull] + if args[-1] == "repos/owner/repo/pulls?state=open&per_page=100" + else responses[args[-1]] + ) + if "--paginate" in args: + payload = [payload] + return CompletedProcess(args, 0, json.dumps(payload), "") + + persistent_snapshot = queue_health.collect_snapshot(["owner/repo"], runner=persistent_empty_runner) + assert persistent_snapshot["repositories"] == [] + assert persistent_snapshot["collection_errors"][0]["repository"] == "owner/repo" + + +def test_collect_snapshot_and_build_report_preserve_linked_head_through_round_trip() -> None: + """The full collect -> build pipeline must not lose the linked head SHA. + + ``build_report`` re-normalises runs loaded from a collected snapshot; + this exercises the entire ``collect_snapshot`` -> ``build_report`` path + for a ``pull_request_target``-shaped run (run-level head_sha is the base + commit, the linked pull-request entry carries the real PR head) and + checks the run still resolves to ``current_head``. + """ + pull_request_target_run = workflow_run( + 70, + head_sha="base-branch-checkout-sha", + status="in_progress", + pull_requests=[{"number": 1, "head": {"sha": "pr-head-sha"}}], + jobs=[job(700, runner_id=9, runner_name="runner-9")], + ) + responses = { + "repos/owner/repo": {"default_branch": "main"}, + "repos/owner/repo/pulls?state=open&per_page=100": [pull_request(1, "pr-head-sha")], + "repos/owner/repo/actions/runs?per_page=50": [pull_request_target_run], + "repos/owner/repo/actions/runs/70/jobs?per_page=100": { + "jobs": [job(700, runner_id=9, runner_name="runner-9")] + }, + } + for status in ("in_progress", "pending", "queued", "requested", "waiting"): + responses[f"repos/owner/repo/actions/runs?status={status}&per_page=50"] = ( + [pull_request_target_run] if status == "in_progress" else [] + ) + responses["repos/owner/repo/actions/runs?status=completed&head_sha=pr-head-sha&per_page=50"] = [] + responses["repos/owner/repo/actions/runs?status=cancelled&event=pull_request_target&per_page=50"] = [] + + def runner(args: list[str], **kwargs: object) -> CompletedProcess[str]: + """Return the deterministic API response for each requested endpoint.""" + payload = responses[args[-1]] + if "--paginate" in args: + payload = [payload] + return CompletedProcess(args, 0, json.dumps(payload), "") + + snapshot = queue_health.collect_snapshot(["owner/repo"], runner=runner, generated_at="2026-08-19T11:00:00Z") + report = queue_health.build_report(snapshot, now=NOW) + row = report["runs"][0] + assert row["identity_state"] == "current_head" + assert row["obsolete"] is False + + +def test_collect_snapshot_isolates_repository_errors_and_reports_incomplete_evidence() -> None: + """Continue healthy collection while recording one repository's failure.""" + def runner(args: list[str], **kwargs: object) -> CompletedProcess[str]: + """Return a rate-limit failure for one repository and valid data for another.""" + path = args[-1] + if path == "repos/bad/repo": + return CompletedProcess(args, 1, "", "rate limit") + if path == "repos/good/repo": + payload: object = {"default_branch": "main"} + elif path == "repos/good/repo/pulls?state=open&per_page=100": + payload = [] + elif "/actions/runs?status=" in path: + payload = [] + else: # pragma: no cover - a new endpoint must be explicitly governed + raise AssertionError(f"unexpected endpoint: {path}") + if "--paginate" in args: + payload = [payload] + return CompletedProcess(args, 0, json.dumps(payload), "") + + snapshot = queue_health.collect_snapshot( + ["bad/repo", "good/repo"], runner=runner, generated_at="2026-08-19T11:00:00Z" + ) + assert [item["full_name"] for item in snapshot["repositories"]] == ["good/repo"] + assert snapshot["collection_errors"] == [ + {"repository": "bad/repo", "error": "GitHub API read failed for repos/bad/repo: rate limit"} + ] + + report = queue_health.build_report(snapshot, now=NOW) + assert report["summary"]["collection_error_count"] == 1 + assert report["collection_errors"] == snapshot["collection_errors"] + assert "bad/repo" in queue_health.render_html(report) + + +@pytest.mark.parametrize( + "collection_errors", + [ + "bad", + [None], + [{"repository": "../..", "error": "bad"}], + [{"repository": "owner/repo", "error": 1}], + ], +) +def test_build_report_rejects_malformed_collection_errors(collection_errors: object) -> None: + """Reject malformed collection errors without inventing missing evidence.""" + snapshot = report_snapshot() + snapshot["collection_errors"] = collection_errors + with pytest.raises(queue_health.QueueHealthError): + queue_health.build_report(snapshot, now=NOW) + + snapshot["collection_errors"] = None + assert queue_health.build_report(snapshot, now=NOW)["summary"]["collection_error_count"] == 0 + + +def test_collect_snapshot_bounds_workflow_run_payloads_to_fifty_items() -> None: + """Ignore large historical totals while retaining bounded active-run pagination.""" + requested_paths: list[str] = [] + + def runner(args: list[str], **kwargs: object) -> CompletedProcess[str]: + """Return empty bounded run pages and record the requested endpoints.""" + path = args[-1] + requested_paths.append(path) + if path == "repos/owner/repo": + payload: object = {"default_branch": "main"} + elif path == "repos/owner/repo/pulls?state=open&per_page=100": + payload = [] + elif path == "repos/owner/repo/actions/runs?per_page=50": + payload = {"total_count": 2_001, "workflow_runs": []} + elif "/actions/runs?status=" in path: + status = path.split("status=", 1)[1].split("&", 1)[0] + if status == "cancelled": + payload = {"total_count": 0, "workflow_runs": []} + else: + run_id = ("in_progress", "pending", "queued", "requested", "waiting").index(status) + payload = { + "total_count": 1, + "workflow_runs": [workflow_run(100 + run_id, status=status)], + } + else: # pragma: no cover - a new endpoint must be explicitly governed + raise AssertionError(f"unexpected endpoint: {path}") + if "--paginate" in args: + payload = [payload] + return CompletedProcess(args, 0, json.dumps(payload), "") + + snapshot = queue_health.collect_snapshot( + ["owner/repo"], runner=runner, generated_at="2026-08-19T11:00:00Z" + ) + + run_paths = [path for path in requested_paths if "/actions/runs?status=" in path] + assert "repos/owner/repo/actions/runs?per_page=50" not in requested_paths + assert run_paths == [ + *[ + f"repos/owner/repo/actions/runs?status={status}&per_page=50" + for status in ("in_progress", "pending", "queued", "requested", "waiting") + ], + *[ + f"repos/owner/repo/actions/runs?status={status}&per_page=50" + for status in ("waiting", "requested", "queued", "pending", "in_progress") + ], + "repos/owner/repo/actions/runs?status=cancelled&event=pull_request_target&per_page=50", + ] + assert not any("page=2" in path for path in requested_paths) + assert {run["status"] for run in snapshot["repositories"][0]["runs"]} == { + "IN_PROGRESS", + "PENDING", + "QUEUED", + "REQUESTED", + "WAITING", + } + + queued_reads = 0 + + def changing_runner(args: list[str], **kwargs: object) -> CompletedProcess[str]: + """Expose a queue transition between the two bounded status sweeps.""" + nonlocal queued_reads + path = args[-1] + if path == "repos/owner/repo": + payload: object = {"default_branch": "main"} + elif path == "repos/owner/repo/pulls?state=open&per_page=100": + payload = [] + elif "/actions/runs?status=" in path: + status = path.split("status=", 1)[1].split("&", 1)[0] + if status == "queued": + queued_reads += 1 + payload = [workflow_run(900, status=status)] if queued_reads == 1 else [] + else: + payload = [] + else: # pragma: no cover - a new endpoint must be explicitly governed + raise AssertionError(f"unexpected endpoint: {path}") + if "--paginate" in args: + payload = [payload] + return CompletedProcess(args, 0, json.dumps(payload), "") + + changing_snapshot = queue_health.collect_snapshot(["owner/repo"], runner=changing_runner) + assert changing_snapshot["repositories"] == [] + assert changing_snapshot["collection_errors"] == [ + { + "repository": "owner/repo", + "error": "active workflow run snapshot changed during collection", + } + ] + + +def test_load_snapshot_and_identity_helpers(tmp_path: Path) -> None: + """Load offline snapshots and classify current, obsolete, and unlinked runs.""" + path = tmp_path / "snapshot.json" + path.write_text(json.dumps(report_snapshot()), encoding="utf-8") + assert queue_health.load_snapshot(path)["generated_at"] == "2026-08-19T11:00:00Z" + path.write_text("[]", encoding="utf-8") + with pytest.raises(queue_health.QueueHealthError): + queue_health.load_snapshot(path) + path.write_text("{", encoding="utf-8") + with pytest.raises(queue_health.QueueHealthError): + queue_health.load_snapshot(path) + with pytest.raises(queue_health.QueueHealthError): + queue_health.load_snapshot(tmp_path / "missing.json") + + current_run = {"head_sha": "head", "pull_requests": [{"number": 1, "head_sha": "head"}]} + assert queue_health._run_identity(current_run, {1: {"head_sha": "head"}}) == ("current_head", 1) + assert queue_health._run_identity(current_run, {1: {"head_sha": "other"}}) == ("obsolete", 1) + assert queue_health._run_identity({"pull_requests": []}, {}) == ("unlinked", None) + + # pull_request_target runs report the *base*-branch commit as their + # run-level head_sha; identity must use the linked entry's head_sha + # (preserved by _normalise_run) instead, or an active run looks obsolete. + base_triggered_run = { + "head_sha": "base-branch-checkout-sha", + "pull_requests": [{"number": 1, "head_sha": "pr-head-sha"}], + } + assert queue_health._run_identity(base_triggered_run, {1: {"head_sha": "pr-head-sha"}}) == ( + "current_head", + 1, + ) + stale_base_triggered_run = { + "head_sha": "base-branch-checkout-sha", + "pull_requests": [{"number": 1, "head_sha": "old-pr-head-sha"}], + } + assert queue_health._run_identity( + stale_base_triggered_run, {1: {"head_sha": "pr-head-sha"}} + ) == ("obsolete", 1) + + +def test_job_state_and_queue_age_cover_pending_terminal_and_unknown_paths() -> None: + """Classify job assignment states and calculate bounded queue age.""" + assert queue_health._job_state({"status": "queued", "runner_id": 1}) == ("queued_assigned", True, True) + assert queue_health._job_state({"status": "in_progress", "runner_name": "runner"}) == ( + "queued_assigned", + True, + True, + ) + assert queue_health._job_state({"status": "queued"}) == ("queued_unassigned", True, False) + assert queue_health._job_state({"status": "pending"}) == ("queued_unassigned", True, False) + assert queue_health._job_state({"status": "requested"}) == ("queued_unassigned", True, False) + assert queue_health._job_state({"status": "completed"}) == ("terminal", False, False) + assert queue_health._job_state({"status": "", "conclusion": "failure"}) == ("terminal", False, False) + # "waiting" (paused on an environment/deployment approval) is pending + # evidence, not unclassified "unknown" evidence that drops off the report. + assert queue_health._job_state({"status": "waiting"}) == ("waiting_approval", True, False) + assert queue_health._job_state({"status": "waiting", "runner_id": 3}) == ( + "waiting_approval", + True, + True, + ) + assert queue_health._format_age("2026-08-19T10:00:00Z", NOW) == 7200 + assert queue_health._format_age("2026-08-19T13:00:00Z", NOW) == 0 + with pytest.raises(queue_health.QueueHealthError): + queue_health._format_age("bad", NOW) + + +def test_build_report_classifies_exact_head_and_external_blockers() -> None: + """Report exact-head, stale, unlinked, terminal, and SLO evidence separately.""" + report = queue_health.build_report(report_snapshot(), now=NOW, queue_age_slo_seconds=900) + assert report["schema_version"] == "actions.queue_health.v1" + assert report["summary"]["observed_job_count"] == 7 + # job 102 (run 10) is "waiting" on an environment/deployment approval; it + # must be visible as pending evidence, not silently dropped. + assert report["summary"]["pending_job_count"] == 6 + assert report["summary"]["current_head_pending_count"] == 4 + assert report["summary"]["unassigned_slo_breached_count"] == 2 + assert report["summary"]["obsolete_job_count"] == 1 + assert report["summary"]["unlinked_job_count"] == 1 + assert report["summary"]["duplicate_pending_lane_count"] == 1 + assert report["summary"]["terminal_job_count"] == 1 + # Run 10 has two pending jobs and run 13 has one fallback row; the metric + # counts concurrent runs, not the number of pending jobs in those runs. + assert report["duplicate_pending_lanes"][0]["count"] == 2 + assert any(row["blocker"] == "obsolete_run_requires_identity_confirmed_cleanup" for row in report["runs"]) + assert any(row["blocker"] == "run_not_linked_to_pull_request" for row in report["runs"]) + waiting_row = next(row for row in report["runs"] if row["job_id"] == 102) + assert waiting_row["execution_state"] == "waiting_approval" + assert waiting_row["is_pending"] is True + assert waiting_row["blocker"] == "environment_or_deployment_approval_required" + assert waiting_row["recommended_action"] == "reviewer_or_owner_approve_pending_environment_deployment" + # A waiting-on-approval job must never be folded into the runner-capacity + # SLO metric, which is reserved for queued_unassigned jobs. + assert report["summary"]["unassigned_slo_breached_count"] == 2 + assert report["runs"] == sorted(report["runs"], key=lambda row: (row["repository"], row["run_id"], row["job_id"])) + assert queue_health.build_report(report_snapshot(), now=NOW, queue_age_slo_seconds=7200)["summary"]["unassigned_slo_breached_count"] == 0 + assert queue_health.build_report(report_snapshot(), queue_age_slo_seconds=0)["summary"]["observed_job_count"] == 7 + + +def test_build_report_treats_pull_request_target_linked_head_as_current() -> None: + """A pull_request_target run's base-commit head_sha must not look obsolete. + + For a ``pull_request_target``-triggered run, GitHub reports the checked + out *base*-branch commit as the run-level ``head_sha``, while the run's + linked pull-request entry still carries the real PR head SHA. The run + must classify as ``current_head`` (and have its job evidence inspected) + whenever that linked head SHA matches the currently open pull request. + """ + snapshot = { + "generated_at": "2026-08-19T11:00:00Z", + "repositories": [ + { + "full_name": "owner/repo", + "pull_requests": [pull_request(1, "pr-head-sha")], + "runs": [ + workflow_run( + 50, + head_sha="base-branch-checkout-sha", + pull_requests=[{"number": 1, "head": {"sha": "pr-head-sha"}}], + jobs=[job(500)], + workflow_name="opencode-review", + ), + ], + } + ], + } + report = queue_health.build_report(snapshot, now=NOW) + row = report["runs"][0] + assert row["identity_state"] == "current_head" + assert row["obsolete"] is False + assert row["blocker"] != "obsolete_run_requires_identity_confirmed_cleanup" + + +def test_build_report_measures_job_wait_from_job_created_at_not_run_creation() -> None: + """A freshly queued job inside an old in-progress run must not inherit its age.""" + old_run_created_at = "2026-08-19T08:00:00Z" + recent_job_created_at = "2026-08-19T11:55:00Z" + snapshot = { + "generated_at": "2026-08-19T11:00:00Z", + "repositories": [ + { + "full_name": "owner/repo", + "pull_requests": [pull_request()], + "runs": [ + workflow_run( + 60, + created_at=old_run_created_at, + status="in_progress", + pull_requests=[{"number": 1, "head": {"sha": "head"}}], + jobs=[ + job(600, status="in_progress", runner_id=1, runner_name="runner-1"), + { + "id": 601, + "name": "second-stage", + "status": "queued", + "conclusion": None, + "runner_id": None, + "runner_name": None, + "created_at": recent_job_created_at, + "steps": [], + }, + ], + ), + ], + } + ], + } + report = queue_health.build_report(snapshot, now=NOW, queue_age_slo_seconds=900) + second_stage = next(row for row in report["runs"] if row["job_id"] == 601) + assert second_stage["queue_age_seconds"] == 300 + assert second_stage["slo_breached"] is False + assert second_stage["blocker"] == "current_head_required_evidence_incomplete" + # The first-stage job is still measured against the run's own (old) + # creation time because it carries no job-level created_at of its own, + # but it is runner-assigned so it never counts as an unassigned breach. + first_stage = next(row for row in report["runs"] if row["job_id"] == 600) + assert first_stage["queue_age_seconds"] == 14400 + assert report["summary"]["unassigned_slo_breached_count"] == 0 + + +@pytest.mark.parametrize( + "snapshot, message", + [ + ({"generated_at": "2026-08-19T11:00:00Z", "repositories": "bad"}, "repositories"), + ({"generated_at": "2026-08-19T11:00:00Z", "repositories": ["bad"]}, "repository entry"), + ( + {"generated_at": "2026-08-19T11:00:00Z", "repositories": [{"full_name": "owner/repo", "pull_requests": "bad", "runs": []}]}, + "pull requests", + ), + ( + {"generated_at": "2026-08-19T11:00:00Z", "repositories": [{"full_name": "owner/repo", "pull_requests": [], "runs": "bad"}]}, + "runs", + ), + ( + {"generated_at": "2026-08-19T11:00:00Z", "repositories": [{"full_name": "owner/repo", "pull_requests": [], "runs": ["bad"]}]}, + "workflow run entry", + ), + ], +) +def test_build_report_rejects_malformed_snapshot_shapes(snapshot: dict, message: str) -> None: + """Reject malformed snapshot containers before counting jobs.""" + with pytest.raises(queue_health.QueueHealthError, match=message): + queue_health.build_report(snapshot, now=NOW) + + +def test_build_report_rejects_duplicate_and_invalid_entries() -> None: + """Reject duplicate identities and invalid report boundaries.""" + duplicate_repository = report_snapshot() + duplicate_repository["repositories"].append( + dict(duplicate_repository["repositories"][0]) + ) + with pytest.raises(queue_health.QueueHealthError, match="duplicate repository entry"): + queue_health.build_report(duplicate_repository, now=NOW) + + duplicate_pr = report_snapshot() + duplicate_pr["repositories"][0]["pull_requests"].append(pull_request(1, "other")) + with pytest.raises(queue_health.QueueHealthError, match="duplicate pull request"): + queue_health.build_report(duplicate_pr, now=NOW) + + duplicate_run = report_snapshot() + duplicate_run["repositories"][0]["runs"].append(workflow_run(10)) + with pytest.raises(queue_health.QueueHealthError, match="duplicate workflow run"): + queue_health.build_report(duplicate_run, now=NOW) + + invalid_jobs = report_snapshot() + invalid_jobs["repositories"][0]["runs"][0]["jobs"] = "bad" + with pytest.raises(queue_health.QueueHealthError, match="jobs"): + queue_health.build_report(invalid_jobs, now=NOW) + + with pytest.raises(queue_health.QueueHealthError, match="negative"): + queue_health.build_report(report_snapshot(), now=NOW, queue_age_slo_seconds=-1) + with pytest.raises(queue_health.QueueHealthError, match="timestamp"): + queue_health.build_report({"generated_at": "bad", "repositories": []}, now=NOW) + with pytest.raises(queue_health.QueueHealthError, match="evaluation time"): + queue_health.build_report(report_snapshot(), now=datetime(2026, 8, 19, 12, 0)) + + for key in ("pull_requests", "runs"): + null_entry = {"generated_at": "2026-08-19T11:00:00Z", "repositories": [{"full_name": "owner/repo", key: None}]} + assert queue_health.build_report(null_entry, now=NOW)["summary"]["observed_job_count"] == 0 + null_jobs = { + "generated_at": "2026-08-19T11:00:00Z", + "repositories": [{"full_name": "owner/repo", "runs": [{"id": 1, "created_at": "2026-08-19T10:00:00Z", "jobs": None}]}], + } + assert queue_health.build_report(null_jobs, now=NOW)["summary"]["observed_job_count"] == 1 + + +def test_render_and_write_reports_escape_fields_and_support_empty_reports(tmp_path: Path) -> None: + """Escape HTML fields and write both populated and empty reports.""" + report = queue_health.build_report(report_snapshot(), now=NOW) + report["runs"][0]["blocker"] = "" + rendered = queue_health.render_html(report) + assert "<script>" in rendered + assert 'owner/repo' in rendered + assert "queue-age SLO: 900 seconds" in rendered + + empty = queue_health.build_report({"generated_at": "2026-08-19T11:00:00Z", "repositories": []}, now=NOW) + assert "No queued or in-progress jobs observed." in queue_health.render_html(empty) + + json_path = tmp_path / "nested" / "report.json" + html_path = tmp_path / "nested" / "report.html" + queue_health.write_reports(report, json_path, html_path) + assert json.loads(json_path.read_text(encoding="utf-8"))["schema_version"] == "actions.queue_health.v1" + assert " None: + """Exercise snapshot mode, allowlist mode, and bounded CLI failures.""" + args = queue_health.parse_args( + ["--snapshot", "snapshot.json", "--output-json", "out.json", "--output-html", "out.html"] + ) + assert args.snapshot == Path("snapshot.json") + args = queue_health.parse_args( + ["--allowlist", "allowlist.json", "--output-json", "out.json", "--output-html", "out.html"] + ) + assert args.allowlist == Path("allowlist.json") + with pytest.raises(SystemExit): + queue_health.parse_args(["--snapshot", "a", "--allowlist", "b", "--output-json", "o", "--output-html", "h"]) + + snapshot_path = tmp_path / "snapshot.json" + snapshot_path.write_text(json.dumps(report_snapshot()), encoding="utf-8") + json_path = tmp_path / "out.json" + html_path = tmp_path / "out.html" + assert queue_health.main( + [ + "--snapshot", + str(snapshot_path), + "--output-json", + str(json_path), + "--output-html", + str(html_path), + "--now", + "2026-08-19T12:00:00Z", + ] + ) == 0 + assert "QUEUE_HEALTH_RESULT=" in capsys.readouterr().out + + empty_snapshot_path = tmp_path / "empty-snapshot.json" + empty_snapshot_path.write_text( + json.dumps({"generated_at": "2026-08-19T11:00:00Z", "repositories": []}), + encoding="utf-8", + ) + assert queue_health.main( + [ + "--snapshot", + str(empty_snapshot_path), + "--output-json", + str(json_path), + "--output-html", + str(html_path), + "--now", + "2026-08-19T12:00:00Z", + ] + ) == 0 + assert "::warning::" not in capsys.readouterr().out + + error = io.StringIO() + assert queue_health.main( + ["--snapshot", str(tmp_path / "missing.json"), "--output-json", "o", "--output-html", "h"], + stderr=error, + ) == 2 + assert "ERROR:" in error.getvalue() + + allowlist_path = tmp_path / "allowlist.json" + allowlist_path.write_text(json.dumps(["owner/repo"]), encoding="utf-8") + original_collect = queue_health.collect_snapshot + queue_health.collect_snapshot = lambda repositories: report_snapshot() # type: ignore[assignment] + try: + assert queue_health.main( + ["--allowlist", str(allowlist_path), "--output-json", str(json_path), "--output-html", str(html_path)] + ) == 0 + finally: + queue_health.collect_snapshot = original_collect + assert "QUEUE_HEALTH_RESULT=" in capsys.readouterr().out diff --git a/tests/test_actions_queue_health_cancelled_before_runner.py b/tests/test_actions_queue_health_cancelled_before_runner.py new file mode 100644 index 0000000000..1827275f68 --- /dev/null +++ b/tests/test_actions_queue_health_cancelled_before_runner.py @@ -0,0 +1,322 @@ +"""Regression coverage for workflow cancellation before runner assignment.""" + +from __future__ import annotations + +import importlib.util +import json +from datetime import datetime, timezone +from pathlib import Path +from subprocess import CompletedProcess + + +ROOT = Path(__file__).resolve().parents[1] +MODULE_PATH = ROOT / "scripts/ci/actions_queue_health.py" +SPEC = importlib.util.spec_from_file_location("actions_queue_health", MODULE_PATH) +assert SPEC and SPEC.loader +queue_health = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(queue_health) + + +def test_collect_snapshot_classifies_cancelled_job_before_runner_assignment() -> None: + """A cancelled current-head job with no runner or steps stays explicit evidence.""" + repository_name = "owner/repo" + pull_request = { + "number": 17, + "state": "open", + "base": {"ref": "main", "repo": {"full_name": repository_name}}, + "head": {"sha": "exact-head"}, + "updated_at": "2026-09-02T13:15:00Z", + } + cancelled_run = { + "id": 1701, + "name": "Repository Metadata Reconcile", + "workflow_id": 9017, + "event": "pull_request", + "status": "completed", + "conclusion": "cancelled", + "head_sha": "exact-head", + "created_at": "2026-09-02T13:00:00Z", + "updated_at": "2026-09-02T13:08:00Z", + "run_attempt": 1, + "pull_requests": [{"number": 17, "head": {"sha": "exact-head"}}], + } + cancelled_job = { + "id": 17001, + "name": "validate", + "status": "completed", + "conclusion": "cancelled", + "runner_id": 0, + "runner_name": "", + "created_at": "2026-09-02T13:00:00Z", + "steps": [], + } + skipped_job = { + "id": 17002, + "name": "publish optional evidence", + "status": "completed", + "conclusion": "skipped", + "runner_id": 0, + "runner_name": "", + "created_at": "2026-09-02T13:00:00Z", + "steps": [], + } + missing_steps_job = { + "id": 17003, + "name": "cancelled without step evidence", + "status": "completed", + "conclusion": "cancelled", + "runner_id": 0, + "runner_name": "", + "created_at": "2026-09-02T13:00:00Z", + } + null_steps_job = { + **missing_steps_job, + "id": 17004, + "name": "cancelled with null step evidence", + "steps": None, + } + terminal_path = ( + f"repos/{repository_name}/actions/runs?status=completed" + "&head_sha=exact-head&per_page=50" + ) + + def runner(args: list[str], **_: object) -> CompletedProcess[str]: + """Return deterministic GitHub REST fixtures for the collector.""" + path = args[-1] + if path == f"repos/{repository_name}": + payload: object = {"default_branch": "main"} + elif path == f"repos/{repository_name}/pulls?state=open&per_page=100": + payload = [pull_request] + elif path == terminal_path: + payload = {"total_count": 1, "workflow_runs": [cancelled_run]} + elif path == f"repos/{repository_name}/actions/runs/1701/jobs?per_page=100": + payload = { + "total_count": 4, + "jobs": [ + cancelled_job, + skipped_job, + missing_steps_job, + null_steps_job, + ], + } + elif "status=startup_failure" in path: + raise AssertionError( + "GitHub workflow-run status filtering does not accept startup_failure" + ) + elif path.startswith(f"repos/{repository_name}/actions/runs?status="): + payload = {"total_count": 0, "workflow_runs": []} + else: # pragma: no cover - unexpected API expansion must fail loudly. + raise AssertionError(f"unexpected GitHub API path: {path}") + return CompletedProcess(args, 0, json.dumps(payload), "") + + snapshot = queue_health.collect_snapshot( + [repository_name], + runner=runner, + generated_at="2026-09-02T13:16:00Z", + ) + + assert snapshot["collection_errors"] == [] + assert [run["id"] for run in snapshot["repositories"][0]["runs"]] == [1701] + + report = queue_health.build_report( + snapshot, + now=datetime(2026, 9, 2, 13, 16, tzinfo=timezone.utc), + ) + cancelled_row = next(row for row in report["runs"] if row["job_id"] == 17001) + assert cancelled_row["identity_state"] == "current_head" + assert cancelled_row["run_conclusion"] == "CANCELLED" + assert cancelled_row["jobs_materialized"] is True + assert cancelled_row["runner_assigned"] is False + assert cancelled_row["admission_state"] == "cancelled_before_runner_assignment" + assert cancelled_row["blocker"] == "cancelled_before_runner_assignment" + assert cancelled_row["recommended_action"] == ( + "inspect_actions_control_plane_without_leaf_bypass" + ) + + skipped_row = next(row for row in report["runs"] if row["job_id"] == 17002) + assert skipped_row["run_conclusion"] == "CANCELLED" + assert skipped_row["admission_state"] != "cancelled_before_runner_assignment" + for unavailable_step_job_id in (17003, 17004): + unavailable_step_row = next( + row for row in report["runs"] if row["job_id"] == unavailable_step_job_id + ) + assert unavailable_step_row["admission_state"] != ( + "cancelled_before_runner_assignment" + ) + assert report["summary"]["cancelled_before_runner_assignment_count"] == 1 + + +def test_collect_snapshot_retains_cancelled_pull_request_target_current_head() -> None: + """A target-triggered cancellation uses linked PR head identity, not base SHA.""" + repository_name = "owner/repo" + pull_request = { + "number": 23, + "state": "open", + "base": {"ref": "main", "repo": {"full_name": repository_name}}, + "head": {"sha": "exact-target-head"}, + "updated_at": "2026-09-02T13:20:00Z", + } + cancelled_run = { + "id": 2301, + "name": "Target Review", + "workflow_id": 9023, + "event": "pull_request_target", + "status": "completed", + "conclusion": "cancelled", + "head_sha": "base-commit-sha", + "created_at": "2026-09-02T13:00:00Z", + "updated_at": "2026-09-02T13:05:00Z", + "run_attempt": 1, + "pull_requests": [ + {"number": 23, "head": {"sha": "exact-target-head"}} + ], + } + cancelled_job = { + "id": 23001, + "name": "review", + "status": "completed", + "conclusion": "cancelled", + "runner_id": 0, + "runner_name": "", + "created_at": "2026-09-02T13:00:00Z", + "steps": [], + } + head_terminal_path = ( + f"repos/{repository_name}/actions/runs?status=completed" + "&head_sha=exact-target-head&per_page=50" + ) + target_cancelled_path = ( + f"repos/{repository_name}/actions/runs?status=cancelled" + "&event=pull_request_target&per_page=50" + ) + + def runner(args: list[str], **_: object) -> CompletedProcess[str]: + """Model GitHub target runs whose run-level SHA is the base commit.""" + path = args[-1] + if path == f"repos/{repository_name}": + payload: object = {"default_branch": "main"} + elif path == f"repos/{repository_name}/pulls?state=open&per_page=100": + payload = [pull_request] + elif path == head_terminal_path: + payload = {"total_count": 0, "workflow_runs": []} + elif path == target_cancelled_path: + payload = {"total_count": 1, "workflow_runs": [cancelled_run]} + elif "status=startup_failure" in path: + raise AssertionError( + "GitHub workflow-run status filtering does not accept startup_failure" + ) + elif path == f"repos/{repository_name}/actions/runs/2301/jobs?per_page=100": + payload = {"total_count": 1, "jobs": [cancelled_job]} + elif path.startswith(f"repos/{repository_name}/actions/runs?status="): + payload = {"total_count": 0, "workflow_runs": []} + else: # pragma: no cover - unexpected API expansion must fail loudly. + raise AssertionError(f"unexpected GitHub API path: {path}") + return CompletedProcess(args, 0, json.dumps(payload), "") + + snapshot = queue_health.collect_snapshot( + [repository_name], + runner=runner, + generated_at="2026-09-02T13:21:00Z", + ) + + assert snapshot["collection_errors"] == [] + assert [run["id"] for run in snapshot["repositories"][0]["runs"]] == [2301] + report = queue_health.build_report( + snapshot, + now=datetime(2026, 9, 2, 13, 21, tzinfo=timezone.utc), + ) + assert report["runs"][0]["identity_state"] == "current_head" + assert report["runs"][0]["admission_state"] == ( + "cancelled_before_runner_assignment" + ) + + +def test_collect_snapshot_rejects_head_change_after_target_evidence_read() -> None: + """Terminal evidence is rejected when its PR identity changes before completion.""" + repository_name = "owner/repo" + original_pull_request = { + "number": 29, + "state": "open", + "base": {"ref": "main", "repo": {"full_name": repository_name}}, + "head": {"sha": "original-head"}, + "updated_at": "2026-09-02T13:22:00Z", + } + changed_pull_request = { + **original_pull_request, + "head": {"sha": "replacement-head"}, + "updated_at": "2026-09-02T13:24:00Z", + } + cancelled_run = { + "id": 2901, + "name": "Target Review", + "workflow_id": 9029, + "event": "pull_request_target", + "status": "completed", + "conclusion": "cancelled", + "head_sha": "base-commit-sha", + "created_at": "2026-09-02T13:00:00Z", + "updated_at": "2026-09-02T13:05:00Z", + "run_attempt": 1, + "pull_requests": [{"number": 29, "head": {"sha": "original-head"}}], + } + cancelled_job = { + "id": 29001, + "name": "review", + "status": "completed", + "conclusion": "cancelled", + "runner_id": 0, + "runner_name": "", + "created_at": "2026-09-02T13:00:00Z", + "steps": [], + } + head_terminal_path = ( + f"repos/{repository_name}/actions/runs?status=completed" + "&head_sha=original-head&per_page=50" + ) + target_cancelled_path = ( + f"repos/{repository_name}/actions/runs?status=cancelled" + "&event=pull_request_target&per_page=50" + ) + pull_read_count = 0 + + def runner(args: list[str], **_: object) -> CompletedProcess[str]: + """Advance the PR head only after terminal/job evidence has been read.""" + nonlocal pull_read_count + path = args[-1] + if path == f"repos/{repository_name}": + payload: object = {"default_branch": "main"} + elif path == f"repos/{repository_name}/pulls?state=open&per_page=100": + pull_read_count += 1 + payload = [ + changed_pull_request if pull_read_count >= 3 else original_pull_request + ] + elif path == head_terminal_path: + payload = {"total_count": 0, "workflow_runs": []} + elif path == target_cancelled_path: + payload = {"total_count": 1, "workflow_runs": [cancelled_run]} + elif path == f"repos/{repository_name}/actions/runs/2901/jobs?per_page=100": + payload = {"total_count": 1, "jobs": [cancelled_job]} + elif "status=startup_failure" in path: + raise AssertionError( + "GitHub workflow-run status filtering does not accept startup_failure" + ) + elif path.startswith(f"repos/{repository_name}/actions/runs?status="): + payload = {"total_count": 0, "workflow_runs": []} + else: # pragma: no cover - unexpected API expansion must fail loudly. + raise AssertionError(f"unexpected GitHub API path: {path}") + return CompletedProcess(args, 0, json.dumps(payload), "") + + snapshot = queue_health.collect_snapshot( + [repository_name], + runner=runner, + generated_at="2026-09-02T13:25:00Z", + ) + + assert snapshot["repositories"] == [] + assert snapshot["collection_errors"] == [ + { + "repository": repository_name, + "error": "pull-request identity snapshot changed during evidence collection", + } + ] + assert pull_read_count == 3 diff --git a/tests/test_actions_queue_health_contract.py b/tests/test_actions_queue_health_contract.py new file mode 100644 index 0000000000..d14a5b44b7 --- /dev/null +++ b/tests/test_actions_queue_health_contract.py @@ -0,0 +1,52 @@ +"""Contract tests for the scheduled read-only Actions queue report.""" + +import json +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[1] + + +def test_queue_health_workflow_is_scheduled_read_only_and_pinned() -> None: + """Keep the scheduled collector bounded, read-only, and supply-chain pinned.""" + workflow = (ROOT / ".github/workflows/actions-queue-health.yml").read_text(encoding="utf-8") + + assert 'cron: "7 * * * *"' in workflow + assert "workflow_dispatch:" not in workflow + assert "cancel-in-progress: false" in workflow + assert "timeout-minutes: 30" in workflow + assert "runs-on: ubuntu-24.04" in workflow + assert "actions: read" in workflow + assert "pull-requests: read" not in workflow + assert "contents: write" not in workflow + assert ( + "GH_TOKEN: ${{ secrets.PR_REVIEW_MERGE_TOKEN || secrets.OPENCODE_APPROVE_TOKEN }}" + in workflow + ) + assert "GH_TOKEN: ${{ github.token }}" not in workflow + assert "required for cross-repository queue reads" in workflow + assert "gh run cancel" not in workflow + assert "gh pr merge" not in workflow + assert "step-security/harden-runner@b09bb98e06d4d774595224525879c09bc6e98c40" in workflow + assert "actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0" in workflow + assert "actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a" in workflow + assert "actions_queue_health.py" in workflow + assert "actions_queue_health_repositories.json" in workflow + + +def test_queue_health_allowlist_is_explicit_and_bounded() -> None: + """Keep the first product slice limited to its reviewed repositories.""" + payload = json.loads( + (ROOT / "config/actions_queue_health_repositories.json").read_text(encoding="utf-8") + ) + assert payload == { + "repositories": [ + "ContextualWisdomLab/.github", + "ContextualWisdomLab/ConceptWeave", + "ContextualWisdomLab/ELUNVERA", + "ContextualWisdomLab/TEPP", + "ContextualWisdomLab/contextual-orchestrator", + "ContextualWisdomLab/fast-mlsirm", + "ContextualWisdomLab/naruon", + ] + } diff --git a/tests/test_actions_queue_health_post_evidence_retry.py b/tests/test_actions_queue_health_post_evidence_retry.py new file mode 100644 index 0000000000..bc266fa96f --- /dev/null +++ b/tests/test_actions_queue_health_post_evidence_retry.py @@ -0,0 +1,84 @@ +"""Regression tests for post-evidence pull-request identity retry semantics.""" + +import importlib.util +import json +from pathlib import Path +from subprocess import CompletedProcess + + +ROOT = Path(__file__).resolve().parents[1] +MODULE_PATH = ROOT / "scripts/ci/actions_queue_health.py" +SPEC = importlib.util.spec_from_file_location("actions_queue_health_post_evidence_retry", MODULE_PATH) +assert SPEC and SPEC.loader +queue_health = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(queue_health) + + +def _pull(head_sha: str = "head") -> dict: + """Return one complete raw open-pull-request identity fixture.""" + return { + "number": 1, + "state": "open", + "base": {"ref": "main", "repo": {"full_name": "owner/repo"}}, + "head": {"sha": head_sha}, + "updated_at": "2026-09-02T14:00:00Z", + } + + +def _incomplete_pull() -> dict: + """Return a transiently incomplete identity fixture.""" + pull_request = _pull() + pull_request["head"] = {"sha": ""} + return pull_request + + +def _runner_with_post_evidence_identity_reads(*, persistent: bool): + """Return a runner that makes the post-evidence identity read incomplete.""" + pull_reads = 0 + + def runner(args: list[str], **kwargs: object) -> CompletedProcess[str]: + """Serve stable queue evidence with a transient or persistent final identity gap.""" + nonlocal pull_reads + path = args[-1] + if path == "repos/owner/repo": + payload: object = {"default_branch": "main"} + elif path == "repos/owner/repo/pulls?state=open&per_page=100": + pull_reads += 1 + if pull_reads == 3 or (persistent and pull_reads >= 3): + payload = [_incomplete_pull()] + else: + payload = [_pull()] + elif "/actions/runs?" in path: + payload = {"total_count": 0, "workflow_runs": []} + else: # pragma: no cover - any new endpoint must be explicitly governed. + raise AssertionError(f"unexpected endpoint: {path}") + return CompletedProcess(args, 0, json.dumps(payload), "") + + return runner + + +def test_post_evidence_identity_read_retries_one_transient_incomplete_snapshot(monkeypatch) -> None: + """A transient incomplete post-evidence identity read receives one bounded retry.""" + monkeypatch.setattr(queue_health.time, "sleep", lambda _: None) + snapshot = queue_health.collect_snapshot( + ["owner/repo"], + runner=_runner_with_post_evidence_identity_reads(persistent=False), + generated_at="2026-09-02T14:00:00Z", + ) + assert snapshot["collection_errors"] == [] + assert len(snapshot["repositories"]) == 1 + assert snapshot["repositories"][0]["pull_requests"][0]["head_sha"] == "head" + + +def test_post_evidence_identity_read_fails_closed_after_retry_remains_incomplete(monkeypatch) -> None: + """Persistent incomplete post-evidence identity is repository-scoped failure.""" + monkeypatch.setattr(queue_health.time, "sleep", lambda _: None) + snapshot = queue_health.collect_snapshot( + ["owner/repo"], + runner=_runner_with_post_evidence_identity_reads(persistent=True), + generated_at="2026-09-02T14:00:00Z", + ) + assert snapshot["repositories"] == [] + assert len(snapshot["collection_errors"]) == 1 + assert snapshot["collection_errors"][0]["repository"] == "owner/repo" + assert "pull-request identity validation failed" in snapshot["collection_errors"][0]["error"] diff --git a/tests/test_actions_queue_health_snapshot_consistency.py b/tests/test_actions_queue_health_snapshot_consistency.py new file mode 100644 index 0000000000..b9711a09dd --- /dev/null +++ b/tests/test_actions_queue_health_snapshot_consistency.py @@ -0,0 +1,195 @@ +"""Regression tests for stable queue-health identity and audit evidence.""" + +from datetime import datetime, timezone +import importlib.util +import json +from pathlib import Path +from subprocess import CompletedProcess + +import pytest + + +ROOT = Path(__file__).resolve().parents[1] +MODULE_PATH = ROOT / "scripts/ci/actions_queue_health.py" +SPEC = importlib.util.spec_from_file_location("actions_queue_health_consistency", MODULE_PATH) +assert SPEC and SPEC.loader +queue_health = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(queue_health) +NOW = datetime(2026, 9, 2, 0, 0, tzinfo=timezone.utc) + + +def _pull(head_sha: str = "head") -> dict: + """Return one complete open pull-request identity fixture.""" + return { + "number": 1, + "state": "open", + "base": {"ref": "main", "repo": {"full_name": "owner/repo"}}, + "head": {"sha": head_sha}, + "updated_at": "2026-09-01T23:00:00Z", + } + + +def _run(run_id: int, workflow_id: int, *, name: str = "shared-name") -> dict: + """Return one current-head queued workflow run with stable workflow identity.""" + return { + "id": run_id, + "workflow_id": workflow_id, + "name": name, + "event": "pull_request", + "status": "queued", + "conclusion": "", + "head_sha": "head", + "created_at": "2026-09-01T23:30:00Z", + "updated_at": "2026-09-01T23:30:00Z", + "run_attempt": 1, + "pull_requests": [{"number": 1, "head": {"sha": "head"}}], + "jobs": [], + } + + +def _runner_with_pull_transition(final_pulls: list[dict]): + """Return a runner whose final pull read differs from its initial read.""" + pull_reads = 0 + + def runner(args: list[str], **kwargs: object) -> CompletedProcess[str]: + """Serve metadata, pull identities, and empty active-run partitions.""" + nonlocal pull_reads + path = args[-1] + if path == "repos/owner/repo": + payload: object = {"default_branch": "main"} + elif path == "repos/owner/repo/pulls?state=open&per_page=100": + pull_reads += 1 + payload = [_pull()] if pull_reads == 1 else final_pulls + elif "/actions/runs?status=" in path: + payload = [] + else: # pragma: no cover - any new endpoint must be explicitly governed. + raise AssertionError(f"unexpected endpoint: {path}") + return CompletedProcess(args, 0, json.dumps(payload), "") + + return runner + + +@pytest.mark.parametrize("final_pulls", [[_pull("new-head")], []]) +def test_collect_snapshot_rejects_pull_identity_changes_during_run_sweep( + final_pulls: list[dict], +) -> None: + """A concurrent push or closure cannot corrupt current-head classification.""" + snapshot = queue_health.collect_snapshot( + ["owner/repo"], + runner=_runner_with_pull_transition(final_pulls), + generated_at="2026-09-02T00:00:00Z", + ) + assert snapshot["repositories"] == [] + assert snapshot["collection_errors"] == [ + { + "repository": "owner/repo", + "error": "pull-request identity snapshot changed during collection", + } + ] + + +def test_distinct_workflow_ids_with_same_display_name_are_not_duplicate_lanes() -> None: + """Duplicate-lane evidence groups by stable workflow identity, not display name.""" + snapshot = { + "generated_at": "2026-09-01T23:45:00Z", + "repositories": [ + { + "full_name": "owner/repo", + "pull_requests": [_pull()], + "runs": [_run(100, 501), _run(101, 502)], + } + ], + } + report = queue_health.build_report(snapshot, now=NOW) + assert report["summary"]["duplicate_pending_lane_count"] == 0 + assert {row["workflow_id"] for row in report["runs"]} == {501, 502} + assert {row["workflow_identity"] for row in report["runs"]} == { + "workflow_id:501", + "workflow_id:502", + } + + +def test_same_workflow_id_across_runs_is_one_duplicate_lane() -> None: + """Two pending runs of one workflow remain a true duplicate execution lane.""" + snapshot = { + "generated_at": "2026-09-01T23:45:00Z", + "repositories": [ + { + "full_name": "owner/repo", + "pull_requests": [_pull()], + "runs": [_run(100, 501), _run(101, 501)], + } + ], + } + report = queue_health.build_report(snapshot, now=NOW) + assert report["summary"]["duplicate_pending_lane_count"] == 1 + assert report["duplicate_pending_lanes"] == [ + { + "repository": "owner/repo", + "pull_request_number": 1, + "workflow_identity": "workflow_id:501", + "workflow_name": "shared-name", + "count": 2, + } + ] + + +def test_queue_age_exports_the_timestamp_and_source_used_for_calculation() -> None: + """Report consumers can reproduce queue age from exported evidence.""" + run = _run(100, 501) + run["status"] = "in_progress" + run["jobs"] = [ + { + "id": 1000, + "name": "second-stage", + "status": "queued", + "conclusion": None, + "runner_id": None, + "runner_name": None, + "created_at": "2026-09-01T23:55:00Z", + "steps": [], + } + ] + report = queue_health.build_report( + { + "generated_at": "2026-09-01T23:56:00Z", + "repositories": [ + { + "full_name": "owner/repo", + "pull_requests": [_pull()], + "runs": [run], + } + ], + }, + now=NOW, + ) + row = report["runs"][0] + assert row["queue_age_started_at"] == "2026-09-01T23:55:00Z" + assert row["queue_age_source"] == "job_created_at" + assert row["queue_age_seconds"] == 300 + + +def test_invalid_present_workflow_id_fails_closed() -> None: + """Malformed stable workflow identity cannot silently fall back to a display name.""" + run = _run(100, 501) + run["workflow_id"] = "501" + with pytest.raises(queue_health.QueueHealthError, match="workflow id"): + queue_health.build_report( + { + "generated_at": "2026-09-01T23:45:00Z", + "repositories": [ + { + "full_name": "owner/repo", + "pull_requests": [_pull()], + "runs": [run], + } + ], + }, + now=NOW, + ) + + +def test_queue_health_workflow_does_not_grant_unused_pull_request_permission() -> None: + """The scheduler token keeps only permissions used outside the cross-repository token.""" + workflow = (ROOT / ".github/workflows/actions-queue-health.yml").read_text(encoding="utf-8") + assert "pull-requests: read" not in workflow diff --git a/tests/test_actions_queue_health_startup_failure.py b/tests/test_actions_queue_health_startup_failure.py new file mode 100644 index 0000000000..7bd9cba97e --- /dev/null +++ b/tests/test_actions_queue_health_startup_failure.py @@ -0,0 +1,165 @@ +"""Regression coverage for pre-job GitHub Actions startup failures.""" + +from __future__ import annotations + +import importlib.util +import json +from datetime import datetime, timezone +from pathlib import Path +from subprocess import CompletedProcess + + +ROOT = Path(__file__).resolve().parents[1] +MODULE_PATH = ROOT / "scripts/ci/actions_queue_health.py" +SPEC = importlib.util.spec_from_file_location("actions_queue_health", MODULE_PATH) +assert SPEC and SPEC.loader +queue_health = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(queue_health) + + +def test_collect_snapshot_preserves_current_head_startup_failure_without_jobs() -> None: + """A terminal startup failure with zero jobs must remain visible and explicit.""" + repository_name = "owner/repo" + pull_request = { + "number": 7, + "state": "open", + "base": {"ref": "main", "repo": {"full_name": repository_name}}, + "head": {"sha": "exact-head"}, + "updated_at": "2026-09-02T10:28:00Z", + } + startup_failure_run = { + "id": 701, + "name": "CodeQL PR", + "workflow_id": 9001, + "event": "pull_request", + "status": "completed", + "conclusion": "startup_failure", + "head_sha": "exact-head", + "created_at": "2026-09-02T10:28:00Z", + "updated_at": "2026-09-02T10:28:00Z", + "run_attempt": 1, + "pull_requests": [{"number": 7, "head": {"sha": "exact-head"}}], + } + requested_paths: list[str] = [] + terminal_path = ( + f"repos/{repository_name}/actions/runs?status=completed" + "&head_sha=exact-head&per_page=50" + ) + + def runner(args: list[str], **_: object) -> CompletedProcess[str]: + """Return deterministic GitHub REST fixtures for the collector.""" + path = args[-1] + requested_paths.append(path) + if path == f"repos/{repository_name}": + payload: object = {"default_branch": "main"} + elif path == f"repos/{repository_name}/pulls?state=open&per_page=100": + payload = [pull_request] + elif path == terminal_path: + payload = {"total_count": 1, "workflow_runs": [startup_failure_run]} + elif path == f"repos/{repository_name}/actions/runs/701/jobs?per_page=100": + payload = {"total_count": 0, "jobs": []} + elif path.startswith(f"repos/{repository_name}/actions/runs?status="): + payload = {"total_count": 0, "workflow_runs": []} + else: # pragma: no cover - unexpected API expansion must fail loudly. + raise AssertionError(f"unexpected GitHub API path: {path}") + return CompletedProcess(args, 0, json.dumps(payload), "") + + snapshot = queue_health.collect_snapshot( + [repository_name], + runner=runner, + generated_at="2026-09-02T10:30:00Z", + ) + + assert snapshot["collection_errors"] == [] + assert snapshot["repositories"][0]["runs"] == [ + { + "repository": repository_name, + "id": 701, + "workflow_name": "CodeQL PR", + "event": "pull_request", + "status": "COMPLETED", + "conclusion": "STARTUP_FAILURE", + "head_sha": "exact-head", + "created_at": "2026-09-02T10:28:00Z", + "updated_at": "2026-09-02T10:28:00Z", + "run_attempt": 1, + "concurrency_group": "unavailable_from_actions_api", + "pull_requests": [{"number": 7, "head_sha": "exact-head"}], + "jobs": [], + "workflow_id": 9001, + "workflow_identity": "workflow_id:9001", + } + ] + assert terminal_path in requested_paths + assert f"repos/{repository_name}/actions/runs/701/jobs?per_page=100" in requested_paths + + report = queue_health.build_report( + snapshot, + now=datetime(2026, 9, 2, 10, 30, tzinfo=timezone.utc), + ) + row = report["runs"][0] + assert row["identity_state"] == "current_head" + assert row["execution_state"] == "terminal" + assert row["run_conclusion"] == "STARTUP_FAILURE" + assert row["jobs_materialized"] is False + assert row["blocker"] == "startup_failure_before_job_materialization" + assert row["recommended_action"] == "inspect_actions_control_plane_without_leaf_bypass" + + +def test_collect_snapshot_retains_old_failure_for_unchanged_current_head() -> None: + """Current-head startup failures must not disappear merely because they are old.""" + repository_name = "owner/repo" + pull_request = { + "number": 8, + "state": "open", + "base": {"ref": "main", "repo": {"full_name": repository_name}}, + "head": {"sha": "unchanged-head"}, + "updated_at": "2026-09-02T10:29:00Z", + } + old_current_failure = { + "id": 801, + "name": "CodeQL PR", + "workflow_id": 9001, + "event": "pull_request", + "status": "completed", + "conclusion": "startup_failure", + "head_sha": "unchanged-head", + "created_at": "2026-08-01T10:00:00Z", + "updated_at": "2026-08-01T10:00:00Z", + "run_attempt": 1, + "pull_requests": [{"number": 8, "head": {"sha": "unchanged-head"}}], + } + terminal_path = ( + f"repos/{repository_name}/actions/runs?status=completed" + "&head_sha=unchanged-head&per_page=50" + ) + requested_paths: list[str] = [] + + def runner(args: list[str], **_: object) -> CompletedProcess[str]: + """Return an old but still current-head terminal failure by exact SHA.""" + path = args[-1] + requested_paths.append(path) + if path == f"repos/{repository_name}": + payload: object = {"default_branch": "main"} + elif path == f"repos/{repository_name}/pulls?state=open&per_page=100": + payload = [pull_request] + elif path == terminal_path: + payload = {"total_count": 1, "workflow_runs": [old_current_failure]} + elif path == f"repos/{repository_name}/actions/runs/801/jobs?per_page=100": + payload = {"total_count": 0, "jobs": []} + elif path.startswith(f"repos/{repository_name}/actions/runs?status="): + payload = {"total_count": 0, "workflow_runs": []} + else: # pragma: no cover - unexpected API expansion must fail loudly. + raise AssertionError(f"unexpected GitHub API path: {path}") + return CompletedProcess(args, 0, json.dumps(payload), "") + + snapshot = queue_health.collect_snapshot( + [repository_name], + runner=runner, + generated_at="2026-09-02T10:30:00Z", + ) + + assert snapshot["collection_errors"] == [] + assert [run["id"] for run in snapshot["repositories"][0]["runs"]] == [801] + assert terminal_path in requested_paths + assert not any("&created=" in path for path in requested_paths)