From 58dfc789a2c347be3f66c0b40b80ddb37990b1d5 Mon Sep 17 00:00:00 2001 From: Bohdan Triapitsyn Date: Sat, 5 Sep 2026 11:21:39 +0300 Subject: [PATCH] fix(queue): clear the queue card after the last queued message is sent When the server delivered the last item of a session's queue it deleted the session entry and broadcast the update with an empty directory. The UI keys its projection by directory, could not build the key, and dropped the event, so the delivered message stayed in the "Queued messages" card until the next full hydration. The server now remembers a session's directory beyond the emptying of its queue so that broadcast names it. The UI additionally treats an empty session without a directory as "this session's queue is done" for every projection keyed under that session id, which covers clients talking to an older server. Claude-Session: https://claude.ai/code/session_01VqV56Hez25hTxXH4ipJfzH --- packages/ui/src/stores/DOCUMENTATION.md | 2 +- .../stores/messageQueueStore.server.test.ts | 17 +++++++++++ packages/ui/src/stores/messageQueueStore.ts | 29 ++++++++++++++++++- .../server/lib/message-queue/DOCUMENTATION.md | 6 +++- .../web/server/lib/message-queue/runtime.js | 10 ++++++- .../server/lib/message-queue/runtime.test.js | 15 ++++++++++ 6 files changed, 75 insertions(+), 4 deletions(-) diff --git a/packages/ui/src/stores/DOCUMENTATION.md b/packages/ui/src/stores/DOCUMENTATION.md index f6bdb396..9f1e1ce3 100644 --- a/packages/ui/src/stores/DOCUMENTATION.md +++ b/packages/ui/src/stores/DOCUMENTATION.md @@ -64,7 +64,7 @@ These stores coordinate persistent project/session metadata across multiple view `useProjectContextStore.ts` caches server-owned project notes, todos, and plan links, keyed by the path-derived project id. It replaced a pair of `window` CustomEvents that made every mounted notes panel re-read the whole project config. Writes are optimistic and roll back on failure; they are serialized per project, because the server's own store does a read-modify-write and two concurrent saves would otherwise race it. A load that resolves while a write is in flight keeps the local value for that field group only, so a slow snapshot cannot undo newer typing while still delivering the plan list it fetched. A failed load sets `error` and preserves the cached snapshot — an unreachable server must never render as "this project has no notes". Note and plan creation are deliberately not optimistic, since ids and timestamps are assigned by the server. Notes, todos, and plans are written through separate routes and tracked by separate in-flight flags, so a todo toggle cannot clobber a note edit in the same window. Pinned notes and plans are assembled into a synthetic context part by `lib/projectContextPinning.ts` at send time; that module tracks per-session what it already sent so an unchanged pinned set is not re-sent every turn. -`messageQueueStore.ts` has two owners, decided by `isServerOwnedMessageQueue()`. On web, desktop, and mobile the OpenChamber server owns the queue (`packages/web/server/lib/message-queue/`): it delivers queued messages when the session goes idle whether or not any UI is open, and the store is a projection of it — `hydrate()` loads the server snapshot for the active runtime, `openchamber:message-queue.updated` broadcasts keep it current, and every mutation is optimistic locally then settled on the server's copy of that session (a failed round-trip re-reads the server instead of guessing). A per-key server revision rejects stale snapshots. Projection items carry attachment metadata only and no captured context; `popToInput()`/`takeForSend()` remove the message on the server and get the full payload back, which is why both are async. +`messageQueueStore.ts` has two owners, decided by `isServerOwnedMessageQueue()`. On web, desktop, and mobile the OpenChamber server owns the queue (`packages/web/server/lib/message-queue/`): it delivers queued messages when the session goes idle whether or not any UI is open, and the store is a projection of it — `hydrate()` loads the server snapshot for the active runtime, `openchamber:message-queue.updated` broadcasts keep it current, and every mutation is optimistic locally then settled on the server's copy of that session (a failed round-trip re-reads the server instead of guessing). A per-key server revision rejects stale snapshots. An empty session that arrives without a directory (servers before 1.22.2 dropped it once the queue emptied) clears every projection of that session id in the runtime, because a session id is unique across directories. Projection items carry attachment metadata only and no captured context; `popToInput()`/`takeForSend()` remove the message on the server and get the full payload back, which is why both are async. A queued message is captured whole, so whoever delivers it sends exactly what the composer would have: `text` (the content with its agent mention stripped and `@file` mentions already resolved into `attachments`), `agentMention`, and `context` — every chip the composer had attached (inline comments, terminal selections, browser annotations, PR comments/checks, quotes, linked issue/PR/Linear references, pending synthetic parts) plus the skill instruction derived from the text. `QueuedContextPart` distinguishes attached items (restored to the chips when the message is edited) from derived instructions (re-derived on send, never restored) and from synthetic parts other surfaces handed the composer (restored as pending). Context is captured by `buildComposerContext` and delivered by `queuedContextToParts` (`components/chat/composer/submit/buildOutgoingMessage.ts`), the same functions the composer uses for its own send. Nothing is re-resolved at delivery: the server has no agent list, no confirmed mentions, and no draft store. Messages a previous build left in this browser are uploaded once on the first hydration of a runtime and then dropped from persistence for that runtime (`partialize` skips server-owned runtime keys). VS Code has no server and keeps the local queue with the foreground auto-send hook (`useQueuedMessageAutoSend`, enabled only there); `useMessageQueueHoldSync` tells the server to hold a session's queue while a UI-driven auto-review run is going. diff --git a/packages/ui/src/stores/messageQueueStore.server.test.ts b/packages/ui/src/stores/messageQueueStore.server.test.ts index 892de098..63e3dda3 100644 --- a/packages/ui/src/stores/messageQueueStore.server.test.ts +++ b/packages/ui/src/stores/messageQueueStore.server.test.ts @@ -233,6 +233,23 @@ describe("server-owned message queue", () => { expect(useMessageQueueStore.getState().queuedMessages[key]?.map((m) => m.content)).toEqual(["newer"]) }) + test("an empty session without a directory still clears the projection it was keyed under", () => { + applyMessageQueueUpdatedEvent(updated(4, session([serverItem("q1", "queued")], "q1")), "runtime-a") + expect(useMessageQueueStore.getState().queuedMessages[key]).toHaveLength(1) + expect(useMessageQueueStore.getState().sendingIds[key]).toEqual(["q1"]) + + // A server that forgot the directory once the queue emptied. + applyMessageQueueUpdatedEvent(updated(5, { sessionId: "session-1", directory: "", items: [], sendingId: null }), "runtime-a") + expect(useMessageQueueStore.getState().queuedMessages[key]).toBe(undefined) + expect(useMessageQueueStore.getState().sendingIds[key]).toBe(undefined) + + // Still never backwards, and never another runtime's projection. + applyMessageQueueUpdatedEvent(updated(6, session([serverItem("q2", "later")])), "runtime-a") + applyMessageQueueUpdatedEvent(updated(3, { sessionId: "session-1", directory: "", items: [], sendingId: null }), "runtime-a") + applyMessageQueueUpdatedEvent(updated(9, { sessionId: "session-1", directory: "", items: [], sendingId: null }), "runtime-b") + expect(useMessageQueueStore.getState().queuedMessages[key]?.map((m) => m.content)).toEqual(["later"]) + }) + test("removeFromQueue and clearQueue update locally and tell the server", async () => { useMessageQueueStore.setState({ queuedMessages: { [key]: [{ id: "q1", content: "a", text: "a", createdAt: 1 }, { id: "q2", content: "b", text: "b", createdAt: 2 }] } }) respond = () => json({ revision: 10, session: session([serverItem("q2", "b")]) }) diff --git a/packages/ui/src/stores/messageQueueStore.ts b/packages/ui/src/stores/messageQueueStore.ts index 3341779e..d6ec620b 100644 --- a/packages/ui/src/stores/messageQueueStore.ts +++ b/packages/ui/src/stores/messageQueueStore.ts @@ -429,6 +429,26 @@ const removeMessageLocally = ( return { queuedMessages: { ...state.queuedMessages, [key]: newQueue } }; }; +/** Every projection of one session in this runtime, whatever directory it was keyed under. */ +const clearSessionProjection = ( + state: Pick, + runtimeKey: string, + sessionId: string, + revision: number, +): Pick => { + let queuedMessages = state.queuedMessages; + let sendingIds = state.sendingIds; + for (const key of new Set([...Object.keys(queuedMessages), ...Object.keys(sendingIds)])) { + const parsed = parseMessageQueueKey(key); + if (parsed?.runtimeKey !== runtimeKey || parsed.sessionId !== sessionId) continue; + if ((appliedRevisions.get(key) ?? -1) > revision) continue; + appliedRevisions.set(key, revision); + queuedMessages = withoutKey(queuedMessages, key); + sendingIds = withoutKey(sendingIds, key); + } + return { queuedMessages, sendingIds }; +}; + export const useMessageQueueStore = create()( devtools( persist( @@ -436,7 +456,14 @@ export const useMessageQueueStore = create()( const applyServerSession = (session: ServerQueueSession, revision: number, expectedRuntimeKey: string) => { if (expectedRuntimeKey !== getRuntimeKey()) return; const target = createMessageQueueTarget(session.sessionId, session.directory, expectedRuntimeKey); - if (!target) return; + if (!target) { + // Servers before 1.22.2 drop a session's directory once its + // queue is empty. A session id is unique across directories, + // so an empty session still says which projection is done. + if (session.items.length > 0) return; + set((state) => clearSessionProjection(state, expectedRuntimeKey, session.sessionId, revision)); + return; + } const key = getMessageQueueKey(target); if ((appliedRevisions.get(key) ?? -1) > revision) return; appliedRevisions.set(key, revision); diff --git a/packages/web/server/lib/message-queue/DOCUMENTATION.md b/packages/web/server/lib/message-queue/DOCUMENTATION.md index 3ce85087..470bef7c 100644 --- a/packages/web/server/lib/message-queue/DOCUMENTATION.md +++ b/packages/web/server/lib/message-queue/DOCUMENTATION.md @@ -134,7 +134,11 @@ allowlists. Every mutation broadcasts `openchamber:message-queue.updated` with `{ revision, session }` to all connected clients (SSE and WS), so several -devices on one server see one queue. +devices on one server see one queue. The session in that payload always names +its `directory`, including the broadcast that removes the last item: the UI +keys its projection by directory, and a broadcast without one left the +delivered message on screen (a session's directory is remembered until the +session is deleted or evicted). Limits: 20 items per session, 50 sessions (oldest evicted, never one with an item in flight), 200k characters of content; attachment payloads are bounded diff --git a/packages/web/server/lib/message-queue/runtime.js b/packages/web/server/lib/message-queue/runtime.js index 370dbe9c..e040b079 100644 --- a/packages/web/server/lib/message-queue/runtime.js +++ b/packages/web/server/lib/message-queue/runtime.js @@ -220,6 +220,10 @@ export function createMessageQueueRuntime({ const failures = new Map(); // sessionId → { itemId, failures, nextAttemptAt } const abortedAt = new Map(); // sessionId → timestamp const holds = new Map(); // sessionId → expiresAt + // sessionId → directory, kept after the queue empties: the UI keys its + // projection by directory, so the broadcast that removes the last item must + // still name it or the client cannot tell which queue just finished. + const directories = new Map(); // --- persistence --------------------------------------------------------- @@ -302,7 +306,7 @@ export function createMessageQueueRuntime({ const queue = queues.get(sessionId); return { sessionId, - directory: queue?.directory ?? '', + directory: queue?.directory ?? directories.get(sessionId) ?? '', items: (queue?.items ?? []).map(toPublicItem), sendingId: sending.get(sessionId) ?? null, }; @@ -329,6 +333,7 @@ export function createMessageQueueRuntime({ }; const setQueueItems = (sessionId, directory, items) => { + directories.set(sessionId, directory); if (items.length === 0) { queues.delete(sessionId); return; @@ -564,6 +569,7 @@ export function createMessageQueueRuntime({ const existing = queues.get(sessionId); const items = [...(existing?.items ?? []), item].slice(-MAX_ITEMS_PER_SESSION); queues.set(sessionId, { directory, items }); + directories.set(sessionId, directory); if (queues.size > MAX_SESSIONS) { const oldest = Array.from(queues.entries()) .filter(([id]) => id !== sessionId && !sending.has(id)) @@ -573,6 +579,7 @@ export function createMessageQueueRuntime({ queues.delete(staleId); clearTimer(staleId); broadcast(staleId); + directories.delete(staleId); } } const result = commit(sessionId); @@ -675,6 +682,7 @@ export function createMessageQueueRuntime({ clearTimer(deletedSessionId); failures.delete(deletedSessionId); commit(deletedSessionId); + directories.delete(deletedSessionId); return; } diff --git a/packages/web/server/lib/message-queue/runtime.test.js b/packages/web/server/lib/message-queue/runtime.test.js index fb64405a..509fb07e 100644 --- a/packages/web/server/lib/message-queue/runtime.test.js +++ b/packages/web/server/lib/message-queue/runtime.test.js @@ -315,6 +315,21 @@ describe('message queue runtime', () => { expect(runtime.snapshot().sessions).toEqual([]); }); + it('names the directory in the broadcast that empties a queue', async () => { + // The UI keys its projection by directory; without it the client cannot + // tell which queue just delivered its last message and keeps showing it. + const { runtime, emit, broadcasts, openCode } = createRuntime(); + runtime.start(); + await runtime.enqueue(SESSION, DIRECTORY, item()); + openCode.state.statuses = {}; + emit({ type: 'session.status', properties: { sessionID: SESSION, status: { type: 'idle' } } }); + await settle(); + + expect(openCode.state.sent).toHaveLength(1); + expect(runtime.snapshot().sessions).toEqual([]); + expect(broadcasts.at(-1).properties.session).toEqual({ sessionId: SESSION, directory: DIRECTORY, items: [], sendingId: null }); + }); + it('reorders only with a complete permutation', async () => { const { runtime } = createRuntime(); runtime.start();