From 3063f2868bf5768a962eb16a27a18bb806ffa1a8 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Mon, 27 Jul 2026 15:22:57 +0200 Subject: [PATCH] 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. --- .../app/e2e/agent-message-submission.spec.ts | 146 ++++++++++++++++++ .../app/e2e/helpers/agent-message-gate.ts | 38 +++-- .../app/e2e/helpers/daemon-websocket-gate.ts | 99 +++++++++++- .../app/e2e/rewind-menu.ui-contract.spec.ts | 68 ++++++++ packages/app/src/composer/actions.test.ts | 57 +------ .../app/src/composer/submission/model.test.ts | 83 +++++----- packages/app/src/composer/submission/model.ts | 101 ++++++------ .../app/src/runtime/replica-cache/index.ts | 11 +- packages/app/src/stores/session-store.ts | 47 +++--- packages/app/src/types/stream.ts | 11 +- .../src/utils/agent-directory-sync.test.ts | 12 +- 11 files changed, 481 insertions(+), 192 deletions(-) diff --git a/packages/app/e2e/agent-message-submission.spec.ts b/packages/app/e2e/agent-message-submission.spec.ts index ea10c8b69e..e3c51e74e3 100644 --- a/packages/app/e2e/agent-message-submission.spec.ts +++ b/packages/app/e2e/agent-message-submission.spec.ts @@ -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>; + agent: Awaited>; +} + 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 { @@ -219,6 +242,79 @@ async function retryRestoredSubmission(page: Page, prompt: string): Promise { + await fillComposerDraft(page, prompt); + await sendDraftToQueue(page); +} + +async function expectQueuedSendFailuresRestored(page: Page, prompts: string[]): Promise { + 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 { + 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 { + 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 { + 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 { 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); + }); }); diff --git a/packages/app/e2e/helpers/agent-message-gate.ts b/packages/app/e2e/helpers/agent-message-gate.ts index ec9d4a346e..866d257706 100644 --- a/packages/app/e2e/helpers/agent-message-gate.ts +++ b/packages/app/e2e/helpers/agent-message-gate.ts @@ -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((resolve) => { - resolveRequest = resolve; - }); + let browserSocket: WebSocketRoute | null = null; + const heldMessages: Array = []; + 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 => { + while (requests.length < count) { + await new Promise((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 { + if (!browserSocket) throw new Error("No browser daemon socket to disconnect"); + await browserSocket.close({ code: 1008, reason: "Dropped by submission test." }); }, }; } diff --git a/packages/app/e2e/helpers/daemon-websocket-gate.ts b/packages/app/e2e/helpers/daemon-websocket-gate.ts index a6c7f38675..d6e56df6d1 100644 --- a/packages/app/e2e/helpers/daemon-websocket-gate.ts +++ b/packages/app/e2e/helpers/daemon-websocket-gate.ts @@ -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(); 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(); + const serverMessageCounts = new Map(); + 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 }; + 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; + } + } + 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 { + if (heldClientRequest) return Promise.resolve(); + return new Promise((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 { + while ((serverMessageCounts.get(type) ?? 0) < count) { + await new Promise((resolve) => serverMessageWaiters.add(resolve)); + } + }, }; } diff --git a/packages/app/e2e/rewind-menu.ui-contract.spec.ts b/packages/app/e2e/rewind-menu.ui-contract.spec.ts index 158f40ae18..982fd314f8 100644 --- a/packages/app/e2e/rewind-menu.ui-contract.spec.ts +++ b/packages/app/e2e/rewind-menu.ui-contract.spec.ts @@ -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 await expect(userMessage(page, text)).toBeVisible(); } +async function rewriteCachedMessageAsLegacyRow(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 }>; + }; + 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> } | 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."; diff --git a/packages/app/src/composer/actions.test.ts b/packages/app/src/composer/actions.test.ts index d45ff769ee..349d15d493 100644 --- a/packages/app/src/composer/actions.test.ts +++ b/packages/app/src/composer/actions.test.ts @@ -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 + Map >(); 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 = new Map(), ): QueueWriter & { state: Map } { @@ -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({ diff --git a/packages/app/src/composer/submission/model.test.ts b/packages/app/src/composer/submission/model.test.ts index 1a18dfe724..a1ef7ea1b6 100644 --- a/packages/app/src/composer/submission/model.test.ts +++ b/packages/app/src/composer/submission/model.test.ts @@ -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, diff --git a/packages/app/src/composer/submission/model.ts b/packages/app/src/composer/submission/model.ts index b9147fe0e1..c9dc60e9e5 100644 --- a/packages/app/src/composer/submission/model.ts +++ b/packages/app/src/composer/submission/model.ts @@ -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, }, }; } diff --git a/packages/app/src/runtime/replica-cache/index.ts b/packages/app/src/runtime/replica-cache/index.ts index 7f8582884e..faafe00dd8 100644 --- a/packages/app/src/runtime/replica-cache/index.ts +++ b/packages/app/src/runtime/replica-cache/index.ts @@ -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, }; diff --git a/packages/app/src/stores/session-store.ts b/packages/app/src/stores/session-store.ts index 89beef3b89..71f53ce8f8 100644 --- a/packages/app/src/stores/session-store.ts +++ b/packages/app/src/stores/session-store.ts @@ -376,7 +376,7 @@ export interface SessionState { // Stream state (head/tail model) agentStreamTail: Map; agentStreamHead: Map; - messageSubmissions: Map; + messageSubmissions: Map; agentTimelineCursor: Map; agentTimelineHasOlder: Map; agentTimelineOlderFetchInFlight: Map; @@ -594,22 +594,22 @@ const agentLastActivityCoalescer = createAgentLastActivityCoalescer(); function applyAgentStateToMessageSubmissions(input: { previousAgents: Map; nextAgents: Map; - submissions: Map; -}): Map { + submissions: Map; +}): Map { 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()( { 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()( }, ); 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()( 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()( 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: { diff --git a/packages/app/src/types/stream.ts b/packages/app/src/types/stream.ts index 7e3bfcd218..f7410c2b2b 100644 --- a/packages/app/src/types/stream.ts +++ b/packages/app/src/types/stream.ts @@ -374,7 +374,6 @@ export function replaceWithCanonicalStream( item.kind === "user_message" && item.clientMessageId !== undefined, ); let nextHead = input.previousHead; - const matchedHeadClientMessageIds = new Set(); 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 diff --git a/packages/app/src/utils/agent-directory-sync.test.ts b/packages/app/src/utils/agent-directory-sync.test.ts index 49afa03a70..18301c9270 100644 --- a/packages/app/src/utils/agent-directory-sync.test.ts +++ b/packages/app/src/utils/agent-directory-sync.test.ts @@ -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,