fix(desktop): drain queued prompts for background sessions (#66001)
Co-authored-by: Jakub Wolniewicz <4850809+frizikk@users.noreply.github.com>
This commit is contained in:
brooklyn!
GitHub
parent
bd00212337
commit
432fca55a7
10 changed files with 573 additions and 86 deletions
|
|
@ -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(
|
||||
|
|
|
|||
|
|
@ -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> | boolean
|
||||
onSubmit: (
|
||||
value: string,
|
||||
options?: { attachments?: ComposerAttachment[]; fromQueue?: boolean }
|
||||
) => Promise<boolean> | boolean
|
||||
onSubmit: (value: string, options?: SubmitTextOptions) => Promise<boolean> | boolean
|
||||
onTranscribeAudio?: (audio: Blob) => Promise<string>
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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<React.ComponentProps<'div'>, 'onSubmit'> {
|
|||
onPickImages: () => void
|
||||
onRemoveAttachment: (id: string) => void
|
||||
onSteer: (text: string) => Promise<boolean> | boolean
|
||||
onSubmit: (
|
||||
text: string,
|
||||
options?: { attachments?: ComposerAttachment[]; fromQueue?: boolean }
|
||||
) => Promise<boolean> | boolean
|
||||
onSubmit: (text: string, options?: SubmitTextOptions) => Promise<boolean> | boolean
|
||||
onThreadMessagesChange: (messages: readonly ThreadMessage[]) => void
|
||||
onEdit: (message: AppendMessage) => Promise<void>
|
||||
onReload: (parentId: string | null) => Promise<void>
|
||||
|
|
|
|||
|
|
@ -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({
|
||||
|
|
|
|||
|
|
@ -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<Map<string, string>>
|
||||
selectedStoredSessionId?: string | null
|
||||
submitText: (text: string, options?: SubmitTextOptions) => Promise<boolean> | 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(<Harness runtimeMap={runtimeMap} submitText={submitText} />)
|
||||
|
||||
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(<Harness runtimeMap={runtimeMap} selectedStoredSessionId="stored-session-a" submitText={submitText} />)
|
||||
|
||||
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(<Harness runtimeMap={runtimeMap} submitText={submitText} />)
|
||||
|
||||
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<string, string>() }
|
||||
const submitText = vi.fn(async () => true)
|
||||
|
||||
enqueueQueuedPrompt('stored-session-a', { text: 'resume then send', attachments: [] })
|
||||
|
||||
render(<Harness runtimeMap={runtimeMap} submitText={submitText} />)
|
||||
|
||||
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(<Harness runtimeMap={runtimeMap} submitText={submitText} />)
|
||||
|
||||
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)
|
||||
})
|
||||
})
|
||||
174
apps/desktop/src/app/session/hooks/use-background-queue-drain.ts
Normal file
174
apps/desktop/src/app/session/hooks/use-background-queue-drain.ts
Normal file
|
|
@ -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> | boolean
|
||||
|
||||
interface BackgroundQueueDrainOptions {
|
||||
enabled: boolean
|
||||
runtimeIdByStoredSessionIdRef: MutableRefObject<Map<string, string>>
|
||||
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<string>())
|
||||
const drainFailuresRef = useRef(new Map<string, number>())
|
||||
const retryTimersRef = useRef<number[]>([])
|
||||
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])
|
||||
}
|
||||
|
|
@ -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<string | null>
|
||||
cancelRun: () => Promise<void>
|
||||
restoreToMessage: (messageId: string, target?: { text?: string; userOrdinal?: number | null }) => Promise<void>
|
||||
steerPrompt: (text: string) => Promise<boolean>
|
||||
submitText: (text: string, options?: { attachments?: ComposerAttachment[]; fromQueue?: boolean }) => Promise<boolean>
|
||||
submitText: (text: string, options?: SubmitTextOptions) => Promise<boolean>
|
||||
}
|
||||
|
||||
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<string, unknown>
|
||||
) => void
|
||||
onReady: (handle: HarnessHandle) => void
|
||||
onSeedState?: (state: Record<string, unknown>) => void
|
||||
openMemoryGraph?: () => void
|
||||
|
|
@ -97,9 +106,8 @@ function Harness({
|
|||
activeSessionId?: null | string
|
||||
createBackendSessionForSend?: () => Promise<null | string>
|
||||
}) {
|
||||
const activeSessionIdRef: MutableRefObject<string | null> = activeSessionIdRefProp ?? {
|
||||
current: activeSessionId === undefined ? RUNTIME_SESSION_ID : activeSessionId
|
||||
}
|
||||
const localActiveSessionIdRef = useRef<string | null>(activeSessionId === undefined ? RUNTIME_SESSION_ID : activeSessionId)
|
||||
const activeSessionIdRef = activeSessionIdRefProp ?? localActiveSessionIdRef
|
||||
|
||||
const selectedStoredSessionIdRef: MutableRefObject<string | null> = 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<string, unknown>
|
||||
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<typeof actions.cancelRun>) =>
|
||||
act(async () => actions.cancelRun(...args)) as Promise<void>,
|
||||
restoreToMessage: (...args: Parameters<typeof actions.restoreToMessage>) =>
|
||||
|
|
@ -152,7 +162,7 @@ function Harness({
|
|||
submitText: (...args: Parameters<typeof actions.submitText>) =>
|
||||
act(async () => actions.submitText(...args)) as Promise<boolean>
|
||||
})
|
||||
}, [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<string, unknown>; storedSessionId: null | string | undefined }[] = []
|
||||
const requestGateway = vi.fn(async () => ({}) as never)
|
||||
|
||||
let handle: HarnessHandle | null = null
|
||||
render(
|
||||
<Harness
|
||||
onReady={h => (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<string, unknown> }[] = []
|
||||
let submitAttempts = 0
|
||||
|
||||
const requestGateway = vi.fn(async (method: string, params?: Record<string, unknown>) => {
|
||||
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(
|
||||
<Harness
|
||||
onReady={h => (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<string, unknown> }[] = []
|
||||
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'
|
||||
})
|
||||
})
|
||||
|
||||
|
|
|
|||
|
|
@ -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> | void
|
||||
startFreshSessionDraft: () => void
|
||||
submitPromptText: (
|
||||
rawText: string,
|
||||
options?: { attachments?: ComposerAttachment[]; fromQueue?: boolean }
|
||||
) => Promise<boolean>
|
||||
submitPromptText: (rawText: string, options?: SubmitTextOptions) => Promise<boolean>
|
||||
}
|
||||
|
||||
/** The /slash command dispatcher, extracted from usePromptActions. */
|
||||
|
|
|
|||
|
|
@ -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
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue