From ce16513ffa8e2f38cb355bd0a97d556f579753fc Mon Sep 17 00:00:00 2001 From: "Fredrik Liljegren (Claude Code Claude Fable 5)" Date: Tue, 1 Sep 2026 14:10:41 +0200 Subject: [PATCH 1/4] feat: the live queue serves the whole instance MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 left both waiting forever. The queue, the listeners and the page's "listening" answer are now instance-wide: the request names its thread, so whichever agent is parked here takes it. Waiting and working stay per session — they describe the requests a page can see. WIP toward #51 — not yet reviewed or pushed. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_018PkYQzbsnMihHesafWvXKs --- packages/cli/src/live.ts | 59 ++++------- packages/cli/src/review-routes.ts | 4 +- packages/cli/src/server.ts | 7 +- .../cli/tests/idle-instance-guards.test.ts | 2 +- packages/cli/tests/live-requests.test.ts | 99 +++++++++---------- 5 files changed, 75 insertions(+), 96 deletions(-) diff --git a/packages/cli/src/live.ts b/packages/cli/src/live.ts index 23b51963..3c7945cb 100644 --- a/packages/cli/src/live.ts +++ b/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 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 { @@ -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,11 +173,7 @@ 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); @@ -198,10 +181,10 @@ export function waitForLiveRequest( 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); }); } diff --git a/packages/cli/src/review-routes.ts b/packages/cli/src/review-routes.ts index 870b7776..646fff1e 100644 --- a/packages/cli/src/review-routes.ts +++ b/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); }); diff --git a/packages/cli/src/server.ts b/packages/cli/src/server.ts index 724d6880..1d2e8510 100644 --- a/packages/cli/src/server.ts +++ b/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 { 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 { 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. diff --git a/packages/cli/tests/idle-instance-guards.test.ts b/packages/cli/tests/idle-instance-guards.test.ts index 684c9db1..d212b540 100644 --- a/packages/cli/tests/idle-instance-guards.test.ts +++ b/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( diff --git a/packages/cli/tests/live-requests.test.ts b/packages/cli/tests/live-requests.test.ts index 9483d0b4..b48955cf 100644 --- a/packages/cli/tests/live-requests.test.ts +++ b/packages/cli/tests/live-requests.test.ts @@ -45,13 +45,12 @@ async function finding(body = 'P2: a finding') { async function drainRequests() { const { claimNextLiveRequest, answerLiveRequest } = await import('../src/live.js'); - const s = await session(); // Answered, not merely claimed: sessions on one branch share their open threads, and a claim // without an answer leaves the session looking like an agent is still working on it. - let taken = claimNextLiveRequest(s.id); + let taken = claimNextLiveRequest(); while (taken) { answerLiveRequest(taken.commentId); - taken = claimNextLiveRequest(s.id); + taken = claimNextLiveRequest(); } } @@ -90,7 +89,7 @@ describe('asking the agent for something', () => { const reply = addReply(thread.id, 'what do you mean by the marker?', you, 'aside'); requestLive(reply.id); - const claimed = claimNextLiveRequest(s.id); + const claimed = claimNextLiveRequest(); expect(claimed?.commentId).toBe(reply.id); expect(claimed?.threadId).toBe(thread.id); @@ -107,8 +106,8 @@ describe('asking the agent for something', () => { const reply = addReply(thread.id, 'only once please', you, 'aside'); requestLive(reply.id); - const first = claimNextLiveRequest(s.id); - const second = claimNextLiveRequest(s.id); + const first = claimNextLiveRequest(); + const second = claimNextLiveRequest(); expect(first?.commentId).toBe(reply.id); expect(second).toBeNull(); @@ -124,8 +123,8 @@ describe('asking the agent for something', () => { requestLive(first.id); requestLive(second.id); - expect(claimNextLiveRequest(s.id)?.body).toBe('asked first'); - expect(claimNextLiveRequest(s.id)?.body).toBe('asked second'); + expect(claimNextLiveRequest()?.body).toBe('asked first'); + expect(claimNextLiveRequest()?.body).toBe('asked second'); }); it('says how many are still waiting', async () => { @@ -138,7 +137,7 @@ describe('asking the agent for something', () => { } expect(pendingLiveCount(s.id)).toBe(3); - claimNextLiveRequest(s.id); + claimNextLiveRequest(); expect(pendingLiveCount(s.id)).toBe(2); }); @@ -149,21 +148,27 @@ describe('asking the agent for something', () => { const thread = await finding(); const reply = addReply(thread.id, 'answer me', you, 'aside'); requestLive(reply.id); - claimNextLiveRequest(s.id); + claimNextLiveRequest(); answerLiveRequest(reply.id); expect(getThread(thread.id)?.comments[1].liveAnsweredAt).toBeTruthy(); }); - it('belongs to its own session', async () => { + it('is served across sessions: the instance is one queue', async () => { const { addReply } = await import('../src/threads.js'); - const { requestLive, claimNextLiveRequest } = await import('../src/live.js'); + const { requestLive, waitForLiveRequest, notifyLiveListeners } = await import('../src/live.js'); + await drainRequests(); + // The reader asks on one session while the only agent sits parked on another — the + // arrangement that once left both waiting forever. const thread = await finding(); - const reply = addReply(thread.id, 'in the work session', you, 'aside'); + const reply = addReply(thread.id, 'asked from the page', you, 'aside'); + + const parked = waitForLiveRequest(60_000); requestLive(reply.id); + notifyLiveListeners(); - expect(claimNextLiveRequest('some-other-session')).toBeNull(); + expect((await parked)?.commentId).toBe(reply.id); }); }); @@ -176,7 +181,7 @@ describe('a listener that died holding a request', () => { const thread = await finding(); const reply = addReply(thread.id, 'nobody came back', you, 'aside'); requestLive(reply.id); - claimNextLiveRequest(s.id); + claimNextLiveRequest(); // What it looks like when the agent was claimed by a process that then went away. getDb() @@ -195,7 +200,7 @@ describe('a listener that died holding a request', () => { const s = await session(); const thread = await finding(); requestLive(addReply(thread.id, 'just claimed', you, 'aside').id); - claimNextLiveRequest(s.id); + claimNextLiveRequest(); expect(reclaimStaleLiveRequests(10)).toBe(0); }); @@ -212,7 +217,7 @@ describe('asking about a line nobody has commented on', () => { ); requestLive(thread.comments[0].id); - const claimed = claimNextLiveRequest(s.id); + const claimed = claimNextLiveRequest(); expect(claimed?.body).toBe('what does this do?'); expect(claimed?.findingBody).toBeNull(); @@ -227,7 +232,7 @@ describe('asking about a line nobody has commented on', () => { const asked = addReply(thread.id, 'why?', you, 'aside'); requestLive(asked.id); - expect(claimNextLiveRequest(s.id)?.findingBody).toBe('P2: the finding'); + expect(claimNextLiveRequest()?.findingBody).toBe('P2: the finding'); }); }); @@ -242,7 +247,7 @@ describe('closing a request', () => { const thread = await finding(); const asked = addReply(thread.id, 'answer me by prefix', you, 'aside'); requestLive(asked.id); - claimNextLiveRequest(s.id); + claimNextLiveRequest(); expect(answerLiveRequest(asked.id.slice(0, 8))).toBe(true); expect(getThread(thread.id)?.comments[1].liveAnsweredAt).toBeTruthy(); @@ -255,43 +260,35 @@ describe('closing a request', () => { }); }); -describe('two sessions in one server', () => { +describe('one queue for the instance', () => { // One diffity serves a whole checkout: the file browser is its own session and each ref gets - // one, so a listener on one must not be disturbed by — or spoken for by — another. - it('does not report a listener on one session as listening on another', async () => { - const { findOrCreateSession } = await import('../src/session.js'); - const { waitForLiveRequest, liveListenerCount } = await import('../src/live.js'); - const other = findOrCreateSession('__tree__'); - const s = await session(); + // one. A reader asking on one session while the only agent sat parked on another once left + // both waiting forever, so a parked listener now counts for — and serves — all of them. + it('counts a parked listener for the whole instance, and only while it is parked', async () => { + const { waitForLiveRequest, liveListenerTotal } = await import('../src/live.js'); + await drainRequests(); - const parked = waitForLiveRequest(s.id, 400); + const parked = waitForLiveRequest(400); await new Promise(resolve => setTimeout(resolve, 50)); - expect(liveListenerCount(s.id)).toBe(1); - expect(liveListenerCount(other.id)).toBe(0); - + expect(liveListenerTotal()).toBe(1); await parked; + expect(liveListenerTotal()).toBe(0); }); - it('does not end a wait because another session was asked something', async () => { - const { findOrCreateSession } = await import('../src/session.js'); - const { createThread } = await import('../src/threads.js'); - const { requestLive, notifyLiveListeners, waitForLiveRequest } = await import('../src/live.js'); + it('does not end a wait when a wake finds nothing claimable', async () => { + const { notifyLiveListeners, waitForLiveRequest } = await import('../src/live.js'); await drainRequests(); - const other = findOrCreateSession('__tree__'); - const s = await session(); let settled = false; - const parked = waitForLiveRequest(s.id, 600).then(request => { + const parked = waitForLiveRequest(600).then(request => { settled = true; return request; }); await new Promise(resolve => setTimeout(resolve, 50)); - // Something asked, but on the other session. - const elsewhere = createThread(other.id, 'a.ts', 'new', 1, 1, 'asked over here', you, undefined, 'aside'); - requestLive(elsewhere.comments[0].id); - notifyLiveListeners(other.id); + // Woken with an empty queue: only a claim may end the wait early. + notifyLiveListeners(); await new Promise(resolve => setTimeout(resolve, 100)); expect(settled).toBe(false); @@ -320,32 +317,32 @@ describe('a listener whose connection goes away', () => { // Presence is a parked connection. Nothing noticed the connection closing, so `listening` kept // saying yes for as long as the wait had left — up to four minutes of promising an answer. it('stops being counted at once, not when its wait runs out', async () => { - const { waitForLiveRequest, liveListenerCount } = await import('../src/live.js'); + const { waitForLiveRequest, liveListenerTotal } = await import('../src/live.js'); // Sessions on one branch share their open threads, so an earlier case's request would be // claimed here and this would never park at all. await drainRequests(); const s = await session(); const controller = new AbortController(); - const parked = waitForLiveRequest(s.id, 60_000, controller.signal); + const parked = waitForLiveRequest(60_000, controller.signal); await new Promise(resolve => setTimeout(resolve, 20)); - expect(liveListenerCount(s.id)).toBe(1); + expect(liveListenerTotal()).toBe(1); controller.abort(); expect(await parked).toBeNull(); - expect(liveListenerCount(s.id)).toBe(0); + expect(liveListenerTotal()).toBe(0); }); it('is already gone if the connection closed before it parked', async () => { - const { waitForLiveRequest, liveListenerCount } = await import('../src/live.js'); + const { waitForLiveRequest, liveListenerTotal } = await import('../src/live.js'); await drainRequests(); const s = await session(); const controller = new AbortController(); controller.abort(); - expect(await waitForLiveRequest(s.id, 60_000, controller.signal)).toBeNull(); - expect(liveListenerCount(s.id)).toBe(0); + expect(await waitForLiveRequest(60_000, controller.signal)).toBeNull(); + expect(liveListenerTotal()).toBe(0); }); it('claims nothing when it arrives already hung up, so the request stays answerable', async () => { @@ -359,7 +356,7 @@ describe('a listener whose connection goes away', () => { const controller = new AbortController(); controller.abort(); - expect(await waitForLiveRequest(s.id, 60_000, controller.signal)).toBeNull(); + expect(await waitForLiveRequest(60_000, controller.signal)).toBeNull(); // A live listener must still find it; claimed-by-nobody costs the reader ten minutes. expect(pendingLiveCount(s.id)).toBe(1); }); @@ -380,7 +377,7 @@ describe('while an agent is busy with a request', () => { // A delta, not an absolute: other cases in this file claim without answering, and every // session on this branch shares its threads. const before = liveWorkingCount(s.id); - claimNextLiveRequest(s.id); + claimNextLiveRequest(); expect(liveWorkingCount(s.id)).toBe(before + 1); }); @@ -393,7 +390,7 @@ describe('while an agent is busy with a request', () => { const thread = await finding(); const asked = addReply(thread.id, 'answer me', you, 'aside'); requestLive(asked.id); - claimNextLiveRequest(s.id); + claimNextLiveRequest(); const busy = liveWorkingCount(s.id); answerLiveRequest(asked.id); From 5e0136d26fa659519a6edec5b3e8582d621411ac Mon Sep 17 00:00:00 2001 From: "Fredrik Liljegren (Claude Code Claude Fable 5.1)" Date: Thu, 3 Sep 2026 14:50:17 +0200 Subject: [PATCH 2/4] chore: bump to 0.10.14 Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01Bwp5QefjsjMFeq6CK6cT6w --- package-lock.json | 12 ++++++------ packages/api/package.json | 2 +- packages/cli/package.json | 2 +- packages/git/package.json | 2 +- packages/github/package.json | 2 +- packages/parser/package.json | 2 +- packages/ui/package.json | 2 +- 7 files changed, 12 insertions(+), 12 deletions(-) diff --git a/package-lock.json b/package-lock.json index 8bf30df8..43a9ccf7 100644 --- a/package-lock.json +++ b/package-lock.json @@ -8457,7 +8457,7 @@ }, "packages/api": { "name": "@diffity/api", - "version": "0.10.13", + "version": "0.10.14", "dependencies": { "@diffity/parser": "*" }, @@ -8468,7 +8468,7 @@ }, "packages/cli": { "name": "@naturalcycles/diffity", - "version": "0.10.13", + "version": "0.10.14", "license": "MIT", "dependencies": { "commander": "^14.0.3", @@ -8492,7 +8492,7 @@ }, "packages/git": { "name": "@diffity/git", - "version": "0.10.13", + "version": "0.10.14", "devDependencies": { "@types/node": "^25.5.0", "typescript": "^5.9.3", @@ -8501,7 +8501,7 @@ }, "packages/github": { "name": "@diffity/github", - "version": "0.10.13", + "version": "0.10.14", "dependencies": { "@diffity/api": "*", "@diffity/parser": "*" @@ -8514,7 +8514,7 @@ }, "packages/parser": { "name": "@diffity/parser", - "version": "0.10.13", + "version": "0.10.14", "devDependencies": { "typescript": "^5.9.3", "vitest": "^4.1.0" @@ -8522,7 +8522,7 @@ }, "packages/ui": { "name": "@diffity/ui", - "version": "0.10.13", + "version": "0.10.14", "dependencies": { "@diffity/api": "*", "@diffity/parser": "*", diff --git a/packages/api/package.json b/packages/api/package.json index 6e6b63b1..589ef7a9 100644 --- a/packages/api/package.json +++ b/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", diff --git a/packages/cli/package.json b/packages/cli/package.json index c8507e86..e2f203e6 100644 --- a/packages/cli/package.json +++ b/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": { diff --git a/packages/git/package.json b/packages/git/package.json index 7500403d..049bc2ff 100644 --- a/packages/git/package.json +++ b/packages/git/package.json @@ -1,6 +1,6 @@ { "name": "@diffity/git", - "version": "0.10.13", + "version": "0.10.14", "private": true, "type": "module", "main": "./dist/index.js", diff --git a/packages/github/package.json b/packages/github/package.json index 82234e4c..55fd0760 100644 --- a/packages/github/package.json +++ b/packages/github/package.json @@ -1,6 +1,6 @@ { "name": "@diffity/github", - "version": "0.10.13", + "version": "0.10.14", "private": true, "type": "module", "main": "./dist/index.js", diff --git a/packages/parser/package.json b/packages/parser/package.json index 4162a789..486d5216 100644 --- a/packages/parser/package.json +++ b/packages/parser/package.json @@ -1,6 +1,6 @@ { "name": "@diffity/parser", - "version": "0.10.13", + "version": "0.10.14", "private": true, "type": "module", "main": "./dist/index.js", diff --git a/packages/ui/package.json b/packages/ui/package.json index f22aacdf..c2c08866 100644 --- a/packages/ui/package.json +++ b/packages/ui/package.json @@ -1,6 +1,6 @@ { "name": "@diffity/ui", - "version": "0.10.13", + "version": "0.10.14", "type": "module", "private": true, "scripts": { From 15911701202226d0575d046b6053438c8a690847 Mon Sep 17 00:00:00 2001 From: "Fredrik Liljegren (Claude Code Claude Fable 5.1)" Date: Thu, 3 Sep 2026 14:58:57 +0200 Subject: [PATCH 3/4] fix: a thread on any session of the instance takes the agent's reply and resolution The queue became instance-wide but the agent's thread commands still refused a thread outside the agent's own session, so a request claimed from another session could be received and not answered. Threads are addressable across the instance now; tours stay with their session. The cross-session claim test puts its thread on a second session, as its comment always said, and a CLI test replies to and resolves a thread from another session. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01Bwp5QefjsjMFeq6CK6cT6w --- packages/cli/src/agent.ts | 23 +++---- packages/cli/src/server.ts | 1 - .../cli/tests/agent-cross-session.test.ts | 61 +++++++++++++++++++ packages/cli/tests/live-requests.test.ts | 7 ++- 4 files changed, 78 insertions(+), 14 deletions(-) create mode 100644 packages/cli/tests/agent-cross-session.test.ts diff --git a/packages/cli/src/agent.ts b/packages/cli/src/agent.ts index 28d371a9..7b4f3a2c 100644 --- a/packages/cli/src/agent.ts +++ b/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 (or 8-char prefix)') .option('--summary ', '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 (or 8-char prefix)') .option('--reason ', '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 ', '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'); diff --git a/packages/cli/src/server.ts b/packages/cli/src/server.ts index 7abd9ca0..3e0404fe 100644 --- a/packages/cli/src/server.ts +++ b/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 { - liveListenerTotal, liveWorkingCount, pendingLiveCount, diff --git a/packages/cli/tests/agent-cross-session.test.ts b/packages/cli/tests/agent-cross-session.test.ts new file mode 100644 index 00000000..25cc2489 --- /dev/null +++ b/packages/cli/tests/agent-cross-session.test.ts @@ -0,0 +1,61 @@ +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']); + expect(replied.stderr).toBe(''); + 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); + }); +}); diff --git a/packages/cli/tests/live-requests.test.ts b/packages/cli/tests/live-requests.test.ts index b48955cf..f0f0f128 100644 --- a/packages/cli/tests/live-requests.test.ts +++ b/packages/cli/tests/live-requests.test.ts @@ -156,12 +156,15 @@ describe('asking the agent for something', () => { }); it('is served across sessions: the instance is one queue', async () => { - const { addReply } = await import('../src/threads.js'); + const { addReply, createThread } = await import('../src/threads.js'); + const { findOrCreateSession } = await import('../src/session.js'); const { requestLive, waitForLiveRequest, notifyLiveListeners } = await import('../src/live.js'); await drainRequests(); // The reader asks on one session while the only agent sits parked on another — the // arrangement that once left both waiting forever. - const thread = await finding(); + const elsewhere = findOrCreateSession('main'); + expect(elsewhere.id).not.toBe((await session()).id); + const thread = createThread(elsewhere.id, 'a.ts', 'new', 1, 1, 'P2: asked over here', agent); const reply = addReply(thread.id, 'asked from the page', you, 'aside'); const parked = waitForLiveRequest(60_000); From c9c2351693c34370630e43a5cb95eb415d18b2c0 Mon Sep 17 00:00:00 2001 From: "Fredrik Liljegren (Claude Code Claude Fable 5.1)" Date: Thu, 3 Sep 2026 15:01:35 +0200 Subject: [PATCH 4/4] test: the cross-session reply test tolerates Node 22's sqlite warning on stderr Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01Bwp5QefjsjMFeq6CK6cT6w --- packages/cli/tests/agent-cross-session.test.ts | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/packages/cli/tests/agent-cross-session.test.ts b/packages/cli/tests/agent-cross-session.test.ts index 25cc2489..56fa3892 100644 --- a/packages/cli/tests/agent-cross-session.test.ts +++ b/packages/cli/tests/agent-cross-session.test.ts @@ -48,7 +48,8 @@ describe('a thread on another session of the same instance', () => { 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']); - expect(replied.stderr).toBe(''); + // 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']);