telemetry: CI emitter for admin.windyword.ai (inert until token)

ci.run (one row per finished job, exactly once via a high-water mark;
branch_kind default|pr|other so the dashboard can show "main is red") and
service.health (interval counts: finished/failed/cancelled, waiting,
running, runners online, oldest wait). Shapes declared with Windy
Telemetry 40; sends nothing until WINDYGIT_TELEMETRY_TOKEN exists.
State is only advanced after a 2xx, so a failed post retries.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

scripts/cancel_unrunnable.sql

@@ -26,4 +26,16 @@ UPDATE action_run r
        stopped = CASE WHEN r.stopped = 0 THEN extract(epoch from now())::bigint ELSE r.stopped END
  WHERE r.id IN (SELECT DISTINCT run_id FROM dead)
 RETURNING r.id;
+
+-- Jobs BLOCKED on `needs:` inside a run that has already finished (a needed job
+-- failed): Gitea leaves them status 7 forever. They were never going to run;
+-- mark them skipped (4), which is what GitHub shows for the same situation.
+UPDATE action_run_job j
+   SET status = 4, updated = extract(epoch from now())::bigint
+  FROM action_run r
+ WHERE r.id = j.run_id
+   AND j.status = 7
+   AND r.status IN (1, 2, 3)
+   AND to_timestamp(j.created) < now() - interval '30 minutes'
+RETURNING j.run_id;
 COMMIT;

scripts/sync_from_github.sh

@@ -90,5 +90,9 @@ if ! python3 "$(dirname "$0")/pr_status_bridge.py"; then
   log "FAILED pr status bridge"; FAILED=1
 fi
 
+# CI telemetry -> admin.windyword.ai (shapes declared with Windy Telemetry 40).
+# Sends nothing until WINDYGIT_TELEMETRY_TOKEN is set; never fails the sync.
+python3 "$(dirname "$0")/telemetry_emit.py" || log "telemetry emit failed (non-fatal)"
+
 [[ "$FAILED" -ne 0 ]] && { log "COMPLETED WITH FAILURES"; exit 1; }
 log "all repos in step with GitHub"

scripts/telemetry_emit.py

