diff --git a/packages/ui/src/components/chat/ChatInput.tsx b/packages/ui/src/components/chat/ChatInput.tsx index feda2c44..ba493f55 100644 --- a/packages/ui/src/components/chat/ChatInput.tsx +++ b/packages/ui/src/components/chat/ChatInput.tsx @@ -159,6 +159,7 @@ const MAX_MOBILE_COMPOSER_LINES = 16; */ const MOBILE_COMPOSER_BOUND_GAP_PX = 4; const EMPTY_QUEUE: QueuedMessage[] = []; +const EMPTY_SENDING_IDS: string[] = []; const COMPACT_CHAT_PLACEHOLDER_MAX_WIDTH = 560; const renameFileForAttachmentCitation = (file: File, filename: string): File => { if (file.name === filename) { @@ -945,9 +946,16 @@ const ChatInputComponent: React.FC = ({ onOpenSettings, scrollTo hasContent: options.presetText.trim().length > 0 || attachedFiles.length > 0 || hasDrafts, } : getCurrentInputSnapshot(); - const queuedMessagesToSend = queuedMessageId + // A queued item stays in the queue until its own send resolves, so the + // auto-send hook may already be delivering one of these. Merging it here + // would send the same message twice (the window is seconds over a relay). + const sendingIds = messageQueueTarget + ? useMessageQueueStore.getState().sendingIds[getMessageQueueKey(messageQueueTarget)] ?? EMPTY_SENDING_IDS + : EMPTY_SENDING_IDS; + const queuedMessagesToSend = (queuedMessageId ? queuedMessages.filter((message) => message.id === queuedMessageId) - : queuedMessages; + : queuedMessages + ).filter((message) => !sendingIds.includes(message.id)); if (queuedOnly && autoReviewRunning) { return; diff --git a/packages/ui/src/hooks/useQueuedMessageAutoSend.ts b/packages/ui/src/hooks/useQueuedMessageAutoSend.ts index bc327ebe..e7b39c97 100644 --- a/packages/ui/src/hooks/useQueuedMessageAutoSend.ts +++ b/packages/ui/src/hooks/useQueuedMessageAutoSend.ts @@ -221,7 +221,9 @@ export function useQueuedMessageAutoSend(enabledOrOptions?: boolean | { enabled? return; } - const payload = buildQueuedAutoSendPayload(queueSnapshot); + // Read the queue back at dispatch time and skip anything already being + // delivered, rather than trusting the render-time snapshot. + const payload = buildQueuedAutoSendPayload(useMessageQueueStore.getState().getSendableQueue(target)); if (!payload) { return; } @@ -248,6 +250,10 @@ export function useQueuedMessageAutoSend(enabledOrOptions?: boolean | { enabled? } inFlightSessionsRef.current.add(targetKey); + // The ref only guards this hook. Publish the dispatch to the store so the + // composer cannot merge the same item into a parallel send while this one + // is still awaiting the server. + useMessageQueueStore.getState().markSending(target, payload.queuedMessageId); try { await sendQueuedAutoSendPayload(sessionId, target.directory, payload, { @@ -271,6 +277,7 @@ export function useQueuedMessageAutoSend(enabledOrOptions?: boolean | { enabled? retryScheduler.schedule(nextAttemptAt); } finally { inFlightSessionsRef.current.delete(targetKey); + useMessageQueueStore.getState().clearSending(target, payload.queuedMessageId); } }; diff --git a/packages/ui/src/stores/DOCUMENTATION.md b/packages/ui/src/stores/DOCUMENTATION.md index bca5d525..65ad473a 100644 --- a/packages/ui/src/stores/DOCUMENTATION.md +++ b/packages/ui/src/stores/DOCUMENTATION.md @@ -47,9 +47,12 @@ Examples: - `useProjectsStore.ts` - `useGlobalSessionsStore.ts` - `useSessionFoldersStore.ts` +- `messageQueueStore.ts` These stores coordinate persistent project/session metadata across multiple views. +`messageQueueStore.ts` keeps a queued message until its own send resolves, so between dispatch and resolution the entry is still visible to every reader. Dispatchers must therefore mark the send (`markSending`/`clearSending`) and read `getSendableQueue()` — or filter `sendingIds` themselves — instead of dispatching straight from `queuedMessages`; otherwise a composer submit merges a message the auto-send hook is already delivering and it is sent twice (the window is seconds over a relay). `clearQueue()` retains in-flight entries for the same reason. `sendingIds` is deliberately not persisted: a restart has no in-flight sends, and a stale flag would strand a queued message. + `useGlobalSessionsStore.ts` owns cold/global active and archived session coverage, including `sessionsByDirectory`. It is complementary to directory child stores: it is not the source of live busy/retry status or session messages. User-visible session ordering is also not owned by the global cache array order. `sync/session-ordering.ts` combines lifecycle rank with timestamp fallbacks, and session surfaces must use that shared comparator instead of independently sorting global sessions by `time.updated`. diff --git a/packages/ui/src/stores/messageQueueStore.test.ts b/packages/ui/src/stores/messageQueueStore.test.ts index 6ec859a0..8d4b237d 100644 --- a/packages/ui/src/stores/messageQueueStore.test.ts +++ b/packages/ui/src/stores/messageQueueStore.test.ts @@ -8,7 +8,7 @@ import { } from "./messageQueueStore" beforeEach(() => { - useMessageQueueStore.setState({ queuedMessages: {}, quarantinedLegacyMessages: {} }) + useMessageQueueStore.setState({ queuedMessages: {}, quarantinedLegacyMessages: {}, sendingIds: {} }) }) describe("message queue runtime ownership", () => { @@ -49,3 +49,48 @@ describe("message queue runtime ownership", () => { expect(queue[0]?.content).toBe("message-5") }) }) + +describe("in-flight queued sends", () => { + test("hides a dispatched message from the sendable queue but keeps it visible", () => { + const target = createMessageQueueTarget("session-1", "/repo", "runtime-a")! + const store = useMessageQueueStore.getState() + store.addToQueue(target, { content: "first" }) + store.addToQueue(target, { content: "second" }) + const [first] = useMessageQueueStore.getState().getQueueForTarget(target) + + useMessageQueueStore.getState().markSending(target, first.id) + + expect(useMessageQueueStore.getState().getQueueForTarget(target)).toHaveLength(2) + const sendable = useMessageQueueStore.getState().getSendableQueue(target) + expect(sendable).toHaveLength(1) + expect(sendable[0]?.content).toBe("second") + + useMessageQueueStore.getState().clearSending(target, first.id) + expect(useMessageQueueStore.getState().getSendableQueue(target)).toHaveLength(2) + expect(useMessageQueueStore.getState().sendingIds).toEqual({}) + }) + + test("clearQueue retains a message whose send is still awaiting the server", () => { + const target = createMessageQueueTarget("session-1", "/repo", "runtime-a")! + const store = useMessageQueueStore.getState() + store.addToQueue(target, { content: "in flight" }) + store.addToQueue(target, { content: "merged by composer" }) + const [inFlight] = useMessageQueueStore.getState().getQueueForTarget(target) + useMessageQueueStore.getState().markSending(target, inFlight.id) + + useMessageQueueStore.getState().clearQueue(target) + + const remaining = useMessageQueueStore.getState().getQueueForTarget(target) + expect(remaining).toHaveLength(1) + expect(remaining[0]?.id).toBe(inFlight.id) + }) + + test("clearQueue drops everything once no send is in flight", () => { + const target = createMessageQueueTarget("session-1", "/repo", "runtime-a")! + useMessageQueueStore.getState().addToQueue(target, { content: "queued" }) + + useMessageQueueStore.getState().clearQueue(target) + + expect(useMessageQueueStore.getState().getQueueForTarget(target)).toHaveLength(0) + }) +}) diff --git a/packages/ui/src/stores/messageQueueStore.ts b/packages/ui/src/stores/messageQueueStore.ts index 2ac9f86f..8ee096ec 100644 --- a/packages/ui/src/stores/messageQueueStore.ts +++ b/packages/ui/src/stores/messageQueueStore.ts @@ -85,6 +85,19 @@ interface MessageQueueState { queuedMessages: Record; // runtime + directory + session → queue quarantinedLegacyMessages: Record; followUpBehavior: FollowUpBehavior; + /** + * Queued messages whose send is currently awaiting the server, per target. + * + * A queued item is removed only after its send resolves, so between + * dispatch and resolution it is still visible to every other reader — and + * a composer submit merges the whole queue into its own send. Over a relay + * that window is seconds, long enough for the same message to be delivered + * twice. Dispatchers must skip entries listed here. + * + * Never persisted: a restart has no in-flight sends, and a stale flag would + * strand a queued message permanently. + */ + sendingIds: Record; } interface MessageQueueActions { @@ -94,6 +107,9 @@ interface MessageQueueActions { popToInput: (target: MessageQueueTarget, messageId: string) => QueuedMessage | null; clearQueue: (target: MessageQueueTarget) => void; clearAllQueues: () => void; + markSending: (target: MessageQueueTarget, messageId: string) => void; + clearSending: (target: MessageQueueTarget, messageId: string) => void; + getSendableQueue: (target: MessageQueueTarget) => QueuedMessage[]; setFollowUpBehavior: (behavior: FollowUpBehavior) => void; getQueueForTarget: (target: MessageQueueTarget) => QueuedMessage[]; } @@ -127,6 +143,7 @@ export const useMessageQueueStore = create()( queuedMessages: {}, quarantinedLegacyMessages: {}, followUpBehavior: DEFAULT_FOLLOW_UP_BEHAVIOR, + sendingIds: {}, addToQueue: (target, message) => { const key = getMessageQueueKey(target); @@ -237,6 +254,14 @@ export const useMessageQueueStore = create()( clearQueue: (target) => { const key = getMessageQueueKey(target); set((state) => { + // Clearing drops what is still queued, never a message + // already handed to the server: that send will resolve + // and must find its entry to remove or restore. + const sending = state.sendingIds[key] ?? []; + const retained = (state.queuedMessages[key] ?? []).filter((m) => sending.includes(m.id)); + if (retained.length > 0) { + return { queuedMessages: { ...state.queuedMessages, [key]: retained } }; + } const { [key]: _removed, ...rest } = state.queuedMessages; void _removed; return { queuedMessages: rest }; @@ -244,7 +269,40 @@ export const useMessageQueueStore = create()( }, clearAllQueues: () => { - set({ queuedMessages: {} }); + set({ queuedMessages: {}, sendingIds: {} }); + }, + + markSending: (target, messageId) => { + const key = getMessageQueueKey(target); + set((state) => { + const current = state.sendingIds[key] ?? []; + if (current.includes(messageId)) return state; + return { sendingIds: { ...state.sendingIds, [key]: [...current, messageId] } }; + }); + }, + + clearSending: (target, messageId) => { + const key = getMessageQueueKey(target); + set((state) => { + const current = state.sendingIds[key]; + if (!current || !current.includes(messageId)) return state; + const next = current.filter((id) => id !== messageId); + if (next.length === 0) { + const { [key]: _removed, ...rest } = state.sendingIds; + void _removed; + return { sendingIds: rest }; + } + return { sendingIds: { ...state.sendingIds, [key]: next } }; + }); + }, + + getSendableQueue: (target) => { + const key = getMessageQueueKey(target); + const state = get(); + const queue = state.queuedMessages[key] ?? []; + const sending = state.sendingIds[key]; + if (!sending || sending.length === 0) return queue; + return queue.filter((message) => !sending.includes(message.id)); }, setFollowUpBehavior: (behavior) => {