fix(agent): preserve stalled-provider escalation

agent/chat_completion_helpers.py

@@ -733,6 +733,34 @@ def _reset_stale_streak(agent) -> None:
         pass
 
 
+_INTERRUPTED_WAIT_STALE_SECONDS = 30.0
+
+
+def _record_interrupted_provider_wait(
+    agent,
+    elapsed: float,
+    *,
+    response_started: bool,
+) -> bool:
+    """Count a user-aborted pre-response stall toward the stale breaker.
+
+    Interactive users commonly send a follow-up while a provider is wedged.
+    Once the same no-output interval that earns a wait notice has elapsed, that
+    interrupt is evidence of an unresponsive attempt rather than a quick user
+    cancellation. Mid-response and early interrupts remain neutral.
+    """
+    if response_started or elapsed < _INTERRUPTED_WAIT_STALE_SECONDS:
+        return False
+    _bump_stale_streak(agent)
+    logger.warning(
+        "Interrupted provider wait counted as stale after %.0fs with no output; "
+        "consecutive stale attempts=%d.",
+        elapsed,
+        _stale_streak(agent),
+    )
+    return True
+
+
 def _report_stale_nonstream_kill(
     agent,
     api_kwargs: dict,
@@ -1751,6 +1779,14 @@ def interruptible_api_call(agent, api_kwargs: dict):
             break
 
         if agent._interrupt_requested:
+            _record_interrupted_provider_wait(
+                agent,
+                _elapsed,
+                response_started=(
+                    _codex_watchdog_enabled
+                    and getattr(agent, "_codex_stream_last_event_ts", None) is not None
+                ),
+            )
             # Mark THIS request cancelled before force-closing so the worker's
             # exception handler recognizes the forced transport error as a
             # cancel and exits cleanly instead of surfacing a network error or
@@ -5125,6 +5161,14 @@ def interruptible_streaming_api_call(agent, api_kwargs: dict, *, on_first_delta=
             )
 
         if agent._interrupt_requested:
+            # The stale branch above already counted this iteration when its
+            # deadline won the race; do not double-count a simultaneous stop.
+            if _stale_elapsed <= _stream_stale_timeout:
+                _record_interrupted_provider_wait(
+                    agent,
+                    _stale_elapsed,
+                    response_started=deltas_were_sent["yes"],
+                )
             # Mark THIS request cancelled before force-closing so the worker's
             # exception handler recognizes the forced transport error as a
             # cancel and exits without retrying or surfacing a network error.

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

@@ -53,6 +53,7 @@ import {
   setSecretRequest,
   setSudoRequest
 } from '@/store/prompts'
+import { providerWaitText, setSessionProviderWait } from '@/store/provider-wait'
 import { recordAgentReaction } from '@/store/reactions-local'
 import {
   $currentCwd,
@@ -213,6 +214,20 @@ const COMPACTION_RESUME_EVENT_TYPES = new Set([
   'tool.complete'
 ])
 
+const PROVIDER_WAIT_SUPERSEDING_EVENT_TYPES = new Set([
+  'error',
+  'message.complete',
+  'message.delta',
+  'message.interim',
+  'message.start',
+  'reasoning.available',
+  'reasoning.delta',
+  'tool.complete',
+  'tool.generating',
+  'tool.progress',
+  'tool.start'
+])
+
 interface GatewayEventDeps {
   activeGatewayProfile: string
   activeSessionIdRef: MutableRefObject<string | null>
@@ -387,6 +402,10 @@ export function useGatewayEventHandler(deps: GatewayEventDeps) {
         setSessionDraftingTool(sessionId, '')
       }
 
+      if (sessionId && PROVIDER_WAIT_SUPERSEDING_EVENT_TYPES.has(event.type)) {
+        setSessionProviderWait(sessionId, '')
+      }
+
       if (event.type === 'gateway.ready') {
         // Seed the active skin into the desktop theme registry without applying,
         // so a fresh connect never overrides the user's persisted desktop theme.
@@ -807,10 +826,13 @@ export function useGatewayEventHandler(deps: GatewayEventDeps) {
           }
         }
       } else if (event.type === 'thinking.delta') {
-        // thinking.delta carries the kawaii spinner status (face + verb from
-        // KawaiiSpinner), not real reasoning. The bottom-of-thread loading
-        // indicator already covers that UX, so we ignore these events to
-        // avoid a duplicative "Thinking" disclosure showing spinner text.
+        // Most thinking.delta frames are kawaii spinner rewrites and stay out
+        // of the transcript. Explained provider waits are different: the core
+        // emits them after prolonged silence, so name that wait in the existing
+        // bottom-of-thread status row instead of leaving only an unlabeled timer.
+        if (sessionId) {
+          setSessionProviderWait(sessionId, providerWaitText(coerceGatewayText(payload?.text)))
+        }
       } else if (event.type === 'reaction') {
         // Core-detected affection (ily / <3 / good bot) on the user's message.
         // Play hearts only for the visible session so background turns stay quiet.

apps/desktop/src/app/session/hooks/use-message-stream/provider-wait-event.test.tsx

@@ -0,0 +1,81 @@
+import { QueryClient } from '@tanstack/react-query'
+import { act, cleanup, render, waitFor } 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 { $providerWaitSessions } from '@/store/provider-wait'
+import type { RpcEvent } from '@/types/hermes'
+
+import { useMessageStream } from './index'
+
+const SID = 'session-1'
+let handleEvent: ((event: RpcEvent) => void) | null = null
+
+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
+  }, [stream.handleGatewayEvent])
+
+  return null
+}
+
+function emit(type: RpcEvent['type'], payload: RpcEvent['payload'] = {}) {
+  act(() => handleEvent!({ payload, session_id: SID, type }))
+}
+
+describe('provider wait visibility', () => {
+  beforeEach(async () => {
+    handleEvent = null
+    $providerWaitSessions.set({})
+    render(<Harness />)
+    await waitFor(() => expect(handleEvent).not.toBeNull())
+  })
+
+  afterEach(() => {
+    cleanup()
+    $providerWaitSessions.set({})
+    vi.restoreAllMocks()
+  })
+
+  it('surfaces explained waits but ignores generic spinner rewrites', () => {
+    emit('thinking.delta', { text: '⏳ waiting on local-model — 30s with no output yet' })
+    expect($providerWaitSessions.get()).toEqual({
+      [SID]: '⏳ waiting on local-model — 30s with no output yet'
+    })
+
+    emit('thinking.delta', { text: '◉_◉ cogitating...' })
+    expect($providerWaitSessions.get()).toEqual({})
+  })
+
+  it.each(['message.delta', 'reasoning.delta', 'tool.start', 'message.complete', 'error'] as const)(
+    'clears the wait when %s proves the turn progressed or ended',
+    type => {
+      emit('thinking.delta', { text: '⚠ no output from provider for 900s — reconnecting...' })
+      emit(type, type === 'tool.start' ? { name: 'terminal', tool_id: 'tool-1' } : { text: 'progress' })
+
+      expect($providerWaitSessions.get()).toEqual({})
+    }
+  )
+})

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

