diff --git a/src/services/acp/__tests__/bridge.test.ts b/src/services/acp/__tests__/bridge.test.ts index 76e82bd20b..bb3a73ddb6 100644 --- a/src/services/acp/__tests__/bridge.test.ts +++ b/src/services/acp/__tests__/bridge.test.ts @@ -11,6 +11,8 @@ import { promptToQueryInput } from '../promptConversion.js' import { markdownEscape, toDisplayPath } from '../utils.js' import type { AgentSideConnection, ToolKind } from '@agentclientprotocol/sdk' import type { SDKMessage } from '../../../entrypoints/sdk/coreTypes.js' +import { createAssistantAPIErrorMessage } from '../../../utils/messages.js' +import { normalizeMessage } from '../../../utils/queryHelpers.js' // ── Helpers ──────────────────────────────────────────────────────── @@ -1871,3 +1873,136 @@ describe('replayHistoryMessages — message-id (RFD)', () => { expect(typeof chunkCall!.messageId).toBe('string') }) }) + +// ── API error messages must survive streaming ──────────────────── +// +// A synthetic API error message (isApiErrorMessage) carries its text only in +// `message.content`. No stream_event ever emitted that text, so the +// streamingActive duplicate filter must not drop it — otherwise the failure +// reaches the client as an empty turn with no error surface. + +describe('forwardSessionUpdates — API error messages', () => { + test('emits agent_message_chunk for API error text while streaming is active', async () => { + const conn = makeConn() + const msgs: SDKMessage[] = [ + { + type: 'stream_event', + parent_tool_use_id: null, + event: { + type: 'content_block_delta', + delta: { type: 'text_delta', text: 'partial answer' }, + }, + }, + { + type: 'assistant', + parent_tool_use_id: null, + isApiErrorMessage: true, + message: { + model: '', + role: 'assistant', + content: [ + { + type: 'text', + text: 'API Error: OpenAI stream ended without returning any data', + }, + ], + }, + } as unknown as SDKMessage, + ] + await forwardSessionUpdates( + 's1', + makeStream(msgs), + conn, + new AbortController().signal, + {}, + ) + const calls = (conn.sessionUpdate as ReturnType).mock.calls + const texts = calls + .map(c => (c[0] as { update: Record }).update) + .filter(u => u.sessionUpdate === 'agent_message_chunk') + .map(u => (u.content as { text: string }).text) + expect(texts).toContain( + 'API Error: OpenAI stream ended without returning any data', + ) + }) + + test('still filters the streamed duplicate of a normal assistant message', async () => { + const conn = makeConn() + const msgs: SDKMessage[] = [ + { + type: 'stream_event', + parent_tool_use_id: null, + event: { + type: 'content_block_delta', + delta: { type: 'text_delta', text: 'Answer' }, + }, + }, + { + type: 'assistant', + parent_tool_use_id: null, + message: { + role: 'assistant', + content: [{ type: 'text', text: 'Answer' }], + }, + } as unknown as SDKMessage, + ] + await forwardSessionUpdates( + 's1', + makeStream(msgs), + conn, + new AbortController().signal, + {}, + ) + const calls = (conn.sessionUpdate as ReturnType).mock.calls + const texts = calls + .map(c => (c[0] as { update: Record }).update) + .filter(u => u.sessionUpdate === 'agent_message_chunk') + .map(u => (u.content as { text: string }).text) + expect(texts).toEqual(['Answer']) + }) + + // Unlike the two tests above, which hand-build the SDKMessage with the flag + // pre-set, this one drives the real pipeline: the same factory the query + // engine uses (createAssistantAPIErrorMessage) → normalizeMessage → bridge. + // The passthrough in normalizeMessage is what marks the error text as + // never-streamed. If that passthrough is ever removed, the bridge can no + // longer tell the difference and the dedup filter drops the text — this + // test fails while the hand-built ones stay green. + test('end-to-end: real createAssistantAPIErrorMessage survives normalizeMessage', async () => { + const conn = makeConn() + const errorMsg = createAssistantAPIErrorMessage({ + content: 'API Error: probe', + }) + const sdkMsgs = [...normalizeMessage(errorMsg)] + + // Assert the bridge-visible flag directly for a precise failure signal. + expect( + (sdkMsgs[0] as unknown as Record).isApiErrorMessage, + ).toBe(true) + + const msgs: SDKMessage[] = [ + { + type: 'stream_event', + parent_tool_use_id: null, + event: { + type: 'content_block_delta', + delta: { type: 'text_delta', text: 'partial answer' }, + }, + } as unknown as SDKMessage, + ...sdkMsgs, + ] + await forwardSessionUpdates( + 's1', + makeStream(msgs), + conn, + new AbortController().signal, + {}, + ) + const calls = (conn.sessionUpdate as ReturnType).mock.calls + const texts = calls + .map(c => (c[0] as { update: Record }).update) + .filter(u => u.sessionUpdate === 'agent_message_chunk') + .map(u => (u.content as { text: string }).text) + expect(texts).toContain('API Error: probe') + }) +}) diff --git a/src/services/acp/bridge/notifications.ts b/src/services/acp/bridge/notifications.ts index f4e7a9592e..cb2b3bde87 100644 --- a/src/services/acp/bridge/notifications.ts +++ b/src/services/acp/bridge/notifications.ts @@ -235,7 +235,11 @@ export function toAcpNotifications( } export function assistantMessageToAcpNotifications( - msg: { message?: unknown; parent_tool_use_id?: string | null }, + msg: { + message?: unknown + parent_tool_use_id?: string | null + isApiErrorMessage?: boolean + }, sessionId: string, toolUseCache: ToolUseCache, conn: AgentSideConnection, @@ -272,13 +276,17 @@ export function assistantMessageToAcpNotifications( // When streaming is active, text/thinking were already sent via stream_event // messages. Filter them out to avoid duplicate agent_message_chunk / - // agent_thought_chunk notifications. String content (synthetic messages) - // is unaffected — those have no corresponding stream_events. - const contentToProcess = options?.streamingActive - ? content.filter( - block => block.type !== 'text' && block.type !== 'thinking', - ) - : content + // agent_thought_chunk notifications. + // + // API error messages bypass this filter. They are synthetic — no stream_event + // ever carried their text — so filtering would drop the only copy and the + // failure would reach the client as a silent, empty turn. + const contentToProcess = + options?.streamingActive && msg.isApiErrorMessage !== true + ? content.filter( + block => block.type !== 'text' && block.type !== 'thinking', + ) + : content if (contentToProcess.length === 0) return [] diff --git a/src/services/acp/bridge/types.ts b/src/services/acp/bridge/types.ts index bdb8031f83..036c761621 100644 --- a/src/services/acp/bridge/types.ts +++ b/src/services/acp/bridge/types.ts @@ -91,6 +91,9 @@ export type BridgeAssistantMessage = { uuid?: string session_id?: string error?: unknown + // Set on synthetic messages that carry an upstream API failure. Their text + // exists only here, so the bridge must not treat it as a streamed duplicate. + isApiErrorMessage?: boolean [key: string]: unknown } diff --git a/src/utils/queryHelpers.ts b/src/utils/queryHelpers.ts index 94dbb74044..ea0a7a138b 100644 --- a/src/utils/queryHelpers.ts +++ b/src/utils/queryHelpers.ts @@ -124,6 +124,9 @@ export function* normalizeMessage(message: Message): Generator { session_id: getSessionId(), uuid: _.uuid, error: _.error, + // Carried through so the ACP bridge can tell a synthetic upstream + // failure from a real assistant turn whose text was already streamed. + ...(_.isApiErrorMessage ? { isApiErrorMessage: true } : {}), } } return