feat: Enhance telemetry store with revision tracking and optimize UI refresh logic

src/observability/store.py

@@ -50,20 +50,26 @@ class TelemetryStore:
         self._spans: deque[SpanRecord] = deque(maxlen=capacity)
         self._metrics: deque[MetricPoint] = deque(maxlen=capacity * 4)
         self._counters: dict[tuple, float] = {}
+        # Monotonic ingest counter — a cheap "has anything changed since I last looked?"
+        # signal so the Telemetry tab can skip recomputing/repainting on an idle tick.
+        self._rev = 0
 
     # ── ingest ──────────────────────────────────────────────────────────────
 
     def add_log(self, record: dict) -> None:
         with self._lock:
             self._logs.append(record)
+            self._rev += 1
 
     def add_span(self, span: SpanRecord) -> None:
         with self._lock:
             self._spans.append(span)
+            self._rev += 1
 
     def add_metric(self, point: MetricPoint, *, counter: bool = False) -> None:
         with self._lock:
             self._metrics.append(point)
+            self._rev += 1
             if counter:
                 key = (point.name, tuple(sorted(point.labels.items())))
                 self._counters[key] = self._counters.get(key, 0.0) + point.value
@@ -78,10 +84,30 @@ class TelemetryStore:
         with self._lock:
             return list(self._spans)[-n:]
 
-    def metric_points(self, name: str | None = None) -> list[MetricPoint]:
+    def metric_points(self, name: str | None = None, limit: int | None = None) -> list[MetricPoint]:
+        """Recorded points, optionally filtered by ``name``.
+
+        With ``limit`` set, only the most recent ``limit`` matching points are returned
+        (still chronological) — this bounds both the scan and the chart payload for
+        high-frequency metrics like agent-turn latency, which would otherwise grow until
+        the buffer is full.
+        """
+        with self._lock:
+            if limit is None:
+                return [p for p in self._metrics if name is None or p.name == name]
+            out: list[MetricPoint] = []
+            for p in reversed(self._metrics):  # newest-first, stop once we have `limit`
+                if name is None or p.name == name:
+                    out.append(p)
+                    if len(out) >= limit:
+                        break
+            out.reverse()
+            return out
+
+    def revision(self) -> int:
+        """Monotonic ingest counter — bumped on every log/span/metric and on ``clear``."""
         with self._lock:
-            points = list(self._metrics)
-        return [p for p in points if name is None or p.name == name]
+            return self._rev
 
     def counter_totals(self) -> dict[str, float]:
         """Cumulative total per metric name, summed across label sets."""
@@ -102,6 +128,7 @@ class TelemetryStore:
             self._spans.clear()
             self._metrics.clear()
             self._counters.clear()
+            self._rev += 1  # a clear is a change too — let the UI repaint to empty
 
 
 def now_ts() -> float:

src/ui/fishbowl/app.py

@@ -276,9 +276,12 @@ except Exception:  # pragma: no cover
 def build_telemetry() -> None:
     """The Telemetry tab: structured log feed + metric charts + per-trace timeline.
 
-    Reads live from the in-memory telemetry store (ADR-0024); a manual Refresh, a
-    3s auto-tick, and the filter dropdowns all repaint it. Clicking a span in the
-    timeline reveals the prompt + memory that agent saw.
+    Reads live from the in-memory telemetry store (ADR-0024). The 3s auto-tick is
+    *non-blocking and idle-cheap*: it gates on the store's revision and returns
+    ``gr.skip()`` (no recompute, no repaint, ``show_progress='hidden'`` so no overlay)
+    whenever nothing was recorded since the last paint. The filter dropdowns and ⟳
+    Refresh force a repaint; "show more traces" pages older traces into the timeline.
+    Clicking a span reveals the prompt + memory that agent saw.
     """
     try:
         from src.ui.fishbowl.render import telemetry as t
@@ -306,24 +309,62 @@ def build_telemetry() -> None:
         row_count=(12, "dynamic"),
         column_count=(5, "fixed"),
     )
-    traces = gr.HTML(t.traces_html())
+    _traces0, _shown0, _total0 = t.render_traces(t.DEFAULT_TRACES)
+    traces = gr.HTML(_traces0)
+    load_more = gr.Button(t.more_label(_shown0, _total0), interactive=_total0 > _shown0, size="sm")
+
+    # Per-user view state: the signature of the last paint (so an idle tick can skip the
+    # work) and how many traces the timeline currently shows ("show more" grows it).
+    sig_state = gr.State(None)
+    trace_limit_state = gr.State(t.DEFAULT_TRACES)
     timer = gr.Timer(3.0)
 
