prose-contracts/scripts/health-log-freshness.py

#!/usr/bin/env python3
"""Health-log freshness watchdog (dead-man's-switch).

Absence of logs raises no alarm. On 2026-09-26 the `gpu/` directory of
SyslogSolution/health-logs went silent for 12 days unnoticed, because the only
thing that would have noticed was the job that had stopped. This check lives on
a DIFFERENT host (CT 100) from the producers, so a producer host that is dead,
or a schedule that was deleted, still raises an alarm.

It asks the Gitea API for the newest commit touching each watched directory and
fails when that commit is older than the directory's threshold.

Usage:
    health-log-freshness.py            # check every watched directory
    health-log-freshness.py --json     # machine-readable output

Exit: 0 = all fresh, 1 = at least one stale or unreachable, 2 = check could not run.

Environment:
    GITEA_URL            default https://git.sysloggh.net
    GITEA_TOKEN          API token (falls back to GITEA_PAT, then to the token
                         embedded in ~/.git-credentials for that host)
    HEALTH_LOG_MAX_AGE_H override thresholds, e.g. HEALTH_LOG_MAX_AGE_GPU=12
"""

from __future__ import annotations

import base64
import json
import os
import re
import sys
import urllib.error
import urllib.request
from datetime import datetime, timezone

REPO = "SyslogSolution/health-logs"
GITEA_URL = os.environ.get("GITEA_URL", "https://git.sysloggh.net").rstrip("/")

# dir -> (threshold_hours, why)
# Thresholds are sized for the producer cadence plus one missed run:
#   gpu/  runs every 6h  -> 12h tolerates one miss, catches a second
#   litellm/ runs every 6h -> 18h (it is proven healthy; a looser bound avoids
#            noise while still catching a real stop)
WATCHED: dict[str, tuple[float, str]] = {
    "gpu": (12.0, "gpu-self-heal.py, CT116 cron 2 */6 * * *"),
    "litellm": (18.0, "litellm-health-check.sh, CT116 cron 0 */6 * * *"),
}


def _threshold(directory: str, default: float) -> float:
    """Allow HEALTH_LOG_MAX_AGE_<DIR> to override a threshold.

    Documented override; used both operationally (tighten a bound while
    investigating) and in tests (force a stale verdict deterministically).
    """
    raw = os.environ.get(f"HEALTH_LOG_MAX_AGE_{directory.upper()}")
    if raw is None:
        return default
    try:
        return float(raw)
    except ValueError:
        print(f"WARN: ignoring non-numeric HEALTH_LOG_MAX_AGE_{directory.upper()}={raw!r}")
        return default

# pm2/ is deliberately NOT watched: pm2-self-heal.prose.md claimed a health-logs
# posting that its executor never implemented, and the claim was retired on
# 2026-09-26 in favour of the contract-runner's per-run logs. The directory is
# left as historical evidence, not as a live obligation.


def _auth_candidates() -> list[str]:
    """Authorization header values to try, in order.

    Gitea accepts either an API token (``token <tok>``) or HTTP basic auth, and
    the value held under ``GITEA_TOKEN`` is not reliably an API token - on this
    host it is the git account's *password*, which sent as a token returns HTTP
    401. So return every candidate and let the caller use the first that works,
    rather than guessing and failing.
    """
    candidates: list[str] = []
    for var in ("GITEA_TOKEN", "GITEA_PAT"):
        if os.environ.get(var):
            candidates.append(f"token {os.environ[var]}")
            candidates.append(
                "Basic "
                + base64.b64encode(
                    f"{_git_user()}:{os.environ[var]}".encode()
                ).decode()
            )
    # ~/.git-credentials may hold this Gitea under several hostnames (the public
    # name and the internal IP both appear in this fleet), so accept any of them
    # rather than filtering on the configured host - that filter produced zero
    # candidates whenever GITEA_URL pointed at the internal address.
    try:
        with open(os.path.expanduser("~/.git-credentials")) as fh:
            for line in fh:
                m = re.match(r"https://([^:]+):([^@]+)@", line.strip())
                if m:
                    raw = f"{m.group(1)}:{m.group(2)}".encode()
                    candidates.append("Basic " + base64.b64encode(raw).decode())
    except OSError:
        pass
    seen, out = set(), []
    for c in candidates:
        if c not in seen:
            seen.add(c)
            out.append(c)
    return out


