Optimize broadcast handling and debounce WS session reloads

bin/ccam.js

@@ -739,7 +739,12 @@ async function cmdSession(positional) {
   const agents = d.agents || [];
   if (agents.length) renderAgentTree(agents);
 
-  const events = (d.events || []).slice(0, 10);
+  // Fetch recent events separately (no longer embedded in session detail)
+  let events = [];
+  try {
+    const eventData = await get(`/api/events?session_id=${id}&limit=10`);
+    events = (eventData.events || []).slice(0, 10);
+  } catch { /* non-fatal */ }
   if (events.length) renderEventLines(events);
 }
 

client/src/pages/SessionDetail.tsx

@@ -686,12 +686,17 @@ export function SessionDetail() {
                       rootAgents.push(a);
                     }
                   }
-                  // Sort roots and children by started_at ascending (chronological order)
-                  rootAgents.sort((a, b) => (a.started_at || "").localeCompare(b.started_at || ""));
+                  // Sort roots: Main Agent first, then subagents by started_at descending (newest first)
+                  rootAgents.sort((a, b) => {
+                    if (a.type === "main") return -1;
+                    if (b.type === "main") return 1;
+                    return (b.started_at || "").localeCompare(a.started_at || "");
+                  });
+                  // Children: newest subagents first within each parent
                   for (const key of childrenByParent.keys()) {
                     childrenByParent
                       .get(key)!
-                      .sort((a, b) => (a.started_at || "").localeCompare(b.started_at || ""));
+                      .sort((a, b) => (b.started_at || "").localeCompare(a.started_at || ""));
                   }
 
                   // Count all descendants (recursive) for collapsed badge.

client/src/pages/Sessions.tsx

@@ -134,19 +134,31 @@ export function Sessions() {
     setPage(0);
   }, [filter, search, cwd, sortBy, sortDesc]);
 
+  // Debounced WS-driven reload: collapse rapid bursts of session_updated /
+  // new_event messages into a single load() call. Without this, 22+ concurrent
+  // Claude processes issuing hooks each trigger an immediate API request,
+  // saturating the server's event loop and making the UI unresponsive.
   useEffect(() => {
+    let timer: number | null = null;
+    let pendingRunStatus = false;
+    const fire = () => {
+      timer = null;
+      load();
+      if (pendingRunStatus) { pendingRunStatus = false; loadDashboardRuns(); }
+    };
     return eventBus.subscribe((msg) => {
       if (msg.type === "session_created" || msg.type === "session_updated") {
-        load();
+        if (!timer) timer = window.setTimeout(fire, 800);
       }
       if (msg.type === "new_event") {
         const ev = msg.data as DashboardEvent;
         if (ev.event_type === "Stop" || ev.event_type === "SessionEnd") {
-          load();
+          if (!timer) timer = window.setTimeout(fire, 800);
         }
       }
       if (msg.type === "run_status") {
-        loadDashboardRuns();
+        pendingRunStatus = true;
+        if (!timer) timer = window.setTimeout(fire, 800);
       }
     });
   }, [load]);

server/__tests__/api.test.js

