feat(monitor): WS client live stream + apply snapshot/delta to UI

Hand-rolled minimal RFC6455 WebSocket client (~330 LOC) tailored to the
Monitor tab's needs. Single endpoint, text frames, masked client→server,
exponential reconnect backoff (0.5s → 8s cap), thread-safe in/out queues.
TLS is intentionally out of scope (localhost-only). Sec-WebSocket-Accept
verification is skipped — the server is controlled, 101 status is enough.

Files:
- ws_client.{h,cpp}: WsClient with start(host,port,path), drain(), send_text(),
  is_connected(), last_event_ts(). Worker thread does connect + handshake +
  read_loop + reconnect.
- CMakeLists.txt: pulls ws_client.cpp into the dashboard target. ws2_32 was
  already linked for http_client.cpp.
- main.cpp: parses host:port from --api URL, spawns a global WsClient, and
  drains its queue once per render frame via poll_ws(). apply_ws_message()
  routes JSON to g_data.claude:
    snapshot → replace KPIs + recent_executions
    delta    → append rows, increment total_calls / total_errors
  monitor_set_ws_state() forwards connection state + last_event_ts to the
  Monitor toolbar LED.

End-to-end smoke test passed against sqlite_api on localhost:8484:
- Snapshot received with KPIs + 100 recent rows.
- After INSERT INTO calls, delta arrives within ~250ms (server ticker).
- Errors (success=0) propagate correctly and bump the Errors KPI.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

main.cpp

@@ -8,18 +8,135 @@
 #include "data.h"
 #include "data_http.h"
 #include "views.h"
+#include "ws_client.h"
+
+#include "nlohmann/json.hpp"
 
 #include <cstdio>
 #include <cstdlib>
 #include <cstring>
 #include <string>
 #include <fstream>
+#include <vector>
 
 static RegistryData g_data;
 static std::string g_db_path;
 static std::string g_api_url;
 static bool g_loaded = false;
 static bool g_using_http = false;
