fix(eve): correlate optimistic message deliveries (#3504)

Signed-off-by: owenkephart <owen.kephart@vercel.com>

.changeset/optimistic-delivery-correlation.md

@@ -0,0 +1,5 @@
+---
+"eve": patch
+---
+
+Reconcile frontend optimistic messages with their server delivery identities instead of stream order. Concurrent, coalesced, identical, and structured message submissions now resolve the correct placeholders, and separately delivered messages within one turn remain separate chat bubbles.

docs/guides/frontend/overview.mdx

@@ -157,7 +157,9 @@ change the option from `false` to `true` again, reset, or send a message to reco
 
 The hook keeps consuming one session stream across turns and idle periods. Replies,
 authorization updates, and turns started elsewhere all update the same projection, including
-background-task results that arrive during another send. Runtime-authored task input remains in the
+background-task results that arrive during another send. Separately delivered participant messages
+remain separate bubbles even when they steer the same active turn; a single event that coalesces
+multiple deliveries remains one bubble. Runtime-authored task input remains in the
 event stream with `data.kind: "execution.background_task"`, but the default reducer does not render
 it as a participant message. A turn settling does not close the connection. The transport reconnects from its cursor after a disconnect or the
 server's renewable 60-second lease; lease renewal does not start a turn or clear UI state.
@@ -322,7 +324,7 @@ const agent = useEveAgent({
 });
 ```
 
-The `optimistic` option (default `true`) projects submitted user messages into `data` before eve confirms them with a `message.received` event. These are reducer-facing projection events only. `events` stays the authoritative eve stream.
+The `optimistic` option (default `true`) projects submitted user messages into `data` before eve confirms them with a `message.received` event. These are reducer-facing projection events only. `events` stays the authoritative eve stream. For an existing session, the hook reconciles each placeholder with the server's delivery ID, so an unrelated or identical message cannot consume it.
 
 ## Custom reducer
 

packages/eve/src/client/eve-agent-projection.ts

@@ -26,6 +26,11 @@ export class EveAgentProjection<TData> {
     this.#data = this.#reducer.reduce(this.#data, event);
   }
 
+  remove(predicate: (event: EveAgentReducerEvent) => boolean): void {
+    this.#events = this.#events.filter((event) => !predicate(event));
+    this.#data = this.#reduce();
+  }
+
   replace(
     predicate: (event: EveAgentReducerEvent) => boolean,
     replacement: EveAgentReducerEvent,

packages/eve/src/client/eve-agent-store-helpers.ts

@@ -85,7 +85,7 @@ export function createActiveTurn(
     completion: completion.promise,
     followUpDispatches: new Set(),
     receivedFollowUps: 0,
-    receivedFollowUpEvents: new Set(),
+    receivedFollowUpEvents: new Map(),
     followUpSubmissionIds: new Set(),
     resolveCompletion: completion.resolve,
     response: response.promise,
@@ -105,7 +105,8 @@ export async function followSteeredTurns(
   if (turn.receivedFollowUps >= turn.acceptedFollowUps) return;
   for await (const event of events) {
     if (!isActive()) return;
-    if (turn.receivedFollowUpEvents.delete(event)) turn.receivedFollowUps += 1;
+    turn.receivedFollowUps += turn.receivedFollowUpEvents.get(event) ?? 0;
+    turn.receivedFollowUpEvents.delete(event);
     if (isCurrentTurnBoundaryEvent(event)) {
       while (turn.followUpDispatches.size > 0) {
         await Promise.allSettled(turn.followUpDispatches);

packages/eve/src/client/eve-agent-store-state.ts

@@ -66,8 +66,11 @@ export interface EveAgentStoreInit<TData> {
 
 export interface PendingMessageSubmission {
   readonly createdAt: number;
+  readonly eventStartIndex: number;
   readonly id: string;
   readonly message: string;
+  readonly requiresDeliveryId: boolean;
+  readonly deliveryId?: string;
 }
 
 export interface ActiveTurn {
@@ -77,7 +80,7 @@ export interface ActiveTurn {
   readonly completion: Promise<void>;
   readonly followUpDispatches: Set<Promise<void>>;
   receivedFollowUps: number;
-  readonly receivedFollowUpEvents: Set<MessageStreamEvent>;
+  readonly receivedFollowUpEvents: Map<MessageStreamEvent, number>;
   readonly followUpSubmissionIds: Set<string>;
   readonly resolveCompletion: () => void;
   readonly response: Promise<MessageResponse | undefined>;

packages/eve/src/client/eve-agent-store.test.ts

@@ -443,6 +443,64 @@ describe("EveAgentStore prewarming", () => {
     expect(store.snapshot.session?.streamIndex).toBe(6);
   });
 
+  it("projects a background turn that arrives while another message is being accepted", async () => {
+    const live = controlledStreamResponse();
+    const accepted = Promise.withResolvers<Response>();
+    const fetchMock = vi
+      .spyOn(globalThis, "fetch")
+      .mockResolvedValueOnce(live.response)
+      .mockReturnValueOnce(accepted.promise);
+    const store = createStore({
+      initialSession: { sessionId: "session_1", streamIndex: 0 },
+      reducer: defaultMessageReducer(),
+    });
+    const sending = store.send({ message: "Next question" });
+    await vi.waitFor(() => expect(fetchMock).toHaveBeenCalledTimes(2));
+
+    const background = stampTestEvents([
+      createTurnStartedEvent({ sequence: 0, turnId: "turn_background" }),
+      createMessageCompletedEvent({
+        finishReason: "stop",
+        message: "Background result",
+        sequence: 1,
+        stepIndex: 0,
+        turnId: "turn_background",
+      }),
+      createSessionWaitingEvent(),
+    ]).map((event) => ({
+      ...event,
+      meta: { ...event.meta, deliveryIds: ["background-delivery"] },
+    }));
+    for (const event of background) live.emit(event);
+    await vi.waitFor(() => expect(store.snapshot.events).toHaveLength(3));
+    expect(store.snapshot.data.messages.some((message) => message.metadata?.optimistic)).toBe(true);
+
+    accepted.resolve(startedResponse("message-delivery"));
+    const messageTurn = turnEvents().map((event) => ({
+      ...event,
+      meta: {
+        ...event.meta,
+        deliveryIds: ["message-delivery"],
+        id: `message-${event.meta.id}`,
+      },
+    }));
+    for (const event of messageTurn) live.emit(event);
+    await vi.waitFor(() => expect(store.snapshot.events).toHaveLength(6));
+    await sending;
+
+    expect(store.snapshot.events).toEqual([...background, ...messageTurn]);
+    expect(store.snapshot.data.messages.some((message) => message.metadata?.optimistic)).toBe(
+      false,
+    );
+    expect(store.snapshot.data.messages).toEqual(
+      expect.arrayContaining([
+        expect.objectContaining({ role: "user" }),
+        expect.objectContaining({ role: "assistant" }),
+      ]),
+    );
+    expect(fetchMock).toHaveBeenCalledTimes(2);
+  });
+
   it("honors a send's disabled reconnect policy on an existing prewarmed stream", async () => {
     const live = controlledStreamResponse();
     const fetchMock = vi
@@ -755,7 +813,9 @@ describe("EveAgentStore session resume", () => {
           turnId: "turn_0",
         }),
         createSessionWaitingEvent(),
-      ]);
+      ]).map((event, index) =>
+        index < 2 ? event : { ...event, meta: { ...event.meta, deliveryIds: ["delivery_1"] } },
+      );
       const live = controlledStreamResponse();
       live.response.headers.set("x-eve-stream-tail-index", "1");
       live.emit(events[0]!);
@@ -1305,6 +1365,55 @@ describe("EveAgentStore steering", () => {
     },
   );
 
+  it("settles steering whose delivery event arrives before its response", async () => {
+    const activeStream = controlledStreamResponse();
+    const steeringAccepted = Promise.withResolvers<Response>();
+    const [firstReceived, firstStarted, steeringReceived, completed, waiting] = stampTestEvents([
+      createMessageReceivedEvent({ message: "First", sequence: 0, turnId: "turn_1" }),
+      createTurnStartedEvent({ sequence: 1, turnId: "turn_1" }),
+      createMessageReceivedEvent({ message: "Instead", sequence: 2, turnId: "turn_1" }),
+      createMessageCompletedEvent({
+        finishReason: "stop",
+        message: "Steered reply.",
+        sequence: 3,
+        stepIndex: 0,
+        turnId: "turn_1",
+      }),
+      createSessionWaitingEvent(),
+    ] as UnstampedMessageStreamEvent[]).map((event, index) => ({
+      ...event,
+      meta: {
+        ...event.meta,
+        deliveryIds: [index === 2 ? "steering-delivery" : "first-delivery"],
+      },
+    }));
+    const fetchMock = vi
+      .spyOn(globalThis, "fetch")
+      .mockResolvedValueOnce(startedResponse("first-delivery"))
+      .mockResolvedValueOnce(activeStream.response)
+      .mockReturnValueOnce(steeringAccepted.promise);
+    const store = createStore({ reducer: defaultMessageReducer() });
+
+    const firstSend = store.send({ message: "First" });
+    await vi.waitFor(() => expect(fetchMock).toHaveBeenCalledTimes(2));
+    activeStream.emit(firstReceived!);
+    activeStream.emit(firstStarted!);
+
+    const steering = store.send({ message: "Instead", turnPolicy: "steer" });
+    await vi.waitFor(() => expect(fetchMock).toHaveBeenCalledTimes(3));
+    activeStream.emit(steeringReceived!);
+    activeStream.emit(completed!);
+    activeStream.emit(waiting!);
+    await vi.waitFor(() => expect(store.snapshot.events).toHaveLength(5));
+    steeringAccepted.resolve(startedResponse("steering-delivery"));
+
+    await Promise.all([firstSend, steering]);
+    expect(store.snapshot.status).toBe("ready");
+    expect(store.snapshot.data.messages.some((message) => message.metadata?.optimistic)).toBe(
+      false,
+    );
+  });
+
   it("follows a late steering delivery after the active turn settles", async () => {
     const activeStream = controlledStreamResponse();
     const [

packages/eve/src/client/eve-agent-store.ts

@@ -5,16 +5,20 @@ import type {
   EveAgentStoreInit,
   EveAgentStoreSnapshot,
   EveAgentStoreStatus,
-  PendingMessageSubmission,
   PrepareSend,
 } from "#client/eve-agent-store-state.js";
-import { consumeMessageResponse, type MessageResponse } from "#client/message-response.js";
+import {
+  consumeMessageResponse,
+  getMessageResponseDeliveryId,
+  type MessageResponse,
+} from "#client/message-response.js";
 import {
   SessionEventStream,
   type SessionEventReader,
   type SessionEventStreamOptions,
 } from "#client/session-event-stream.js";
 import { EveAgentProjection } from "#client/eve-agent-projection.js";
+import { OptimisticMessageSubmissions } from "#client/optimistic-message-submissions.js";
 import type { ClientSession } from "#client/session.js";
 import { createEventDeduper } from "#protocol/event-dedupe.js";
 import { isCurrentTurnBoundaryEvent, type MessageStreamEvent } from "#protocol/message.js";
@@ -23,10 +27,8 @@ import {
   createAbortSignal,
   createActiveTurn,
   followSteeredTurns,
-  createSubmissionId,
   isAbortError,
   isSettledSessionTail,
-  summarizeUserContent,
   toTerminalStreamFailureError,
   waitWithSignal,
 } from "#client/eve-agent-store-helpers.js";
@@ -68,7 +70,7 @@ export class EveAgentStore<TData> {
   #callbacks: EveAgentStoreCallbacks<TData> = {};
   #error: Error | undefined;
   #events: readonly MessageStreamEvent[];
-  #pendingMessageSubmissions: readonly PendingMessageSubmission[] = [];
+  readonly #messageSubmissions: OptimisticMessageSubmissions<TData>;
   #prewarmGeneration = 0;
   #prewarmPromise: Promise<void> | undefined;
   #prewarmController: AbortController | undefined;
@@ -97,6 +99,7 @@ export class EveAgentStore<TData> {
     this.#events = initialEvents;
     this.#projection = new EveAgentProjection(init.reducer, this.#events);
     this.#optimistic = init.optimistic ?? true;
+    this.#messageSubmissions = new OptimisticMessageSubmissions(this.#projection, this.#optimistic);
     this.#session =
       init.session ??
       (init.initialSession === undefined
@@ -218,7 +221,7 @@ export class EveAgentStore<TData> {
         return;
       }
 
-      this.#projectOptimisticMessage(preparedInput);
+      const submissionId = this.#messageSubmissions.submit(preparedInput, this.#events.length);
       this.#projectInputResponses(preparedInput);
       this.#publish();
 
@@ -233,9 +236,21 @@ export class EveAgentStore<TData> {
       if (!this.#isActiveTurn(turn)) return;
       turn.resolveResponse(response);
 
+      if (
+        this.#handleReconciliation(
+          this.#messageSubmissions.correlate(
+            submissionId,
+            getMessageResponseDeliveryId(response),
+            this.#events,
+          ),
+        )
+      ) {
+        this.#publish();
+      }
       for await (const event of consumeMessageResponse(response, reader)) {
         if (!this.#isActiveTurn(turn)) return;
-        if (turn.receivedFollowUpEvents.delete(event)) turn.receivedFollowUps += 1;
+        turn.receivedFollowUps += turn.receivedFollowUpEvents.get(event) ?? 0;
+        turn.receivedFollowUpEvents.delete(event);
       }
 
       if (!this.#isActiveTurn(turn)) {
@@ -252,12 +267,12 @@ export class EveAgentStore<TData> {
 
       if (isAbortError(error)) {
         this.#status = "ready";
-        this.#failPendingMessageSubmission(toError(error));
+        this.#messageSubmissions.fail(toError(error));
       } else {
         const reported = this.#error !== undefined;
         this.#error ??= toError(error);
         this.#status = "error";
-        this.#failPendingMessageSubmission(this.#error);
+        this.#messageSubmissions.fail(this.#error);
         if (!reported) this.#callbacks.onError?.(this.#error);
       }
     } finally {
@@ -316,7 +331,8 @@ export class EveAgentStore<TData> {
         this.#publish();
         for await (const event of reader) {
           if (!this.#isActiveTurn(turn)) return;
-          if (turn.receivedFollowUpEvents.delete(event)) turn.receivedFollowUps += 1;
+          turn.receivedFollowUps += turn.receivedFollowUpEvents.get(event) ?? 0;
+          turn.receivedFollowUpEvents.delete(event);
           if (isCurrentTurnBoundaryEvent(event) && this.#pendingAuthorizations.size === 0) break;
         }
       }
@@ -380,7 +396,7 @@ export class EveAgentStore<TData> {
     if (!this.#externalSession) this.#session = undefined;
     this.#events = [];
     this.#seenEvents = createEventDeduper();
-    this.#pendingMessageSubmissions = [];
+    this.#messageSubmissions.reset();
     this.#projection.reset();
     this.#error = undefined;
     this.#status = "ready";
@@ -411,7 +427,7 @@ export class EveAgentStore<TData> {
     }
     if (!this.#isActiveTurn(turn)) return await this.#submit(preparedInput);
 
-    const submissionId = this.#projectOptimisticMessage(preparedInput);
+    const submissionId = this.#messageSubmissions.submit(preparedInput, this.#events.length);
     if (submissionId !== undefined) turn.followUpSubmissionIds.add(submissionId);
     this.#publish();
     this.#ensureStream({
@@ -428,11 +444,22 @@ export class EveAgentStore<TData> {
           throw new Error("The active eve turn ended before the follow-up could be sent.");
         }
         const { message, ...options } = preparedInput;
-        await this.#session.send(message, { ...options, signal });
+        const response = await this.#session.send(message, { ...options, signal });
         turn.acceptedFollowUps += 1;
+        if (
+          this.#handleReconciliation(
+            this.#messageSubmissions.correlate(
+              submissionId,
+              getMessageResponseDeliveryId(response),
+              this.#events,
+            ),
+          )
+        ) {
+          this.#publish();
+        }
       } catch (error) {
         if (this.#isActiveTurn(turn)) {
-          this.#failPendingMessageSubmission(toError(error), submissionId);
+          this.#messageSubmissions.fail(toError(error), submissionId);
           this.#publish();
         }
         throw error;
@@ -547,30 +574,6 @@ export class EveAgentStore<TData> {
     return this.#activeTurn === turn;
   }
 
-  #projectOptimisticMessage(input: SendTurnPayload): string | undefined {
-    if (input.message === undefined) {
-      return undefined;
-    }
-
-    const id = createSubmissionId();
-    const pending = {
-      createdAt: Date.now(),
-      id,
-      message: summarizeUserContent(input.message),
-    };
-    this.#pendingMessageSubmissions = [...this.#pendingMessageSubmissions, pending];
-    if (this.#optimistic)
-      this.#projection.append({
-        data: {
-          createdAt: pending.createdAt,
-          message: pending.message,
-          submissionId: pending.id,
-        },
-        type: "client.message.submitted",
-      });
-    return id;
-  }
-
   #projectInputResponses(input: SendTurnPayload): void {
     if (input.inputResponses === undefined || input.inputResponses.length === 0) {
       return;
@@ -590,7 +593,7 @@ export class EveAgentStore<TData> {
     const wasStreaming = this.#status === "streaming";
     updatePendingAuthorizations(this.#pendingAuthorizations, event);
     this.#events = [...this.#events, event];
-    this.#applyServerEvent(event);
+    this.#handleReconciliation(this.#messageSubmissions.apply(event));
     this.#callbacks.onEvent?.(event);
     this.#applyTerminalStreamFailure(event);
     const settled = isCurrentTurnBoundaryEvent(event) && this.#pendingAuthorizations.size === 0;
@@ -605,23 +608,23 @@ export class EveAgentStore<TData> {
     }
   }
 
-  #applyServerEvent(event: MessageStreamEvent): void {
+  #handleReconciliation(
-    const pendingSubmission = this.#pendingMessageSubmissions[0];
+    reconciliation: ReturnType<OptimisticMessageSubmissions<TData>["apply"]>,
-    if (event.type === "message.received" && pendingSubmission !== undefined) {
+  ): boolean {
-      const submissionId = pendingSubmission.id;
+    if (reconciliation === undefined) return false;
-      if (this.#activeTurn?.followUpSubmissionIds.delete(submissionId))
+    let followed = 0;
-        this.#activeTurn.receivedFollowUpEvents.add(event);
+    for (const id of reconciliation.ids) {
-      this.#pendingMessageSubmissions = this.#pendingMessageSubmissions.slice(1);
+      if (this.#activeTurn?.followUpSubmissionIds.delete(id)) followed += 1;
-      this.#projection.replace(
-        (candidate) =>
-          candidate.type === "client.message.submitted" &&
-          candidate.data.submissionId === submissionId,
-        event,
-      );
-      return;
     }
-
+    if (followed > 0 && this.#activeTurn !== undefined) {
-    this.#projection.append(event);
+      if (reconciliation.alreadyProjected) {
+        this.#activeTurn.receivedFollowUps += followed;
+      } else {
+        const previous = this.#activeTurn.receivedFollowUpEvents.get(reconciliation.event) ?? 0;
+        this.#activeTurn.receivedFollowUpEvents.set(reconciliation.event, previous + followed);
+      }
+    }
+    return true;
   }
 
   #applyTerminalStreamFailure(event: MessageStreamEvent): void {
@@ -631,7 +634,7 @@ export class EveAgentStore<TData> {
     }
 
     this.#status = "error";
-    this.#failPendingMessageSubmission(error);
+    this.#messageSubmissions.failAll(error);
 
     if (this.#error === undefined) {
       this.#error = error;
@@ -639,33 +642,6 @@ export class EveAgentStore<TData> {
     }
   }
 
-  #failPendingMessageSubmission(error: Error, submissionId?: string): void {
-    const pending =
-      submissionId === undefined
-        ? this.#pendingMessageSubmissions[0]
-        : this.#pendingMessageSubmissions.find((candidate) => candidate.id === submissionId);
-    if (pending === undefined) return;
-
-    this.#pendingMessageSubmissions = this.#pendingMessageSubmissions.filter(
-      (candidate) => candidate.id !== pending.id,
-    );
-    this.#projection.replace(
-      (event) =>
-        event.type === "client.message.submitted" && event.data.submissionId === pending.id,
-      {
-        data: {
-          createdAt: pending.createdAt,
-          error: {
-            message: error.message,
-          },
-          message: pending.message,
-          submissionId: pending.id,
-        },
-        type: "client.message.failed",
-      },
-    );
-  }
-
   #createSnapshot(): EveAgentStoreSnapshot<TData> {
     return {
       data: this.#projection.data,

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

@@ -13,12 +13,14 @@ import {
   createInputRequestedEvent,
   createMessageAppendedEvent,
   createMessageCompletedEvent,
+  createMessageReceivedEvent,
   createReasoningAppendedEvent,
   createReasoningCompletedEvent,
   createResultCompletedEvent,
   createStepStartedEvent,
   createTurnCancelledEvent,
   createTurnFailedEvent,
+  type MessageStreamEvent,
   type UnstampedMessageStreamEvent,
 } from "#protocol/message.js";
 
@@ -1265,6 +1267,72 @@ describe("defaultMessageReducer", () => {
     ]);
   });
 
+  it("preserves separate participant messages received within one turn", () => {
+    const reducer = defaultMessageReducer();
+    const events = stampTestEvents([
+      createMessageReceivedEvent({ message: "test message", sequence: 0, turnId: "turn_1" }),
+      createMessageReceivedEvent({ message: "a", sequence: 1, turnId: "turn_1" }),
+    ]).map((event, index) => ({
+      ...event,
+      meta: { ...event.meta, deliveryIds: [`delivery_${index}`] },
+    }));
+    const reduce = () =>
+      events.reduce((data, event) => reducer.reduce(data, event), reducer.initial());
+
+    const data = reduce();
+    expect(data.messages.map((message) => message.id)).toEqual(
+      events.map((event) => `${event.meta.id}:user`),
+    );
+    expect(data.messages.map((message) => message.parts)).toEqual([
+      [{ state: "done", text: "test message", type: "text" }],
+      [{ state: "done", text: "a", type: "text" }],
+    ]);
+    expect(reduce().messages.map((message) => message.id)).toEqual(
+      data.messages.map((message) => message.id),
+    );
+  });
+
+  it("projects one bubble for one coalesced participant event", () => {
+    const reducer = defaultMessageReducer();
+    const [event] = stampTestEvents([
+      createMessageReceivedEvent({ message: "first\n\nsecond", sequence: 0, turnId: "turn_1" }),
+    ]).map((candidate) => ({
+      ...candidate,
+      meta: { ...candidate.meta, deliveryIds: ["delivery_1", "delivery_2"] },
+    }));
+    const data = reducer.reduce(reducer.initial(), event!);
+
+    expect(data.messages).toHaveLength(1);
+    expect(data.messages[0]?.parts).toEqual([
+      { state: "done", text: "first\n\nsecond", type: "text" },
+    ]);
+  });
+
+  it("uses a stable fallback id for legacy received events", () => {
+    const reducer = defaultMessageReducer();
+    const event = {
+      ...createMessageReceivedEvent({ message: "legacy", sequence: 2, turnId: "turn_1" }),
+      meta: { at: "2026-07-27T18:04:11.912Z" },
+    } as MessageStreamEvent;
+
+    const data = reducer.reduce(reducer.initial(), event);
+    expect(data.messages[0]?.id).toBe("turn_1:2:user");
+  });
+
+  it("does not project framework-authored task input", () => {
+    const reducer = defaultMessageReducer();
+    const [event] = stampTestEvents([
+      createMessageReceivedEvent({
+        kind: "execution.background_task",
+        message: "Task completed",
+        sequence: 1,
+        turnId: "turn_1",
+      }),
+    ]);
+
+    expect(reducer.reduce(reducer.initial(), event!).messages).toEqual([]);
+  });
+
   it("projects structured file parts from message.received onto the user message", () => {
     const reducer = defaultMessageReducer();
     const data = reduceServerEvents(reducer, reducer.initial(), [

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

@@ -45,6 +45,12 @@ export type {
 } from "#client/message-reducer-types.js";
 
 type EveAssistantMessage = EveMessage & { readonly role: "assistant" };
+type MessageReceivedEvent = Extract<EveAgentReducerEvent, { readonly type: "message.received" }>;
+
+function receivedMessageEventId(event: MessageReceivedEvent): string {
+  const eventId: string | undefined = event.meta.id;
+  return eventId ?? `${event.data.turnId}:${event.data.sequence}`;
+}
 
 /**
  * Creates a UIMessage-compatible eve reducer for chat and agent UIs.
@@ -108,7 +114,7 @@ function reduceMessageData(data: EveMessageData, event: EveAgentReducerEvent): E
     case "message.received":
       if (event.data.kind === "execution.background_task") return data;
       return upsertMessage(data, {
-        id: `${event.data.turnId}:user`,
+        id: `${receivedMessageEventId(event)}:user`,
         metadata: {
           status: "complete",
           turnId: event.data.turnId,

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

@@ -11,10 +11,12 @@ interface MessageResponseInput {
   readonly createStream: (
     source?: AsyncIterable<MessageStreamEvent>,
   ) => AsyncGenerator<MessageStreamEvent>;
+  readonly deliveryId?: string;
   readonly sessionId: string;
 }
 
 const consumeResponse = Symbol("consumeMessageResponse");
+const acceptedDeliveryId = Symbol("acceptedDeliveryId");
 
 /**
  * The response from {@link ClientSession.send}.
@@ -28,6 +30,7 @@ export class MessageResponse<TOutput = unknown> implements AsyncIterable<Message
    * Session ID assigned by the server.
    */
   readonly sessionId: string;
+  readonly [acceptedDeliveryId]: string | undefined;
 
   readonly #cancelTurn: (turnId: string) => Promise<CancelSessionResult>;
   #cancellation: Promise<CancelSessionResult> | undefined;
@@ -40,6 +43,7 @@ export class MessageResponse<TOutput = unknown> implements AsyncIterable<Message
   constructor(input: MessageResponseInput) {
     this.#cancelTurn = input.cancelTurn;
     this.sessionId = input.sessionId;
+    this[acceptedDeliveryId] = input.deliveryId;
     this.#createStream = input.createStream;
   }
 
@@ -124,6 +128,11 @@ export class MessageResponse<TOutput = unknown> implements AsyncIterable<Message
   }
 }
 
+/** @internal Returns the accepted message identity without consuming its response stream. */
+export function getMessageResponseDeliveryId(response: MessageResponse): string | undefined {
+  return response[acceptedDeliveryId];
+}
+
 /** @internal Observe a turn through the frontend's existing session stream. */
 export function consumeMessageResponse(
   response: MessageResponse,

packages/eve/src/client/optimistic-message-submissions.test.ts

@@ -0,0 +1,126 @@
+import { describe, expect, it } from "vitest";
+import { EveAgentProjection } from "#client/eve-agent-projection.js";
+import { defaultMessageReducer } from "#client/message-reducer.js";
+import { OptimisticMessageSubmissions } from "#client/optimistic-message-submissions.js";
+import { stampTestEvents } from "#internal/testing/events.js";
+import { createMessageReceivedEvent, type MessageStreamEvent } from "#protocol/message.js";
+
+function received(
+  message: string,
+  deliveryIds: readonly string[],
+): Extract<MessageStreamEvent, { readonly type: "message.received" }> {
+  const event = createMessageReceivedEvent({
+    message,
+    sequence: 0,
+    turnId: `turn_${deliveryIds[0]}`,
+  });
+  return {
+    ...event,
+    meta: { at: new Date().toISOString(), deliveryIds, id: `event_${deliveryIds[0]}` },
+  };
+}
+
+function setup(optimistic = true) {
+  const projection = new EveAgentProjection(defaultMessageReducer(), []);
+  return { projection, submissions: new OptimisticMessageSubmissions(projection, optimistic) };
+}
+
+describe("OptimisticMessageSubmissions", () => {
+  it("reconciles by delivery identity rather than message text", () => {
+    const { projection, submissions } = setup();
+    const id = submissions.submit({ message: "Same text" }, 0)!;
+    const other = received("Same text", ["other"]);
+    submissions.apply(other);
+    expect(projection.data.messages.filter((message) => message.metadata?.optimistic)).toHaveLength(
+      1,
+    );
+
+    const own = received("Same text", ["mine"]);
+    submissions.apply(own);
+    submissions.correlate(id, "mine", [other, own]);
+
+    expect(projection.data.messages.filter((message) => message.role === "user")).toHaveLength(2);
+    expect(projection.data.messages.some((message) => message.metadata?.optimistic)).toBe(false);
+  });
+
+  it("does not reconcile framework-authored task input", () => {
+    const { projection, submissions } = setup();
+    const id = submissions.submit({ message: "Hello" }, 0)!;
+    submissions.correlate(id, "mine", []);
+    const taskWake = received("Task completed", ["mine"]);
+    submissions.apply({
+      ...taskWake,
+      data: { ...taskWake.data, kind: "execution.background_task" },
+    });
+
+    expect(projection.data.messages).toHaveLength(1);
+    expect(projection.data.messages[0]?.metadata?.optimistic).toBe(true);
+  });
+
+  it("reconciles events that arrive before the POST response", () => {
+    const { projection, submissions } = setup();
+    const id = submissions.submit({ message: "Hello" }, 0)!;
+    const event = received("Hello", ["mine"]);
+    submissions.apply(event);
+
+    expect(projection.data.messages.some((message) => message.metadata?.optimistic)).toBe(true);
+    submissions.correlate(id, "mine", [event]);
+    expect(projection.data.messages).toHaveLength(1);
+    expect(projection.data.messages[0]?.metadata?.optimistic).toBeUndefined();
+  });
+
+  it.each([true, false])("folds coalesced deliveries once (optimistic=%s)", (optimistic) => {
+    const { projection, submissions } = setup(optimistic);
+    const first = submissions.submit({ message: "First" }, 0)!;
+    const second = submissions.submit({ message: "Second" }, 0)!;
+    submissions.correlate(first, "first", []);
+    submissions.correlate(second, "second", []);
+
+    submissions.apply(received("First\n\nSecond", ["first", "second"]));
+
+    expect(projection.data.messages).toHaveLength(1);
+    expect(projection.data.messages[0]?.parts).toContainEqual({
+      state: "done",
+      text: "First\n\nSecond",
+      type: "text",
+    });
+  });
+
+  it("reconciles structured file input without comparing summaries", () => {
+    const { projection, submissions } = setup();
+    const input = {
+      message: [
+        { text: "Review", type: "text" as const },
+        {
+          data: "data:text/plain;base64,SGVsbG8=",
+          filename: "note.txt",
+          mediaType: "text/plain",
+          type: "file" as const,
+        },
+      ],
+    };
+    const id = submissions.submit(input, 0)!;
+    submissions.correlate(id, "mine", []);
+    submissions.apply(
+      stampTestEvents([
+        createMessageReceivedEvent({ message: input.message, sequence: 0, turnId: "turn_mine" }),
+      ]).map((event) => ({ ...event, meta: { ...event.meta, deliveryIds: ["mine"] } }))[0]!,
+    );
+
+    expect(projection.data.messages).toHaveLength(1);
+    expect(projection.data.messages[0]?.parts).toMatchObject([
+      { state: "done", text: "Review", type: "text" },
+      { filename: "note.txt", mediaType: "text/plain", type: "file" },
+    ]);
+  });
+
+  it("uses the first message boundary only for a newly created session", () => {
+    const { projection, submissions } = setup();
+    const id = submissions.submit({ message: "Hello" }, 0);
+    const event = received("Hello", []);
+    submissions.apply(event);
+    submissions.correlate(id, undefined, [event]);
+    expect(projection.data.messages).toHaveLength(1);
+    expect(projection.data.messages[0]?.metadata?.optimistic).toBeUndefined();
+  });
+});

packages/eve/src/client/optimistic-message-submissions.ts

@@ -0,0 +1,149 @@
+import { EveAgentProjection } from "#client/eve-agent-projection.js";
+import type { PendingMessageSubmission } from "#client/eve-agent-store-state.js";
+import { createSubmissionId, summarizeUserContent } from "#client/eve-agent-store-helpers.js";
+import type { MessageStreamEvent } from "#protocol/message.js";
+import type { SendTurnPayload } from "#client/types.js";
+
+interface ReconciledSubmissions {
+  readonly alreadyProjected: boolean;
+  readonly event: Extract<MessageStreamEvent, { readonly type: "message.received" }>;
+  readonly ids: readonly string[];
+}
+
+/** Owns optimistic message projection and server-delivery reconciliation. */
+export class OptimisticMessageSubmissions<TData> {
+  readonly #optimistic: boolean;
+  readonly #projection: EveAgentProjection<TData>;
+  #pending: readonly PendingMessageSubmission[] = [];
+
+  constructor(projection: EveAgentProjection<TData>, optimistic: boolean) {
+    this.#projection = projection;
+    this.#optimistic = optimistic;
+  }
+
+  reset(): void {
+    this.#pending = [];
+  }
+
+  submit(input: SendTurnPayload, eventStartIndex: number): string | undefined {
+    if (input.message === undefined) return undefined;
+    const pending = {
+      createdAt: Date.now(),
+      eventStartIndex,
+      id: createSubmissionId(),
+      message: summarizeUserContent(input.message),
+      requiresDeliveryId: true,
+    };
+    this.#pending = [...this.#pending, pending];
+    if (this.#optimistic) {
+      this.#projection.append({
+        data: { createdAt: pending.createdAt, message: pending.message, submissionId: pending.id },
+        type: "client.message.submitted",
+      });
+    }
+    return pending.id;
+  }
+
+  apply(event: MessageStreamEvent): ReconciledSubmissions | undefined {
+    if (event.type !== "message.received") {
+      this.#projection.append(event);
+      return undefined;
+    }
+    if (event.data.kind === "execution.background_task") {
+      this.#projection.append(event);
+      return undefined;
+    }
+    const matching = this.#matching(event);
+    if (matching.length === 0) {
+      this.#projection.append(event);
+      return undefined;
+    }
+    return this.#reconcile(matching, event, false);
+  }
+
+  correlate(
+    submissionId: string | undefined,
+    deliveryId: string | undefined,
+    events: readonly MessageStreamEvent[],
+  ): ReconciledSubmissions | undefined {
+    if (submissionId === undefined) return undefined;
+    this.#pending = this.#pending.map((pending) =>
+      pending.id === submissionId
+        ? { ...pending, deliveryId, requiresDeliveryId: deliveryId !== undefined }
+        : pending,
+    );
+    const pending = this.#pending.find((candidate) => candidate.id === submissionId);
+    if (pending === undefined) return undefined;
+    for (const event of events.slice(pending.eventStartIndex)) {
+      if (event.type !== "message.received" || event.data.kind === "execution.background_task") {
+        continue;
+      }
+      const matching = this.#matching(event);
+      if (matching.some((candidate) => candidate.id === submissionId)) {
+        return this.#reconcile(matching, event, true);
+      }
+    }
+    return undefined;
+  }
+
+  fail(error: Error, submissionId?: string): void {
+    const pending =
+      submissionId === undefined
+        ? this.#pending[0]
+        : this.#pending.find((candidate) => candidate.id === submissionId);
+    if (pending === undefined) return;
+    this.#pending = this.#pending.filter((candidate) => candidate.id !== pending.id);
+    this.#projection.replace(
+      (event) =>
+        event.type === "client.message.submitted" && event.data.submissionId === pending.id,
+      {
+        data: {
+          createdAt: pending.createdAt,
+          error: { message: error.message },
+          message: pending.message,
+          submissionId: pending.id,
+        },
+        type: "client.message.failed",
+      },
+    );
+  }
+
+  failAll(error: Error): void {
+    for (const pending of this.#pending) this.fail(error, pending.id);
+  }
+
+  #matching(event: Extract<MessageStreamEvent, { readonly type: "message.received" }>) {
+    return this.#pending.filter((pending) =>
+      pending.deliveryId === undefined
+        ? !pending.requiresDeliveryId
+        : event.meta.deliveryIds?.includes(pending.deliveryId) === true,
+    );
+  }
+
+  #reconcile(
+    submissions: readonly PendingMessageSubmission[],
+    event: Extract<MessageStreamEvent, { readonly type: "message.received" }>,
+    alreadyProjected: boolean,
+  ): ReconciledSubmissions {
+    const ids = submissions.map((pending) => pending.id);
+    const idSet = new Set(ids);
+    this.#pending = this.#pending.filter((pending) => !idSet.has(pending.id));
+    if (alreadyProjected) {
+      this.#projection.remove(
+        (candidate) =>
+          candidate.type === "client.message.submitted" && idSet.has(candidate.data.submissionId),
+      );
+    } else {
+      this.#projection.replace(
+        (candidate) =>
+          candidate.type === "client.message.submitted" && candidate.data.submissionId === ids[0],
+        event,
+      );
+      this.#projection.remove(
+        (candidate) =>
+          candidate.type === "client.message.submitted" && idSet.has(candidate.data.submissionId),
+      );
+    }
+    return { alreadyProjected, event, ids };
+  }
+}

packages/eve/src/client/session.ts

@@ -221,6 +221,7 @@ export class ClientSession {
       cancelTurn: async (turnId) => await this.cancel({ turnId }),
       createStream: (source) =>
         this.#createEventStream(initialStreamIndex, input, deliveryId, source),
+      deliveryId,
       sessionId: this.#state.sessionId,
     });
   }

packages/eve/src/react/use-eve-agent.test.ts

@@ -98,7 +98,7 @@ function completedTurnData(input: {
   return {
     messages: [
       {
-        id: `${input.turnId}:user`,
+        id: expect.stringMatching(/^evt_.+:user$/),
         metadata: {
           status: "complete",
           turnId: input.turnId,

packages/eve/src/vue/use-eve-agent.test.ts

@@ -78,7 +78,7 @@ function completedTurnData(input: {
   return {
     messages: [
       {
-        id: `${input.turnId}:user`,
+        id: expect.stringMatching(/^evt_.+:user$/),
         metadata: {
           status: "complete",
           turnId: input.turnId,