diff --git a/packages/app/e2e/agent-message-submission.spec.ts b/packages/app/e2e/agent-message-submission.spec.ts index 509f408de7..092159b9bb 100644 --- a/packages/app/e2e/agent-message-submission.spec.ts +++ b/packages/app/e2e/agent-message-submission.spec.ts @@ -454,6 +454,55 @@ async function expectStaleCanonicalPagePreservesNewerLiveOutput( } } +async function expectCanonicalOrderWinsAcrossOverlappingClients( + page: Page, + testInfo: { workerIndex: number }, +): Promise { + const gate = await installDaemonWebSocketGate(page); + const agent = await seedMockAgentWorkspace({ + repoPrefix: `submission-cross-client-order-${testInfo.workerIndex}-`, + title: "Cross-client submission order", + model: "ten-second-stream", + }); + const localPrompt = "Send this after the other client turn."; + const remotePrompt = "Commit this other client turn first."; + try { + await openAgentRoute(page, { workspaceId: agent.workspaceId, agentId: agent.agentId }); + await expectComposerVisible(page); + await expectAgentIdle(page); + gate.holdNextClientRequest("send_agent_message_request"); + const localRow = await submitMessageWithImage(page, localPrompt); + await gate.waitForHeldClientRequest(); + + await agent.client.sendAgentMessage(agent.agentId, remotePrompt); + await agent.client.waitForFinish(agent.agentId, 30_000); + const remoteRow = page.getByTestId("user-message").filter({ hasText: remotePrompt }); + await expect(remoteRow).toBeVisible(); + + const userMessageCount = gate.getAgentStreamItemCount("user_message"); + gate.releaseHeldClientRequest(); + await gate.waitForAgentStreamItem("user_message", userMessageCount + 1); + await expect(localRow).toHaveAttribute("aria-busy", "false"); + await expect(localRow.getByRole("button", { name: "Open image attachment" })).toBeVisible(); + await expect + .poll(async () => { + const localElement = await localRow.elementHandle(); + if (!localElement) return false; + return remoteRow.evaluate( + (remoteElement, localNode) => + Boolean( + remoteElement.compareDocumentPosition(localNode) & Node.DOCUMENT_POSITION_FOLLOWING, + ), + localElement, + ); + }) + .toBe(true); + } 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"); @@ -627,4 +676,10 @@ test.describe("Agent message submission", () => { test.setTimeout(90_000); await expectStaleCanonicalPagePreservesNewerLiveOutput(page, testInfo); }); + + test("uses canonical order when another client turn overtakes a held submission", async ({ + page, + }, testInfo) => { + await expectCanonicalOrderWinsAcrossOverlappingClients(page, testInfo); + }); }); diff --git a/packages/app/e2e/helpers/daemon-websocket-gate.ts b/packages/app/e2e/helpers/daemon-websocket-gate.ts index 9d18c398d1..c1e57f7adc 100644 --- a/packages/app/e2e/helpers/daemon-websocket-gate.ts +++ b/packages/app/e2e/helpers/daemon-websocket-gate.ts @@ -403,6 +403,9 @@ export async function installDaemonWebSocketGate(page: Page) { getClientRequestCount(type: string): number { return clientRequestCounts.get(type) ?? 0; }, + getAgentStreamItemCount(type: string): number { + return agentStreamItemCounts.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/src/timeline/session-stream-reducers.test.ts b/packages/app/src/timeline/session-stream-reducers.test.ts index 0b321aa655..10eed0188c 100644 --- a/packages/app/src/timeline/session-stream-reducers.test.ts +++ b/packages/app/src/timeline/session-stream-reducers.test.ts @@ -556,7 +556,7 @@ describe("processTimelineResponse", () => { expect(result.tail).toEqual([]); }); - it("does not move an unmatched submission during timeline replacement", () => { + it("keeps an unmatched submission after the canonical replacement range", () => { const unmatched = makeSubmittedUserMessage("first submission", "client-first"); const acknowledged: StreamItem[] = [ { @@ -604,10 +604,10 @@ describe("processTimelineResponse", () => { }, }, { - ...makeTimelineEntry(4, "response to all three submissions"), + ...makeTimelineEntry(4, "response to canonical submissions"), item: { type: "assistant_message", - text: "response to all three submissions", + text: "response to canonical submissions", messageId: "assistant-response", }, }, @@ -622,14 +622,14 @@ describe("processTimelineResponse", () => { text: "text" in item ? item.text : undefined, })), ).toEqual([ - { kind: "user_message", id: "client-first", text: "first submission" }, { kind: "user_message", id: "provider-second", text: "second submission" }, { kind: "user_message", id: "provider-third", text: "third submission" }, { kind: "assistant_message", id: "assistant-response", - text: "response to all three submissions", + text: "response to canonical submissions", }, + { kind: "user_message", id: "client-first", text: "first submission" }, ]); }); @@ -1195,7 +1195,7 @@ describe("processTimelineResponse", () => { ]); }); - it("acknowledges a head prompt in place while catch-up history arrives", () => { + it("moves an acknowledged head prompt to its catch-up sequence position", () => { const prompt = makeSubmittedUserMessage("New prompt", "new-prompt"); const result = processTimelineResponse({ @@ -1227,8 +1227,8 @@ describe("processTimelineResponse", () => { expect([...result.tail, ...result.head].map((item) => item.kind)).toEqual([ "assistant_message", - "user_message", "tool_call", + "user_message", ]); expect( [...result.tail, ...result.head] @@ -1427,7 +1427,7 @@ describe("processTimelineResponse", () => { ]); }); - it("acknowledges a local prompt in place when a remote user row also arrives", () => { + it("places a local prompt after an earlier remote canonical row", () => { const prompt = makeSubmittedUserMessage("Local prompt", "local-prompt"); const result = processTimelineResponse({ @@ -1462,7 +1462,7 @@ describe("processTimelineResponse", () => { expect( result.tail.filter((item) => item.kind === "user_message").map((item) => item.text), - ).toEqual(["Local prompt", "Remote prompt"]); + ).toEqual(["Remote prompt", "Local prompt"]); }); it("keeps an unmatched submitted prompt when catch-up contains only a remote user row", () => { diff --git a/packages/app/src/types/stream.ts b/packages/app/src/types/stream.ts index 2b39e498ce..8b66ab80bd 100644 --- a/packages/app/src/types/stream.ts +++ b/packages/app/src/types/stream.ts @@ -367,6 +367,25 @@ export function upsertUserMessageAcrossStream( }; } +function placeCanonicalUserMessageAtTail( + tail: StreamItem[], + message: UserMessageItem, + insertWhenUnmatched: boolean, +): Pick { + const produced = produceUserMessage(tail, message, null, "existing"); + if (!produced.matched && !insertWhenUnmatched) { + return produced; + } + const preceding = produced.matched + ? [...produced.items.slice(0, produced.index), ...produced.items.slice(produced.index + 1)] + : produced.items; + return { + items: [...preceding, produced.message], + message: produced.message, + matched: produced.matched, + }; +} + export interface CanonicalStreamReplacementInput { canonical: StreamItem[]; previousTail: StreamItem[]; @@ -483,10 +502,7 @@ export function replaceWithCanonicalStream( if (!local.clientMessageId || !sendingClientMessageIds.has(local.clientMessageId)) { continue; } - const insertionIndex = nextTail.findIndex( - (item) => item.timestamp.getTime() > local.timestamp.getTime(), - ); - nextTail.splice(insertionIndex < 0 ? nextTail.length : insertionIndex, 0, local); + nextTail.push(local); } nextHead = nextHead.filter((item) => { @@ -1510,21 +1526,15 @@ function applyCanonicalUserMessageEvent(params: { text: normalized.chunk, timestamp, }); - const reconciled = upsertUserMessageAcrossStream({ - tail: flushedTail, - head: flushedHead, - message: canonical, - insert: normalized.hasContent ? "tail" : "none", - presentation: "existing", - }); + const reconciled = placeCanonicalUserMessageAtTail(flushedTail, canonical, normalized.hasContent); return { - tail: reconciled.tail, - head: reconciled.head, - changedTail: flushedTail !== tail || reconciled.changedTail, - changedHead: flushedHead !== head || reconciled.changedHead, + tail: reconciled.items, + head: flushedHead, + changedTail: flushedTail !== tail || reconciled.items !== flushedTail, + changedHead: flushedHead !== head, acknowledgedClientMessageIds: - reconciled.location?.matched && reconciled.location.message.clientMessageId - ? [reconciled.location.message.clientMessageId] + reconciled.matched && reconciled.message.clientMessageId + ? [reconciled.message.clientMessageId] : [], }; }