fix: ACP 转发不再丢弃上游 API 错误消息

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: '<synthetic>',
+          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<typeof mock>).mock.calls
+    const texts = calls
+      .map(c => (c[0] as { update: Record<string, unknown> }).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<typeof mock>).mock.calls
+    const texts = calls
+      .map(c => (c[0] as { update: Record<string, unknown> }).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<string, unknown>).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<typeof mock>).mock.calls
+    const texts = calls
+      .map(c => (c[0] as { update: Record<string, unknown> }).update)
+      .filter(u => u.sessionUpdate === 'agent_message_chunk')
+      .map(u => (u.content as { text: string }).text)
+    expect(texts).toContain('API Error: probe')
+  })
+})

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 []
 

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
 }
 

src/utils/queryHelpers.ts

@@ -124,6 +124,9 @@ export function* normalizeMessage(message: Message): Generator<SDKMessage> {
           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