"""Telemetry tab renderers — read the in-memory store, shape it for Gradio. Pure functions over :func:`src.observability.telemetry_store`: a filterable log feed, metric dataframes for the charts, and a per-trace timeline that surfaces the prompt + memory each agent saw. No Gradio components are created here (the app shell owns those, Unit 9 in ``app.py``); these just produce the markdown / HTML / dataframes the components render. See ADR-0024. """ from __future__ import annotations import html from typing import Any from src import observability as obs #: Layer prefixes used by the feed's layer filter (matched against the ``logger`` #: name and the ``event`` namespace). LAYERS = [ "all", "llm", "agent", "memory", "ledger", "event", "projection", "tool", "governor", "router", "session", "modal", "run", "config", "manifest", "context", ] LEVELS = ["all", "DEBUG", "INFO", "WARNING", "ERROR"] def _short(value: Any, limit: int = 160) -> str: text = value if isinstance(value, str) else repr(value) return text if len(text) <= limit else text[: limit - 1] + "…" # ── log feed ────────────────────────────────────────────────────────────────── def log_rows(level: str = "all", layer: str = "all", limit: int = 250) -> list[list[str]]: """Recent structured logs as table rows: [time, level, agent/turn, event, detail].""" rows: list[list[str]] = [] for rec in reversed(obs.telemetry_store().recent_logs(2000)): lvl = str(rec.get("level", "")) event = str(rec.get("event", "")) if level != "all" and lvl != level: continue if layer != "all" and not (event.startswith(layer) or str(rec.get("logger", "")).startswith(layer)): continue ctx = "/".join(str(rec[k]) for k in ("agent", "turn") if rec.get(k) is not None) if not ctx: continue # Detail = the most informative extra fields (skip the bookkeeping keys). skip = {"ts", "level", "logger", "event", "msg", "src", "run_id", "turn", "agent"} detail = " ".join(f"{k}={_short(v, 80)}" for k, v in rec.items() if k not in skip) ts = str(rec.get("ts", ""))[-12:] rows.append([ts, lvl, ctx, event, _short(detail, 200)]) if len(rows) >= limit: break return rows # ── metrics (chart data) ──────────────────────────────────────────────────── def kpi_markdown(counters: dict | None = None) -> str: """A one-line headline of the key counters. Pass a pre-read ``counters`` snapshot (see :func:`snapshot`) to avoid re-locking the store; omit it and one is fetched. """ c = counters if counters is not None else obs.telemetry_store().counter_totals() calls = int(c.get("llm.calls", 0)) tin, tout = int(c.get("llm.tokens.input", 0)), int(c.get("llm.tokens.output", 0)) cost = c.get("llm.cost_usd", 0.0) tools = int(c.get("tool.calls", 0)) trips = int(c.get("governor.trips", 0)) events = int(c.get("ledger.events", 0)) return ( f"### Telemetry — live\n" f"**LLM calls** {calls} · **tokens** {tin:,} in / {tout:,} out · " f"**cost** ${cost:.4f} · **tool calls** {tools} · " f"**events** {events} · **governor trips** {trips}" ) def _df(rows: list[dict], columns: list[str]): """Build a pandas DataFrame (Gradio plots want one); empty-safe.""" import pandas as pd return pd.DataFrame(rows or [{c: None for c in columns}][:0], columns=columns) def calls_frame(counters: dict | None = None): """Counts by metric (calls / tool calls / events / trips) for a bar chart.""" c = counters if counters is not None else obs.telemetry_store().counter_totals() keep = { "llm.calls": "llm calls", "tool.calls": "tool calls", "ledger.events": "events", "governor.trips": "gov trips", } rows = [{"metric": label, "count": float(c.get(key, 0))} for key, label in keep.items()] return _df(rows, ["metric", "count"]) def tokens_frame(counters: dict | None = None): """Input vs output tokens for a bar chart.""" c = counters if counters is not None else obs.telemetry_store().counter_totals() rows = [ {"kind": "input", "tokens": float(c.get("llm.tokens.input", 0))}, {"kind": "output", "tokens": float(c.get("llm.tokens.output", 0))}, ] return _df(rows, ["kind", "tokens"]) #: Cap on how many latency observations the line chart plots — a rolling window that #: keeps both the store scan and the browser-side chart light on long runs. _LATENCY_POINTS = 300 def latency_frame(): """Agent-turn latency over time (rolling window: seq index → seconds, by agent).""" points = obs.telemetry_store().metric_points("agent.turn.seconds", limit=_LATENCY_POINTS) rows = [ {"n": i, "seconds": round(p.value, 4), "agent": str(p.labels.get("agent", "?"))} for i, p in enumerate(points) ] return _df(rows, ["n", "seconds", "agent"]) # ── trace timeline ──────────────────────────────────────────────────────────── _PROMPT_KEYS = ("llm.prompt", "agent.prompt", "memory.query") _OUTPUT_KEYS = ("llm.completion", "memory.visible_count") #: How many traces the timeline shows at first, and how many more each "show more" adds. DEFAULT_TRACES = 8 TRACE_PAGE = 8 #: Span buffer slice scanned to assemble the timeline (≤ store capacity). _SPAN_SCAN = 3000 def render_traces(limit_traces: int = DEFAULT_TRACES) -> tuple[str, int, int]: """Recent spans grouped by trace, newest first; render the first ``limit_traces``. Returns ``(html, shown, total)``. Each span shows name + duration + status; spans that carry a prompt or memory (``llm.prompt`` / ``agent`` / ``memory.*`` attributes) expand to reveal exactly what the agent sent and saw — the heart of the 'what did each agent do' view. Only ``limit_traces`` traces are turned into HTML, so the payload stays bounded while ``total`` lets the caller offer a "show more" control. """ spans = obs.telemetry_store().recent_spans(_SPAN_SCAN) if not spans: return "
{html.escape(str(attrs[key]))}{html.escape(str(attrs['llm.completion']))}