Add weekly parse attribution and keepalive NDJSON repair

.github/workflows/agents-keepalive-loop.yml

@@ -914,7 +914,7 @@ jobs:
           tasks_completed=$(( tasks_total - tasks_unchecked ))
           if [ "$tasks_completed" -lt 0 ]; then tasks_completed=0; fi
 
-          metrics_json=$(jq -n \
+          metrics_json=$(jq -cn \
             --arg pr "${PR_NUMBER:-0}" \
             --arg iteration "${ITERATION:-0}" \
             --arg action "${ACTION:-}" \

.github/workflows/agents-weekly-metrics.yml

@@ -124,6 +124,7 @@ jobs:
         env:
           METRICS_DIR: artifacts
           OUTPUT_PATH: agent-weekly-metrics.md
+          OUTPUT_JSON_PATH: agent-weekly-metrics.json
         run: |
           python scripts/aggregate_agent_metrics.py
 
@@ -236,6 +237,7 @@ jobs:
           name: agent-weekly-metrics
           path: |
             agent-weekly-metrics.md
+            agent-weekly-metrics.json
             artifacts/metric-artifacts-selection.json
             artifacts/metric-artifacts-selection.md
             terminal-disposition-coverage.json

scripts/aggregate_agent_metrics.py

@@ -6,16 +6,61 @@
 import datetime as _dt
 import json
 import os
+import re
 import sys
 from collections import Counter
 from collections.abc import Iterable
+from dataclasses import dataclass
 from pathlib import Path
 from typing import Any
 
 _DEFAULT_METRICS_DIR = "agent-metrics"
 _DEFAULT_OUTPUT = "agent-metrics-summary.md"
+_DEFAULT_JSON_OUTPUT = "agent-metrics-summary.json"
 _DEFAULT_UNSUPPORTED_VERIFIER_MODELS = {"gpt-5.2-codex"}
 _DEFAULT_VERIFIER_MODEL_METADATA_REQUIRED_AFTER = "2026-04-26T04:25:00Z"
+_EXACT_ARTIFACT_FAMILIES = {
+    "keepalive-metrics",
+    "agents-autofix-metrics",
+    "agents-verifier-metrics",
+    "agents-verifier-disposition-metrics",
+}
+_PREFIXED_ARTIFACT_FAMILIES = (
+    "autopilot-metrics-",
+    "issue-optimizer-metrics-",
+    "issue-intake-format-metrics-",
+    "verifier-terminal-disposition-",
+    "review-thread-terminal-disposition-",
+)
+_PATTERNED_ARTIFACT_FAMILIES = (
+    (
+        "bot-comment-auth-coverage-wrapper",
+        re.compile(r"^bot-comment-auth-coverage-wrapper(?:-[A-Za-z0-9][A-Za-z0-9._-]*)?$"),
+    ),
+    (
+        "bot-comment-auth-coverage-reusable",
+        re.compile(r"^bot-comment-auth-coverage-reusable(?:-[A-Za-z0-9][A-Za-z0-9._-]*)?$"),
+    ),
+)
+_MAX_PARSE_ERROR_ROWS = 25
+
+
+@dataclass(frozen=True)
+class ParseErrorDetail:
+    path: str
+    artifact: str
+    artifact_family: str
+    line: int | None
+    reason: str
+
+    def as_dict(self) -> dict[str, Any]:
+        return {
+            "path": self.path,
+            "artifact": self.artifact,
+            "artifact_family": self.artifact_family,
+            "line": self.line,
+            "reason": self.reason,
+        }
 
 
 def _parse_timestamp(value: Any) -> _dt.datetime | None:
@@ -52,29 +97,82 @@ def _gather_metrics_files(metrics_paths: list[str], metrics_dir: str) -> list[Pa
     return sorted(path for path in root.rglob("*.ndjson") if path.is_file())
 
 
-def _read_ndjson(files: Iterable[Path]) -> tuple[list[dict[str, Any]], int]:
+def _artifact_family(artifact: str) -> str:
+    if artifact in _EXACT_ARTIFACT_FAMILIES:
+        return artifact
+    for family, pattern in _PATTERNED_ARTIFACT_FAMILIES:
+        if pattern.match(artifact):
+            return family
+    for prefix in _PREFIXED_ARTIFACT_FAMILIES:
+        if artifact.startswith(prefix):
+            return prefix.rstrip("-")
+    return "unknown"
+
+
+def _infer_artifact_name(path: Path) -> str:
+    parts = path.parts
+    for index, part in enumerate(parts):
+        if part == "agent-metrics" and index > 0:
+            return parts[index - 1]
+    if path.parent.name:
+        return path.parent.name
+    return "unknown"
+
+
+def _parse_error_detail(path: Path, line: int | None, reason: str) -> ParseErrorDetail:
+    artifact = _infer_artifact_name(path)
+    return ParseErrorDetail(
+        path=path.as_posix(),
+        artifact=artifact,
+        artifact_family=_artifact_family(artifact),
+        line=line,
+        reason=reason,
+    )
+
+
+def _read_ndjson(files: Iterable[Path]) -> tuple[list[dict[str, Any]], list[ParseErrorDetail]]:
     entries: list[dict[str, Any]] = []
-    errors = 0
+    errors: list[ParseErrorDetail] = []
     for path in files:
         try:
             handle = path.open("r", encoding="utf-8")
         except OSError:
-            errors += 1
+            errors.append(_parse_error_detail(path, None, "unreadable-file"))
             continue
+        file_entries: list[dict[str, Any]] = []
+        file_errors: list[ParseErrorDetail] = []
+        raw_lines: list[str] = []
         with handle:
-            for line in handle:
+            for line_number, line in enumerate(handle, start=1):
                 raw = line.strip()
                 if not raw:
                     continue
+                raw_lines.append(raw)
                 try:
                     parsed = json.loads(raw)
                 except json.JSONDecodeError:
-                    errors += 1
+                    file_errors.append(_parse_error_detail(path, line_number, "invalid-json"))
                     continue
                 if isinstance(parsed, dict):
-                    entries.append(parsed)
+                    file_entries.append(parsed)
                 else:
-                    errors += 1
+                    file_errors.append(_parse_error_detail(path, line_number, "non-object-json"))
+        if file_errors and not file_entries and raw_lines:
+            try:
+                parsed_file = json.loads("\n".join(raw_lines))
+            except json.JSONDecodeError:
+                pass
+            else:
+                if isinstance(parsed_file, dict):
+                    file_entries.append(parsed_file)
+                    file_errors = []
+                elif isinstance(parsed_file, list) and all(
+                    isinstance(item, dict) for item in parsed_file
+                ):
+                    file_entries.extend(parsed_file)
+                    file_errors = []
+        entries.extend(file_entries)
+        errors.extend(file_errors)
     return entries, errors
 
 
@@ -433,7 +531,52 @@ def _format_rate(numerator: int, denominator: int) -> str:
     return f"{rate:.1f}% ({numerator}/{denominator})"
 
 
-def build_summary(entries: list[dict[str, Any]], errors: int) -> str:
+def _format_parse_error_details(parse_error_details: list[ParseErrorDetail]) -> list[str]:
+    if not parse_error_details:
+        return []
+    family_counts = Counter(detail.artifact_family for detail in parse_error_details)
+    artifact_counts = Counter(detail.artifact for detail in parse_error_details)
+    lines = [
+        "",
+        "## Parse Error Details",
+        f"- By artifact family: {_format_counter(family_counts)}",
+        f"- By artifact: {_format_counter(artifact_counts)}",
+        "",
+        "| Artifact family | Artifact | File | Line | Reason |",
+        "|-----------------|----------|------|------|--------|",
+    ]
+    for detail in parse_error_details[:_MAX_PARSE_ERROR_ROWS]:
+        line = str(detail.line) if detail.line is not None else "n/a"
+        lines.append(
+            "| "
+            f"{detail.artifact_family} | {detail.artifact} | {detail.path} | {line} | "
+            f"{detail.reason} |"
+        )
+    remaining = len(parse_error_details) - _MAX_PARSE_ERROR_ROWS
+    if remaining > 0:
+        lines.append("")
+        lines.append(f"- Additional parse errors omitted from table: {remaining}")
+    return lines
+
+
+def _parse_error_contract(parse_error_details: list[ParseErrorDetail]) -> dict[str, Any]:
+    family_counts = Counter(detail.artifact_family for detail in parse_error_details)
+    artifact_counts = Counter(detail.artifact for detail in parse_error_details)
+    reason_counts = Counter(detail.reason for detail in parse_error_details)
+    return {
+        "count": len(parse_error_details),
+        "by_artifact_family": dict(sorted(family_counts.items())),
+        "by_artifact": dict(sorted(artifact_counts.items())),
+        "by_reason": dict(sorted(reason_counts.items())),
+        "details": [detail.as_dict() for detail in parse_error_details],
+    }
+
+
+def build_summary(
+    entries: list[dict[str, Any]],
+    errors: int,
+    parse_error_details: list[ParseErrorDetail] | None = None,
+) -> str:
     buckets: dict[str, list[dict[str, Any]]] = {
         "keepalive": [],
         "autofix": [],
@@ -479,6 +622,9 @@ def build_summary(entries: list[dict[str, Any]], errors: int) -> str:
         latest = max(timestamps).isoformat().replace("+00:00", "Z")
         lines.append(f"Range: {earliest} to {latest}")
 
+    if parse_error_details:
+        lines.extend(_format_parse_error_details(parse_error_details))
+
     lines.extend(
         [
             "",
@@ -563,25 +709,67 @@ def build_summary(entries: list[dict[str, Any]], errors: int) -> str:
     return "\n".join(lines) + "\n"
 
 
+def build_summary_contract(
+    entries: list[dict[str, Any]], parse_error_details: list[ParseErrorDetail]
+) -> dict[str, Any]:
+    buckets: dict[str, int] = Counter(_classify_entry(entry) for entry in entries)
+    timestamps: list[_dt.datetime] = []
+    for entry in entries:
+        for key in ("timestamp", "created_at", "time", "run_started_at"):
+            ts = _parse_timestamp(entry.get(key))
+            if ts is not None:
+                timestamps.append(ts)
+                break
+    generated_at = (
+        _dt.datetime.now(_dt.UTC).replace(microsecond=0).isoformat().replace("+00:00", "Z")
+    )
+    contract: dict[str, Any] = {
+        "schema": "workflows-agent-metrics-summary/v1",
+        "generated_at": generated_at,
+        "record_count": len(entries),
+        "record_buckets": dict(sorted(buckets.items())),
+        "parse_errors": _parse_error_contract(parse_error_details),
+    }
+    if timestamps:
+        contract["range"] = {
+            "earliest": min(timestamps).isoformat().replace("+00:00", "Z"),
+            "latest": max(timestamps).isoformat().replace("+00:00", "Z"),
+        }
+    return contract
+
+
 def main() -> int:
     metrics_paths_raw = os.environ.get("METRICS_PATHS", "")
     metrics_paths = [item.strip() for item in metrics_paths_raw.split(",") if item.strip()]
     metrics_dir = os.environ.get("METRICS_DIR", _DEFAULT_METRICS_DIR)
     output_path = Path(os.environ.get("OUTPUT_PATH", _DEFAULT_OUTPUT))
+    output_json_path = Path(os.environ.get("OUTPUT_JSON_PATH", _DEFAULT_JSON_OUTPUT))
 
     files = _gather_metrics_files(metrics_paths, metrics_dir)
     if not files:
         output_path.parent.mkdir(parents=True, exist_ok=True)
         output_path.write_text("No metrics files found to aggregate.\n", encoding="utf-8")
+        output_json_path.parent.mkdir(parents=True, exist_ok=True)
+        output_json_path.write_text(
+            json.dumps(build_summary_contract([], []), indent=2, sort_keys=True) + "\n",
+            encoding="utf-8",
+        )
         print("No metrics files found to aggregate.", file=sys.stderr)
         return 0
 
-    entries, errors = _read_ndjson(files)
-    summary = build_summary(entries, errors)
+    entries, parse_error_details = _read_ndjson(files)
+    summary = build_summary(entries, len(parse_error_details), parse_error_details)
+    summary_contract = build_summary_contract(entries, parse_error_details)
 
     output_path.parent.mkdir(parents=True, exist_ok=True)
     output_path.write_text(summary, encoding="utf-8")
+    output_json_path.parent.mkdir(parents=True, exist_ok=True)
+    output_json_path.write_text(
+        json.dumps(summary_contract, indent=2, sort_keys=True) + "\n",
+        encoding="utf-8",
+    )
     print(f"Wrote metrics summary to {output_path}")
+    print(f"Wrote metrics summary JSON to {output_json_path}")
     return 0
 
 

templates/consumer-repo/.github/workflows/agents-81-gate-followups.yml

@@ -402,7 +402,7 @@ jobs:
           tasks_completed=$(( tasks_total - tasks_unchecked ))
           if [ "$tasks_completed" -lt 0 ]; then tasks_completed=0; fi
 
-          metrics_json=$(jq -n \
+          metrics_json=$(jq -cn \
             --arg pr "${PR_NUMBER:-0}" \
             --arg iteration "${ITERATION:-0}" \
             --arg action "${ACTION:-}" \

templates/consumer-repo/.github/workflows/agents-weekly-metrics.yml

@@ -135,6 +135,7 @@ jobs:
         env:
           METRICS_DIR: artifacts
           OUTPUT_PATH: agent-weekly-metrics.md
+          OUTPUT_JSON_PATH: agent-weekly-metrics.json
         run: |
           python scripts/aggregate_agent_metrics.py
 
@@ -228,6 +229,7 @@ jobs:
           name: agent-weekly-metrics
           path: |
             agent-weekly-metrics.md
+            agent-weekly-metrics.json
             artifacts/metric-artifacts-selection.json
             artifacts/metric-artifacts-selection.md
             terminal-disposition-coverage.json