diff --git a/packages/app/e2e/agent-message-submission.spec.ts b/packages/app/e2e/agent-message-submission.spec.ts index e3c51e74e3..5673e9229e 100644 --- a/packages/app/e2e/agent-message-submission.spec.ts +++ b/packages/app/e2e/agent-message-submission.spec.ts @@ -1,6 +1,6 @@ import type { Locator, Page } from "@playwright/test"; import { expect, test as baseTest } from "./fixtures"; -import { expectAgentIdle } from "./helpers/agent-stream"; +import { awaitToolCall, expectAgentIdle } from "./helpers/agent-stream"; import { gateNextAgentMessage } from "./helpers/agent-message-gate"; import { attachImageFromMenu, @@ -301,6 +301,111 @@ async function expectInterruptedTurnOrderAfterReconnect( } } +async function expectCompletedSubmissionClearsAfterMissedRunningTransition( + page: Page, + testInfo: { workerIndex: number }, +): Promise { + const gate = await installDaemonWebSocketGate(page); + const agent = await seedMockAgentWorkspace({ + repoPrefix: `submission-missed-running-${testInfo.workerIndex}-`, + title: "Submission missed running transition", + model: "ten-second-stream", + }); + const prompt = "Clear this submission from canonical history."; + try { + await openAgentRoute(page, { workspaceId: agent.workspaceId, agentId: agent.agentId }); + await expectComposerVisible(page); + await expectAgentIdle(page); + gate.holdNextClientRequest("send_agent_message_request"); + const userMessage = await submitMessageWithImage(page, prompt); + await gate.waitForHeldClientRequest(); + gate.setServerMessageSuppressed("agent_status", true); + gate.setServerMessageSuppressed("agent_update", true); + gate.releaseHeldClientRequest(); + await gate.waitForServerMessage("send_agent_message_response"); + await expect(userMessage).toHaveAttribute("aria-busy", "false"); + await gate.drop(); + await agent.client.waitForFinish(agent.agentId, 30_000); + gate.setServerMessageSuppressed("agent_status", false); + gate.setServerMessageSuppressed("agent_update", false); + gate.restoreFresh(); + await gate.waitForServerMessage("fetch_agent_timeline_response", 2); + await expect(page.getByText("(end of synthetic stream)", { exact: true }).last()).toBeVisible(); + await expect(page.getByTestId("turn-working-indicator")).toHaveCount(0); + await expect(userMessage).toHaveAttribute("aria-busy", "false"); + } finally { + gate.restore(); + await agent.cleanup(); + } +} + +async function expectLegacyAssistantStartsAfterInterruptedPrompt( + page: Page, + testInfo: { workerIndex: number }, +): Promise { + const gate = await installDaemonWebSocketGate(page); + const agent = await seedMockAgentWorkspace({ + repoPrefix: `submission-legacy-assistant-${testInfo.workerIndex}-`, + title: "Legacy assistant interrupt boundary", + model: "ten-second-stream", + }); + const prompt = "Start the replacement answer after this prompt."; + try { + await openAgentRoute(page, { workspaceId: agent.workspaceId, agentId: agent.agentId }); + await expectComposerVisible(page); + await agent.client.sendAgentMessage(agent.agentId, "Start the interrupted answer."); + await expect(page.getByText("Cycle 1", { exact: true })).toBeVisible(); + await queueMessage(page, prompt); + gate.setAssistantMessageIdsStripped(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(); + await expect(promptRow).toBeVisible(); + await expect(replacementAnswer).toBeVisible({ timeout: 30_000 }); + await expectRenderedBefore(promptRow, replacementAnswer); + } finally { + gate.setAssistantMessageIdsStripped(false); + await agent.cleanup(); + } +} + +async function expectStaleCanonicalPagePreservesNewerLiveOutput( + page: Page, + testInfo: { workerIndex: number }, +): Promise { + const gate = await installDaemonWebSocketGate(page); + const agent = await seedMockAgentWorkspace({ + repoPrefix: `submission-stale-canonical-${testInfo.workerIndex}-`, + title: "Stale canonical page race", + model: "one-minute-stream", + }); + try { + await openAgentRoute(page, { workspaceId: agent.workspaceId, agentId: agent.agentId }); + await expectComposerVisible(page); + await agent.client.sendAgentMessage(agent.agentId, "End the snapshot at a tool call."); + await awaitToolCall(page, "read"); + await page + .getByRole("button", { name: /stop|cancel/i }) + .first() + .click(); + await expectAgentIdle(page); + + gate.holdNextServerMessage("fetch_agent_timeline_response"); + gate.requestTimelineTail(agent.agentId); + await gate.waitForHeldServerMessage(); + 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); + gate.releaseHeldServerMessage(); + await expect(cycleHeadings).toHaveCount(2); + } finally { + await agent.cleanup(); + } +} + async function expectRenderedBefore(first: Locator, second: Locator): Promise { const secondElement = await second.elementHandle(); if (!secondElement) throw new Error("Expected the second timeline item to be rendered"); @@ -457,4 +562,25 @@ test.describe("Agent message submission", () => { test.setTimeout(90_000); await expectInterruptedTurnOrderAfterReconnect(page, testInfo); }); + + test("clears an accepted submission when canonical history arrives after a missed running transition", async ({ + page, + }, testInfo) => { + test.setTimeout(90_000); + await expectCompletedSubmissionClearsAfterMissedRunningTransition(page, testInfo); + }); + + test("keeps an old-daemon replacement answer after its interrupted prompt", async ({ + page, + }, testInfo) => { + test.setTimeout(90_000); + await expectLegacyAssistantStartsAfterInterruptedPrompt(page, testInfo); + }); + + test("preserves newer live output when a stale canonical page arrives", async ({ + page, + }, testInfo) => { + test.setTimeout(90_000); + await expectStaleCanonicalPagePreservesNewerLiveOutput(page, testInfo); + }); }); diff --git a/packages/app/e2e/helpers/daemon-websocket-gate.ts b/packages/app/e2e/helpers/daemon-websocket-gate.ts index d6e56df6d1..d0d4258df7 100644 --- a/packages/app/e2e/helpers/daemon-websocket-gate.ts +++ b/packages/app/e2e/helpers/daemon-websocket-gate.ts @@ -52,15 +52,50 @@ function directoryForRequest(request: ClientRequest): keyof DirectoryBootstrapCo return null; } +function stripAssistantMessageId( + message: string | Buffer, + enabled: boolean, + messageType: unknown, +): string | Buffer { + if (!enabled || messageType !== "agent_stream" || typeof message !== "string") return message; + const envelope = JSON.parse(message) as { + message?: { payload?: { event?: { type?: unknown; item?: Record } } }; + payload?: { event?: { type?: unknown; item?: Record } }; + }; + const event = (envelope.message?.payload ?? envelope.payload)?.event; + if (event?.type !== "timeline" || event.item?.type !== "assistant_message") return message; + delete event.item.messageId; + return JSON.stringify(envelope); +} + +function forceTimelineReset(message: string | Buffer, enabled: boolean): string | Buffer { + if (!enabled || typeof message !== "string") return message; + const envelope = JSON.parse(message) as { + message?: { payload?: Record }; + payload?: Record; + }; + const payload = envelope.message?.payload ?? envelope.payload; + if (!payload) return message; + payload.epoch = `playwright-reset-${Date.now()}`; + payload.reset = true; + return JSON.stringify(envelope); +} + export async function installDaemonWebSocketGate(page: Page) { let acceptingConnections = true; let reconnectWithFreshClient = false; let suppressAgentStream = false; let forceTimelineEpochReset = false; + let stripAssistantMessageIds = false; let heldClientRequestType: string | null = null; let heldClientRequest: { server: WebSocketRoute; message: string | Buffer } | null = null; let resolveHeldClientRequest: (() => void) | null = null; + let heldServerMessageType: string | null = null; + let heldServerMessage: { browser: WebSocketRoute; message: string | Buffer } | null = null; + let resolveHeldServerMessage: (() => void) | null = null; + const suppressedServerMessageTypes = new Set(); const activeSockets = new Set(); + let latestServer: WebSocketRoute | null = null; const directoryStarts: DirectoryRequestStartCounts = { subscribed: { agents: 0, workspaces: 0 }, unsubscribed: { agents: 0, workspaces: 0 }, @@ -78,6 +113,7 @@ export async function installDaemonWebSocketGate(page: Page) { activeSockets.add(ws); const server = ws.connectToServer(); + latestServer = server; ws.onMessage((message) => { if (!acceptingConnections) return; @@ -117,24 +153,15 @@ 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 }; - payload?: Record; - }; - const payload = envelope.message?.payload ?? envelope.payload; - if (payload) { - payload.epoch = `playwright-reset-${Date.now()}`; - payload.reset = true; - outboundMessage = JSON.stringify(envelope); - forceTimelineEpochReset = false; - } - } + let outboundMessage = stripAssistantMessageId( + message, + stripAssistantMessageIds, + serverMessage?.type, + ); + const shouldForceTimelineReset = + forceTimelineEpochReset && serverMessage?.type === "fetch_agent_timeline_response"; + outboundMessage = forceTimelineReset(outboundMessage, shouldForceTimelineReset); + if (shouldForceTimelineReset) forceTimelineEpochReset = false; if (typeof serverMessage?.type === "string") { serverMessageCounts.set( serverMessage.type, @@ -143,6 +170,18 @@ export async function installDaemonWebSocketGate(page: Page) { for (const resolve of serverMessageWaiters) resolve(); serverMessageWaiters.clear(); } + if (serverMessage?.type === heldServerMessageType) { + heldServerMessage = { browser: ws, message: outboundMessage }; + resolveHeldServerMessage?.(); + resolveHeldServerMessage = null; + return; + } + if ( + typeof serverMessage?.type === "string" && + suppressedServerMessageTypes.has(serverMessage.type) + ) { + return; + } if (suppressAgentStream && serverMessage?.type === "agent_stream") return; try { ws.send(outboundMessage); @@ -186,6 +225,89 @@ export async function installDaemonWebSocketGate(page: Page) { heldClientRequest = null; heldClientRequestType = null; }, + holdNextServerMessage(type: string): void { + heldServerMessageType = type; + heldServerMessage = null; + }, + waitForHeldServerMessage(): Promise { + if (heldServerMessage) return Promise.resolve(); + return new Promise((resolve) => { + resolveHeldServerMessage = resolve; + }); + }, + releaseHeldServerMessage(): void { + if (!heldServerMessage) throw new Error("No held server message to release"); + heldServerMessage.browser.send(heldServerMessage.message); + heldServerMessage = null; + heldServerMessageType = null; + }, + requestTimelineTail(agentId: string): void { + if (!latestServer) throw new Error("No daemon WebSocket is connected"); + latestServer.send( + JSON.stringify({ + type: "session", + message: { + type: "fetch_agent_timeline_request", + agentId, + requestId: `playwright-timeline-${Date.now()}`, + direction: "tail", + limit: 0, + projection: "projected", + }, + }), + ); + }, + getHeldTimelineLastItemType(): string | null { + if (!heldServerMessage) throw new Error("No held server message to inspect"); + const response = readSessionMessage(heldServerMessage.message); + const payload = response?.payload; + if (!payload || typeof payload !== "object") return null; + const entries = (payload as { entries?: unknown }).entries; + if (!Array.isArray(entries)) return null; + const last = entries.at(-1) as { item?: { type?: unknown } } | undefined; + return typeof last?.item?.type === "string" ? last.item.type : null; + }, + truncateHeldTimelineAfterLast(itemType: string): void { + if (!heldServerMessage || typeof heldServerMessage.message !== "string") { + throw new Error("No held text server message to truncate"); + } + const envelope = JSON.parse(heldServerMessage.message) as { + message?: { payload?: Record }; + payload?: Record; + }; + const payload = envelope.message?.payload ?? envelope.payload; + if (!payload) throw new Error("Held message has no payload"); + const entries = payload.entries; + if (!Array.isArray(entries)) throw new Error("Held message is not a timeline response"); + const index = entries.findLastIndex( + (entry) => + typeof entry === "object" && + entry !== null && + (entry as { item?: { type?: unknown } }).item?.type === itemType, + ); + if (index < 0) throw new Error(`Timeline response has no ${itemType} item`); + const retained = entries.slice(0, index + 1) as Array<{ seqEnd?: unknown }>; + const lastSeq = retained.at(-1)?.seqEnd; + if (typeof lastSeq !== "number") throw new Error("Timeline entry has no sequence end"); + payload.entries = retained; + payload.endCursor = { epoch: payload.epoch, seq: lastSeq }; + payload.hasNewer = false; + if (payload.window && typeof payload.window === "object") { + (payload.window as Record).maxSeq = lastSeq; + (payload.window as Record).nextSeq = lastSeq + 1; + } + heldServerMessage.message = JSON.stringify(envelope); + }, + setServerMessageSuppressed(type: string, suppressed: boolean): void { + if (suppressed) { + suppressedServerMessageTypes.add(type); + } else { + suppressedServerMessageTypes.delete(type); + } + }, + setAssistantMessageIdsStripped(stripped: boolean): void { + stripAssistantMessageIds = stripped; + }, setAgentStreamSuppressed(suppressed: boolean): void { suppressAgentStream = suppressed; },