fix(creation): recover draft retries without accepting partial workspaces

This commit is contained in:
Mohamed Boudra
2026-09-14 10:38:49 +02:00
parent 7a09bd9f12
commit 4b8977aa67
12 changed files with 456 additions and 348 deletions
@@ -1,304 +1,92 @@
import { installDaemonWebSocketGate } from "../support/helpers/daemon-websocket-gate";
import { expect, test, type Page } from "../support/fixtures";
import type { SessionInboundMessage, SessionOutboundMessage } from "@getpaseo/protocol/messages";
import { gotoAppShell } from "../support/helpers/app";
import { gotoWorkspace } from "../support/helpers/launcher";
import { fillComposerDraft } from "../support/helpers/composer";
import { createAgentTabFromMenu } from "../support/helpers/workspace-tabs";
import { daemonWsRoutePattern } from "../support/helpers/daemon-port";
import {
delayBrowserAgentCreatedStatus,
openNewWorkspaceComposer,
selectWorkspaceIsolation,
} from "../support/helpers/new-workspace";
import { seedWorkspace } from "../support/helpers/seed-client";
import { waitForSidebarHydration } from "../support/helpers/workspace-ui";
async function pressSubmitBeforeTheNextRender(page: Page, name: string): Promise<void> {
const create = page.getByRole("button", { name, exact: true });
await expect(create).toBeEnabled();
// Dispatch the queued clicks in one JS task, before pending state can paint.
// Exercise DOM events rather than calling the app's submit handler directly.
await create.evaluate((button) => {
for (let click = 0; click < 3; click++) {
button.dispatchEvent(new MouseEvent("click", { bubbles: true }));
}
});
}
function observeCreationRequests(page: Page) {
const pending = new Set<string>();
page.on("websocket", (socket) => {
socket.on("framesent", ({ payload }) => {
const envelope = JSON.parse(payload.toString()) as { message?: SessionInboundMessage };
const request = envelope.message;
if (
request?.type === "workspace.create.request" ||
request?.type === "create_agent_request"
) {
pending.add(request.requestId);
}
});
socket.on("framereceived", ({ payload }) => {
const envelope = JSON.parse(payload.toString()) as { message?: SessionOutboundMessage };
const response = envelope.message;
if (
response?.type === "workspace.create.response" ||
(response?.type === "status" &&
(response.payload.status === "agent_created" ||
response.payload.status === "agent_create_failed"))
) {
const requestId = response.payload.requestId;
if (typeof requestId === "string") pending.delete(requestId);
}
});
});
return {
async settled() {
await expect.poll(() => pending.size).toBe(0);
},
};
}
async function retryNextAgentCreation(page: Page) {
const retryIds = new Set<string>();
const results: Array<{ status: string; agentId?: string }> = [];
let repeated = false;
await page.routeWebSocket(daemonWsRoutePattern(), (browser) => {
const server = browser.connectToServer();
browser.onMessage((frame) => {
const envelope = JSON.parse(frame.toString()) as { message?: SessionInboundMessage };
const request = envelope.message;
if (!repeated && request?.type === "create_agent_request") {
repeated = true;
for (let attempt = 1; attempt <= 3; attempt++) {
const requestId = attempt === 1 ? request.requestId : `${request.requestId}-${attempt}`;
retryIds.add(requestId);
// Keep the app's operation key and payload; only RPC correlation changes.
server.send(JSON.stringify({ ...envelope, message: { ...request, requestId } }));
}
return;
}
server.send(frame);
});
server.onMessage((frame) => {
const envelope = JSON.parse(frame.toString()) as { message?: SessionOutboundMessage };
const response = envelope.message;
if (
response?.type === "status" &&
(response.payload.status === "agent_created" ||
response.payload.status === "agent_create_failed") &&
typeof response.payload.requestId === "string" &&
retryIds.delete(response.payload.requestId)
) {
results.push(response.payload);
}
browser.send(frame);
});
});
return {
async completedAgentIds() {
await expect.poll(() => results.length).toBe(3);
expect(results.map((result) => result.status)).toEqual([
"agent_created",
"agent_created",
"agent_created",
]);
return results.map((result) => result.agentId);
},
};
}
import { expect } from "../support/fixtures";
import { test } from "../support/creation-fixtures";
for (const isolation of ["local", "worktree"] as const) {
test(`repeated Create clicks before a render create only one ${isolation} workspace`, async ({
page,
creation,
}) => {
const requests = observeCreationRequests(page);
const project = await seedWorkspace({ repoPrefix: "creation-idempotency-" });
try {
await gotoAppShell(page);
await waitForSidebarHydration(page);
await openNewWorkspaceComposer(page, project);
await selectWorkspaceIsolation(page, isolation);
await pressSubmitBeforeTheNextRender(page, "Create");
await expect(page).toHaveURL(/\/workspace\//);
await requests.settled();
const workspaces = await project.client.fetchWorkspaces();
expect(
workspaces.entries
.filter(
(workspace) =>
workspace.projectId === project.projectId && workspace.id !== project.workspaceId,
)
.map((workspace) => workspace.id),
).toHaveLength(1);
} finally {
await project.cleanup();
}
await creation.openWorkspaceForm(isolation);
await creation.submitRepeatedly("Create");
await creation.expectOneCreatedWorkspace();
});
}
test("repeated Send clicks before a render create only one agent", async ({ page }) => {
const requests = observeCreationRequests(page);
const project = await seedWorkspace({ repoPrefix: "agent-creation-idempotency-" });
try {
await gotoWorkspace(page, project.workspaceId);
await createAgentTabFromMenu(page);
await fillComposerDraft(page, "Start exactly one agent for this prompt.");
await pressSubmitBeforeTheNextRender(page, "Send message");
await expect(page.getByTestId("user-message").first()).toBeVisible();
await requests.settled();
const agents = await project.client.fetchAgents();
expect(
agents.entries
.filter(({ agent }) => agent.workspaceId === project.workspaceId)
.map(({ agent }) => agent.id),
).toHaveLength(1);
} finally {
await project.cleanup();
}
test("repeated Send clicks before a render create only one agent", async ({ creation }) => {
await creation.openAgentDraft();
await creation.submitRepeatedly("Send message", "Start exactly one agent for this prompt.");
await creation.expectPromptVisible();
await creation.expectAgentCount(1);
});
test("retrying the app's agent creation returns the same agent for every attempt", async ({
page,
creation,
agentRetries,
}) => {
const retries = await retryNextAgentCreation(page);
const project = await seedWorkspace({ repoPrefix: "agent-creation-retries-" });
try {
await gotoWorkspace(page, project.workspaceId);
await createAgentTabFromMenu(page);
await fillComposerDraft(page, "Start exactly one agent for this prompt.");
await page.getByRole("button", { name: "Send message", exact: true }).click();
const agentIds = await retries.completedAgentIds();
expect(new Set(agentIds).size).toBe(1);
const agents = await project.client.fetchAgents();
expect(
agents.entries.filter(({ agent }) => agent.workspaceId === project.workspaceId),
).toHaveLength(1);
} finally {
await project.cleanup();
}
await creation.openAgentDraft();
await creation.submitPrompt("Start exactly one agent for this prompt.");
expect(new Set(await agentRetries.completedAgentIds()).size).toBe(1);
await creation.expectAgentCount(1);
});
test("the first prompt still names an agent created with a receipt", async ({ page }) => {
const project = await seedWorkspace({ repoPrefix: "creation-title-" });
try {
await gotoWorkspace(page, project.workspaceId);
await createAgentTabFromMenu(page);
await fillComposerDraft(page, "Name this new agent from its first prompt.");
await page.getByRole("button", { name: "Send message", exact: true }).click();
await expect(page.getByTestId("user-message").first()).toBeVisible();
await expect
.poll(async () => {
const agents = await project.client.fetchAgents();
return agents.entries.find(({ agent }) => agent.workspaceId === project.workspaceId)?.agent
.title;
})
.toBe("Name this new agent from its first prompt.");
} finally {
await project.cleanup();
}
test("the first prompt still names an agent created with a receipt", async ({ creation }) => {
await creation.openAgentDraft();
await creation.submitPrompt("Name this new agent from its first prompt.");
await creation.expectPromptVisible();
await creation.expectAgentTitle("Name this new agent from its first prompt.");
});
test("retrying after a lost first-prompt acknowledgement reuses the agent and message", async ({
page,
creation,
promptRetry,
}) => {
const gate = await installDaemonWebSocketGate(page);
const project = await seedWorkspace({ repoPrefix: "creation-lost-ack-" });
try {
await gotoWorkspace(page, project.workspaceId);
await createAgentTabFromMenu(page);
gate.holdNextServerMessage("send_agent_message_response");
await fillComposerDraft(page, "Deliver this initial prompt once.");
await page.getByRole("button", { name: "Send message", exact: true }).click();
await gate.waitForHeldServerMessage("send_agent_message_response");
const firstMessage = gate.getClientRequests("send_agent_message_request").at(-1);
const agentFetches = gate.getClientRequestCount("fetch_agents_request");
await gate.drop();
gate.restore();
await expect
.poll(() => gate.getClientRequestCount("fetch_agents_request"))
.toBeGreaterThan(agentFetches);
await expect(page.getByRole("button", { name: "Send message", exact: true })).toBeEnabled();
await fillComposerDraft(page, "Deliver this initial prompt once.");
await page.getByRole("button", { name: "Send message", exact: true }).click();
await expect.poll(() => gate.getClientRequestCount("send_agent_message_request")).toBe(2);
const secondMessage = gate.getClientRequests("send_agent_message_request").at(-1);
expect(secondMessage).toMatchObject({
agentId: firstMessage?.agentId,
messageId: firstMessage?.messageId,
});
await expect(page.getByTestId("user-message")).toHaveCount(1);
const agents = await project.client.fetchAgents();
expect(
agents.entries.filter(({ agent }) => agent.workspaceId === project.workspaceId),
).toHaveLength(1);
} finally {
gate.restore();
await project.cleanup();
}
await creation.openAgentDraft();
promptRetry.holdAcknowledgement();
await creation.submitPrompt("Deliver this initial prompt once.");
await promptRetry.waitForDeliveredPrompt();
await promptRetry.disconnectAndReconnect();
await creation.submitPrompt("Deliver this initial prompt once.");
await promptRetry.expectSameAgentAndMessage();
await creation.expectAgentCount(1);
});
test("a remounted draft unlocks when its original submission fails", async ({
creation,
promptRetry,
}) => {
await creation.openAgentDraft();
promptRetry.holdAcknowledgement();
await creation.submitPrompt("Deliver this initial prompt once.");
await promptRetry.waitForDeliveredPrompt();
await creation.evictAndReturnToDraft();
await creation.expectPromptVisible("Deliver this initial prompt once.");
await promptRetry.disconnectAndReconnect();
await creation.submitPrompt("Deliver this initial prompt once.");
await promptRetry.expectSameAgentAndMessage();
await creation.expectAgentCount(1);
});
test("repeated new-workspace prompt submissions create one workspace and one agent", async ({
page,
creation,
}) => {
const requests = observeCreationRequests(page);
const project = await seedWorkspace({ repoPrefix: "creation-workspace-prompt-" });
try {
await gotoAppShell(page);
await waitForSidebarHydration(page);
await openNewWorkspaceComposer(page, project);
await selectWorkspaceIsolation(page, "local");
await fillComposerDraft(page, "Start one agent in one new workspace.");
await pressSubmitBeforeTheNextRender(page, "Create");
await expect(page).toHaveURL(/\/workspace\//);
await expect(page.getByTestId("user-message").first()).toBeVisible();
await requests.settled();
const workspaces = (await project.client.fetchWorkspaces()).entries.filter(
(workspace) =>
workspace.projectId === project.projectId && workspace.id !== project.workspaceId,
);
expect(workspaces).toHaveLength(1);
const agents = await project.client.fetchAgents();
expect(
agents.entries.filter(({ agent }) => agent.workspaceId === workspaces[0]!.id),
).toHaveLength(1);
} finally {
await project.cleanup();
}
await creation.openWorkspaceForm("local");
await creation.submitRepeatedly("Create", "Start one agent in one new workspace.");
await creation.expectPromptVisible();
await creation.expectOneCreatedWorkspace();
await creation.expectAgentCount(1);
});
test("separate drafts can intentionally create two agents in the same workspace", async ({
page,
creation,
delayedCreation,
}) => {
const delayed = await delayBrowserAgentCreatedStatus(page);
const project = await seedWorkspace({ repoPrefix: "creation-distinct-drafts-" });
try {
await gotoWorkspace(page, project.workspaceId);
for (const prompt of ["Start the first agent.", "Start the second agent."]) {
await createAgentTabFromMenu(page);
await fillComposerDraft(page, prompt);
await page.getByRole("button", { name: "Send message", exact: true }).click();
await expect(page.getByTestId("user-message").filter({ hasText: prompt })).toBeVisible();
await delayed.waitForDelayedCreatedStatus();
}
await expect
.poll(
async () =>
(await project.client.fetchAgents()).entries.filter(
({ agent }) => agent.workspaceId === project.workspaceId,
).length,
)
.toBe(2);
delayed.release();
} finally {
delayed.release();
await project.cleanup();
}
await creation.openAgentDraft();
await creation.submitPrompt("Start the first agent.");
await creation.expectPromptVisible("Start the first agent.");
await delayedCreation.waitForDelayedCreatedStatus();
await creation.startAnotherDraft();
await creation.submitPrompt("Start the second agent.");
await creation.expectPromptVisible("Start the second agent.");
await delayedCreation.waitForDelayedCreatedStatus();
delayedCreation.release();
await creation.expectAgentCount(2);
});
@@ -0,0 +1,43 @@
import { test as base } from "./fixtures";
import {
createCreationScenario,
createPromptRetryScenario,
retryNextAgentCreation,
} from "./helpers/creation";
import { installDaemonWebSocketGate } from "./helpers/daemon-websocket-gate";
import { delayBrowserAgentCreatedStatus } from "./helpers/new-workspace";
export const test = base.extend<{
creation: Awaited<ReturnType<typeof createCreationScenario>>;
promptRetry: ReturnType<typeof createPromptRetryScenario>;
agentRetries: Awaited<ReturnType<typeof retryNextAgentCreation>>;
delayedCreation: Awaited<ReturnType<typeof delayBrowserAgentCreatedStatus>>;
}>({
creation: async ({ page }, provide) => {
const creation = await createCreationScenario(page);
try {
await provide(creation);
} finally {
await creation.cleanup();
}
},
promptRetry: async ({ page }, provide) => {
const gate = await installDaemonWebSocketGate(page);
try {
await provide(createPromptRetryScenario(page, gate));
} finally {
gate.restore();
}
},
agentRetries: async ({ page }, provide) => {
await provide(await retryNextAgentCreation(page));
},
delayedCreation: async ({ page }, provide) => {
const delayed = await delayBrowserAgentCreatedStatus(page);
try {
await provide(delayed);
} finally {
delayed.release();
}
},
});
@@ -0,0 +1,234 @@
import { expect, type Page } from "@playwright/test";
import type { SessionInboundMessage, SessionOutboundMessage } from "@getpaseo/protocol/messages";
import { daemonWsRoutePattern } from "./daemon-port";
import { gotoAppShell } from "./app";
import { gotoWorkspace } from "./launcher";
import { fillComposerDraft } from "./composer";
import { createAgentTabFromMenu } from "./workspace-tabs";
import { openNewWorkspaceComposer, selectWorkspaceIsolation } from "./new-workspace";
import { seedWorkspace } from "./seed-client";
import {
waitForSidebarHydration,
switchWorkspaceViaSidebar,
workspaceDeckEntryLocator,
} from "./workspace-ui";
import { getServerId } from "./server-id";
import { WORKSPACE_DECK_MAX_MOUNTED_WORKSPACES } from "@/screens/workspace/workspace-deck-retention";
import type { installDaemonWebSocketGate } from "./daemon-websocket-gate";
export async function pressSubmitBeforeTheNextRender(page: Page, name: string): Promise<void> {
const create = page.getByRole("button", { name, exact: true });
await expect(create).toBeEnabled();
// Dispatch the queued clicks in one JS task, before pending state can paint.
// Exercise DOM events rather than calling the app's submit handler directly.
await create.evaluate((button) => {
for (let click = 0; click < 3; click++) {
button.dispatchEvent(new MouseEvent("click", { bubbles: true }));
}
});
}
export function observeCreationRequests(page: Page) {
const pending = new Set<string>();
page.on("websocket", (socket) => {
socket.on("framesent", ({ payload }) => {
const envelope = JSON.parse(payload.toString()) as { message?: SessionInboundMessage };
const request = envelope.message;
if (
request?.type === "workspace.create.request" ||
request?.type === "create_agent_request"
) {
pending.add(request.requestId);
}
});
socket.on("framereceived", ({ payload }) => {
const envelope = JSON.parse(payload.toString()) as { message?: SessionOutboundMessage };
const response = envelope.message;
if (
response?.type === "workspace.create.response" ||
(response?.type === "status" &&
(response.payload.status === "agent_created" ||
response.payload.status === "agent_create_failed"))
) {
const requestId = response.payload.requestId;
if (typeof requestId === "string") pending.delete(requestId);
}
});
});
return {
async settled() {
await expect.poll(() => pending.size).toBe(0);
},
};
}
export async function retryNextAgentCreation(page: Page) {
const retryIds = new Set<string>();
const results: Array<{ status: string; agentId?: string }> = [];
let repeated = false;
await page.routeWebSocket(daemonWsRoutePattern(), (browser) => {
const server = browser.connectToServer();
browser.onMessage((frame) => {
const envelope = JSON.parse(frame.toString()) as { message?: SessionInboundMessage };
const request = envelope.message;
if (!repeated && request?.type === "create_agent_request") {
repeated = true;
for (let attempt = 1; attempt <= 3; attempt++) {
const requestId = attempt === 1 ? request.requestId : `${request.requestId}-${attempt}`;
retryIds.add(requestId);
// Keep the app's operation key and payload; only RPC correlation changes.
server.send(JSON.stringify({ ...envelope, message: { ...request, requestId } }));
}
return;
}
server.send(frame);
});
server.onMessage((frame) => {
const envelope = JSON.parse(frame.toString()) as { message?: SessionOutboundMessage };
const response = envelope.message;
if (
response?.type === "status" &&
(response.payload.status === "agent_created" ||
response.payload.status === "agent_create_failed") &&
typeof response.payload.requestId === "string" &&
retryIds.delete(response.payload.requestId)
) {
results.push(response.payload);
}
browser.send(frame);
});
});
return {
async completedAgentIds() {
await expect.poll(() => results.length).toBe(3);
expect(results.map((result) => result.status)).toEqual([
"agent_created",
"agent_created",
"agent_created",
]);
return results.map((result) => result.agentId);
},
};
}
export async function createCreationScenario(page: Page) {
const requests = observeCreationRequests(page);
const project = await seedWorkspace({ repoPrefix: "creation-idempotency-" });
let workspaceId = project.workspaceId;
return {
cleanup: project.cleanup,
async openWorkspaceForm(isolation: "local" | "worktree") {
await gotoAppShell(page);
await waitForSidebarHydration(page);
await openNewWorkspaceComposer(page, project);
await selectWorkspaceIsolation(page, isolation);
},
async openAgentDraft() {
await gotoWorkspace(page, project.workspaceId);
await createAgentTabFromMenu(page);
},
async startAnotherDraft() {
await createAgentTabFromMenu(page);
},
async submitPrompt(prompt: string, button = "Send message") {
await fillComposerDraft(page, prompt);
await page.getByRole("button", { name: button, exact: true }).click();
},
async submitRepeatedly(button: string, prompt?: string) {
if (prompt) await fillComposerDraft(page, prompt);
await pressSubmitBeforeTheNextRender(page, button);
},
async expectPromptVisible(prompt?: string) {
const rows = page.getByTestId("user-message");
await expect(prompt ? rows.filter({ hasText: prompt }) : rows.first()).toBeVisible();
},
async expectOneCreatedWorkspace() {
await expect(page).toHaveURL(/\/workspace\//);
await requests.settled();
const workspaces = (await project.client.fetchWorkspaces()).entries.filter(
(workspace) =>
workspace.projectId === project.projectId && workspace.id !== project.workspaceId,
);
expect(workspaces).toHaveLength(1);
workspaceId = workspaces[0]!.id;
},
async expectAgentCount(count: number) {
await requests.settled();
await expect
.poll(
async () =>
(await project.client.fetchAgents()).entries.filter(
({ agent }) => agent.workspaceId === workspaceId,
).length,
)
.toBe(count);
},
async expectAgentTitle(title: string) {
await expect
.poll(
async () =>
(await project.client.fetchAgents()).entries.find(
({ agent }) => agent.workspaceId === workspaceId,
)?.agent.title,
)
.toBe(title);
},
async evictAndReturnToDraft() {
const draft = page
.locator('[data-testid^="workspace-tab-draft_"][aria-selected="true"]')
.filter({ visible: true })
.first();
const draftTestId = await draft.getAttribute("data-testid");
expect(draftTestId).not.toBeNull();
for (let index = 0; index < WORKSPACE_DECK_MAX_MOUNTED_WORKSPACES; index++) {
const result = await project.client.createWorkspace({
source: { kind: "directory", path: project.repoPath },
});
if (!result.workspace) throw new Error(result.error ?? "Failed to seed eviction workspace");
await switchWorkspaceViaSidebar({
page,
serverId: getServerId(),
workspaceId: result.workspace.id,
});
}
await expect(workspaceDeckEntryLocator(page, getServerId(), workspaceId)).toHaveCount(0);
await switchWorkspaceViaSidebar({ page, serverId: getServerId(), workspaceId });
await page
.getByTestId(draftTestId!)
.filter({ visible: true })
.click({ position: { x: 12, y: 13 } });
},
};
}
export function createPromptRetryScenario(
page: Page,
gate: Awaited<ReturnType<typeof installDaemonWebSocketGate>>,
) {
let firstMessage: ReturnType<typeof gate.getClientRequests>[number] | undefined;
return {
holdAcknowledgement() {
gate.holdNextServerMessage("send_agent_message_response");
},
async waitForDeliveredPrompt() {
await gate.waitForHeldServerMessage("send_agent_message_response");
firstMessage = gate.getClientRequests("send_agent_message_request").at(-1);
},
async disconnectAndReconnect() {
const fetches = gate.getClientRequestCount("fetch_agents_request");
await gate.drop();
gate.restore();
await expect
.poll(() => gate.getClientRequestCount("fetch_agents_request"))
.toBeGreaterThan(fetches);
},
async expectSameAgentAndMessage() {
await expect.poll(() => gate.getClientRequestCount("send_agent_message_request")).toBe(2);
expect(gate.getClientRequests("send_agent_message_request").at(-1)).toMatchObject({
agentId: firstMessage?.agentId,
messageId: firstMessage?.messageId,
});
await expect(page.getByTestId("user-message").filter({ visible: true })).toHaveCount(1);
},
};
}
@@ -250,40 +250,4 @@ describe("useDraftAgentCreateFlow", () => {
});
expect(onCreateSuccess).toHaveBeenCalledTimes(1);
});
it("unlocks a remounted draft when the original request fails", async () => {
let rejectCreation!: (error: Error) => void;
const creation = new Promise<never>((_resolve, reject) => {
rejectCreation = reject;
});
const options = {
draftId: "draft-remount",
getPendingServerId: () => "server-1",
buildDraftAgent: () => ({ provider: "codex" }),
createRequest: async () => creation,
onCreateSuccess: () => undefined,
};
const original = renderHook(() => useDraftAgentCreateFlow(options));
let submission!: Promise<unknown>;
const captureError = (error: unknown) => error;
await act(async () => {
submission = original.result.current
.handleCreateFromInput({ text: "hello", attachments: [], cwd: "/repo" })
.catch(captureError);
});
const pending = useCreateFlowStore.getState().pendingByDraftId[options.draftId]!;
const restored = renderHook(() =>
useDraftAgentCreateFlow({
...options,
initialAttempt: { ...pending, timestamp: new Date(pending.timestamp) },
}),
);
expect(restored.result.current.isSubmitting).toBe(true);
await act(async () => {
rejectCreation(new Error("Provider unavailable"));
await submission;
});
expect(restored.result.current.isSubmitting).toBe(false);
expect(restored.result.current.formErrorMessage).toBe("Provider unavailable");
});
});
+11 -8
View File
@@ -135,14 +135,17 @@ export function useDraftAgentCreateFlow<TDraftAgent, TCreateResult>({
);
const pending = useCreateFlowStore((state) => state.pendingByDraftId[draftId]);
// A remounted view observes completion from the same attempt as its original view.
const machine = useMemo<DraftAgentMachineState<TDraftAgent>>(
() =>
localMachine.tag === "creating" && pending?.lifecycle === "abandoned"
? { tag: "draft", errorMessage: pending.errorMessage ?? "" }
: localMachine,
[pending, localMachine],
);
// Remounts can precede model hydration. Rebuild the preview when its inputs
// arrive, and observe the original request's failure through shared state.
const machine = useMemo<DraftAgentMachineState<TDraftAgent>>(() => {
if (pending?.lifecycle === "abandoned") {
return { tag: "draft", errorMessage: pending.errorMessage ?? "" };
}
if (pending?.lifecycle === "active" && localMachine.tag === "draft" && initialAttempt) {
return prepareCreateAttempt(initialAttempt, buildDraftAgent);
}
return localMachine;
}, [pending, localMachine, initialAttempt, buildDraftAgent]);
const setPendingCreateAttempt = useCreateFlowStore((state) => state.setPending);
const updatePendingAgentId = useCreateFlowStore((state) => state.updateAgentId);
@@ -1068,8 +1068,6 @@ async function submitWorkspaceDraft(input: SubmitDraftInput): Promise<SubmitOutc
initialSetup,
} = input;
const draftId = draftIdInput?.trim() || generateDraftId();
const existing = useCreateFlowStore.getState().pendingByDraftId[draftId];
if (existing?.serverId === serverId) return;
const clientMessageId = `${draftId}:initial-message`;
const timestamp = Date.now();
const wirePayload = splitComposerAttachmentsForSubmit(attachments, {
@@ -1111,7 +1109,7 @@ async function submitWorkspaceDraft(input: SubmitDraftInput): Promise<SubmitOutc
return "background";
}
useCreateFlowStore.getState().setPending({
const started = useCreateFlowStore.getState().trySetPending({
serverId,
draftId,
workspaceId,
@@ -1122,6 +1120,7 @@ async function submitWorkspaceDraft(input: SubmitDraftInput): Promise<SubmitOutc
...(wirePayload.images.length > 0 ? { images: wirePayload.images } : {}),
...(wirePayload.attachments.length > 0 ? { attachments: wirePayload.attachments } : {}),
});
if (!started) return "navigated";
useWorkspaceDraftSubmissionStore.getState().setPending({
serverId,
workspaceId,
@@ -6,6 +6,33 @@ describe("create-flow-store", () => {
useCreateFlowStore.setState({ pendingByDraftId: {} });
});
it("accepts a retry after failure while rejecting repeated active or completed submissions", () => {
const store = useCreateFlowStore.getState();
const submission = {
draftId: "draft-retry",
serverId: "server-1",
workspaceId: "workspace-1",
agentId: null,
clientMessageId: "draft-retry:initial-message",
text: "hello",
timestamp: 1,
};
expect(store.trySetPending(submission)).toBe(true);
expect(store.trySetPending(submission)).toBe(false);
store.markLifecycle({
draftId: submission.draftId,
lifecycle: "abandoned",
errorMessage: "Disconnected",
});
expect(store.trySetPending(submission)).toBe(true);
expect(useCreateFlowStore.getState().pendingByDraftId[submission.draftId]).toEqual({
...submission,
lifecycle: "active",
});
store.markLifecycle({ draftId: submission.draftId, lifecycle: "sent" });
expect(store.trySetPending(submission)).toBe(false);
});
it("tracks lifecycle transitions explicitly", () => {
const store = useCreateFlowStore.getState();
store.setPending({
+8 -1
View File
@@ -35,6 +35,7 @@ export function isActiveCreateFlowForDraft(input: {
interface CreateFlowState {
pendingByDraftId: Record<string, PendingCreateAttempt>;
setPending: (pending: Omit<PendingCreateAttempt, "lifecycle">) => void;
trySetPending: (pending: Omit<PendingCreateAttempt, "lifecycle">) => boolean;
updateAgentId: (input: { draftId: string; agentId: string }) => void;
markLifecycle: (input: {
draftId: string;
@@ -47,7 +48,7 @@ interface CreateFlowState {
clearAll: () => void;
}
export const useCreateFlowStore = create<CreateFlowState>((set) => ({
export const useCreateFlowStore = create<CreateFlowState>((set, get) => ({
pendingByDraftId: {},
setPending: (pending) =>
set((state) => ({
@@ -59,6 +60,12 @@ export const useCreateFlowStore = create<CreateFlowState>((set) => ({
},
},
})),
trySetPending: (pending) => {
const existing = get().pendingByDraftId[pending.draftId];
if (existing?.serverId === pending.serverId && existing.lifecycle !== "abandoned") return false;
get().setPending(pending);
return true;
},
updateAgentId: ({ draftId, agentId }) =>
set((state) => {
const current = state.pendingByDraftId[draftId];
@@ -180,20 +180,22 @@ test("workspace receipts serialize concurrent requests and survive reconstructio
).not.toBe("wks_first");
});
test("a workspace persisted before a lost acknowledgement is recovered", async () => {
test("a workspace record left by failed provisioning is not mistaken for completion", async () => {
const { requests, directory } = await fixture();
const workspaces = new Set<string>();
const input = {
key: "lost-workspace",
key: "partial-workspace",
workspaceId: "wks_created",
request: {},
findWorkspace: async (id: string) => workspaces.has(id),
create: async (id: string) => {
workspaces.add(id);
throw new Error("acknowledgement lost");
throw new Error("checkout rolled back");
},
};
await expect(requests.createWorkspace(input)).rejects.toThrow("acknowledgement lost");
expect(await new RequestReceipts(directory).createWorkspace(input)).toBe("wks_created");
await expect(requests.createWorkspace(input)).rejects.toThrow("checkout rolled back");
await expect(new RequestReceipts(directory).createWorkspace(input)).rejects.toThrow(
"workspace_request_outcome_unknown",
);
expect([...workspaces]).toEqual(["wks_created"]);
});
@@ -46,7 +46,9 @@ export class RequestReceipts {
}): Promise<string> {
return this.execute(["create-workspace", input.key], input.request, {
resource: { kind: "workspace", id: input.workspaceId },
recover: input.findWorkspace,
// A registry entry can survive failed provisioning and checkout rollback.
// Only a completed receipt proves that the entire operation succeeded.
recover: async () => false,
run: input.create,
retrySafe: async (workspaceId) => !(await input.findWorkspace(workspaceId)),
});
+27 -15
View File
@@ -6350,19 +6350,6 @@ export class Session {
request: Extract<SessionInboundMessage, { type: "workspace.create.request" }>,
): Promise<void> {
try {
const create = async (workspaceId?: string) => {
let creationRequest = request;
// Hooks belong to the operation: retries fingerprint the caller's input
// and must not rerun hooks or compare their potentially changing output.
if (this.pluginRuntime) {
const { type, requestId, ...input } = request;
const transformed = await this.pluginRuntime.before("workspace.create", input);
creationRequest = { ...transformed, type, requestId };
}
return creationRequest.source.kind === "directory"
? this.handleWorkspaceCreateLocal(creationRequest, workspaceId)
: this.handleWorkspaceCreateWorktree(creationRequest, workspaceId);
};
let descriptor: WorkspaceDescriptorPayload | undefined;
if (request.idempotencyKey !== undefined) {
const { requestId: _requestId, idempotencyKey, ...payload } = request;
@@ -6372,16 +6359,24 @@ export class Session {
workspaceId: generateWorkspaceId(),
findWorkspace: async (id) => (await this.workspaceRegistry.get(id)) !== null,
create: async (id) => {
descriptor = await create(id);
descriptor = await this.createRequestedWorkspace(request, id);
},
});
if (!descriptor) {
const workspace = await this.workspaceRegistry.get(workspaceId);
if (!workspace) throw new Error("Previously created workspace no longer exists");
if (!(await this.filesystem.isDirectory(workspace.cwd))) {
throw new SessionRequestError(
"directory_not_found",
`Directory not found: ${workspace.cwd}`,
);
}
await this.syncWorkspaceGitObserverForWorkspace(workspace);
descriptor = await this.describeWorkspaceRecordWithGitData(workspace);
await this.emitCreatedWorkspaceUpdate(descriptor);
}
} else {
descriptor = await create();
descriptor = await this.createRequestedWorkspace(request);
}
const workspace = await this.workspaceRegistry.get(descriptor.id);
this.emit({
@@ -6423,6 +6418,23 @@ export class Session {
}
}
private async createRequestedWorkspace(
request: Extract<SessionInboundMessage, { type: "workspace.create.request" }>,
workspaceId?: string,
): Promise<WorkspaceDescriptorPayload> {
let creationRequest = request;
// Hooks belong to the operation: retries fingerprint the caller's input
// and must not rerun hooks or compare their potentially changing output.
if (this.pluginRuntime) {
const { type, requestId, ...input } = request;
const transformed = await this.pluginRuntime.before("workspace.create", input);
creationRequest = { ...transformed, type, requestId };
}
return creationRequest.source.kind === "directory"
? this.handleWorkspaceCreateLocal(creationRequest, workspaceId)
: this.handleWorkspaceCreateWorktree(creationRequest, workspaceId);
}
private async handleWorkspaceCreateLocal(
request: Extract<SessionInboundMessage, { type: "workspace.create.request" }>,
workspaceId?: string,
@@ -1,5 +1,5 @@
import { test, expect } from "vitest";
import { rmSync } from "node:fs";
import { mkdtempSync, rmSync } from "node:fs";
import { tmpdir } from "node:os";
import path from "node:path";
@@ -49,3 +49,30 @@ test("workspace.create surfaces each early-reject error branch", async () => {
rmSync(missingDir, { recursive: true, force: true });
}
}, 180000);
test("workspace creation replay rejects a checkout removed after completion", async () => {
const daemon = await createTestPaseoDaemon();
const directory = mkdtempSync(path.join(tmpdir(), "workspace-replay-"));
const client = new DaemonClient({
url: `ws://127.0.0.1:${daemon.port}/ws`,
appVersion: "0.1.82",
});
try {
await client.connect();
const request = {
source: { kind: "directory" as const, path: directory },
idempotencyKey: "removed-checkout",
};
const created = await client.createWorkspace(request);
expect(created.error).toBeNull();
expect(created.workspace).not.toBeNull();
rmSync(directory, { recursive: true, force: true });
const replayed = await client.createWorkspace(request);
expect(replayed.workspace).toBeNull();
expect(replayed.errorCode).toBe("directory_not_found");
} finally {
await client.close().catch(() => undefined);
await daemon.close();
rmSync(directory, { recursive: true, force: true });
}
}, 180000);