Merge pull request #75966 from NousResearch/bb/dither-tail-only

fix(desktop): thinking indicator can no longer appear mid-transcript

apps/desktop/src/app/session/hooks/use-message-stream/gateway-event.ts

@@ -72,6 +72,7 @@ import { ingestBackendSkin } from '@/themes/backend-sync'
 import type { RpcEvent } from '@/types/hermes'
 
 import type { ClientSessionState } from '../../../types'
+import { finalizeInterruptedMessages } from '../use-prompt-actions/rewind'
 
 import { hasSessionInfoStatePatch, sessionInfoStatePatch, SUBAGENT_EVENT_TYPES, toTodoPayload } from './utils'
 
@@ -492,6 +493,16 @@ export function useGatewayEventHandler(deps: GatewayEventDeps) {
                 ...state,
                 awaitingResponse: false,
                 busy,
+                // The turn is over but its streaming bubble may still say
+                // pending — running=false from the agent loop's finally block
+                // is the ONLY settle signal when message.complete never
+                // arrives (turn crash, reconnect gap). Left pending, that
+                // bubble shows a thinking indicator forever, stranded
+                // mid-transcript once the next user message lands after it.
+                // finalizeInterruptedMessages un-pends kept text and drops
+                // empty placeholders; on the normal path message.complete
+                // already settled everything and this is a no-op.
+                messages: finalizeInterruptedMessages(state.messages, state.streamId),
                 pendingBranchGroup: null,
                 streamId: null,
                 turnStartedAt: null

apps/desktop/src/app/session/hooks/use-message-stream/stale-pending-settle.test.tsx

@@ -0,0 +1,128 @@
+// A turn that ends WITHOUT its message.complete (turn crash, reconnect gap,
+// steer race) used to leave its streaming bubble pending:true forever. The
+// next user message then landed after it, stranding a live thinking indicator
+// mid-transcript — the dither block anywhere but the tail. session.info
+// running=false is the turn's finally-block signal and the only settle edge
+// those paths still emit, so it must finalize the bubble.
+import { QueryClient } from '@tanstack/react-query'
+import { act, cleanup, render } from '@testing-library/react'
+import { useEffect, useRef } from 'react'
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+
+import type { ClientSessionState } from '@/app/types'
+import { createClientSessionState } from '@/lib/chat-runtime'
+import type { RpcEvent } from '@/types/hermes'
+
+import { STREAM_DELTA_FLUSH_MS } from './utils'
+
+import { useMessageStream } from './index'
+
+const SID = 'stale-pending-session'
+
+let handleEvent: ((event: RpcEvent) => void) | null = null
+let states: Map<string, ClientSessionState>
+
+function Harness() {
+  const activeSessionIdRef = useRef<string | null>(SID)
+  const sessionStateByRuntimeIdRef = useRef(new Map<string, ClientSessionState>())
+  const queryClientRef = useRef(new QueryClient())
+
+  const stream = useMessageStream({
+    activeSessionIdRef,
+    hydrateFromStoredSession: vi.fn(async () => undefined),
+    queryClient: queryClientRef.current,
+    refreshHermesConfig: vi.fn(async () => undefined),
+    refreshSessions: vi.fn(async () => undefined),
+    sessionStateByRuntimeIdRef,
+    updateSessionState: (sessionId, updater) => {
+      const current = sessionStateByRuntimeIdRef.current.get(sessionId) ?? createClientSessionState()
+      const next = updater(current)
+      sessionStateByRuntimeIdRef.current.set(sessionId, next)
+
+      return next
+    }
+  })
+
+  useEffect(() => {
+    handleEvent = stream.handleGatewayEvent
+    states = sessionStateByRuntimeIdRef.current
+  }, [stream.handleGatewayEvent])
+
+  return null
+}
+
+async function mountHarness() {
+  vi.useFakeTimers()
+  render(<Harness />)
+  await act(async () => {
+    await Promise.resolve()
+  })
+}
+
+const flushDeltas = async () => {
+  await act(async () => {
+    await vi.advanceTimersByTimeAsync(STREAM_DELTA_FLUSH_MS)
+  })
+}
+
+const emit = (event: RpcEvent) => act(() => handleEvent?.(event))
+
+describe('turn end without message.complete (session.info running=false)', () => {
+  beforeEach(() => {
+    handleEvent = null
+    states = new Map()
+  })
+
+  afterEach(() => {
+    cleanup()
+    vi.useRealTimers()
+    vi.restoreAllMocks()
+  })
+
+  it('settles a streaming bubble that kept text', async () => {
+    await mountHarness()
+
+    emit({ session_id: SID, type: 'message.start', payload: {} })
+    emit({ payload: { text: 'partial answer' }, session_id: SID, type: 'message.delta' })
+    await flushDeltas()
+
+    expect(states.get(SID)?.messages.at(-1)?.pending).toBe(true)
+
+    emit({ payload: { running: false }, session_id: SID, type: 'session.info' })
+
+    const state = states.get(SID)
+    const tail = state?.messages.at(-1)
+
+    expect(tail?.role).toBe('assistant')
+    expect(tail?.pending).toBe(false)
+    expect(tail?.parts).toEqual([{ type: 'text', text: 'partial answer' }])
+    expect(state?.streamId).toBeNull()
+    expect(state?.busy).toBe(false)
+  })
+
+  it('drops an empty streaming placeholder instead of stranding it', async () => {
+    await mountHarness()
+
+    emit({ session_id: SID, type: 'message.start', payload: {} })
+    // A tool row seeds the bubble but no text ever arrives.
+    emit({
+      payload: { args: { command: 'true' }, name: 'terminal', tool_id: 't1' },
+      session_id: SID,
+      type: 'tool.start'
+    })
+    emit({
+      payload: { name: 'terminal', result: 'ok', tool_id: 't1' },
+      session_id: SID,
+      type: 'tool.complete'
+    })
+
+    emit({ payload: { running: false }, session_id: SID, type: 'session.info' })
+
+    const state = states.get(SID)
+
+    // Same math as Stop: an empty-text placeholder is dropped, nothing stays
+    // pending, and the stream binding is released.
+    expect(state?.messages.every(message => !message.pending)).toBe(true)
+    expect(state?.streamId).toBeNull()
+  })
+})

apps/desktop/src/app/session/hooks/use-prompt-actions/submit.ts

@@ -34,6 +34,7 @@ import type { ClientSessionState } from '../../../types'
 import { sessionContextDrift } from '../session-context-drift'
 import { resolveSessionProfile } from '../use-session-actions/utils'
 
+import { finalizeInterruptedMessages } from './rewind'
 import {
   _submitInFlight,
   type GatewayRequest,
@@ -347,13 +348,18 @@ export function useSubmitPrompt(deps: SubmitPromptDeps) {
           sid,
           state => ({
             ...state,
+            // A fresh user message may never land after a still-pending
+            // assistant bubble — settle any leftover (drop it when empty)
+            // before appending, or a stale spinner gets stranded
+            // mid-transcript above this message forever.
             messages: state.messages.some(m => m.id === optimisticId)
               ? state.messages
-              : [...state.messages, buildUserMessage()],
+              : [...finalizeInterruptedMessages(state.messages, state.streamId), buildUserMessage()],
             busy: true,
             awaitingResponse: true,
             pendingBranchGroup: null,
             sawAssistantPayload: false,
+            streamId: null,
             // Fresh submit = new turn — clear any leftover interrupt flag, else
             // mutateStream/completeAssistantMessage drop every delta of this turn
             // (what made drained-after-interrupt sends go silent).

apps/desktop/src/components/assistant-ui/thread/assistant-message.tsx

@@ -71,6 +71,13 @@ export const AssistantMessage: FC<{
   // ChatMessage.interim).
   const isInterim = useAuiState(s => s.message.metadata?.custom?.interim === true)
 
+  // The thinking/stall indicator belongs to the TAIL of the thread, period. A
+  // stale pending bubble mid-transcript (a turn that ended without its settle
+  // event, a steer race) must never show one — a spinner above a later user
+  // message reads as the agent answering out of order. Booleans are stable
+  // across token flushes, so this selector adds no streaming re-renders.
+  const isLastMessage = useAuiState(s => s.thread.messages[s.thread.messages.length - 1]?.id === s.message.id)
+
   // Preview targets only materialize once the turn completes — while running
   // the selector returns '' (stable), so per-token flushes skip the regex
   // scan and the re-render it would cause.
@@ -124,7 +131,7 @@ export const AssistantMessage: FC<{
       >
         {/* Todos render in the composer status stack now, not inline. */}
         <MessagePrimitive.Parts components={MESSAGE_PARTS_COMPONENTS} />
-        {isPlaceholder ? <ResponseLoadingIndicator /> : isRunning && <StreamStallIndicator />}
+        {isLastMessage && (isPlaceholder ? <ResponseLoadingIndicator /> : isRunning && <StreamStallIndicator />)}
         {previewTargets.length > 0 && (
           <div className="mt-3 flex flex-wrap gap-2">
             {previewTargets.map(target => (

apps/desktop/src/components/assistant-ui/thread/status-tail-only.test.tsx

@@ -0,0 +1,106 @@
+// The thinking indicator (dither block) may only ever render at the TAIL of
+// the thread. A message stuck status:running mid-transcript — however it got
+// there (missed settle event, steer race, upstream state bug) — must render
+// its content with no spinner: a live indicator above a later user message
+// reads as the agent answering out of order.
+import { AssistantRuntimeProvider, type ThreadMessage, useExternalStoreRuntime } from '@assistant-ui/react'
+import { cleanup, render, screen } from '@testing-library/react'
+import { afterEach, describe, expect, it, vi } from 'vitest'
+
+import { Thread } from '.'
+
+const createdAt = new Date('2026-05-01T00:00:00.000Z')
+
+class TestResizeObserver {
+  observe() {}
+  unobserve() {}
+  disconnect() {}
+}
+vi.stubGlobal('ResizeObserver', TestResizeObserver)
+vi.stubGlobal('requestAnimationFrame', (callback: FrameRequestCallback) =>
+  window.setTimeout(() => callback(performance.now()), 0)
+)
+vi.stubGlobal('cancelAnimationFrame', (id: number) => window.clearTimeout(id))
+vi.stubGlobal('CSS', { escape: (str: string) => str })
+
+Element.prototype.scrollTo = function scrollTo() {}
+// Enter animation fires for running messages; jsdom has no WAAPI.
+Element.prototype.animate = function animate() {
+  return { cancel() {}, finished: Promise.resolve() } as unknown as Animation
+}
+
+afterEach(() => {
+  cleanup()
+})
+
+const assistantMetadata = {
+  unstable_state: null,
+  unstable_annotations: [],
+  unstable_data: [],
+  steps: [],
+  custom: {}
+}
+
+function user(id: string, text: string): ThreadMessage {
+  return {
+    id,
+    role: 'user',
+    content: [{ type: 'text', text }],
+    attachments: [],
+    createdAt,
+    metadata: { custom: {} }
+  } as ThreadMessage
+}
+
+function assistant(id: string, text: string, running: boolean): ThreadMessage {
+  return {
+    id,
+    role: 'assistant',
+    content: text ? [{ type: 'text', text }] : [],
+    status: running ? { type: 'running' } : { type: 'complete', reason: 'stop' },
+    createdAt,
+    metadata: assistantMetadata
+  } as ThreadMessage
+}
+
+function Harness({ messages }: { messages: ThreadMessage[] }) {
+  const runtime = useExternalStoreRuntime<ThreadMessage>({
+    messages,
+    isRunning: messages.at(-1)?.status?.type === 'running',
+    onNew: async () => {}
+  })
+
+  return (
+    <AssistantRuntimeProvider runtime={runtime}>
+      <Thread />
+    </AssistantRuntimeProvider>
+  )
+}
+
+describe('thinking indicator is tail-only', () => {
+  it('shows the loading indicator on a running placeholder at the tail', async () => {
+    const { container } = render(<Harness messages={[user('u1', 'question'), assistant('a1', '', true)]} />)
+
+    expect(await screen.findByRole('status', { name: 'Hermes is loading a response' })).toBeTruthy()
+    expect(container.querySelector('[data-slot="aui_response-loading"]')).toBeTruthy()
+  })
+
+  it('never shows an indicator on a stale running message mid-transcript', async () => {
+    // A stranded pending bubble from an earlier turn, then a newer exchange.
+    const { container } = render(
+      <Harness
+        messages={[
+          user('u1', 'first question'),
+          assistant('a1', '', true),
+          user('u2', 'second question'),
+          assistant('a2', 'answered', false)
+        ]}
+      />
+    )
+
+    await screen.findByText('answered')
+
+    expect(container.querySelector('[data-slot="aui_response-loading"]')).toBeNull()
+    expect(container.querySelector('[data-slot="aui_stream-stall"]')).toBeNull()
+  })
+})