@@ -3,6 +3,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
 
 import { __resetElapsedTimerRegistryForTests } from '@/components/chat/activity-timer'
 import { I18nProvider } from '@/i18n'
+import { $providerWaitSessions, setSessionProviderWait } from '@/store/provider-wait'
 import { $activeSessionId, $turnStartedAt } from '@/store/session'
 
 import { ResponseLoadingIndicator } from './status'
@@ -22,7 +23,7 @@ describe('ResponseLoadingIndicator timer', () => {
     // useViewedInterval gates ticking on document focus + visibility; jsdom's
     // hasFocus() is unreliable across runners, so pin it (same as the
     // background-sync backstop tests).
-    vi.spyOn(document, 'hasFocus').mockReturnValue(true)
+    vi.spyOn(globalThis.document, 'hasFocus').mockReturnValue(true)
     __resetElapsedTimerRegistryForTests()
   })
 
@@ -30,6 +31,7 @@ describe('ResponseLoadingIndicator timer', () => {
     cleanup()
     $activeSessionId.set(null)
     $turnStartedAt.set(null)
+    $providerWaitSessions.set({})
     __resetElapsedTimerRegistryForTests()
     vi.restoreAllMocks()
     vi.useRealTimers()
@@ -58,6 +60,16 @@ describe('ResponseLoadingIndicator timer', () => {
 
     expect(screen.getAllByText((_, node) => node?.textContent === '8s').length).toBeGreaterThan(0)
   })
+
+  it('names a prolonged provider wait in the existing response status row', () => {
+    $activeSessionId.set('session-a')
+    $turnStartedAt.set(Date.now())
+    setSessionProviderWait('session-a', '⏳ waiting on local-model — 30s with no output yet')
+
+    renderIndicator()
+
+    expect(screen.getByText('⏳ waiting on local-model — 30s with no output yet')).toBeTruthy()
+  })
 })
 
 // The status line sits between tool rows and thinking headers, which the

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

