WIP: the live queue serves the whole instance (#51)

packages/cli/src/live.ts

@@ -27,24 +27,25 @@ export function requestLive(commentId: string, intent: LiveIntent = 'ask'): Live
 }
 
 /**
- * Takes the oldest request nobody has picked up, in one statement: two listeners on one session
- * must not both go and answer the same question.
+ * Takes the oldest request nobody has picked up, in one statement: two listeners must not both
+ * go and answer the same question.
+ *
+ * The whole instance is one queue. The file browser is its own session and every ref gets one,
+ * and a reader asking on one session while the only agent sat parked on another waited forever —
+ * the request names its thread, so whichever agent is parked here can take it.
  */
-export function claimNextLiveRequest(sessionId: string): LiveRequest | null {
+export function claimNextLiveRequest(): LiveRequest | null {
   const claimed = queryOne<{ id: string }>(
     `UPDATE comments SET live_claimed_at = datetime('now')
       WHERE id = (
         SELECT c.id FROM comments c
-          JOIN comment_threads t ON t.id = c.thread_id
-         WHERE t.session_id = ?
-           AND c.live_requested_at IS NOT NULL
+         WHERE c.live_requested_at IS NOT NULL
            AND c.live_claimed_at IS NULL
            AND c.live_answered_at IS NULL
          ORDER BY c.live_requested_at ASC, c.rowid ASC
          LIMIT 1
       )
       RETURNING id`,
-    sessionId,
   );
 
   if (!claimed) {
@@ -128,30 +129,17 @@ export function reclaimStaleLiveRequests(olderThanMinutes: number): number {
  * because the connection closing is what ends the wait — which only became true once something
  * listened for it: before that a dead listener stayed counted until its wait ran out.
  *
- * Kept per session rather than per process, because one server holds a whole checkout — the file
- * browser is its own session and each ref gets one. A single set would have every session claiming
- * an agent that is parked on one of them.
+ * One set for the instance, matching the queue: a parked agent answers whichever session asks.
  */
-const listeners = new Map<string, Set<() => void>>();
-
-export function liveListenerCount(sessionId: string): number {
-  return listeners.get(sessionId)?.size ?? 0;
-}
+const listeners = new Set<() => void>();
 
-/** Every parked listener on this instance, whichever session each one waits on. */
+/** Every parked listener on this instance. */
 export function liveListenerTotal(): number {
-  let total = 0;
-  for (const forSession of listeners.values()) {
-    total += forSession.size;
-  }
-  return total;
+  return listeners.size;
 }
 
-export function notifyLiveListeners(sessionId: string | null): void {
-  if (!sessionId) {
-    return;
-  }
-  for (const wake of [...(listeners.get(sessionId) ?? [])]) {
+export function notifyLiveListeners(): void {
+  for (const wake of [...listeners]) {
     wake();
   }
 }
@@ -165,7 +153,6 @@ export function notifyLiveListeners(sessionId: string | null): void {
  * until its wait ran out, and the page went on saying an agent was there for up to that long.
  */
 export function waitForLiveRequest(
-  sessionId: string,
   waitMs: number,
   signal?: AbortSignal,
 ): Promise<LiveRequest | null> {
@@ -174,7 +161,7 @@ export function waitForLiveRequest(
   if (signal?.aborted) {
     return Promise.resolve(null);
   }
-  const claimed = claimNextLiveRequest(sessionId);
+  const claimed = claimNextLiveRequest();
   if (claimed || waitMs <= 0) {
     return Promise.resolve(claimed);
   }
@@ -186,22 +173,18 @@ export function waitForLiveRequest(
         return;
       }
       settled = true;
-      const forSession = listeners.get(sessionId);
-      forSession?.delete(wake);
-      if (forSession?.size === 0) {
-        listeners.delete(sessionId);
-      }
+      listeners.delete(wake);
       clearTimeout(timer);
       signal?.removeEventListener('abort', giveUp);
       resolve(request);
     };
 
     const giveUp = () => finish(null);
 
-    // Only a request this listener could take ends its wait. Waking on anything else would end it
-    // with "nothing asked" and send the agent round the loop for someone else's question.
+    // Only a claim ends the wait early. Waking without one would end it with "nothing asked" and
+    // send the agent round the loop for a question that was never there.
     const wake = () => {
-      const claimed = claimNextLiveRequest(sessionId);
+      const claimed = claimNextLiveRequest();
       if (claimed) {
         finish(claimed);
       }
@@ -210,8 +193,6 @@ export function waitForLiveRequest(
     timer.unref?.();
     signal?.addEventListener('abort', giveUp, { once: true });
 
-    const forSession = listeners.get(sessionId) ?? new Set<() => void>();
-    forSession.add(wake);
-    listeners.set(sessionId, forSession);
+    listeners.add(wake);
   });
 }

packages/cli/src/review-routes.ts

@@ -58,7 +58,7 @@ export function handleReviewRoute(req: IncomingMessage, res: ServerResponse, pat
       if (body.live === true && kind === 'aside') {
         const stamp = requestLive(thread.comments[0].id, body.intent ?? 'ask');
         thread.comments[0].liveRequestedAt = stamp.requestedAt;
-        notifyLiveListeners(stamp.sessionId);
+        notifyLiveListeners();
       }
       sendJson(res, thread);
     });
@@ -76,7 +76,7 @@ export function handleReviewRoute(req: IncomingMessage, res: ServerResponse, pat
       if (body.live === true && kind === 'aside') {
         const stamp = requestLive(comment.id, body.intent ?? 'ask');
         requestedAt = stamp.requestedAt;
-        notifyLiveListeners(stamp.sessionId);
+        notifyLiveListeners();
       }
       sendJson(res, { ...comment, liveRequestedAt: requestedAt } satisfies Comment);
     });

packages/cli/src/server.ts

@@ -67,7 +67,7 @@ import {
 import { findOrCreateSession, resolveSessionId, agentSeenAt, markAgentSeen } from './session.js';
 import { resolveMayChangeCode, type SessionPurpose } from './live-permissions.js';
 import {
-  liveListenerCount,
+
   liveListenerTotal,
   liveWorkingCount,
   pendingLiveCount,
@@ -451,7 +451,8 @@ export function startServer(options: ServerOptions): Promise<ServerResult> {
           const sid = liveSessionId();
           sendJson(res, {
             enabled: isLoopbackBind(getBindHost()),
-            listening: sid ? liveListenerCount(sid) > 0 : false,
+            // Instance-wide like the queue itself: a parked agent answers whichever session asks.
+            listening: liveListenerTotal() > 0,
             working: sid ? liveWorkingCount(sid) > 0 : false,
             waiting: sid ? pendingLiveCount(sid) : 0,
             mayChangeCode: resolveMayChangeCode(purpose, await authorship()),
@@ -512,7 +513,7 @@ export function startServer(options: ServerOptions): Promise<ServerResult> {
           const stopWatching = (): void => clearInterval(viewerWatch);
           req.on('close', stopWatching);
 
-          waitForLiveRequest(sid, waitMs, listenerGone.signal).then(
+          waitForLiveRequest(waitMs, listenerGone.signal).then(
             request => {
               stopWatching();
               // The connection may already be gone; writing to it would throw rather than help.

packages/cli/tests/idle-instance-guards.test.ts

@@ -103,7 +103,7 @@ describe('a server judging whether it is still needed', () => {
 
     const tree = findOrCreateSession('__tree__');
     const abort = new AbortController();
-    const waiting = waitForLiveRequest(tree.id, 5_000, abort.signal);
+    const waiting = waitForLiveRequest(5_000, abort.signal);
 
     expect(liveListenerTotal()).toBe(1);
     expect(