From 177bf1f0c8b2b4103a2daab5c5998598ef1ada7b Mon Sep 17 00:00:00 2001
From: naindev <193035803+Nain9Dev@users.noreply.github.com>
Date: Wed, 14 Oct 2026 16:31:10 +0200
Subject: [PATCH] feat(report): add the CI health report

Print, per repository and period, the failure rate of conclusive runs,
the share of failures caused by the infrastructure rather than the code,
the median duration and the median queue time, followed by a snapshot of
today: runners, queued runs and open pull requests. A failed run counts
as infrastructure only when every failed job never reached its own steps
or is the shared preflight. Cancelled runs are not failures.
---
 README.md               |   9 +-
 report/ci_report.py     | 310 ++++++++++++++++++++++++++++++++++++++++
 tests/test_ci_report.py | 255 +++++++++++++++++++++++++++++++++
 3 files changed, 571 insertions(+), 3 deletions(-)
 create mode 100644 report/ci_report.py
 create mode 100644 tests/test_ci_report.py

diff --git a/README.md b/README.md
index 6eb468f..5529e1f 100644
--- a/README.md
+++ b/README.md
@@ -40,9 +40,12 @@ Status: in progress. The design is in `specs/next-ci-reliability` of
   the example only if missing, and `train.log`). Run it from a normal PowerShell window of the owner's
   session; `-Remove` retires the task.
 
-## Planned contents
-
-- The weekly CI health report (task T16 of the spec).
+- `report/ci_report.py` (available): the CI health report. Per repository and period it prints the
+  failure rate of conclusive runs, the share of failures caused by the infrastructure (a job that
+  never reached its steps, or the shared preflight) and not the code, the median duration and the
+  median queue time, followed by a snapshot of today: runners online, offline and busy, queued runs
+  and open pull requests. Use it to plan from the current state, not from memory:
+  `py -3 report/ci_report.py --config train/repos.json --repo Nain9Dev/crm-platform --days 7`.
 
 ## Rules
 
