diff --git a/src/acp-agent.ts b/src/acp-agent.ts index 93c8d066..dac05529 100644 --- a/src/acp-agent.ts +++ b/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(); + const streamedThinkingIds = new Set(); 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 = - message.type === "assistant" - ? // Handled by stream events above - message.message.content.filter( - (item) => !["text", "thinking"].includes(item.type), - ) - : message.message.content; + let content: typeof message.message.content; + if (message.type === "assistant" && message.parent_tool_use_id === null) { + // Top-level assistant message: drop text/thinking blocks already + // streamed live as chunks, and forward (as a fallback) any that were + // not, so non-streaming gateways still deliver the final answer. + const id = messageIdForGrouping(message); + 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, diff --git a/src/tests/acp-agent.test.ts b/src/tests/acp-agent.test.ts index 20157889..8367ac68 100644 --- a/src/tests/acp-agent.test.ts +++ b/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(); + 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[] = [];