+static WsClient g_ws;
+
+// Parse "http://host:port" → host, port. Devuelve false si no encaja.
+static bool parse_http_url(const std::string& url, std::string& host, int& port) {
+    static const char* kPrefix = "http://";
+    if (url.rfind(kPrefix, 0) != 0) return false;
+    std::string rest = url.substr(std::strlen(kPrefix));
+    auto colon = rest.find(':');
+    if (colon == std::string::npos) {
+        host = rest;
+        port = 80;
+        return true;
+    }
+    host = rest.substr(0, colon);
+    auto slash = rest.find('/', colon + 1);
+    std::string port_str = (slash == std::string::npos)
+        ? rest.substr(colon + 1)
+        : rest.substr(colon + 1, slash - colon - 1);
+    port = std::atoi(port_str.c_str());
+    return port > 0;
+}
+
+// Aplica un mensaje JSON recibido por WS a g_data.claude. Tipos:
+//   - "snapshot": reemplaza KPIs y la lista entera de recent_executions.
+//   - "delta":    append rows (front), dedup por id, recalcula KPIs.
+// Devuelve true si el mensaje era valido.
+static bool apply_ws_message(const std::string& raw) {
+    using nlohmann::json;
+    json msg = json::parse(raw, nullptr, false);
+    if (!msg.is_object()) return false;
+    const std::string type = msg.value("type", "");
+    if (type != "snapshot" && type != "delta") return false;
+
+    if (msg.contains("server_time") && msg["server_time"].is_number_integer()) {
+        g_data.claude.last_event_ts = msg["server_time"].get<long long>();
+    }
+    if (msg.contains("watermark") && msg["watermark"].is_number_integer()) {
+        long long w = msg["watermark"].get<long long>();
+        if (w > g_data.claude.last_seen_call_id) g_data.claude.last_seen_call_id = w;
+    }
+
+    // Snapshot reemplaza KPIs. Delta los actualiza por incremento.
+    if (type == "snapshot" && msg.contains("stats") && msg["stats"].is_object()) {
+        const auto& s = msg["stats"];
+        g_data.claude.total_calls      = s.value("total_calls", 0);
+        g_data.claude.total_errors     = s.value("total_errors", 0);
+        g_data.claude.total_violations = s.value("total_violations", 0);
+        g_data.claude.total_copies     = s.value("total_copies", 0);
+        g_data.claude.total_versions   = s.value("total_versions", 0);
+        g_data.claude.available        = true;
+    }
+
+    if (!msg.contains("calls") || !msg["calls"].is_array()) return true;
+
+    if (type == "snapshot") {
+        g_data.claude.recent_executions.clear();
+    }
+
+    // Construye filas nuevas
+    std::vector<RecentExecutionRow> incoming;
+    incoming.reserve(msg["calls"].size());
+    int new_errors = 0;
+    for (const auto& c : msg["calls"]) {
+        RecentExecutionRow row;
+        row.id          = c.value("id", 0LL);
+        row.ts          = c.value("ts", 0LL);
+        row.function_id = c.value("function_id", "");
+        row.tool_used   = c.value("tool_used", "");
+        row.duration_ms = c.value("duration_ms", 0);
+        row.success     = c.value("success", true);
+        row.error_class = c.value("error_class", "");
+        row.session_id  = c.value("session_id", "");
+        if (!row.success) new_errors++;
+        incoming.push_back(std::move(row));
+    }
+
+    if (type == "delta") {
+        g_data.claude.total_calls  += static_cast<int>(incoming.size());
+        g_data.claude.total_errors += new_errors;
+    }
+
+    // Prepend (newer al frente). Para delta: filas vienen ASC del server,
+    // las anadimos al frente en orden inverso. Para snapshot: ya vienen DESC.
+    if (type == "delta") {
+        for (auto it = incoming.rbegin(); it != incoming.rend(); ++it) {
+            g_data.claude.recent_executions.insert(
+                g_data.claude.recent_executions.begin(), std::move(*it));
+        }
+    } else {
+        for (auto& row : incoming) {
+            g_data.claude.recent_executions.push_back(std::move(row));
+        }
+    }
+
+    // Cap list (UI muestra ~100). Evita crecer indefinidamente con deltas.
+    const size_t kCap = 200;
+    if (g_data.claude.recent_executions.size() > kCap) {
+        g_data.claude.recent_executions.resize(kCap);
+    }
+    return true;
+}
+
+static void poll_ws() {
+    bool connected = g_ws.is_connected();
+    long long ts = g_ws.last_event_ts();
+    monitor_set_ws_state(connected, ts);
+
+    std::vector<std::string> msgs;
+    g_ws.drain(msgs, 32);
+    for (const auto& m : msgs) {
+        apply_ws_message(m);
+    }
+}
 
 static void reload_data() {
     // Conservar la ventana del Monitor entre recargas (no se pierde al refrescar).
@@ -73,6 +190,11 @@ static void render() {
         reload_monitor();
     }
 
+    // Issue 0086: drena la cola de mensajes WS y aplica deltas a g_data.
+    // No bloquea — drain es O(N) sobre los mensajes encolados desde el
+    // ultimo frame (tipicamente 0-3).
+    poll_ws();
+
     if (!g_loaded) {
         fullscreen_window_begin("##error");
         ImGui::TextColored(ImVec4(1, 0.3f, 0.3f, 1),
@@ -187,6 +309,17 @@ int main(int argc, char** argv) {
 
     reload_data();
 
+    // Issue 0086: lanza el cliente WS al hub de eventos. El hub solo arranca
+    // su ticker cuando recibe el primer subscriber, asi que esta conexion
+    // tambien le dice al servidor "empieza a streamear".
+    {
+        std::string ws_host;
+        int ws_port = 0;
+        if (parse_http_url(g_api_url, ws_host, ws_port)) {
+            g_ws.start(ws_host, ws_port, "/api/events/call_monitor");
+        }
+    }
+
     return fn::run_app(
         {.title     = "fn_registry Dashboard",
          .width     = 1600,