diff --git a/report/ci_report.py b/report/ci_report.py
new file mode 100644
index 0000000..6e67312
--- /dev/null
+++ b/report/ci_report.py
@@ -0,0 +1,310 @@
+"""Weekly CI health report: failure rate by cause, durations, queue time and a snapshot of today.
+
+    py -3 ci_report.py --repo owner/name --repo owner/other --days 7
+    py -3 ci_report.py --config ..\\train\\repos.json --days 30 --output report.md
+
+Per repository, over the period: conclusive runs (success or failure) of one workflow, how many
+failed, which share of those failures was the infrastructure and not the code, the median duration
+and the median time a run waited for its first job to start. Cancelled runs are counted apart, never
+as failures. The snapshot reports what is true now (runners, queue, open pull requests), because a
+plan made from remembered state is how stale diagnoses happen.
+
+A failed run counts as infrastructure only when every failed job never reached its own steps
+(`Set up job`, a job with no steps) or is the shared runner preflight. Anything else is code.
+"""
+
+from __future__ import annotations
+
+import argparse
+import datetime as dt
+import json
+import statistics
+import subprocess
+import sys
+from collections.abc import Callable
+from dataclasses import dataclass
+from pathlib import Path
+
+RUN_FIELDS = "databaseId,conclusion,status,workflowName,createdAt,startedAt,updatedAt,event,headBranch"
+INFRA_STEPS = ("Set up job", "Complete job")
+
+
+def run_gh(command: list[str]) -> str:
+    # Every command is built in this file from fixed words and GitHub's own numbers.
+    no_window = getattr(subprocess, "CREATE_NO_WINDOW", 0)
+    return subprocess.run(
+        command, check=True, capture_output=True, text=True, creationflags=no_window
+    ).stdout
+
+
+class Github:
+    """The few GitHub reads the report needs, over the `gh` command line."""
+
+    def __init__(self, run: Callable[[list[str]], str] = run_gh):
+        self._run = run
+
+    def runs(self, repo: str, limit: int) -> list[dict]:
+        out = self._run(
+            [
+                "gh",
+                "run",
+                "list",
+                "--repo",
+                repo,
+                "--limit",
+                str(limit),
+                "--json",
+                RUN_FIELDS,
+            ]
+        )
+        return json.loads(out or "[]")
+
+    def jobs(self, repo: str, run_id: int) -> list[dict]:
+        out = self._run(
+            ["gh", "run", "view", str(run_id), "--repo", repo, "--json", "jobs"]
+        )
+        return json.loads(out or "{}").get("jobs", [])
+
+    def runners(self, repo: str) -> list[dict]:
+        try:
+            out = self._run(["gh", "api", f"repos/{repo}/actions/runners"])
+        except subprocess.CalledProcessError:
+            return []  # no permission to list runners, or none registered
+        return json.loads(out or "{}").get("runners", [])
+
+    def queued_runs(self, repo: str) -> list[dict]:
+        out = self._run(
+            [
+                "gh",
+                "run",
+                "list",
+                "--repo",
+                repo,
+                "--status",
+                "queued",
+                "--json",
+                "databaseId,createdAt",
+                "--limit",
+                "50",
+            ]
+        )
+        return json.loads(out or "[]")
+
+    def open_prs(self, repo: str) -> list[dict]:
+        out = self._run(
+            [
+                "gh",
+                "pr",
+                "list",
+                "--repo",
+                repo,
+                "--state",
+                "open",
+                "--json",
+                "number,labels,createdAt",
+                "--limit",
+                "100",
+            ]
+        )
+        return json.loads(out or "[]")
+
+
+@dataclass
+class RepoStats:
+    repo: str
+    conclusive: int = 0
+    failed: int = 0
+    cancelled: int = 0
+    infra_failures: int = 0
+    median_duration_s: float | None = None
+    median_queue_s: float | None = None
+    runners_online: int = 0
+    runners_busy: int = 0
+    runners_offline: int = 0
+    queued_now: int = 0
+    oldest_queued_min: int | None = None
+    open_prs: int = 0
+    automerge_prs: int = 0
+
+    @property
+    def fail_rate(self) -> float | None:
+        return self.failed / self.conclusive if self.conclusive else None
+
+    @property
+    def infra_share(self) -> float | None:
+        return self.infra_failures / self.failed if self.failed else None
+
+
+def classify_failure(jobs: list[dict]) -> str:
+    """'infra' when every failed job never reached its own steps, otherwise 'code'."""
+    failed = [j for j in jobs if j.get("conclusion") == "failure"]
+    if not failed:
+        return "code"
+    return "infra" if all(_is_infra(j) for j in failed) else "code"
+
+
+def _is_infra(job: dict) -> bool:
+    if "preflight" in (job.get("name") or "").lower():
+        return True
+    failed_steps = [
+        s["name"] for s in job.get("steps", []) if s.get("conclusion") == "failure"
+    ]
+    return not failed_steps or failed_steps[0] in INFRA_STEPS
+
+
+def _parse(moment: str) -> dt.datetime:
+    return dt.datetime.fromisoformat(moment)
+
+
+def collect(
+    gh, repo: str, workflow: str, days: int, now: dt.datetime, sample: int = 30
+) -> RepoStats:
+    since = now - dt.timedelta(days=days)
+    runs = [
+        r
+        for r in gh.runs(repo, 300)
+        if r.get("workflowName") == workflow and _parse(r["createdAt"]) >= since
+    ]
+    conclusive = [r for r in runs if r.get("conclusion") in ("success", "failure")]
+    stats = RepoStats(repo=repo)
+    stats.conclusive = len(conclusive)
+    stats.cancelled = sum(1 for r in runs if r.get("conclusion") == "cancelled")
+
+    for r in conclusive:
+        if r["conclusion"] != "failure":
+            continue
+        stats.failed += 1
+        if classify_failure(gh.jobs(repo, r["databaseId"])) == "infra":
+            stats.infra_failures += 1
+
+    durations = [
+        (_parse(r["updatedAt"]) - _parse(r["startedAt"])).total_seconds()
+        for r in conclusive
+    ]
+    if durations:
+        stats.median_duration_s = statistics.median(durations)
+
+    waits = []
+    for r in sorted(conclusive, key=lambda item: item["createdAt"], reverse=True)[
+        :sample
+    ]:
+        starts = [
+            j["startedAt"] for j in gh.jobs(repo, r["databaseId"]) if j.get("startedAt")
+        ]
+        if starts:
+            first = min(_parse(s) for s in starts)
+            waits.append((first - _parse(r["createdAt"])).total_seconds())
+    if waits:
+        stats.median_queue_s = statistics.median(waits)
+
+    for runner in gh.runners(repo):
+        if runner.get("status") == "online":
+            stats.runners_online += 1
+            stats.runners_busy += 1 if runner.get("busy") else 0
+        else:
+            stats.runners_offline += 1
+
+    queue = gh.queued_runs(repo)
+    stats.queued_now = len(queue)
+    if queue:
+        stats.oldest_queued_min = int(
+            max((now - _parse(q["createdAt"])).total_seconds() for q in queue) // 60
+        )
+
+    prs = gh.open_prs(repo)
+    stats.open_prs = len(prs)
+    stats.automerge_prs = sum(
+        1
+        for p in prs
+        if "automerge" in {label["name"] for label in p.get("labels", [])}
+    )
+    return stats
+
+
+def fmt_duration(seconds: float | None) -> str:
+    if seconds is None:
+        return "-"
+    total = round(seconds)
+    if total < 60:
+        return f"{total}s"
+    if total < 3600:
+        return f"{total // 60}m{total % 60:02d}s"
+    return f"{total // 3600}h{(total % 3600) // 60:02d}m"
+
+
+def _pct(value: float | None) -> str:
+    return "-" if value is None else f"{round(value * 100)}%"
+
+
+def render(stats: list[RepoStats], days: int, now: dt.datetime) -> str:
+    since = (now - dt.timedelta(days=days)).strftime("%Y-%m-%d")
+    lines = [
+        "# CI health report",
+        "",
+        (
+            f"Period: {since} to {now.strftime('%Y-%m-%d')} ({days} days). "
+            "Conclusive runs only; cancelled runs are not failures."
+        ),
+        "",
+        "| Repository | Runs | Failed | Fail rate | Infra share | Median duration | Median queue |",
+        "| --- | ---: | ---: | ---: | ---: | ---: | ---: |",
+    ]
+    for s in stats:
+        lines.append(
+            f"| {s.repo} | {s.conclusive} | {s.failed} | {_pct(s.fail_rate)} | {_pct(s.infra_share)}"
+            f" | {fmt_duration(s.median_duration_s)} | {fmt_duration(s.median_queue_s)} |"
+        )
+    lines += ["", "## Snapshot", ""]
+    for s in stats:
+        lines.append(f"### {s.repo}")
+        if s.runners_online or s.runners_offline:
+            lines.append(
+                f"- Runners: {s.runners_online} online ({s.runners_busy} busy), {s.runners_offline} offline"
+            )
+        else:
+            lines.append("- Runners: none registered (GitHub-hosted)")
+        oldest = (
+            f" (oldest {s.oldest_queued_min} min)"
+            if s.oldest_queued_min is not None
+            else ""
+        )
+        lines.append(f"- Queued now: {s.queued_now}{oldest}")
+        lines.append(
+            f"- Open pull requests: {s.open_prs} ({s.automerge_prs} labelled automerge)"
+        )
+        lines.append("")
+    return "\n".join(lines)
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
+    parser.add_argument(
+        "--repo", action="append", default=[], help="owner/name; repeatable"
+    )
+    parser.add_argument("--config", help="train repos.json: its repositories are added")
+    parser.add_argument("--days", type=int, default=7)
+    parser.add_argument("--workflow", default="CI")
+    parser.add_argument("--output", help="write the report here instead of printing it")
+    args = parser.parse_args(argv)
+
+    repos = list(args.repo)
+    if args.config:
+        data = json.loads(Path(args.config).read_text(encoding="utf-8"))
+        repos += [entry["repo"] for entry in data.get("repos", [])]
+    if not repos:
+        print("give at least one --repo or a --config", file=sys.stderr)
+        return 2
+
+    now = dt.datetime.now(dt.UTC)
+    gh = Github()
+    stats = [collect(gh, repo, args.workflow, args.days, now) for repo in repos]
+    text = render(stats, args.days, now)
+    if args.output:
+        Path(args.output).write_text(text + "\n", encoding="utf-8")
+    else:
+        print(text)
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())
diff --git a/tests/test_ci_report.py b/tests/test_ci_report.py
new file mode 100644
index 0000000..c251e95
--- /dev/null
+++ b/tests/test_ci_report.py
@@ -0,0 +1,255 @@
+"""Tests for report/ci_report.py: failure classification, the metrics and the rendered report."""
+
+from __future__ import annotations
+
+import datetime as dt
+import json
+import subprocess
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "report"))
+
+import ci_report as report
+
+REPO = "naindev-contrast3d/example"
+NOW = dt.datetime(2026, 10, 14, 12, 0, 0, tzinfo=dt.UTC)
+
+
+def iso(minutes_ago: float, seconds: int = 0) -> str:
+    moment = NOW - dt.timedelta(minutes=minutes_ago) + dt.timedelta(seconds=seconds)
+    return moment.strftime("%Y-%m-%dT%H:%M:%SZ")
+
+
+def run(run_id, conclusion, created_min_ago, duration_s, name="CI", status="completed"):
+    return {
+        "databaseId": run_id,
+        "conclusion": conclusion,
+        "status": status,
+        "workflowName": name,
+        "createdAt": iso(created_min_ago),
+        "startedAt": iso(created_min_ago),
+        "updatedAt": iso(created_min_ago, duration_s),
+        "event": "pull_request",
+        "headBranch": "feat/x",
+    }
+
+
+def job(name, conclusion, started_after_s, failed_step=None):
+    steps = [{"name": failed_step, "conclusion": "failure"}] if failed_step else []
+    return {
+        "name": name,
+        "conclusion": conclusion,
+        "startedAt": started_after_s,
+        "steps": steps,
+    }
+
+
+class FakeGithub:
+    def __init__(self, runs, jobs, runners=None, queued=None, prs=None):
+        self._runs = runs
+        self._jobs = jobs  # run id -> list of (name, conclusion, delay_s, failed_step)
+        self._runners = runners if runners is not None else []
+        self._queued = queued or []
+        self._prs = prs or []
+
+    def runs(self, repo, limit):
+        return self._runs
+
+    def jobs(self, repo, run_id):
+        created = next(r for r in self._runs if r["databaseId"] == run_id)["createdAt"]
+        base = dt.datetime.fromisoformat(created)
+        out = []
+        for name, conclusion, delay, failed_step in self._jobs.get(run_id, []):
+            started = (base + dt.timedelta(seconds=delay)).strftime(
+                "%Y-%m-%dT%H:%M:%SZ"
+            )
+            out.append(job(name, conclusion, started, failed_step))
+        return out
+
+    def runners(self, repo):
+        return self._runners
+
+    def queued_runs(self, repo):
+        return self._queued
+
+    def open_prs(self, repo):
+        return self._prs
+
+
+def sample_github():
+    runs = [
+        run(1, "success", 60, 300),
+        run(2, "success", 120, 600),
+        run(3, "failure", 180, 120),  # a real test failure
+        run(4, "failure", 240, 60),  # the runner could not start the job
+        run(5, "cancelled", 300, 30),
+        run(6, "skipped", 360, 0),
+        run(7, "success", 60 * 24 * 10, 300),  # older than the window
+        run(8, "success", 70, 100, name="Other"),  # another workflow
+    ]
+    jobs = {
+        1: [("Verify", "success", 30, None)],
+        2: [("Verify", "success", 10, None)],
+        3: [("Verify", "success", 90, None), ("Tests", "failure", 95, "Pruebas")],
+        4: [("Verify", "failure", 200, "Set up job")],
+    }
+    runners = [
+        {"name": "vm-01", "status": "online", "busy": True},
+        {"name": "vm-02", "status": "offline", "busy": False},
+    ]
+    queued = [
+        {"databaseId": 90, "createdAt": iso(45)},
+        {"databaseId": 91, "createdAt": iso(5)},
+    ]
+    prs = [
+        {"number": 1, "labels": [{"name": "automerge"}], "createdAt": iso(60 * 5)},
+        {"number": 2, "labels": [], "createdAt": iso(60)},
+    ]
+    return FakeGithub(runs, jobs, runners, queued, prs)
+
+
+class Classification(unittest.TestCase):
+    def test_a_job_that_never_started_its_steps_is_infrastructure(self):
+        jobs = [job("Verify", "failure", iso(1), "Set up job")]
+        self.assertEqual(report.classify_failure(jobs), "infra")
+
+    def test_a_failure_with_no_steps_at_all_is_infrastructure(self):
+        self.assertEqual(
+            report.classify_failure([job("Verify", "failure", iso(1))]), "infra"
+        )
+
+    def test_a_failed_test_step_is_code(self):
+        jobs = [job("Tests", "failure", iso(1), "Pruebas")]
+        self.assertEqual(report.classify_failure(jobs), "code")
+
+    def test_the_shared_preflight_failing_is_infrastructure(self):
+        jobs = [
+            job("Runner preflight", "failure", iso(1), "Check tools, memory and disk")
+        ]
+        self.assertEqual(report.classify_failure(jobs), "infra")
+
+    def test_a_run_is_infrastructure_only_when_every_failed_job_is(self):
+        jobs = [
+            job("Verify", "failure", iso(1), "Set up job"),
+            job("Tests", "failure", iso(1), "Pruebas"),
+        ]
+        self.assertEqual(report.classify_failure(jobs), "code")
+
+    def test_no_failed_job_is_reported_as_code(self):
+        self.assertEqual(
+            report.classify_failure([job("Verify", "success", iso(1))]), "code"
+        )
+
+
+class Metrics(unittest.TestCase):
+    def setUp(self):
+        self.stats = report.collect(sample_github(), REPO, "CI", days=7, now=NOW)
+
+    def test_only_conclusive_runs_of_the_workflow_inside_the_window_are_counted(self):
+        self.assertEqual(self.stats.conclusive, 4)
+        self.assertEqual(self.stats.failed, 2)
+        self.assertEqual(self.stats.cancelled, 1)
+
+    def test_failures_are_split_by_cause(self):
+        self.assertEqual(self.stats.infra_failures, 1)
+
+    def test_the_median_duration_is_over_conclusive_runs(self):
+        self.assertEqual(self.stats.median_duration_s, 210)  # 60, 120, 300, 600
+
+    def test_the_median_queue_time_is_the_first_job_start_after_the_run_was_created(
+        self,
+    ):
+        self.assertEqual(self.stats.median_queue_s, 60)  # 10, 30, 90, 200
+
+    def test_the_snapshot_reports_runners_the_queue_and_open_pull_requests(self):
+        self.assertEqual((self.stats.runners_online, self.stats.runners_busy), (1, 1))
+        self.assertEqual(self.stats.runners_offline, 1)
+        self.assertEqual(self.stats.queued_now, 2)
+        self.assertEqual(self.stats.oldest_queued_min, 45)
+        self.assertEqual((self.stats.open_prs, self.stats.automerge_prs), (2, 1))
+
+    def test_a_repository_without_runs_has_no_rates_instead_of_dividing_by_zero(self):
+        stats = report.collect(FakeGithub([], {}), REPO, "CI", days=7, now=NOW)
+        self.assertEqual(stats.conclusive, 0)
+        self.assertIsNone(stats.fail_rate)
+        self.assertIsNone(stats.median_duration_s)
+        self.assertIsNone(stats.median_queue_s)
+
+
+class Rendering(unittest.TestCase):
+    def test_the_table_row_is_pinned(self):
+        stats = report.collect(sample_github(), REPO, "CI", days=7, now=NOW)
+        text = report.render([stats], days=7, now=NOW)
+        self.assertIn(
+            f"| {REPO} | 4 | 2 | 50% | 50% | 3m30s | 1m00s |",
+            text,
+        )
+
+    def test_the_snapshot_lines_are_present(self):
+        stats = report.collect(sample_github(), REPO, "CI", days=7, now=NOW)
+        text = report.render([stats], days=7, now=NOW)
+        self.assertIn("Runners: 1 online (1 busy), 1 offline", text)
+        self.assertIn("Queued now: 2 (oldest 45 min)", text)
+        self.assertIn("Open pull requests: 2 (1 labelled automerge)", text)
+
+    def test_a_repository_without_data_renders_dashes(self):
+        stats = report.collect(FakeGithub([], {}), REPO, "CI", days=7, now=NOW)
+        text = report.render([stats], days=7, now=NOW)
+        self.assertIn(f"| {REPO} | 0 | 0 | - | - | - | - |", text)
+
+    def test_the_header_states_the_period(self):
+        text = report.render([], days=7, now=NOW)
+        self.assertIn("2026-10-07", text)
+        self.assertIn("2026-10-14", text)
+
+    def test_durations_are_formatted_for_people(self):
+        self.assertEqual(report.fmt_duration(45), "45s")
+        self.assertEqual(report.fmt_duration(210), "3m30s")
+        self.assertEqual(report.fmt_duration(3720), "1h02m")
+
+
+class RealCommands(unittest.TestCase):
+    def setUp(self):
+        self.commands = []
+        self.outputs = {}
+        self.gh = report.Github(self.fake_run)
+
+    def fake_run(self, command):
+        self.commands.append(command)
+        key = " ".join(command)
+        for needle, output in self.outputs.items():
+            if needle in key:
+                return output
+        return "[]"
+
+    def test_runs_asks_for_the_fields_the_metrics_need(self):
+        self.gh.runs(REPO, 200)
+        command = self.commands[0]
+        self.assertEqual(command[:4], ["gh", "run", "list", "--repo"])
+        fields = (
+            "databaseId,conclusion,status,workflowName,createdAt,startedAt,updatedAt"
+        )
+        self.assertTrue(any(fields in part for part in command))
+
+    def test_jobs_reads_the_jobs_of_one_run(self):
+        self.outputs["run view"] = json.dumps(
+            {"jobs": [job("Verify", "success", iso(1))]}
+        )
+        self.assertEqual(self.gh.jobs(REPO, 7)[0]["name"], "Verify")
+        self.assertIn("7", self.commands[0])
+
+    def test_runners_reads_the_repository_runners(self):
+        self.outputs["actions/runners"] = json.dumps({"runners": [{"name": "vm-01"}]})
+        self.assertEqual(self.gh.runners(REPO)[0]["name"], "vm-01")
+
+    def test_a_repository_without_runner_access_reports_none_instead_of_failing(self):
+        def refuse(command):
+            raise subprocess.CalledProcessError(1, command, stderr="HTTP 403")
+
+        self.assertEqual(report.Github(refuse).runners(REPO), [])
+
+
+if __name__ == "__main__":
+    unittest.main()