mirror of
https://github.com/getpaseo/paseo.git
synced 2026-11-06 16:34:36 +00:00
test(app): cover canonical submission races in browser
This commit is contained in:
Mohamed Boudra
@@ -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<void> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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);
|
||||
});
|
||||
});
|
||||
|
||||
@@ -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<string, unknown> } } };
|
||||
payload?: { event?: { type?: unknown; item?: Record<string, unknown> } };
|
||||
};
|
||||
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<string, unknown> };
|
||||
payload?: Record<string, unknown>;
|
||||
};
|
||||
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<string>();
|
||||
const activeSockets = new Set<WebSocketRoute>();
|
||||
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<string, unknown> };
|
||||
payload?: Record<string, unknown>;
|
||||
};
|
||||
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<void> {
|
||||
if (heldServerMessage) return Promise.resolve();
|
||||
return new Promise<void>((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<string, unknown> };
|
||||
payload?: Record<string, unknown>;
|
||||
};
|
||||
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<string, unknown>).maxSeq = lastSeq;
|
||||
(payload.window as Record<string, unknown>).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;
|
||||
},
|
||||
|
||||
Reference in New Issue
Block a user