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;