feat(train): mark merges, write a heartbeat and report the train health

README.md

@@ -32,6 +32,9 @@ Status: in progress. The design is in `specs/next-ci-reliability` of
   logs once any workflow run that has been queued longer than `queue_alert_minutes` (default 30):
   a runner that is offline or saturated otherwise holds jobs for up to 24 hours before they fail.
   The runs already reported are remembered in `queue-alerts.json` beside the configuration.
+  After every merge it comments `<!-- merge-train:merged -->` on the pull request, so a merge made
+  around the train can be counted, and it writes a `heartbeat:` line (repositories, labelled pull
+  requests, version of the script) at most once an hour as proof of life.
   `--dry-run` changes nothing.
   Tests: `python3 -m unittest discover -s tests`.
 
@@ -45,7 +48,11 @@ Status: in progress. The design is in `specs/next-ci-reliability` of
   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`.
+  `py -3 report/ci_report.py --config train/repos.json --repo naindev-contrast3d/contrast-hub --days 7`.
+  For the repositories of `--config` (the shared train) the snapshot also counts the pull requests
+  merged outside the train since `--train-since`, and `--train-dir D:\Development\Tools\ci-train`
+  adds a section on the installed train: age of its log (silent after 90 minutes) and whether its
+  script differs from the one in `ci-shared` main.
   `--until 2026-10-14T11:00` ends the period in the past, so a baseline can be reproduced after later
   changes; `--limit` raises the number of runs read per repository (default 300).
 

report/ci_report.py

@@ -17,6 +17,7 @@
 
 import argparse
 import datetime as dt
+import hashlib
 import importlib.util
 import json
 import statistics
@@ -28,6 +29,14 @@
 
 RUN_FIELDS = "databaseId,conclusion,status,workflowName,createdAt,startedAt,updatedAt,event,headBranch"
 INFRA_STEPS = ("Set up job", "Complete job")
+# The comment the merge train writes on every pull request it merges (train/automerge.py).
+TRAIN_MARK = "<!-- merge-train:merged -->"
+TRAIN_SILENT_AFTER_MIN = 90  # the train writes a heartbeat every 60 minutes
+
+
+def text_version(text: str) -> str:
+    """A short fingerprint of a script, the same for LF and CRLF copies (as in the train)."""
+    return hashlib.sha256(text.replace("\r\n", "\n").encode("utf-8")).hexdigest()[:12]
 
 
 def run_gh(command: list[str]) -> str:
@@ -91,6 +100,38 @@ def queued_runs(self, repo: str) -> list[dict]:
         )
         return json.loads(out or "[]")
 
+    def merged_prs(self, repo: str, limit: int) -> list[dict]:
+        out = self._run(
+            [
+                "gh",
+                "pr",
+                "list",
+                "--repo",
+                repo,
+                "--state",
+                "merged",
+                "--limit",
+                str(limit),
+                "--json",
+                "number,mergedAt,comments",
+            ]
+        )
+        return json.loads(out or "[]")
+
+    def file_text(self, repo: str, path: str) -> str | None:
+        try:
+            return self._run(
+                [
+                    "gh",
+                    "api",
+                    "-H",
+                    "Accept: application/vnd.github.raw+json",
+                    f"repos/{repo}/contents/{path}",
+                ]
+            )
+        except subprocess.CalledProcessError:
+            return None
+
     def open_prs(self, repo: str) -> list[dict]:
         out = self._run(
             [
@@ -126,6 +167,9 @@ class RepoStats:
     oldest_queued_min: int | None = None
     open_prs: int = 0
     automerge_prs: int = 0
+    # Only for repositories served by the shared merge train; None elsewhere.
+    merged_total: int | None = None
+    merged_by_hand: int | None = None
 
     @property
     def fail_rate(self) -> float | None:
@@ -172,6 +216,8 @@ def collect(
     sample: int = 30,
     until: dt.datetime | None = None,
     limit: int = 300,
+    train_repos: frozenset[str] | set[str] = frozenset(),
+    train_since: dt.datetime | None = None,
 ) -> RepoStats:
     """Metrics for runs created in the `days` before `until` (default `now`).
 
@@ -237,9 +283,89 @@ def collect(
         for p in prs
         if "automerge" in {label["name"] for label in p.get("labels", [])}
     )
+
+    if repo in train_repos:
+        floor = max(since, train_since) if train_since else since
+        merged = [
+            m for m in gh.merged_prs(repo, 200) if floor <= _parse(m["mergedAt"]) <= end
+        ]
+        stats.merged_total = len(merged)
+        stats.merged_by_hand = sum(
+            1
+            for m in merged
+            if not any(
+                TRAIN_MARK in (c.get("body") or "") for c in m.get("comments", [])
+            )
+        )
     return stats
 
 
+@dataclass
+class TrainHealth:
+    log_age_min: (
+        int | None
+    )  # minutes since the log was last written; None without a log
+    silent: bool  # no log line for too long
+    copy: str  # current, stale, missing or unknown
+    installed_version: str | None = None
+    repository_version: str | None = None
+
+
+def train_health(
+    train_dir: Path,
+    gh,
+    now: dt.datetime,
+    repo: str = "naindev-contrast3d/ci-shared",
+) -> TrainHealth:
+    """Is the installed train alive and is it the script of `ci-shared` main?"""
+    train_dir = Path(train_dir)
+    script = train_dir / "automerge.py"
+    log = train_dir / "train.log"
+
+    age = None
+    if log.exists():
+        modified = dt.datetime.fromtimestamp(log.stat().st_mtime, dt.UTC)
+        age = int((now - modified).total_seconds() // 60)
+    silent = age is not None and age > TRAIN_SILENT_AFTER_MIN
+
+    if not script.exists():
+        return TrainHealth(age, silent, "missing")
+    installed = text_version(script.read_text(encoding="utf-8"))
+    published = gh.file_text(repo, "train/automerge.py")
+    if published is None:
+        return TrainHealth(age, silent, "unknown", installed_version=installed)
+    current = text_version(published)
+    return TrainHealth(
+        age, silent, "current" if installed == current else "stale", installed, current
+    )
+
+
+def render_train(health: TrainHealth) -> str:
+    lines = ["## Merge train", ""]
+    if health.copy == "missing":
+        lines.append("- Installed copy not found (the train is not installed here).")
+        return "\n".join(lines)
+    if health.log_age_min is None:
+        lines.append("- No log yet.")
+    elif health.silent:
+        lines.append(
+            f"- Warning: no log line for {health.log_age_min} min "
+            f"(the train writes a heartbeat every 60 minutes)."
+        )
+    else:
+        lines.append(f"- Last log activity {health.log_age_min} min ago.")
+    if health.copy == "current":
+        lines.append("- The installed script matches ci-shared main.")
+    elif health.copy == "stale":
+        lines.append(
+            f"- Warning: the installed script differs from ci-shared main "
+            f"(installed {health.installed_version}, repository {health.repository_version})."
+        )
+    else:
+        lines.append("- Could not read the script of ci-shared main to compare.")
+    return "\n".join(lines)
+
+
 def fmt_duration(seconds: float | None) -> str:
     if seconds is None:
         return "-"
@@ -302,6 +428,10 @@ def render(
         lines.append(
             f"- Open pull requests: {s.open_prs} ({s.automerge_prs} labelled automerge)"
         )
+        if s.merged_by_hand is not None:
+            lines.append(
+                f"- Merged outside the train: {s.merged_by_hand} of {s.merged_total}"
+            )
         lines.append("")
     return "\n".join(lines)
 
@@ -348,25 +478,49 @@ def main(argv: list[str] | None = None) -> int:
         action="store_true",
         help="append the standards audit of the organisation",
     )
+    parser.add_argument(
+        "--train-dir",
+        help="folder of the installed merge train (automerge.py and train.log): adds its health",
+    )
+    parser.add_argument(
+        "--train-since",
+        help="UTC date the shared train started serving the repositories of --config; "
+        "earlier merges are not counted as made around it",
+    )
     parser.add_argument("--output", help="write the report here instead of printing it")
     args = parser.parse_args(argv)
 
     repos = list(args.repo)
+    train_repos: set[str] = set()
     if args.config:
         data = json.loads(Path(args.config).read_text(encoding="utf-8"))
-        repos += [entry["repo"] for entry in data.get("repos", [])]
+        train_repos = {entry["repo"] for entry in data.get("repos", [])}
+        repos += sorted(train_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()
     until = parse_moment(args.until) if args.until else None
+    train_since = parse_moment(args.train_since) if args.train_since else None
     stats = [
-        collect(gh, repo, args.workflow, args.days, now, until=until, limit=args.limit)
+        collect(
+            gh,
+            repo,
+            args.workflow,
+            args.days,
+            now,
+            until=until,
+            limit=args.limit,
+            train_repos=train_repos,
+            train_since=train_since,
+        )
         for repo in repos
     ]
     text = render(stats, args.days, now, until=until)
+    if args.train_dir:
+        text += "\n\n" + render_train(train_health(Path(args.train_dir), gh, now))
     if args.audit:
         audit = load_audit()
         standard_path = Path(audit.__file__).with_name("standards.json")

tests/test_automerge.py

@@ -50,7 +50,10 @@ def __init__(
         merge_error=False,
         queue=None,
         queue_error=False,
+        comment_error=False,
     ):
+        self.comment_error = comment_error
+        self.comments = []
         self.prs = prs or []
         self.queue = queue or {}
         self.queue_error = queue_error
@@ -90,6 +93,11 @@ def merge(self, repo, number, sha, method):
         if self.merge_error:
             raise subprocess.CalledProcessError(1, ["gh"], stderr="not allowed")
 
+    def comment(self, repo, number, body):
+        if self.comment_error:
+            raise subprocess.CalledProcessError(1, ["gh"], stderr="comment boom")
+        self.comments.append((repo, number, body))
+
     def update_branch(self, repo, number, sha):
         self.calls.append(("update", repo, number, sha))
         return self.update_result
@@ -101,19 +109,95 @@ def mutations(self):
         return [c for c in self.calls if c[0] != "open_prs"]
 
 
-def run(gh, repos=None, dry_run=False, alerts=None, now=None):
+def run(gh, repos=None, dry_run=False, alerts=None, now=None, heartbeat=None):
     logs = []
     settings = train.Settings()
     configs = repos or [train.RepoConfig(repo=REPO)]
     code = train.run_cycle(
-        gh, configs, settings, dry_run=dry_run, log=logs.append, alerts=alerts, now=now
+        gh,
+        configs,
+        settings,
+        dry_run=dry_run,
+        log=logs.append,
+        alerts=alerts,
+        now=now,
+        heartbeat=heartbeat,
     )
     return code, logs
 
 
 NOW = dt.datetime(2026, 10, 14, 13, 0, 0, tzinfo=dt.UTC)
 
 
+class MergeMark(unittest.TestCase):
+    """The train marks what it merges, so a merge made around it can be counted (FR13)."""
+
+    def test_a_merged_pull_request_gets_the_train_mark_with_its_head(self):
+        gh = FakeGithub(prs=[pr(7)])
+        run(gh)
+        self.assertEqual(len(gh.comments), 1)
+        repo, number, body = gh.comments[0]
+        self.assertEqual((repo, number), (REPO, 7))
+        self.assertIn(train.MERGE_MARK, body)
+        self.assertIn(SHA[:12], body)
+
+    def test_nothing_is_marked_in_a_dry_run_or_when_the_merge_is_refused(self):
+        dry = FakeGithub(prs=[pr(7)])
+        run(dry, dry_run=True)
+        refused = FakeGithub(prs=[pr(7)], merge_error=True)
+        run(refused)
+        self.assertEqual((dry.comments, refused.comments), ([], []))
+
+    def test_a_failing_comment_is_logged_and_does_not_undo_or_fail_the_merge(self):
+        gh = FakeGithub(prs=[pr(7)], comment_error=True)
+        code, logs = run(gh)
+        self.assertEqual(code, 0)
+        self.assertEqual(gh.mutations(), [("merge", REPO, 7, SHA, "merge")])
+        self.assertTrue(any("mark failed" in line for line in logs))
+
+
+class HeartbeatLine(unittest.TestCase):
+    """A line at least once an hour proves the train is alive and says which script it runs (FR14)."""
+
+    def state(self):
+        return Path(tempfile.mkdtemp()) / "heartbeat.json"
+
+    def test_the_first_cycle_logs_a_heartbeat_with_counts_and_the_script_version(self):
+        gh = FakeGithub(prs=[pr(1), pr(2, "b" * 40)], behind=2)
+        _, logs = run(gh, heartbeat=train.Heartbeat(self.state()), now=NOW)
+        line = next(item for item in logs if "heartbeat" in item)
+        self.assertIn("1 repositories", line)
+        self.assertIn("2 labelled pull requests", line)
+        self.assertIn(train.script_version(), line)
+
+    def test_no_second_heartbeat_within_the_hour_but_one_after_it(self):
+        path = self.state()
+        gh = FakeGithub(prs=[])
+        _, first = run(gh, heartbeat=train.Heartbeat(path), now=NOW)
+        _, soon = run(
+            gh, heartbeat=train.Heartbeat(path), now=NOW + dt.timedelta(minutes=59)
+        )
+        _, later = run(
+            gh, heartbeat=train.Heartbeat(path), now=NOW + dt.timedelta(minutes=61)
+        )
+        counts = [
+            sum("heartbeat" in item for item in logs) for logs in (first, soon, later)
+        ]
+        self.assertEqual(counts, [1, 0, 1])
+
+    def test_a_dry_run_does_not_write_the_heartbeat_state(self):
+        path = self.state()
+        run(FakeGithub(prs=[]), dry_run=True, heartbeat=train.Heartbeat(path), now=NOW)
+        self.assertFalse(path.exists())
+
+    def test_the_script_version_is_the_same_for_lf_and_crlf_copies(self):
+        text = "a\nb\n"
+        self.assertEqual(
+            train.text_version(text), train.text_version(text.replace("\n", "\r\n"))
+        )
+        self.assertNotEqual(train.text_version(text), train.text_version(text + "c\n"))
+
+
 def queued(run_id, minutes_ago, branch="feat/x", workflow="CI"):
     created = NOW - dt.timedelta(minutes=minutes_ago)
     return {
@@ -440,6 +524,13 @@ def test_merge_pins_the_head_commit_and_makes_a_merge_commit(self):
             ],
         )
 
+    def test_comment_posts_the_body_on_the_pull_request(self):
+        self.gh.comment(REPO, 9, f"{train.MERGE_MARK}\nMerged")
+        self.assertEqual(
+            self.commands[0][:7], ["gh", "pr", "comment", "9", "--repo", REPO, "--body"]
+        )
+        self.assertIn(train.MERGE_MARK, self.commands[0][7])
+
     def test_queued_runs_asks_for_queued_status_and_parses_json(self):
         self.outputs["run list"] = json.dumps([queued(5, 40)])
         runs = self.gh.queued_runs(REPO)