fix: Duplicate assistant messages in feed (#785)

src/acp-agent.ts

@@ -959,9 +959,8 @@ export class ClaudeAcpAgent implements Agent {
     // via `stream_event` deltas, split by block type, so the consolidated
     // `assistant` message can drop the duplicate blocks that already reached the
     // client as chunks while still forwarding any that a non-streaming gateway
-    // delivered only as an assembled block.
-    let streamedTextIds = new Set<string>();
-    let streamedThinkingIds = new Set<string>();
+    const streamedTextIds = new Set<string>();
+    const streamedThinkingIds = new Set<string>();
     // Stop reason accumulated for the active turn (result subtype, refusal,
     // max_tokens, …). Reset per turn; read when the turn settles at idle.
     let stopReason: StopReason = "end_turn";
@@ -974,8 +973,6 @@ export class ClaudeAcpAgent implements Agent {
       lastRefusalExplanation = null;
       compactionInProgress = false;
       currentStreamMessageId = undefined;
-      streamedTextIds = new Set<string>();
-      streamedThinkingIds = new Set<string>();
       stopReason = "end_turn";
       session.accumulatedUsage = {
         inputTokens: 0,
@@ -1864,6 +1861,13 @@ export class ClaudeAcpAgent implements Agent {
                 }
                 return true;
               });
+              // The consolidated message is the last place this id is needed for
+              // dedupe — drop it so the streamed-id sets stay bounded to in-flight
+              // streams rather than growing for the session's whole life.
+              if (id !== undefined) {
+                streamedTextIds.delete(id);
+                streamedThinkingIds.delete(id);
+              }
             } 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

src/tests/acp-agent.test.ts

@@ -3679,6 +3679,36 @@ describe("assembled assistant text fallback", () => {
     });
   }
 
+  // Like injectSession, but the user-message echo is yielded at the position of
+  // the "ECHO" sentinel in `messages` rather than always first — so a test can
+  // reproduce the production ordering where the assistant stream arrives before
+  // the SDK replays the user message.
+  function injectSessionEchoAt(agent: ClaudeAcpAgent, messages: any[]) {
+    const input = new Pushable<any>();
+    async function* messageGenerator() {
+      const iter = input[Symbol.asyncIterator]();
+      const { value: userMessage } = await iter.next();
+      for (const m of messages) {
+        if (m === "ECHO") {
+          yield {
+            type: "user",
+            message: userMessage.message,
+            parent_tool_use_id: null,
+            uuid: userMessage.uuid,
+            session_id: "test-session",
+            isReplay: true,
+          };
+        } else {
+          yield m;
+        }
+      }
+    }
+    agent.sessions["test-session"] = mockSessionState({
+      query: wrapQuery(messageGenerator()),
+      input,
+    });
+  }
+
   function messageChunkTexts(updates: any[]): string[] {
     return updates
       .filter((u) => u.update?.sessionUpdate === "agent_message_chunk")
@@ -3723,6 +3753,28 @@ describe("assembled assistant text fallback", () => {
     expect(messageChunkTexts(updates)).toEqual(["hello ", "world"]);
   });
 
+  it("dedupes streamed text even when the stream arrives before the user echo", async () => {
+    const { agent, updates } = createMockAgentWithCapture();
+    // Production ordering: the SDK emits the assistant's stream events before it
+    // replays the user message that activates the turn. The streamed-id tracking
+    // must survive activation, or the consolidated block is re-emitted as a
+    // duplicate (regression from the persistent-consumer rework).
+    injectSessionEchoAt(agent, [
+      messageStart("msg-streamed"),
+      textDelta("hello "),
+      textDelta("world"),
+      "ECHO",
+      assistantMessage("msg-streamed", [{ type: "text", text: "hello world" }]),
+      result(),
+      idle,
+    ]);
+
+    await agent.prompt({ sessionId: "test-session", prompt: [{ type: "text", text: "hi" }] });
+
+    // Still just the two streamed deltas — no duplicated assembled block.
+    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