ci: live-updating PR review comment with structured job statuses

Replace the static comment-pending + comment-results two-job pattern
with a live-updating comment system that polls the GitHub Actions API
every 15s, re-assembles the review comment from whatever results are
available, and upserts it via the <!-- hermes-ci-review-bot --> marker.
The comment updates in real time as each job finishes — no waiting for
the full pipeline.

Every CI job that wants to appear in the review comment emits a
review_status output — a JSON array of objects, each with a source
and a results array:

    [
      {
        "source": "review-label-gate",
        "results": [
          {"kind": "action_required", "title": "...", "summary": "...",
           "how_to_fix": "..."},
          {"kind": "info", "title": "...", "summary": "..."}
        ]
      },
      {
        "source": "ci timing",
        "results": [
          {"kind": "warning", "title": "CI timings", "summary": "...",
           "detail": "...", "link": "..."}
        ]
      }
    ]

One job can emit multiple results of different kinds. The source field
is used to exclude the corresponding job from the synthesized error
list (case-insensitive, hyphen-normalized matching against GitHub
Actions job display names).

| job                        | source                   | kind (on failure)         | section              |
|----------------------------|--------------------------|---------------------------|----------------------|
| review-labels              | review label gate        | action_required / info    | Action required      |
| lockfile-diff              | lockfile-diff            | action_required           | Action required      |
| ci-timings                 | ci timing                | warning / info            | Warnings             |
| supply-chain scan          | supply chain             | error / (none)            | Job failures         |
| supply-chain dep-bounds    | supply chain             | action_required / (none)  | Action required      |
| osv-scanner                | osv scan                 | warning / (none)          | Warnings             |
| uv-lockfile-check          | uv.lock check            | action_required / (none)  | Action required      |
| history-check              | unrelated histories      | action_required           | Action required      |
| contributor-check          | contributor attribution  | action_required           | Action required      |

Jobs that find nothing emit [] (empty array) — no noise info items.

A single comment-live job polls the GitHub Actions API every 15s,
classifies jobs into (completed, pending), assembles the comment, and
upserts it. Merges review_status outputs from all needs jobs via
toJSON(needs.*.outputs.review_status), and downloads the ci-timings
artifact when it becomes available. Shows commit SHA + message below
the header.

The assembler has ZERO job-specific knowledge. It just:
1. collect_from_statuses() — flattens all nested status objects into ReviewItems
2. collect_failed_jobs() — synthesizes errors for failed jobs with no declared status
3. _attach_job_urls() — fills in per-job log links for ALL items
4. render_comment() — groups by severity, renders with group headers

Each item shows links inline next to the title: View report (job-emitted
URL) and View job (auto-attached logs link). Each info item is its own
collapsible <details> block.

    # ૮ >ﻌ< ა ci review

    running on abc1234 — commit message first line

    ## ❌ Job failures
    ### {title} · [View job](url)
    {summary}

    ## ⚠️ Action required
    ### {title} · [View job](url)
    {summary}
    **How to fix:**
    {how_to_fix}

    ## ⚠️ Warnings
    ### {title} · [View report](url) · [View job](url)
    {summary}
    {detail}

    <details><summary>{title}</summary>
    {content}
    </details>

    Still running 3 jobs: ci-timings, docker

- test_assemble_review_comment.py (48 tests): collect_from_statuses,
  collect_failed_jobs with exclude_sources, _attach_job_urls,
  render_comment (group headers, inline links, commit info, per-item
  details, pending footer), assemble integration
- test_live_comment.py (16 tests): classify_jobs pure function
- test_timings_report.py (10 tests): generate_review_status nested format
- test_lockfile_diff.py (6 tests)
- test_classify_changes.py (32 tests, pre-existing)

scripts/ci/assemble_review_comment.py