@@ -0,0 +1,196 @@
+#!/usr/bin/env python3
+"""Emit Windy Git CI telemetry to admin.windyword.ai (Windy Telemetry 40's ledger).
+
+Runs on Veron after every sync (root; reads the gitea DB via `docker exec`).
+Shapes are declared with Telemetry 40 (2026-09-23) — do not add keys or enum
+values without re-declaring: a declared family quarantines any row that
+doesn't match.
+
+  ci.run          one row per FINISHED job, exactly once (high-water mark on
+                  action_run_job.id in STATE)
+  service.health  one row per invocation: CI plane counts for the interval
+
+Privacy: ids, names of repos/jobs, codes, counts, durations. No commit
+messages, no logs, no author names.
+
+  --dry-run   print the batch instead of posting (and don't advance STATE)
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import subprocess
+import sys
+import time
+import urllib.error
+import urllib.request
+from datetime import datetime, timezone
+
+INGEST = os.environ.get("TELEMETRY_INGEST_URL", "https://admin.windyword.ai/v1/events")
+TOKEN = os.environ.get("WINDYGIT_TELEMETRY_TOKEN", "")
+STATE = os.environ.get("TELEMETRY_STATE", "/var/lib/windy-git/telemetry-state.json")
+PLATFORM, SERVICE = "windy-git", "ci"
+OUTCOME = {1: "success", 2: "failure", 3: "cancelled", 4: "skipped"}
+EVENTS = {"push", "pull_request", "pull_request_sync", "schedule", "workflow_dispatch"}
+RUNNERS_EXPECTED = 6
+
+
+def sql(query: str) -> list[dict]:
+    """Rows as dicts, via psql's json_agg — no driver needed on the host."""
+    wrapped = f"select coalesce(json_agg(t), '[]'::json) from ({query}) t;"
+    out = subprocess.run(
+        [
+            "docker",
+            "exec",
+            "-i",
+            "windy-git-db-1",
+            "sh",
+            "-c",
+            'psql -U "$POSTGRES_USER" -d gitea -At -v ON_ERROR_STOP=1',
+        ],
+        input=wrapped,
+        capture_output=True,
+        text=True,
+        check=True,
+    ).stdout.strip()
+    return json.loads(out or "[]")
+
+
+def load_state() -> dict:
+    try:
+        with open(STATE) as f:
+            return json.load(f)
+    except (OSError, ValueError):
+        return {}
+
+
+def iso(epoch: float) -> str:
+    return datetime.fromtimestamp(epoch, timezone.utc).isoformat().replace("+00:00", "Z")
+
+
+def main() -> int:
+    dry = "--dry-run" in sys.argv
+    state = load_state()
+    now = time.time()
+    last_job = int(state.get("last_job_id", 0))
+    since = float(state.get("last_ts", now - 300))
+
+    if not last_job:
+        # First run: start at the current high-water mark rather than replaying
+        # a month of history into the ledger as if it happened now.
+        last_job = int(sql("select coalesce(max(id),0) as m from action_run_job")[0]["m"])
+
+    jobs = sql(f"""
+        select j.id, j.name as job, j.status, j.started, j.stopped,
+               p.lower_name as repo, p.default_branch, r.workflow_id, r.event,
+               r.ref, r.index as run, left(r.commit_sha, 7) as sha
+          from action_run_job j
+          join action_run r on r.id = j.run_id
+          join repository p on p.id = r.repo_id
+         where j.id > {last_job} and j.status in (1, 2, 3, 4) and j.stopped > 0
+         order by j.id
+         limit 2000""")
+
+    events = []
+    for j in jobs:
+        ref = j["ref"] or ""
+        if ref.startswith("refs/pull/"):
+            kind = "pr"
+        elif ref == f"refs/heads/{j['default_branch']}":
+            kind = "default"
+        else:
+            kind = "other"
+        dur = (j["stopped"] - j["started"]) * 1000 if j["started"] else None
+        ev = {
+            "ts": iso(j["stopped"]),
+            "platform": PLATFORM,
+            "service": SERVICE,
+            "event_type": "ci.run",
+            "actor_type": "system",
+            "metadata": {
+                "repo": j["repo"],
+                "workflow": (j["workflow_id"] or "").removesuffix(".yml").removesuffix(".yaml"),
+                "job": j["job"],
+                "outcome": OUTCOME[j["status"]],
+                "event": j["event"] if j["event"] in EVENTS else "other",
+                "branch_kind": kind,
+                "run": j["run"],
+                "sha": j["sha"],
+            },
+        }
+        if dur is not None and dur >= 0:
+            ev["duration_ms"] = int(dur)
+        events.append(ev)
+
+    # --- heartbeat: counts since the previous invocation --------------------
+    h = sql(f"""
+        select
+          (select count(*) from action_run_job where stopped >= {int(since)} and status in (1,2,3,4)) as jobs_finished,
+          (select count(*) from action_run_job where stopped >= {int(since)} and status = 2) as jobs_failed,
+          (select count(*) from action_run_job where stopped >= {int(since)} and status = 3) as jobs_cancelled,
+          (select count(*) from action_run_job where status in (5, 7)) as jobs_waiting,
+          (select count(*) from action_run_job where status = 6) as jobs_running,
+          (select count(*) from action_runner where deleted is null and last_online >= {int(now) - 120}) as runners_online,
+          (select coalesce(extract(epoch from now())::bigint - min(created), 0)
+             from action_run_job where status in (5, 7)) as oldest_waiting_s""")[0]
+    meta = {k: int(v) for k, v in h.items()}
+    for k in ("repos_synced", "repos_sync_failed", "statuses_posted", "bridge_errors"):
+        v = os.environ.get(f"TELEMETRY_{k.upper()}")
+        if v is not None and v.isdigit():  # absent = couldn't count; never invent 0
+            meta[k] = int(v)
+    events.append(
+        {
+            "ts": iso(now),
+            "platform": PLATFORM,
+            "service": SERVICE,
+            "event_type": "service.health",
+            "actor_type": "system",
+            "metadata": meta,
+        }
+    )
+
+    if dry:
+        print(json.dumps({"events": events}, indent=1)[:4000])
+        print(f"[telemetry] DRY RUN: {len(events)} events ({len(jobs)} ci.run)")
+        return 0
+    if not TOKEN:
+        print("[telemetry] WINDYGIT_TELEMETRY_TOKEN unset — not sending (not a failure)")
+        return 0
+
+    for i in range(0, len(events), 500):
+        req = urllib.request.Request(
+            INGEST,
+            data=json.dumps({"events": events[i : i + 500]}).encode(),
+            method="POST",
+            headers={
+                "Authorization": f"Bearer {TOKEN}",
+                "Content-Type": "application/json",
+                "User-Agent": "windy-git-telemetry/1",
+            },
+        )
+        try:
+            with urllib.request.urlopen(req, timeout=30) as r:
+                body = r.read()[:300]
+                if r.status >= 300:
+                    raise urllib.error.HTTPError(
+                        INGEST, r.status, body.decode(errors="replace"), None, None
+                    )
+        except urllib.error.HTTPError as e:
+            print(f"[telemetry] FAILED ingest HTTP {e.code}: {e.read()[:200]!r}")
+            return 1  # state NOT advanced: the same rows retry next run
+        except urllib.error.URLError as e:
+            print(f"[telemetry] FAILED ingest: {e.reason}")
+            return 1
+
+    os.makedirs(os.path.dirname(STATE), exist_ok=True)
+    new_last = max([j["id"] for j in jobs], default=last_job)
+    with open(STATE + ".tmp", "w") as f:
+        json.dump({"last_job_id": new_last, "last_ts": now}, f)
+    os.replace(STATE + ".tmp", STATE)
+    print(f"[telemetry] sent {len(events)} events ({len(jobs)} ci.run)")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())