fix(client): skip replayed message.appended / message.completed when the step's text run is already done

.changeset/swift-pandas-tap.md

@@ -0,0 +1,11 @@
+---
+"eve": patch
+---
+
+fix(client): skip replayed message.appended / message.completed when the step's text run is already done (#1507)
+
+`upsertRun` previously appended a fresh part whenever the latest same-step run was `done`, so a stale resume-stream cursor replaying past events could duplicate a completed text part for the same `stepIndex`. The reducer now checks: if the incoming snapshot's text is a prefix of the last done run's recorded text (including exact equality), the upsert is declined as a replay. New turns producing different text for the same step continue to append a new part (`text → tool call → more text` multi-run pattern, unchanged).
+
+Complements the prior partial fix in the same stream (#1507 input-requested preservation already shipped separately); together the reducer is idempotent against both replay classes the issue identifies.
+
+Includes regression coverage in `message-reducer.test.ts` driving `message.appended("Hel") → message.appended("Hello") → message.completed("Hello")` twice and asserting a single done part survives; plus a control test asserting different text on the same step still appends a new run.

packages/eve/src/client/message-reducer.test.ts

@@ -829,4 +829,90 @@ describe("defaultMessageReducer", () => {
     const userMessage = data.messages.find((message) => message.role === "user");
     expect(userMessage?.parts).toEqual([{ state: "done", text: "hello there", type: "text" }]);
   });
+
+  it("skips replayed message.appended / message.completed snapshots that are already done (#1507)", () => {
+    const reducer = defaultMessageReducer();
+    const stream: UnstampedMessageStreamEvent[] = [
+      createMessageAppendedEvent({
+        messageDelta: "Hel",
+        messageSoFar: "Hel",
+        sequence: 0,
+        stepIndex: 0,
+        turnId: "turn_1",
+      }),
+      createMessageAppendedEvent({
+        messageDelta: "lo",
+        messageSoFar: "Hello",
+        sequence: 1,
+        stepIndex: 0,
+        turnId: "turn_1",
+      }),
+      createMessageCompletedEvent({
+        finishReason: "stop",
+        message: "Hello",
+        sequence: 2,
+        stepIndex: 0,
+        turnId: "turn_1",
+      }),
+    ];
+
+    // One delivery: a single done text part with the final snapshot.
+    let data = reduceServerEvents(reducer, reducer.initial(), stream);
+    const initialTextParts = data.messages.flatMap((message) =>
+      message.parts.filter((part) => part.type === "text"),
+    );
+    expect(initialTextParts).toHaveLength(1);
+    expect(initialTextParts[0]).toMatchObject({
+      state: "done",
+      stepIndex: 0,
+      text: "Hello",
+    });
+
+    // Replay the same stream — the second pass must not append a second
+    // done text part for the same step. (The resume stream can sit behind
+    // the session cursor while a send opens a turn stream at a stale
+    // streamIndex; see #1507.)
+    data = reduceServerEvents(reducer, data, stream);
+    const replayedTextParts = data.messages.flatMap((message) =>
+      message.parts.filter((part) => part.type === "text"),
+    );
+    expect(replayedTextParts).toHaveLength(1);
+    expect(replayedTextParts[0]).toMatchObject({
+      state: "done",
+      stepIndex: 0,
+      text: "Hello",
+    });
+  });
+
+  it("still appends a new text run for the same step when the message differs", () => {
+    const reducer = defaultMessageReducer();
+    // First run: text part completes.
+    let data = reduceServerEvents(reducer, reducer.initial(), [
+      createMessageCompletedEvent({
+        finishReason: "stop",
+        message: "First response.",
+        sequence: 0,
+        stepIndex: 0,
+        turnId: "turn_1",
+      }),
+    ]);
+    // Second run for the same stepIndex: must NOT be dropped just because
+    // a "done" run exists — this is the multi-run pattern the reducer
+    // explicitly supports (text → tool call → more text).
+    data = reduceServerEvents(reducer, data, [
+      createMessageCompletedEvent({
+        finishReason: "stop",
+        message: "Second response.",
+        sequence: 1,
+        stepIndex: 0,
+        turnId: "turn_1",
+      }),
+    ]);
+
+    const textParts = data.messages.flatMap((message) =>
+      message.parts.filter((part) => part.type === "text"),
+    );
+    expect(textParts).toHaveLength(2);
+    expect(textParts.map((part) => part.text)).toEqual(["First response.", "Second response."]);
+  });
 });

packages/eve/src/client/message-reducer.ts

@@ -401,6 +401,14 @@ type EveRunPart = Extract<EveMessagePart, { readonly type: "text" | "reasoning"
 // We find the latest same-step run of this type: while it is still streaming,
 // its snapshots replace it in place; once it is done (or there is none), `next`
 // begins a new run appended in arrival order.
+//
+// Replay-safety (#1507): if the latest same-step run is already `done` and the
+// incoming snapshot is a prefix of its recorded text, the stream is replaying
+// events we've already consumed — decline the upsert. This covers the case
+// where `message.completed` marked the run done and a stale cursor replays
+// `message.appended` with the same (or shorter) snapshot. New turns producing
+// *different* text for the same stepIndex still append a new run, preserving
+// the multi-run pattern (text → tool call → more text).
 function upsertRun(message: EveAssistantMessage, next: EveRunPart): EveAssistantMessage {
   let lastIndex = -1;
   for (let index = message.parts.length - 1; index >= 0; index -= 1) {
@@ -411,6 +419,17 @@ function upsertRun(message: EveAssistantMessage, next: EveRunPart): EveAssistant
     }
   }
 
+  if (lastIndex !== -1) {
+    const last = message.parts[lastIndex] as EveRunPart;
+    // A done run's recorded text is the terminal snapshot. If `next` is a
+    // strict prefix of it, this is a replay of an earlier streaming event;
+    // the terminal state is already correct, so keep the message as-is.
+    if (last.state === "done" && next.text.length <= last.text.length
+        && last.text.startsWith(next.text)) {
+      return message;
+    }
+  }
+
   const openRun =
     lastIndex !== -1 && (message.parts[lastIndex] as EveRunPart).state === "streaming";
   const parts = openRun