From 9ea95e7a4307f62a2ae5faf00dc24235649864e3 Mon Sep 17 00:00:00 2001 From: iroiro147 Date: Sun, 2 Aug 2026 10:39:42 +0530 Subject: [PATCH] fix(client): skip replayed message.appended / message.completed when the step's text run is already done MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `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`. (#1507 describes this as the second half of the broader replay-idempotency class.) 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 — the `text → tool call → more text` multi-run pattern the reducer explicitly supports is unchanged. Complements the prior partial fix for #1507 (input.requested preservation). Together they cover both non-idempotent branches the issue identifies. Regression coverage in `message-reducer.test.ts`: - drives `message.appended("Hel") → message.appended("Hello") → message.completed("Hello")` twice, asserts a single done part survives - control test: different text on the same step still appends a new run Verified: - `pnpm exec vitest run --config vitest.unit.config.ts src/client` — 90/90 tests green - `pnpm exec tsc -p tsconfig.json --noEmit` — clean Refs #1507 (partial — covers the second reducer branch; the resume-from- stale-streamIndex fix is a useEveAgent hook concern and out of scope here) Signed-off-by: Sarthak Singh --- .changeset/swift-pandas-tap.md | 11 +++ .../eve/src/client/message-reducer.test.ts | 86 +++++++++++++++++++ packages/eve/src/client/message-reducer.ts | 19 ++++ 3 files changed, 116 insertions(+) create mode 100644 .changeset/swift-pandas-tap.md diff --git a/.changeset/swift-pandas-tap.md b/.changeset/swift-pandas-tap.md new file mode 100644 index 000000000..5d235e7c7 --- /dev/null +++ b/.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. diff --git a/packages/eve/src/client/message-reducer.test.ts b/packages/eve/src/client/message-reducer.test.ts index fe0886a1c..3a43aae16 100644 --- a/packages/eve/src/client/message-reducer.test.ts +++ b/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."]); + }); }); diff --git a/packages/eve/src/client/message-reducer.ts b/packages/eve/src/client/message-reducer.ts index a033a6085..4b0a9a65b 100644 --- a/packages/eve/src/client/message-reducer.ts +++ b/packages/eve/src/client/message-reducer.ts @@ -401,6 +401,14 @@ type EveRunPart = Extract= 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