Merge pull request #45414 from NousResearch/bb/fix-desktop-queue-drain-strand

fix(desktop): stop stranding queued prompts across backend bounces

apps/desktop/src/store/composer-queue.test.ts

@@ -7,9 +7,10 @@ import {
   dequeueQueuedPrompt,
   enqueueQueuedPrompt,
   getQueuedPrompts,
+  migrateQueuedPrompts,
   promoteQueuedPrompt,
   removeQueuedPrompt,
-  shouldAutoDrainOnSettle,
+  shouldAutoDrain,
   updateQueuedPrompt,
   updateQueuedPromptText
 } from './composer-queue'
@@ -117,32 +118,53 @@ describe('composer queue store', () => {
   })
 })
 
-describe('shouldAutoDrainOnSettle', () => {
-  const base = { isBusy: false, queueLength: 1, wasBusy: true }
-
-  it('drains the next queued prompt when a turn settles', () => {
-    expect(shouldAutoDrainOnSettle(base)).toBe(true)
+describe('migrateQueuedPrompts', () => {
+  beforeEach(() => {
+    window.localStorage.removeItem(QUEUE_STORAGE_KEY)
+    $queuedPromptsBySession.set({})
   })
 
-  it('drains after an interrupt — the settle edge is the same', () => {
-    // Interrupting to reach a queued message is the point of the queue; the
-    // gateway emits the same settle whether the turn finished or was stopped.
-    expect(shouldAutoDrainOnSettle(base)).toBe(true)
+  it('moves entries from a dead runtime key onto the live one', () => {
+    enqueueQueuedPrompt('rt-old', { attachments: [], text: 'stranded' })
+
+    expect(migrateQueuedPrompts('rt-old', 'rt-new')).toBe(true)
+    expect(getQueuedPrompts('rt-old')).toEqual([])
+    expect(getQueuedPrompts('rt-new').map(e => e.text)).toEqual(['stranded'])
+    // The dead key is dropped from the store entirely.
+    expect($queuedPromptsBySession.get()['rt-old']).toBeUndefined()
   })
 
-  it('does not drain when the queue is empty', () => {
-    expect(shouldAutoDrainOnSettle({ ...base, queueLength: 0 })).toBe(false)
+  it('appends after existing target entries (FIFO preserved)', () => {
+    enqueueQueuedPrompt('rt-new', { attachments: [], text: 'already here' })
+    enqueueQueuedPrompt('rt-old', { attachments: [], text: 'migrated' })
+
+    migrateQueuedPrompts('rt-old', 'rt-new')
+
+    expect(getQueuedPrompts('rt-new').map(e => e.text)).toEqual(['already here', 'migrated'])
   })
 
-  it('ignores steady busy state (no true → false transition)', () => {
-    expect(shouldAutoDrainOnSettle({ ...base, isBusy: true })).toBe(false)
+  it('is a no-op when source is empty or keys match', () => {
+    expect(migrateQueuedPrompts('rt-old', 'rt-new')).toBe(false)
+    expect(migrateQueuedPrompts('rt-x', 'rt-x')).toBe(false)
   })
+})
 
-  it('ignores busy entry (false → true, not a settle)', () => {
-    expect(shouldAutoDrainOnSettle({ ...base, isBusy: true, wasBusy: false })).toBe(false)
+describe('shouldAutoDrain', () => {
+  it('drains whenever idle with a non-empty queue', () => {
+    expect(shouldAutoDrain({ isBusy: false, queueLength: 1 })).toBe(true)
   })
 
-  it('ignores steady idle state (was not busy)', () => {
-    expect(shouldAutoDrainOnSettle({ ...base, wasBusy: false })).toBe(false)
+  it('drains on mount/reconnect with no observed busy edge', () => {
+    // The whole point of dropping the edge: a remount resets the busy ref, so an
+    // edge-gated drain would strand the entry. Idle + non-empty must still fire.
+    expect(shouldAutoDrain({ isBusy: false, queueLength: 2 })).toBe(true)
+  })
+
+  it('does not drain mid-turn', () => {
+    expect(shouldAutoDrain({ isBusy: true, queueLength: 1 })).toBe(false)
+  })
+
+  it('does not drain an empty queue', () => {
+    expect(shouldAutoDrain({ isBusy: false, queueLength: 0 })).toBe(false)
   })
 })

apps/desktop/src/store/composer-queue.ts

@@ -209,31 +209,58 @@ export const clearQueuedPrompts = (key: string | null | undefined) => {
   writeSession(sid, [])
 }
 
-/** Inputs to {@link shouldAutoDrainOnSettle}, captured at a `busy` transition. */
-export interface AutoDrainSettleInput {
-  wasBusy: boolean
+/**
+ * Move pending entries from a dead session key onto a live one, preserving FIFO
+ * (existing target entries first, migrated entries appended). A backend bounce /
+ * resume can mint a fresh runtime session id for the *same* conversation; the
+ * entries enqueued under the old id would otherwise be stranded under a key
+ * nothing reads anymore. No-op unless both keys resolve and differ.
+ */
+export const migrateQueuedPrompts = (
+  fromKey: string | null | undefined,
+  toKey: string | null | undefined
+): boolean => {
+  const from = sidOf(fromKey)
+  const to = sidOf(toKey)
+
+  if (!from || !to || from === to) {
+    return false
+  }
+
+  const pending = queueFor(from)
+
+  if (pending.length === 0) {
+    return false
+  }
+
+  const next = { ...$queuedPromptsBySession.get() }
+  delete next[from]
+  next[to] = [...queueFor(to), ...pending]
+
+  $queuedPromptsBySession.set(next)
+  save(next)
+
+  return true
+}
+
+/** Inputs to {@link shouldAutoDrain}. */
+export interface AutoDrainInput {
   isBusy: boolean
   queueLength: number
 }
 
 /**
- * Decide whether the composer should auto-drain the next queued prompt when a
- * turn settles (busy transitions true → false).
+ * Decide whether the composer should auto-drain the next queued prompt.
  *
- * Queued turns always advance once the session is idle again, whether the turn
- * finished naturally or the user interrupted it. Interrupting to reach a queued
- * message is the whole point of the queue, so we never suppress the drain. The
- * gateway guarantees a settle (message.complete + session.info running:false)
- * even after an interrupt, so this single edge reliably advances the queue. To
- * cancel queued turns the user deletes them from the panel.
+ * Edge-independent on purpose: the queue must advance whenever the session is
+ * idle and has pending entries, NOT only on an observed busy true → false edge.
+ * A backend bounce / websocket reconnect remounts the composer and resets the
+ * busy ref to the current value, swallowing the settle edge — an edge-gated
+ * drain would then strand the entry forever. The caller's drain lock
+ * (`drainingQueueRef`) serializes sends so being edge-free can't double-submit.
  */
-export const shouldAutoDrainOnSettle = (params: AutoDrainSettleInput): boolean => {
-  const { isBusy, queueLength, wasBusy } = params
+export const shouldAutoDrain = ({ isBusy, queueLength }: AutoDrainInput): boolean => !isBusy && queueLength > 0
 
-  // Only react to a true → false transition; ignore steady state and entry.
-  if (isBusy || !wasBusy) {
-    return false
-  }
-
-  return queueLength > 0
-}
+/** Auto-drain attempts for one entry before we stop retrying and toast. The
+ * entry stays queued for a manual send; a remount/reconnect resets the count. */
+export const MAX_AUTO_DRAIN_ATTEMPTS = 4