From 432fca55a7d28514e3f419a0ac7249557ba73673 Mon Sep 17 00:00:00 2001 From: brooklyn! Date: Thu, 16 Jul 2026 20:39:55 -0400 Subject: [PATCH] fix(desktop): drain queued prompts for background sessions (#66001) Co-authored-by: Jakub Wolniewicz <4850809+frizikk@users.noreply.github.com> --- .../chat/composer/hooks/use-composer-queue.ts | 18 +- apps/desktop/src/app/chat/composer/types.ts | 7 +- apps/desktop/src/app/chat/index.tsx | 7 +- apps/desktop/src/app/contrib/wiring.tsx | 10 + .../hooks/use-background-queue-drain.test.tsx | 139 ++++++++++++++ .../hooks/use-background-queue-drain.ts | 174 ++++++++++++++++++ .../hooks/use-prompt-actions/index.test.tsx | 118 +++++++++++- .../session/hooks/use-prompt-actions/slash.ts | 15 +- .../hooks/use-prompt-actions/submit.ts | 164 +++++++++++------ .../session/hooks/use-prompt-actions/utils.ts | 7 + 10 files changed, 573 insertions(+), 86 deletions(-) create mode 100644 apps/desktop/src/app/session/hooks/use-background-queue-drain.test.tsx create mode 100644 apps/desktop/src/app/session/hooks/use-background-queue-drain.ts diff --git a/apps/desktop/src/app/chat/composer/hooks/use-composer-queue.ts b/apps/desktop/src/app/chat/composer/hooks/use-composer-queue.ts index 761d6830a..9d54e17bd 100644 --- a/apps/desktop/src/app/chat/composer/hooks/use-composer-queue.ts +++ b/apps/desktop/src/app/chat/composer/hooks/use-composer-queue.ts @@ -8,6 +8,7 @@ import { resetBrowseState } from '@/store/composer-input-history' import { $queuedPromptsBySession, enqueueQueuedPrompt, + getQueuedPrompts, MAX_AUTO_DRAIN_ATTEMPTS, migrateQueuedPrompts, promoteQueuedPrompt, @@ -189,7 +190,9 @@ export function useComposerQueue({ return false } - const entry = pickEntry(queuedPrompts) + const drainQueueSessionKey = activeQueueSessionKey + const drainRuntimeSessionId = sessionId ?? null + const entry = pickEntry(getQueuedPrompts(drainQueueSessionKey)) if (!entry) { return false @@ -199,7 +202,12 @@ export function useComposerQueue({ try { const accepted = await Promise.resolve( - onSubmit(entry.text, { attachments: entry.attachments, fromQueue: true }) + onSubmit(entry.text, { + attachments: entry.attachments, + fromQueue: true, + sessionId: drainRuntimeSessionId, + storedSessionId: drainQueueSessionKey + }) ) if (accepted === false) { @@ -207,15 +215,15 @@ export function useComposerQueue({ } drainFailuresRef.current.delete(entry.id) - removeQueuedPrompt(activeQueueSessionKey, entry.id) - resetBrowseState(sessionId) + removeQueuedPrompt(drainQueueSessionKey, entry.id) + resetBrowseState(drainRuntimeSessionId) return true } finally { drainingQueueRef.current = false } }, - [activeQueueSessionKey, onSubmit, queuedPrompts, sessionId] + [activeQueueSessionKey, onSubmit, sessionId] ) const pickDrainHead = useCallback( diff --git a/apps/desktop/src/app/chat/composer/types.ts b/apps/desktop/src/app/chat/composer/types.ts index 59c7c1727..24abcf373 100644 --- a/apps/desktop/src/app/chat/composer/types.ts +++ b/apps/desktop/src/app/chat/composer/types.ts @@ -1,7 +1,7 @@ import type { ReactNode } from 'react' +import type { SubmitTextOptions } from '@/app/session/hooks/use-prompt-actions/utils' import type { HermesGateway } from '@/hermes' -import type { ComposerAttachment } from '@/store/composer' import type { DroppedFile } from '../hooks/use-composer-actions' @@ -52,10 +52,7 @@ export interface ChatBarProps { onPickImages?: () => void onRemoveAttachment?: (id: string) => void onSteer?: (text: string) => Promise | boolean - onSubmit: ( - value: string, - options?: { attachments?: ComposerAttachment[]; fromQueue?: boolean } - ) => Promise | boolean + onSubmit: (value: string, options?: SubmitTextOptions) => Promise | boolean onTranscribeAudio?: (audio: Blob) => Promise } diff --git a/apps/desktop/src/app/chat/index.tsx b/apps/desktop/src/app/chat/index.tsx index 4b417404b..681c6cf0a 100644 --- a/apps/desktop/src/app/chat/index.tsx +++ b/apps/desktop/src/app/chat/index.tsx @@ -5,6 +5,7 @@ import type * as React from 'react' import { Suspense, useCallback, useMemo } from 'react' import { useLocation } from 'react-router-dom' +import type { SubmitTextOptions } from '@/app/session/hooks/use-prompt-actions/utils' import { Thread } from '@/components/assistant-ui/thread' import { Backdrop } from '@/components/Backdrop' import { COMPOSER_HEART_CONFIG, HeartField } from '@/components/chat/vibe-hearts' @@ -19,7 +20,6 @@ import type { ChatMessage } from '@/lib/chat-messages' import { quickModelOptions, sessionTitle } from '@/lib/chat-runtime' import { useIncrementalExternalStoreRuntime } from '@/lib/incremental-external-store-runtime' import { cn } from '@/lib/utils' -import type { ComposerAttachment } from '@/store/composer' import { $pinnedSessionIds } from '@/store/layout' import { $petActive } from '@/store/pet' import { $petOverlayActive } from '@/store/pet-overlay' @@ -74,10 +74,7 @@ interface ChatViewProps extends Omit, 'onSubmit'> { onPickImages: () => void onRemoveAttachment: (id: string) => void onSteer: (text: string) => Promise | boolean - onSubmit: ( - text: string, - options?: { attachments?: ComposerAttachment[]; fromQueue?: boolean } - ) => Promise | boolean + onSubmit: (text: string, options?: SubmitTextOptions) => Promise | boolean onThreadMessagesChange: (messages: readonly ThreadMessage[]) => void onEdit: (message: AppendMessage) => Promise onReload: (parentId: string | null) => Promise diff --git a/apps/desktop/src/app/contrib/wiring.tsx b/apps/desktop/src/app/contrib/wiring.tsx index 3f7529fff..f47a2301b 100644 --- a/apps/desktop/src/app/contrib/wiring.tsx +++ b/apps/desktop/src/app/contrib/wiring.tsx @@ -75,6 +75,7 @@ import { PersistentTerminal } from '../right-sidebar/terminal/persistent' import { CRON_ROUTE, routeSessionId, sessionRoute, SETTINGS_ROUTE, syncWorkspaceIsPage } from '../routes' import { SessionPickerOverlay } from '../session-picker-overlay' import { SessionSwitcher } from '../session-switcher' +import { useBackgroundQueueDrain } from '../session/hooks/use-background-queue-drain' import { useContextSuggestions } from '../session/hooks/use-context-suggestions' import { useCwdActions } from '../session/hooks/use-cwd-actions' import { useHermesConfig } from '../session/hooks/use-hermes-config' @@ -536,6 +537,15 @@ export function ContribWiring({ children }: { children: ReactNode }) { updateSessionState }) + // Runs outside the selected ChatBar so queues belonging to background + // sessions continue once those sessions are idle. + useBackgroundQueueDrain({ + enabled: gatewayState === 'open', + runtimeIdByStoredSessionIdRef, + selectedStoredSessionId, + submitText + }) + // Session-tile delegate (resume/submit/interrupt/slash + the session verbs // the tile TAB menu needs, without touching the primary view). useSessionTileDelegate({ diff --git a/apps/desktop/src/app/session/hooks/use-background-queue-drain.test.tsx b/apps/desktop/src/app/session/hooks/use-background-queue-drain.test.tsx new file mode 100644 index 000000000..e6cd960c1 --- /dev/null +++ b/apps/desktop/src/app/session/hooks/use-background-queue-drain.test.tsx @@ -0,0 +1,139 @@ +import { act, cleanup, render, waitFor } from '@testing-library/react' +import type { MutableRefObject } from 'react' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' + +import { $queuedPromptsBySession, enqueueQueuedPrompt, getQueuedPrompts } from '@/store/composer-queue' +import { $workingSessionIds } from '@/store/session' + +import { useBackgroundQueueDrain } from './use-background-queue-drain' +import type { SubmitTextOptions } from './use-prompt-actions/utils' + +function Harness({ + enabled = true, + runtimeMap, + selectedStoredSessionId = 'stored-session-b', + submitText +}: { + enabled?: boolean + runtimeMap: MutableRefObject> + selectedStoredSessionId?: string | null + submitText: (text: string, options?: SubmitTextOptions) => Promise | boolean +}) { + useBackgroundQueueDrain({ + enabled, + runtimeIdByStoredSessionIdRef: runtimeMap, + selectedStoredSessionId, + submitText + }) + + return null +} + +describe('useBackgroundQueueDrain', () => { + beforeEach(() => { + vi.useRealTimers() + }) + + afterEach(() => { + cleanup() + vi.restoreAllMocks() + vi.useRealTimers() + $queuedPromptsBySession.set({}) + $workingSessionIds.set([]) + }) + + it('drains an idle queued prompt for a non-selected background session', async () => { + const runtimeMap = { current: new Map([['stored-session-a', 'rt-session-a']]) } + const submitText = vi.fn(async () => true) + + enqueueQueuedPrompt('stored-session-a', { text: 'continue in the background', attachments: [] }) + $workingSessionIds.set([]) + + render() + + await waitFor(() => { + expect(submitText).toHaveBeenCalledWith('continue in the background', { + attachments: [], + fromQueue: true, + sessionId: 'rt-session-a', + storedSessionId: 'stored-session-a' + }) + }) + + await waitFor(() => expect(getQueuedPrompts('stored-session-a')).toHaveLength(0)) + }) + + it('leaves the selected session queue to the mounted ChatBar drainer', async () => { + const runtimeMap = { current: new Map([['stored-session-a', 'rt-session-a']]) } + const submitText = vi.fn(async () => true) + + enqueueQueuedPrompt('stored-session-a', { text: 'visible queue entry', attachments: [] }) + $workingSessionIds.set([]) + + render() + + await new Promise(resolve => window.setTimeout(resolve, 0)) + + expect(submitText).not.toHaveBeenCalled() + expect(getQueuedPrompts('stored-session-a')).toHaveLength(1) + }) + + it('does not drain a background session that is still marked working', async () => { + const runtimeMap = { current: new Map([['stored-session-a', 'rt-session-a']]) } + const submitText = vi.fn(async () => true) + + enqueueQueuedPrompt('stored-session-a', { text: 'wait for current turn', attachments: [] }) + $workingSessionIds.set(['stored-session-a']) + + render() + + await new Promise(resolve => window.setTimeout(resolve, 0)) + + expect(submitText).not.toHaveBeenCalled() + expect(getQueuedPrompts('stored-session-a')).toHaveLength(1) + }) + + it('passes a null runtime id so submitText can resume stale background sessions by stored id', async () => { + const runtimeMap = { current: new Map() } + const submitText = vi.fn(async () => true) + + enqueueQueuedPrompt('stored-session-a', { text: 'resume then send', attachments: [] }) + + render() + + await waitFor(() => { + expect(submitText).toHaveBeenCalledWith('resume then send', { + attachments: [], + fromQueue: true, + sessionId: null, + storedSessionId: 'stored-session-a' + }) + }) + }) + + it('retries a rejected background drain without waiting for another queue or busy-state change', async () => { + vi.useFakeTimers() + + const runtimeMap = { current: new Map([['stored-session-a', 'rt-session-a']]) } + const submitText = vi.fn().mockResolvedValueOnce(false).mockResolvedValueOnce(true) + + enqueueQueuedPrompt('stored-session-a', { text: 'retry me', attachments: [] }) + + render() + + await act(async () => { + await Promise.resolve() + }) + + expect(submitText).toHaveBeenCalledTimes(1) + expect(getQueuedPrompts('stored-session-a')).toHaveLength(1) + + await act(async () => { + await vi.advanceTimersByTimeAsync(750) + await Promise.resolve() + }) + + expect(submitText).toHaveBeenCalledTimes(2) + expect(getQueuedPrompts('stored-session-a')).toHaveLength(0) + }) +}) diff --git a/apps/desktop/src/app/session/hooks/use-background-queue-drain.ts b/apps/desktop/src/app/session/hooks/use-background-queue-drain.ts new file mode 100644 index 000000000..83d389e97 --- /dev/null +++ b/apps/desktop/src/app/session/hooks/use-background-queue-drain.ts @@ -0,0 +1,174 @@ +import { useStore } from '@nanostores/react' +import { type MutableRefObject, useCallback, useEffect, useRef, useState } from 'react' + +import { useI18n } from '@/i18n' +import { resetBrowseState } from '@/store/composer-input-history' +import { + $queuedPromptsBySession, + getQueuedPrompts, + MAX_AUTO_DRAIN_ATTEMPTS, + type QueuedPromptEntry, + removeQueuedPrompt, + shouldAutoDrain +} from '@/store/composer-queue' +import { notify } from '@/store/notifications' +import { $workingSessionIds } from '@/store/session' + +import type { SubmitTextOptions } from './use-prompt-actions/utils' + +type SubmitQueuedPrompt = (text: string, options?: SubmitTextOptions) => Promise | boolean + +interface BackgroundQueueDrainOptions { + enabled: boolean + runtimeIdByStoredSessionIdRef: MutableRefObject> + selectedStoredSessionId: string | null + submitText: SubmitQueuedPrompt +} + +const BACKGROUND_DRAIN_RETRY_MS = 750 + +/** + * Drain queued prompts for sessions that are not currently rendered by ChatBar. + * + * The visible ChatBar owns the interactive queue panel for the selected session. + * Without this background drain, a prompt queued in Session A can sit forever + * after the user switches to Session B: the only auto-drain effect lives inside + * the mounted ChatBar, so Session A's queue is not observed when A is offscreen. + */ +export function useBackgroundQueueDrain({ + enabled, + runtimeIdByStoredSessionIdRef, + selectedStoredSessionId, + submitText +}: BackgroundQueueDrainOptions) { + const { t } = useI18n() + const queuedPromptsBySession = useStore($queuedPromptsBySession) + const workingSessionIds = useStore($workingSessionIds) + const submitTextRef = useRef(submitText) + const drainingSessionIdsRef = useRef(new Set()) + const drainFailuresRef = useRef(new Map()) + const retryTimersRef = useRef([]) + const [retryTick, setRetryTick] = useState(0) + + useEffect(() => { + submitTextRef.current = submitText + }, [submitText]) + + const scheduleRetry = useCallback(() => { + if (typeof window === 'undefined') { + return + } + + const timer = window.setTimeout(() => { + retryTimersRef.current = retryTimersRef.current.filter(id => id !== timer) + setRetryTick(tick => tick + 1) + }, BACKGROUND_DRAIN_RETRY_MS) + + retryTimersRef.current.push(timer) + }, []) + + useEffect( + () => () => { + for (const timer of retryTimersRef.current) { + window.clearTimeout(timer) + } + + retryTimersRef.current = [] + }, + [] + ) + + const drainSessionQueue = useCallback( + (sessionKey: string, entry: QueuedPromptEntry) => { + if (drainingSessionIdsRef.current.has(sessionKey)) { + return + } + + drainingSessionIdsRef.current.add(sessionKey) + + const onFail = () => { + const failures = (drainFailuresRef.current.get(entry.id) ?? 0) + 1 + drainFailuresRef.current.set(entry.id, failures) + + if (failures >= MAX_AUTO_DRAIN_ATTEMPTS) { + notify({ + id: `composer-background-queue-stuck-${sessionKey}`, + kind: 'error', + title: t.composer.queueStuckTitle, + message: t.composer.queueStuckBody + }) + + return + } + + scheduleRetry() + } + + void Promise.resolve() + .then(async () => { + const liveEntry = getQueuedPrompts(sessionKey).find(candidate => candidate.id === entry.id) + + if (!liveEntry) { + return true + } + + const runtimeSessionId = runtimeIdByStoredSessionIdRef.current.get(sessionKey) ?? null + + const accepted = await Promise.resolve( + submitTextRef.current(liveEntry.text, { + attachments: liveEntry.attachments, + fromQueue: true, + sessionId: runtimeSessionId, + storedSessionId: sessionKey + }) + ) + + if (accepted === false) { + return false + } + + drainFailuresRef.current.delete(liveEntry.id) + removeQueuedPrompt(sessionKey, liveEntry.id) + resetBrowseState(runtimeSessionId) + + return true + }) + .then(accepted => { + if (!accepted) { + onFail() + } + }) + .catch(onFail) + .finally(() => { + drainingSessionIdsRef.current.delete(sessionKey) + }) + }, + [runtimeIdByStoredSessionIdRef, scheduleRetry, t] + ) + + useEffect(() => { + if (!enabled) { + return + } + + const working = new Set(workingSessionIds) + + for (const [sessionKey, entries] of Object.entries(queuedPromptsBySession)) { + if ( + sessionKey === selectedStoredSessionId || + drainingSessionIdsRef.current.has(sessionKey) || + !shouldAutoDrain({ isBusy: working.has(sessionKey), queueLength: entries.length }) + ) { + continue + } + + const entry = entries[0] + + if (!entry || (drainFailuresRef.current.get(entry.id) ?? 0) >= MAX_AUTO_DRAIN_ATTEMPTS) { + continue + } + + drainSessionQueue(sessionKey, entry) + } + }, [drainSessionQueue, enabled, queuedPromptsBySession, retryTick, selectedStoredSessionId, workingSessionIds]) +} diff --git a/apps/desktop/src/app/session/hooks/use-prompt-actions/index.test.tsx b/apps/desktop/src/app/session/hooks/use-prompt-actions/index.test.tsx index d721498d2..46fc10290 100644 --- a/apps/desktop/src/app/session/hooks/use-prompt-actions/index.test.tsx +++ b/apps/desktop/src/app/session/hooks/use-prompt-actions/index.test.tsx @@ -8,6 +8,8 @@ import { $composerAttachments, $composerDraft, type ComposerAttachment, setCompo import { $busy, $connection, $messages, $sessions, $turnStartedAt, setSessions } from '@/store/session' import type { SessionInfo } from '@/types/hermes' +import type { SubmitTextOptions } from './utils' + import { uploadComposerAttachment, usePromptActions } from '.' vi.mock('@/hermes', () => ({ @@ -56,10 +58,11 @@ async function actRender(ui: React.ReactElement) { } interface HarnessHandle { + activeSessionIdRef: MutableRefObject cancelRun: () => Promise restoreToMessage: (messageId: string, target?: { text?: string; userOrdinal?: number | null }) => Promise steerPrompt: (text: string) => Promise - submitText: (text: string, options?: { attachments?: ComposerAttachment[]; fromQueue?: boolean }) => Promise + submitText: (text: string, options?: SubmitTextOptions) => Promise } function Harness({ @@ -68,6 +71,7 @@ function Harness({ getRoutedStoredSessionId, getRuntimeIdForStoredSession, getRouteToken, + onUpdateState, onReady, onSeedState, openMemoryGraph, @@ -85,6 +89,11 @@ function Harness({ getRoutedStoredSessionId?: () => null | string getRuntimeIdForStoredSession?: (storedSessionId: string) => null | string getRouteToken?: () => string + onUpdateState?: ( + sessionId: string, + storedSessionId: null | string | undefined, + state: Record + ) => void onReady: (handle: HarnessHandle) => void onSeedState?: (state: Record) => void openMemoryGraph?: () => void @@ -97,9 +106,8 @@ function Harness({ activeSessionId?: null | string createBackendSessionForSend?: () => Promise }) { - const activeSessionIdRef: MutableRefObject = activeSessionIdRefProp ?? { - current: activeSessionId === undefined ? RUNTIME_SESSION_ID : activeSessionId - } + const localActiveSessionIdRef = useRef(activeSessionId === undefined ? RUNTIME_SESSION_ID : activeSessionId) + const activeSessionIdRef = activeSessionIdRefProp ?? localActiveSessionIdRef const selectedStoredSessionIdRef: MutableRefObject = selectedStoredSessionIdRefProp ?? { current: storedSessionId === undefined ? RUNTIME_SESSION_ID : storedSessionId @@ -131,11 +139,12 @@ function Harness({ selectedStoredSessionIdRef, startFreshSessionDraft: () => undefined, sttEnabled: false, - updateSessionState: (_sessionId, updater) => { + updateSessionState: (sessionId, updater, storedSessionId) => { // Seed with interrupted:true so we can prove a fresh submit clears it. const next = updater(stateRef.current) as unknown as Record stateRef.current = next as never onSeedState?.(next) + onUpdateState?.(sessionId, storedSessionId, next) return next as never } @@ -143,6 +152,7 @@ function Harness({ useEffect(() => { onReady({ + activeSessionIdRef, cancelRun: (...args: Parameters) => act(async () => actions.cancelRun(...args)) as Promise, restoreToMessage: (...args: Parameters) => @@ -152,7 +162,7 @@ function Harness({ submitText: (...args: Parameters) => act(async () => actions.submitText(...args)) as Promise }) - }, [actions.cancelRun, actions.restoreToMessage, actions.steerPrompt, actions.submitText, onReady]) + }, [actions.cancelRun, actions.restoreToMessage, actions.steerPrompt, actions.submitText, activeSessionIdRef, onReady]) return null } @@ -540,6 +550,45 @@ describe('usePromptActions submit / queue drain semantics', () => { ) }) + it('a fromQueue drain sends to its queued session even after the active session changes', async () => { + $busy.set(false) + + const updates: { sessionId: string; state: Record; storedSessionId: null | string | undefined }[] = [] + const requestGateway = vi.fn(async () => ({}) as never) + + let handle: HarnessHandle | null = null + render( + (handle = h)} + onUpdateState={(sessionId, storedSessionId, state) => updates.push({ sessionId, state, storedSessionId })} + refreshSessions={async () => undefined} + requestGateway={requestGateway} + /> + ) + + const accepted = await handle!.submitText('queued for background session', { + fromQueue: true, + sessionId: 'rt-session-a', + storedSessionId: 'stored-session-a' + }) + + expect(accepted).toBe(true) + expect(requestGateway).toHaveBeenCalledWith( + 'prompt.submit', + { + session_id: 'rt-session-a', + text: 'queued for background session' + }, + 1_800_000 + ) + expect(requestGateway).not.toHaveBeenCalledWith('session.resume', expect.anything()) + expect( + updates.some(update => update.sessionId === 'rt-session-a' && update.storedSessionId === 'stored-session-a') + ).toBe(true) + // Offscreen queue drains must not flip the foreground composer into Thinking. + expect($busy.get()).toBe(false) + }) + it('a rejected fromQueue drain returns false (entry stays queued) and a later retry sends it', async () => { // A stale-session 404 must not strand the queued entry: submitPrompt returns // false on failure so the composer keeps it, and the edge-independent @@ -1132,6 +1181,58 @@ describe('usePromptActions sleep/wake session recovery', () => { expect(calls[2]?.params).toEqual({ session_id: RECOVERED_SESSION_ID, text: 'message after wake' }) }) + it('background queue resume uses the queued stored id and leaves foreground runtime selected', async () => { + const calls: { method: string; params?: Record }[] = [] + let submitAttempts = 0 + + const requestGateway = vi.fn(async (method: string, params?: Record) => { + calls.push({ method, params }) + + if (method === 'prompt.submit') { + submitAttempts += 1 + + if (submitAttempts === 1) { + throw new Error('session not found') + } + + return {} as never + } + + if (method === 'session.resume') { + return { session_id: RECOVERED_SESSION_ID } as never + } + + return {} as never + }) + + let handle: HarnessHandle | null = null + render( + (handle = h)} + refreshSessions={async () => undefined} + requestGateway={requestGateway} + storedSessionId="stored-foreground" + /> + ) + await waitFor(() => expect(handle).not.toBeNull()) + + const ok = await handle!.submitText('queued background message after wake', { + fromQueue: true, + sessionId: 'rt-background-stale', + storedSessionId: STORED_SESSION_ID + }) + + expect(ok).toBe(true) + expect(calls.map(c => c.method)).toEqual(['prompt.submit', 'session.resume', 'prompt.submit']) + expect(calls[0]?.params).toEqual({ session_id: 'rt-background-stale', text: 'queued background message after wake' }) + expect(calls[1]?.params).toEqual({ session_id: STORED_SESSION_ID, source: 'desktop' }) + expect(calls[2]?.params).toEqual({ + session_id: RECOVERED_SESSION_ID, + text: 'queued background message after wake' + }) + expect(handle!.activeSessionIdRef.current).toBe(RUNTIME_SESSION_ID) + }) + it('resumes the stored session and retries once when session.interrupt reports "session not found"', async () => { const calls: { method: string; params?: Record }[] = [] let interruptAttempts = 0 @@ -1346,7 +1447,7 @@ describe('usePromptActions sleep/wake session recovery', () => { expect(ok).toBe(true) expect(createBackendSessionForSend).not.toHaveBeenCalled() expect(calls.map(c => c.method)).toEqual(['session.resume', 'prompt.submit']) - expect(calls[0]?.params).toEqual({ session_id: STORED_SESSION_ID }) + expect(calls[0]?.params).toEqual({ session_id: STORED_SESSION_ID, source: 'desktop' }) expect(calls[1]?.params).toMatchObject({ session_id: RECOVERED_SESSION_ID }) }) @@ -1655,7 +1756,8 @@ describe('usePromptActions submit session-context isolation (#54527)', () => { expect(await submitting).toBe(false) expect(calls.some(c => c.method === 'prompt.submit')).toBe(false) expect(calls.find(c => c.method === 'session.resume')?.params).toEqual({ - session_id: STORED_SESSION_A + session_id: STORED_SESSION_A, + source: 'desktop' }) }) diff --git a/apps/desktop/src/app/session/hooks/use-prompt-actions/slash.ts b/apps/desktop/src/app/session/hooks/use-prompt-actions/slash.ts index def08fe6e..d80c65f6b 100644 --- a/apps/desktop/src/app/session/hooks/use-prompt-actions/slash.ts +++ b/apps/desktop/src/app/session/hooks/use-prompt-actions/slash.ts @@ -14,7 +14,7 @@ import { } from '@/lib/desktop-slash-commands' import { setSessionYolo } from '@/lib/yolo-session' import { openCommandPalettePage } from '@/store/command-palette' -import { type ComposerAttachment, setComposerDraft } from '@/store/composer' +import { setComposerDraft } from '@/store/composer' import { notify, notifyError } from '@/store/notifications' import { setPetScale } from '@/store/pet-gallery' import { $petGenInput, openPetGenerate } from '@/store/pet-generate' @@ -31,7 +31,13 @@ import { import type { BrowserManageResponse, SessionTitleResponse, SlashExecResponse } from '../../../types' -import { type GatewayRequest, isSessionIdCandidate, renderCommandsCatalog, slashStatusText } from './utils' +import { + type GatewayRequest, + isSessionIdCandidate, + renderCommandsCatalog, + slashStatusText, + type SubmitTextOptions +} from './utils' /** Everything a slash handler needs about the invocation it's serving. */ interface SlashActionCtx { @@ -59,10 +65,7 @@ interface SlashCommandDeps { requestGateway: GatewayRequest resumeStoredSession: (storedSessionId: string) => Promise | void startFreshSessionDraft: () => void - submitPromptText: ( - rawText: string, - options?: { attachments?: ComposerAttachment[]; fromQueue?: boolean } - ) => Promise + submitPromptText: (rawText: string, options?: SubmitTextOptions) => Promise } /** The /slash command dispatcher, extracted from usePromptActions. */ diff --git a/apps/desktop/src/app/session/hooks/use-prompt-actions/submit.ts b/apps/desktop/src/app/session/hooks/use-prompt-actions/submit.ts index ce13b6a65..a973f82f4 100644 --- a/apps/desktop/src/app/session/hooks/use-prompt-actions/submit.ts +++ b/apps/desktop/src/app/session/hooks/use-prompt-actions/submit.ts @@ -141,9 +141,19 @@ export function useSubmitPrompt(deps: SubmitPromptDeps) { return false } - // Pin the session context for the whole async submit pipeline. Without - // this, a fast session switch during session.resume / file.attach can - // redirect the user's text into a different chat (#54527). Mutable — + // Queue drains carry their source session explicitly. A background drain + // must never inherit the currently selected session after the user moves + // to another chat. + const targetStoredSessionId = options?.storedSessionId ?? selectedStoredSessionIdRef.current + + const targetStartedInCurrentView = + !targetStoredSessionId || targetStoredSessionId === selectedStoredSessionIdRef.current + + let sessionId: null | string = options?.sessionId ?? activeSessionIdRef.current + + // Pin the foreground session context for the whole async submit pipeline. + // Without this, a fast session switch during session.resume / file.attach + // can redirect the user's text into a different chat (#54527). Mutable — // not const — because a new-chat submit legitimately re-homes to the // session it creates (see the re-pin after createBackendSessionForSend). const startingActiveSessionId = activeSessionIdRef.current @@ -166,11 +176,16 @@ export function useSubmitPrompt(deps: SubmitPromptDeps) { let startingRouteToken = getRouteToken() const sessionContextDrifted = (): boolean => - selectedStoredSessionIdRef.current !== startingStoredSessionId || getRouteToken() !== startingRouteToken + targetStartedInCurrentView && + (selectedStoredSessionIdRef.current !== startingStoredSessionId || getRouteToken() !== startingRouteToken) + + const targetIsCurrentView = (): boolean => targetStartedInCurrentView && !sessionContextDrifted() // One submit in flight per session — drop any concurrent re-fire so a - // stalled turn can't stack the same prompt into multiple real turns. - const submitLockKey = startingStoredSessionId || startingActiveSessionId || '__pending_new__' + // stalled turn can't stack the same prompt into multiple real turns. The + // foreground ChatBar and background drainers can briefly overlap during a + // session switch; this per-session lock makes that safe. + const submitLockKey = targetStoredSessionId || sessionId || startingActiveSessionId || '__pending_new__' if (_submitInFlight.has(submitLockKey)) { return false @@ -197,9 +212,12 @@ export function useSubmitPrompt(deps: SubmitPromptDeps) { const releaseBusy = () => { releaseSubmitLock() - setMutableRef(busyRef, false) - scope.setBusy(false) - scope.setAwaitingResponse(false) + + if (targetIsCurrentView()) { + setMutableRef(busyRef, false) + scope.setBusy(false) + scope.setAwaitingResponse(false) + } } // Idempotent optimistic insert — re-running with the resolved sessionId @@ -221,7 +239,7 @@ export function useSubmitPrompt(deps: SubmitPromptDeps) { // (what made drained-after-interrupt sends go silent). interrupted: false }), - startingStoredSessionId + targetStoredSessionId ) // After sync rewrites refs, refresh the optimistic message in place so the @@ -233,12 +251,14 @@ export function useSubmitPrompt(deps: SubmitPromptDeps) { ...state, messages: state.messages.map(message => (message.id === optimisticId ? buildUserMessage() : message)) }), - startingStoredSessionId + targetStoredSessionId ) const dropOptimistic = (sid: null | string) => { if (!sid) { - scope.setMessages(current => current.filter(m => m.id !== optimisticId)) + if (targetIsCurrentView()) { + scope.setMessages(current => current.filter(m => m.id !== optimisticId)) + } return } @@ -252,7 +272,7 @@ export function useSubmitPrompt(deps: SubmitPromptDeps) { awaitingResponse: false, pendingBranchGroup: null }), - startingStoredSessionId + targetStoredSessionId ) } @@ -263,19 +283,26 @@ export function useSubmitPrompt(deps: SubmitPromptDeps) { return false } - setMutableRef(busyRef, true) - scope.setBusy(true) - scope.setAwaitingResponse(true) - clearNotifications() + // Foreground-only state: a background queue drain must never write the + // selected view's busy/awaiting flags or clear its notifications. + if (targetIsCurrentView()) { + setMutableRef(busyRef, true) + scope.setBusy(true) + scope.setAwaitingResponse(true) + clearNotifications() + } // A route whose selected/runtime binding is incomplete or cross-wired - // outranks any stale render-time runtime id (often from the previous - // profile). Force the full routed resume path below in that case. - let sessionId: null | string = routedSessionNeedsResume ? null : activeSessionIdRef.current + // outranks a stale render-time runtime id (often from the previous + // profile): force the full routed resume path below. An explicit queued + // runtime id (background drain) is authoritative and is left untouched. + if (!options?.sessionId && routedSessionNeedsResume) { + sessionId = null + } if (sessionId) { seedOptimistic(sessionId) - } else { + } else if (targetIsCurrentView()) { scope.setMessages(current => [...current, buildUserMessage()]) } @@ -313,17 +340,15 @@ export function useSubmitPrompt(deps: SubmitPromptDeps) { seedOptimistic(sessionId) } - if (!sessionId && startingStoredSessionId) { - // A stored session is SELECTED but its runtime binding is gone (the - // live session was orphan-reaped, or a timeout/reconnect cleared - // activeSessionId). Continuing the selected conversation must mean - // resuming it — minting a brand-new backend session here silently - // splits the user's chat in two (#55578 symptom b). Only fall through - // to session creation when NO stored session is selected (a genuine - // new-chat draft). + if (!sessionId && targetStoredSessionId) { + // A target stored session exists but its runtime binding is gone (the + // live session was orphan-reaped, a timeout/reconnect cleared it, or a + // background queue drain only has the durable id). Continue that target + // conversation; only a genuine new-chat draft may create a new session. try { const resumed = await requestGateway<{ session_id: string }>('session.resume', { - session_id: startingStoredSessionId + session_id: targetStoredSessionId, + source: 'desktop' }) if (sessionContextDrifted()) { @@ -332,12 +357,17 @@ export function useSubmitPrompt(deps: SubmitPromptDeps) { if (resumed?.session_id) { sessionId = resumed.session_id - activeSessionIdRef.current = sessionId + + if (targetIsCurrentView()) { + activeSessionIdRef.current = sessionId + } } } catch { - // A selected stored conversation is not a new-chat draft. If its + // A target stored conversation is not a new-chat draft. If its // runtime cannot be rebound, stop here rather than silently replacing - // it with a contextless session. + // it with a contextless session (#55578). For a background/queued + // drain this abort is a no-op on foreground state (both helpers are + // targetIsCurrentView-guarded) and simply drops the queued send. return abortForSessionSwitch(null) } @@ -358,7 +388,10 @@ export function useSubmitPrompt(deps: SubmitPromptDeps) { } catch (err) { dropOptimistic(null) releaseBusy() - notifyError(err, copy.sessionUnavailable) + + if (targetIsCurrentView()) { + notifyError(err, copy.sessionUnavailable) + } return false } @@ -373,7 +406,10 @@ export function useSubmitPrompt(deps: SubmitPromptDeps) { dropOptimistic(null) releaseBusy() - notify({ kind: 'error', title: copy.sessionUnavailable, message: copy.createSessionFailed }) + + if (targetIsCurrentView()) { + notify({ kind: 'error', title: copy.sessionUnavailable, message: copy.createSessionFailed }) + } return false } @@ -424,14 +460,19 @@ export function useSubmitPrompt(deps: SubmitPromptDeps) { requestGateway('prompt.submit', { session_id: sessionId, text }, PROMPT_SUBMIT_REQUEST_TIMEOUT_MS) ) } catch (firstErr) { - if ((isSessionNotFoundError(firstErr) || isGatewayTimeoutError(firstErr)) && startingStoredSessionId) { + const recoverStoredSessionId = targetStoredSessionId ?? selectedStoredSessionIdRef.current + + if ( + (isSessionNotFoundError(firstErr) || isGatewayTimeoutError(firstErr)) && + recoverStoredSessionId + ) { // Re-register the session in the gateway and get a fresh live ID. // Timeouts recover the same way as "session not found": a starved // backend loop (#55578 symptom d) rejects the submit even though // the stored session is fine — resume + retry instead of erroring // out and losing the session binding. const resumed = await requestGateway<{ session_id: string }>('session.resume', { - session_id: startingStoredSessionId, + session_id: recoverStoredSessionId, source: 'desktop' }) @@ -442,7 +483,10 @@ export function useSubmitPrompt(deps: SubmitPromptDeps) { const recoveredId = resumed?.session_id if (recoveredId) { - activeSessionIdRef.current = recoveredId + if (targetIsCurrentView()) { + activeSessionIdRef.current = recoveredId + } + await withSessionBusyRetry(() => requestGateway('prompt.submit', { session_id: recoveredId, text }, PROMPT_SUBMIT_REQUEST_TIMEOUT_MS) ) @@ -479,31 +523,37 @@ export function useSubmitPrompt(deps: SubmitPromptDeps) { const message = inlineErrorMessage(err, copy.promptFailed) - updateSessionState(sessionId, state => ({ - ...state, - messages: [ - ...state.messages, - { - id: `assistant-error-${Date.now()}`, - role: 'assistant', - parts: [], - error: message || copy.promptFailed, - branchGroupId: state.pendingBranchGroup ?? undefined - } - ], - busy: false, - awaitingResponse: false, - pendingBranchGroup: null, - sawAssistantPayload: true - })) + updateSessionState( + sessionId, + state => ({ + ...state, + messages: [ + ...state.messages, + { + id: `assistant-error-${Date.now()}`, + role: 'assistant', + parts: [], + error: message || copy.promptFailed, + branchGroupId: state.pendingBranchGroup ?? undefined + } + ], + busy: false, + awaitingResponse: false, + pendingBranchGroup: null, + sawAssistantPayload: true + }), + targetStoredSessionId + ) - if (isProviderSetupError(err)) { + if (targetIsCurrentView() && isProviderSetupError(err)) { requestDesktopOnboarding(copy.providerCredentialRequired) return false } - notifyError(err, copy.promptFailed) + if (targetIsCurrentView()) { + notifyError(err, copy.promptFailed) + } return false } diff --git a/apps/desktop/src/app/session/hooks/use-prompt-actions/utils.ts b/apps/desktop/src/app/session/hooks/use-prompt-actions/utils.ts index de501a52b..b1d41893c 100644 --- a/apps/desktop/src/app/session/hooks/use-prompt-actions/utils.ts +++ b/apps/desktop/src/app/session/hooks/use-prompt-actions/utils.ts @@ -226,4 +226,11 @@ export function visibleUserIndexAtOrdinal(messages: readonly ChatMessage[], targ export interface SubmitTextOptions { attachments?: ComposerAttachment[] fromQueue?: boolean + /** Runtime session id to submit into. Queue drains pass this so a + * backgrounded/source session cannot be replaced by the current foreground + * session between enqueue and drain. */ + sessionId?: string | null + /** Stable stored session id for optimistic/cache updates and stale-runtime + * recovery. Distinct from the runtime session id minted by the gateway. */ + storedSessionId?: string | null }