@@ -0,0 +1,424 @@
+#!/usr/bin/env python3
+"""Assemble the unified CI review comment for a pull request.
+
+Every CI job that wants to appear in the review comment emits a
+``review_status`` output: a JSON array of objects, each with a ``source``
+(the workflow name, used for dedup) and a ``results`` array of typed
+result objects::
+
+    [
+      {
+        "source": "review-label-gate",
+        "results": [
+          {"kind": "action_required", "title": "...", "summary": "...",
+           "how_to_fix": "..."},
+          {"kind": "info", "title": "...", "summary": "..."}
+        ]
+      },
+      {
+        "source": "ci-timings",
+        "results": [
+          {"kind": "warning", "title": "CI timings", "summary": "...",
+           "detail": "...", "link": "..."}
+        ]
+      }
+    ]
+
+Each result object has:
+
+    kind:        "error" | "action_required" | "warning" | "info"
+    title:       section heading
+    summary:     one-line description
+    detail:      markdown detail (optional)
+    how_to_fix:  markdown checklist (optional)
+    link:        URL (optional)
+    link_label:  label for the link (optional, default "View logs")
+
+The assembler flattens all results into a flat list of ReviewItems,
+grouped by severity in the comment. Jobs that failed (from the
+``needs`` context) but didn't emit any status get synthesized ❌ Error
+items. Jobs that DID emit a status are excluded from the synthesized
+error list — their own output is the authority for their classification.
+
+Exits 0 always — comment posting is best-effort (fork PRs are read-only).
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from dataclasses import dataclass
+from pathlib import Path
+
+# Hidden marker the comment system uses to find-and-edit its
+# previous comment instead of stacking new ones on each run.
+MARKER = "<!-- hermes-ci-review-bot -->"
+
+# Severity ordering for display.
+_SEVERITY_ORDER = ["error", "action_required", "warning", "info"]
+
+# Severities that trigger the "blocking issues" layout (vs. the
+# "looks good!" banner).
+_BLOCKING_SEVERITIES = ("error", "action_required", "warning")
+
+_SEVERITY_GROUP_HEADER = {
+    "error": "## ❌ Job failures",
+    "action_required": "## ⚠️ Action required",
+    "warning": "## ⚠️ Warnings",
+    "info": "## ℹ️ Details",
+}
+
+
+@dataclass
+class ReviewItem:
+    """A single piece of review information with a severity tag."""
+
+    severity: str  # "error" | "action_required" | "warning" | "info"
+    title: str  # short section title, e.g. "package-lock.json"
+    summary: str  # one-line summary
+    detail: str = ""  # optional markdown detail (tables, bullet lists, etc.)
+    link: str = ""  # optional URL emitted by the job (e.g. report URL)
+    link_label: str = "View report"  # label for the emitted link
+    how_to_fix: str = ""  # optional markdown checklist for action_required items
+    source: str = ""  # workflow that declared this status (for dedup)
+    job_url: str = ""  # auto-attached per-job log link (from the live poller)
+
+
+# ---------------------------------------------------------------------------
+# Collectors — each returns a list of ReviewItems (possibly empty)
+# ---------------------------------------------------------------------------
+
+
+def collect_from_statuses(review_statuses_json: str) -> tuple[list[ReviewItem], set[str]]:
+    """Parse the nested review_status JSON into flat ReviewItems.
+
+    The input is a JSON array of ``{source, results: [...]}`` objects.
+    Each entry in ``results`` becomes one ReviewItem, tagged with the
+    parent's ``source``.
+
+    Returns ``(items, sources)`` where ``sources`` is the set of source
+    values — used by :func:`collect_failed_jobs` to exclude jobs that
+    already declared their own status (so a failing job that emitted an
+    ``action_required`` status doesn't also show as a synthesized ❌ Error).
+    """
+    if not review_statuses_json:
+        return [], set()
+    try:
+        data = json.loads(review_statuses_json)
+    except (json.JSONDecodeError, TypeError):
+        return [], set()
+    if not isinstance(data, list):
+        return [], set()
+
+    items: list[ReviewItem] = []
+    sources: set[str] = set()
+
+    for entry in data:
+        if not isinstance(entry, dict):
+            continue
+        source = entry.get("source", "")
+        if source:
+            sources.add(source)
+        for r in entry.get("results", []):
+            if not isinstance(r, dict):
+                continue
+            kind = r.get("kind", "info")
+            if kind not in _SEVERITY_ORDER:
+                kind = "info"
+            items.append(ReviewItem(
+                severity=kind,
+                title=r.get("title", "Unknown"),
+                summary=r.get("summary", ""),
+                detail=r.get("detail", ""),
+                link=r.get("link", ""),
+                link_label=r.get("link_label", "View logs"),
+                how_to_fix=r.get("how_to_fix", ""),
+                source=source,
+            ))
+
+    return items, sources
+
+
+def collect_failed_jobs(
+    needs_json: str,
+    run_url: str,
+    exclude_sources: set[str] | None = None,
+    job_urls: dict[str, str] | None = None,
+) -> list[ReviewItem]:
+    """Build error items for failed CI jobs from the ``needs`` context.
+
+    ``needs_json`` is the JSON string emitted by ``all-checks-pass`` — a
+    ``{job_name: result}`` dict where result is ``success`` / ``failure``
+    / ``skipped``. Only ``failure`` entries become error items.
+
+    ``exclude_sources`` is a set of ``source`` values from status objects
+    declared by workflow_call jobs. Job names containing any of these
+    source strings are excluded — their failure is already covered by their
+    own status output.
+
+    ``job_urls`` is an optional ``{job_name: html_url}`` dict from the
+    live poller. When a job's name is in this dict, the ❌ Error link
+    points directly to that job's logs page instead of the whole run.
+    Falls back to ``run_url`` when no per-job URL is available.
+    """
+    if not needs_json:
+        return []
+    try:
+        needs = json.loads(needs_json)
+    except (json.JSONDecodeError, TypeError):
+        return []
+
+    # Pre-normalize exclude sources once: lowercase + hyphens→spaces, so
+    # "review-label-gate" matches "Review label gate / Review label gate".
+    norm_sources = {
+        src.lower().replace("-", " ") for src in (exclude_sources or set())
+    }
+
+    items: list[ReviewItem] = []
+    for name, result in sorted(needs.items()):
+        if result != "failure":
+            continue
+        if norm_sources:
+            norm = name.lower().replace("-", " ")
+            if any(src in norm for src in norm_sources):
+                continue
+        job_url = (job_urls or {}).get(name, run_url)
+        items.append(ReviewItem(
+            severity="error",
+            title=name,
+            summary=f"Job **{name}** failed.",
+            job_url=job_url,
+        ))
+    return items
+
+
+# ---------------------------------------------------------------------------
+# Rendering
+# ---------------------------------------------------------------------------
+
+
+def _render_item(item: ReviewItem) -> str:
+    """Render a single ReviewItem as a markdown block.
+
+    The group header (``## ❌ Job failures`` etc.) carries the severity
+    emoji, so items don't repeat it. Links are shown inline next to the
+    title. Layout per item::
+
+        ### {title} · [View report](url) · [View job](url)
+
+        {summary}
+
+        {detail}
+
+        **How to fix:**
+
+        {how_to_fix}
+    """
+    title = f"### {item.title}"
+    # Build inline links next to the title.
+    links: list[str] = []
+    if item.link:
+        links.append(f"[{item.link_label}]({item.link})")
+    if item.job_url:
+        links.append(f"[View job]({item.job_url})")
+    if links:
+        title += " · " + " · ".join(links)
+
+    parts = [title, "", item.summary]
+
+    if item.detail:
+        parts += ["", item.detail]
+    if item.how_to_fix:
+        parts += ["", "**How to fix:**", "", item.how_to_fix]
+
+    return "\n".join(parts)
+
+
+def _render_group(header: str, items: list[ReviewItem]) -> str:
+    """Render a severity group: ``##`` header + items separated by ``---``."""
+    blocks = [_render_item(i) for i in items]
+    return f"{header}\n\n" + "\n\n---\n\n".join(blocks)
+
+
+def _render_info_details(items: list[ReviewItem]) -> str:
+    """Render each info item as its own collapsible ``<details>`` block."""
+    blocks = []
+    for item in items:
+        inner = _render_item(item)
+        blocks.append(
+            f"<details>\n<summary>{item.title}</summary>\n\n{inner}\n\n</details>"
+        )
+    return "\n\n".join(blocks)
+
+
+def _render_pending_items(pending_jobs: list[str]) -> str:
+    """Render the dimmed ``<sub>`` items for jobs still running."""
+    job_list = ", ".join(f"`{j}`" for j in sorted(pending_jobs))
+    return f"\n\n---\n\n<sub>Still running {len(pending_jobs)} job{'s' if len(pending_jobs) != 1 else ''}: {job_list}</sub>\n"
+
+
+def render_comment(items: list[ReviewItem], pending_jobs: list[str] | None = None, commit_info: str = "") -> str:
+    """Render the full comment body from a list of review items.
+
+    Items are grouped by severity under ``##`` group headers, separated
+    by ``---``. Errors and action_required items are always visible.
+    Warnings are shown only when present. Info items are in a collapsible
+    ``<details>`` block. If ``pending_jobs`` is non-empty, a dimmed
+    ``<sub>`` footer is appended listing jobs still running.
+
+    When there are no errors, action_required, or warnings (only info
+    items, or nothing at all), a "looks good!" banner is shown at the top,
+    and info items (if any) follow in a collapsible ``<details>`` block.
+    """
+    pending = pending_jobs or []
+
+    # Group by severity
+    by_severity: dict[str, list[ReviewItem]] = {s: [] for s in _SEVERITY_ORDER}
+    for item in items:
+        by_severity.setdefault(item.severity, []).append(item)
+
+    info = by_severity.get("info", [])
+    has_blocking = any(by_severity.get(s) for s in _BLOCKING_SEVERITIES)
+
+    body = f"{MARKER}\n# ૮ >ﻌ< ა ci review\n\n"
+
+    if commit_info:
+        body += f"{commit_info}\n\n"
+
+    if not items and not pending:
+        return f"{body}looks good to me!"
+
+    sections: list[str] = []
+
+    for sev in _BLOCKING_SEVERITIES:
+        group = by_severity.get(sev, [])
+        if group:
+            sections.append(_render_group(_SEVERITY_GROUP_HEADER[sev], group))
+
+    # Info: collapsible <details>
+    if info:
+        sections.append(_render_info_details(info))
+
+    if pending:
+        body += _render_pending_items(pending)
+
+    if sections:
+        body += "\n\n---\n\n".join(sections)
+
+    return body
+
+
+# ---------------------------------------------------------------------------
+# Assembly
+# ---------------------------------------------------------------------------
+
+
+def _attach_job_urls(items: list[ReviewItem], job_urls: dict[str, str], run_url: str) -> None:
+    """Fill in per-job log links for all items.
+
+    Uses the same case-insensitive, hyphen-normalized matching as
+    :func:`collect_failed_jobs`: the item's ``source`` is matched against
+    job names in ``job_urls``. Sets ``job_url`` on the item — this is
+    separate from ``link`` (the job-emitted URL, e.g. a report artifact),
+    so both can appear in the rendered comment.
+    """
+    if not job_urls and not run_url:
+        return
+    # Pre-normalize job_url keys once.
+    norm_urls: dict[str, str] = {}
+    for name, url in job_urls.items():
+        norm_urls[name.lower().replace("-", " ")] = url
+
+    for item in items:
+        if item.job_url:
+            continue
+        src = item.source.lower().replace("-", " ")
+        # Try exact match first, then substring match.
+        if src in norm_urls:
+            item.job_url = norm_urls[src]
+            continue
+        for norm_name, url in norm_urls.items():
+            if src and src in norm_name:
+                item.job_url = url
+                break
+        # If no per-job URL found, fall back to run_url for items with a source.
+        if not item.job_url and item.source and run_url:
+            item.job_url = run_url
+
+
+def assemble(
+    needs_json: str = "",
+    run_url: str = "",
+    job_urls: dict[str, str] | None = None,
+    review_statuses_json: str = "",
+    pending_jobs: list[str] | None = None,
+    commit_info: str = "",
+) -> str:
+    """Assemble the full comment body from all available inputs."""
+    items: list[ReviewItem] = []
+
+    # 1. Structured statuses from workflow_call jobs (review-labels, etc.)
+    status_items, sources = collect_from_statuses(review_statuses_json)
+    items.extend(status_items)
+
+    # 2. Synthesized error items for failed jobs not covered by statuses
+    items.extend(collect_failed_jobs(needs_json, run_url, exclude_sources=sources, job_urls=job_urls))
+
+    # 3. Attach per-job log links to all items (not just synthesized errors)
+    _attach_job_urls(items, job_urls or {}, run_url)
+
+    return render_comment(items, pending_jobs, commit_info)
+
+
+# ---------------------------------------------------------------------------
+# CLI
+# ---------------------------------------------------------------------------
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument(
+        "--needs-json",
+        default="",
+        help="JSON string of {job_name: result} from the all-checks-pass job.",
+    )
+    parser.add_argument(
+        "--run-url",
+        default="",
+        help="URL to the CI run summary page (for failed job links).",
+    )
+    parser.add_argument(
+        "--review-statuses-json",
+        default="",
+        help="JSON array of {source, results: [...]} objects from workflow_call jobs.",
+    )
+    parser.add_argument(
+        "--pending-jobs",
+        default="",
+        help="Comma-separated list of job names still running (shown in a dimmed footer).",
+    )
+    parser.add_argument(
+        "--output",
+        type=Path,
+        required=True,
+        help="Output file for the assembled comment body.",
+    )
+    args = parser.parse_args()
+
+    pending = [j.strip() for j in args.pending_jobs.split(",") if j.strip()] if args.pending_jobs else None
+
+    body = assemble(
+        needs_json=args.needs_json,
+        run_url=args.run_url,
+        review_statuses_json=args.review_statuses_json,
+        pending_jobs=pending,
+    )
+
+    args.output.write_text(body)
+    print(f"Wrote {len(body)} chars to {args.output}")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

scripts/ci/emit_review_status.py

@@ -0,0 +1,143 @@
+#!/usr/bin/env python3
+"""Emit review_status JSON for the review-labels workflow.
+
+Builds a JSON array with one entry::
+
+    [
+      {
+        "source": "review-label-gate",
+        "results": [
+          {"kind": "action_required", "title": "...", "summary": "...",
+           "how_to_fix": "..."},
+          {"kind": "info", "title": "...", "summary": "..."}
+        ]
+      }
+    ]
+
+The ``source`` field is the workflow name that declared the status; the
+assembler uses it to exclude the corresponding job from the synthesized
+❌ Error list (the job already has its own status section).
+
+The array can contain 0, 1, or 2 results — one per lane that ran
+(``ci_review``, ``mcp_catalog``). When the ``ci-reviewed`` label is
+present, the kind is ``info``; when missing, it's ``action_required``
+with the verification checklist.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+
+# The source identifier used for error-synthesis exclusion. This must
+# match (as a normalized substring) the job name as it appears in the
+# GitHub Actions API. The ci.yml job key is ``review-labels`` with
+# ``name: Review label gate``, and the reusable workflow's job is also
+# ``name: Review label gate``, so the API shows the job as
+# "Review label gate / Review label gate". Normalizing "review-label-gate"
+# (lowercase, hyphens→spaces) gives "review label gate", which is a
+# substring of "review label gate / review label gate".
+SOURCE = "review-label-gate"
+
+
+def build_results(
+    ci_review: bool,
+    mcp_catalog: bool,
+    label_present: bool,
+) -> list[dict]:
+    """Build the list of result objects for this source."""
+    results: list[dict] = []
+
+    if ci_review:
+        if label_present:
+            results.append({
+                "kind": "info",
+                "title": "CI-sensitive file review",
+                "summary": "`ci-reviewed` label is present.",
+            })
+        else:
+            results.append({
+                "kind": "action_required",
+                "title": "CI-sensitive file review",
+                "summary": (
+                    "This PR changes CI-sensitive files (eslint config, "
+                    "workflow YAMLs, or composite actions). These influence "
+                    "what the js-autofix job executes and pushes to main."
+                ),
+                "how_to_fix": (
+                    "Add the `ci-reviewed` label after verifying:\n"
+                    "- no new eslint rules with custom `fix` functions that write outside linted paths,\n"
+                    "- no workflow changes that widen permissions or remove guards,\n"
+                    "- no composite action changes that alter what gets executed."
+                ),
+            })
+
+    if mcp_catalog:
+        if label_present:
+            results.append({
+                "kind": "info",
+                "title": "MCP catalog security review",
+                "summary": "`ci-reviewed` label is present.",
+            })
+        else:
+            results.append({
+                "kind": "action_required",
+                "title": "MCP catalog security review",
+                "summary": (
+                    "This PR changes the bundled MCP catalog or MCP catalog "
+                    "installer code. MCP entries can define local commands "
+                    "that users later install into `mcp_servers`, so this "
+                    "needs explicit maintainer review before merge."
+                ),
+                "how_to_fix": (
+                    "Add the `ci-reviewed` label after verifying:\n"
+                    "- any new/changed `optional-mcps/**/manifest.yaml` command and args are expected,\n"
+                    "- stdio transports do not use shell+egress/exfiltration payloads,\n"
+                    "- git install refs are pinned and bootstrap commands are minimal,\n"
+                    "- requested env vars/secrets match the upstream MCP's documented needs."
+                ),
+            })
+
+    return results
+
+
+def build_statuses(
+    ci_review: bool,
+    mcp_catalog: bool,
+    label_present: bool,
+) -> list[dict]:
+    """Build the full review_status array (one entry with a results list)."""
+    results = build_results(ci_review, mcp_catalog, label_present)
+    if not results:
+        return []
+    return [{"source": SOURCE, "results": results}]
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--ci-review", action="store_true",
+                        help="Whether CI-sensitive files changed.")
+    parser.add_argument("--mcp-catalog", action="store_true",
+                        help="Whether the MCP catalog / installer changed.")
+    parser.add_argument("--label-present", action="store_true",
+                        help="Whether the ci-reviewed label is present.")
+    parser.add_argument("--output", default="-",
+                        help="Output file ('-' for stdout, or a GITHUB_OUTPUT path).")
+    args = parser.parse_args()
+
+    statuses = build_statuses(args.ci_review, args.mcp_catalog, args.label_present)
+    json_str = json.dumps(statuses)
+
+    if args.output == "-":
+        print(json_str)
+    else:
+        # GITHUB_OUTPUT format: key=value\n
+        with open(args.output, "a", encoding="utf-8") as f:
+            f.write(f"review_status={json_str}\n")
+
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

scripts/ci/live_comment.py

@@ -0,0 +1,526 @@
+#!/usr/bin/env python3
+"""Live-updating CI review comment.
+
+Polls the GitHub Actions API for job statuses in the current run, assembles
+the review comment from whatever results are available, and upserts it as a
+PR comment. Repeats every ``--interval`` seconds until all jobs are
+completed (or ``--timeout`` is reached), so the comment updates in real time
+as each job finishes.
+
+The comment is identified by the ``<!-- hermes-ci-review-bot -->`` marker
+— the same one ``assemble_review_comment.py`` uses — so it replaces any
+previous comment from an earlier run.
+
+Architecture:
+
+  - :func:`classify_jobs` (pure, testable) — takes a list of raw API job
+    dicts and returns ``(completed, pending, job_urls)`` where ``completed``
+    is a ``{name: result}`` dict (for :func:`assemble_review_comment.assemble`)
+    and ``pending`` is a list of job names still running.
+
+  - :func:`find_comment_id` / :func:`upsert_comment` — thin API wrappers.
+
+  - :func:`_fetch_timings_statuses` — downloads the ci-timings artifact
+    (if available) and parses the ``review_status=`` line from it, merging
+    the status objects into the review statuses array.
+
+  - :func:`run` — the polling loop. Calls the API, classifies, assembles,
+    upserts, sleeps, repeats. Exits when all jobs are completed.
+
+The orchestrator job names (detect, all-checks-pass, comment-live, etc.)
+are excluded from the comment — they're infrastructure, not review signal.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import subprocess
+import sys
+import time
+import urllib.error
+import urllib.request
+from pathlib import Path
+
+API_BASE = "https://api.github.com"
+
+# Job names that are infrastructure (this script, the gate, the detector)
+# and should never appear in the review comment.
+_INFRA_JOBS = frozenset({
+    "detect",
+    "all-checks-pass",
+    "comment-pending",
+    "comment-results",
+    "comment-live",
+    "CI review comment (pending)",
+    "CI review comment (results)",
+    "CI review comment (live)",
+    "All required checks pass",
+    "Detect affected areas",
+})
+
+# Map GitHub API conclusion values to our result strings.
+_CONCLUSION_MAP = {
+    "success": "success",
+    "failure": "failure",
+    "skipped": "skipped",
+    "cancelled": "skipped",
+    "neutral": "skipped",
+    "timed_out": "failure",
+    "action_required": "skipped",
+}
+
+
+def classify_jobs(api_jobs: list[dict]) -> tuple[dict[str, str], list[str], dict[str, str]]:
+    """Classify raw API job dicts into completed + pending + job_urls.
+
+    Returns ``(completed, pending, job_urls)``:
+
+    - ``completed``: ``{job_name: result}`` where result is
+      ``"success"`` / ``"failure"`` / ``"skipped"``. Only non-infra jobs
+      that have finished.
+    - ``pending``: list of job names still running (in_progress / queued
+      / waiting). Excludes infra jobs.
+    - ``job_urls``: ``{job_name: html_url}`` — direct links to each
+      job's logs page, for the assembler to use in ❌ Error links.
+
+    The API returns orchestrator-level jobs and sub-workflow jobs
+    (workflow_call) in separate runs — :func:`collect_run_jobs` merges
+    them. Each sub-workflow job has a ``_workflow_name`` prefix so the
+    display name is ``"Workflow / job"``.
+    """
+    completed: dict[str, str] = {}
+    pending: list[str] = []
+    job_urls: dict[str, str] = {}
+
+    for job in api_jobs:
+        name = job.get("name", "unknown")
+        if job.get("_workflow_name"):
+            name = f"{job['_workflow_name']} / {name}"
+        if name in _INFRA_JOBS:
+            continue
+        status = job.get("status", "")
+        conclusion = job.get("conclusion", "")
+        html_url = job.get("html_url", "")
+
+        if html_url:
+            job_urls[name] = html_url
+
+        if status in ("in_progress", "queued", "waiting"):
+            pending.append(name)
+        elif status == "completed":
+            result = _CONCLUSION_MAP.get(conclusion, "skipped")
+            completed[name] = result
+        # else: unknown status → skip
+
+    return completed, pending, job_urls
+
+
+# ---------------------------------------------------------------------------
+# API helpers
+# ---------------------------------------------------------------------------
+
+
+def _api_request(url: str, token: str) -> dict:
+    """Authenticated GitHub API GET (single page)."""
+    req = urllib.request.Request(url, headers={
+        "Authorization": f"Bearer {token}",
+        "Accept": "application/vnd.github+json",
+        "X-GitHub-Api-Version": "2022-11-28",
+        "User-Agent": "ci-live-comment",
+    })
+    with urllib.request.urlopen(req) as resp:
+        data: dict = json.loads(resp.read())
+        return data
+
+
+def _api_get_paginated(url: str, token: str, list_key: str | None = None) -> list:
+    """Authenticated GitHub API GET with pagination."""
+    results: list = []
+    while url:
+        req = urllib.request.Request(url, headers={
+            "Authorization": f"Bearer {token}",
+            "Accept": "application/vnd.github+json",
+            "X-GitHub-Api-Version": "2022-11-28",
+            "User-Agent": "ci-live-comment",
+        })
+        with urllib.request.urlopen(req) as resp:
+            data = json.loads(resp.read())
+            link_header = resp.headers.get("Link", "")
+
+        if list_key:
+            results.extend(data.get(list_key, []))
+        elif isinstance(data, list):
+            results.extend(data)
+        else:
+            return data
+
+        next_url = None
+        for part in link_header.split(","):
+            part = part.strip()
+            if 'rel="next"' in part:
+                next_url = part[part.find("<") + 1:part.find(">")]
+                break
+        url = next_url
+
+    return results
+
+
+def collect_run_jobs(token: str, repo: str, run_id: str) -> list[dict]:
+    """Collect all jobs in the orchestrator run + sub-workflow runs.
+
+    Returns a flat list of job dicts (same shape as the API returns, plus
+    ``_workflow_name`` on sub-workflow jobs).
+    """
+    owner, repo_name = repo.split("/")
+    run_info = _api_request(f"{API_BASE}/repos/{owner}/{repo_name}/actions/runs/{run_id}", token)
+    created_at = run_info.get("created_at", "")
+    head_sha = run_info.get("head_sha", "")
+
+    # Orchestrator jobs
+    orch_jobs = _api_get_paginated(
+        f"{API_BASE}/repos/{owner}/{repo_name}/actions/runs/{run_id}/jobs",
+        token, list_key="jobs",
+    )
+
+    # Sub-workflow runs (workflow_call)
+    sub_runs = _api_get_paginated(
+        f"{API_BASE}/repos/{owner}/{repo_name}/actions/runs?head_sha={head_sha}&event=workflow_call&per_page=100",
+        token, list_key="workflow_runs",
+    )
+    sub_runs = [r for r in sub_runs if r.get("created_at", "") >= created_at]
+
+    all_jobs: list[dict] = []
+    # Orchestrator jobs: skip workflow-call placeholder steps (they're
+    # sub-workflow triggers, not review signal), but KEEP in_progress /
+    # queued jobs so the poller knows they're still running.
+    for job in orch_jobs:
+        steps = job.get("steps") or []
+        if any(s.get("name", "").startswith("Run ./.github/") for s in steps):
+            continue
+        all_jobs.append(job)
+
+    # Sub-workflow jobs (workflow_call).
+    # These runs may not exist yet on the first few polls — that's fine,
+    # classify_jobs() will just show 0 pending for them.
+    for sr in sub_runs:
+        sr_id = sr["id"]
+        sr_name = sr.get("name", "")
+        sr_jobs = _api_get_paginated(
+            f"{API_BASE}/repos/{owner}/{repo_name}/actions/runs/{sr_id}/jobs",
+            token, list_key="jobs",
+        )
+        for j in sr_jobs:
+            j["_workflow_name"] = sr_name
+            all_jobs.append(j)
+
+    return all_jobs
+
+
+def find_comment_id(token: str, repo: str, pr_number: str) -> int | None:
+    """Find our existing review comment by marker prefix."""
+    owner, repo_name = repo.split("/")
+    comments = _api_get_paginated(
+        f"{API_BASE}/repos/{owner}/{repo_name}/issues/{pr_number}/comments",
+        token,
+    )
+    for c in comments:
+        body = c.get("body", "") if isinstance(c, dict) else ""
+        if body.startswith("<!-- hermes-ci-review-bot -->"):
+            return c.get("id") if isinstance(c, dict) else None
+    return None
+
+
+def upsert_comment(
+    token: str, repo: str, pr_number: str, body: str, comment_id: int | None = None
+) -> int | None:
+    """Create or update the review comment. Returns the comment ID."""
+    owner, repo_name = repo.split("/")
+    if comment_id is None:
+        comment_id = find_comment_id(token, repo, pr_number)
+
+    if comment_id:
+        url = f"{API_BASE}/repos/{owner}/{repo_name}/issues/comments/{comment_id}"
+        method = "PATCH"
+    else:
+        url = f"{API_BASE}/repos/{owner}/{repo_name}/issues/{pr_number}/comments"
+        method = "POST"
+
+    data = json.dumps({"body": body}).encode("utf-8")
+    req = urllib.request.Request(url, data=data, method=method, headers={
+        "Authorization": f"Bearer {token}",
+        "Accept": "application/vnd.github+json",
+        "X-GitHub-Api-Version": "2022-11-28",
+        "Content-Type": "application/json",
+        "User-Agent": "ci-live-comment",
+    })
+    try:
+        with urllib.request.urlopen(req) as resp:
+            result = json.loads(resp.read())
+            return result.get("id")
+    except urllib.error.HTTPError as e:
+        print(f"  API error {e.code}: {e.reason}", file=sys.stderr)
+        return None
+
+
+# ---------------------------------------------------------------------------
+# Artifact fetching (ci-timings review_status)
+# ---------------------------------------------------------------------------
+
+
+def _fetch_artifact_statuses(
+    token: str, repo: str, run_id: str, artifact_name: str,
+) -> list[dict]:
+    """Download a workflow artifact and extract review_status entries.
+
+    The ci-timings job writes a ``review-status.json`` file containing
+    ``review_status=<json>`` (GITHUB_OUTPUT format) into its artifact.
+    This function downloads the artifact, parses the line, and returns
+    the parsed status array. Returns ``[]`` if the artifact doesn't exist
+    yet or can't be parsed.
+    """
+    try:
+        result = subprocess.run(
+            ["gh", "run", "download", run_id, "--repo", repo,
+             "--name", artifact_name, "--dir", "/tmp/artifact-dl"],
+            capture_output=True, timeout=30,
+        )
+        if result.returncode != 0:
+            return []
+    except Exception:
+        return []
+
+    status_file = Path("/tmp/artifact-dl/review-status.json")
+    if not status_file.exists():
+        return []
+
+    try:
+        content = status_file.read_text(encoding="utf-8").strip()
+        # GITHUB_OUTPUT format: review_status=<json>
+        if content.startswith("review_status="):
+            content = content[len("review_status="):]
+        statuses = json.loads(content)
+        if isinstance(statuses, list):
+            return statuses
+    except (json.JSONDecodeError, OSError):
+        pass
+
+    return []
+
+
+# ---------------------------------------------------------------------------
+# Comment assembly
+# ---------------------------------------------------------------------------
+
+
+def _import_assembler():
+    """Import assemble_review_comment.py from the same directory."""
+    here = Path(__file__).resolve().parent
+    sys.path.insert(0, str(here))
+    import assemble_review_comment as asm
+    return asm
+
+
+def build_comment_body(
+    asm_mod,
+    completed: dict[str, str],
+    pending: list[str],
+    run_url: str,
+    job_urls: dict[str, str],
+    review_statuses_json: str,
+    commit_info: str = "",
+) -> str:
+    """Assemble the comment body from current job states + static inputs."""
+    needs_json = json.dumps(completed) if completed else ""
+
+    return asm_mod.assemble(
+        needs_json=needs_json,
+        run_url=run_url,
+        job_urls=job_urls,
+        review_statuses_json=review_statuses_json,
+        pending_jobs=pending if pending else None,
+        commit_info=commit_info,
+    )
+
+
+def _merge_statuses(
+    base_statuses: list[dict], extra_statuses: list[dict]
+) -> str:
+    """Merge two status arrays into one JSON string."""
+    merged = list(base_statuses) + list(extra_statuses)
+    return json.dumps(merged) if merged else ""
+
+
+# ---------------------------------------------------------------------------
+# Polling loop
+# ---------------------------------------------------------------------------
+
+
+def run(
+    token: str,
+    repo: str,
+    run_id: str,
+    pr_number: str,
+    run_url: str,
+    review_statuses_json: str = "",
+    commit_info: str = "",
+    interval: int = 15,
+    timeout: int = 1800,
+    dry_run: bool = False,
+) -> int:
+    """Poll for job statuses and update the PR comment until all done.
+
+    Returns 0 always — comment posting is best-effort.
+    """
+    asm = _import_assembler()
+    start = time.time()
+    last_body = ""
+
+    # Parse the base statuses once (from review-labels, lockfile-diff, etc.)
+    try:
+        base_statuses = json.loads(review_statuses_json) if review_statuses_json else []
+    except (json.JSONDecodeError, TypeError):
+        base_statuses = []
+    print(f"  Loaded {len(base_statuses)} base review status entries")
+
+    while True:
+        elapsed = time.time() - start
+        if elapsed > timeout:
+            print(f"Timeout ({timeout}s) reached — stopping poll.", file=sys.stderr)
+            break
+
+        try:
+            jobs = collect_run_jobs(token, repo, run_id)
+        except Exception as e:
+            print(f"  API error collecting jobs: {e}", file=sys.stderr)
+            time.sleep(interval)
+            continue
+
+        completed, pending, job_urls = classify_jobs(jobs)
+        total = len(completed) + len(pending)
+        print(f"  [{elapsed:.0f}s] {len(completed)} completed, {len(pending)} pending "
+              f"({total} total jobs)")
+
+        # Try to fetch ci-timings artifact statuses (may not exist yet).
+        artifact_statuses = _fetch_artifact_statuses(
+            token, repo, run_id, "ci-timings-report",
+        )
+        if artifact_statuses:
+            print(f"  Found ci-timings artifact with {len(artifact_statuses)} status entries")
+
+        merged_json = _merge_statuses(base_statuses, artifact_statuses)
+
+        body = build_comment_body(
+            asm, completed, pending, run_url, job_urls,
+            merged_json,
+            commit_info,
+        )
+
+        if body != last_body:
+            if dry_run:
+                print("--- DRY RUN — comment body ---")
+                print(body)
+                print("--- END ---")
+            else:
+                cid = upsert_comment(token, repo, pr_number, body)
+                if cid:
+                    print(f"  Updated comment {cid}")
+                else:
+                    print("  Failed to update comment (will retry)", file=sys.stderr)
+            last_body = body
+        else:
+            print("  No change since last poll.")
+
+        if not pending:
+            # Check if any dependency failed. If so, exit non-zero so the
+            # run shows as failed — this lets ``gh run rerun --failed``
+            # (e.g. from label-rerun.yml) pick up and rerun the failed jobs.
+            failed_deps = [name for name, result in completed.items() if result == "failure"]
+            if failed_deps:
+                print(f"  All jobs done, but {len(failed_deps)} failed: {', '.join(failed_deps)}")
+                print("  Exiting with error so the run can be rerun via --failed.")
+                return 1
+            print("  All jobs completed — done.")
+            break
+
+        time.sleep(interval)
+
+    return 0
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--interval", type=int, default=15,
+                        help="Seconds between polls (default: 15).")
+    parser.add_argument("--timeout", type=int, default=1800,
+                        help="Max seconds to poll before giving up (default: 1800).")
+    parser.add_argument("--review-statuses-file", type=Path, default=None,
+                        help="Path to a JSON file with merged review statuses from workflow_call jobs.")
+    parser.add_argument("--dry-run", action="store_true",
+                        help="Print comment body instead of posting to PR.")
+    args = parser.parse_args()
+
+    token = os.environ.get("GITHUB_TOKEN", "")
+    repo = os.environ.get("GITHUB_REPOSITORY", "")
+    run_id = os.environ.get("GITHUB_RUN_ID", "")
+    pr_number = os.environ.get("PR_NUMBER", "")
+    run_url = os.environ.get("RUN_URL", "")
+
+    if not args.dry_run:
+        if not token:
+            print("GITHUB_TOKEN is required", file=sys.stderr)
+            return 1
+        if not repo:
+            print("GITHUB_REPOSITORY is required", file=sys.stderr)
+            return 1
+        if not run_id:
+            print("GITHUB_RUN_ID is required", file=sys.stderr)
+            return 1
+        if not pr_number:
+            print("PR_NUMBER is required", file=sys.stderr)
+            return 1
+
+    # Read merged review statuses from file (prepared by the ci.yml step).
+    review_statuses_json = ""
+    if args.review_statuses_file:
+        try:
+            review_statuses_json = args.review_statuses_file.read_text(encoding="utf-8")
+        except OSError as e:
+            print(f"Warning: could not read review statuses file: {e}", file=sys.stderr)
+
+    # Build commit info line from env vars (set by ci.yml).
+    commit_sha = os.environ.get("COMMIT_SHA", "")
+    commit_msg = os.environ.get("COMMIT_MESSAGE", "")
+    commit_url = os.environ.get("COMMIT_URL", "")
+    commit_info = ""
+    if commit_sha:
+        short_sha = commit_sha[:7]
+        if commit_msg:
+            # Truncate commit message to first line, max 60 chars.
+            first_line = commit_msg.split("\n")[0][:60]
+            if commit_url:
+                commit_info = f"<sub>running on [{short_sha}]({commit_url}) — {first_line}</sub>"
+            else:
+                commit_info = f"<sub>running on {short_sha} — {first_line}</sub>"
+        elif commit_url:
+            commit_info = f"<sub>running on [{short_sha}]({commit_url})</sub>"
+        else:
+            commit_info = f"<sub>running on {short_sha}</sub>"
+
+    return run(
+        token=token,
+        repo=repo,
+        run_id=run_id,
+        pr_number=pr_number,
+        run_url=run_url,
+        review_statuses_json=review_statuses_json,
+        commit_info=commit_info,
+        interval=args.interval,
+        timeout=args.timeout,
+        dry_run=args.dry_run,
+    )
+
+
+if __name__ == "__main__":
+    sys.exit(main())

scripts/ci/lockfile_diff.py

@@ -15,11 +15,11 @@ Usage (from a checkout that still has the base ref available):
         --output diff.md [--repo-root .]
 
 Reads every ``package-lock.json`` tracked at either ref (top-level and
-nested — the repo has several), diffs each, and writes a Markdown report
-to ``--output``. Exits 0 always; an empty report file means "no version
-changes" (the caller uses that to decide whether to post/update the PR
-comment). The report embeds ``COMMENT_MARKER`` so the workflow can find
-and update its own previous comment instead of stacking new ones.
+nested — the repo has several), diffs each, and writes a Markdown fragment
+to ``--output``. Exits 0 always; an empty output file means "no version
+changes" (the caller uses that to decide whether to include the section).
+The fragment is consumed by ``scripts/ci/assemble_review_comment.py``,
+which wraps it in a section with a header and action note.
 """
 
 from __future__ import annotations
@@ -29,9 +29,6 @@ import json
 import subprocess
 import sys
 
-# Hidden marker used to locate the bot's previous comment for in-place update.
-COMMENT_MARKER = "<!-- hermes-lockfile-diff -->"
-
 
 def parse_lockfile(text: str) -> dict[str, str]:
     """Reduce lockfile JSON to ``{install path: version}``.