@@ -15,6 +15,7 @@ import { cn } from '@/lib/utils'
 import { $backgroundResume } from '@/store/background-delegation'
 import { sessionCompacting } from '@/store/compaction'
 import { sessionAwaitingInput } from '@/store/prompts'
+import { sessionProviderWait } from '@/store/provider-wait'
 import { $turnStartedAt } from '@/store/session'
 import { type DraftingTool, sessionDraftingTool } from '@/store/tool-drafting'
 
@@ -58,6 +59,7 @@ function useThreadSessionStatus() {
   const turnStartedAt = useStore($turnStartedAt)
   const compacting = useStore(useMemo(() => sessionCompacting(sessionId), [sessionId]))
   const drafting = useStore(useMemo(() => sessionDraftingTool(sessionId), [sessionId]))
+  const providerWait = useStore(useMemo(() => sessionProviderWait(sessionId), [sessionId]))
   // A pending clarify / approval / sudo / secret means the turn is paused on the
   // user, not working — so don't resurrect the "thinking" timer while they
   // decide (matches the pet's awaitingInput pose taking priority over busy).
@@ -67,6 +69,7 @@ function useThreadSessionStatus() {
     awaitingInput,
     compacting,
     drafting,
+    providerWait,
     turnTimerKey: sessionId && turnStartedAt ? `turn:${sessionId}:${turnStartedAt}` : undefined
   }
 }
@@ -79,7 +82,7 @@ const DRAFTING_REVEAL_MS = 200
  * What to call the wait, if it deserves a name. Compaction outranks a draft —
  * it's rarer, slower, and explains a transcript that looks like it reset.
  */
-function useStatusHint(compacting: boolean, drafting: DraftingTool | null): string {
+function useStatusHint(compacting: boolean, drafting: DraftingTool | null, providerWait: string): string {
   const [revealed, setRevealed] = useState(false)
   const name = drafting?.name ?? ''
 
@@ -99,6 +102,10 @@ function useStatusHint(compacting: boolean, drafting: DraftingTool | null): stri
     return COMPACTION_LABEL
   }
 
+  if (providerWait) {
+    return providerWait
+  }
+
   return revealed && name ? toolPresentVerb(name) : ''
 }
 
