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
This commit is contained in:
@@ -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
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
|
||||
@@ -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();
|
||||
|
||||
Reference in New Issue
Block a user