fix(app): keep replacement submissions authoritative

Keep current pending rows out of legacy cache migration and leave ambiguous terminal lifecycle events to the daemon snapshot. This prevents fabricated rewind identities and stale completion events from marking replacement turns idle.

packages/app/e2e/agent-message-submission.spec.ts

@@ -357,6 +357,7 @@ async function expectLegacyAssistantStartsAfterInterruptedPrompt(
     await expect(page.getByText("Cycle 1", { exact: true })).toBeVisible();
     await queueMessage(page, prompt);
     gate.setAssistantMessageIdsStripped(true);
+    gate.setAgentStreamEventSuppressed("turn_canceled", true);
     await page.getByRole("button", { name: "Send queued message now" }).click();
     const promptRow = page.getByTestId("user-message").filter({ hasText: prompt });
     const replacementAnswer = page.getByText("(end of synthetic stream)", { exact: true }).last();
@@ -365,6 +366,7 @@ async function expectLegacyAssistantStartsAfterInterruptedPrompt(
     await expectRenderedBefore(promptRow, replacementAnswer);
   } finally {
     gate.setAssistantMessageIdsStripped(false);
+    gate.setAgentStreamEventSuppressed("turn_canceled", false);
     await agent.cleanup();
   }
 }
@@ -396,11 +398,16 @@ async function expectStaleCanonicalPagePreservesNewerLiveOutput(
     gate.truncateHeldTimelineAfterLast("tool_call");
     expect(gate.getHeldTimelineLastItemType()).toBe("tool_call");
 
-    await agent.client.sendAgentMessage(agent.agentId, "Stream after the stale snapshot.");
-    const cycleHeadings = page.getByText("Cycle 1", { exact: true });
-    await expect(cycleHeadings).toHaveCount(2);
+    const nextPrompt = "Stream after the stale snapshot.";
+    await agent.client.sendAgentMessage(agent.agentId, nextPrompt);
+    const nextPromptRow = page.getByTestId("user-message").filter({ hasText: nextPrompt });
+    const liveAssistant = nextPromptRow.locator(
+      'xpath=following::*[@data-testid="assistant-message"][1]',
+    );
+    await expect(nextPromptRow).toBeVisible();
+    await expect(liveAssistant).toContainText("Cycle 1");
     gate.releaseHeldServerMessage();
-    await expect(cycleHeadings).toHaveCount(2);
+    await expect(liveAssistant).toContainText("Cycle 1");
   } finally {
     await agent.cleanup();
   }

packages/app/e2e/helpers/daemon-websocket-gate.ts

@@ -81,6 +81,27 @@ function forceTimelineReset(message: string | Buffer, enabled: boolean): string
   return JSON.stringify(envelope);
 }
 
+function readAgentStreamEventType(message: ClientRequest | null): string | null {
+  if (message?.type !== "agent_stream" || !message.payload || typeof message.payload !== "object") {
+    return null;
+  }
+  const event = (message.payload as { event?: { type?: unknown } }).event;
+  return typeof event?.type === "string" ? event.type : null;
+}
+
+function shouldSuppressServerMessage(input: {
+  message: ClientRequest | null;
+  messageTypes: ReadonlySet<string>;
+  agentStreamEventTypes: ReadonlySet<string>;
+  suppressAgentStream: boolean;
+}): boolean {
+  const messageType = typeof input.message?.type === "string" ? input.message.type : null;
+  if (messageType && input.messageTypes.has(messageType)) return true;
+  if (input.suppressAgentStream && messageType === "agent_stream") return true;
+  const eventType = readAgentStreamEventType(input.message);
+  return Boolean(eventType && input.agentStreamEventTypes.has(eventType));
+}
+
 export async function installDaemonWebSocketGate(page: Page) {
   let acceptingConnections = true;
   let reconnectWithFreshClient = false;
@@ -94,6 +115,7 @@ export async function installDaemonWebSocketGate(page: Page) {
   let heldServerMessage: { browser: WebSocketRoute; message: string | Buffer } | null = null;
   let resolveHeldServerMessage: (() => void) | null = null;
   const suppressedServerMessageTypes = new Set<string>();
+  const suppressedAgentStreamEventTypes = new Set<string>();
   const activeSockets = new Set<WebSocketRoute>();
   let latestServer: WebSocketRoute | null = null;
   const directoryStarts: DirectoryRequestStartCounts = {
@@ -177,12 +199,14 @@ export async function installDaemonWebSocketGate(page: Page) {
         return;
       }
       if (
-        typeof serverMessage?.type === "string" &&
-        suppressedServerMessageTypes.has(serverMessage.type)
-      ) {
+        shouldSuppressServerMessage({
+          message: serverMessage,
+          messageTypes: suppressedServerMessageTypes,
+          agentStreamEventTypes: suppressedAgentStreamEventTypes,
+          suppressAgentStream,
+        })
+      )
         return;
-      }
-      if (suppressAgentStream && serverMessage?.type === "agent_stream") return;
       try {
         ws.send(outboundMessage);
       } catch {
@@ -305,6 +329,13 @@ export async function installDaemonWebSocketGate(page: Page) {
         suppressedServerMessageTypes.delete(type);
       }
     },
+    setAgentStreamEventSuppressed(type: string, suppressed: boolean): void {
+      if (suppressed) {
+        suppressedAgentStreamEventTypes.add(type);
+      } else {
+        suppressedAgentStreamEventTypes.delete(type);
+      }
+    },
     setAssistantMessageIdsStripped(stripped: boolean): void {
       stripAssistantMessageIds = stripped;
     },

packages/app/e2e/rewind-menu.ui-contract.spec.ts

@@ -61,7 +61,62 @@ async function rewriteCachedMessageAsLegacyRow(page: Page, prompt: string): Prom
   }, prompt);
 }
 