-    def _refresh(level, layer):
-        return (
-            t.kpi_markdown(),
-            t.log_rows(level, layer),
-            t.calls_frame(),
-            t.tokens_frame(),
-            t.latency_frame(),
-            t.traces_html(),
-        )
+    outs = [kpi, feed, calls_plot, tokens_plot, latency_plot, traces, load_more]
 
-    outs = [kpi, feed, calls_plot, tokens_plot, latency_plot, traces]
-    refresh.click(_refresh, [level_dd, layer_dd], outs)
-    timer.tick(_refresh, [level_dd, layer_dd], outs)
-    level_dd.change(_refresh, [level_dd, layer_dd], outs)
-    layer_dd.change(_refresh, [level_dd, layer_dd], outs)
+    def _values(level, layer, trace_limit):
+        """Recompute all panel outputs in one store pass (single counter read)."""
+        snap = t.snapshot(level, layer, trace_limit)
+        more = gr.update(
+            value=t.more_label(snap["trace_shown"], snap["trace_total"]),
+            interactive=snap["trace_total"] > snap["trace_shown"],
+        )
+        return (snap["kpi"], snap["feed"], snap["calls"], snap["tokens"], snap["latency"], snap["traces"], more)
+
+    def _sig(level, layer, trace_limit):
+        # Read the revision BEFORE rendering: if data lands mid-render the stored sig
+        # stays behind, so the next tick repaints to catch up (never a missed update).
+        return (t.revision(), level, layer, int(trace_limit))
+
+    def _tick(level, layer, trace_limit, last_sig):
+        """Auto-refresh: repaint only when the store advanced or a filter/limit changed."""
+        sig = _sig(level, layer, trace_limit)
+        if sig == last_sig:
+            return (gr.skip(),) * (len(outs) + 1)  # idle → leave every output (and sig) as-is
+        return _values(level, layer, trace_limit) + (sig,)
+
+    def _force(level, layer, trace_limit):
+        """Manual Refresh / filter change: always repaint with the current filters."""
+        sig = _sig(level, layer, trace_limit)
+        return _values(level, layer, trace_limit) + (sig,)
+
+    def _load_more(level, layer, trace_limit):
+        """Grow the timeline by one page and repaint (carries the new limit back to state)."""
+        new_limit = int(trace_limit) + t.TRACE_PAGE
+        sig = _sig(level, layer, new_limit)
+        return _values(level, layer, new_limit) + (sig, new_limit)
+
+    # Auto-tick is hidden (no progress overlay → no 3s flicker); user-driven actions show
+    # a minimal indicator so the click registers.
+    timer.tick(_tick, [level_dd, layer_dd, trace_limit_state, sig_state], outs + [sig_state], show_progress="hidden")
+    refresh.click(_force, [level_dd, layer_dd, trace_limit_state], outs + [sig_state], show_progress="minimal")
+    level_dd.change(_force, [level_dd, layer_dd, trace_limit_state], outs + [sig_state], show_progress="minimal")
+    layer_dd.change(_force, [level_dd, layer_dd, trace_limit_state], outs + [sig_state], show_progress="minimal")
+    load_more.click(
+        _load_more,
+        [level_dd, layer_dd, trace_limit_state],
+        outs + [sig_state, trace_limit_state],
+        show_progress="minimal",
+    )
 
 
 # ── scenario registry (assembled from config/, not hardcoded) ───────────────────

src/ui/fishbowl/render/telemetry.py

@@ -71,9 +71,13 @@ def log_rows(level: str = "all", layer: str = "all", limit: int = 250) -> list[l
 # ── metrics (chart data) ────────────────────────────────────────────────────
 
 
-def kpi_markdown() -> str:
-    """A one-line headline of the key counters."""
-    c = obs.telemetry_store().counter_totals()
+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)
@@ -95,9 +99,9 @@ def _df(rows: list[dict], columns: list[str]):
     return pd.DataFrame(rows or [{c: None for c in columns}][:0], columns=columns)
 
 
-def calls_frame():
+def calls_frame(counters: dict | None = None):
     """Counts by metric (calls / tool calls / events / trips) for a bar chart."""
