From 0c5a052980f83497536085b2725cdc26fe8105c8 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Mon, 27 Jul 2026 16:17:12 +0200 Subject: [PATCH] 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. --- .../app/e2e/agent-message-submission.spec.ts | 15 +++-- .../app/e2e/helpers/daemon-websocket-gate.ts | 41 ++++++++++++-- .../app/e2e/rewind-menu.ui-contract.spec.ts | 55 +++++++++++++++++++ .../app/src/runtime/replica-cache/index.ts | 2 +- .../timeline/session-stream-reducers.test.ts | 28 ++++------ .../src/timeline/session-stream-reducers.ts | 7 +-- 6 files changed, 117 insertions(+), 31 deletions(-) diff --git a/packages/app/e2e/agent-message-submission.spec.ts b/packages/app/e2e/agent-message-submission.spec.ts index 5673e9229e..608447d663 100644 --- a/packages/app/e2e/agent-message-submission.spec.ts +++ b/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(); } diff --git a/packages/app/e2e/helpers/daemon-websocket-gate.ts b/packages/app/e2e/helpers/daemon-websocket-gate.ts index d0d4258df7..d376c84916 100644 --- a/packages/app/e2e/helpers/daemon-websocket-gate.ts +++ b/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; + agentStreamEventTypes: ReadonlySet; + 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(); + const suppressedAgentStreamEventTypes = new Set(); const activeSockets = new Set(); 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; }, diff --git a/packages/app/e2e/rewind-menu.ui-contract.spec.ts b/packages/app/e2e/rewind-menu.ui-contract.spec.ts index 982fd314f8..2e53357be4 100644 --- a/packages/app/e2e/rewind-menu.ui-contract.spec.ts +++ b/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 { + 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> } | 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 { + 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); diff --git a/packages/app/src/runtime/replica-cache/index.ts b/packages/app/src/runtime/replica-cache/index.ts index faafe00dd8..2a57a40120 100644 --- a/packages/app/src/runtime/replica-cache/index.ts +++ b/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 diff --git a/packages/app/src/timeline/session-stream-reducers.test.ts b/packages/app/src/timeline/session-stream-reducers.test.ts index c53b24419e..abbda69d71 100644 --- a/packages/app/src/timeline/session-stream-reducers.test.ts +++ b/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", () => { diff --git a/packages/app/src/timeline/session-stream-reducers.ts b/packages/app/src/timeline/session-stream-reducers.ts index 43c96e2802..4683b41c89 100644 --- a/packages/app/src/timeline/session-stream-reducers.ts +++ b/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;