@@ -79,21 +76,24 @@ def _display_name(path: str) -> str:
 
 
 def render_markdown(diffs: dict[str, dict[str, list]]) -> str:
-    """Render per-lockfile diffs as a Markdown PR comment body.
+    """Render per-lockfile diffs as a Markdown fragment.
 
     ``diffs`` maps lockfile repo-path → the output of :func:`diff_locks`.
     Lockfiles with no version changes are omitted. Returns ``""`` when
-    nothing changed anywhere (caller skips commenting entirely).
+    nothing changed anywhere (caller skips the section entirely).
+
+    The output is a fragment — per-lockfile ``####`` subsections with
+    tables — not a standalone comment. The ``assemble_review_comment``
+    script wraps this in a section with its own header and action note,
+    so no top-level header or comment marker is emitted here.
     """
     sections = []
-    total = 0
     for lockfile, d in sorted(diffs.items()):
         added, removed, updated = d["added"], d["removed"], d["updated"]
         n = len(added) + len(removed) + len(updated)
         if n == 0:
             continue
-        total += n
-        lines = [f"### `{lockfile}`", ""]
+        lines = [f"#### `{lockfile}`", ""]
         lines.append("| Package | Before | After |")
         lines.append("| --- | --- | --- |")
         for path, old, new in updated:
@@ -107,13 +107,7 @@ def render_markdown(diffs: dict[str, dict[str, list]]) -> str:
     if not sections:
         return ""
 
-    header = (
-        f"{COMMENT_MARKER}\n"
-        f"## ⚠️ `package-lock.json` changes ({total} package"
-        f"{'s' if total != 1 else ''})\n\n"
-        "This PR changes locked npm dependency versions."
-    )
-    return header + "\n" + "\n\n".join(sections) + "\n"
+    return "\n\n".join(sections) + "\n"
 
 
 def _git_show(ref: str, path: str, repo_root: str) -> str | None:

scripts/ci/timings_report.py

@@ -895,6 +895,86 @@ def generate_summary(timings: dict, baseline: dict | None = None) -> str:
     return "\n".join(lines)
 
 
+# ---------------------------------------------------------------------------
+# Review status JSON for the unified PR comment
+# ---------------------------------------------------------------------------
+
+# Wall-time regressions above this fraction of baseline are "warning" severity.
+_TIMINGS_WARN_PCT = 0.25
+
+
+def generate_review_status(
+    timings: dict, baseline: dict | None, report_url: str | None = None
+) -> list[dict]:
+    """Produce a review_status JSON array for the CI timings review section.
+
+    Returns a list with one ``{source, results: [...]}`` entry. The
+    result kind is ``"info"`` or ``"warning"`` (timings is never error —
+    it's an observability job). *summary* is a single short line suitable
+    for the PR comment. *detail* has the per-job deltas as a markdown
+    fragment.
+    """
+    stats = compute_stats(timings, baseline)
+
+    if baseline is None:
+        severity = "info"
+        summary = f"Wall time {fmt_dur(stats['wall'])} (no baseline yet)."
+    else:
+        wall = stats["wall"]
+        bl_wall = stats["bl_wall"] or 0
+        if bl_wall > 0:
+            pct = (wall - bl_wall) / bl_wall * 100
+            wall_str = f"Wall time {fmt_dur(wall)} vs {fmt_dur(bl_wall)} ({pct:+.1f}%)."
+            if pct > _TIMINGS_WARN_PCT * 100:
+                severity = "warning"
+            else:
+                severity = "info"
+        else:
+            wall_str = f"Wall time {fmt_dur(wall)}."
+            severity = "info"
+
+        if stats["slower"]:
+            wall_str += f" {stats['slower']} job(s) slower,"
+        if stats["faster"]:
+            wall_str += f" {stats['faster']} faster,"
+        if stats["unchanged"]:
+            wall_str += f" {stats['unchanged']} unchanged."
+        summary = wall_str
+
+    # Per-job delta detail (top 5 by absolute change)
+    detail_lines: list[str] = []
+    if baseline:
+        bl_map = {j["name"]: j for j in baseline.get("jobs", [])}
+        deltas: list[tuple[float, str, str]] = []
+        for j in timings.get("jobs", []):
+            if is_skipped(j):
+                continue
+            bl = bl_map.get(j["name"])
+            if not bl or is_skipped(bl):
+                continue
+            cur = j.get("duration_s") or 0
+            bl_d = bl.get("duration_s") or 0
+            diff = cur - bl_d
+            if abs(diff) < 1.0:
+                continue
+            deltas.append((abs(diff), j["name"], f"{diff:+.1f}s"))
+        deltas.sort(reverse=True)
+        for _, name, delta_str in deltas[:5]:
+            detail_lines.append(f"- {name}: {delta_str}")
+
+    result: dict = {
+        "kind": severity,
+        "title": "CI timings",
+        "summary": summary,
+        "detail": "\n".join(detail_lines),
+    }
+    if report_url:
+        result["link"] = report_url
+        result["link_label"] = "View report"
+
+    return [{"source": "ci timing", "results": [result]}]
+
+
 # ---------------------------------------------------------------------------
 # Main
 # ---------------------------------------------------------------------------
@@ -916,6 +996,8 @@ def main():
                         help="JSON output path (default: ci-timings.json)")
     parser.add_argument("--summary-out", default="ci-timings-summary.md",
                         help="Markdown summary output path (default: ci-timings-summary.md)")
+    parser.add_argument("--review-status-out", default="",
+                        help="If set, write a review-status JSON for the unified PR comment.")
     args = parser.parse_args()
 
     # Collect or load timings
@@ -974,6 +1056,17 @@ def main():
         f.write(summary)
         print(f"Wrote summary to {args.summary_out}")
 
+    # Write review status for the unified PR comment.
+    # The output goes to GITHUB_OUTPUT (or a file with the same key=value
+    # format) so the ci-timings job can expose it as a workflow_call output.
+    if args.review_status_out:
+        report_url = os.environ.get("CI_TIMINGS_REPORT_URL", "")
+        statuses = generate_review_status(timings, baseline, report_url)
+        json_str = json.dumps(statuses)
+        with open(args.review_status_out, "a", encoding="utf-8") as f:
+            f.write(f"review_status={json_str}\n")
+        print(f"Wrote review status to {args.review_status_out}")
+
 
 if __name__ == "__main__":
     main()