feat(observability): add runtime-state gauges for SSE, circuits, list-cache (#4177)

Three subsystems hold meaningful per-pod runtime state but emitted no
metric for it. Add OpenTelemetry observable gauges (sampled at scrape
time, exported via the existing Prometheus reader on /metrics):

- sse_hub.connections — active SSE listeners (bounded by
  MAX_TOTAL_CONNECTIONS=500; chart vs the cap, catch cleanup leaks)
- mcp_circuit_breaker.circuits{state=open|half_open} — MCP connection
  circuits by state; a rising `open` count = downstream servers
  unreachable / requests failing fast
- mcp_list_cache.pending_revalidations — in-flight background list
  revalidations; complements the existing fetches counter

All are thin reads over already-held state — no new tracking, no hot-path
cost beyond an O(n≤cap) scan per scrape for the circuit gauge.

Author: pedrofrxncx

apps/mesh/src/event-bus/sse-hub.ts

@@ -17,6 +17,8 @@
  * - Pluggable broadcast: strategy handles cross-process replication
  */
 
+import { meter } from "../observability";
+
 // ============================================================================
 // Types
 // ============================================================================
@@ -234,3 +236,13 @@ function matchesAnyPattern(eventType: string, patterns: string[]): boolean {
 
 /** Global SSE hub instance */
 export const sseHub = new SSEHub();
+
+// Live SSE connection count (sampled at scrape time). Bounded by
+// MAX_TOTAL_CONNECTIONS — chart against that cap to spot saturation, and watch
+// for a count that never drops (a listener-cleanup leak).
+meter
+  .createObservableGauge("sse_hub.connections", {
+    description: "Active SSE listeners held by this pod",
+    unit: "{connections}",
+  })
+  .addCallback((r) => r.observe(sseHub.count));

apps/mesh/src/mcp-clients/circuit-breaker.ts

@@ -15,6 +15,7 @@ import {
   CIRCUIT_BREAKER_FAILURE_THRESHOLD,
   CIRCUIT_BREAKER_MAX_ENTRIES,
 } from "../core/constants";
+import { meter } from "../observability";
 
 type CircuitState = "CLOSED" | "OPEN" | "HALF_OPEN";
 
@@ -27,6 +28,26 @@ interface CircuitEntry {
 
 const circuits = new Map<string, CircuitEntry>();
 
+// Live circuit counts by state (sampled at scrape time). A rising `open` count
+// means downstream MCP servers are unreachable and requests are failing fast —
+// the signal an operator wants when proxied tools start erroring. O(n) over the
+// map (n ≤ CIRCUIT_BREAKER_MAX_ENTRIES) per scrape.
+meter
+  .createObservableGauge("mcp_circuit_breaker.circuits", {
+    description: "MCP connection circuits by state (open, half_open)",
+    unit: "{circuits}",
+  })
+  .addCallback((r) => {
+    let open = 0;
+    let halfOpen = 0;
+    for (const c of circuits.values()) {
+      if (c.state === "OPEN") open++;
+      else if (c.state === "HALF_OPEN") halfOpen++;
+    }
+    r.observe(open, { state: "open" });
+    r.observe(halfOpen, { state: "half_open" });
+  });
+
 export class CircuitOpenError extends Error {
   readonly cooldownRemainingMs: number;
   constructor(connectionId: string, cooldownRemainingMs: number) {

apps/mesh/src/mcp-clients/mcp-list-cache.ts

@@ -97,6 +97,17 @@ export class JetStreamKVMcpListCache implements McpListCache {
 // Module-level revalidation tracking (prevents thundering herd)
 const revalidating = new Set<string>();
 
+// In-flight background list revalidations (sampled at scrape time). Sits
+// alongside the mcp_list_cache.fetches counter: a count that stays high means
+// upstream list calls aren't keeping up with staleness — refresh load worth
+// watching next to hit/miss/stale rates.
+meter
+  .createObservableGauge("mcp_list_cache.pending_revalidations", {
+    description: "In-flight background MCP list revalidations on this pod",
+    unit: "{revalidations}",
+  })
+  .addCallback((r) => r.observe(revalidating.size));
+
 // Per-(type,connection) timestamp of the last revalidation we scheduled, used to
 // throttle background revalidation. Without it, every cache hit reconnects
 // downstream — so a UI polling tools/list every few seconds hammers the upstream