validate.py

#!/usr/bin/env python3
"""Validate telemetry records (JSONL) against the field contract.

Stdlib only (Python 3.9+). Checks shape, types, patterns, the latency
identity, and re-computes cost_usd from token counts and a price table.

Usage:
    python3 validate.py samples/sample_events.jsonl
    python3 validate.py my_capture.jsonl --prices prices.json

prices.json maps model -> {"in": USD/token, "cached": USD/token, "out": USD/token}.
Without --prices, the built-in table for the bundled samples is used.
"""

from __future__ import annotations

import argparse
import json
import re
import sys

REQ_ID = re.compile(r"^[a-zA-Z0-9_-]{8,64}$")
TS = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(\.\d+)?Z$")

# USD per token, for the bundled samples only.
DEFAULT_PRICES = {
    "sample-model-a": {"in": 5e-7, "cached": 1e-7, "out": 2e-6},
}


def check_record(rec: dict, line_no: int, prices: dict) -> list[str]:
    errs = []

    def err(msg: str) -> None:
        errs.append(f"line {line_no}: {msg}")

    required = [
        "request_id", "model", "prompt_tokens", "completion_tokens",
        "cached_tokens", "latency_ms", "cost_usd", "failover", "ts",
    ]
    missing = [k for k in required if k not in rec]
    extra = [k for k in rec if k not in required]
    if missing:
        err(f"missing fields: {missing}")
    if extra:
        err(f"unexpected fields: {extra}")
    if errs:
        return errs

    if not isinstance(rec["request_id"], str) or not REQ_ID.match(rec["request_id"]):
        err("request_id must match ^[a-zA-Z0-9_-]{8,64}$")
    if not isinstance(rec["model"], str) or not rec["model"]:
        err("model must be a non-empty string")
    for k in ("prompt_tokens", "completion_tokens", "cached_tokens"):
        v = rec[k]
        if not isinstance(v, int) or isinstance(v, bool) or v < 0:
            err(f"{k} must be an integer >= 0")
    if isinstance(rec["cached_tokens"], int) and isinstance(rec["prompt_tokens"], int):
        if rec["cached_tokens"] > rec["prompt_tokens"]:
            err("cached_tokens cannot exceed prompt_tokens")
    if not isinstance(rec["failover"], bool):
        err("failover must be a boolean")
    if not isinstance(rec["ts"], str) or not TS.match(rec["ts"]):
        err("ts must be ISO-8601 UTC (e.g. 2026-07-21T08:14:03.512Z)")
    if not isinstance(rec["cost_usd"], (int, float)) or isinstance(rec["cost_usd"], bool) or rec["cost_usd"] < 0:
        err("cost_usd must be a number >= 0")

    lat = rec["latency_ms"]
    if not isinstance(lat, dict) or set(lat) != {"gateway_overhead", "upstream", "total"}:
        err("latency_ms must contain exactly gateway_overhead, upstream, total")
    else:
        if any(not isinstance(v, int) or isinstance(v, bool) or v < 0 for v in lat.values()):
            err("latency_ms values must be integers >= 0")
        elif lat["total"] != lat["gateway_overhead"] + lat["upstream"]:
            err("latency_ms.total != gateway_overhead + upstream")

    # Re-compute cost if a price entry exists for this model.
    p = prices.get(rec["model"])
    if p and all(isinstance(rec[k], int) for k in ("prompt_tokens", "completion_tokens", "cached_tokens")):
        fresh = rec["prompt_tokens"] - rec["cached_tokens"]
        expected = fresh * p["in"] + rec["cached_tokens"] * p["cached"] + rec["completion_tokens"] * p["out"]
        if abs(expected - rec["cost_usd"]) > 1e-9:
            err(f"cost_usd {rec['cost_usd']} does not match recomputed {expected:.6f}")

    return errs


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("file", help="JSONL file with one telemetry record per line")
    ap.add_argument("--prices", help="JSON file with per-model unit prices (USD per token)")
    args = ap.parse_args()

    prices = DEFAULT_PRICES
    if args.prices:
        with open(args.prices, encoding="utf-8") as f:
            prices = json.load(f)

    all_errs: list[str] = []
    n = 0
    with open(args.file, encoding="utf-8") as f:
        for i, line in enumerate(f, start=1):
            line = line.strip()
            if not line:
                continue
            n += 1
            try:
                rec = json.loads(line)
            except json.JSONDecodeError as e:
                all_errs.append(f"line {i}: invalid JSON ({e})")
                continue
            if not isinstance(rec, dict):
                all_errs.append(f"line {i}: record must be a JSON object")
                continue
            all_errs.extend(check_record(rec, i, prices))

    if all_errs:
        for e in all_errs:
            print(f"INVALID: {e}", file=sys.stderr)
        print(f"FAIL: {n} records, {len(all_errs)} problems", file=sys.stderr)
        return 1

    print(f"OK: {n} records, all fields valid, recomputed cost matches cost_usd (6 dp)")
    return 0


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