fix(acp): preserve final assistant message snapshot before end_turn (#44597)

Process messageData via handleDeltaEvent for both delta and final states
before resolving the turn, so ACP clients no longer drop the last visible
assistant text when the gateway sends the final message body on the
terminal chat event.

Closes #15377
Based on #17615

Co-authored-by: PJ Eby <3527052+pjeby@users.noreply.github.com>

src/acp/translator.session-rate-limit.test.ts

@@ -1020,3 +1020,144 @@ describe("acp prompt size hardening", () => {
     });
   });
 });
+
+describe("acp final chat snapshots", () => {
+  async function createSnapshotHarness() {
+    const sessionStore = createInMemorySessionStore();
+    const connection = createAcpConnection();
+    const sessionUpdate = connection.__sessionUpdateMock;
+    const request = vi.fn(async (method: string) => {
+      if (method === "chat.send") {
+        return new Promise(() => {});
+      }
+      return { ok: true };
+    }) as GatewayClient["request"];
+    const agent = new AcpGatewayAgent(connection, createAcpGateway(request), {
+      sessionStore,
+    });
+    await agent.loadSession(createLoadSessionRequest("snapshot-session"));
+    sessionUpdate.mockClear();
+    const promptPromise = agent.prompt(createPromptRequest("snapshot-session", "hello"));
+    const runId = sessionStore.getSession("snapshot-session")?.activeRunId;
+    if (!runId) {
+      throw new Error("Expected ACP prompt run to be active");
+    }
+    return { agent, sessionUpdate, promptPromise, runId, sessionStore };
+  }
+
+  it("emits final snapshot text before resolving end_turn", async () => {
+    const { agent, sessionUpdate, promptPromise, runId, sessionStore } =
+      await createSnapshotHarness();
+
+    await agent.handleGatewayEvent({
+      event: "chat",
+      payload: {
+        sessionKey: "snapshot-session",
+        runId,
+        state: "final",
+        stopReason: "end_turn",
+        message: {
+          content: [{ type: "text", text: "FINAL TEXT SHOULD BE EMITTED" }],
+        },
+      },
+    } as unknown as EventFrame);
+
+    await expect(promptPromise).resolves.toEqual({ stopReason: "end_turn" });
+    expect(sessionUpdate).toHaveBeenCalledWith({
+      sessionId: "snapshot-session",
+      update: {
+        sessionUpdate: "agent_message_chunk",
+        content: { type: "text", text: "FINAL TEXT SHOULD BE EMITTED" },
+      },
+    });
+    expect(sessionStore.getSession("snapshot-session")?.activeRunId).toBeNull();
+    sessionStore.clearAllSessionsForTest();
+  });
+
+  it("does not duplicate text when final repeats the last delta snapshot", async () => {
+    const { agent, sessionUpdate, promptPromise, runId, sessionStore } =
+      await createSnapshotHarness();
+
+    await agent.handleGatewayEvent({
+      event: "chat",
+      payload: {
+        sessionKey: "snapshot-session",
+        runId,
+        state: "delta",
+        message: {
+          content: [{ type: "text", text: "Hello world" }],
+        },
+      },
+    } as unknown as EventFrame);
+
+    await agent.handleGatewayEvent({
+      event: "chat",
+      payload: {
+        sessionKey: "snapshot-session",
+        runId,
+        state: "final",
+        stopReason: "end_turn",
+        message: {
+          content: [{ type: "text", text: "Hello world" }],
+        },
+      },
+    } as unknown as EventFrame);
+
+    await expect(promptPromise).resolves.toEqual({ stopReason: "end_turn" });
+    const chunks = sessionUpdate.mock.calls.filter(
+      (call: unknown[]) =>
+        (call[0] as Record<string, unknown>)?.update &&
+        (call[0] as Record<string, Record<string, unknown>>).update?.sessionUpdate ===
+          "agent_message_chunk",
+    );
+    expect(chunks).toHaveLength(1);
+    sessionStore.clearAllSessionsForTest();
+  });
+
+  it("emits only the missing tail when the final snapshot extends prior deltas", async () => {
+    const { agent, sessionUpdate, promptPromise, runId, sessionStore } =
+      await createSnapshotHarness();
+
+    await agent.handleGatewayEvent({
+      event: "chat",
+      payload: {
+        sessionKey: "snapshot-session",
+        runId,
+        state: "delta",
+        message: {
+          content: [{ type: "text", text: "Hello" }],
+        },
+      },
+    } as unknown as EventFrame);
+
+    await agent.handleGatewayEvent({
+      event: "chat",
+      payload: {
+        sessionKey: "snapshot-session",
+        runId,
+        state: "final",
+        stopReason: "max_tokens",
+        message: {
+          content: [{ type: "text", text: "Hello world" }],
+        },
+      },
+    } as unknown as EventFrame);
+
+    await expect(promptPromise).resolves.toEqual({ stopReason: "max_tokens" });
+    expect(sessionUpdate).toHaveBeenCalledWith({
+      sessionId: "snapshot-session",
+      update: {
+        sessionUpdate: "agent_message_chunk",
+        content: { type: "text", text: "Hello" },
+      },
+    });
+    expect(sessionUpdate).toHaveBeenCalledWith({
+      sessionId: "snapshot-session",
+      update: {
+        sessionUpdate: "agent_message_chunk",
+        content: { type: "text", text: " world" },
+      },
+    });
+    sessionStore.clearAllSessionsForTest();
+  });
+});

src/acp/translator.ts

@@ -800,9 +800,15 @@ export class AcpGatewayAgent implements Agent {
       return;
     }
 
-    if (state === "delta" && messageData) {
+    const shouldHandleMessageSnapshot = messageData && (state === "delta" || state === "final");
+    if (shouldHandleMessageSnapshot) {
+      // Gateway chat events can carry the latest full assistant snapshot on both
+      // incremental updates and the terminal final event. Process the snapshot
+      // first so ACP clients never drop the last visible assistant text.
       await this.handleDeltaEvent(pending.sessionId, messageData);
-      return;
+      if (state === "delta") {
+        return;
+      }
     }
 
     if (state === "final") {