def _git_user() -> str:
    return os.environ.get("GITEA_USER", "abiba-bot")


def newest_commit_iso(directory: str, auth: list[str] | None) -> tuple[str | None, str]:
    """Return (iso_timestamp, detail) for the newest commit touching `directory`."""
    url = (
        f"{GITEA_URL}/api/v1/repos/{REPO}/commits"
        f"?path={directory}&limit=1&stat=false"
    )
    req = urllib.request.Request(url, headers={"Accept": "application/json"})
    headers = list(auth or [])
    last = "no credential"
    for i, hdr in enumerate(headers or [None]):
        r = urllib.request.Request(url, headers={"Accept": "application/json"})
        if hdr:
            r.add_header("Authorization", hdr)
        try:
            with urllib.request.urlopen(r, timeout=20) as resp:
                data = json.loads(resp.read().decode("utf-8", "replace"))
            break
        except urllib.error.HTTPError as exc:
            last = f"HTTP {exc.code}"
            if exc.code not in (401, 403):
                return None, last
        except Exception as exc:  # noqa: BLE001
            return None, repr(exc)
    else:
        return None, last

    if not data:
        return None, "no commits"
    commit = data[0].get("commit", {})
    when = (
        (commit.get("committer") or {}).get("date")
        or (commit.get("author") or {}).get("date")
    )
    msg = (commit.get("message") or "").splitlines()[0][:60]
    return when, msg


def main() -> int:
    as_json = "--json" in sys.argv
    auth = _auth_candidates()
    now = datetime.now(timezone.utc)
    failures: list[str] = []
    report: dict[str, dict] = {}

    if not auth:
        print("FAIL: no Gitea credential available (GITEA_TOKEN/GITEA_PAT/~/.git-credentials)")
        return 2
    for directory, (default_age_h, why) in WATCHED.items():
        max_age_h = _threshold(directory, default_age_h)
        when, detail = newest_commit_iso(directory, auth)
        entry: dict = {"directory": directory, "producer": why, "max_age_h": max_age_h}

        if when is None:
            entry.update(ok=False, reason=f"could not read newest commit: {detail}")
            failures.append(
                f"health-logs/{directory}/ could not be read ({detail}) — "
                f"a directory that cannot be read is indistinguishable from one that stopped"
            )
        else:
            try:
                ts = datetime.fromisoformat(when.replace("Z", "+00:00"))
            except ValueError:
                entry.update(ok=False, reason=f"unparseable timestamp {when!r}")
                failures.append(f"health-logs/{directory}/ timestamp unparseable: {when!r}")
                report[directory] = entry
                continue
            age_h = (now - ts).total_seconds() / 3600.0
            stale = age_h > max_age_h
            entry.update(
                ok=not stale,
                newest=when,
                age_h=round(age_h, 2),
                newest_commit=detail,
            )
            if stale:
                entry["reason"] = f"stale: {age_h:.1f}h > {max_age_h}h"
                failures.append(
                    f"health-logs/{directory}/ is STALE: newest entry {when} "
                    f"({age_h:.1f}h old, limit {max_age_h}h) from {why}"
                )
        report[directory] = entry

    if as_json:
        print(json.dumps({"checked": report, "failures": failures}, indent=2))
        return 1 if failures else 0

    print("Health-log freshness — dead-man's-switch")
    print("=" * 66)
    for directory, entry in report.items():
        mark = "✅" if entry.get("ok") else "❌"
        if entry.get("newest"):
            print(
                f"{mark} health-logs/{directory}/  newest {entry['newest']} "
                f"({entry['age_h']}h old, limit {entry['max_age_h']}h)"
            )
            print(f"     last commit: {entry.get('newest_commit')}")
        else:
            print(f"{mark} health-logs/{directory}/  {entry.get('reason')}")
        print(f"     producer: {entry['producer']}")

    print("=" * 66)
    if failures:
        print("VERDICT: FAIL")
        for f in failures:
            print(f"  - {f}")
        return 1
    print("VERDICT: PASS — every watched health-log directory is advancing")
    return 0


if __name__ == "__main__":
    sys.exit(main())