From 67660ab4e5fb7ade5060b912576c047fea64ab7b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=A8=8B=E5=BA=8F=E5=91=98=E9=98=BF=E6=B1=9F=28Relakkes?= =?UTF-8?q?=29?= Date: Fri, 12 Jun 2026 18:32:28 +0800 Subject: [PATCH] fix(desktop): surface non-streaming fallback in chat status (#766) api_retry heartbeats already reach the desktop status bar, but the streaming-to-non-streaming fallback had no signal at all: the CLI only flipped an internal flag, and the one-shot fallback response can take minutes with zero incremental output, so the UI showed a bare spinner the whole time. - CLI: yield a {type:'system', subtype:'streaming_fallback', cause} message at both fallback sites (stream error/watchdog and 404 stream creation), mirroring the existing api_error -> api_retry path through query.ts passthrough, QueryEngine SDK output, and the SDK schema. - Server: translate it to a streaming_fallback ServerMessage; unrecognized causes normalize to 'unknown' instead of dropping the event. - Desktop: track it as active-turn state (cleared at the same 12 sites as apiRetry; a fallback notice supersedes the stale retry banner) and render a neutral pill with the turn timer - expected state, not an error, so no amber styling and nothing in the diagnostics panel. - Retry banner now shows "retrying now" once the countdown elapses instead of sticking at "waiting 0s". With 62241a31 disabling the non-streaming fallback for desktop CLI sessions, this notice mainly covers the 404 gateway path (which skips the disable check), callers that re-enable fallback via env, and non-desktop SDK consumers. Constraint: Retries and fallbacks are expected states per the diagnostics severity standard - lightweight active-turn UI only, no error-panel entries, no transcript persistence. Tested: bun test src/server/__tests__/ws-memory-events.test.ts Tested: bun run check:server Tested: cd desktop && bun run test -- --run Tested: cd desktop && bun run lint Not-tested: live provider outage reproduction; verified via unit coverage of the translation, store lifecycle, and indicator rendering. --- .../src/components/chat/MessageList.test.tsx | 19 ++++++ .../chat/StreamingIndicator.test.tsx | 67 +++++++++++++++++++ .../components/chat/StreamingIndicator.tsx | 31 ++++++++- desktop/src/i18n/locales/en.ts | 3 + desktop/src/i18n/locales/jp.ts | 3 + desktop/src/i18n/locales/kr.ts | 3 + desktop/src/i18n/locales/zh-TW.ts | 3 + desktop/src/i18n/locales/zh.ts | 3 + desktop/src/stores/chatStore.test.ts | 66 ++++++++++++++++++ desktop/src/stores/chatStore.ts | 34 +++++++++- desktop/src/types/chat.ts | 10 +++ src/QueryEngine.ts | 9 +++ src/entrypoints/sdk/coreSchemas.ts | 15 +++++ src/server/__tests__/ws-memory-events.test.ts | 39 +++++++++++ src/server/ws/events.ts | 7 ++ src/server/ws/handler.ts | 20 +++++- src/services/api/claude.ts | 14 +++- src/utils/messages.ts | 26 +++++++ 18 files changed, 367 insertions(+), 5 deletions(-) diff --git a/desktop/src/components/chat/MessageList.test.tsx b/desktop/src/components/chat/MessageList.test.tsx index 928c838c..e8e6db3d 100644 --- a/desktop/src/components/chat/MessageList.test.tsx +++ b/desktop/src/components/chat/MessageList.test.tsx @@ -896,6 +896,25 @@ describe('MessageList nested tool calls', () => { expect(screen.getByText(/waiting \d+s/)).toBeTruthy() }) + it('shows the non-streaming fallback notice in the active turn indicator', () => { + useChatStore.setState({ + sessions: { + [ACTIVE_TAB]: makeSessionState({ + chatState: 'thinking', + streamingFallback: { + cause: 'watchdog', + receivedAt: Date.now(), + }, + }), + }, + }) + + render() + + expect(screen.getByTestId('streaming-fallback-indicator')).toBeTruthy() + expect(screen.getByText(/switched to non-streaming mode/)).toBeTruthy() + }) + it('renders compact completion as an expandable timeline divider', () => { useChatStore.setState({ sessions: { diff --git a/desktop/src/components/chat/StreamingIndicator.test.tsx b/desktop/src/components/chat/StreamingIndicator.test.tsx index ba32c732..5ce3edbd 100644 --- a/desktop/src/components/chat/StreamingIndicator.test.tsx +++ b/desktop/src/components/chat/StreamingIndicator.test.tsx @@ -91,4 +91,71 @@ describe('StreamingIndicator', () => { expect(screen.queryByText(/tokens/)).toBeNull() }) + + it('shows the non-streaming fallback notice while waiting for the one-shot response', () => { + useChatStore.setState({ + sessions: { + [ACTIVE_TAB]: makeSession({ + chatState: 'thinking', + streamingFallback: { cause: 'watchdog', receivedAt: Date.now() }, + elapsedSeconds: 95, + }), + }, + }) + + render() + + const notice = screen.getByTestId('streaming-fallback-indicator') + expect(notice.textContent).toContain('switched to non-streaming mode') + // 回合计时保留，证明"还在跑"而不是卡死。 + expect(notice.textContent).toContain('1m 35s') + }) + + it('prefers the retry banner over the fallback notice when both are active', () => { + useChatStore.setState({ + sessions: { + [ACTIVE_TAB]: makeSession({ + chatState: 'thinking', + apiRetry: { + attempt: 2, + maxRetries: 10, + retryDelayMs: 60_000, + errorStatus: 529, + receivedAt: Date.now(), + }, + streamingFallback: { cause: 'stream_error', receivedAt: Date.now() }, + }), + }, + }) + + render() + + // api_retry 携带更具体的进度（第 N 次/倒计时），优先级高于泛化的降级提示。 + expect(screen.getByTestId('api-retry-indicator')).toBeTruthy() + expect(screen.queryByTestId('streaming-fallback-indicator')).toBeNull() + }) + + it('switches the retry banner from a countdown to "retrying now" once the delay elapses', () => { + useChatStore.setState({ + sessions: { + [ACTIVE_TAB]: makeSession({ + chatState: 'thinking', + apiRetry: { + attempt: 3, + maxRetries: 10, + retryDelayMs: 1000, + errorStatus: null, + // 倒计时早已结束：请求已在途，不该再显示"waiting 0s"。 + receivedAt: Date.now() - 10_000, + }, + }), + }, + }) + + render() + + const banner = screen.getByTestId('api-retry-indicator') + expect(banner.textContent).toContain('retrying now') + expect(banner.textContent).not.toContain('waiting') + }) }) diff --git a/desktop/src/components/chat/StreamingIndicator.tsx b/desktop/src/components/chat/StreamingIndicator.tsx index 6d885647..efd9370f 100644 --- a/desktop/src/components/chat/StreamingIndicator.tsx +++ b/desktop/src/components/chat/StreamingIndicator.tsx @@ -41,6 +41,7 @@ export function StreamingIndicator() { const chatState = sessionState?.chatState ?? 'idle' const statusVerb = sessionState?.statusVerb ?? '' const apiRetry = sessionState?.apiRetry ?? null + const streamingFallback = sessionState?.streamingFallback ?? null const elapsedSeconds = sessionState?.elapsedSeconds ?? 0 // chars ÷ 4 estimates output tokens for this turn, mirroring the CLI spinner. const streamingTokens = Math.round((sessionState?.streamingResponseChars ?? 0) / 4) @@ -75,7 +76,9 @@ export function StreamingIndicator() { {statusText} - {t('chat.retry.waiting', { seconds: formatRetrySeconds(remainingMs) })} + {remainingMs > 0 + ? t('chat.retry.waiting', { seconds: formatRetrySeconds(remainingMs) }) + : t('chat.retry.retrying')} {detailText && ( @@ -86,6 +89,32 @@ export function StreamingIndicator() { ) } + if (streamingFallback) { + // 预期内的降级等待（非错误）：非流式响应一次性返回，期间无增量输出。 + // 用中性样式的轻提示 + 回合计时，与 api_retry 的警示横幅区分开。 + return ( +
+
+ ) + } + let verb: string if (statusVerb) { verb = translateServerVerb(t, statusVerb) diff --git a/desktop/src/i18n/locales/en.ts b/desktop/src/i18n/locales/en.ts index 528681f9..9574589b 100644 --- a/desktop/src/i18n/locales/en.ts +++ b/desktop/src/i18n/locales/en.ts @@ -1862,6 +1862,9 @@ export const en = { 'chat.retry.httpStatus': 'HTTP {status}', 'chat.retry.networkError': 'network error', 'chat.retry.waiting': 'waiting {seconds}s', + 'chat.retry.retrying': 'retrying now…', + 'chat.fallback.title': 'Network hiccup — switched to non-streaming mode', + 'chat.fallback.detail': 'response arrives in one piece, this can take a while', // ─── Tabs ────────────────────────────────────── 'tabs.close': 'Close', diff --git a/desktop/src/i18n/locales/jp.ts b/desktop/src/i18n/locales/jp.ts index a4507e7f..90cf61d2 100644 --- a/desktop/src/i18n/locales/jp.ts +++ b/desktop/src/i18n/locales/jp.ts @@ -1864,6 +1864,9 @@ export const jp: Record = { 'chat.retry.httpStatus': 'HTTP {status}', 'chat.retry.networkError': 'ネットワークエラー', 'chat.retry.waiting': '{seconds} 秒待機中', + 'chat.retry.retrying': '再試行しています…', + 'chat.fallback.title': 'ネットワーク不調のため非ストリーミングに切替', + 'chat.fallback.detail': '応答は一括で届くため時間がかかることがあります', // ─── Tabs ────────────────────────────────────── 'tabs.close': '閉じる', diff --git a/desktop/src/i18n/locales/kr.ts b/desktop/src/i18n/locales/kr.ts index 4fd1db08..62f6e0d7 100644 --- a/desktop/src/i18n/locales/kr.ts +++ b/desktop/src/i18n/locales/kr.ts @@ -1864,6 +1864,9 @@ export const kr: Record = { 'chat.retry.httpStatus': 'HTTP {status}', 'chat.retry.networkError': '네트워크 오류', 'chat.retry.waiting': '{seconds}초 대기 중', + 'chat.retry.retrying': '지금 다시 시도 중…', + 'chat.fallback.title': '네트워크 불안정으로 비스트리밍 모드로 전환됨', + 'chat.fallback.detail': '응답이 한 번에 도착하므로 시간이 걸릴 수 있습니다', // ─── Tabs ────────────────────────────────────── 'tabs.close': '닫기', diff --git a/desktop/src/i18n/locales/zh-TW.ts b/desktop/src/i18n/locales/zh-TW.ts index 3c39aa7c..54bbcc0d 100644 --- a/desktop/src/i18n/locales/zh-TW.ts +++ b/desktop/src/i18n/locales/zh-TW.ts @@ -1864,6 +1864,9 @@ export const zh: Record = { 'chat.retry.httpStatus': 'HTTP {status}', 'chat.retry.networkError': '網路錯誤', 'chat.retry.waiting': '等待 {seconds}s', + 'chat.retry.retrying': '正在重試…', + 'chat.fallback.title': '網路波動，已切換為非串流請求', + 'chat.fallback.detail': '回應將一次性返回，可能需要較長時間', // ─── Tabs ────────────────────────────────────── 'tabs.close': '關閉', diff --git a/desktop/src/i18n/locales/zh.ts b/desktop/src/i18n/locales/zh.ts index 396ba4d5..c36a657f 100644 --- a/desktop/src/i18n/locales/zh.ts +++ b/desktop/src/i18n/locales/zh.ts @@ -1864,6 +1864,9 @@ export const zh: Record = { 'chat.retry.httpStatus': 'HTTP {status}', 'chat.retry.networkError': '网络错误', 'chat.retry.waiting': '等待 {seconds}s', + 'chat.retry.retrying': '正在重试…', + 'chat.fallback.title': '网络波动，已切换为非流式请求', + 'chat.fallback.detail': '响应将一次性返回，可能需要较长时间', // ─── Tabs ────────────────────────────────────── 'tabs.close': '关闭', diff --git a/desktop/src/stores/chatStore.test.ts b/desktop/src/stores/chatStore.test.ts index 60b31d25..7694708e 100644 --- a/desktop/src/stores/chatStore.test.ts +++ b/desktop/src/stores/chatStore.test.ts @@ -2888,6 +2888,72 @@ describe('chatStore history mapping', () => { expect(useChatStore.getState().sessions[TEST_SESSION_ID]?.apiRetry).toBeNull() }) + it('tracks the streaming fallback notice and supersedes a stale retry banner', () => { + useChatStore.setState({ + sessions: { + [TEST_SESSION_ID]: makeSession({ + messages: [], + chatState: 'thinking', + statusVerb: 'Thinking', + apiRetry: { + attempt: 10, + maxRetries: 10, + retryDelayMs: 1000, + errorStatus: 529, + receivedAt: Date.now() - 5_000, + }, + }), + }, + }) + + useChatStore.getState().handleServerMessage(TEST_SESSION_ID, { + type: 'streaming_fallback', + cause: 'watchdog', + }) + + const fallbackSession = useChatStore.getState().sessions[TEST_SESSION_ID] + expect(fallbackSession?.streamingFallback).toMatchObject({ cause: 'watchdog' }) + // 旧的流式重试横幅针对已放弃的请求，必须被降级提示接管。 + expect(fallbackSession?.apiRetry).toBeNull() + expect(fallbackSession?.chatState).toBe('thinking') + expect(fallbackSession?.statusVerb).toBe('') + expect(updateTabStatusMock).toHaveBeenLastCalledWith(TEST_SESSION_ID, 'running') + + // 非流式响应的首个内容块到达即清除降级提示。 + useChatStore.getState().handleServerMessage(TEST_SESSION_ID, { + type: 'content_start', + blockType: 'text', + }) + expect(useChatStore.getState().sessions[TEST_SESSION_ID]?.streamingFallback).toBeNull() + }) + + it('keeps the fallback notice when idle and clears it on turn completion', () => { + useChatStore.setState({ + sessions: { + [TEST_SESSION_ID]: makeSession({ + messages: [], + chatState: 'idle', + }), + }, + }) + + useChatStore.getState().handleServerMessage(TEST_SESSION_ID, { + type: 'streaming_fallback', + cause: '404_stream_creation', + }) + + // idle 会话收到降级信号说明回合仍在跑，状态条要回到 thinking。 + const session = useChatStore.getState().sessions[TEST_SESSION_ID] + expect(session?.chatState).toBe('thinking') + expect(session?.streamingFallback).toMatchObject({ cause: '404_stream_creation' }) + + useChatStore.getState().handleServerMessage(TEST_SESSION_ID, { + type: 'message_complete', + usage: { input_tokens: 1, output_tokens: 0 }, + }) + expect(useChatStore.getState().sessions[TEST_SESSION_ID]?.streamingFallback).toBeNull() + }) + it('renders memory saved notifications as chat memory events', () => { useChatStore.setState({ sessions: { diff --git a/desktop/src/stores/chatStore.ts b/desktop/src/stores/chatStore.ts index 77b728a4..1685ef63 100644 --- a/desktop/src/stores/chatStore.ts +++ b/desktop/src/stores/chatStore.ts @@ -26,6 +26,7 @@ import type { ComputerUsePermissionResponse, GoalEventAction, MemoryEventFile, + StreamingFallbackState, UIAttachment, UIMessage, ServerMessage, @@ -104,6 +105,8 @@ export type PerSessionState = { elapsedSeconds: number statusVerb: string apiRetry?: ApiRetryState | null + // 流式→非流式降级提示（活动回合状态，与 apiRetry 同清除时机）。 + streamingFallback?: StreamingFallbackState | null slashCommands: Array<{ name: string; description: string; argumentHint?: string }> agentTaskNotifications: Record backgroundAgentTasks?: Record @@ -139,6 +142,7 @@ const DEFAULT_SESSION_STATE: PerSessionState = { elapsedSeconds: 0, statusVerb: '', apiRetry: null, + streamingFallback: null, slashCommands: [], agentTaskNotifications: {}, backgroundAgentTasks: {}, @@ -1008,6 +1012,7 @@ export const useChatStore = create((set, get) => ({ streamingResponseChars: 0, statusVerb: isMemberSession ? '' : randomSpinnerVerb(), apiRetry: null, + streamingFallback: null, elapsedTimer: timer, connectionState: isMemberSession ? 'connected' : session.connectionState, }, @@ -1100,6 +1105,7 @@ export const useChatStore = create((set, get) => ({ pendingPermission: null, pendingComputerUsePermission: null, apiRetry: null, + streamingFallback: null, elapsedTimer: null, }, }, @@ -1229,6 +1235,7 @@ export const useChatStore = create((set, get) => ({ elapsedTimer: null, statusVerb: '', apiRetry: null, + streamingFallback: null, })), } }) @@ -1397,6 +1404,7 @@ export const useChatStore = create((set, get) => ({ streamingText: '', chatState: 'idle', apiRetry: null, + streamingFallback: null, queuedUserMessages: [], })) })) }, @@ -1445,7 +1453,7 @@ export const useChatStore = create((set, get) => ({ ? msg.verb : '', ...(msg.state === 'idle' ? { activeThinkingId: null } : {}), - ...(msg.state === 'idle' ? { apiRetry: null } : {}), + ...(msg.state === 'idle' ? { apiRetry: null, streamingFallback: null } : {}), ...(nextMessages !== session.messages ? { messages: nextMessages } : {}), ...(shouldFlush ? { streamingText: '', @@ -1491,6 +1499,7 @@ export const useChatStore = create((set, get) => ({ chatState: 'streaming', activeThinkingId: null, apiRetry: null, + streamingFallback: null, })) } else if (msg.blockType === 'tool_use') { clearPendingToolInputDelta(sessionId) @@ -1519,6 +1528,7 @@ export const useChatStore = create((set, get) => ({ chatState: 'tool_executing', activeThinkingId: null, apiRetry: null, + streamingFallback: null, })) } break @@ -1546,6 +1556,23 @@ export const useChatStore = create((set, get) => ({ break } + case 'streaming_fallback': { + // 进入非流式降级阶段：旧的重试横幅（针对失败的流式请求）已过时， + // 清掉换成降级提示；后续非流式重试到来的 api_retry 会重新接管显示。 + update((session) => ({ + streamingFallback: { + cause: msg.cause, + receivedAt: Date.now(), + }, + apiRetry: null, + chatState: session.chatState === 'idle' ? 'thinking' : session.chatState, + activeThinkingId: null, + statusVerb: '', + })) + useTabStore.getState().updateTabStatus(sessionId, 'running') + break + } + case 'content_delta': if (msg.text !== undefined) { if (!get().sessions[sessionId]) break @@ -1722,6 +1749,7 @@ export const useChatStore = create((set, get) => ({ chatState: 'permission_pending', activeThinkingId: null, apiRetry: null, + streamingFallback: null, messages: msg.toolName === 'AskUserQuestion' ? s.messages @@ -1756,6 +1784,7 @@ export const useChatStore = create((set, get) => ({ chatState: 'permission_pending', activeThinkingId: null, apiRetry: null, + streamingFallback: null, })) break @@ -1783,6 +1812,7 @@ export const useChatStore = create((set, get) => ({ pendingComputerUsePermission: null, elapsedTimer: null, apiRetry: null, + streamingFallback: null, })) useTabStore.getState().updateTabStatus(sessionId, 'idle') const appendedCompletionMessage = completionMessages !== session.messages @@ -1848,6 +1878,7 @@ export const useChatStore = create((set, get) => ({ pendingPermission: null, pendingComputerUsePermission: null, apiRetry: null, + streamingFallback: null, } }) useTabStore.getState().updateTabStatus(sessionId, 'error') @@ -1911,6 +1942,7 @@ export const useChatStore = create((set, get) => ({ elapsedSeconds: 0, statusVerb: '', apiRetry: null, + streamingFallback: null, tokenUsage: { input_tokens: 0, output_tokens: 0 }, streamingResponseChars: 0, slashCommands: [], diff --git a/desktop/src/types/chat.ts b/desktop/src/types/chat.ts index e3525458..a4e7c614 100644 --- a/desktop/src/types/chat.ts +++ b/desktop/src/types/chat.ts @@ -108,6 +108,8 @@ export type ServerMessage = errorType?: string errorMessage?: string } + // 流式请求失败、CLI 已降级为非流式重试：完整响应一次性返回，期间无增量输出。 + | { type: 'streaming_fallback'; cause: StreamingFallbackCause } | { type: 'error'; message: string; code: string; retryable?: boolean; businessErrorCode?: string } | { type: 'system_notification'; subtype: string; message?: string; data?: unknown } | { type: 'pong' } @@ -136,6 +138,14 @@ export type ApiRetryState = { receivedAt: number } +export type StreamingFallbackCause = 'watchdog' | 'stream_error' | '404_stream_creation' | 'unknown' + +// 活动回合状态（与 apiRetry 同生命周期），不进消息历史。 +export type StreamingFallbackState = { + cause: StreamingFallbackCause + receivedAt: number +} + export type TeamMemberStatus = { agentId: string role: string diff --git a/src/QueryEngine.ts b/src/QueryEngine.ts index 5118634f..1d47bf33 100644 --- a/src/QueryEngine.ts +++ b/src/QueryEngine.ts @@ -985,6 +985,15 @@ export class QueryEngine { uuid: message.uuid, } } + if (message.subtype === 'streaming_fallback') { + yield { + type: 'system', + subtype: 'streaming_fallback' as const, + cause: message.cause, + session_id: getSessionId(), + uuid: message.uuid, + } + } // Don't yield other system messages in headless mode break } diff --git a/src/entrypoints/sdk/coreSchemas.ts b/src/entrypoints/sdk/coreSchemas.ts index e20a749e..78d4b6d0 100644 --- a/src/entrypoints/sdk/coreSchemas.ts +++ b/src/entrypoints/sdk/coreSchemas.ts @@ -1587,6 +1587,20 @@ export const SDKAPIRetryMessageSchema = lazySchema(() => ), ) +export const SDKStreamingFallbackMessageSchema = lazySchema(() => + z + .object({ + type: z.literal('system'), + subtype: z.literal('streaming_fallback'), + cause: z.enum(['watchdog', 'stream_error', '404_stream_creation']), + uuid: UUIDPlaceholder(), + session_id: z.string(), + }) + .describe( + 'Emitted when a streaming request fails and the query falls back to a non-streaming request. The fallback response arrives in one piece after a potentially long wait with no incremental output.', + ), +) + export const SDKLocalCommandOutputMessageSchema = lazySchema(() => z .object({ @@ -1863,6 +1877,7 @@ export const SDKMessageSchema = lazySchema(() => SDKCompactBoundaryMessageSchema(), SDKStatusMessageSchema(), SDKAPIRetryMessageSchema(), + SDKStreamingFallbackMessageSchema(), SDKLocalCommandOutputMessageSchema(), SDKHookStartedMessageSchema(), SDKHookProgressMessageSchema(), diff --git a/src/server/__tests__/ws-memory-events.test.ts b/src/server/__tests__/ws-memory-events.test.ts index b022e462..dd84bb3e 100644 --- a/src/server/__tests__/ws-memory-events.test.ts +++ b/src/server/__tests__/ws-memory-events.test.ts @@ -270,6 +270,45 @@ describe('WebSocket API retry events', () => { }, ]) }) + + it('forwards CLI streaming_fallback messages with a recognized cause', () => { + // 形状对齐 QueryEngine 的 SDK 输出：{type:'system', subtype:'streaming_fallback', cause, ...} + expect(translateCliMessage({ + type: 'system', + subtype: 'streaming_fallback', + cause: 'watchdog', + session_id: 'session-1', + uuid: 'uuid-1', + }, 'session-1')).toEqual([ + { type: 'streaming_fallback', cause: 'watchdog' }, + ]) + + expect(translateCliMessage({ + type: 'system', + subtype: 'streaming_fallback', + cause: '404_stream_creation', + }, 'session-1')).toEqual([ + { type: 'streaming_fallback', cause: '404_stream_creation' }, + ]) + }) + + it('normalizes unrecognized streaming_fallback causes to unknown instead of dropping the event', () => { + // 新 CLI + 旧枚举：提示本身比成因重要，不能丢消息。 + expect(translateCliMessage({ + type: 'system', + subtype: 'streaming_fallback', + cause: 'some_future_cause', + }, 'session-1')).toEqual([ + { type: 'streaming_fallback', cause: 'unknown' }, + ]) + + expect(translateCliMessage({ + type: 'system', + subtype: 'streaming_fallback', + }, 'session-1')).toEqual([ + { type: 'streaming_fallback', cause: 'unknown' }, + ]) + }) }) describe('WebSocket background task events', () => { diff --git a/src/server/ws/events.ts b/src/server/ws/events.ts index 1b66c236..b0abafe4 100644 --- a/src/server/ws/events.ts +++ b/src/server/ws/events.ts @@ -79,6 +79,9 @@ export type ServerMessage = errorType?: string errorMessage?: string } + // 流式请求失败、CLI 已降级为非流式重试。非流式响应要等完整生成才返回， + // 期间没有任何增量输出，前端据此显示"慢速模式"轻提示而不是裸转圈。 + | { type: 'streaming_fallback'; cause: StreamingFallbackCause } | { type: 'error'; message: string; code: string; retryable?: boolean; businessErrorCode?: string } | { type: 'system_notification'; subtype: string; message?: string; data?: unknown } | { type: 'pong' } @@ -97,6 +100,10 @@ export type TokenUsage = { export type ChatState = 'idle' | 'thinking' | 'compacting' | 'tool_executing' | 'streaming' | 'permission_pending' +// 与 CLI 的 streaming_fallback cause 对齐；unknown 兜底未来新增的 cause 值， +// 避免新 CLI + 旧 server 组合下丢消息。 +export type StreamingFallbackCause = 'watchdog' | 'stream_error' | '404_stream_creation' | 'unknown' + export type TeamMemberStatus = { agentId: string role: string diff --git a/src/server/ws/handler.ts b/src/server/ws/handler.ts index c0bdcde0..5879d942 100644 --- a/src/server/ws/handler.ts +++ b/src/server/ws/handler.ts @@ -7,7 +7,7 @@ */ import type { ServerWebSocket } from 'bun' -import type { ClientMessage, ServerMessage } from './events.js' +import type { ClientMessage, ServerMessage, StreamingFallbackCause } from './events.js' import * as os from 'node:os' import { ConversationStartupError, @@ -1692,6 +1692,9 @@ export function translateCliMessage(cliMsg: any, sessionId: string): ServerMessa const apiRetryMessage = toApiRetryServerMessage(cliMsg) return apiRetryMessage ? [apiRetryMessage] : [] } + if (subtype === 'streaming_fallback') { + return [toStreamingFallbackServerMessage(cliMsg)] + } if (subtype === 'init') { // CLI 初始化完成 — 缓存 slash commands 并发送模型信息 // NOTE: Do NOT send status:idle here — the CLI init fires while @@ -1902,6 +1905,21 @@ function toApiRetryServerMessage(cliMsg: any): ServerMessage | null { } } +const STREAMING_FALLBACK_CAUSES: ReadonlySet = new Set([ + 'watchdog', + 'stream_error', + '404_stream_creation', +]) + +function toStreamingFallbackServerMessage(cliMsg: any): ServerMessage { + // 未识别的 cause 兜底为 unknown 而不是丢消息：提示本身比成因重要。 + const cause: StreamingFallbackCause = + typeof cliMsg.cause === 'string' && STREAMING_FALLBACK_CAUSES.has(cliMsg.cause as StreamingFallbackCause) + ? (cliMsg.cause as StreamingFallbackCause) + : 'unknown' + return { type: 'streaming_fallback', cause } +} + function sendMessage(ws: ServerWebSocket, message: ServerMessage) { ws.send(JSON.stringify(message)) } diff --git a/src/services/api/claude.ts b/src/services/api/claude.ts index 4fe93092..5824bd3e 100644 --- a/src/services/api/claude.ts +++ b/src/services/api/claude.ts @@ -47,6 +47,7 @@ import type { Message, StreamEvent, SystemAPIErrorMessage, + SystemStreamingFallbackMessage, UserMessage, } from "../../types/message.js"; import { @@ -74,6 +75,7 @@ import { computeFingerprintFromMessages } from "../../utils/fingerprint.js"; import { captureAPIRequest, logError } from "../../utils/log.js"; import { createAssistantAPIErrorMessage, + createSystemStreamingFallbackMessage, createUserMessage, ensureToolResultPairing, normalizeContentFromAPI, @@ -767,7 +769,7 @@ export async function* queryModelWithStreaming({ signal: AbortSignal; options: Options; }): AsyncGenerator< - StreamEvent | AssistantMessage | SystemAPIErrorMessage, + StreamEvent | AssistantMessage | SystemAPIErrorMessage | SystemStreamingFallbackMessage, void > { return yield* withStreamingVCR(messages, async function* () { @@ -1025,7 +1027,7 @@ async function* queryModel( signal: AbortSignal, options: Options, ): AsyncGenerator< - StreamEvent | AssistantMessage | SystemAPIErrorMessage, + StreamEvent | AssistantMessage | SystemAPIErrorMessage | SystemStreamingFallbackMessage, void > { if (getAPIProvider() === "azureOpenAI") { @@ -2574,6 +2576,13 @@ async function* queryModel( if (options.onStreamingFallback) { options.onStreamingFallback(); } + // Surface the mode switch to consumers (SDK stream → desktop status + // bar): the non-streaming response arrives in one piece after a + // potentially long silent wait, so without this signal the UI shows a + // bare spinner the whole time. + yield createSystemStreamingFallbackMessage( + streamIdleAborted ? "watchdog" : "stream_error", + ); logEvent("tengu_streaming_fallback_to_non_streaming", { model: @@ -2695,6 +2704,7 @@ async function* queryModel( if (options.onStreamingFallback) { options.onStreamingFallback(); } + yield createSystemStreamingFallbackMessage("404_stream_creation"); logEvent("tengu_streaming_fallback_to_non_streaming", { model: diff --git a/src/utils/messages.ts b/src/utils/messages.ts index 266b21d4..a1b3b17a 100644 --- a/src/utils/messages.ts +++ b/src/utils/messages.ts @@ -71,6 +71,7 @@ import type { SystemPermissionRetryMessage, SystemScheduledTaskFireMessage, SystemStopHookSummaryMessage, + SystemStreamingFallbackMessage, SystemTurnDurationMessage, TombstoneMessage, ToolUseSummaryMessage, @@ -4718,6 +4719,31 @@ export function createSystemAPIErrorMessage( } } +export type StreamingFallbackCause = + | 'watchdog' + | 'stream_error' + | '404_stream_creation' + +/** + * Marks the switch from a failed streaming request to the non-streaming + * fallback. The fallback response arrives in one piece after a potentially + * long wait with zero incremental output, so UIs surface this as a lightweight + * active-turn status (level info — an expected state, not an error). + */ +export function createSystemStreamingFallbackMessage( + cause: StreamingFallbackCause, +): SystemStreamingFallbackMessage { + return { + type: 'system', + subtype: 'streaming_fallback', + level: 'info', + content: `Streaming request failed (${cause.replace(/_/g, ' ')}); retrying in non-streaming mode`, + cause, + timestamp: new Date().toISOString(), + uuid: randomUUID(), + } +} + /** * Checks if a message is a compact boundary marker */