-    c = obs.telemetry_store().counter_totals()
+    c = counters if counters is not None else obs.telemetry_store().counter_totals()
     keep = {
         "llm.calls": "llm calls",
         "tool.calls": "tool calls",
@@ -108,9 +112,9 @@ def calls_frame():
     return _df(rows, ["metric", "count"])
 
 
-def tokens_frame():
+def tokens_frame(counters: dict | None = None):
     """Input vs output tokens for a bar chart."""
-    c = obs.telemetry_store().counter_totals()
+    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))},
@@ -118,9 +122,14 @@ def tokens_frame():
     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 observations over time (seq index → seconds, by agent)."""
-    points = obs.telemetry_store().metric_points("agent.turn.seconds")
+    """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)
     ]
@@ -132,26 +141,36 @@ def latency_frame():
 _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 traces_html(limit_traces: int = 8) -> str:
-    """Recent spans grouped by trace, rendered as an indented timeline.
+def render_traces(limit_traces: int = DEFAULT_TRACES) -> tuple[str, int, int]:
+    """Recent spans grouped by trace, newest first; render the first ``limit_traces``.
 
-    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.
+    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(3000)
+    spans = obs.telemetry_store().recent_spans(_SPAN_SCAN)
     if not spans:
-        return "<div class='tele-empty'>No traces yet — run the show to populate the timeline.</div>"
+        return "<div class='tele-empty'>No traces yet — run the show to populate the timeline.</div>", 0, 0
 
     by_trace: dict[str, list] = {}
     for sp in spans:
         by_trace.setdefault(sp.trace_id, []).append(sp)
     # Most recent traces first (by max end time within the trace).
     ordered = sorted(by_trace.items(), key=lambda kv: max(s.end_ms for s in kv[1]), reverse=True)
+    total = len(ordered)
+    limit = max(0, int(limit_traces))
 
     out: list[str] = []
-    for trace_id, group in ordered[:limit_traces]:
+    for trace_id, group in ordered[:limit]:
         depth = _depth_map(group)
         root_dur = max(s.end_ms for s in group) - min(s.start_ms for s in group)
         out.append(
@@ -161,7 +180,12 @@ def traces_html(limit_traces: int = 8) -> str:
         for sp in sorted(group, key=lambda s: s.start_ms):
             out.append(_span_html(sp, depth.get(sp.span_id, 0)))
         out.append("</div>")
-    return "\n".join(out)
+    return "\n".join(out), min(limit, total), total
+
+
+def traces_html(limit_traces: int = DEFAULT_TRACES) -> str:
+    """The trace timeline HTML alone (no counts) — for callers that don't paginate."""
+    return render_traces(limit_traces)[0]
 
 
 def _depth_map(group: list) -> dict[str, int]:
@@ -210,6 +234,42 @@ def _span_html(sp, depth: int) -> str:
     return head + (f"<div class='tele-span-body'>{body}</div>" if body else "") + "</div>"
 
 
+# ── panel orchestration ─────────────────────────────────────────────────────
+# One entry point the Telemetry tab drives: a single store read per repaint, plus a
+# cheap change signal so an idle auto-tick can skip the work entirely (see app.py).
+
+
+def revision() -> int:
+    """Cheap 'has anything been recorded since I last painted?' signal for the UI."""
+    return obs.telemetry_store().revision()
+
+
+def more_label(shown: int, total: int) -> str:
+    """Label for the 'show more traces' button given how many are shown vs. available."""
+    return f"↓ Show more traces · {shown}/{total}" if total > shown else f"All {total} traces shown"
+
+
+def snapshot(level: str = "all", layer: str = "all", trace_limit: int = DEFAULT_TRACES) -> dict:
+    """Every Telemetry output computed in one pass — one counter read, one span scan.
+
+    Centralising the reads here means the metric panels share a single
+    ``counter_totals()`` snapshot (instead of three separate locks) and the auto-refresh
+    can recompute the whole panel only when :func:`revision` says the store advanced.
+    """
+    counters = obs.telemetry_store().counter_totals()
+    traces, shown, total = render_traces(trace_limit)
+    return {
+        "kpi": kpi_markdown(counters),
+        "feed": log_rows(level, layer),
+        "calls": calls_frame(counters),
+        "tokens": tokens_frame(counters),
+        "latency": latency_frame(),
+        "traces": traces,
+        "trace_shown": shown,
+        "trace_total": total,
+    }
+
+
 TELEMETRY_CSS = """
 .tele-trace { border:1px solid #2e3d25; border-radius:8px; margin:8px 0; padding:8px; background:#0b0f0a; }
 .tele-trace-hd { color:#5fd0d0; font-weight:700; margin-bottom:6px; }