@@ -125,9 +132,9 @@ export const CenteredThreadSpinner: FC = () => {
 
 export const ResponseLoadingIndicator: FC = () => {
   const { t } = useI18n()
-  const { compacting, drafting, turnTimerKey } = useThreadSessionStatus()
+  const { compacting, drafting, providerWait, turnTimerKey } = useThreadSessionStatus()
   const elapsed = useElapsedSeconds(true, turnTimerKey)
-  const hint = useStatusHint(compacting, drafting)
+  const hint = useStatusHint(compacting, drafting, providerWait)
 
   return (
     <StatusRow data-slot="aui_response-loading" label={hint || t.assistant.thread.loadingResponse}>
@@ -205,8 +212,8 @@ export const StreamStallIndicator: FC = () => {
   // what lets the timer read "quiet for 12s" rather than the age of this
   // component, which is the whole turn so far.
   const [quietSince, setQuietSince] = useState<number | undefined>(undefined)
-  const { awaitingInput, compacting, drafting, turnTimerKey } = useThreadSessionStatus()
-  const hint = useStatusHint(compacting, drafting)
+  const { awaitingInput, compacting, drafting, providerWait, turnTimerKey } = useThreadSessionStatus()
+  const hint = useStatusHint(compacting, drafting, providerWait)
 
   // A tool run at the tail already narrates the wait — its summary counts the
   // calls, its ticker names the current one, and it carries its own timer. A

apps/desktop/src/store/provider-wait.ts

@@ -0,0 +1,41 @@
+import { atom, computed } from 'nanostores'
+
+const keyFor = (sessionId: string | null | undefined): string => sessionId ?? ''
+
+export const $providerWaitSessions = atom<Record<string, string>>({})
+
+export function sessionProviderWait(sessionId: null | string) {
+  return computed($providerWaitSessions, sessions => sessions[keyFor(sessionId)] ?? '')
+}
+
+export function setSessionProviderWait(sessionId: string | null | undefined, text: string): void {
+  const key = keyFor(sessionId)
+  const sessions = $providerWaitSessions.get()
+  const nextText = text.trim()
+
+  if (!nextText) {
+    if (!(key in sessions)) {
+      return
+    }
+
+    const next = { ...sessions }
+    delete next[key]
+    $providerWaitSessions.set(next)
+
+    return
+  }
+
+  if (sessions[key] === nextText) {
+    return
+  }
+
+  $providerWaitSessions.set({ ...sessions, [key]: nextText })
+}
+
+/** Only the core's explained wait/reconnect frames belong in Desktop's status
+ * row. Generic kawaii spinner rewrites remain presentation noise. */
+export function providerWaitText(text: string): string {
+  const value = text.trim()
+
+  return /^(?:⏳|⚠|↻)\s*(?:waiting on|no (?:output|response)|model returned)/i.test(value) ? value : ''
+}

tests/run_agent/test_stream_stale_circuit_breaker.py

@@ -61,6 +61,29 @@ def _good_stream_cm():
 
 
 class TestStreamStaleCircuitBreaker:
+    def test_interrupted_pre_response_wait_advances_streak(self, monkeypatch):
+        """Qualified pre-response interrupts advance the breaker, while early
+        and mid-stream user cancellations remain neutral."""
+        from agent.chat_completion_helpers import (
+            _check_stale_giveup,
+            _record_interrupted_provider_wait,
+        )
+
+        monkeypatch.setenv("HERMES_STREAM_STALE_GIVEUP", "2")
+        agent = _make_anthropic_agent()
+        agent._consecutive_stale_streams = 0
+
+        assert _record_interrupted_provider_wait(agent, 29.9, response_started=False) is False
+        assert _record_interrupted_provider_wait(agent, 45.0, response_started=True) is False
+        assert agent._consecutive_stale_streams == 0
+
+        assert _record_interrupted_provider_wait(agent, 45.0, response_started=False) is True
+        assert agent._consecutive_stale_streams == 1
+
+        assert _record_interrupted_provider_wait(agent, 60.0, response_started=False) is True
+        with pytest.raises(RuntimeError, match="2 consecutive stale attempts"):
+            _check_stale_giveup(agent)
+
     @pytest.mark.filterwarnings("ignore::pytest.PytestUnhandledThreadExceptionWarning")
     def test_short_circuits_when_streak_at_threshold(self, monkeypatch):
         """A session already past the consecutive-stale threshold must abort