diff --git a/server/modules/websocket/services/chat-run-registry.service.ts b/server/modules/websocket/services/chat-run-registry.service.ts index a5e51b5f..101d3bac 100644 --- a/server/modules/websocket/services/chat-run-registry.service.ts +++ b/server/modules/websocket/services/chat-run-registry.service.ts @@ -318,6 +318,22 @@ export const chatRunRegistry = { run.writer.sendComplete(opts); }, + /** + * Safety-net variant of `completeRun` scoped to one specific run: a no-op + * unless `run` is still the session's current, running run. A runtime + * promise can resolve after its own `complete` already streamed AND a new + * run has replaced it in the registry (a queued message sends within + * milliseconds of the previous turn ending) — the session-keyed + * `completeRun` would terminate that newer run. + */ + completeRunIfCurrent(run: ChatRun, opts: { exitCode: number; aborted?: boolean }): void { + if (runs.get(run.appSessionId) !== run || run.status !== 'running') { + return; + } + + run.writer.sendComplete(opts); + }, + /** * Test-only escape hatch: clears every tracked run. */ diff --git a/server/modules/websocket/services/chat-websocket.service.ts b/server/modules/websocket/services/chat-websocket.service.ts index eb0f7ce1..48db6750 100644 --- a/server/modules/websocket/services/chat-websocket.service.ts +++ b/server/modules/websocket/services/chat-websocket.service.ts @@ -212,8 +212,10 @@ async function handleChatSend( } finally { // Safety net: a runtime that crashed (or resolved) without emitting its // terminal `complete` would otherwise leave the session stuck in - // "processing" forever on every connected client. - chatRunRegistry.completeRun(sessionId, { exitCode: 1 }); + // "processing" forever on every connected client. Scoped to THIS run — + // a queued message can start the session's next run before this promise + // settles, and the session-keyed completeRun would kill that new run. + chatRunRegistry.completeRunIfCurrent(run, { exitCode: 1 }); } } diff --git a/server/modules/websocket/tests/chat-run-registry.test.ts b/server/modules/websocket/tests/chat-run-registry.test.ts index d8ea8a58..1e911622 100644 --- a/server/modules/websocket/tests/chat-run-registry.test.ts +++ b/server/modules/websocket/tests/chat-run-registry.test.ts @@ -129,6 +129,44 @@ test('complete marks the run finished and duplicate completes are dropped', asyn }); }); +test('a finished run\'s safety net cannot complete the session\'s next run', async () => { + await withIsolatedDatabase(() => { + sessionsDb.createAppSession('app-run-9', 'codex', '/workspace/demo'); + const connection = new FakeConnection(); + + const firstRun = chatRunRegistry.startRun({ + appSessionId: 'app-run-9', + provider: 'codex', + providerSessionId: null, + connection, + userId: null, + }); + assert.ok(firstRun); + firstRun.writer.send({ kind: 'complete', provider: 'codex', sessionId: 'native-9', exitCode: 0 }); + + // A queued message starts the next run before the first run's runtime + // promise settles (the chat handler's `finally` hasn't executed yet). + const secondRun = chatRunRegistry.startRun({ + appSessionId: 'app-run-9', + provider: 'codex', + providerSessionId: null, + connection, + userId: null, + }); + assert.ok(secondRun); + + // First run's safety net fires late: it must not touch the new run. + chatRunRegistry.completeRunIfCurrent(firstRun, { exitCode: 1 }); + assert.equal(chatRunRegistry.isProcessing('app-run-9'), true); + assert.equal(connection.frames.filter((frame) => frame.kind === 'complete').length, 1); + + // The second run's own safety net still works while it is current. + chatRunRegistry.completeRunIfCurrent(secondRun, { exitCode: 1 }); + assert.equal(chatRunRegistry.isProcessing('app-run-9'), false); + assert.equal(connection.frames.filter((frame) => frame.kind === 'complete').length, 2); + }); +}); + test('listRunningRuns returns only currently running app sessions', async () => { await withIsolatedDatabase(() => { sessionsDb.createAppSession('app-run-7', 'claude', '/workspace/demo'); diff --git a/src/components/app/AppContent.tsx b/src/components/app/AppContent.tsx index 6cae890e..d94b75d9 100644 --- a/src/components/app/AppContent.tsx +++ b/src/components/app/AppContent.tsx @@ -10,6 +10,7 @@ import { PaletteOpsProvider, usePaletteOpsRegister } from '../../contexts/Palett import { useDeviceSettings } from '../../hooks/useDeviceSettings'; import { useSessionProtection } from '../../hooks/useSessionProtection'; import { useProjectsState } from '../../hooks/useProjectsState'; +import { useQueuedMessageAutoSend } from '../../hooks/useQueuedMessageAutoSend'; import { api } from '../../utils/api'; type RunningSessionApiItem = { @@ -84,6 +85,17 @@ function AppContentInner() { activeSessions: processingSessions, }); + // Queued messages for sessions that finish while another session (or none) + // is being viewed are sent from here; the viewed session's composer handles + // its own queue. + useQueuedMessageAutoSend({ + processingSessions, + activeSessionId: selectedSession?.id ?? sessionId ?? null, + ws, + sendMessage, + markSessionProcessing, + }); + const refreshRunningSessions = useCallback(async () => { try { const response = await api.runningSessions(); diff --git a/src/components/chat/hooks/useChatComposerState.ts b/src/components/chat/hooks/useChatComposerState.ts index 170055ac..7543ebae 100644 --- a/src/components/chat/hooks/useChatComposerState.ts +++ b/src/components/chat/hooks/useChatComposerState.ts @@ -14,7 +14,13 @@ import { useDropzone } from 'react-dropzone'; import { authenticatedFetch } from '../../../utils/api'; import type { MarkSessionProcessing } from '../../../hooks/useSessionProtection'; import { grantClaudeToolPermission } from '../utils/chatPermissions'; -import { safeLocalStorage } from '../utils/chatStorage'; +import { + clearQueuedMessage, + readQueuedMessage, + safeLocalStorage, + writeQueuedMessage, + type QueuedSendOptions, +} from '../utils/chatStorage'; import type { ChatMessage, PendingPermissionRequest, @@ -147,6 +153,18 @@ const createFakeSubmitEvent = () => { export type QueuedDraft = { content: string; images: File[]; + /** + * Send options snapshotted at queue time. Persisted with the draft so the + * app-level auto-send can dispatch the message with the right model and + * permission settings while another session is being viewed. + */ + options?: QueuedSendOptions; +}; + +const restoreQueuedDraft = (sessionKey: string): QueuedDraft | null => { + const saved = readQueuedMessage(sessionKey); + // Image attachments can't survive a reload; only text and options persist. + return saved ? { content: saved.content, images: [], options: saved.options } : null; }; const getNotificationSessionSummary = ( @@ -227,9 +245,13 @@ export function useChatComposerState({ if (typeof window === 'undefined' || !sessionKey) { return null; } - const saved = safeLocalStorage.getItem(`queued_message_${sessionKey}`); - return saved ? { content: saved, images: [] } : null; + return restoreQueuedDraft(sessionKey); }); + // Which session the in-memory `queuedDraft` belongs to. On a session switch + // there is one commit where `sessionKey` already points at the new session + // while `queuedDraft` still holds the old session's draft; the persistence + // effect must not write across that gap. + const queuedDraftSessionRef = useRef(sessionKey); const handleBuiltInCommand = useCallback( (result: CommandExecutionResult) => { @@ -561,6 +583,66 @@ export function useChatComposerState({ noKeyboard: true, }); + // Snapshot of everything `chat.send` needs beyond the text itself. Built at + // send time for immediate sends and at queue time for queued ones, so a + // queued message keeps the provider settings it was composed under even if + // it is later dispatched outside this composer (app-level auto-send). + const buildSendOptions = useCallback((currentInput: string): QueuedSendOptions => { + const getToolsSettings = () => { + try { + const settingsKey = + provider === 'cursor' + ? 'cursor-tools-settings' + : provider === 'codex' + ? 'codex-settings' + : provider === 'opencode' + ? 'opencode-settings' + : 'claude-settings'; + const savedSettings = safeLocalStorage.getItem(settingsKey); + if (savedSettings) { + return JSON.parse(savedSettings); + } + } catch (error) { + console.error('Error loading tools settings:', error); + } + + return { + allowedTools: [], + disallowedTools: [], + skipPermissions: false, + }; + }; + + const toolsSettings = getToolsSettings(); + const model = + provider === 'cursor' + ? cursorModel + : provider === 'codex' + ? codexModel + : provider === 'opencode' + ? opencodeModel + : claudeModel; + + return { + model, + effort: currentProviderEffort, + permissionMode: resolvePermissionModeForProvider(provider, permissionMode), + toolsSettings, + skipPermissions: toolsSettings?.skipPermissions || false, + sessionSummary: getNotificationSessionSummary(selectedSession, currentInput), + }; + }, [ + claudeModel, + codexModel, + currentProviderEffort, + cursorModel, + opencodeModel, + permissionMode, + provider, + resolvePermissionModeForProvider, + selectedSession, + ]); + const handleSubmit = useCallback( async ( event: FormEvent | MouseEvent | TouchEvent | KeyboardEvent, @@ -575,7 +657,12 @@ export function useChatComposerState({ // It's auto-flushed (re-running this same function) once the turn ends, // so it still goes through slash-command interception, image upload, etc. if (isLoading) { - setQueuedDraft({ content: currentInput, images: attachedImages }); + queuedDraftSessionRef.current = sessionKey; + setQueuedDraft({ + content: currentInput, + images: attachedImages, + options: buildSendOptions(currentInput), + }); setInput(''); inputValueRef.current = ''; setAttachedImages([]); @@ -728,42 +815,6 @@ export function useChatComposerState({ setIsUserScrolledUp(false); setTimeout(() => scrollToBottom(), 100); - const getToolsSettings = () => { - try { - const settingsKey = - provider === 'cursor' - ? 'cursor-tools-settings' - : provider === 'codex' - ? 'codex-settings' - : provider === 'opencode' - ? 'opencode-settings' - : 'claude-settings'; - const savedSettings = safeLocalStorage.getItem(settingsKey); - if (savedSettings) { - return JSON.parse(savedSettings); - } - } catch (error) { - console.error('Error loading tools settings:', error); - } - - return { - allowedTools: [], - disallowedTools: [], - skipPermissions: false, - }; - }; - - const toolsSettings = getToolsSettings(); - const model = - provider === 'cursor' - ? cursorModel - : provider === 'codex' - ? codexModel - : provider === 'opencode' - ? opencodeModel - : claudeModel; - const effort = currentProviderEffort; - // One message shape for every provider. The backend resolves the // provider, project path, and provider-native resume id from the // session row; `options` only carries composer-level preferences. @@ -772,12 +823,7 @@ export function useChatComposerState({ sessionId: targetSessionId, content: messageContent, options: { - model, - effort, - permissionMode: resolvePermissionModeForProvider(provider, permissionMode), - toolsSettings, - skipPermissions: toolsSettings?.skipPermissions || false, - sessionSummary, + ...buildSendOptions(messageContent), images: uploadedImages, }, }); @@ -799,23 +845,18 @@ export function useChatComposerState({ [ selectedSession, attachedImages, - claudeModel, - codexModel, - currentProviderEffort, + buildSendOptions, currentSessionId, - cursorModel, executeCommand, - opencodeModel, isLoading, onSessionProcessing, onSessionEstablished, - permissionMode, provider, - resolvePermissionModeForProvider, resetCommandMenuState, scrollToBottom, selectedProject, sendMessage, + sessionKey, addMessage, setIsUserScrolledUp, slashCommands, @@ -829,21 +870,47 @@ export function useChatComposerState({ // Once the in-flight turn ends, replay the queued draft through the normal // submit path (slash commands, image upload, etc. all still apply). const wasLoadingRef = useRef(isLoading); + const flushSessionKeyRef = useRef(sessionKey); useEffect(() => { const wasLoading = wasLoadingRef.current; wasLoadingRef.current = isLoading; - if (!wasLoading || isLoading || !queuedDraft) { + + // A session switch changes which session `isLoading` describes, so this + // transition says nothing about the queued draft's own session. Never + // flush across it — the swap effect below replaces `queuedDraft` with the + // new session's saved draft right after this. + if (flushSessionKeyRef.current !== sessionKey) { + flushSessionKeyRef.current = sessionKey; return; } - setQueuedDraft(null); - setInput(queuedDraft.content); - inputValueRef.current = queuedDraft.content; - setAttachedImages(queuedDraft.images); - setTimeout(() => { - handleSubmitRef.current?.(createFakeSubmitEvent()); - }, 0); - }, [isLoading, queuedDraft]); + if (isLoading || !queuedDraft) { + return; + } + + // Turn just ended in this session: flush immediately. Otherwise this is a + // saved draft restored into an apparently idle session — hold it briefly + // so the `chat_subscribed` ack can flip `isLoading` if a run is actually + // still live (the cleanup below cancels the send in that case). + const delay = wasLoading ? 0 : 750; + const timer = setTimeout(() => { + // The saved key is the claim ticket shared with the app-level auto-send + // (which handles sessions that finish while not viewed). If it's gone, + // the message was already dispatched — don't send it twice. + if (sessionKey && !readQueuedMessage(sessionKey)) { + setQueuedDraft(null); + return; + } + setQueuedDraft(null); + setInput(queuedDraft.content); + inputValueRef.current = queuedDraft.content; + setAttachedImages(queuedDraft.images); + setTimeout(() => { + handleSubmitRef.current?.(createFakeSubmitEvent()); + }, 0); + }, delay); + return () => clearTimeout(timer); + }, [isLoading, queuedDraft, sessionKey, setInput]); const editQueuedDraft = useCallback(() => { if (!queuedDraft) { @@ -898,28 +965,33 @@ export function useChatComposerState({ } }, [input, selectedProjectId]); - // Switching sessions swaps in that session's queued draft (image - // attachments can't survive a reload, so only the text is restored). + // Persist the queued draft under its session's key. Must be defined BEFORE + // the swap effect below: on a session switch there is one commit where + // `sessionKey` already points at the new session while `queuedDraft` (and + // the owner ref) still describe the old one — the ref mismatch makes this + // effect skip that commit instead of writing/clearing across sessions. useEffect(() => { + if (!sessionKey || queuedDraftSessionRef.current !== sessionKey) { + return; + } + if (queuedDraft?.content) { + writeQueuedMessage(sessionKey, { content: queuedDraft.content, options: queuedDraft.options }); + } else { + clearQueuedMessage(sessionKey); + } + }, [queuedDraft, sessionKey]); + + // Switching sessions swaps in that session's queued draft (image + // attachments can't survive a reload, so only text and options restore). + useEffect(() => { + queuedDraftSessionRef.current = sessionKey; if (!sessionKey) { setQueuedDraft(null); return; } - const saved = safeLocalStorage.getItem(`queued_message_${sessionKey}`); - setQueuedDraft(saved ? { content: saved, images: [] } : null); + setQueuedDraft(restoreQueuedDraft(sessionKey)); }, [sessionKey]); - useEffect(() => { - if (!sessionKey) { - return; - } - if (queuedDraft?.content) { - safeLocalStorage.setItem(`queued_message_${sessionKey}`, queuedDraft.content); - } else { - safeLocalStorage.removeItem(`queued_message_${sessionKey}`); - } - }, [queuedDraft, sessionKey]); - useEffect(() => { if (!textareaRef.current) { return; diff --git a/src/components/chat/utils/chatStorage.ts b/src/components/chat/utils/chatStorage.ts index 0365679b..3e3636bb 100644 --- a/src/components/chat/utils/chatStorage.ts +++ b/src/components/chat/utils/chatStorage.ts @@ -43,6 +43,52 @@ export const safeLocalStorage = { }, }; +/** + * Composer options captured when a message is queued, so the message can be + * sent later with the exact settings (model, permission mode, tools) the + * session's composer had at queue time — even from outside the composer, + * e.g. the app-level auto-send that fires while another session is viewed. + */ +export type QueuedSendOptions = Record; + +export type StoredQueuedMessage = { + content: string; + options?: QueuedSendOptions; +}; + +export const queuedMessageKey = (sessionId: string) => `queued_message_${sessionId}`; + +/** + * Reads a session's queued message. Understands both the JSON + * `{ content, options }` format and the legacy raw-text format. + */ +export function readQueuedMessage(sessionId: string): StoredQueuedMessage | null { + const raw = safeLocalStorage.getItem(queuedMessageKey(sessionId)); + if (!raw) { + return null; + } + + try { + const parsed = JSON.parse(raw) as unknown; + if (parsed && typeof parsed === 'object' && typeof (parsed as StoredQueuedMessage).content === 'string') { + const { content, options } = parsed as StoredQueuedMessage; + return content.trim() ? { content, options } : null; + } + } catch { + // Legacy format: the raw draft text itself. + } + + return raw.trim() ? { content: raw } : null; +} + +export function writeQueuedMessage(sessionId: string, message: StoredQueuedMessage): void { + safeLocalStorage.setItem(queuedMessageKey(sessionId), JSON.stringify(message)); +} + +export function clearQueuedMessage(sessionId: string): void { + safeLocalStorage.removeItem(queuedMessageKey(sessionId)); +} + export function getClaudeSettings(): ClaudeSettings { const raw = safeLocalStorage.getItem(CLAUDE_SETTINGS_KEY); if (!raw) { diff --git a/src/hooks/useQueuedMessageAutoSend.ts b/src/hooks/useQueuedMessageAutoSend.ts new file mode 100644 index 00000000..7be47377 --- /dev/null +++ b/src/hooks/useQueuedMessageAutoSend.ts @@ -0,0 +1,70 @@ +import { useEffect, useRef } from 'react'; + +import { clearQueuedMessage, readQueuedMessage } from '../components/chat/utils/chatStorage'; + +import type { MarkSessionProcessing, SessionActivityMap } from './useSessionProtection'; + +interface UseQueuedMessageAutoSendArgs { + processingSessions: SessionActivityMap; + /** + * The session currently open in the chat view. Its queued draft is owned by + * the composer (which also handles image attachments and slash commands), + * so this hook never touches it. + */ + activeSessionId: string | null; + ws: WebSocket | null; + sendMessage: (message: unknown) => void; + markSessionProcessing: MarkSessionProcessing; +} + +/** + * Dispatches queued messages for sessions the user is NOT currently viewing. + * + * The composer persists each queued draft (text + send options snapshotted at + * queue time) under `queued_message_`. When a session's run leaves + * the processing map — its previous response completed — this hook sends that + * session's queued message immediately instead of waiting for the user to + * open the session again. Removing the storage key before sending is the + * claim that keeps the composer's own flush from double-sending. + */ +export function useQueuedMessageAutoSend({ + processingSessions, + activeSessionId, + ws, + sendMessage, + markSessionProcessing, +}: UseQueuedMessageAutoSendArgs) { + const prevProcessingRef = useRef>(new Set()); + + useEffect(() => { + const prev = prevProcessingRef.current; + const current = new Set(processingSessions.keys()); + prevProcessingRef.current = current; + + for (const sessionId of prev) { + if (current.has(sessionId) || sessionId === activeSessionId) { + continue; + } + + const queued = readQueuedMessage(sessionId); + if (!queued) { + continue; + } + + // A closed socket would drop the send silently; keep the draft so the + // composer (or a later completion) can retry once we're connected. + if (!ws || ws.readyState !== WebSocket.OPEN) { + continue; + } + + clearQueuedMessage(sessionId); + sendMessage({ + type: 'chat.send', + sessionId, + content: queued.content, + options: { ...(queued.options ?? {}), images: [] }, + }); + markSessionProcessing(sessionId, { statusText: null, canInterrupt: true }); + } + }, [processingSessions, activeSessionId, ws, sendMessage, markSessionProcessing]); +} diff --git a/src/stores/useSessionStore.ts b/src/stores/useSessionStore.ts index 8645c618..6117cbb7 100644 --- a/src/stores/useSessionStore.ts +++ b/src/stores/useSessionStore.ts @@ -97,6 +97,16 @@ export interface SessionSlot { /** @internal Cache-invalidation refs for computeMerged */ _lastServerRef: NormalizedMessage[]; _lastRealtimeRef: NormalizedMessage[]; + /** + * @internal Monotonic ticket per server fetch (fetch/refresh/fetchMore) and + * the ticket of the last response applied. Concurrent fetches for the same + * session can resolve out of order — e.g. the `complete` refresh racing the + * watcher-triggered refresh right as a queued message is flushed — and a + * stale response applied last would wind `serverMessages` back to a + * transcript that no longer matches what the user already saw. + */ + _fetchSeq: number; + _appliedFetchSeq: number; status: SessionStatus; fetchedAt: number; total: number; @@ -120,6 +130,8 @@ function createEmptySlot(): SessionSlot { hasMore: false, offset: 0, tokenUsage: null, + _fetchSeq: 0, + _appliedFetchSeq: 0, }; } @@ -459,6 +471,7 @@ export function useSessionStore() { } = {}, ) => { const slot = getSlot(sessionId); + const fetchTicket = ++slot._fetchSeq; slot.status = 'loading'; notify(sessionId); @@ -481,6 +494,12 @@ export function useSessionStore() { const data = body?.data ?? body; const messages: NormalizedMessage[] = data.messages || []; + // A later-started fetch already applied: this response is stale. + if (fetchTicket <= slot._appliedFetchSeq) { + return slot; + } + slot._appliedFetchSeq = fetchTicket; + slot.serverMessages = messages; slot.total = data.total ?? messages.length; slot.hasMore = Boolean(data.hasMore); @@ -496,8 +515,11 @@ export function useSessionStore() { return slot; } catch (error) { console.error(`[SessionStore] fetch failed for ${sessionId}:`, error); - slot.status = 'error'; - notify(sessionId); + // Don't clobber a newer fetch's result with a stale failure. + if (fetchTicket > slot._appliedFetchSeq) { + slot.status = 'error'; + notify(sessionId); + } return slot; } }, [getSlot, notify]); @@ -514,6 +536,7 @@ export function useSessionStore() { const slot = getSlot(sessionId); if (!slot.hasMore) return slot; + const fetchTicket = ++slot._fetchSeq; const params = new URLSearchParams(); const limit = opts.limit ?? 20; params.append('limit', String(limit)); @@ -529,6 +552,13 @@ export function useSessionStore() { const data = body?.data ?? body; const olderMessages: NormalizedMessage[] = data.messages || []; + // A full fetch/refresh replaced serverMessages while this page was in + // flight — prepending onto the new array would duplicate or misorder. + if (fetchTicket <= slot._appliedFetchSeq) { + return slot; + } + slot._appliedFetchSeq = fetchTicket; + // Prepend older messages (they're earlier in the conversation) slot.serverMessages = [...olderMessages, ...slot.serverMessages]; slot.hasMore = Boolean(data.hasMore); @@ -588,6 +618,7 @@ export function useSessionStore() { sessionId: string, ) => { const slot = getSlot(sessionId); + const fetchTicket = ++slot._fetchSeq; try { const url = `/api/providers/sessions/${encodeURIComponent(sessionId)}/messages`; const response = await authenticatedFetch(url); @@ -596,6 +627,14 @@ export function useSessionStore() { const body = await response.json(); const data = body?.data ?? body; + // A later-started fetch already applied: applying this stale transcript + // would erase rows the user has already seen (and re-prune realtime + // rows against an outdated snapshot). + if (fetchTicket <= slot._appliedFetchSeq) { + return; + } + slot._appliedFetchSeq = fetchTicket; + slot.serverMessages = data.messages || []; slot.total = data.total ?? slot.serverMessages.length; slot.hasMore = Boolean(data.hasMore);