windy-git/api/app/telemetry.py

"""Field telemetry to the admin ledger (admin.windyword.ai) — step 2, 2026-09-23.

Shapes are DECLARED with Telemetry Boss (the ledger owner); the server
quarantines any row that doesn't match, so never add a key or a code here
without re-declaring it first:

  service.boot       once per process start   {commit_sha, version, environment}
  service.health     hourly, in-process       interval_s, uptime_s, requests,
                                              errors_5xx, errors_4xx,
                                              refusals_4xx, p95_ms
  forge.auth.failed  every refused request    {code, http_status, caller,
                                              route?, upstream_status?}

Refusals come first: a refused caller is the most expensive silent failure
("the button did nothing"). An UNAUTHENTICATED caller has no trustworthy id, so
per the all-lanes actor rule the row is actor_type "system", no actor_id, and
the caller class goes in metadata.caller.

Privacy: codes, statuses, route TEMPLATES, counts, durations. Never a passport
number, an email, a token fragment or a concrete path with names in it.

No token → nothing is sent and nothing is buffered. A failed flush keeps the
rows (bounded) and retries on the next tick; it never raises into a request.
"""

from __future__ import annotations

import asyncio
import json
import logging
import time
import urllib.request
from datetime import UTC, datetime

log = logging.getLogger("windy-git.telemetry")

PLATFORM, SERVICE = "windy-git", "api"
FLUSH_EVERY_S = 60
HEALTH_EVERY_S = 3600
MAX_BUFFER = 5000
MAX_LATENCY_SAMPLES = 20000

# The declared forge.auth.failed code enum (Telemetry Boss, 2026-09-23). A code
# outside this set is NOT a refusal row — it counts in errors_* instead.
AUTH_CODES = frozenset(
    {
        "not_signed_in",
        "token_invalid",
        "token_unrecognised",
        "ept_invalid",
        "passport_revoked",
        "passport_unresolvable",
        "agent_read_only",
        "agent_rate_limited",
        "quota_exceeded",
        "trust_unavailable",
        "throttle_unavailable",
        "service_token_invalid",
        "service_auth_unconfigured",
    }
)


def _iso(epoch: float) -> str:
    return datetime.fromtimestamp(epoch, UTC).isoformat().replace("+00:00", "Z")


def caller_class(headers) -> str:
    """Declared values: anonymous_human | anonymous_agent | unknown."""
    from api.app.ept import looks_like_ept

    if headers.get("x-service-token"):
        return "unknown"
    auth = headers.get("authorization") or ""
    if not auth.lower().startswith("bearer "):
        return "unknown"
    return "anonymous_agent" if looks_like_ept(auth.split(" ", 1)[1].strip()) else "anonymous_human"


class Telemetry:
    def __init__(
        self,
        url: str,
        token: str,
        *,
        environment: str = "",
        commit_sha: str | None = None,
        version: str = "",
    ) -> None:
        self.url, self.token = url, token
        self.environment, self.commit_sha, self.version = environment, commit_sha, version
        self.started = time.time()
        self.buffer: list[dict] = []
        self.auth_codes = AUTH_CODES
        self._reset_window()

    @property
    def enabled(self) -> bool:
        return bool(self.token)

    def _reset_window(self) -> None:
        self.window_start = time.time()
        self.requests = self.errors_5xx = self.errors_4xx = self.refusals_4xx = 0
        self.latencies_ms: list[float] = []

    # ---- recording (never raises into a request) --------------------------
    def record_request(self, status: int, duration_ms: float, *, refused: bool = False) -> None:
        self.requests += 1
        if status >= 500:
            self.errors_5xx += 1
        elif refused:
            self.refusals_4xx += 1
        elif status >= 400:
            self.errors_4xx += 1
        if len(self.latencies_ms) < MAX_LATENCY_SAMPLES:
            self.latencies_ms.append(duration_ms)

    def _event(self, event_type: str, metadata: dict, *, ts: float | None = None) -> None:
        if not self.enabled:
            return
        self.buffer.append(
            {
                "ts": _iso(ts or time.time()),
                "platform": PLATFORM,
                "service": SERVICE,
                "event_type": event_type,
                "actor_type": "system",
                "metadata": metadata,
            }
        )
        if len(self.buffer) > MAX_BUFFER:
            del self.buffer[: len(self.buffer) - MAX_BUFFER]

    def boot(self) -> None:
        meta = {"version": self.version, "environment": self.environment}
        if self.commit_sha:  # unknown is absent, never invented (I-12)
            meta["commit_sha"] = self.commit_sha
        self._event("service.boot", meta, ts=self.started)

    def auth_failed(
        self,
        *,
        code: str,
        http_status: int,
        caller: str,
        route: str | None = None,
        upstream_status: int | None = None,
    ) -> None:
        if code not in AUTH_CODES:
            return
        meta: dict = {"code": code, "http_status": int(http_status), "caller": caller}
        if route:
            meta["route"] = route
        if upstream_status is not None:
            meta["upstream_status"] = int(upstream_status)
        self._event("forge.auth.failed", meta)

    def health_row(self) -> dict:
        now = time.time()
        meta = {
            "interval_s": int(now - self.window_start),
            "uptime_s": int(now - self.started),
            "requests": self.requests,
            "errors_5xx": self.errors_5xx,
            "errors_4xx": self.errors_4xx,
            "refusals_4xx": self.refusals_4xx,
        }
        if self.latencies_ms:  # no traffic = no p95, not a fake 0
            s = sorted(self.latencies_ms)
            meta["p95_ms"] = int(s[min(len(s) - 1, int(0.95 * (len(s) - 1) + 0.5))])
        return meta

    def health(self) -> None:
        self._event("service.health", self.health_row())
        self._reset_window()

    # ---- sending ------------------------------------------------------------
    def _post(self, batch: list[dict]) -> int:
        req = urllib.request.Request(
            self.url,
            data=json.dumps({"events": batch}).encode(),
            method="POST",
            headers={
                "Authorization": f"Bearer {self.token}",
                "Content-Type": "application/json",
                "User-Agent": "windy-git-api-telemetry/1",
            },
        )
        with urllib.request.urlopen(req, timeout=20) as r:
            return r.status

    async def flush(self) -> None:
        if not self.enabled or not self.buffer:
            return
        batch = self.buffer[:500]
        try:
            status = await asyncio.to_thread(self._post, batch)
        except Exception as exc:  # noqa: BLE001 - telemetry must never take the API down
            log.warning("telemetry flush failed (%d rows kept): %s", len(self.buffer), exc)
            return
        if 200 <= status < 300:
            del self.buffer[: len(batch)]

    async def run(self) -> None:
        """The one in-process timer: flush every minute, heartbeat every hour."""
        last_health = time.monotonic()
        while True:
            await asyncio.sleep(FLUSH_EVERY_S)
            if time.monotonic() - last_health >= HEALTH_EVERY_S:
                self.health()
                last_health = time.monotonic()
            await self.flush()