feat: the live queue serves the whole instance

packages/api/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@diffity/api",
-  "version": "0.10.13",
+  "version": "0.10.14",
   "private": true,
   "type": "module",
   "main": "./dist/index.js",

packages/cli/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@naturalcycles/diffity",
-  "version": "0.10.13",
+  "version": "0.10.14",
   "description": "Agent-agnostic, GitHub-style diff viewer and code review tool with a live agent loop",
   "type": "module",
   "bin": {

packages/cli/src/agent.ts

@@ -129,16 +129,17 @@ function assertFileExists(filePath: string): void {
   }
 }
 
-function resolveThreadId(shortId: string, sessionId: string): Thread {
+/**
+ * Any thread on this instance, whichever session holds it: the live queue is instance-wide, so the
+ * request an agent took may sit on a session other than the one it is parked on, and the reply has
+ * to be allowed to follow it there. Tours stay with their session.
+ */
+function resolveThreadId(shortId: string): Thread {
   const thread = getThread(shortId);
   if (!thread) {
     console.error(pc.red(`Error: Thread not found: ${shortId}`));
     process.exit(1);
   }
-  if (thread.sessionId !== sessionId) {
-    console.error(pc.red(`Error: Thread ${shortId} does not belong to current session`));
-    process.exit(1);
-  }
   return thread;
 }
 
@@ -340,8 +341,8 @@ Examples:
     .argument('<thread-id>', 'Thread ID (or 8-char prefix)')
     .option('--summary <text>', 'What was done to resolve it')
     .action(async (id: string, opts) => {
-      const session = await requireSession(agent.opts().session);
-      const thread = resolveThreadId(id, session.id);
+      await requireSession(agent.opts().session);
+      const thread = resolveThreadId(id);
       const author = opts.summary ? { name: 'Agent', type: 'agent' as const } : undefined;
       updateThreadStatus(thread.id, 'resolved', opts.summary ?? '', author);
       console.log(pc.green(`Resolved thread ${thread.id.slice(0, 8)}`));
@@ -353,8 +354,8 @@ Examples:
     .argument('<thread-id>', 'Thread ID (or 8-char prefix)')
     .option('--reason <text>', 'Why the thread is being dismissed')
     .action(async (id: string, opts) => {
-      const session = await requireSession(agent.opts().session);
-      const thread = resolveThreadId(id, session.id);
+      await requireSession(agent.opts().session);
+      const thread = resolveThreadId(id);
       const author = opts.reason ? { name: 'Agent', type: 'agent' as const } : undefined;
       updateThreadStatus(thread.id, 'dismissed', opts.reason ?? '', author);
       console.log(pc.green(`Dismissed thread ${thread.id.slice(0, 8)}`));
@@ -369,8 +370,8 @@ Examples:
     .option('--aside', 'A note for the reader that never goes to the forge')
     .option('--answers <comment-id>', 'The request this answers, so the page stops waiting on it')
     .action(async (id: string, opts: { body?: string; bodyFile?: string; aside?: boolean; answers?: string }) => {
-      const session = await requireSession(agent.opts().session);
-      const thread = resolveThreadId(id, session.id);
+      await requireSession(agent.opts().session);
+      const thread = resolveThreadId(id);
       const stillOpen = unansweredRequest(thread.comments);
       const body = bodyTextOrExit(opts, true);
       addReply(thread.id, body, { name: 'Agent', type: 'agent' }, opts.aside ? 'aside' : 'review');

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,6 @@ import {
 import { findOrCreateSession, resolveSessionId, agentSeenAt, markAgentSeen } from './session.js';
 import { resolveMayChangeCode, type SessionPurpose } from './live-permissions.js';
 import {
-  liveListenerCount,
   liveListenerTotal,
   liveWorkingCount,
   pendingLiveCount,
@@ -470,7 +469,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()),
@@ -531,7 +531,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/agent-cross-session.test.ts

@@ -0,0 +1,62 @@
+import { describe, it, expect, beforeAll, afterAll } from 'vitest';
+import { execFileSync, spawnSync } from 'node:child_process';
+import { mkdtempSync, writeFileSync, rmSync } from 'node:fs';
+import { dirname, join } from 'node:path';
+import { fileURLToPath } from 'node:url';
+import { tmpdir } from 'node:os';
+
+const ENTRY = join(dirname(fileURLToPath(import.meta.url)), '..', 'dist', 'index.js');
+
+let root: string;
+let repo: string;
+let origCwd: string;
+
+beforeAll(() => {
+  origCwd = process.cwd();
+  root = mkdtempSync(join(tmpdir(), 'diffity-cross-session-'));
+  repo = join(root, 'repo');
+  execFileSync('git', ['init', '-b', 'main', repo], { stdio: 'pipe' });
+  execFileSync('git', ['config', 'user.email', 't@t'], { cwd: repo, stdio: 'pipe' });
+  execFileSync('git', ['config', 'user.name', 'T'], { cwd: repo, stdio: 'pipe' });
+  writeFileSync(join(repo, 'a.ts'), 'const a = 1;\n');
+  execFileSync('git', ['add', '.'], { cwd: repo, stdio: 'pipe' });
+  execFileSync('git', ['commit', '-m', 'init'], { cwd: repo, stdio: 'pipe' });
+  process.env.DIFFITY_DATA_DIR = join(root, 'notes');
+  process.chdir(repo);
+});
+
+afterAll(() => {
+  process.chdir(origCwd);
+  delete process.env.DIFFITY_DATA_DIR;
+  rmSync(root, { recursive: true, force: true });
+});
+
+function cli(args: string[]) {
+  return spawnSync(process.execPath, [ENTRY, '--repo', repo, 'agent', ...args], {
+    cwd: repo, encoding: 'utf-8', env: { ...process.env, DIFFITY_DATA_DIR: join(root, 'notes') },
+  });
+}
+
+describe('a thread on another session of the same instance', () => {
+  it('takes the agent\'s reply and resolution, since the live queue can hand it one from anywhere', async () => {
+    const { findOrCreateSession } = await import('../src/session.js');
+    const { createThread, getThread } = await import('../src/threads.js');
+    // The thread lives on one session; the agent is parked on another, which is the current one.
+    const elsewhere = findOrCreateSession('main');
+    const parkedOn = findOrCreateSession('work');
+    expect(parkedOn.id).not.toBe(elsewhere.id);
+    const thread = createThread(elsewhere.id, 'a.ts', 'new', 1, 1, 'P2: asked over here', { name: 'Agent', type: 'agent' });
+
+    const replied = cli(['reply', thread.id, '--body', 'answered from where the agent sits']);
+    // Node 22 prints an ExperimentalWarning for node:sqlite on stderr; only a refusal matters here.
+    expect(replied.stderr).not.toContain('Error');
+    expect(replied.status).toBe(0);
+    expect(replied.stdout).toContain('Replied to thread');
+    expect(getThread(thread.id)!.comments.map(comment => comment.body)).toEqual(['P2: asked over here', 'answered from where the agent sits']);
+
+    const resolved = cli(['resolve', thread.id, '--summary', 'done']);
+    expect(resolved.status).toBe(0);
+    expect(getThread(thread.id)!.status).toBe('resolved');
+    expect(getThread(thread.id)!.sessionId).toBe(elsewhere.id);
+  });
+});

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(