fix(review-followup-fed): reconcile compaction-queued send acknowledgements

6 files changed, 84 insertions, 4 deletions

diff --git a/src/core/chunks/index.ts b/src/core/chunks/index.ts
index 3822a79..49381d5 100644
--- a/src/core/chunks/index.ts
+++ b/src/core/chunks/index.ts
@@ -7,6 +7,7 @@ export {
   clearGenerating,
   foldEvent,
   initialState,
+  removeLocalEcho,
 } from "./reducer";
 export type { ProviderRetryView } from "./retry-banner";
 export { formatRetryDelay, viewProviderRetry } from "./retry-banner";
diff --git a/src/core/chunks/local-echo.test.ts b/src/core/chunks/local-echo.test.ts
new file mode 100644
index 0000000..4b5eed4
--- /dev/null
+++ b/src/core/chunks/local-echo.test.ts
@@ -0,0 +1,45 @@
+import { describe, expect, it } from "vitest";
+import { appendUserMessage, applyHistory, foldEvent, initialState, removeLocalEcho } from "./index";
+
+describe("request-local optimistic echo", () => {
+  it("retains identity through turn-start tags without conflating identical sends", () => {
+    let state = appendUserMessage(initialState(), "same", [{ url: "data:first" }], "r1");
+    state = appendUserMessage(state, "same", [{ url: "data:second" }], "r2");
+    state = foldEvent(state, { type: "turn-start", conversationId: "c1", turnId: "t1" });
+    state = removeLocalEcho(state, "r1");
+    expect(state.provisional.map((p) => p.localRequestId)).toEqual(["r2", "r2"]);
+    expect(state.provisional.map((p) => p.chunk)).toEqual([
+      { type: "text", text: "same" },
+      { type: "image", url: "data:second" },
+    ]);
+    expect(state.generating).toBe(true);
+    expect(removeLocalEcho(state, "r1")).toBe(state);
+  });
+
+  it("a live user-message confirms only its deduplicated text, not every identical local send", () => {
+    let state = appendUserMessage(initialState(), "same", [{ url: "data:first" }], "r1");
+    state = appendUserMessage(state, "same", [{ url: "data:second" }], "r2");
+    state = foldEvent(state, { type: "turn-start", conversationId: "c1", turnId: "t1" });
+    state = foldEvent(state, {
+      type: "user-message",
+      conversationId: "c1",
+      turnId: "t1",
+      text: "same",
+    });
+    state = removeLocalEcho(removeLocalEcho(state, "r1"), "r2");
+    expect(state.provisional).toEqual([
+      { role: "user", turnId: "t1", chunk: { type: "text", text: "same" } },
+    ]);
+  });
+
+  it("does not remove committed rows even if their content is identical", () => {
+    let state = appendUserMessage(initialState(), "same", undefined, "r1");
+    state = applyHistory(state, [{ seq: 1, role: "user", chunk: { type: "text", text: "same" } }]);
+    const committed = state.committed;
+    state = removeLocalEcho(state, "r1");
+    expect(state.committed).toBe(committed);
+    expect(state.provisional).toEqual([]);
+    expect(removeLocalEcho(state, "r1")).toBe(state);
+    expect(removeLocalEcho(initialState(), "r1")).toEqual(initialState());
+  });
+});
diff --git a/src/core/chunks/reducer.ts b/src/core/chunks/reducer.ts
index 0af2ddd..53af9c1 100644
--- a/src/core/chunks/reducer.ts
+++ b/src/core/chunks/reducer.ts
@@ -246,11 +246,17 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState
       // text) emits no `user-message` and is not de-duped here.
       if (event.text.length === 0) return state;
       const run = trailingRun(state.provisional, "user");
