fix: Forward unstreamed assistant text blocks (#757)

src/acp-agent.ts

@@ -862,6 +862,21 @@ export class ClaudeAcpAgent implements Agent {
     // `parent_tool_use_id === null` (subagent work is folded into tool-result
     // messages, never surfaced as partial streams).
     let currentStreamMessageId: string | undefined;
+    // Per-message-id record of which assistant content actually streamed live via
+    // `stream_event` deltas, split by block type. The `assistant` case below
+    // normally drops `text`/`thinking` blocks on the assumption they already
+    // reached the client as `agent_message_chunk`/`agent_thought_chunk`. That
+    // breaks behind Anthropic-protocol gateways that return a turn as a single
+    // non-streamed block (common with OpenAI-compatible proxies): no
+    // `content_block_delta` fires, so the assembled block is the only copy the
+    // client will ever see. We keep the filter per (id, block type) for content
+    // that did stream, and fall back to forwarding what did not. Split by type so
+    // a gateway that streams text but not thinking (or vice versa) doesn't lose
+    // the un-streamed block. Only top-level (`parent_tool_use_id === null`)
+    // streams are recorded — subagent text is never streamed and must stay
+    // filtered, as it is internal to the tool call.
+    const streamedTextIds = new Set<string>();
+    const streamedThinkingIds = new Set<string>();
 
     const userMessage = promptToClaude(params);
 
@@ -1309,6 +1324,22 @@ export class ClaudeAcpAgent implements Agent {
             if (message.event.type === "message_start") {
               currentStreamMessageId = message.event.message.id || undefined;
             }
+            // Record that this top-level message id actually streamed text/
+            // thinking, so the `assistant` case below knows its assembled blocks
+            // are duplicates (filter them) rather than the only copy (forward
+            // them). Gated on `parent_tool_use_id === null` so a subagent stream
+            // can't attribute its content to the top-level message id.
+            if (
+              currentStreamMessageId &&
+              message.parent_tool_use_id === null &&
+              message.event.type === "content_block_delta"
+            ) {
+              if (message.event.delta.type === "text_delta") {
+                streamedTextIds.add(currentStreamMessageId);
+              } else if (message.event.delta.type === "thinking_delta") {
+                streamedThinkingIds.add(currentStreamMessageId);
+              }
+            }
             if (
               message.parent_tool_use_id === null &&
               (message.event.type === "message_start" || message.event.type === "message_delta")
@@ -1498,13 +1529,47 @@ export class ClaudeAcpAgent implements Agent {
               throw RequestError.authRequired();
             }
 
-            const content =
+            let content: typeof message.message.content;
-              message.type === "assistant"
+            if (message.type === "assistant" && message.parent_tool_use_id === null) {
-                ? // Handled by stream events above
+              // Top-level assistant message: drop text/thinking blocks already
-                  message.message.content.filter(
+              // streamed live as chunks, and forward (as a fallback) any that were
-                    (item) => !["text", "thinking"].includes(item.type),
+              // not, so non-streaming gateways still deliver the final answer.
-                  )
+              const id = messageIdForGrouping(message);
-                : message.message.content;
+              content = message.message.content.filter((item) => {
+                // Non-text blocks (tool_use, etc.) always pass through; their own
+                // dedupe (`toolUseCache`) collapses the streamed/assembled pair.
+                if (item.type !== "text" && item.type !== "thinking") {
+                  return true;
+                }
+                // Already delivered live as a chunk of this exact type — drop the
+                // duplicate. Checked per type so streaming one (e.g. text) doesn't
+                // suppress an un-streamed block of the other (e.g. thinking).
+                const streamedLive =
+                  id !== undefined &&
+                  (item.type === "text" ? streamedTextIds : streamedThinkingIds).has(id);
+                if (streamedLive) {
+                  return false;
+                }
+                // Empty assembled blocks carry nothing (some gateways emit an empty
+                // `thinking` block before the real text) — don't forward stray
+                // empty chunks.
+                const text = item.type === "text" ? item.text : item.thinking;
+                if (text.length === 0) {
+                  return false;
+                }
+                return true;
+              });
+            } else if (message.type === "assistant") {
+              // Subagent assistant message (`parent_tool_use_id !== null`). It is
+              // never streamed live and its text/thinking is internal to the tool
+              // call — keep dropping it so subagent prose doesn't leak into the
+              // top-level feed.
+              content = message.message.content.filter(
+                (item) => item.type !== "text" && item.type !== "thinking",
+              );
+            } else {
+              content = message.message.content;
+            }
 
             for (const notification of toAcpNotifications(
               content,

src/tests/acp-agent.test.ts

@@ -3323,6 +3323,266 @@ describe("usage_update computation", () => {
   });
 });
 
+describe("assembled assistant text fallback", () => {
+  const ZERO_USAGE = {
+    input_tokens: 0,
+    output_tokens: 0,
+    cache_read_input_tokens: 0,
+    cache_creation_input_tokens: 0,
+  };
+
+  function createMockAgentWithCapture() {
+    const updates: any[] = [];
+    const mockClient = {
+      sessionUpdate: async (notification: any) => {
+        updates.push(notification);
+      },
+    } as unknown as AgentSideConnection;
+    const agent = new ClaudeAcpAgent(mockClient, { log: () => {}, error: () => {} });
+    return { agent, updates };
+  }
+
+  function messageStart(apiId: string) {
+    return {
+      type: "stream_event" as const,
+      parent_tool_use_id: null,
+      uuid: randomUUID(),
+      session_id: "test-session",
+      event: {
+        type: "message_start" as const,
+        message: { id: apiId, model: "claude-sonnet-4-20250514", usage: ZERO_USAGE },
+      },
+    };
+  }
+
+  function textDelta(text: string) {
+    return {
+      type: "stream_event" as const,
+      parent_tool_use_id: null,
+      uuid: randomUUID(),
+      session_id: "test-session",
+      event: {
+        type: "content_block_delta" as const,
+        index: 0,
+        delta: { type: "text_delta" as const, text },
+      },
+    };
+  }
+
+  function assistantMessage(apiId: string, content: any[], parentToolUseId: string | null = null) {
+    return {
+      type: "assistant" as const,
+      parent_tool_use_id: parentToolUseId,
+      uuid: randomUUID(),
+      session_id: "test-session",
+      message: {
+        id: apiId,
+        role: "assistant" as const,
+        model: "claude-sonnet-4-20250514",
+        content,
+        usage: ZERO_USAGE,
+      },
+    };
+  }
+
+  function result() {
+    return {
+      type: "result" as const,
+      subtype: "success" as const,
+      stop_reason: "end_turn",
+      is_error: false,
+      result: "",
+      errors: [],
+      duration_ms: 0,
+      duration_api_ms: 0,
+      num_turns: 1,
+      total_cost_usd: 0,
+      usage: ZERO_USAGE,
+      modelUsage: {},
+      permission_denials: [],
+      uuid: randomUUID(),
+      session_id: "test-session",
+    };
+  }
+
+  const idle = { type: "system", subtype: "session_state_changed", state: "idle" };
+
+  function injectSession(agent: ClaudeAcpAgent, messages: any[]) {
+    const input = new Pushable<any>();
+    async function* messageGenerator() {
+      const iter = input[Symbol.asyncIterator]();
+      const { value: userMessage, done } = await iter.next();
+      if (!done && userMessage) {
+        yield {
+          type: "user",
+          message: userMessage.message,
+          parent_tool_use_id: null,
+          uuid: userMessage.uuid,
+          session_id: "test-session",
+          isReplay: true,
+        };
+      }
+      yield* messages;
+    }
+    agent.sessions["test-session"] = {
+      query: messageGenerator() as any,
+      input,
+      cancelled: false,
+      cwd: "/test",
+      sessionFingerprint: JSON.stringify({ cwd: "/test", mcpServers: [] }),
+      modes: { currentModeId: "default", availableModes: [] },
+      models: { currentModelId: "default", availableModels: [] },
+      modelInfos: [],
+      settingsManager: { dispose: vi.fn() } as any,
+      accumulatedUsage: {
+        inputTokens: 0,
+        outputTokens: 0,
+        cachedReadTokens: 0,
+        cachedWriteTokens: 0,
+      },
+      configOptions: [],
+      promptRunning: false,
+      pendingMessages: new Map(),
+      nextPendingOrder: 0,
+      abortController: new AbortController(),
+      emitRawSDKMessages: false,
+      contextWindowSize: 200000,
+      taskState: new Map(),
+      toolUseCache: {},
+      messageIdToUuid: new Map(),
+    };
+  }
+
+  function messageChunkTexts(updates: any[]): string[] {
+    return updates
+      .filter((u) => u.update?.sessionUpdate === "agent_message_chunk")
+      .map((u) => u.update.content.text);
+  }
+
+  function thoughtChunkTexts(updates: any[]): string[] {
+    return updates
+      .filter((u) => u.update?.sessionUpdate === "agent_thought_chunk")
+      .map((u) => u.update.content.text);
+  }
+
+  it("emits the assembled text when no content_block_delta was streamed", async () => {
+    const { agent, updates } = createMockAgentWithCapture();
+    // Gateway delivers a fully assembled message with no preceding deltas.
+    injectSession(agent, [
+      assistantMessage("msg-no-stream", [{ type: "text", text: "the final answer" }]),
+      result(),
+      idle,
+    ]);
+
+    await agent.prompt({ sessionId: "test-session", prompt: [{ type: "text", text: "hi" }] });
+
+    expect(messageChunkTexts(updates)).toEqual(["the final answer"]);
+  });
+
+  it("does not re-emit text already streamed via content_block_delta", async () => {
+    const { agent, updates } = createMockAgentWithCapture();
+    // Normal streaming: deltas arrive, then the consolidated message repeats them.
+    injectSession(agent, [
+      messageStart("msg-streamed"),
+      textDelta("hello "),
+      textDelta("world"),
+      assistantMessage("msg-streamed", [{ type: "text", text: "hello world" }]),
+      result(),
+      idle,
+    ]);
+
+    await agent.prompt({ sessionId: "test-session", prompt: [{ type: "text", text: "hi" }] });
+
+    // Only the two streamed deltas — the assembled block is filtered out.
+    expect(messageChunkTexts(updates)).toEqual(["hello ", "world"]);
+  });
+
+  it("dedupes per block type: streamed text is dropped but an un-streamed thinking block in the same message is forwarded", async () => {
+    const { agent, updates } = createMockAgentWithCapture();
+    // Gateway streams the text live but delivers the thinking block only in the
+    // assembled message (no thinking_delta). The dedupe must be per-type so the
+    // thinking survives. This also makes the test non-vacuous: if the fallback
+    // were removed (text/thinking always dropped) the thought chunk disappears.
+    injectSession(agent, [
+      messageStart("msg-mixed"),
+      textDelta("streamed text"),
+      assistantMessage("msg-mixed", [
+        { type: "text", text: "streamed text" },
+        { type: "thinking", thinking: "private reasoning" },
+      ]),
+      result(),
+      idle,
+    ]);
+
+    await agent.prompt({ sessionId: "test-session", prompt: [{ type: "text", text: "hi" }] });
+
+    // Streamed text appears once (delta only — assembled copy deduped).
+    expect(messageChunkTexts(updates)).toEqual(["streamed text"]);
+    // The un-streamed thinking block is forwarded despite text having streamed.
+    expect(thoughtChunkTexts(updates)).toEqual(["private reasoning"]);
+  });
+
+  it("does not leak subagent assistant text into the top-level feed", async () => {
+    const { agent, updates } = createMockAgentWithCapture();
+    // Subagent assistant messages (parent_tool_use_id !== null) are never
+    // streamed live; their text/thinking is internal to the tool call and must
+    // stay filtered out, not surface as a fallback chunk.
+    injectSession(agent, [
+      assistantMessage(
+        "msg-subagent",
+        [{ type: "text", text: "subagent internal prose" }],
+        "tool_use_1",
+      ),
+      result(),
+      idle,
+    ]);
+
+    await agent.prompt({ sessionId: "test-session", prompt: [{ type: "text", text: "hi" }] });
+
+    expect(messageChunkTexts(updates)).toEqual([]);
+    expect(thoughtChunkTexts(updates)).toEqual([]);
+  });
+
+  it("forwards distinct blocks that a gateway splits across same-id messages", async () => {
+    const { agent, updates } = createMockAgentWithCapture();
+    // Observed with OpenAI-compatible gateways: one response id split into an
+    // empty thinking block, then the real text — both with no deltas.
+    injectSession(agent, [
+      assistantMessage("msg-split", [{ type: "thinking", thinking: "" }]),
+      assistantMessage("msg-split", [{ type: "text", text: "the real answer" }]),
+      result(),
+      idle,
+    ]);
+
+    await agent.prompt({ sessionId: "test-session", prompt: [{ type: "text", text: "hi" }] });
+
+    // The text survives even though an earlier same-id message already triggered
+    // the fallback for a different (thinking) block.
+    expect(messageChunkTexts(updates)).toEqual(["the real answer"]);
+    // The empty thinking block carries nothing and must not produce a stray
+    // empty thought chunk.
+    expect(thoughtChunkTexts(updates)).toEqual([]);
+  });
+
+  it("re-forwards a block a gateway re-delivers (no content-keyed dedupe)", async () => {
+    const { agent, updates } = createMockAgentWithCapture();
+    // The fallback intentionally keys only on whether the id streamed live, not
+    // on block content — so a gateway re-delivering the same assembled block
+    // emits it twice. This is the accepted, cosmetic tradeoff for not caching
+    // every fallback block's full text; see `streamedTextMessageIds`.
+    injectSession(agent, [
+      assistantMessage("msg-dup", [{ type: "text", text: "answer" }]),
+      assistantMessage("msg-dup", [{ type: "text", text: "answer" }]),
+      result(),
+      idle,
+    ]);
+
+    await agent.prompt({ sessionId: "test-session", prompt: [{ type: "text", text: "hi" }] });
+
+    expect(messageChunkTexts(updates)).toEqual(["answer", "answer"]);
+  });
+});
+
 describe("emitRawSDKMessages", () => {
   function createMockAgentWithExtNotification() {
     const updates: any[] = [];