platform/harness/telemetry.py

"""harness/telemetry.py — usage telemetry v0 (OM-3; principle 7: the improvement loop is
structural). Append-only JSONL of what people actually do — pages opened, explore queries run,
Analyst questions asked — the observe-half of just-in-time modeling: what's used gets promoted
into the governed model; what's ignored gets pruned.

Fail-silent by design: telemetry must never break a page or a query. Local file v0
(data/store/usage.jsonl, tenant data, git-ignored); mined in-session for now, store-backed later.
"""
import json
import sys
import time
import traceback
from pathlib import Path

PATH = Path(__file__).resolve().parents[1] / "data" / "store" / "usage.jsonl"
ERR_PATH = PATH.parent / "errors.jsonl"


def log(kind, user=None, tenant=None, **fields):
    """Append one usage event. Never raises.

    ⭢ WAVE 19 (R4): `user` and `tenant` are FIRST-CLASS PARAMETERS, not just two more `**fields`.
    They already worked as free-form fields — the point of naming them is that attribution stops
    being a convention each call site remembers or forgets. The Analyst's token line is the first
    caller to pass them (`harness/analyst.py`), which is what makes "who spent these tokens, for
    which customer?" answerable AT ALL: before this wave the log recorded a token count attached
    to nobody.

    Both are OPTIONAL and omitted from the line when absent, so a system-initiated run (the eval
    ladder, `harness/routines.py` — neither runs on behalf of a person) writes exactly the record
    it wrote before. A missing key is honest; `"user": null` would invite a reader to count it.

    ⚠ HONEST LIMIT — this file is CONTAINER-LOCAL JSONL (`data/store/usage.jsonl`, git-ignored,
    lost on a Space restart). Attribution starts accruing now; it is not yet a durable, queryable
    usage ledger, which is why the per-user token PANEL stays deferred (DEBT D-8) rather than
    being built against a log that a redeploy erases.
    """
    try:
        PATH.parent.mkdir(parents=True, exist_ok=True)
        rec = {"ts": time.strftime("%Y-%m-%d %H:%M:%S"), "kind": kind}
        if user:
            rec["user"] = str(user)
        if tenant:
            rec["tenant"] = str(tenant)
        rec.update(fields)
        with PATH.open("a", encoding="utf-8") as f:
            f.write(json.dumps(rec, default=str) + "\n")
    except Exception:
        pass


def error(where, exc=None, **fields):
    """THE central error sink (owner rule 2026-07-23: ALL errors are logged). One JSONL line per
    caught error (data/store/errors.jsonl — per-deployment, git-ignored) + a mirror line on
    stderr so it also lands in the Space container logs. Call it from EVERY except block that
    swallows or downgrades an exception — a failure with nowhere to land is how the app breaks
    silently (the 2026-07-23 drawer fire). Never raises."""
    try:
        rec = {"ts": time.strftime("%Y-%m-%d %H:%M:%S"), "where": str(where)[:120], **fields}
        if exc is not None:
            rec["type"] = type(exc).__name__
            rec["msg"] = str(exc)[:400]
            try:
                tb = "".join(traceback.format_exception(type(exc), exc, exc.__traceback__,
                                                        limit=8))
                if tb.strip():
                    rec["trace"] = tb[-1800:]
            except Exception:
                pass
        ERR_PATH.parent.mkdir(parents=True, exist_ok=True)
        with ERR_PATH.open("a", encoding="utf-8") as f:
            f.write(json.dumps(rec, default=str) + "\n")
        print(f"[app-error] {rec['ts']} {rec['where']}: "
              f"{rec.get('type', '')} {rec.get('msg', '')}", file=sys.stderr, flush=True)
    except Exception:
        pass


def read_errors(limit=500):
    """Tail of the error log (newest last) — the Data Health admin section reads this."""
    if not ERR_PATH.exists():
        return []
    out = []
    try:
        with ERR_PATH.open(encoding="utf-8") as f:
            for line in f:
                try:
                    out.append(json.loads(line))
                except json.JSONDecodeError:
                    continue
    except Exception:
        return out
    return out[-limit:]


def read(limit=5000):
    if not PATH.exists():
        return []
    out = []
    with PATH.open(encoding="utf-8") as f:
        for line in f:
            try:
                out.append(json.loads(line))
            except json.JSONDecodeError:
                continue
    return out[-limit:]


def summary():
    """Counts by kind + the most-touched pages/topics — the promote-or-prune shortlist."""
    from collections import Counter
    ev = read()
    kinds = Counter(e.get("kind") for e in ev)
    pages = Counter(e.get("page") for e in ev if e.get("kind") == "page")
    topics = Counter(e.get("topic") for e in ev if e.get("topic"))
    return {"events": len(ev), "by_kind": dict(kinds),
            "top_pages": pages.most_common(10), "top_topics": topics.most_common(10)}