From 0fd059e745af666e2bb42f52d75fbd959799df53 Mon Sep 17 00:00:00 2001 From: fangliquanflq Date: Sun, 16 Aug 2026 02:36:49 +0800 Subject: [PATCH] fix(agent): preserve stalled-provider escalation --- agent/chat_completion_helpers.py | 44 ++++++++++ .../hooks/use-message-stream/gateway-event.ts | 30 ++++++- .../provider-wait-event.test.tsx | 81 +++++++++++++++++++ .../assistant-ui/thread/status.test.tsx | 14 +++- .../components/assistant-ui/thread/status.tsx | 17 ++-- apps/desktop/src/store/provider-wait.ts | 41 ++++++++++ .../test_stream_stale_circuit_breaker.py | 23 ++++++ 7 files changed, 240 insertions(+), 10 deletions(-) create mode 100644 apps/desktop/src/app/session/hooks/use-message-stream/provider-wait-event.test.tsx create mode 100644 apps/desktop/src/store/provider-wait.ts diff --git a/agent/chat_completion_helpers.py b/agent/chat_completion_helpers.py index e998f1a85d..aea9277fb9 100644 --- a/agent/chat_completion_helpers.py +++ b/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. diff --git a/apps/desktop/src/app/session/hooks/use-message-stream/gateway-event.ts b/apps/desktop/src/app/session/hooks/use-message-stream/gateway-event.ts index 50291775e6..2290e31504 100644 --- a/apps/desktop/src/app/session/hooks/use-message-stream/gateway-event.ts +++ b/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 @@ -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. diff --git a/apps/desktop/src/app/session/hooks/use-message-stream/provider-wait-event.test.tsx b/apps/desktop/src/app/session/hooks/use-message-stream/provider-wait-event.test.tsx new file mode 100644 index 0000000000..01a0d17f2f --- /dev/null +++ b/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(SID) + const sessionStateByRuntimeIdRef = useRef(new Map()) + 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() + 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({}) + } + ) +}) diff --git a/apps/desktop/src/components/assistant-ui/thread/status.test.tsx b/apps/desktop/src/components/assistant-ui/thread/status.test.tsx index 51604b4dc4..dc513d4f10 100644 --- a/apps/desktop/src/components/assistant-ui/thread/status.test.tsx +++ b/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 diff --git a/apps/desktop/src/components/assistant-ui/thread/status.tsx b/apps/desktop/src/components/assistant-ui/thread/status.tsx index d48c04e2cc..1eb53d3ea3 100644 --- a/apps/desktop/src/components/assistant-ui/thread/status.tsx +++ b/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 ( @@ -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(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 diff --git a/apps/desktop/src/store/provider-wait.ts b/apps/desktop/src/store/provider-wait.ts new file mode 100644 index 0000000000..f81e4b93e2 --- /dev/null +++ b/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>({}) + +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 : '' +} diff --git a/tests/run_agent/test_stream_stale_circuit_breaker.py b/tests/run_agent/test_stream_stale_circuit_breaker.py index 19207e1f27..af6e6c26d6 100644 --- a/tests/run_agent/test_stream_stale_circuit_breaker.py +++ b/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