@@ -251,7 +251,7 @@ describe("Sessions API", () => {
     assert.equal(res.status, 200);
     assert.equal(res.body.session.id, "sess-1");
     assert.ok(Array.isArray(res.body.agents));
-    assert.ok(Array.isArray(res.body.events));
+    // events are no longer embedded in session detail — use /api/events instead
   });
 
   it("should return 404 for nonexistent session", async () => {

server/__tests__/session-liveness.test.js

@@ -39,6 +39,7 @@ const liveness = require("../lib/session-liveness");
 const hooksRouter = require("../routes/hooks");
 
 const realProbe = liveness.probeLiveCwds;
+const _clearProbeCache = liveness._clearCache;
 
 const enc = (cwd) => cwd.replace(/[^a-zA-Z0-9]/g, "-");
 const PROJECTS = path.join(CLAUDE_HOME, "projects");
@@ -127,6 +128,7 @@ after(() => {
 
 beforeEach(() => {
   liveness.probeLiveCwds = realProbe;
+  _clearProbeCache();
 });
 
 describe("isClaudeCommand — claude CLI process matcher", () => {

server/db.js

@@ -125,7 +125,11 @@ const db = new Database(DB_PATH);
 
 db.pragma("journal_mode = WAL");
 db.pragma("foreign_keys = ON");
-db.pragma("busy_timeout = 5000");
+// Temporary mitigation for lock contention under heavy concurrent writes.
+// 60 s busy_timeout prevents SQLITE_BUSY failures when the WAL writer and
+// checkpoint thread compete with the event loop, but the root cause (long
+// transactions holding the write lock) should be addressed separately.
+db.pragma("busy_timeout = 60000");
 
 db.exec(`
   CREATE TABLE IF NOT EXISTS sessions (

server/lib/broadcast-queue.js

@@ -0,0 +1,130 @@
+/**
+ * @file Deferred broadcast queue that decouples WebSocket notifications from SQLite transactions. Enqueue type+ref pairs inside a transaction, then flush after commit — reducing lock hold time and deduplicating redundant broadcasts.
+ * @author Son Nguyen <hoangson091104@gmail.com>
+ */
+
+const { stmts } = require("../db");
+const { broadcast } = require("../websocket");
+
+/**
+ * Pending broadcast items. Each item is { type, ref } where ref contains
+ * just enough info to load the full payload on flush (sessionId / agentId / inline data).
+ */
+let pending = [];
+
+/**
+ * Enqueue a broadcast to be sent later (outside the current transaction).
+ * Call this inside `db.transaction()` instead of `broadcast()` directly.
+ *
+ * @param {string} type  - WebSocket message type ("session_updated", "agent_updated", etc.)
+ * @param {object} ref   - Reference to locate the data on flush.
+ *   For session/agent types: { sessionId } or { agentId }
+ *   For inline data (e.g. "new_event"): { data } — the full payload, no DB read needed
+ */
+function enqueue(type, ref) {
+  pending.push({ type, ref });
+}
+
+/**
+ * Flush all pending broadcasts: deduplicate, load data from DB,
+ * and call the real `broadcast()`. Should be called outside any transaction
+ * (typically via `setImmediate` after `processEvent` returns).
+ */
+function flush() {
+  if (pending.length === 0) return;
+
+  const items = pending;
+  pending = [];
+
+  // Deduplicate: for DB-backed types, keep only the last entry per (type, id)
+  // but preserve the original insertion order. Inline-data types ("new_event")
+  // are never deduped.
+  // Two-pass approach: first pass records the last index for each dedup key;
+  // second pass filters to only those indices (plus keyless inline items),
+  // preserving the relative order items were enqueued in.
+  const seen = new Map(); // key → last index
+  for (let i = 0; i < items.length; i++) {
+    const key = dedupeKey(items[i]);
+    if (key) seen.set(key, i); // overwrite — keeps the last occurrence
+  }
+  const kept = new Set(seen.values()); // indices of dedup-able items to keep
+  const deduped = [];
+  for (let i = 0; i < items.length; i++) {
+    const key = dedupeKey(items[i]);
+    if (!key || kept.has(i)) {
+      deduped.push(items[i]);
+    }
+  }
+
+  // Send broadcasts (DB reads happen here, outside the transaction).
+  // Each item is dispatched independently so a single failure doesn't
+  // block the rest of the queue or crash the process.
+  for (const item of deduped) {
+    try {
+      const data = loadData(item);
+      if (data !== undefined && data !== null) {
+        broadcast(item.type, data);
+      }
+    } catch (err) {
+      console.error("[broadcast-queue] flush error for %s:", item.type, err?.message || err);
+    }
+  }
+}
+
+/**
+ * Clear all pending items without sending. Useful for error recovery.
+ */
+function clear() {
+  pending = [];
+}
+
+/**
+ * Number of pending items (for diagnostics / testing).
+ */
+function size() {
+  return pending.length;
+}
+
+// ---------------------------------------------------------------------------
+// Internal helpers
+// ---------------------------------------------------------------------------
+
+/**
+ * Build a dedup key for DB-backed types. Returns null for inline-data types.
+ */
+function dedupeKey(item) {
+  const { type, ref } = item;
+  if (!ref) return null; // defensive: null/undefined ref — treat as non-dedup-able
+  if (ref.data !== undefined) return null; // inline payload, don't dedup
+  if (ref.sessionId) return `${type}:session:${ref.sessionId}`;
+  if (ref.agentId) return `${type}:agent:${ref.agentId}`;
+  return null;
+}
+
+/**
+ * Load the full payload for a queued broadcast item.
+ * DB-backed types fetch fresh data; inline types return ref.data directly.
+ */
+function loadData(item) {
+  const { type, ref } = item;
+
+  // Defensive: null/undefined ref means nothing to load
+  if (!ref) return null;
+
+  // Inline payload — no DB read needed
+  if (ref.data !== undefined) return ref.data;
+
+  // Session types
+  if (ref.sessionId) {
+    return stmts.getSession.get(ref.sessionId) || null;
+  }
+
+  // Agent types
+  if (ref.agentId) {
+    return stmts.getAgent.get(ref.agentId) || null;
+  }
+
+  return null;
+}
+
+module.exports = { enqueue, flush, clear, size };

server/lib/session-liveness.js

@@ -23,6 +23,13 @@ const { isInsideContainer } = require("../../scripts/install-hooks");
 
 const UNAVAILABLE = () => ({ available: false, cwds: new Set() });
 
+// Cache probe results for 30s to avoid running ps + lsof every 15s watchdog
+// tick. These commands are synchronous and expensive on macOS with 20+ claude
+// processes, blocking the event loop and starving HTTP requests.
+const PROBE_CACHE_TTL_MS = 30_000;
+let cachedProbe = null;
+let cachedProbeTime = 0;
+
 /**
  * True when a `ps` args string is a Claude Code CLI process. Matches the
  * bare binary (`claude`, `/usr/local/bin/claude`) and interpreter-launched
@@ -60,6 +67,13 @@ function probeLiveCwds() {
   if (process.platform === "win32") return UNAVAILABLE();
   if (isInsideContainer()) return UNAVAILABLE();
 
+  // Return cached result if still fresh (these early-return checks are not
+  // cached because the env/container state may change between calls in tests)
+  const now = Date.now();
+  if (cachedProbe && now - cachedProbeTime < PROBE_CACHE_TTL_MS) return cachedProbe;
+  if (process.platform === "win32") return UNAVAILABLE();
+  if (isInsideContainer()) return UNAVAILABLE();
+
   let psOut;
   try {
     psOut = execFileSync("ps", ["-Ao", "pid=,args="], {
@@ -110,7 +124,10 @@ function probeLiveCwds() {
   for (const line of lsofOut.split("\n")) {
     if (line.startsWith("n") && line.length > 1) cwds.add(path.resolve(line.slice(1)));
   }
-  return { available: true, cwds };
+  const result = { available: true, cwds };
+  cachedProbe = result;
+  cachedProbeTime = now;
+  return result;
 }
 
-module.exports = { probeLiveCwds, isClaudeCommand };
+module.exports = { probeLiveCwds, isClaudeCommand, _clearCache: () => { cachedProbe = null; cachedProbeTime = 0; } };