+async function waitForCurrentSubmissionInCache(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 }>;
+        };
+        return cache.hosts
+          ?.flatMap((host) => host.timeline?.items ?? [])
+          .some(
+            (item) =>
+              item.kind === "user_message" &&
+              item.text === messageText &&
+              typeof item.clientMessageId === "string" &&
+              item.messageId === undefined,
+          );
+      }, prompt),
+    )
+    .toBe(true);
+}
+
+async function expectPendingSubmissionNotRewindableAfterReload(page: Page): Promise<void> {
+  const prompt = "Keep this cached submission pending.";
+  const gate = await installDaemonWebSocketGate(page);
+  const session = await seedMockAgentWorkspace({
+    repoPrefix: "rewind-current-cache-e2e-",
+    title: "Current cache submission e2e",
+  });
+
+  try {
+    await openAgentRoute(page, session);
+    await expectComposerVisible(page);
+    gate.holdNextClientRequest("send_agent_message_request");
+    await submitMessage(page, prompt);
+    await gate.waitForHeldClientRequest();
+    await waitForCurrentSubmissionInCache(page, prompt);
+    await gate.drop();
+    await page.reload();
+
+    const restoredMessage = userMessage(page, prompt);
+    await expect(restoredMessage).toBeVisible();
+    await restoredMessage.hover();
+    await expect(restoredMessage.getByTestId("rewind-menu-trigger")).toHaveCount(0);
+  } finally {
+    gate.restore();
+    await session.cleanup();
+  }
+}
+
 test.describe("Rewind sheet", () => {
+  test("does not restore a pending submission as a rewindable legacy message", async ({ page }) => {
+    await expectPendingSubmissionNotRewindableAfterReload(page);
+  });
+
   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);

packages/app/src/runtime/replica-cache/index.ts

@@ -130,7 +130,7 @@ function decodeDates(value: unknown): unknown {
 }
 
 function restoreCachedStreamItem(item: StreamItem): StreamItem {
-  if (item.kind !== "user_message" || item.messageId) return item;
+  if (item.kind !== "user_message" || item.messageId || item.clientMessageId) 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

packages/app/src/timeline/session-stream-reducers.test.ts

@@ -2403,7 +2403,7 @@ describe("processAgentStreamEvent", () => {
     });
   });
 
-  it("derives submitted idle status on turn_completed for running agent", () => {
+  it("does not mark a running replacement idle on an earlier turn_completed", () => {
     const turnCompletedEvent: AgentStreamEventPayload = {
       type: "turn_completed",
       provider: "claude",
@@ -2420,11 +2420,8 @@ describe("processAgentStreamEvent", () => {
       timestamp: new Date(2000),
     });
 
-    expect(result.agentChanged).toBe(true);
-    expect(result.agent).not.toBe(null);
-    expect(result.agent!.status).toBe("idle");
-    expect(result.agent!.updatedAt.getTime()).toBe(2000);
-    expect(result.agent!.lastActivityAt.getTime()).toBe(2000);
+    expect(result.agentChanged).toBe(false);
+    expect(result.agent).toBe(null);
   });
 
   it("derives submitted error status on turn_failed for running agent", () => {
@@ -2509,15 +2506,16 @@ describe("processAgentStreamEvent", () => {
     expect(result.agent).toBe(null);
   });
 
-  it("preserves updatedAt when agent timestamp is newer than event", () => {
-    const turnCompletedEvent: AgentStreamEventPayload = {
-      type: "turn_completed",
+  it("preserves updatedAt when a failing event is older than the agent", () => {
+    const turnFailedEvent: AgentStreamEventPayload = {
+      type: "turn_failed",
       provider: "claude",
+      error: "something broke",
     };
 
     const result = processAgentStreamEvent({
       ...baseStreamInput,
-      event: turnCompletedEvent,
+      event: turnFailedEvent,
       currentAgent: {
         status: "running",
         updatedAt: new Date(5000),
@@ -2833,7 +2831,7 @@ describe("processAgentStreamEvents", () => {
     ]);
   });
 
-  it("returns the final submitted lifecycle patch across a batch", () => {
+  it("does not derive lifecycle state from a terminal event in a batch", () => {
     const result = processAgentStreamEvents({
       events: [
         makeStreamReducerEvent(makeTimelineEvent("Done"), 1),
@@ -2856,12 +2854,8 @@ describe("processAgentStreamEvents", () => {
 
     expect(result.head).toEqual([]);
     expect(result.tail).toHaveLength(1);
-    expect(result.agentChanged).toBe(true);
-    expect(result.agent).toMatchObject({
-      status: "idle",
-      updatedAt: new Date(3000),
-      lastActivityAt: new Date(3000),
-    });
+    expect(result.agentChanged).toBe(false);
+    expect(result.agent).toBe(null);
   });
 
   it("keeps a live Claude assistant paragraph contiguous when init tail hydration lands mid-stream", () => {

packages/app/src/timeline/session-stream-reducers.ts

@@ -212,13 +212,12 @@ function deriveProjectedLifecycleStatus(
     return null;
   }
   switch (event.type) {
-    case "turn_completed":
-      return "idle";
     case "turn_failed":
       return "error";
+    case "turn_completed":
     case "turn_canceled":
-      // A canceled turn can be either a final user cancel or an interrupt before
-      // a replacement turn starts. The daemon snapshot is authoritative here.
+      // A terminal event can belong to an interrupted turn while its replacement
+      // is still running. The daemon snapshot is authoritative here.
       return null;
     default:
       return null;