-      const alreadyEchoed = run.some((p) => p.chunk.type === "text" && p.chunk.text === event.text);
-      if (alreadyEchoed) {
+      const echoed = run.find((p) => p.chunk.type === "text" && p.chunk.text === event.text);
+      if (echoed !== undefined) {
         return {
           ...state,
-          provisional: tagTrailingUserEcho(state.provisional, event.turnId),
+          // Preserve the live text independently of the local send. A delayed
+          // queued ack may remove its images, but cannot erase this server event.
+          provisional: tagTrailingUserEcho(state.provisional, event.turnId).map((p) => {
+            if (p.chunk !== echoed.chunk || p.localRequestId === undefined) return p;
+            const { localRequestId: _local, ...confirmed } = p;
+            return confirmed;
+          }),
           generating: true,
         };
       }
@@ -463,6 +469,7 @@ export function appendUserMessage(
   state: TranscriptState,
   text: string,
   images?: readonly ImageInput[],
+  localRequestId?: string,
 ): TranscriptState {
   const provisional = flushAccumulating(state.provisional, state.accumulating);
   const userChunks: Chunk[] = [];
@@ -484,7 +491,23 @@ export function appendUserMessage(
   }
   return {
     ...state,
-    provisional: [...provisional, ...userChunks.map((chunk) => ({ role: "user", chunk }) as const)],
+    provisional: [
+      ...provisional,
+      ...userChunks.map(
+        (chunk) =>
+          ({
+            role: "user",
+            chunk,
+            ...(localRequestId !== undefined ? { localRequestId } : {}),
+          }) as const,
+      ),
+    ],
     accumulating: null,
   };
 }
+
+/** Remove only still-local artifacts of one accepted queued send; never history/live text. */
+export function removeLocalEcho(state: TranscriptState, requestId: string): TranscriptState {
+  const provisional = state.provisional.filter((p) => p.localRequestId !== requestId);
+  return provisional.length === state.provisional.length ? state : { ...state, provisional };
+}
diff --git a/src/core/chunks/types.ts b/src/core/chunks/types.ts
index 949857a..f7bfe01 100644
--- a/src/core/chunks/types.ts
+++ b/src/core/chunks/types.ts
@@ -10,6 +10,8 @@ export interface AccumulatingChunk {
 
 /** A provisional chunk that has no authoritative seq yet. */
 export interface ProvisionalChunk {
+  /** Local send identity only; turn-start tagging does NOT confirm this content. */
+  readonly localRequestId?: string;
   readonly role: Role;
   readonly chunk: Chunk;
   /**
diff --git a/src/core/wire/conformance.test.ts b/src/core/wire/conformance.test.ts
index 1c5829b..1179fbc 100644
--- a/src/core/wire/conformance.test.ts
+++ b/src/core/wire/conformance.test.ts
@@ -174,6 +174,12 @@ describe("classifies every WsServerMessage type", () => {
         event: { type: "done" as const, conversationId: "c", turnId: "t", reason: "r" },
       },
       { type: "chat.error" as const, message: "e" },
+      {
+        type: "chat.queued" as const,
+        conversationId: "conv-1",
+        reason: "compaction" as const,
+        messageId: "queued-1",
+      },
       { type: "conversation.open" as const, conversationId: "c1", workspaceId: "w1" },
       {
         type: "conversation.statusChanged" as const,
@@ -198,6 +204,7 @@ describe("classifies every WsServerMessage type", () => {
       "error",
       "chat.delta",
       "chat.error",
+      "chat.queued",
       "conversation.open",
       "conversation.statusChanged",
       "conversation.compacted",
diff --git a/src/core/wire/conformance.ts b/src/core/wire/conformance.ts
index 9847081..78313f9 100644
--- a/src/core/wire/conformance.ts
+++ b/src/core/wire/conformance.ts
@@ -85,6 +85,8 @@ export function assertWsServerMessageExhaustive(msg: WsServerMessage): string {
       return "error";
     case "chat.delta":
       return "chat.delta";
+    case "chat.queued":
+      return "chat.queued";
     case "chat.error":
       return "chat.error";
     case "conversation.open":