puter/paseo
1
0
Fork 0
mirror of https://github.com/getpaseo/paseo.git synced 2026-11-06 08:24:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(app): preserve canonical prompt order

Browse Source
This commit is contained in:
Mohamed Boudra
2026-07-27 19:50:26 +02:00
parent 73119d0f36
commit e8d151015b
4 changed files with 94 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/app/e2e/agent-message-submission.spec.ts
+55
View File
@@ -454,6 +454,55 @@ async function expectStaleCanonicalPagePreservesNewerLiveOutput(
}
}
async function expectCanonicalOrderWinsAcrossOverlappingClients(
page: Page,
testInfo: { workerIndex: number },
): Promise<void> {
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<void> {
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);
});
});
packages/app/e2e/helpers/daemon-websocket-gate.ts
+3
View File
@@ -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<void> {
while ((serverMessageCounts.get(type) ?? 0) < count) {
await new Promise<void>((resolve) => serverMessageWaiters.add(resolve));
packages/app/src/timeline/session-stream-reducers.test.ts
+9 -9
View File
@@ -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", () => {
packages/app/src/types/stream.ts
+27 -17
View File
@@ -367,6 +367,25 @@ export function upsertUserMessageAcrossStream(
};
}
function placeCanonicalUserMessageAtTail(
tail: StreamItem[],
message: UserMessageItem,
insertWhenUnmatched: boolean,
): Pick<UserMessageProductionResult, "items" | "message" | "matched"> {
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]
: [],
};
}