diff --git a/.github/workflows/actions-queue-health.yml b/.github/workflows/actions-queue-health.yml
new file mode 100644
index 000000000..629fc0781
--- /dev/null
+++ b/.github/workflows/actions-queue-health.yml
@@ -0,0 +1,56 @@
+name: GitHub Actions queue health
+
+on:
+ schedule:
+ - cron: "7 * * * *"
+
+concurrency:
+ group: github-actions-queue-health
+ cancel-in-progress: false
+
+permissions:
+ contents: read
+ actions: read
+ pull-requests: read
+
+jobs:
+ collect:
+ name: Collect exact-head queue evidence
+ runs-on: ubuntu-24.04
+ permissions:
+ contents: read
+ actions: read
+ pull-requests: read
+ steps:
+ - name: Harden the runner (Audit all outbound calls)
+ uses: step-security/harden-runner@bf7454d06d71f1098171f2acdf0cd4708d7b5920 # v2.20.0
+ 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/CHANGELOG.md b/CHANGELOG.md
index 7bf8ad766..9c44a9a16 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -8,6 +8,11 @@ Semantic Versioning where the repository publishes a release.
### Added
+- Added a read-only hourly GitHub Actions queue-health report for the central
+ repository, TEPP, naruon, and contextual-orchestrator, with exact pull-request
+ head classification, runner-assignment evidence, bounded queue-age SLO
+ warnings, deterministic JSON, accessible HTML, and no cancellation or merge
+ mutation.
- Added an hourly organization commercial-readiness coordinator that discovers writable repositories, honors enabled dedicated writer leases and fully paginated live writer runs, refetches exact repository/workflow/run/PR state before dispatch, rotates bounded review-repair and opt-in NVIDIA OpenCode product-development targets, fails nonzero on fleet-wide inspection or dispatch outages, retains three-day JSON receipts, and keeps the existing 15-minute merge scheduler authoritative.
- Added a dedicated OriginWeave hourly caller that invokes the product-neutral central scheduler with the exact repository, protected `main` branch, one-dispatch budget, two-hour same-head retry floor, non-cancelling single-flight heartbeat, job-scoped OIDC, and only the established scheduler credentials.
- Added a trusted pull-request comment router for `@cwl-noema-review` and review-only `@opencode-agent` dispatches, with an organization sweep, exact-head receipts, repository allowlisting, fixed runners, immutable checkout pins, and a permanent 100% statement/branch/docstring quality gate.
@@ -32,6 +37,9 @@ Semantic Versioning where the repository publishes a release.
### Fixed
+- Validate collector-produced canonical pull-request identity at report time and
+ retry one transient incomplete raw response before failing queue-health collection
+ with repository context; exact-head classification remains fail-closed.
- Parsed `opencode.jsonc` as JSONC (stripping `//` and `/* */` comments outside string literals) in the reasoning-effort guard and its contract tests, instead of raw `json.loads`, which rejected the file the moment it carried its first explanatory comment (added for the `contextual-orchestrator` provider block) with `Expecting property name enclosed in double quotes`. Comment markers inside string values, such as the `$schema` URL, are left untouched.
- Download the pinned `uv` 0.12.1 exporter from the official GitHub Releases URL instead of `releases.astral.sh`, which now returns HTTP 403 and blocks org-wide OpenCode `coverage-evidence`. The SHA-256 pin is unchanged. The opener may follow one hop onto `release-assets.githubusercontent.com` or `objects.githubusercontent.com` and still rejects every other host, userinfo, non-HTTPS scheme, and nondefault port (ContextualWisdomLab/.github#1109).
- Compared the trusted `uv` executable's post-install `--version` output against the real GitHub Releases build's full string, `uv 0.12.1 (x86_64-unknown-linux-gnu)`, instead of the bare `uv 0.12.1` the prior check required; the genuine release binary always prints the target triple, so every installation was failing the pin check immediately after the archive download itself was fixed (ContextualWisdomLab/.github#1109).
diff --git a/config/actions_queue_health_repositories.json b/config/actions_queue_health_repositories.json
new file mode 100644
index 000000000..a6b6f6ee5
--- /dev/null
+++ b/config/actions_queue_health_repositories.json
@@ -0,0 +1,8 @@
+{
+ "repositories": [
+ "ContextualWisdomLab/.github",
+ "ContextualWisdomLab/TEPP",
+ "ContextualWisdomLab/contextual-orchestrator",
+ "ContextualWisdomLab/naruon"
+ ]
+}
diff --git a/docs/doctoring/actions-queue-health.md b/docs/doctoring/actions-queue-health.md
new file mode 100644
index 000000000..7f7b6e87b
--- /dev/null
+++ b/docs/doctoring/actions-queue-health.md
@@ -0,0 +1,40 @@
+# 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), queue age, job state, and runner
+assignment. A run is `current_head` only when its linked open pull request and
+head SHA match. Stale linked runs are `obsolete`; runs without a pull-request
+link are `unlinked`. Queued evidence remains incomplete even when a report is
+successfully produced.
+
+Queued runs use run-level evidence because GitHub has not assigned their jobs;
+only current-head `in_progress` runs make the additional jobs API read needed
+to inspect a concrete runner assignment.
+
+List endpoints use GitHub CLI pagination with at most 20 pages of 100 records.
+An incomplete, malformed, or larger response fails closed instead of silently
+claiming that the visible page is the whole 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.
+
+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.
diff --git a/scripts/ci/actions_queue_health.py b/scripts/ci/actions_queue_health.py
new file mode 100644
index 000000000..3b825d91d
--- /dev/null
+++ b/scripts/ci/actions_queue_health.py
@@ -0,0 +1,612 @@
+#!/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 collections import Counter
+from datetime import datetime, timezone
+import html
+import json
+from pathlib import Path
+import re
+import subprocess
+import sys
+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"}
+TERMINAL_STATES = {"COMPLETED"}
+DEFAULT_QUEUE_AGE_SLO_SECONDS = 900
+SCHEMA_VERSION = "actions.queue_health.v1"
+MAX_API_PAGE_SIZE = 100
+MAX_API_PAGES = 20
+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."""
+
+
+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}")
+ 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) -> list[dict[str, Any]]:
+ """Extract a bounded GitHub list response without guessing its shape."""
+ 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 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_API_PAGE_SIZE * MAX_API_PAGES:
+ raise QueueHealthError(f"GitHub response field {key!r} exceeds the bounded page size")
+ return values
+
+
+def github_json(path: str, *, paginate: bool = False, 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}")
+ command = ["gh", "api"]
+ if paginate:
+ command.extend(("--paginate", "--slurp"))
+ command.append(path)
+ result = runner(
+ command,
+ capture_output=True,
+ text=True,
+ check=False,
+ )
+ if result.returncode != 0:
+ detail = (result.stderr or result.stdout or "GitHub API read failed").strip()
+ raise QueueHealthError(f"GitHub API read failed for {path}: {detail[:400]}")
+ try:
+ payload = json.loads(result.stdout)
+ except json.JSONDecodeError as exc:
+ raise QueueHealthError(f"GitHub API returned invalid JSON for {path}") from exc
+ if paginate:
+ if not isinstance(payload, list) or not payload or len(payload) > MAX_API_PAGES:
+ raise QueueHealthError(f"GitHub API returned an unbounded page set for {path}")
+ return {PAGINATED_PAGES_KEY: payload}
+ return payload
+
+
+def _normalise_pull_request(pull_request: dict[str, Any]) -> dict[str, Any]:
+ """Keep only exact-head identity fields needed for queue classification."""
+ if not isinstance(pull_request, dict):
+ raise QueueHealthError("pull request entry must be an object")
+ number = pull_request.get("number")
+ if isinstance(number, bool) or not isinstance(number, int) or number <= 0:
+ raise QueueHealthError("pull request number must be a positive integer")
+ if "head_sha" in pull_request or "base_ref" in pull_request:
+ canonical = {
+ "base_ref": pull_request.get("base_ref", ""),
+ "base_repository": pull_request.get("base_repository", ""),
+ "head_sha": pull_request.get("head_sha", ""),
+ "updated_at": pull_request.get("updated_at", ""),
+ }
+ if not all(isinstance(value, str) for value in canonical.values()):
+ raise QueueHealthError("normalized pull request identity fields must be strings")
+ return {
+ "number": number,
+ "state": pull_request.get("state", "open"),
+ **canonical,
+ }
+ head = pull_request.get("head")
+ base = pull_request.get("base")
+ if not isinstance(head, dict) or not isinstance(base, dict):
+ raise QueueHealthError("pull request head and base must be objects")
+ return {
+ "number": number,
+ "state": pull_request.get("state", "open"),
+ "base_ref": base.get("ref", ""),
+ "base_repository": (base.get("repo") or {}).get("full_name", "")
+ if isinstance(base.get("repo"), dict)
+ else "",
+ "head_sha": head.get("sha", ""),
+ "updated_at": pull_request.get("updated_at", ""),
+ }
+
+
+def _normalise_job(job: dict[str, Any]) -> dict[str, Any]:
+ """Keep job state and runner assignment evidence without log contents."""
+ 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
+ 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 ""),
+ "steps_count": len(job.get("steps") or []) if isinstance(job.get("steps"), list) else 0,
+ }
+
+
+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."""
+ 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")
+ 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]] = []
+ for repository in validated:
+ 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",
+ )
+ try:
+ normalized_pull_requests = sorted(
+ (_normalise_pull_request(item) for item in pull_requests),
+ key=lambda item: item["number"],
+ )
+ except QueueHealthError as exc:
+ if str(exc) != "pull request head and base must be objects":
+ raise QueueHealthError(
+ f"pull-request identity validation failed for {repository}: {exc}"
+ ) from exc
+ retry_pull_requests = _list_payload(
+ github_json(pulls_endpoint, paginate=True, runner=runner),
+ "pulls",
+ )
+ try:
+ normalized_pull_requests = sorted(
+ (_normalise_pull_request(item) for item in retry_pull_requests),
+ key=lambda item: item["number"],
+ )
+ except QueueHealthError as retry_exc:
+ raise QueueHealthError(
+ f"pull-request identity validation failed for {repository}: {retry_exc}"
+ ) from retry_exc
+ pull_requests_by_number = {item["number"]: item for item in normalized_pull_requests}
+ runs_by_id: dict[int, dict[str, Any]] = {}
+ for status in ("in_progress", "queued"):
+ runs = _list_payload(
+ github_json(
+ f"repos/{repository}/actions/runs?status={status}&per_page={MAX_API_PAGE_SIZE}",
+ paginate=True,
+ runner=runner,
+ ),
+ "workflow_runs",
+ )
+ 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")
+ if run_id in runs_by_id:
+ continue
+ candidate = _normalise_run(repository, run, [])
+ identity, _ = _run_identity(candidate, pull_requests_by_number)
+ if identity != "current_head" or candidate["status"] == "QUEUED":
+ 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")
+ runs_by_id[run_id] = _normalise_run(repository, run, jobs)
+ 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}
+
+
+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."""
+ 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") == run.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."""
+ 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 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")
+
+ rows: list[dict[str, Any]] = []
+ 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"))
+ 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)
+ 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)
+ age_seconds = _format_age(run.get("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 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,
+ "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["execution_state"].startswith("queued_")]
+ current_pending = [row for row in pending if row["identity_state"] == "current_head"]
+ lane_counts = Counter(
+ (row["repository"], row["pull_request_number"], row["workflow_name"])
+ for row in current_pending
+ if row["pull_request_number"] is not None
+ )
+ duplicate_lanes = [
+ {"repository": key[0], "pull_request_number": key[1], "workflow_name": key[2], "count": count}
+ for key, count in sorted(lane_counts.items())
+ if count > 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),
+ "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),
+ "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",
+ )
+ )
+ + "