fix(app): keep submitted messages consistent through reconnects

Track each in-flight send until its own RPC establishes acceptance, and keep canonical timeline placement authoritative. Restore legacy cached rewind IDs during cache deserialization.
2026-11-06 16:34:36 +00:00 · 2026-07-27 15:22:57 +02:00
parent 979aa1c64c
commit 3063f2868b
11 changed files with 481 additions and 192 deletions
@@ -8,6 +8,9 @@ import {
  expectComposerEditable,
  expectAttachmentPill,
  expectComposerVisible,
+  fillComposerDraft,
+  sendDraftToQueue,
+  startRunningMockAgent,
 } from "./helpers/composer";
 import { openAgentRoute, seedMockAgentWorkspace } from "./helpers/mock-agent";
 import { readScrollMetrics } from "./helpers/agent-bottom-anchor";
@@ -16,6 +19,7 @@ import { waitForWorkspaceTabsVisible } from "./helpers/workspace-tabs";
 import { getServerId } from "./helpers/server-id";
 import { buildHostWorkspaceRoute } from "@/utils/host-routes";
 import { delayBrowserAgentCreatedStatus } from "./helpers/new-workspace";
+import { installDaemonWebSocketGate } from "./helpers/daemon-websocket-gate";

 const IMAGE = {
  name: "message-submission.png",
@@ -46,10 +50,16 @@ interface RejectionScenario {
  errorMessage: string;
 }

+interface UnrelatedRunningScenario {
+  gate: Awaited<ReturnType<typeof gateNextAgentMessage>>;
+  agent: Awaited<ReturnType<typeof seedMockAgentWorkspace>>;
+}
+
 const test = baseTest.extend<{
  submissionScenario: SubmissionScenario;
  draftCreateScenario: DraftCreateScenario;
  rejectionScenario: RejectionScenario;
+  unrelatedRunningScenario: UnrelatedRunningScenario;
 }>({
  submissionScenario: async ({ page }, provide, testInfo) => {
    const gate = await gateNextAgentMessage(page);
@@ -87,6 +97,19 @@ const test = baseTest.extend<{
    await provide({ errorMessage });
    await agent.cleanup();
  },
+  unrelatedRunningScenario: async ({ page }, provide, testInfo) => {
+    const gate = await gateNextAgentMessage(page);
+    const agent = await seedMockAgentWorkspace({
+      repoPrefix: `unrelated-running-${testInfo.workerIndex}-`,
+      title: "Unrelated running transition",
+      model: "one-minute-stream",
+    });
+    await openAgentRoute(page, { workspaceId: agent.workspaceId, agentId: agent.agentId });
+    await expectComposerVisible(page);
+    await expectAgentIdle(page);
+    await provide({ gate, agent });
+    await agent.cleanup();
+  },
 });

 async function submitMessageWithImage(page: Page, prompt: string): Promise<Locator> {
@@ -219,6 +242,79 @@ async function retryRestoredSubmission(page: Page, prompt: string): Promise<void
  await expect(page.getByTestId("composer-image-attachment-pill")).toHaveCount(0);
 }

+async function queueMessage(page: Page, prompt: string): Promise<void> {
+  await fillComposerDraft(page, prompt);
+  await sendDraftToQueue(page);
+}
+
+async function expectQueuedSendFailuresRestored(page: Page, prompts: string[]): Promise<void> {
+  await expect(page.getByRole("button", { name: "Send queued message now" })).toHaveCount(
+    prompts.length,
+  );
+  for (const prompt of prompts) {
+    await expect(page.getByTestId("user-message").filter({ hasText: prompt })).toHaveCount(0);
+  }
+}
+
+async function expectFailedSubmissionRestored(page: Page, prompt: string): Promise<void> {
+  await expectComposerDraft(page, prompt);
+  await expectComposerEditable(page);
+  await expect(page.getByTestId("user-message").filter({ hasText: prompt })).toHaveCount(0);
+}
+
+async function expectInterruptedTurnOrderAfterReconnect(
+  page: Page,
+  testInfo: { workerIndex: number },
+): Promise<void> {
+  const gate = await installDaemonWebSocketGate(page);
+  const agent = await seedMockAgentWorkspace({
+    repoPrefix: `submission-reconnect-${testInfo.workerIndex}-`,
+    title: "Submission reconnect ordering",
+    model: "ten-second-stream",
+  });
+  const prompt = "Keep this prompt before its response.";
+  try {
+    await openAgentRoute(page, { workspaceId: agent.workspaceId, agentId: agent.agentId });
+    await expectComposerVisible(page);
+    await agent.client.sendAgentMessage(agent.agentId, "Start the turn that will be interrupted.");
+    await expect(page.getByRole("button", { name: /stop|cancel/i }).first()).toBeVisible();
+    await expect(page.getByText("Cycle 1", { exact: true })).toBeVisible();
+    await queueMessage(page, prompt);
+    gate.setAgentStreamSuppressed(true);
+    await page.getByRole("button", { name: "Send queued message now" }).click();
+    const promptRow = page.getByTestId("user-message").filter({ hasText: prompt });
+    await expect(promptRow).toBeVisible();
+    await gate.waitForServerMessage("send_agent_message_response");
+    await gate.drop();
+    await agent.client.waitForFinish(agent.agentId, 30_000);
+    gate.setAgentStreamSuppressed(false);
+    gate.forceNextTimelineEpochReset();
+    gate.restoreFresh();
+    await gate.waitForServerMessage("fetch_agent_timeline_response", 2);
+    const response = page.getByText("(end of synthetic stream)", { exact: true }).last();
+    await expect(promptRow).toBeVisible();
+    await expect(response).toBeVisible();
+    await expectRenderedBefore(promptRow, response);
+  } finally {
+    gate.restore();
+    await agent.cleanup();
+  }
+}
+
+async function expectRenderedBefore(first: Locator, second: Locator): Promise<void> {
+  const secondElement = await second.elementHandle();
+  if (!secondElement) throw new Error("Expected the second timeline item to be rendered");
+  expect(
+    await first.evaluate(
+      (firstElement, secondNode) =>
+        Boolean(
+          firstElement.compareDocumentPosition(secondNode) & Node.DOCUMENT_POSITION_FOLLOWING,
+        ),
+      secondElement,
+    ),
+  ).toBe(true);
+}
+
 async function openWorkspaceDraft(page: Page, workspaceId: string): Promise<void> {
  await page.goto(buildHostWorkspaceRoute(getServerId(), workspaceId));
  await waitForWorkspaceTabsVisible(page);
@@ -311,4 +407,54 @@ test.describe("Agent message submission", () => {
    await expectRejectedSubmissionRestored(page, { prompt, ...rejectionScenario });
    await retryRestoredSubmission(page, prompt);
  });
+
+  test("restores overlapping queued sends when their connection fails", async ({
+    page,
+  }, testInfo) => {
+    test.setTimeout(120_000);
+    const gate = await gateNextAgentMessage(page);
+    const agent = await startRunningMockAgent(page, {
+      prefix: `overlapping-queued-send-${testInfo.workerIndex}-`,
+      model: "one-minute-stream",
+      prompt: "Keep the agent running while messages queue.",
+    });
+    const prompts = ["Restore the first queued send.", "Restore the second queued send."];
+    try {
+      await queueMessage(page, prompts[0]);
+      await queueMessage(page, prompts[1]);
+      await page.getByRole("button", { name: "Send queued message now" }).first().click();
+      await gate.waitForRequest(1);
+      await page.getByRole("button", { name: "Send queued message now" }).first().click();
+      await gate.waitForRequest(2);
+      await gate.disconnect();
+      await expectQueuedSendFailuresRestored(page, prompts);
+    } finally {
+      await agent.cleanup();
+    }
+  });
+
+  test("does not accept a failed submission from an unrelated running turn", async ({
+    page,
+    unrelatedRunningScenario,
+  }) => {
+    const prompt = "Restore this unsent prompt.";
+    await submitMessageThatWillBeRejected(page, prompt);
+    await unrelatedRunningScenario.gate.waitForRequest();
+    await unrelatedRunningScenario.agent.client.sendAgentMessage(
+      unrelatedRunningScenario.agent.agentId,
+      "Start an unrelated turn.",
+    );
+    await expect(
+      page.getByTestId("user-message").filter({ hasText: "Start an unrelated turn." }),
+    ).toBeVisible();
+    await unrelatedRunningScenario.gate.disconnect();
+    await expectFailedSubmissionRestored(page, prompt);
+  });
+
+  test("keeps a submitted prompt before its response when canonical history arrives", async ({
+    page,
+  }, testInfo) => {
+    test.setTimeout(90_000);
+    await expectInterruptedTurnOrderAfterReconnect(page, testInfo);
+  });
 });
@@ -36,23 +36,23 @@ function readSendRequest(message: WebSocketMessage): SendAgentMessageRequest | n

 export async function gateNextAgentMessage(page: Page) {
  let serverSocket: WebSocketRoute | null = null;
-  let heldMessage: WebSocketMessage | null = null;
-  let hasHeldRequest = false;
-  let resolveRequest: ((request: SendAgentMessageRequest) => void) | null = null;
-  const requestSeen = new Promise<SendAgentMessageRequest>((resolve) => {
-    resolveRequest = resolve;
-  });
+  let browserSocket: WebSocketRoute | null = null;
+  const heldMessages: Array<WebSocketMessage | null> = [];
+  const requests: SendAgentMessageRequest[] = [];
+  const requestWaiters = new Set<() => void>();

  await page.routeWebSocket(daemonWsRoutePattern(), (ws) => {
+    browserSocket = ws;
    const server = ws.connectToServer();
    serverSocket = server;

    ws.onMessage((message) => {
      const request = readSendRequest(message);
-      if (request && !hasHeldRequest) {
-        heldMessage = message;
-        hasHeldRequest = true;
-        resolveRequest?.(request);
+      if (request) {
+        heldMessages.push(message);
+        requests.push(request);
+        for (const resolve of requestWaiters) resolve();
+        requestWaiters.clear();
        return;
      }
      server.send(message);
@@ -61,14 +61,26 @@ export async function gateNextAgentMessage(page: Page) {
    server.onMessage((message) => ws.send(message));
  });

+  const waitForRequest = async (count = 1): Promise<SendAgentMessageRequest> => {
+    while (requests.length < count) {
+      await new Promise<void>((resolve) => requestWaiters.add(resolve));
+    }
+    return requests[count - 1];
+  };
+
  return {
-    waitForRequest: () => requestSeen,
-    accept() {
+    waitForRequest,
+    accept(index = 0) {
+      const heldMessage = heldMessages[index];
      if (!serverSocket || !heldMessage) {
        throw new Error("No held send-agent-message request to accept");
      }
      serverSocket.send(heldMessage);
-      heldMessage = null;
+      heldMessages[index] = null;
+    },
+    async disconnect(): Promise<void> {
+      if (!browserSocket) throw new Error("No browser daemon socket to disconnect");
+      await browserSocket.close({ code: 1008, reason: "Dropped by submission test." });
    },
  };
 }
@@ -16,6 +16,20 @@ interface ClientRequest {
  type?: unknown;
  subscribe?: unknown;
  page?: { cursor?: unknown };
+  payload?: unknown;
+}
+
+function readSessionMessage(message: string | Buffer): ClientRequest | null {
+  if (typeof message !== "string") return null;
+  try {
+    const envelope = JSON.parse(message) as {
+      type?: unknown;
+      message?: ClientRequest;
+    };
+    return envelope.message ?? envelope;
+  } catch {
+    return null;
+  }
 }

 function readClientRequest(message: string | Buffer): ClientRequest | null {
@@ -40,6 +54,12 @@ function directoryForRequest(request: ClientRequest): keyof DirectoryBootstrapCo

 export async function installDaemonWebSocketGate(page: Page) {
  let acceptingConnections = true;
+  let reconnectWithFreshClient = false;
+  let suppressAgentStream = false;
+  let forceTimelineEpochReset = false;
+  let heldClientRequestType: string | null = null;
+  let heldClientRequest: { server: WebSocketRoute; message: string | Buffer } | null = null;
+  let resolveHeldClientRequest: (() => void) | null = null;
  const activeSockets = new Set<WebSocketRoute>();
  const directoryStarts: DirectoryRequestStartCounts = {
    subscribed: { agents: 0, workspaces: 0 },
@@ -47,6 +67,8 @@ export async function installDaemonWebSocketGate(page: Page) {
    total: { agents: 0, workspaces: 0 },
  };
  const clientRequestCounts = new Map<string, number>();
+  const serverMessageCounts = new Map<string, number>();
+  const serverMessageWaiters = new Set<() => void>();

  await page.routeWebSocket(daemonWsRoutePattern(), (ws) => {
    if (!acceptingConnections) {
@@ -59,6 +81,16 @@ export async function installDaemonWebSocketGate(page: Page) {

    ws.onMessage((message) => {
      if (!acceptingConnections) return;
+      if (reconnectWithFreshClient && typeof message === "string") {
+        const hello = readClientRequest(message);
+        if (hello?.type === "hello") {
+          const parsed = JSON.parse(message) as { clientId?: string };
+          parsed.clientId = `${parsed.clientId ?? "playwright"}-fresh-${Date.now()}`;
+          reconnectWithFreshClient = false;
+          server.send(JSON.stringify(parsed));
+          return;
+        }
+      }
      const request = readClientRequest(message);
      if (typeof request?.type === "string") {
        clientRequestCounts.set(request.type, (clientRequestCounts.get(request.type) ?? 0) + 1);
@@ -69,6 +101,12 @@ export async function installDaemonWebSocketGate(page: Page) {
          directoryStarts.total[directory] += 1;
        }
      }
+      if (request?.type === heldClientRequestType) {
+        heldClientRequest = { server, message };
+        resolveHeldClientRequest?.();
+        resolveHeldClientRequest = null;
+        return;
+      }
      try {
        server.send(message);
      } catch {
@@ -78,8 +116,36 @@ export async function installDaemonWebSocketGate(page: Page) {

    server.onMessage((message) => {
      if (!acceptingConnections) return;
+      const serverMessage = readSessionMessage(message);
+      let outboundMessage = message;
+      if (
+        forceTimelineEpochReset &&
+        serverMessage?.type === "fetch_agent_timeline_response" &&
+        typeof message === "string"
+      ) {
+        const envelope = JSON.parse(message) as {
+          message?: { payload?: Record<string, unknown> };
+          payload?: Record<string, unknown>;
+        };
+        const payload = envelope.message?.payload ?? envelope.payload;
+        if (payload) {
+          payload.epoch = `playwright-reset-${Date.now()}`;
+          payload.reset = true;
+          outboundMessage = JSON.stringify(envelope);
+          forceTimelineEpochReset = false;
+        }
+      }
+      if (typeof serverMessage?.type === "string") {
+        serverMessageCounts.set(
+          serverMessage.type,
+          (serverMessageCounts.get(serverMessage.type) ?? 0) + 1,
+        );
+        for (const resolve of serverMessageWaiters) resolve();
+        serverMessageWaiters.clear();
+      }
+      if (suppressAgentStream && serverMessage?.type === "agent_stream") return;
      try {
-        ws.send(message);
+        ws.send(outboundMessage);
      } catch {
        activeSockets.delete(ws);
      }
@@ -100,6 +166,32 @@ export async function installDaemonWebSocketGate(page: Page) {
    restore(): void {
      acceptingConnections = true;
    },
+    restoreFresh(): void {
+      reconnectWithFreshClient = true;
+      acceptingConnections = true;
+    },
+    holdNextClientRequest(type: string): void {
+      heldClientRequestType = type;
+      heldClientRequest = null;
+    },
+    waitForHeldClientRequest(): Promise<void> {
+      if (heldClientRequest) return Promise.resolve();
+      return new Promise<void>((resolve) => {
+        resolveHeldClientRequest = resolve;
+      });
+    },
+    releaseHeldClientRequest(): void {
+      if (!heldClientRequest) throw new Error("No held client request to release");
+      heldClientRequest.server.send(heldClientRequest.message);
+      heldClientRequest = null;
+      heldClientRequestType = null;
+    },
+    setAgentStreamSuppressed(suppressed: boolean): void {
+      suppressAgentStream = suppressed;
+    },
+    forceNextTimelineEpochReset(): void {
+      forceTimelineEpochReset = true;
+    },
    getDirectoryRequestStartCounts(): DirectoryRequestStartCounts {
      return {
        subscribed: { ...directoryStarts.subscribed },
@@ -110,5 +202,10 @@ export async function installDaemonWebSocketGate(page: Page) {
    getClientRequestCount(type: string): number {
      return clientRequestCounts.get(type) ?? 0;
    },
+    async waitForServerMessage(type: string, count = 1): Promise<void> {
+      while ((serverMessageCounts.get(type) ?? 0) < count) {
+        await new Promise<void>((resolve) => serverMessageWaiters.add(resolve));
+      }
+    },
  };
 }
@@ -1,6 +1,7 @@
 import type { Locator } from "@playwright/test";
 import { expect, test, type Page } from "./fixtures";
 import { openAgentRoute, seedMockAgentWorkspace } from "./helpers/mock-agent";
+import { installDaemonWebSocketGate } from "./helpers/daemon-websocket-gate";
 import {
  composerLocator,
  expectComposerDraft,
@@ -23,7 +24,74 @@ async function expectUserMessageVisible(page: Page, text: string): Promise<void>
  await expect(userMessage(page, text)).toBeVisible();
 }

+async function rewriteCachedMessageAsLegacyRow(page: Page, prompt: string): Promise<void> {
+  await expect
+    .poll(() =>
+      page.evaluate((messageText) => {
+        const raw = localStorage.getItem("@paseo:replica-cache");
+        if (!raw) return false;
+        const cache = JSON.parse(raw) as {
+          hosts?: Array<{ timeline?: { items?: Array<Record<string, unknown>> } | null }>;
+        };
+        for (const host of cache.hosts ?? []) {
+          for (const item of host.timeline?.items ?? []) {
+            if (item.kind === "user_message" && item.text === messageText && item.messageId) {
+              return true;
+            }
+          }
+        }
+        return false;
+      }, prompt),
+    )
+    .toBe(true);
+
+  await page.evaluate((messageText) => {
+    const key = "@paseo:replica-cache";
+    const raw = localStorage.getItem(key);
+    if (!raw) throw new Error("Replica cache was not persisted");
+    const cache = JSON.parse(raw) as {
+      hosts?: Array<{ timeline?: { items?: Array<Record<string, unknown>> } | null }>;
+    };
+    const cachedMessage = cache.hosts
+      ?.flatMap((host) => host.timeline?.items ?? [])
+      .find((item) => item.kind === "user_message" && item.text === messageText);
+    if (!cachedMessage) throw new Error("Cached user message was not found");
+    delete cachedMessage.messageId;
+    localStorage.setItem(key, JSON.stringify(cache));
+  }, prompt);
+}
+
 test.describe("Rewind sheet", () => {
+  test("keeps rewind available for a message restored from the legacy cache", async ({ page }) => {
+    const prompt = "Restore this rewind identity from the legacy cache.";
+    const gate = await installDaemonWebSocketGate(page);
+    const session = await seedMockAgentWorkspace({
+      repoPrefix: "rewind-cache-upgrade-e2e-",
+      title: "Rewind cache upgrade e2e",
+      initialPrompt: prompt,
+    });
+    let heldTimelineRequest = false;
+
+    try {
+      await openAgentRoute(page, session);
+      await expectUserMessageVisible(page, prompt);
+      await rewriteCachedMessageAsLegacyRow(page, prompt);
+      gate.holdNextClientRequest("fetch_agent_timeline_request");
+      await page.reload();
+      await gate.waitForHeldClientRequest();
+      heldTimelineRequest = true;
+
+      const restoredMessage = userMessage(page, prompt);
+      await expect(restoredMessage).toBeVisible();
+      await restoredMessage.hover();
+      await expect(restoredMessage.getByTestId("rewind-menu-trigger")).toBeVisible();
+    } finally {
+      if (heldTimelineRequest) gate.releaseHeldClientRequest();
+      gate.restore();
+      await session.cleanup();
+    }
+  });
+
  test("rewinds from a user message sheet option", async ({ page }) => {
    const firstPrompt = "emit 1 coalesced agent stream updates for first rewind turn.";
    const secondPrompt = "Prepare deleted rewind turn assistant content.";
@@ -10,7 +10,6 @@ import type { StreamItem } from "@/types/stream";
 import {
  acceptMessageSubmission,
  beginMessageSubmission,
-  observeMessageSubmissionRunning,
  rejectMessageSubmission,
  type MessageSubmissionAcceptance,
  type MessageSubmissionState,
@@ -233,14 +232,14 @@ function createFakeStream(

 const submissionsByFakeStream = new WeakMap<
  FakeStream,
-  Map<string, MessageSubmissionState["submission"]>
+  Map<string, MessageSubmissionState["submissions"]>
 >();

 function readSubmission(fake: FakeStream, agentId: string): MessageSubmissionState {
  return {
    tail: fake.tail.get(agentId) ?? [],
    head: fake.head.get(agentId) ?? [],
-    submission: submissionsByFakeStream.get(fake)?.get(agentId) ?? null,
+    submissions: submissionsByFakeStream.get(fake)?.get(agentId) ?? [],
  };
 }

@@ -248,25 +247,10 @@ function writeSubmission(fake: FakeStream, agentId: string, state: MessageSubmis
  fake.tail = new Map(fake.tail).set(agentId, state.tail);
  fake.head = new Map(fake.head).set(agentId, state.head);
  const submissions = submissionsByFakeStream.get(fake) ?? new Map();
-  submissions.set(agentId, state.submission);
+  submissions.set(agentId, state.submissions);
  submissionsByFakeStream.set(fake, submissions);
 }

-function applyFakeAuthoritativeRunning(
-  fake: FakeStream,
-  agentId: string,
-  clientMessageId: string,
-): void {
-  const state = readSubmission(fake, agentId);
-  if (state.submission?.clientMessageId !== clientMessageId) {
-    return;
-  }
-  writeSubmission(fake, agentId, {
-    ...state,
-    submission: observeMessageSubmissionRunning(state.submission),
-  });
-}
-
 function createFakeQueue(
  initial: Map<string, QueuedComposerMessage[]> = new Map(),
 ): QueueWriter & { state: Map<string, QueuedComposerMessage[]> } {
@@ -420,43 +404,10 @@ describe("dispatchComposerAgentMessage", () => {
    expect(stream.tail.get("agent") ?? []).toEqual([]);
  });

-  it("keeps the submitted prompt when authoritative running wins a late transport error", async () => {
-    const stream = createFakeStream();
-    const transportError = new Error("Connection lost after authoritative running");
-    const client = createFakeSendClient({
-      rejection: transportError,
-      beforeRejection: ({ agentId, options }) => {
-        applyFakeAuthoritativeRunning(stream, agentId, options.messageId);
-      },
-    });
-
-    await expect(
-      dispatchComposerAgentMessage({
-        client,
-        agentId: "agent",
-        text: "accepted before transport loss",
-        attachments: [],
-        encodeImages: passthroughEncodeImages,
-        submission: stream,
-      }),
-    ).resolves.toBeUndefined();
-
-    expect(stream.tail.get("agent")).toHaveLength(1);
-    expect(stream.tail.get("agent")?.[0]).toMatchObject({
-      kind: "user_message",
-      text: "accepted before transport loss",
-    });
-  });
-
  it("rolls back an already-running force send when its RPC fails", async () => {
    const stream = createFakeStream(new Map(), { acceptance: "rpc-only" });
    const transportError = new Error("Force send failed while the prior turn was running");
-    const client = createFakeSendClient({
-      rejection: transportError,
-      beforeRejection: ({ agentId, options }) => {
-        applyFakeAuthoritativeRunning(stream, agentId, options.messageId);
-      },
-    });
+    const client = createFakeSendClient({ rejection: transportError });

    await expect(
      dispatchComposerAgentMessage({
@@ -10,7 +10,7 @@ import {
 } from "./model";

 function initialState(head: StreamItem[] = []): MessageSubmissionState {
-  return { tail: [], head, submission: null };
+  return { tail: [], head, submissions: [] };
 }

 describe("message submission model", () => {
@@ -27,18 +27,20 @@ describe("message submission model", () => {
    });

    const accepted = acceptMessageSubmission(begun, "client-1");
-    const runningObserved = observeMessageSubmissionRunning(accepted.submission);
+    const runningObserved = observeMessageSubmissionRunning(accepted.submissions);

-    expect(accepted.submission).toEqual({
-      clientMessageId: "client-1",
-      submittedAt: message.timestamp,
-      phase: "waiting-for-running",
-    });
-    expect(getPendingMessageSubmission(accepted.submission)).toEqual(accepted.submission);
-    expect(runningObserved).toBeNull();
+    expect(accepted.submissions).toEqual([
+      {
+        clientMessageId: "client-1",
+        submittedAt: message.timestamp,
+        phase: "waiting-for-running",
+      },
+    ]);
+    expect(getPendingMessageSubmission(accepted.submissions)).toEqual(accepted.submissions[0]);
+    expect(runningObserved).toEqual([]);
  });

-  it("keeps a row when authoritative running precedes a late RPC error", () => {
+  it("rejects a row when an unrelated running transition precedes its RPC error", () => {
    const message = createUserMessage({
      clientMessageId: "client-1",
      text: "hello",
@@ -50,18 +52,17 @@ describe("message submission model", () => {
      acceptance: "rpc-and-running",
    });

-    const runningObserved = observeMessageSubmissionRunning(begun.submission);
-    const rejected = rejectMessageSubmission({ ...begun, submission: runningObserved }, "client-1");
+    const runningObserved = observeMessageSubmissionRunning(begun.submissions);
+    const rejected = rejectMessageSubmission(
+      { ...begun, submissions: runningObserved },
+      "client-1",
+    );

-    expect(runningObserved).toEqual({
-      clientMessageId: "client-1",
-      submittedAt: message.timestamp,
-      phase: "running-observed",
-    });
-    expect(getPendingMessageSubmission(runningObserved)).toBeNull();
+    expect(runningObserved).toBe(begun.submissions);
+    expect(getPendingMessageSubmission(runningObserved)).toEqual(runningObserved[0]);
    expect(rejected).toEqual({
-      outcome: "accepted",
-      state: { ...begun, submission: null },
+      outcome: "rejected",
+      state: { tail: [], head: [], submissions: [] },
    });
  });

@@ -77,7 +78,7 @@ describe("message submission model", () => {
      acceptance: "rpc-only",
    });

-    expect(observeMessageSubmissionRunning(begun.submission)).toBe(begun.submission);
+    expect(observeMessageSubmissionRunning(begun.submissions)).toBe(begun.submissions);
  });

  it("begins and accepts without changing the submitted row", () => {
@@ -98,16 +99,18 @@ describe("message submission model", () => {
    expect(begun).toEqual({
      tail: [message],
      head: [],
-      submission: {
-        clientMessageId: "client-1",
-        submittedAt,
-        phase: "waiting-for-rpc",
-        acceptance: "rpc-only",
-      },
+      submissions: [
+        {
+          clientMessageId: "client-1",
+          submittedAt,
+          phase: "waiting-for-rpc",
+          acceptance: "rpc-only",
+        },
+      ],
    });
    expect(accepted.tail).toBe(begun.tail);
    expect(accepted.head).toBe(begun.head);
-    expect(accepted.submission).toBeNull();
+    expect(accepted.submissions).toEqual([]);
  });

  it("rejects by removing the submitted row and closing pending", () => {
@@ -134,12 +137,12 @@ describe("message submission model", () => {
      state: {
        tail: [],
        head: [existingHead],
-        submission: null,
+        submissions: [],
      },
    });
  });

-  it("treats running-observed as authoritatively accepted", () => {
+  it("treats RPC acceptance waiting for running as authoritatively accepted", () => {
    const message = createUserMessage({
      clientMessageId: "client-1",
      text: "hello",
@@ -147,18 +150,20 @@ describe("message submission model", () => {
    });
    const state = initialState([message]);

-    const runningObserved = {
+    const waitingForRunning = {
      ...state,
-      submission: {
-        clientMessageId: "client-1",
-        submittedAt: message.timestamp,
-        phase: "running-observed" as const,
-      },
+      submissions: [
+        {
+          clientMessageId: "client-1",
+          submittedAt: message.timestamp,
+          phase: "waiting-for-running" as const,
+        },
+      ],
    };

-    expect(rejectMessageSubmission(runningObserved, "client-1")).toEqual({
+    expect(rejectMessageSubmission(waitingForRunning, "client-1")).toEqual({
      outcome: "accepted",
-      state: { ...runningObserved, submission: null },
+      state: { ...waitingForRunning, submissions: [] },
    });
  });

@@ -190,7 +195,7 @@ describe("submitting alongside repeated history", () => {
    });

    const begun = beginMessageSubmission(
-      { tail: [olderRow], head: [], submission: null },
+      { tail: [olderRow], head: [], submissions: [] },
      {
        message: resubmitted,
        submittedAt: resubmitted.timestamp,
@@ -11,19 +11,18 @@ export type MessageSubmissionRecord = PendingMessageSubmission &
  (
    | { phase: "waiting-for-rpc"; acceptance: MessageSubmissionAcceptance }
    | { phase: "waiting-for-running" }
-    | { phase: "running-observed" }
  );

 export function getPendingMessageSubmission(
-  submission: MessageSubmissionRecord | null | undefined,
+  submissions: readonly MessageSubmissionRecord[] | null | undefined,
 ): PendingMessageSubmission | null {
-  return submission && submission.phase !== "running-observed" ? submission : null;
+  return submissions?.at(-1) ?? null;
 }

 export interface MessageSubmissionState {
  tail: StreamItem[];
  head: StreamItem[];
-  submission: MessageSubmissionRecord | null;
+  submissions: MessageSubmissionRecord[];
 }

 export type MessageSubmissionRejectionOutcome = "rejected" | "accepted" | "unknown";
@@ -44,63 +43,58 @@ export function beginMessageSubmission(
  if (!input.message.clientMessageId) {
    throw new Error("Submitted user message requires client identity");
  }
-  const pending = {
+  const pending: MessageSubmissionRecord = {
    clientMessageId: input.message.clientMessageId,
    submittedAt: input.submittedAt,
-    phase: "waiting-for-rpc" as const,
+    phase: "waiting-for-rpc",
    acceptance: input.acceptance,
  };
  // A submission carries a freshly generated client identity, so it cannot already be in
  // the timeline. Append it. Reconciling here would let the legacy content fallback adopt
  // an older row that happens to repeat the same text.
-  if (state.head.length > 0) {
-    return {
-      tail: state.tail,
-      head: [...state.head, input.message],
-      submission: pending,
-    };
-  }
-  return {
-    tail: [...state.tail, input.message],
-    head: state.head,
-    submission: pending,
-  };
+  return state.head.length > 0
+    ? {
+        tail: state.tail,
+        head: [...state.head, input.message],
+        submissions: [...state.submissions, pending],
+      }
+    : {
+        tail: [...state.tail, input.message],
+        head: state.head,
+        submissions: [...state.submissions, pending],
+      };
 }

 export function acceptMessageSubmission(
  state: MessageSubmissionState,
  clientMessageId: string,
+  isAgentRunning = false,
 ): MessageSubmissionState {
-  if (state.submission?.clientMessageId !== clientMessageId) {
-    return state;
-  }
-  if (state.submission.phase === "running-observed") {
-    return { ...state, submission: null };
-  }
-  if (state.submission.phase === "waiting-for-running") return state;
-  return state.submission.acceptance === "rpc-only"
-    ? { ...state, submission: null }
-    : {
-        ...state,
-        submission: {
-          clientMessageId: state.submission.clientMessageId,
-          submittedAt: state.submission.submittedAt,
+  const index = state.submissions.findIndex(
+    (submission) => submission.clientMessageId === clientMessageId,
+  );
+  if (index === -1) return state;
+  const submission = state.submissions[index];
+  if (submission.phase === "waiting-for-running") return state;
+  const nextSubmission: MessageSubmissionRecord | null =
+    submission.acceptance === "rpc-only" || isAgentRunning
+      ? null
+      : {
+          clientMessageId: submission.clientMessageId,
+          submittedAt: submission.submittedAt,
          phase: "waiting-for-running",
-        },
-      };
+        };
+  const submissions = state.submissions.slice();
+  if (nextSubmission) submissions[index] = nextSubmission;
+  else submissions.splice(index, 1);
+  return { ...state, submissions };
 }

 export function observeMessageSubmissionRunning(
-  submission: MessageSubmissionRecord | null,
-): MessageSubmissionRecord | null {
-  if (!submission || submission.phase === "running-observed") return submission;
-  if (submission.phase === "waiting-for-running") return null;
-  if (submission.acceptance === "rpc-only") return submission;
-  return {
-    clientMessageId: submission.clientMessageId,
-    submittedAt: submission.submittedAt,
-    phase: "running-observed",
-  };
+  submissions: readonly MessageSubmissionRecord[],
+): MessageSubmissionRecord[] {
+  const next = submissions.filter((submission) => submission.phase !== "waiting-for-running");
+  return next.length === submissions.length ? (submissions as MessageSubmissionRecord[]) : next;
 }

 function removeSubmittedMessage(items: StreamItem[], clientMessageId: string): StreamItem[] {
@@ -114,21 +108,22 @@ export function rejectMessageSubmission(
  state: MessageSubmissionState,
  clientMessageId: string,
 ): MessageSubmissionRejectionResult {
-  if (state.submission?.clientMessageId !== clientMessageId) {
-    return { outcome: "unknown", state };
-  }
-  if (
-    state.submission.phase === "running-observed" ||
-    state.submission.phase === "waiting-for-running"
-  ) {
-    return { outcome: "accepted", state: { ...state, submission: null } };
+  const index = state.submissions.findIndex(
+    (submission) => submission.clientMessageId === clientMessageId,
+  );
+  if (index === -1) return { outcome: "unknown", state };
+  const submission = state.submissions[index];
+  const submissions = state.submissions.slice();
+  submissions.splice(index, 1);
+  if (submission.phase === "waiting-for-running") {
+    return { outcome: "accepted", state: { ...state, submissions } };
  }
  return {
    outcome: "rejected",
    state: {
      tail: removeSubmittedMessage(state.tail, clientMessageId),
      head: removeSubmittedMessage(state.head, clientMessageId),
-      submission: null,
+      submissions,
    },
  };
 }
@@ -129,6 +129,15 @@ function decodeDates(value: unknown): unknown {
  return Object.fromEntries(Object.entries(value).map(([key, entry]) => [key, decodeDates(entry)]));
 }

+function restoreCachedStreamItem(item: StreamItem): StreamItem {
+  if (item.kind !== "user_message" || item.messageId) return item;
+  const legacyItem = item as StreamItem & { optimistic?: true };
+  if (legacyItem.optimistic) return item;
+  // COMPAT(replicaCacheUserMessageId): before v0.2.3, cache v1 stored the provider ID
+  // only in `id`. Remove after 2027-01-27, once cache v1 predating v0.2.3 is unsupported.
+  return { ...item, messageId: item.id };
+}
+
 function deserializeTimeline(stored: StoredHost["timeline"]): SessionReplica["timeline"] {
  if (!stored) {
    return null;
@@ -139,7 +148,7 @@ function deserializeTimeline(stored: StoredHost["timeline"]): SessionReplica["ti
  }
  return {
    agentId: stored.agentId,
-    items: decoded,
+    items: decoded.map(restoreCachedStreamItem),
    cursor: stored.cursor,
    hasOlder: stored.hasOlder,
  };
@@ -376,7 +376,7 @@ export interface SessionState {
  // Stream state (head/tail model)
  agentStreamTail: Map<string, StreamItem[]>;
  agentStreamHead: Map<string, StreamItem[]>;
-  messageSubmissions: Map<string, MessageSubmissionRecord>;
+  messageSubmissions: Map<string, MessageSubmissionRecord[]>;
  agentTimelineCursor: Map<string, AgentTimelineCursorState>;
  agentTimelineHasOlder: Map<string, boolean>;
  agentTimelineOlderFetchInFlight: Map<string, boolean>;
@@ -594,22 +594,22 @@ const agentLastActivityCoalescer = createAgentLastActivityCoalescer();
 function applyAgentStateToMessageSubmissions(input: {
  previousAgents: Map<string, Agent>;
  nextAgents: Map<string, Agent>;
-  submissions: Map<string, MessageSubmissionRecord>;
-}): Map<string, MessageSubmissionRecord> {
+  submissions: Map<string, MessageSubmissionRecord[]>;
+}): Map<string, MessageSubmissionRecord[]> {
  let nextSubmissions = input.submissions;
-  const write = (agentId: string, submission: MessageSubmissionRecord | null): void => {
+  const write = (agentId: string, submissions: MessageSubmissionRecord[]): void => {
    if (nextSubmissions === input.submissions) nextSubmissions = new Map(input.submissions);
-    if (submission) {
-      nextSubmissions.set(agentId, submission);
+    if (submissions.length > 0) {
+      nextSubmissions.set(agentId, submissions);
    } else {
      nextSubmissions.delete(agentId);
    }
  };

-  for (const [agentId, submission] of input.submissions) {
+  for (const [agentId, submissions] of input.submissions) {
    const nextAgent = input.nextAgents.get(agentId);
    if (!nextAgent) {
-      write(agentId, null);
+      write(agentId, []);
      continue;
    }
    const previousAgent = input.previousAgents.get(agentId);
@@ -618,8 +618,8 @@ function applyAgentStateToMessageSubmissions(input: {
      previousAgent.status !== "running" &&
      nextAgent.status === "running";
    if (!observedRunningTransition) continue;
-    const nextSubmission = observeMessageSubmissionRunning(submission);
-    if (nextSubmission !== submission) write(agentId, nextSubmission);
+    const nextSubmission = observeMessageSubmissionRunning(submissions);
+    if (nextSubmission !== submissions) write(agentId, nextSubmission);
  }
  return nextSubmissions;
 }
@@ -1133,7 +1133,7 @@ export const useSessionStore = create<SessionStore>()(
            {
              tail: session.agentStreamTail.get(agentId) ?? [],
              head: session.agentStreamHead.get(agentId) ?? [],
-              submission: session.messageSubmissions.get(agentId) ?? null,
+              submissions: session.messageSubmissions.get(agentId) ?? [],
            },
            {
              message,
@@ -1145,10 +1145,10 @@ export const useSessionStore = create<SessionStore>()(
            },
          );
          const messageSubmissions = new Map(session.messageSubmissions);
-          if (!next.submission) {
+          if (next.submissions.length === 0) {
            throw new Error("Beginning a message submission must create lifecycle state");
          }
-          messageSubmissions.set(agentId, next.submission);
+          messageSubmissions.set(agentId, next.submissions);
          return {
            ...prev,
            sessions: {
@@ -1168,19 +1168,20 @@ export const useSessionStore = create<SessionStore>()(
        set((prev) => {
          const session = prev.sessions[serverId];
          if (!session) return prev;
-          const currentSubmission = session.messageSubmissions.get(agentId) ?? null;
+          const currentSubmissions = session.messageSubmissions.get(agentId) ?? [];
          const next = acceptMessageSubmission(
            {
              tail: session.agentStreamTail.get(agentId) ?? [],
              head: session.agentStreamHead.get(agentId) ?? [],
-              submission: currentSubmission,
+              submissions: currentSubmissions,
            },
            clientMessageId,
+            session.agents.get(agentId)?.status === "running",
          );
-          if (next.submission === currentSubmission) return prev;
+          if (next.submissions === currentSubmissions) return prev;
          const messageSubmissions = new Map(session.messageSubmissions);
-          if (next.submission) {
-            messageSubmissions.set(agentId, next.submission);
+          if (next.submissions.length > 0) {
+            messageSubmissions.set(agentId, next.submissions);
          } else {
            messageSubmissions.delete(agentId);
          }
@@ -1201,16 +1202,20 @@ export const useSessionStore = create<SessionStore>()(
          if (!session) return prev;
          const currentTail = session.agentStreamTail.get(agentId) ?? [];
          const currentHead = session.agentStreamHead.get(agentId) ?? [];
-          const currentSubmission = session.messageSubmissions.get(agentId) ?? null;
+          const currentSubmissions = session.messageSubmissions.get(agentId) ?? [];
          const result = rejectMessageSubmission(
-            { tail: currentTail, head: currentHead, submission: currentSubmission },
+            { tail: currentTail, head: currentHead, submissions: currentSubmissions },
            clientMessageId,
          );
          outcome = result.outcome;
          const next = result.state;
          if (outcome === "unknown") return prev;
          const messageSubmissions = new Map(session.messageSubmissions);
-          messageSubmissions.delete(agentId);
+          if (next.submissions.length > 0) {
+            messageSubmissions.set(agentId, next.submissions);
+          } else {
+            messageSubmissions.delete(agentId);
+          }
          return {
            ...prev,
            sessions: {
@@ -374,7 +374,6 @@ export function replaceWithCanonicalStream(
      item.kind === "user_message" && item.clientMessageId !== undefined,
  );
  let nextHead = input.previousHead;
-  const matchedHeadClientMessageIds = new Set<string>();
  const nextTail: StreamItem[] = [];

  for (const item of input.canonical) {
@@ -392,10 +391,11 @@ export function replaceWithCanonicalStream(

    const headResult = produceUserMessage(nextHead, item, null, "existing");
    if (headResult.matched) {
-      nextHead = headResult.items;
-      if (headResult.message.clientMessageId) {
-        matchedHeadClientMessageIds.add(headResult.message.clientMessageId);
-      }
+      nextHead = [
+        ...headResult.items.slice(0, headResult.index),
+        ...headResult.items.slice(headResult.index + 1),
+      ];
+      nextTail.push(headResult.message);
      continue;
    }

@@ -417,7 +417,6 @@ export function replaceWithCanonicalStream(

  nextHead = nextHead.filter((item) => {
    if (item.kind !== "user_message" || !item.clientMessageId) return true;
-    if (matchedHeadClientMessageIds.has(item.clientMessageId)) return true;
    return (
      input.policy === "preserve-local-submissions" ||
      item.clientMessageId === input.pendingClientMessageId
@@ -114,11 +114,13 @@ describe("message submission lifecycle ordering", () => {

    const session = useSessionStore.getState().sessions[serverId];
    expect(session?.agents.get(agentId)?.status).toBe("idle");
-    expect(session?.messageSubmissions.get(agentId)).toEqual({
-      clientMessageId,
-      submittedAt: new Date("2026-07-27T10:00:00.000Z"),
-      phase: "waiting-for-running",
-    });
+    expect(session?.messageSubmissions.get(agentId)).toEqual([
+      {
+        clientMessageId,
+        submittedAt: new Date("2026-07-27T10:00:00.000Z"),
+        phase: "waiting-for-running",
+      },
+    ]);

    applyAgentStatus({
      serverId,