diff --git a/docs/architecture.md b/docs/architecture.md index 96d8cdcc56..160b5cf35c 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -319,6 +319,30 @@ Example: adding a new enum value // 4. Gate the new emitted value: session.supports(CLIENT_CAPS.newThing) ? "new_value" : "old_value" ``` +### Creation ownership + +Creation is owned by the daemon across socket lifetimes. A workspace request can include +its initial agent and prompt. Workspace readiness is published before provider startup; +the app keeps its existing disk-ready navigation and observes the remaining creation. +Callbacks do not advance the workflow. Resource reservations in acknowledgement are +identities, not ready workspace records. + +The creation journal (`server/creation/`) keeps cumulative milestones and permanent IDs +under an operation kind and idempotency key. Reconnect subscribes to that key; retries +join active work or reuse committed stages. A persisted resource alone cannot prove +that a provider accepted its initial prompt. Interrupted side effects with no conclusive +receipt return an unknown outcome instead of being repeated. + +`packages/client/src/creation/` owns capability selection and legacy orchestration. +Callers always pass the initial prompt to agent creation. On an older host, the client +adapts keyed creation to the legacy create/send sequence; it cannot continue that +sequence after the client process disappears. Explicitly requested IDs or receipts +that an old host cannot honor produce an unsupported error. Keep these adapters inside +the client package, as an exception to the default no-fallback feature policy. + +Creation updates require an explicit subscription and go only to the requesting socket. +Legacy consumers, including Hub, keep their existing response contract. + ## Agent lifecycle The lifecycle states are defined in `shared/agent-lifecycle.ts`: diff --git a/packages/app/e2e/browser/creation-idempotency.spec.ts b/packages/app/e2e/browser/creation-idempotency.spec.ts index 2976f00459..d3fb5f47c9 100644 --- a/packages/app/e2e/browser/creation-idempotency.spec.ts +++ b/packages/app/e2e/browser/creation-idempotency.spec.ts @@ -35,7 +35,7 @@ test("the first prompt still names an agent created with a receipt", async ({ cr 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 ({ +test("reconnecting after a lost creation response recovers the agent and initial message", async ({ creation, promptRetry, }) => { @@ -44,12 +44,11 @@ test("retrying after a lost first-prompt acknowledgement reuses the agent and me 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 ({ +test("a remounted draft reconciles its original creation after reconnect", async ({ creation, promptRetry, }) => { @@ -60,7 +59,6 @@ test("a remounted draft unlocks when its original submission fails", async ({ 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); }); @@ -90,3 +88,31 @@ test("separate drafts can intentionally create two agents in the same workspace" delayedCreation.release(); await creation.expectAgentCount(2); }); + +test("new workspace navigation and optimistic prompt precede agent completion", async ({ + creation, + delayedCreation, +}) => { + await creation.openWorkspaceForm("worktree"); + await creation.submitPrompt("Show this prompt while the agent is starting.", "Create"); + await delayedCreation.waitForDelayedCreatedStatus(); + await creation.expectWorkspaceReadyBeforeAgentCompletion(); + delayedCreation.expectSingleWorkspaceIntent(); + delayedCreation.release(); + await creation.expectOneCreatedWorkspace(); + await creation.expectAgentCount(1); +}); + +test("retrying a failed combined result keeps the workspace and original creation intent", async ({ + creation, + delayedCreation, +}) => { + await creation.openWorkspaceForm("local"); + await creation.submitPrompt("Retry this workspace and agent together.", "Create"); + await delayedCreation.waitForDelayedCreatedStatus(); + await creation.expectWorkspaceReadyBeforeAgentCompletion(); + delayedCreation.fail("Creation response was lost"); + await creation.submitPrompt("Retry this workspace and agent together."); + await creation.expectOneCreatedWorkspace(); + await creation.expectAgentCount(1); +}); diff --git a/packages/app/e2e/support/helpers/creation.ts b/packages/app/e2e/support/helpers/creation.ts index f4b94fa056..4231edf118 100644 --- a/packages/app/e2e/support/helpers/creation.ts +++ b/packages/app/e2e/support/helpers/creation.ts @@ -36,7 +36,7 @@ export function observeCreationRequests(page: Page) { const request = envelope.message; if ( request?.type === "workspace.create.request" || - request?.type === "create_agent_request" + request?.type === "agent.create.request" ) { pending.add(request.requestId); } @@ -46,6 +46,7 @@ export function observeCreationRequests(page: Page) { const response = envelope.message; if ( response?.type === "workspace.create.response" || + response?.type === "agent.create.response" || (response?.type === "status" && (response.payload.status === "agent_created" || response.payload.status === "agent_create_failed")) @@ -71,7 +72,7 @@ export async function retryNextAgentCreation(page: Page) { browser.onMessage((frame) => { const envelope = JSON.parse(frame.toString()) as { message?: SessionInboundMessage }; const request = envelope.message; - if (!repeated && request?.type === "create_agent_request") { + if (!repeated && request?.type === "agent.create.request") { repeated = true; for (let attempt = 1; attempt <= 3; attempt++) { const requestId = attempt === 1 ? request.requestId : `${request.requestId}-${attempt}`; @@ -87,13 +88,13 @@ export async function retryNextAgentCreation(page: Page) { 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" && + response?.type === "agent.create.response" && retryIds.delete(response.payload.requestId) ) { - results.push(response.payload); + results.push({ + status: response.payload.error ? "agent_create_failed" : "agent_created", + agentId: response.payload.agent?.id, + }); } browser.send(frame); }); @@ -142,6 +143,15 @@ export async function createCreationScenario(page: Page) { const rows = page.getByTestId("user-message"); await expect(prompt ? rows.filter({ hasText: prompt }) : rows.first()).toBeVisible(); }, + async expectWorkspaceReadyBeforeAgentCompletion() { + await expect(page).toHaveURL(/\/workspace\//); + await expect(page.getByTestId("user-message").first()).toBeVisible(); + await expect( + page + .locator('[data-testid^="workspace-tab-draft_"][aria-selected="true"]') + .filter({ visible: true }), + ).toBeVisible(); + }, async expectOneCreatedWorkspace() { await expect(page).toHaveURL(/\/workspace\//); await requests.settled(); @@ -208,11 +218,11 @@ export function createPromptRetryScenario( let firstMessage: ReturnType[number] | undefined; return { holdAcknowledgement() { - gate.holdNextServerMessage("send_agent_message_response"); + gate.holdNextServerMessage("agent.create.response"); }, async waitForDeliveredPrompt() { - await gate.waitForHeldServerMessage("send_agent_message_response"); - firstMessage = gate.getClientRequests("send_agent_message_request").at(-1); + await gate.waitForHeldServerMessage("agent.create.response"); + firstMessage = gate.getClientRequests("agent.create.request").at(-1); }, async disconnectAndReconnect() { const fetches = gate.getClientRequestCount("fetch_agents_request"); @@ -223,11 +233,12 @@ export function createPromptRetryScenario( .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, - }); + expect(gate.getClientRequestCount("agent.create.request")).toBe(1); + await expect + .poll(() => gate.getClientRequestCount("creation.subscribe.request")) + .toBeGreaterThan(0); + expect(gate.getClientRequests("agent.create.request").at(-1)).toEqual(firstMessage); + expect(gate.getClientRequestCount("send_agent_message_request")).toBe(0); await expect(page.getByTestId("user-message").filter({ visible: true })).toHaveCount(1); }, }; diff --git a/packages/app/e2e/support/helpers/new-workspace.ts b/packages/app/e2e/support/helpers/new-workspace.ts index 5cebc309b1..69d50c1d10 100644 --- a/packages/app/e2e/support/helpers/new-workspace.ts +++ b/packages/app/e2e/support/helpers/new-workspace.ts @@ -559,6 +559,8 @@ export interface AgentCreatedDelayControl { release(): void; waitForCreateRequest(): Promise; waitForDelayedCreatedStatus(): Promise; + expectSingleWorkspaceIntent(): void; + fail(error: string): void; } export async function delayBrowserAgentCreatedStatus( @@ -566,8 +568,10 @@ export async function delayBrowserAgentCreatedStatus( ): Promise { const daemonPortPattern = daemonWsRoutePattern(); const createRequestIds = new Set(); + const creationRequests: Record[] = []; const delayedForwards: Array<() => void> = []; let releaseRequested = false; + let responseError: string | null = null; let resolveCreateRequest: (() => void) | null = null; let resolveDelayedCreatedStatus: (() => void) | null = null; const createRequestSeen = new Promise((resolve) => { @@ -582,7 +586,14 @@ export async function delayBrowserAgentCreatedStatus( ws.onMessage((message) => { const sessionMessage = getSessionMessage(message); - if (sessionMessage?.type === "create_agent_request") { + if (sessionMessage?.type === "send_agent_message_request") + creationRequests.push(sessionMessage); + if ( + sessionMessage?.type === "create_agent_request" || + sessionMessage?.type === "agent.create.request" || + sessionMessage?.type === "workspace.create.request" + ) { + creationRequests.push(sessionMessage); const requestId = getStringField(sessionMessage, "requestId"); if (requestId) { createRequestIds.add(requestId); @@ -595,18 +606,34 @@ export async function delayBrowserAgentCreatedStatus( server.onMessage((message) => { const sessionMessage = getSessionMessage(message); const payload = - sessionMessage?.type === "status" && typeof sessionMessage.payload === "object" + sessionMessage && typeof sessionMessage.payload === "object" ? (sessionMessage.payload as Record) : null; const requestId = payload ? getStringField(payload, "requestId") : null; - if (payload?.status === "agent_created" && requestId && createRequestIds.has(requestId)) { + if ( + (payload?.status === "agent_created" || + sessionMessage?.type === "agent.create.response" || + sessionMessage?.type === "workspace.create.response") && + requestId && + createRequestIds.has(requestId) + ) { resolveDelayedCreatedStatus?.(); if (releaseRequested) { ws.send(message); return; } - delayedForwards.push(() => ws.send(message)); + delayedForwards.push(() => { + if (!responseError) { + ws.send(message); + return; + } + const envelope = JSON.parse(message.toString()); + const response = envelope.message ?? envelope; + response.payload.error = responseError; + delete response.payload.creation; + ws.send(JSON.stringify(envelope)); + }); return; } @@ -621,6 +648,19 @@ export async function delayBrowserAgentCreatedStatus( forward(); } }, + fail(error: string) { + responseError = error; + releaseRequested = true; + for (const forward of delayedForwards.splice(0)) forward(); + responseError = null; + }, + expectSingleWorkspaceIntent() { + expect(creationRequests).toHaveLength(1); + expect(creationRequests[0]).toMatchObject({ + type: "workspace.create.request", + agent: { initialPrompt: expect.any(String) }, + }); + }, waitForCreateRequest: () => createRequestSeen, waitForDelayedCreatedStatus: () => delayedCreatedStatusSeen, }; diff --git a/packages/app/src/agent-creation/index.ts b/packages/app/src/agent-creation/index.ts deleted file mode 100644 index 0a643e83c9..0000000000 --- a/packages/app/src/agent-creation/index.ts +++ /dev/null @@ -1,24 +0,0 @@ -import type { - DaemonClient, - CreateAgentRequestOptions, -} from "@getpaseo/client/internal/daemon-client"; - -/** Creation and prompt delivery have separate durable receipts on the daemon. */ -export async function createAgentWithInitialMessage( - client: Pick, - options: Omit & { - idempotencyKey: string; - clientMessageId: string; - }, -) { - const { initialPrompt = "", clientMessageId, images, attachments, ...creation } = options; - const agent = await client.createAgent(creation); - if (initialPrompt || images?.length || attachments?.length) { - await client.sendMessage(agent.id, initialPrompt, { - messageId: clientMessageId, - images, - attachments, - }); - } - return agent; -} diff --git a/packages/app/src/components/workspace-setup-dialog.tsx b/packages/app/src/components/workspace-setup-dialog.tsx index 4b96b8469a..4b2da86ea5 100644 --- a/packages/app/src/components/workspace-setup-dialog.tsx +++ b/packages/app/src/components/workspace-setup-dialog.tsx @@ -1,4 +1,5 @@ -import { createAgentWithInitialMessage } from "@/agent-creation"; +import type { CreateWorkspaceRequestOptions } from "@getpaseo/client/internal/daemon-client"; +import type { AgentSnapshotPayload } from "@getpaseo/protocol/messages"; import { getHostRuntimeStore } from "@/runtime/host-runtime"; import { useCallback, useEffect, useMemo, useState } from "react"; import { Text, View } from "react-native"; @@ -32,7 +33,6 @@ import type { DaemonClient, } from "@getpaseo/client/internal/daemon-client"; import { projectIconPlaceholderLabelFromDisplayName } from "@/utils/project-display-name"; -import { requireWorkspaceDirectory } from "@/utils/workspace-directory"; import { navigateToAgent } from "@/utils/navigate-to-agent"; import { navigateToWorkspace } from "@/stores/navigation-active-workspace-store"; import type { MessagePayload } from "@/composer/types"; @@ -93,10 +93,11 @@ async function callWorkspaceCreation({ connectedClient: DaemonClient; creationId: string; worktreeSlug: string; - input: { cwd: string }; + input: { cwd: string; agent?: CreateWorkspaceRequestOptions["agent"] }; }) { return connectedClient.createWorkspace({ idempotencyKey: creationId, + agent: input.agent, source: creationMethod === "create_worktree" ? { kind: "worktree", cwd: input.cwd, worktreeSlug } @@ -251,12 +252,17 @@ export function WorkspaceSetupDialog() { }, [client, isConnected, t]); const ensureWorkspace = useCallback( - async (input: { cwd: string; attachments: MessagePayload["attachments"] }) => { + async (input: { + cwd: string; + attachments: MessagePayload["attachments"]; + agent?: CreateWorkspaceRequestOptions["agent"]; + onAgentCreated?: (agent: AgentSnapshotPayload) => void; + }) => { if (!pendingWorkspaceSetup) { throw new Error(t("workspaceSetup.errors.pendingRequired")); } - if (createdWorkspace) { + if (createdWorkspace && !input.agent) { return createdWorkspace; } @@ -275,6 +281,7 @@ export function WorkspaceSetupDialog() { ); } + if (payload.agent) input.onAgentCreated?.(payload.agent); const normalizedWorkspace = normalizeWorkspaceDescriptor(payload.workspace); mergeWorkspaces(pendingWorkspaceSetup.serverId, [normalizedWorkspace]); if (pendingWorkspaceSetup.creationMethod === "open_project") { @@ -311,8 +318,7 @@ export function WorkspaceSetupDialog() { try { setPendingAction("chat"); setErrorMessage(null); - const ensuredWorkspace = await ensureWorkspace({ cwd, attachments }); - const connectedClient = withConnectedClient(); + if (!composerState) { throw new Error(t("workspaceSetup.errors.composerStateRequired")); } @@ -326,24 +332,30 @@ export function WorkspaceSetupDialog() { }), }); const encodedImages = await encodeImages(wirePayload.images); - const workspaceDirectory = requireWorkspaceDirectory({ - workspaceId: ensuredWorkspace.id, - workspaceDirectory: ensuredWorkspace.workspaceDirectory, - }); if (!pendingWorkspaceSetup) throw new Error(t("workspaceSetup.errors.pendingRequired")); - const agent = await createAgentWithInitialMessage(connectedClient, { - idempotencyKey: pendingWorkspaceSetup.creationId, - clientMessageId: `${pendingWorkspaceSetup.creationId}:initial-message`, - ...buildCreateAgentOptions({ - composerState, - text, - attachments: wirePayload.attachments, - encodedImages: encodedImages ?? null, - workspaceDirectory, - workspaceId: ensuredWorkspace.id, - provider: composerState.selectedProvider, - }), + let createdAgent: AgentSnapshotPayload | undefined; + const { workspaceId: _workspaceId, ...agentInput } = buildCreateAgentOptions({ + composerState, + text, + attachments: wirePayload.attachments, + encodedImages: encodedImages ?? null, + workspaceDirectory: cwd, + workspaceId: "", + provider: composerState.selectedProvider, }); + const ensuredWorkspace = await ensureWorkspace({ + cwd, + attachments, + agent: { + ...agentInput, + clientMessageId: `${pendingWorkspaceSetup.creationId}:initial-message`, + }, + onAgentCreated: (value) => { + createdAgent = value; + }, + }); + if (!createdAgent) throw new Error("The daemon did not create the requested agent"); + const agent = createdAgent; if (!getIsStillActive()) { return; @@ -376,7 +388,6 @@ export function WorkspaceSetupDialog() { ensureWorkspace, t, toast, - withConnectedClient, supportsForgeSearch, ], ); diff --git a/packages/app/src/composer/draft/workspace-tab.tsx b/packages/app/src/composer/draft/workspace-tab.tsx index 7864165b7f..c3c9820fe9 100644 --- a/packages/app/src/composer/draft/workspace-tab.tsx +++ b/packages/app/src/composer/draft/workspace-tab.tsx @@ -1,4 +1,3 @@ -import { createAgentWithInitialMessage } from "@/agent-creation"; import { useCallback, useEffect, useMemo, useRef } from "react"; import { Keyboard, ScrollView, StyleSheet as RNStyleSheet, Text, View } from "react-native"; import { useTranslation } from "react-i18next"; @@ -198,15 +197,17 @@ async function submitDraftCreateRequest(input: { const attachmentsArray = Array.isArray(attachments) ? attachments : undefined; const imagesData = await encodeImages(images); - const result = await createAgentWithInitialMessage(client, { + const options = { idempotencyKey: input.draftId, config, workspaceId, initialPrompt: text, clientMessageId: attempt.clientMessageId, ...(imagesData && imagesData.length > 0 ? { images: imagesData } : {}), - ...(attachmentsArray ? { attachments: attachmentsArray } : {}), - }); + ...(attachmentsArray && attachmentsArray.length > 0 ? { attachments: attachmentsArray } : {}), + }; + const creation = useWorkspaceDraftSubmissionStore.getState().creationByDraftId[input.draftId]; + const result = creation ? await creation.retry(options) : await client.createAgent(options); return { agentId: result.id, @@ -487,8 +488,12 @@ export function WorkspaceDraftAgentTab({ composerState, selectModelMessage: t("workspaceSetup.errors.selectModel"), }), - createRequest: async ({ attempt, text, images, attachments, cwd }) => - submitDraftCreateRequest({ + createRequest: async ({ attempt, text, images, attachments, cwd }) => { + if (pendingAutoSubmit?.agentCreation) { + const result = await pendingAutoSubmit.agentCreation.result; + return { agentId: result.id, result }; + } + return submitDraftCreateRequest({ draftId, attempt, text, @@ -502,7 +507,8 @@ export function WorkspaceDraftAgentTab({ composerState, hostDisconnectedMessage: t("workspace.terminal.hostDisconnected"), selectModelMessage: t("workspaceSetup.errors.selectModel"), - }), + }); + }, onCreateSuccess: ({ result }) => { clearDraftInput("sent"); clearWorkspaceAttachments({ scopeKey: draftAttachmentScopeKey }); diff --git a/packages/app/src/screens/new-workspace-screen.tsx b/packages/app/src/screens/new-workspace-screen.tsx index c30aebbbd4..86315e8208 100644 --- a/packages/app/src/screens/new-workspace-screen.tsx +++ b/packages/app/src/screens/new-workspace-screen.tsx @@ -1,3 +1,9 @@ +import type { + CreateAgentRequestOptions, + CreateWorkspaceRequestOptions, +} from "@getpaseo/client/internal/daemon-client"; +import type { AgentSnapshotPayload, CreationSnapshot } from "@getpaseo/protocol/messages"; +import { encodeImages } from "@/utils/encode-images"; import { getHostRuntimeStore } from "@/runtime/host-runtime"; import { useCallback, useEffect, useMemo, useReducer, useRef, useState } from "react"; import type { ReactElement, RefObject } from "react"; @@ -84,15 +90,10 @@ import { useProjectIcons } from "@/projects/icons"; import { ICON_SIZE, type Theme } from "@/styles/theme"; import type { ComposerAttachment } from "@/attachments/types"; import { useDraftWorkspaceAttachmentScopeKey } from "@/attachments/workspace-attachments-store"; -import { requestWorkspaceDraftAgent } from "@/composer/draft/create-agent-request"; -import { buildWorkspaceDraftAgentConfig } from "@/screens/workspace/workspace-draft-agent-config"; import type { MessagePayload } from "@/composer/types"; import type { UserComposerAttachment } from "@/attachments/types"; import type { AgentAttachment, ForgeSearchItem } from "@getpaseo/protocol/messages"; -import type { - CreatePaseoWorktreeInput, - DaemonClient, -} from "@getpaseo/client/internal/daemon-client"; +import type { DaemonClient } from "@getpaseo/client/internal/daemon-client"; import type { AgentProvider } from "@getpaseo/protocol/agent-types"; import type { WorkspaceDraftTabSetup, WorkspaceTabTarget } from "@/workspace-tabs/model"; import { isEmptyWorkspaceSubmission, runCreateEmptyWorkspace } from "./new-workspace-empty"; @@ -127,10 +128,7 @@ import { type ListTerminalsPayload, upsertCreatedTerminalPayload, } from "./workspace/terminals/state"; -import { - captureWorkspaceDraftCleanup, - createWorkspaceAgentInBackground, -} from "./new-workspace/background-handoff"; +import { captureWorkspaceDraftCleanup } from "./new-workspace/background-handoff"; import { useNewWorkspaceScreenPresence } from "./new-workspace/screen-presence"; const ThemedFolderPlus = withUnistyles(FolderPlus); @@ -764,6 +762,10 @@ type SubmitOutcome = "navigated" | "background"; interface SubmitDraftInput { clearConsumedDraft: () => void; + agentCreation?: { + result: Promise; + retry: (input: CreateAgentRequestOptions) => Promise; + }; serverId: string; draftKey: string; clearDraft: (lifecycle: "sent" | "abandoned") => void; @@ -795,30 +797,6 @@ interface WorkspaceDraftSubmissionConfig { target: WorkspaceTabTarget; } -async function createAndMergeWorkspace(input: { - client: NonNullable>; - createInput: Parameters< - NonNullable>["createPaseoWorktree"] - >[0]; - mergeWorkspaces: ( - serverId: string, - workspaces: ReturnType[], - ) => void; - serverId: string; - createFailedMessage: string; -}): Promise> { - const payload = await input.client.createPaseoWorktree(input.createInput); - if (payload.error || !payload.workspace) { - throw new Error(payload.error ?? input.createFailedMessage); - } - const normalizedWorkspace = normalizeWorkspaceDescriptor(payload.workspace); - const workspaceForInitialMerge = input.createInput.firstAgentContext - ? { ...normalizedWorkspace, status: "running" as const, statusEnteredAt: new Date() } - : normalizedWorkspace; - input.mergeWorkspaces(input.serverId, [workspaceForInitialMerge]); - return normalizedWorkspace; -} - async function createMultiplicityWorkspace(input: { idempotencyKey: string; worktreeSlug: string; @@ -828,6 +806,9 @@ async function createMultiplicityWorkspace(input: { sourceDirectory: string; checkoutRequest: PickerCheckoutRequest | undefined; withInitialAgent: boolean; + agent?: CreateWorkspaceRequestOptions["agent"]; + onEvent?: (snapshot: CreationSnapshot) => void; + onAgentCreated?: (agent: AgentSnapshotPayload) => void; prompt: string; attachments: AgentAttachment[]; mergeWorkspaces: ( @@ -846,6 +827,8 @@ async function createMultiplicityWorkspace(input: { }); const payload = await input.client.createWorkspace({ idempotencyKey: input.idempotencyKey, + agent: input.agent, + onEvent: input.onEvent, source: isWorktree ? { kind: "worktree", @@ -864,6 +847,7 @@ async function createMultiplicityWorkspace(input: { if (payload.error || !payload.workspace) { throw new Error(payload.error ?? input.createFailedMessage); } + if (payload.agent) input.onAgentCreated?.(payload.agent); const normalizedWorkspace = normalizeWorkspaceDescriptor(payload.workspace); const workspaceForInitialMerge = input.withInitialAgent ? { ...normalizedWorkspace, status: "running" as const, statusEnteredAt: new Date() } @@ -881,6 +865,9 @@ interface CreateChatAgentInput { prompt: string; attachments: AgentAttachment[]; withInitialAgent: boolean; + agent?: CreateWorkspaceRequestOptions["agent"]; + onEvent?: (snapshot: CreationSnapshot) => void; + onAgentCreated?: (agent: AgentSnapshotPayload) => void; }) => Promise>; serverId: string; draftKey: string; @@ -945,7 +932,18 @@ function buildComposerInitialValues(input: { return undefined; } -async function runCreateChatAgent(input: CreateChatAgentInput): Promise { +const pendingWorkspaceSubmissions = new Map>(); +function runCreateChatAgent(input: CreateChatAgentInput): Promise { + const key = JSON.stringify([input.serverId, input.draftId]); + const pending = pendingWorkspaceSubmissions.get(key); + if (pending) return pending; + const submission = Promise.resolve().then(() => createWorkspaceChatAgent(input)); + pendingWorkspaceSubmissions.set(key, submission); + void submission.finally(() => pendingWorkspaceSubmissions.delete(key)).catch(() => undefined); + return submission; +} + +async function createWorkspaceChatAgent(input: CreateChatAgentInput): Promise { const { payload, composerState, ensureWorkspace, serverId, clearDraft } = input; const clearConsumedDraft = captureWorkspaceDraftCleanup(input); const { text, attachments, cwd } = payload; @@ -963,36 +961,103 @@ async function runCreateChatAgent(input: CreateChatAgentInput): Promise void; + let rejectAgent!: (error: unknown) => void; + const result = new Promise((resolve, reject) => { + resolveAgent = resolve; + rejectAgent = reject; }); + // The destination may mount after creation finishes or fails. + void result.catch(() => undefined); + let navigated = false; + let outcome: SubmitOutcome = "background"; + const initialAgent: NonNullable = { + config: { + provider, + cwd, + modeId: composerState.selectedMode || undefined, + model: composerState.effectiveModelId || undefined, + thinkingOptionId: composerState.effectiveThinkingOptionId || undefined, + featureValues: composerState.featureValues, + }, + initialPrompt: text, + clientMessageId: `${input.draftId}:initial-message`, + images: images?.length ? images : undefined, + attachments: wirePayload.attachments?.length ? wirePayload.attachments : undefined, + }; + const execute = async (requestedAgent = initialAgent): Promise => { + let createdAgent: AgentSnapshotPayload | undefined; + await ensureWorkspace({ + cwd, + prompt: text, + attachments: workspaceNamingAttachments, + withInitialAgent: true, + agent: requestedAgent, + onAgentCreated: (agent) => { + createdAgent = agent; + }, + onEvent: (snapshot) => { + if (!snapshot.workspace || navigated) return; + navigated = true; + if (!input.isStillOnCreateScreen()) return; + const workspace = normalizeWorkspaceDescriptor(snapshot.workspace); + getHostRuntimeStore().acceptWorkspaceSnapshots(serverId, [ + { ...workspace, status: "running" }, + ]); + const initialSetup = buildWorkspaceDraftSetupForCreatedWorkspace({ + forkDraftSetup: input.forkDraftSetup, + workspaceDirectory: workspace.workspaceDirectory, + provider, + composerState, + }); + outcome = submitWorkspaceDraft({ + clearConsumedDraft, + draftKey: input.draftKey, + draftContextScopeKey: input.draftContextScopeKey, + resolveClient: input.resolveClient, + isStillOnCreateScreen: input.isStillOnCreateScreen, + serverId, + clearDraft, + draftId: input.draftId, + initialSetup, + workspaceId: workspace.id, + workspaceDirectory: workspace.workspaceDirectory, + text, + attachments, + provider, + composerState, + supportsForgeSearch: input.supportsForgeSearch, + agentCreation, + }); + }, + }); + if (!createdAgent) throw new Error("Workspace creation returned no agent"); + return createdAgent; + }; + const agentCreation = { + result, + retry: (request: CreateAgentRequestOptions) => + execute({ + ...initialAgent, + config: { ...request.config!, cwd }, + initialPrompt: request.initialPrompt ?? "", + clientMessageId: initialAgent.clientMessageId, + images: request.images, + attachments: request.attachments, + }), + }; + try { + resolveAgent(await execute()); + if (outcome === "background") clearConsumedDraft(); + return outcome; + } catch (error) { + rejectAgent(error); + throw error; + } } function buildComposerConfig(input: { @@ -1054,7 +1119,7 @@ function resolveWorkspaceDraftSubmissionConfig(input: { }; } -async function submitWorkspaceDraft(input: SubmitDraftInput): Promise { +function submitWorkspaceDraft(input: SubmitDraftInput): SubmitOutcome { const { serverId, clearDraft, @@ -1082,33 +1147,6 @@ async function submitWorkspaceDraft(input: SubmitDraftInput): Promise - requestWorkspaceDraftAgent(input.resolveClient(), { - workspaceId, - config: buildWorkspaceDraftAgentConfig({ - provider: submission.provider, - cwd: submission.cwd, - ...(submission.modeId ? { modeId: submission.modeId } : {}), - ...(submission.model ? { model: submission.model } : {}), - ...(submission.thinkingOptionId - ? { thinkingOptionId: submission.thinkingOptionId } - : {}), - ...(submission.featureValues ? { featureValues: submission.featureValues } : {}), - }), - text: text.trim(), - clientMessageId, - ...(wirePayload.images.length > 0 ? { images: wirePayload.images } : {}), - ...(wirePayload.attachments.length > 0 ? { attachments: wirePayload.attachments } : {}), - }), - }); - return "background"; - } - const started = useCreateFlowStore.getState().trySetPending({ serverId, draftId, @@ -1136,6 +1174,7 @@ async function submitWorkspaceDraft(input: SubmitDraftInput): Promise { - if (!selectedProject) { - throw new Error("Choose a project"); - } - if (!selectedSourceDirectory) { - throw new Error("Choose a host for this project"); - } - const firstAgentContext = buildFirstAgentContext(input); - const hostProjectId = getHostProjectId(selectedProject, selectedServerId); - if (!hostProjectId) { - throw new Error("Project is not available on the selected host"); - } - - return { - cwd: selectedSourceDirectory, - projectId: hostProjectId, - worktreeSlug: createNameId(), - ...(firstAgentContext ? { firstAgentContext } : {}), - ...input.checkoutRequest, - }; - }, - [selectedProject, selectedServerId, selectedSourceDirectory], - ); - const ensureWorkspace = useCallback( async (input: { cwd: string; prompt: string; attachments: AgentAttachment[]; withInitialAgent: boolean; + agent?: CreateWorkspaceRequestOptions["agent"]; + onEvent?: (snapshot: CreationSnapshot) => void; + onAgentCreated?: (agent: AgentSnapshotPayload) => void; }) => { if (createdWorkspace) { return createdWorkspace; @@ -2063,34 +2075,28 @@ export function NewWorkspaceScreen({ selectedItem ?? defaultBasePickerItem(checkoutStatusForCreate), ) : undefined; - const normalizedWorkspace = supportsWorkspaceMultiplicity - ? await createMultiplicityWorkspace({ - idempotencyKey: creationIdentity.draftId, - worktreeSlug: creationIdentity.worktreeSlug, - client: connectedClient, - isolation: effectiveIsolation, - project: selectedProject, - sourceDirectory: selectedSourceDirectory, - checkoutRequest, - withInitialAgent: input.withInitialAgent, - prompt: input.prompt, - attachments: input.attachments, - mergeWorkspaces, - serverId: selectedServerId, - createFailedMessage: t("newWorkspace.errors.createWorktreeFailed"), - }) - : await createAndMergeWorkspace({ - client: connectedClient, - createInput: buildCreateWorktreeInput({ ...input, checkoutRequest }), - mergeWorkspaces, - serverId: selectedServerId, - createFailedMessage: t("newWorkspace.errors.createWorktreeFailed"), - }); + const normalizedWorkspace = await createMultiplicityWorkspace({ + idempotencyKey: creationIdentity.draftId, + worktreeSlug: creationIdentity.worktreeSlug, + client: connectedClient, + isolation: createsWorktree ? "worktree" : "local", + project: selectedProject, + sourceDirectory: selectedSourceDirectory, + checkoutRequest, + withInitialAgent: input.withInitialAgent, + prompt: input.prompt, + attachments: input.attachments, + agent: input.agent, + onEvent: input.onEvent, + onAgentCreated: input.onAgentCreated, + mergeWorkspaces, + serverId: selectedServerId, + createFailedMessage: t("newWorkspace.errors.createWorktreeFailed"), + }); setCreatedWorkspace(normalizedWorkspace); return normalizedWorkspace; }, [ - buildCreateWorktreeInput, creationIdentity, createdWorkspace, effectiveIsolation, diff --git a/packages/app/src/stores/workspace-draft-submission-store.test.ts b/packages/app/src/stores/workspace-draft-submission-store.test.ts index 87fa3fad74..73e1a40899 100644 --- a/packages/app/src/stores/workspace-draft-submission-store.test.ts +++ b/packages/app/src/stores/workspace-draft-submission-store.test.ts @@ -9,6 +9,7 @@ function loadProductionStore() { const filename = fileURLToPath(new URL("./workspace-draft-submission-store.ts", import.meta.url)); const caller = { name: "metro", platform: "web", isDev: false, supportsStaticESM: false }; const result = transformFileSync(filename, { + cwd: fileURLToPath(new URL("../../", import.meta.url)), envName: "production", caller, }); @@ -44,3 +45,27 @@ test("the production app consumes a draft once and leaves other drafts pending", expect(store.getState().consumePending(submission)).toBeNull(); expect(Object.keys(store.getState().pendingByDraftId)).toEqual(["draft-2"]); }); + +test("consuming draft presentation retains its running creation through remount and retry", () => { + const store = loadProductionStore(); + const result = new Promise(() => {}); + const creation = { result, retry: () => result }; + const submission = { + serverId: "server", + workspaceId: "workspace", + draftId: "draft-one", + text: "hello", + attachments: [], + cwd: "/repo", + provider: "codex", + clientMessageId: "message", + timestamp: 1, + agentCreation: creation, + }; + store.getState().setPending(submission); + store.getState().consumePending(submission); + expect(store.getState().pendingByDraftId[submission.draftId]).toBeUndefined(); + expect(store.getState().creationByDraftId[submission.draftId]).toBe(creation); + store.getState().clearDraftSetup({ draftId: submission.draftId }); + expect(store.getState().creationByDraftId[submission.draftId]).toBeUndefined(); +}); diff --git a/packages/app/src/stores/workspace-draft-submission-store.ts b/packages/app/src/stores/workspace-draft-submission-store.ts index a37c3a1f80..639701d740 100644 --- a/packages/app/src/stores/workspace-draft-submission-store.ts +++ b/packages/app/src/stores/workspace-draft-submission-store.ts @@ -1,9 +1,16 @@ +import type { CreateAgentRequestOptions } from "@getpaseo/client/internal/daemon-client"; +import type { AgentSnapshotPayload } from "@getpaseo/protocol/messages"; import { create } from "zustand"; import type { ComposerAttachment } from "@/attachments/types"; import type { AgentProvider } from "@getpaseo/protocol/agent-types"; import type { WorkspaceDraftTabSetup } from "@/workspace-tabs/model"; export interface PendingWorkspaceDraftSubmission { + /** Already-running creation. Mounting the draft only observes its result. */ + agentCreation?: { + result: Promise; + retry: (input: CreateAgentRequestOptions) => Promise; + }; serverId: string; workspaceId: string; draftId: string; @@ -28,6 +35,7 @@ export interface PendingWorkspaceDraftSetup { interface WorkspaceDraftSubmissionState { pendingByDraftId: Record; setupByDraftId: Record; + creationByDraftId: Record>; setPending: (submission: PendingWorkspaceDraftSubmission) => void; setDraftSetup: (input: { draftId: string; @@ -61,8 +69,12 @@ export const useWorkspaceDraftSubmissionStore = create ({ pendingByDraftId: {}, setupByDraftId: {}, + creationByDraftId: {}, setPending: (submission) => set((state) => ({ + creationByDraftId: submission.agentCreation + ? { ...state.creationByDraftId, [submission.draftId]: submission.agentCreation } + : state.creationByDraftId, pendingByDraftId: { ...state.pendingByDraftId, [submission.draftId]: submission, @@ -82,9 +94,10 @@ export const useWorkspaceDraftSubmissionStore = create { - if (!state.setupByDraftId[normalizedDraftId]) return state; + const creationByDraftId = { ...state.creationByDraftId }; + delete creationByDraftId[normalizedDraftId]; const { [normalizedDraftId]: _removed, ...setupByDraftId } = state.setupByDraftId; - return { setupByDraftId }; + return { setupByDraftId, creationByDraftId }; }); }, consumePending: (input) => { diff --git a/packages/client/src/creation/index.test.ts b/packages/client/src/creation/index.test.ts new file mode 100644 index 0000000000..38d1548474 --- /dev/null +++ b/packages/client/src/creation/index.test.ts @@ -0,0 +1,201 @@ +import { expect, test } from "vitest"; +import type { + AgentSnapshotPayload, + CreationSnapshot, + WorkspaceDescriptorPayload, +} from "@getpaseo/protocol/messages"; +import { CreationClient, type CreationResult } from "./index.js"; + +const workspace: WorkspaceDescriptorPayload = { + id: "wks_0123456789abcdef", + projectId: "project", + projectDisplayName: "Project", + projectRootPath: "/project", + workspaceDirectory: "/project/worktree", + projectKind: "git", + workspaceKind: "worktree", + name: "Worktree", + status: "running", + statusEnteredAt: null, + activityAt: null, + archivingAt: null, + scripts: [], +}; +const agent: AgentSnapshotPayload = { + id: "00000000-0000-4000-8000-000000000001", + workspaceId: workspace.id, + provider: "codex", + cwd: "/project/worktree", + model: null, + createdAt: "2026-09-11T00:00:00Z", + updatedAt: "2026-09-11T00:00:00Z", + lastUserMessageAt: null, + status: "idle", + capabilities: { + supportsStreaming: true, + supportsSessionPersistence: true, + supportsDynamicModes: false, + supportsMcpServers: false, + supportsReasoningStream: false, + supportsToolInvocations: true, + supportsRewindConversation: false, + supportsRewindFiles: false, + supportsRewindBoth: false, + }, + currentModeId: null, + availableModes: [], + pendingPermissions: [], + persistence: null, + title: "Start once", + labels: {}, +}; + +function fixture(modern: boolean) { + const requests: Array<{ kind: string; input: Record }> = []; + const legacy: Array<{ kind: string; input: unknown }> = []; + let resolve!: (result: CreationResult) => void; + const result = new Promise((done) => { + resolve = done; + }); + const client = new CreationClient({ + supports: () => modern, + connected: () => true, + requestId: () => "generated-key", + request: async (kind, input) => { + requests.push({ kind, input }); + return result; + }, + subscribe: async () => null, + legacyWorkspace: async (input) => { + legacy.push({ kind: "workspace", input }); + return { requestId: "legacy-workspace", workspace, error: null, setupTerminalId: null }; + }, + legacyAgent: async (input) => { + legacy.push({ kind: "agent", input }); + return agent; + }, + sendMessage: async (id, text, options) => { + legacy.push({ kind: "message", input: { id, text, ...options } }); + }, + }); + const input = { + idempotencyKey: "intent-one", + source: { kind: "directory" as const, path: "/project" }, + agent: { + config: { provider: "codex", cwd: "/project" }, + initialPrompt: "Start once", + clientMessageId: "message-one", + }, + }; + return { client, requests, legacy, resolve, input }; +} + +test("duplicate client submissions join one complete intent and cumulative updates never regress", async () => { + const f = fixture(true); + const phases: string[] = []; + const first = f.client.createWorkspace({ + ...f.input, + onEvent: (snapshot) => phases.push(snapshot.phase), + }); + const duplicate = f.client.createWorkspace(f.input); + expect(duplicate).toBe(first); + expect(f.requests).toEqual([{ kind: "workspace", input: { ...f.input, subscribe: true } }]); + await expect( + f.client.createWorkspace({ + ...f.input, + agent: { ...f.input.agent, initialPrompt: "Different intent" }, + }), + ).rejects.toThrow("workspace_request_key_conflict"); + const snapshot: CreationSnapshot = { + kind: "workspace", + idempotencyKey: "intent-one", + revision: 2, + phase: "workspace_ready", + workspaceId: workspace.id, + agentId: agent.id, + workspace, + error: null, + }; + f.client.receive(snapshot); + f.client.receive({ ...snapshot, revision: 1, phase: "accepted" }); + expect(phases).toEqual(["workspace_ready"]); + const late: string[] = []; + f.client.createWorkspace({ ...f.input, onEvent: (update) => late.push(update.phase) }); + expect(late).toEqual(["workspace_ready"]); + f.resolve({ + workspace, + agent, + error: null, + setupSkippedReason: "Untrusted source", + requestId: "response-one", + }); + expect(await first).toMatchObject({ + workspace, + agent, + setupSkippedReason: "Untrusted source", + requestId: "response-one", + }); + expect(await duplicate).toEqual(await first); + expect(f.legacy).toEqual([]); + f.client.close(); +}); + +test("legacy adaptation keeps workspace, agent and initial prompt sequencing inside the client", async () => { + const f = fixture(false); + const phases: string[] = []; + const result = await f.client.createWorkspace({ + ...f.input, + onEvent: (snapshot) => phases.push(snapshot.phase), + }); + expect(result).toMatchObject({ workspace, agent, error: null }); + expect(phases).toEqual(["workspace_ready"]); + expect(f.legacy).toEqual([ + { kind: "workspace", input: { idempotencyKey: "intent-one", source: f.input.source } }, + { + kind: "agent", + input: { + config: { provider: "codex", cwd: workspace.workspaceDirectory }, + workspaceId: workspace.id, + idempotencyKey: "intent-one:agent", + clientMessageId: "message-one", + }, + }, + { + kind: "message", + input: { + id: agent.id, + text: "Start once", + messageId: "message-one", + images: undefined, + attachments: undefined, + }, + }, + ]); + expect(f.requests).toEqual([]); + f.client.close(); +}); + +test("legacy daemons cannot silently replace caller-selected IDs", async () => { + const f = fixture(false); + await expect(f.client.createWorkspace({ ...f.input, workspaceId: workspace.id })).rejects.toThrow( + "caller-selected", + ); + await expect( + f.client.createAgent({ agentId: agent.id, config: f.input.agent.config }), + ).rejects.toThrow("caller-selected"); + expect(f.legacy).toEqual([]); + f.client.close(); +}); + +test("keyed creation with an empty prompt works on legacy daemons without sending an empty message", async () => { + const f = fixture(false); + await f.client.createAgent({ + idempotencyKey: "empty-agent", + config: f.input.agent.config, + initialPrompt: "", + }); + expect(f.legacy).toEqual([ + { kind: "agent", input: { idempotencyKey: "empty-agent", config: f.input.agent.config } }, + ]); + f.client.close(); +}); diff --git a/packages/client/src/creation/index.ts b/packages/client/src/creation/index.ts new file mode 100644 index 0000000000..07d8247359 --- /dev/null +++ b/packages/client/src/creation/index.ts @@ -0,0 +1,270 @@ +import type { + AgentSnapshotPayload, + CreationSnapshot, + WorkspaceCreateResponse, +} from "@getpaseo/protocol/messages"; +import type { CreateAgentRequestOptions, CreateWorkspaceRequestOptions } from "../daemon-client.js"; + +type Kind = CreationSnapshot["kind"]; +type Observer = (snapshot: CreationSnapshot) => void; +export interface CreationResult { + agent?: AgentSnapshotPayload | null; + workspace?: WorkspaceCreateResponse["payload"]["workspace"]; + creation?: CreationSnapshot; + error: string | null; + errorCode?: string; + setupTerminalId?: string | null; + setupSkippedReason?: string; + requestId?: string; +} +interface Dependencies { + supports: () => boolean; + connected: () => boolean; + requestId: () => string; + request: (kind: Kind, input: Record) => Promise; + subscribe: (kind: Kind, key: string, enabled?: boolean) => Promise; + legacyAgent: (input: CreateAgentRequestOptions) => Promise; + legacyWorkspace: ( + input: CreateWorkspaceRequestOptions, + ) => Promise; + sendMessage: ( + id: string, + text: string, + options: { + messageId: string; + images?: CreateAgentRequestOptions["images"]; + attachments?: CreateAgentRequestOptions["attachments"]; + }, + ) => Promise; +} +interface Operation { + kind: Kind; + key: string; + fingerprint: string; + input: Record; + observers: Set; + snapshot?: CreationSnapshot; + result: Promise; + resolve: (result: CreationResult) => void; + reject: (error: Error) => void; + settled: boolean; + modern: boolean; + recovering: boolean; +} + +/** Client-owned observation and compatibility. UI callbacks never advance creation. */ +export class CreationClient { + private readonly operations = new Map(); + constructor(private readonly deps: Dependencies) {} + + createAgent(input: CreateAgentRequestOptions): Promise { + const { onEvent, ...request } = input; + return this.start("agent", request, onEvent, () => this.legacyAgent(request)); + } + createWorkspace(input: CreateWorkspaceRequestOptions): Promise { + const { onEvent, ...request } = input; + return this.start("workspace", request, onEvent, async (operation) => { + if (request.workspaceId || request.agent?.agentId) + throw new Error("Update the host to use caller-selected creation IDs."); + const { agent, ...workspaceInput } = request; + const workspace = await this.deps.legacyWorkspace(workspaceInput); + if (workspace.error || !workspace.workspace) return workspace; + this.receive({ + kind: "workspace", + idempotencyKey: operation.key, + revision: 1, + phase: "workspace_ready", + workspaceId: workspace.workspace.id, + agentId: null, + workspace: workspace.workspace, + setupSkippedReason: workspace.setupSkippedReason, + error: null, + }); + if (!agent) return workspace; + const sourceCwd = + request.source.kind === "directory" ? request.source.path : request.source.cwd; + const cwd = sourceCwd + ? remapDirectory(agent.config!.cwd, sourceCwd, workspace.workspace.workspaceDirectory!) + : workspace.workspace.workspaceDirectory!; + const created = await this.legacyAgent({ + ...agent, + config: { ...agent.config!, cwd }, + workspaceId: workspace.workspace.id, + idempotencyKey: `${operation.key}:agent`, + }); + return { ...workspace, agent: created.agent }; + }); + } + + private async legacyAgent(input: CreateAgentRequestOptions): Promise { + if (input.agentId) throw new Error("Update the host to use caller-selected creation IDs."); + // COMPAT(creationLifecycle): added in v0.8.0, remove after 2027-03-11 once the daemon floor supports creationLifecycle. + // Old keyed creates reject initialPrompt. Keep this adaptation private to the SDK. + if (input.idempotencyKey) { + const { initialPrompt = "", images, attachments, ...creation } = input; + const hasPrompt = Boolean(initialPrompt.trim() || images?.length || attachments?.length); + if (hasPrompt && input.outputSchema) + throw new Error("Update the host to create a keyed agent with structured initial output."); + const agent = await this.deps.legacyAgent(creation); + if (hasPrompt) + await this.deps.sendMessage(agent.id, initialPrompt, { + messageId: input.clientMessageId ?? `${input.idempotencyKey}:initial-message`, + images, + attachments, + }); + return { agent, error: null }; + } + return { agent: await this.deps.legacyAgent(input), error: null }; + } + + private start( + kind: Kind, + input: Record, + observer: Observer | undefined, + legacy: (operation: Operation) => Promise, + ): Promise { + const key = + typeof input.idempotencyKey === "string" ? input.idempotencyKey : this.deps.requestId(); + const identity = JSON.stringify([kind, key]); + const { requestId: _requestId, ...intent } = input; + const fingerprint = JSON.stringify(intent, (_key, value: unknown) => + value && typeof value === "object" && !Array.isArray(value) + ? Object.fromEntries(Object.entries(value).sort(([a], [b]) => a.localeCompare(b))) + : value, + ); + const previous = this.operations.get(identity); + if (previous) { + if (previous.fingerprint !== fingerprint) + return Promise.reject(new Error(`${kind}_request_key_conflict`)); + if (observer) { + previous.observers.add(observer); + if (previous.snapshot) this.notify(observer, previous.snapshot); + } + return previous.result; + } + let resolve!: Operation["resolve"]; + let reject!: Operation["reject"]; + const result = new Promise((done, fail) => { + resolve = done; + reject = fail; + }); + const operation: Operation = { + kind, + key, + input: { ...input, idempotencyKey: key }, + fingerprint, + observers: new Set(observer ? [observer] : []), + result, + resolve, + reject, + settled: false, + modern: this.deps.supports(), + recovering: false, + }; + this.operations.set(identity, operation); + if (operation.modern) void this.submit(operation); + else + void legacy(operation).then( + (value) => this.finish(operation, value), + (error) => this.fail(operation, error), + ); + return result; + } + + private async submit(operation: Operation): Promise { + try { + const result = await this.deps.request(operation.kind, { + ...operation.input, + subscribe: true, + }); + if (result.creation) this.receive(result.creation); + this.finish(operation, result); + } catch (error) { + if (operation.settled || !this.deps.connected()) return; + this.fail(operation, error); + } + } + receive(snapshot: CreationSnapshot): void { + const operation = this.operations.get(JSON.stringify([snapshot.kind, snapshot.idempotencyKey])); + if (!operation || (operation.snapshot && operation.snapshot.revision >= snapshot.revision)) + return; + operation.snapshot = snapshot; + for (const observer of operation.observers) this.notify(observer, snapshot); + if (operation.recovering && (snapshot.phase === "completed" || snapshot.phase === "failed")) { + this.finish(operation, { + agent: snapshot.agent, + workspace: snapshot.workspace, + creation: snapshot, + setupSkippedReason: snapshot.setupSkippedReason, + error: snapshot.error, + errorCode: snapshot.errorCode, + }); + } + } + private async recover(operation: Operation): Promise { + operation.recovering = true; + const snapshot = await this.deps.subscribe(operation.kind, operation.key); + if (!snapshot) { + await this.submit(operation); + return; + } + this.receive(snapshot); + if (snapshot.phase === "completed" || snapshot.phase === "failed") { + this.finish(operation, { + agent: snapshot.agent, + workspace: snapshot.workspace, + creation: snapshot, + setupSkippedReason: snapshot.setupSkippedReason, + error: snapshot.error, + errorCode: snapshot.errorCode, + }); + } + } + + reconnect(): void { + for (const operation of this.operations.values()) { + if (operation.settled || !operation.modern) continue; + void this.recover(operation).catch((error) => { + if (this.deps.connected()) this.fail(operation, error); + }); + } + } + close(): void { + for (const operation of this.operations.values()) + if (!operation.settled) this.fail(operation, new Error("Daemon client closed")); + this.operations.clear(); + } + private finish(operation: Operation, result: CreationResult): void { + if (operation.settled) return; + operation.settled = true; + if (operation.recovering) + void this.deps.subscribe(operation.kind, operation.key, false).catch(() => undefined); + operation.resolve(result); + operation.observers.clear(); + this.operations.delete(JSON.stringify([operation.kind, operation.key])); + } + private fail(operation: Operation, error: unknown): void { + if (operation.settled) return; + operation.settled = true; + operation.reject(error instanceof Error ? error : new Error(String(error))); + operation.observers.clear(); + this.operations.delete(JSON.stringify([operation.kind, operation.key])); + } + private notify(observer: Observer, snapshot: CreationSnapshot): void { + try { + observer(snapshot); + } catch { + /* Observation does not control execution. */ + } + } +} + +function remapDirectory(cwd: string, source: string, target: string): string { + const normalize = (value: string) => value.replace(/\\/g, "/").replace(/\/$/, ""); + const normalized = normalize(cwd), + root = normalize(source); + if (normalized === root) return target; + if (!normalized.startsWith(`${root}/`)) + throw new Error("Agent directory must be inside the workspace source"); + return `${normalize(target)}${normalized.slice(root.length)}`; +} diff --git a/packages/client/src/daemon-client.test.ts b/packages/client/src/daemon-client.test.ts index 7ecf6130b8..5ed4264957 100644 --- a/packages/client/src/daemon-client.test.ts +++ b/packages/client/src/daemon-client.test.ts @@ -6322,3 +6322,86 @@ test("wire snapshot callers own expansion and receive hash references unchanged" ); expect(await request).toEqual(body); }); + +test("creation lifecycle sends the keyed agent and initial prompt as one intent", async () => { + const transport = createMockTransport(); + const client = new DaemonClient({ + url: "ws://test", + clientId: "creation-contract", + transportFactory: () => transport.transport, + reconnect: { enabled: false }, + }); + clients.push(client); + const connected = client.connect(); + transport.triggerOpen({ features: { creationLifecycle: true, agentRequestReceipts: true } }); + await connected; + const creation = client.createAgent({ + idempotencyKey: "draft-one", + provider: "codex", + cwd: "/project", + workspaceId: "wks_0123456789abcdef", + initialPrompt: "Start once", + clientMessageId: "first-message", + }); + void creation.catch(() => undefined); + const request = parseSentFrame(transport.sent[0]); + expect(request).toMatchObject({ + type: "agent.create.request", + idempotencyKey: "draft-one", + initialPrompt: "Start once", + clientMessageId: "first-message", + }); + transport.triggerMessage( + wrapSessionMessage({ + type: "agent.create.response", + payload: { requestId: request.requestId, agent: null, error: "provider unavailable" }, + }), + ); + await expect(creation).rejects.toThrow("provider unavailable"); + expect(transport.sent).toHaveLength(1); +}); + +test("creation lifecycle acknowledgement reaches the observer before the final response", async () => { + const transport = createMockTransport(); + const client = new DaemonClient({ + url: "ws://test", + clientId: "creation-progress", + transportFactory: () => transport.transport, + reconnect: { enabled: false }, + }); + clients.push(client); + const connected = client.connect(); + transport.triggerOpen({ features: { creationLifecycle: true } }); + await connected; + const phases: string[] = []; + const creation = client.createAgent({ + provider: "codex", + cwd: "/project", + idempotencyKey: "observe-one", + onEvent: (snapshot) => phases.push(snapshot.phase), + }); + void creation.catch(() => undefined); + const request = parseSentFrame(transport.sent[0]); + transport.triggerMessage( + wrapSessionMessage({ + type: "agent.create.update", + payload: { + kind: "agent", + idempotencyKey: "observe-one", + revision: 0, + phase: "accepted", + workspaceId: null, + agentId: "00000000-0000-4000-8000-000000000001", + error: null, + }, + }), + ); + expect(phases).toEqual(["accepted"]); + transport.triggerMessage( + wrapSessionMessage({ + type: "agent.create.response", + payload: { requestId: request.requestId, agent: null, error: "provider unavailable" }, + }), + ); + await expect(creation).rejects.toThrow("provider unavailable"); +}); diff --git a/packages/client/src/daemon-client.ts b/packages/client/src/daemon-client.ts index eb040bee25..78a386dba1 100644 --- a/packages/client/src/daemon-client.ts +++ b/packages/client/src/daemon-client.ts @@ -6,6 +6,8 @@ import { DEFAULT_CLIENT_CAPABILITIES, type TimelineSubscription, } from "./connection/index.js"; +import { CreationClient } from "./creation/index.js"; +import type { CreationSnapshot } from "@getpaseo/protocol/messages"; import type { z } from "zod"; import type { SessionEventSubscription } from "@getpaseo/protocol/messages"; import type { ClientCapability } from "@getpaseo/protocol/client-capabilities"; @@ -370,6 +372,8 @@ export interface AgentAttentionRequiredNotification { type AgentConfigOverrides = Partial>; export interface CreateAgentRequestOptions extends AgentConfigOverrides { + agentId?: string; + onEvent?: (snapshot: CreationSnapshot) => void; config?: AgentSessionConfig; provider?: AgentProvider; cwd?: string; @@ -392,6 +396,20 @@ export interface CreateAgentRequestOptions extends AgentConfigOverrides { labels?: Record; } +export interface CreateWorkspaceRequestOptions { + source: WorkspaceCreateRequest["source"]; + title?: string; + idempotencyKey?: string; + workspaceId?: string; + agent?: Omit< + CreateAgentRequestOptions, + "workspaceId" | "onEvent" | "worktree" | "git" | "worktreeName" | "idempotencyKey" | "requestId" + >; + onEvent?: (snapshot: CreationSnapshot) => void; + firstAgentContext?: WorkspaceCreateRequest["firstAgentContext"]; + requestId?: string; +} + export interface CreatePaseoWorktreeInput extends Pick< CreatePaseoWorktreeRequest, | "cwd" @@ -1416,6 +1434,7 @@ export class DaemonClient { this.providerSnapshotUpdates.clear(); this.clearWaiters(new Error("Daemon client closed")); await this.owned.close(); + this.creations.close(); this.rejectPendingSendQueue(new Error("Daemon client closed")); this.rejectPingProbe(new Error("Daemon client closed")); this.terminalStreams.clearSlots(); @@ -2656,7 +2675,49 @@ export class DaemonClient { // Agent Lifecycle // ============================================================================ + private readonly creations = new CreationClient({ + supports: () => this.lastServerInfoMessage?.features?.creationLifecycle === true, + connected: () => this.connectionState.status === "connected", + requestId: () => this.createRequestId(), + request: (kind, input) => + kind === "workspace" + ? this.sendCorrelatedSessionRequest({ + requestId: input.requestId as string | undefined, + message: { ...input, type: "workspace.create.request" }, + responseType: "workspace.create.response", + timeout: 0, + }) + : this.sendCorrelatedSessionRequest({ + requestId: input.requestId as string | undefined, + message: { ...input, type: "agent.create.request" }, + responseType: "agent.create.response", + timeout: 0, + }), + subscribe: async (kind, idempotencyKey, subscribe = true) => { + const result = await this.sendCorrelatedSessionRequest({ + message: { type: "creation.subscribe.request", kind, idempotencyKey, subscribe }, + responseType: "creation.subscribe.response", + }); + if (result.error) throw new Error(result.error); + return result.snapshot; + }, + legacyAgent: (input) => this.createLegacyAgent(input), + legacyWorkspace: (input) => this.createLegacyWorkspace(input, input.requestId), + sendMessage: (id, text, options) => this.sendMessage(id, text, options), + }); + async createAgent(options: CreateAgentRequestOptions): Promise { + const result = await this.creations.createAgent({ + ...options, + config: resolveAgentConfig(options), + }); + if (result.error || !result.agent) throw new Error(result.error ?? "Agent creation failed"); + return result.agent; + } + + private async createLegacyAgent( + options: CreateAgentRequestOptions, + ): Promise { if (options.idempotencyKey !== undefined) this.requireAgentRequestReceipts(); const requestId = this.createRequestId(options.requestId); const config = resolveAgentConfig(options); @@ -4330,12 +4391,28 @@ export class DaemonClient { } async createWorkspace( - input: { - source: WorkspaceCreateRequest["source"]; - title?: string; - idempotencyKey?: string; - firstAgentContext?: WorkspaceCreateRequest["firstAgentContext"]; - }, + input: CreateWorkspaceRequestOptions, + requestId?: string, + ): Promise { + const resolvedRequestId = this.createRequestId(requestId ?? input.requestId); + const result = await this.creations.createWorkspace({ + ...input, + requestId: resolvedRequestId, + ...(input.agent + ? { agent: { ...input.agent, config: resolveAgentConfig(input.agent) } } + : {}), + }); + return { + ...result, + workspace: result.workspace ?? null, + agent: result.agent ?? undefined, + setupTerminalId: result.setupTerminalId ?? null, + requestId: result.requestId ?? resolvedRequestId, + }; + } + + private async createLegacyWorkspace( + input: CreateWorkspaceRequestOptions, requestId?: string, ): Promise { // COMPAT(workspaceRequestReceipts): added in v0.8.0; remove gate after 2027-03-07. @@ -6281,6 +6358,7 @@ export class DaemonClient { ...DEFAULT_CLIENT_CAPABILITIES, ...this.config.capabilities, }); + this.creations.reconnect(); this.flushPendingSendQueue(); this.resolveConnect(); } @@ -6315,6 +6393,11 @@ export class DaemonClient { } } + if ( + consumerMessage.type === "workspace.create.update" || + consumerMessage.type === "agent.create.update" + ) + this.creations.receive(consumerMessage.payload); this.resolveWaiters(consumerMessage); this.owned.receive(consumerMessage); } @@ -6498,6 +6581,15 @@ function resolveAgentConfig(options: CreateAgentRequestOptions): AgentSessionCon config, provider, cwd, + agentId: _agentId, + onEvent: _onEvent, + idempotencyKey: _idempotencyKey, + clientMessageId: _clientMessageId, + callerAgentId: _callerAgentId, + outputSchema: _outputSchema, + attachments: _attachments, + worktree: _worktree, + autoArchive: _autoArchive, env: _env, workspaceId: _workspaceId, initialPrompt: _initialPrompt, diff --git a/packages/client/src/index.ts b/packages/client/src/index.ts index 8485b063af..e113d48667 100644 --- a/packages/client/src/index.ts +++ b/packages/client/src/index.ts @@ -4,6 +4,7 @@ import type { DaemonClientConfig } from "./daemon-client.js"; import type { AgentPermissionResponse } from "@getpaseo/protocol/agent-types"; import type { AgentSnapshotPayload, + CreationSnapshot, CreateAgentRequestMessage, FetchWorkspacesRequestMessage, FetchWorkspacesResponseMessage, @@ -28,7 +29,7 @@ import type { WorkspaceDescriptorPayload, WorkspaceCreateRequest, } from "@getpaseo/protocol/messages"; -import { DaemonClient } from "./daemon-client.js"; +import { DaemonClient, type CreateAgentRequestOptions } from "./daemon-client.js"; import { createTerminalActions, type PaseoTerminalActions, @@ -142,8 +143,16 @@ export interface PaseoWorkspaceOpenOptions { requestId?: string; } -export type PaseoWorkspaceCreateOptions = Omit & { +export type PaseoWorkspaceCreateOptions = Omit< + WorkspaceCreateRequest, + "type" | "requestId" | "agent" | "subscribe" +> & { requestId?: string; + agent?: Omit< + PaseoAgentCreateOptions, + "worktree" | "git" | "onEvent" | "idempotencyKey" | "requestId" + >; + onEvent?: (snapshot: CreationSnapshot) => void; }; export interface PaseoWorkspaceArchiveResult { @@ -232,6 +241,9 @@ export interface PaseoAgentConfig { } export interface PaseoAgentCreateOptions { + idempotencyKey?: string; + agentId?: string; + onEvent?: (snapshot: CreationSnapshot) => void; config: PaseoAgentConfig; cwd: string; parent?: string | PaseoAgentHandle; @@ -513,6 +525,29 @@ export function createPaseoClient(config: PaseoClientConfig): PaseoClient { }; } +function toDaemonAgentCreateOptions( + options: PaseoAgentCreateOptions, + placement?: { workspaceId: string; cwd: string }, +): CreateAgentRequestOptions { + const { config: agentConfig, cwd, parent, title, prompt, ...requestOptions } = options; + const { provider: providerModel, options: providerOptions, ...runtimeConfig } = agentConfig; + const { provider, model } = parseProviderModel(providerModel); + return { + ...requestOptions, + config: { + ...runtimeConfig, + provider, + model, + cwd: placement?.cwd ?? cwd, + ...(title !== undefined ? { title } : {}), + ...(providerOptions !== undefined ? { providerOptions } : {}), + }, + ...(placement ? { workspaceId: placement.workspaceId } : {}), + ...(parent ? { callerAgentId: resolveAgentId(parent) } : {}), + ...(prompt !== undefined ? { initialPrompt: prompt } : {}), + }; +} + export function createPaseoApi( daemonClient: DaemonClient, scopeOptions?: { signal?: AbortSignal }, @@ -555,24 +590,7 @@ export function createPaseoApi( options: PaseoAgentCreateOptions, placement?: { workspaceId: string; cwd: string }, ) => { - const { config: agentConfig, cwd, parent, title, prompt, ...requestOptions } = options; - const { provider: providerModel, options: providerOptions, ...runtimeConfig } = agentConfig; - const { provider, model } = parseProviderModel(providerModel); - const effectiveCwd = placement?.cwd ?? cwd; - const agent = await daemonClient.createAgent({ - ...requestOptions, - config: { - ...runtimeConfig, - provider, - model, - cwd: effectiveCwd, - ...(title !== undefined ? { title } : {}), - ...(providerOptions !== undefined ? { providerOptions } : {}), - }, - ...(placement ? { workspaceId: placement.workspaceId } : {}), - ...(parent ? { callerAgentId: resolveAgentId(parent) } : {}), - ...(prompt !== undefined ? { initialPrompt: prompt } : {}), - }); + const agent = await daemonClient.createAgent(toDaemonAgentCreateOptions(options, placement)); return createAgentHandle(agent); }; const terminals = createTerminalActions(daemonClient, async (workspaceId) => { @@ -694,8 +712,11 @@ export function createPaseoApi( ref: (workspace) => createWorkspaceHandle(workspace), open: (input, requestId) => openWorkspace(daemonClient, createWorkspaceHandle, input, requestId), - create: async ({ requestId, ...options }) => { - const result = await daemonClient.createWorkspace(options, requestId); + create: async ({ requestId, agent, ...options }) => { + const result = await daemonClient.createWorkspace( + { ...options, ...(agent ? { agent: toDaemonAgentCreateOptions(agent) } : {}) }, + requestId, + ); if (result.error || !result.workspace) { throw new Error(result.error ?? "The daemon did not create a workspace"); } diff --git a/packages/protocol/src/messages.ts b/packages/protocol/src/messages.ts index d22834849a..d1e2963985 100644 --- a/packages/protocol/src/messages.ts +++ b/packages/protocol/src/messages.ts @@ -1681,7 +1681,7 @@ export type CreateAgentWorktreeTarget = z.infer; +export type AgentCreateRequest = z.infer; +export type WorkspaceInitialAgent = z.infer; + +export const WorkspaceCreationUpdateSchema = z.object({ + type: z.literal("workspace.create.update"), + payload: CreationSnapshotSchema, +}); +export const AgentCreationUpdateSchema = WorkspaceCreationUpdateSchema.extend({ + type: z.literal("agent.create.update"), +}); +export const CreationSubscribeResponseSchema = z.object({ + type: z.literal("creation.subscribe.response"), + payload: z.object({ + requestId: z.string(), + snapshot: CreationSnapshotSchema.nullable(), + error: z.string().nullable(), + }), +}); +export const AgentCreateResponseSchema = z.object({ + type: z.literal("agent.create.response"), + payload: z.object({ + requestId: z.string(), + agent: AgentSnapshotPayloadSchema.nullable(), + error: z.string().nullable(), + creation: CreationSnapshotSchema.optional(), + }), +}); + export const WorkspaceCreateResponseSchema = z.object({ type: z.literal("workspace.create.response"), payload: z.object({ + agent: AgentSnapshotPayloadSchema.optional(), + creation: CreationSnapshotSchema.optional(), workspace: WorkspaceDescriptorPayloadSchema.nullable(), setupTerminalId: z.string().nullable(), setupSkippedReason: z.string().optional(), @@ -6637,6 +6726,10 @@ export const SessionOutboundMessageSchema = z.discriminatedUnion("type", [ CancelAgentResponseMessageSchema, ClearAgentAttentionResponseMessageSchema, WorkspaceCreateResponseSchema, + AgentCreateResponseSchema, + WorkspaceCreationUpdateSchema, + AgentCreationUpdateSchema, + CreationSubscribeResponseSchema, WorkspaceClearAttentionResponseSchema, WorkspaceMarkUnreadResponseSchema, SendAgentMessageResponseMessageSchema, diff --git a/packages/server/src/server/agent/create-agent/create.ts b/packages/server/src/server/agent/create-agent/create.ts index bd4dcb16d8..814a8f3aec 100644 --- a/packages/server/src/server/agent/create-agent/create.ts +++ b/packages/server/src/server/agent/create-agent/create.ts @@ -55,6 +55,7 @@ export type EnsureWorkspaceForCreate = ( export interface CreateAgentFromSessionInput { kind: "session"; + onAgentReady?: (agent: ManagedAgent) => Promise; agentId?: string; config: AgentSessionConfig; workspaceId: string; @@ -190,6 +191,8 @@ export async function createAgentCommand( agentId: snapshot.id, }); + if (input.kind === "session") await input.onAgentReady?.(snapshot); + let liveSnapshot = snapshot; let initialPromptStarted = false; let initialPromptError: unknown | null = null; diff --git a/packages/server/src/server/authorization/operation-permissions.ts b/packages/server/src/server/authorization/operation-permissions.ts index 6a6a1e78ec..89d0d3f030 100644 --- a/packages/server/src/server/authorization/operation-permissions.ts +++ b/packages/server/src/server/authorization/operation-permissions.ts @@ -62,6 +62,8 @@ const INBOUND_PERMISSION = { client_heartbeat: "workspace.read", close_items_request: "workspace.manage", create_agent_request: ["workspace.write", "hub.execute"], + "agent.create.request": ["workspace.write", "hub.execute"], + "creation.subscribe.request": "workspace.read", create_paseo_worktree_request: "workspace.manage", create_terminal_request: "workspace.write", "daemon.config.reload.request": "daemon.manage", @@ -209,6 +211,10 @@ const INBOUND_PERMISSION = { } as const satisfies Record; const OUTBOUND_PERMISSION = { + "agent.create.response": ["workspace.write", "hub.execute"], + "agent.create.update": ["workspace.write", "hub.execute"], + "workspace.create.update": "workspace.manage", + "creation.subscribe.response": "workspace.read", activity_log: "workspace.read", "agent.config.apply.response": ["workspace.write", "hub.execute"], "agent.detach.response": "workspace.write", diff --git a/packages/server/src/server/creation/creation.e2e.test.ts b/packages/server/src/server/creation/creation.e2e.test.ts new file mode 100644 index 0000000000..13cb5b1109 --- /dev/null +++ b/packages/server/src/server/creation/creation.e2e.test.ts @@ -0,0 +1,104 @@ +import { randomUUID } from "node:crypto"; +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { expect, test } from "vitest"; +import type { CreationSnapshot, SessionOutboundMessage } from "@getpaseo/protocol/messages"; +import { DaemonClient } from "../test-utils/daemon-client.js"; +import { createTestAgentClients } from "../test-utils/fake-agent-client.js"; +import { createTestPaseoDaemon } from "../test-utils/paseo-daemon.js"; + +function deferred() { + let resolve!: (value: T) => void; + const promise = new Promise((done) => { + resolve = done; + }); + return { promise, resolve }; +} + +test("the daemon continues the combined intent after its initiating client closes", async () => { + const provider = deferred(); + const ready = deferred(); + const directory = await mkdtemp(join(tmpdir(), "creation-wire-")); + let agents = 0; + let prompts = 0; + const daemon = await createTestPaseoDaemon({ + agentClients: createTestAgentClients({ + beforeCreateSession: async () => { + agents++; + await provider.promise; + }, + onStartTurn: () => { + prompts++; + }, + }), + }); + const client = new DaemonClient({ + url: `ws://127.0.0.1:${daemon.port}/ws`, + appVersion: "0.7.2", + clientId: "creation-subscriber", + }); + const observer = new DaemonClient({ + url: `ws://127.0.0.1:${daemon.port}/ws`, + appVersion: "0.7.2", + clientId: "creation-subscriber", + }); + const observerMessages: SessionOutboundMessage[] = []; + const unsubscribe = observer.subscribeRawMessages((message) => observerMessages.push(message)); + const workspaceId = "wks_0123456789abcdef"; + const agentId = randomUUID(); + const request = { + idempotencyKey: "workspace-one", + workspaceId, + source: { kind: "directory" as const, path: directory }, + agent: { + agentId, + provider: "codex", + cwd: directory, + initialPrompt: "Create once", + clientMessageId: "initial-one", + }, + }; + try { + await client.connect(); + await observer.connect(); + const first = client.createWorkspace({ + ...request, + onEvent: (snapshot) => { + if (snapshot.phase === "workspace_ready") ready.resolve(snapshot); + }, + }); + void first.catch(() => undefined); + const snapshot = await ready.promise; + expect(snapshot).toMatchObject({ workspaceId, agentId, phase: "workspace_ready" }); + expect((await observer.fetchAgents()).entries).toHaveLength(0); + expect( + observerMessages.filter((message) => message.type === "workspace.create.update"), + ).toHaveLength(0); + await client.close(); + await expect(first).rejects.toThrow("closed"); + // A disconnected form cannot prevent the provider or initial prompt from starting. + provider.resolve(); + await expect.poll(() => prompts).toBe(1); + const replay = await observer.createWorkspace(request); + expect(replay.error).toBeNull(); + expect(replay.workspace?.id).toBe(workspaceId); + expect(replay.agent?.id).toBe(agentId); + expect(replay.agent?.title).toBe("Create once"); + expect(agents).toBe(1); + expect(prompts).toBe(1); + const conflict = await observer.createWorkspace({ + ...request, + idempotencyKey: "another-intent", + }); + expect(conflict.error).toBe("workspace_id_conflict"); + expect(agents).toBe(1); + } finally { + provider.resolve(); + unsubscribe(); + await client.close(); + await observer.close(); + await daemon.close(); + await rm(directory, { recursive: true, force: true }); + } +}, 60000); diff --git a/packages/server/src/server/creation/index.test.ts b/packages/server/src/server/creation/index.test.ts new file mode 100644 index 0000000000..3762e38923 --- /dev/null +++ b/packages/server/src/server/creation/index.test.ts @@ -0,0 +1,251 @@ +import { cp, mkdtemp, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, expect, test } from "vitest"; +import type { + AgentSnapshotPayload, + CreationSnapshot, + WorkspaceDescriptorPayload, +} from "@getpaseo/protocol/messages"; +import { CreationService, type CreationInput } from "./index.js"; + +const directories: string[] = []; +afterEach(async () => { + await Promise.all( + directories.splice(0).map((directory) => rm(directory, { recursive: true, force: true })), + ); +}); +function deferred() { + let resolve!: () => void; + const promise = new Promise((done) => { + resolve = done; + }); + return { promise, resolve }; +} +const workspace: WorkspaceDescriptorPayload = { + id: "wks_0123456789abcdef", + projectId: "project", + projectDisplayName: "Project", + projectRootPath: "/project", + workspaceDirectory: "/project/worktree", + projectKind: "git", + workspaceKind: "worktree", + name: "Worktree", + status: "running", + statusEnteredAt: null, + activityAt: null, + archivingAt: null, + scripts: [], +}; +const agent: AgentSnapshotPayload = { + id: "00000000-0000-4000-8000-000000000001", + workspaceId: workspace.id, + provider: "codex", + cwd: "/project/worktree", + model: null, + createdAt: "2026-09-11T00:00:00Z", + updatedAt: "2026-09-11T00:00:00Z", + lastUserMessageAt: null, + status: "idle", + capabilities: { + supportsStreaming: true, + supportsSessionPersistence: true, + supportsDynamicModes: false, + supportsMcpServers: false, + supportsReasoningStream: false, + supportsToolInvocations: true, + supportsRewindConversation: false, + supportsRewindFiles: false, + supportsRewindBoth: false, + }, + currentModeId: null, + availableModes: [], + pendingPermissions: [], + persistence: null, + title: "Start once", + labels: {}, +}; +async function fixture() { + const directory = await mkdtemp(join(tmpdir(), "creation-lifecycle-")); + directories.push(directory); + const service = new CreationService(directory); + const provider = deferred(); + const ready = deferred(); + const calls: string[] = []; + const input: CreationInput = { + kind: "workspace", + key: "submit-one", + request: { source: "/project", prompt: "Start once" }, + workspaceId: workspace.id, + agentId: agent.id, + hasAgent: true, + hasPrompt: true, + exists: async () => false, + provision: async () => { + calls.push("workspace"); + return { workspace }; + }, + createAgent: async (_id, _workspace, onReady) => { + await provider.promise; + calls.push("agent"); + await onReady(agent); + calls.push("prompt"); + return agent; + }, + }; + const updates: CreationSnapshot[] = []; + const observe = (snapshot: CreationSnapshot) => { + updates.push(snapshot); + if (snapshot.phase === "workspace_ready") ready.resolve(); + }; + return { directory, service, input, provider, ready, calls, updates, observe }; +} + +test("workspace readiness is observable before provider startup and duplicate requests join the same work", async () => { + const f = await fixture(); + const first = f.service.create(f.input, f.observe); + await f.ready.promise; + expect(f.calls).toEqual(["workspace"]); + expect(f.updates.map((s) => s.phase)).toEqual(["accepted", "workspace_ready"]); + expect(f.updates[0]).toMatchObject({ workspaceId: workspace.id, agentId: agent.id }); + const duplicate = f.service.create(f.input); + f.provider.resolve(); + expect(await first).toMatchObject({ phase: "completed", workspace, agent }); + expect(await duplicate).toEqual(await first); + expect(f.calls).toEqual(["workspace", "agent", "prompt"]); + expect(await new CreationService(f.directory).create(f.input)).toEqual(await first); + expect(f.calls).toEqual(["workspace", "agent", "prompt"]); +}); + +test("a failed agent startup retries only that stage with the reserved IDs", async () => { + const f = await fixture(); + let attempts = 0; + f.input.createAgent = async (id, readyWorkspace, onReady) => { + attempts++; + expect(id).toBe(agent.id); + expect(readyWorkspace).toEqual(workspace); + if (attempts === 1) throw new Error("provider unavailable"); + await onReady(agent); + return agent; + }; + expect(await f.service.create(f.input)).toMatchObject({ + phase: "failed", + failedStage: "agent", + outcomeUnknown: false, + workspace, + }); + expect(await new CreationService(f.directory).create(f.input)).toMatchObject({ + phase: "completed", + agent, + }); + expect(f.calls).toEqual(["workspace"]); + expect(attempts).toBe(2); +}); + +test("an uncertain prompt outcome never resends or removes its workspace and agent", async () => { + const f = await fixture(); + let prompts = 0; + f.input.createAgent = async (_id, _workspace, onReady) => { + await onReady(agent); + prompts++; + throw new Error("connection lost after provider accepted prompt"); + }; + const result = await f.service.create(f.input); + expect(result).toMatchObject({ + phase: "failed", + failedStage: "prompt", + outcomeUnknown: true, + workspace, + agent, + }); + expect(await new CreationService(f.directory).create(f.input)).toEqual(result); + expect(prompts).toBe(1); +}); + +test("conflicting intents and resource IDs cannot start more work", async () => { + const f = await fixture(); + const first = f.service.create(f.input, f.observe); + await f.ready.promise; + await expect(f.service.create({ ...f.input, request: { prompt: "changed" } })).rejects.toThrow( + "workspace_request_key_conflict", + ); + await expect(f.service.create({ ...f.input, key: "another-intent" })).rejects.toThrow( + "workspace_id_conflict", + ); + f.provider.resolve(); + await first; + expect(f.calls).toEqual(["workspace", "agent", "prompt"]); +}); + +test("late subscribers recover readiness and removing an observer cannot stop creation", async () => { + const f = await fixture(); + const first = f.service.create(f.input, f.observe); + await f.ready.promise; + const updates: CreationSnapshot[] = []; + const subscription = await f.service.subscribe("workspace", f.input.key, (snapshot) => + updates.push(snapshot), + ); + expect(subscription.snapshot).toMatchObject({ phase: "workspace_ready", workspace }); + subscription.unsubscribe(); + f.provider.resolve(); + await first; + expect(updates).toEqual([]); + const completed = await new CreationService(f.directory).subscribe( + "workspace", + f.input.key, + () => {}, + ); + expect(completed.snapshot).toMatchObject({ phase: "completed", agent, workspace }); + completed.unsubscribe(); +}); + +test("omitted resource IDs are generated once and preserved on replay", async () => { + const f = await fixture(); + f.input.workspaceId = undefined; + f.input.agentId = undefined; + f.input.provision = async (id) => ({ workspace: { ...workspace, id } }); + f.input.createAgent = async (id, createdWorkspace, onReady) => { + const created = { ...agent, id, workspaceId: createdWorkspace!.id }; + await onReady(created); + return created; + }; + const result = await f.service.create(f.input); + expect(result.workspaceId).toMatch(/^wks_[a-f0-9]{16}$/); + expect(result.agentId).toMatch( + /^[a-f0-9]{8}-[a-f0-9]{4}-4[a-f0-9]{3}-[89ab][a-f0-9]{3}-[a-f0-9]{12}$/, + ); + expect(await new CreationService(f.directory).create(f.input)).toEqual(result); +}); + +test("restart after a committed agent without a prompt completes without recreating the agent", async () => { + const f = await fixture(); + const recoveredDirectory = await mkdtemp(join(tmpdir(), "creation-checkpoint-")); + directories.push(recoveredDirectory); + f.input.hasPrompt = false; + let starts = 0; + f.input.createAgent = async (_id, _workspace, onReady) => { + starts++; + await onReady(agent); + // Capture the durable state at the exact milestone a process could stop after. + await cp(f.directory, recoveredDirectory, { recursive: true }); + return agent; + }; + await f.service.create(f.input); + const recovered = await new CreationService(recoveredDirectory).create(f.input); + expect(recovered).toMatchObject({ phase: "completed", workspace, agent }); + expect(starts).toBe(1); +}); + +test("partial provisioning without a registered workspace is not repeated", async () => { + const f = await fixture(); + let provisions = 0; + f.input.provision = async () => { + provisions++; + await writeFile(join(f.directory, "partial-checkout"), "disk was changed"); + throw new Error("Git failed after creating the checkout"); + }; + const result = await f.service.create(f.input); + expect(result).toMatchObject({ phase: "failed", failedStage: "workspace", outcomeUnknown: true }); + expect(await new CreationService(f.directory).create(f.input)).toEqual(result); + expect(provisions).toBe(1); +}); diff --git a/packages/server/src/server/creation/index.ts b/packages/server/src/server/creation/index.ts new file mode 100644 index 0000000000..73e348af62 --- /dev/null +++ b/packages/server/src/server/creation/index.ts @@ -0,0 +1,302 @@ +import { createHash, randomUUID } from "node:crypto"; +import { mkdir, readFile, writeFile } from "node:fs/promises"; +import { join } from "node:path"; +import { z } from "zod"; +import { + CreationSnapshotSchema, + type AgentSnapshotPayload, + type CreationSnapshot, + type WorkspaceDescriptorPayload, +} from "@getpaseo/protocol/messages"; +import { writeJsonFileAtomic } from "../atomic-file.js"; +import { generateWorkspaceId } from "../workspace-registry-model.js"; + +type Observer = (snapshot: CreationSnapshot) => void; +export interface CreationInput { + kind: "workspace" | "agent"; + key: string; + request: unknown; + workspaceId?: string; + agentId?: string; + hasAgent: boolean; + hasPrompt: boolean; + exists: (kind: "workspace" | "agent", id: string) => Promise; + provision?: ( + workspaceId: string, + ) => Promise<{ workspace: WorkspaceDescriptorPayload; setupSkippedReason?: string }>; + createAgent?: ( + agentId: string, + workspace: WorkspaceDescriptorPayload | undefined, + onReady: (agent: AgentSnapshotPayload) => Promise, + ) => Promise; +} +const RecordSchema = z.object({ + fingerprint: z.string(), + snapshot: CreationSnapshotSchema, + inFlight: z.enum(["workspace", "agent", "prompt"]).nullable(), +}); +type Record = z.infer; + +/** Owns creation across socket lifetimes. Resource owners perform work; this module commits its milestones. */ +export class CreationService { + private admission: Promise = Promise.resolve(); + private readonly active = new Map>(); + private readonly observers = new Map>(); + constructor( + private readonly directory: string, + private readonly validateCompleted: ( + snapshot: CreationSnapshot, + ) => Promise = async () => {}, + ) {} + + async create(input: CreationInput, observer?: Observer): Promise { + const identity = identityFor(input.kind, input.key); + const admitted = this.admission.then(async () => { + await mkdir(this.directory, { recursive: true }); + let record = await this.read(identity); + const fingerprint = digest(input.request); + if (record && record.fingerprint !== fingerprint) + throw new Error(`${input.kind}_request_key_conflict`); + const running = this.active.get(identity); + if (running && record) return { completion: running, snapshot: record.snapshot }; + if (record?.snapshot.phase === "completed") await this.validateCompleted(record.snapshot); + if (record?.snapshot.phase === "completed" || record?.snapshot.outcomeUnknown) { + return { completion: Promise.resolve(record.snapshot), snapshot: record.snapshot }; + } + if (record?.inFlight) { + await this.publish(identity, record, { + phase: "failed", + error: `${input.kind}_request_outcome_unknown`, + failedStage: record.inFlight, + outcomeUnknown: true, + }); + return { completion: Promise.resolve(record.snapshot), snapshot: record.snapshot }; + } + if (!record) { + record = initialRecord(input, fingerprint); + // Persist identity before claiming resources. A failed admission can only retry these IDs. + await this.write(identity, record); + } + await this.claimResources(identity, record.snapshot, input); + if (record.snapshot.phase === "failed") { + await this.publish(identity, record, { + phase: record.snapshot.workspace ? "workspace_ready" : "accepted", + error: null, + errorCode: undefined, + failedStage: undefined, + outcomeUnknown: undefined, + }); + } + const snapshot = record.snapshot; + // Defer the runner until admission has installed the initiating observer. + const completion = Promise.resolve().then(() => this.run(identity, record, input)); + this.active.set(identity, completion); + void completion.finally(() => this.active.delete(identity)).catch(() => undefined); + return { completion, snapshot }; + }); + this.admission = admitted.then( + () => undefined, + () => undefined, + ); + // Install before admission starts, so even a fast local provision cannot outrun observation. + const unsubscribe = observer ? this.observe(identity, observer) : () => {}; + try { + const { completion, snapshot } = await admitted; + if (observer) this.notify(observer, snapshot); + return await completion; + } finally { + unsubscribe(); + } + } + + async subscribe( + kind: CreationSnapshot["kind"], + key: string, + observer: Observer, + ): Promise<{ snapshot: CreationSnapshot | null; unsubscribe: () => void }> { + const identity = identityFor(kind, key); + const unsubscribe = this.observe(identity, observer); + try { + await this.admission; + const record = await this.read(identity); + if (record?.inFlight && !this.active.has(identity)) { + await this.publish(identity, record, { + phase: "failed", + failedStage: record.inFlight, + outcomeUnknown: true, + error: `${kind}_request_outcome_unknown`, + }); + } + if (record?.snapshot.phase === "completed") await this.validateCompleted(record.snapshot); + return { snapshot: record?.snapshot ?? null, unsubscribe }; + } catch (error) { + unsubscribe(); + throw error; + } + } + + private async run( + identity: string, + record: Record, + input: CreationInput, + ): Promise { + try { + if (input.provision && !record.snapshot.workspace) { + record.inFlight = "workspace"; + await this.write(identity, record); + const { workspace, setupSkippedReason } = await input.provision( + record.snapshot.workspaceId!, + ); + record.inFlight = null; + await this.publish(identity, record, { + phase: "workspace_ready", + workspace, + workspaceId: workspace.id, + setupSkippedReason, + }); + } + if (input.hasAgent && input.createAgent && !record.snapshot.agent) { + record.inFlight = "agent"; + await this.write(identity, record); + const agent = await input.createAgent( + record.snapshot.agentId!, + record.snapshot.workspace, + async (readyAgent) => { + record.inFlight = input.hasPrompt ? "prompt" : null; + await this.publish(identity, record, { + phase: "agent_ready", + agent: readyAgent, + workspaceId: readyAgent.workspaceId ?? record.snapshot.workspaceId, + }); + }, + ); + record.inFlight = null; + if (input.hasPrompt) + await this.publish(identity, record, { phase: "prompt_started", agent }); + else record.snapshot = { ...record.snapshot, agent }; + } + record.inFlight = null; + await this.publish(identity, record, { phase: "completed" }); + } catch (error) { + const stage = record.inFlight ?? "agent"; + const resourceId = + stage === "workspace" ? record.snapshot.workspaceId : record.snapshot.agentId; + const code = + error instanceof Error && "code" in error && typeof error.code === "string" + ? error.code + : undefined; + const rejectedBeforeProvision = code === "directory_not_found" || code === "source_required"; + const unknown = + stage === "prompt" || + (stage === "workspace" && !rejectedBeforeProvision) || + (resourceId !== null && (await input.exists(stage, resourceId))); + // A failed provider startup with no registered resource can retry under the same ID. + // A partially provisioned resource or an attempted prompt needs reconciliation. + if (!unknown) record.inFlight = null; + await this.publish(identity, record, { + phase: "failed", + error: error instanceof Error ? error.message : String(error), + errorCode: code, + failedStage: stage, + outcomeUnknown: unknown, + }); + } + return record.snapshot; + } + + private async claimResources( + identity: string, + snapshot: CreationSnapshot, + input: CreationInput, + ): Promise { + if (input.kind === "workspace" && snapshot.workspaceId) + await this.claim(identity, "workspace", snapshot.workspaceId, input.exists); + if (snapshot.agentId) await this.claim(identity, "agent", snapshot.agentId, input.exists); + } + + private async claim( + identity: string, + kind: "workspace" | "agent", + id: string, + exists: CreationInput["exists"], + ): Promise { + const file = join(this.directory, `${digest([kind, id])}.claim`); + const owner = await readOptional(file); + if (owner !== null) { + if (owner !== identity) throw new Error(`${kind}_id_conflict`); + return; + } + if (await exists(kind, id)) throw new Error(`${kind}_id_conflict`); + await writeFile(file, identity, { flag: "wx", mode: 0o600 }); + } + private observe(identity: string, observer: Observer): () => void { + const observers = this.observers.get(identity) ?? new Set(); + observers.add(observer); + this.observers.set(identity, observers); + return () => { + observers.delete(observer); + if (!observers.size) this.observers.delete(identity); + }; + } + private notify(observer: Observer, snapshot: CreationSnapshot): void { + try { + observer(snapshot); + } catch { + /* A disconnected observer cannot cancel accepted work. */ + } + } + private async publish( + identity: string, + record: Record, + update: Partial, + ): Promise { + record.snapshot = { ...record.snapshot, ...update, revision: record.snapshot.revision + 1 }; + await this.write(identity, record); + for (const observer of this.observers.get(identity) ?? []) + this.notify(observer, record.snapshot); + } + private write(identity: string, record: Record): Promise { + return writeJsonFileAtomic(join(this.directory, `${identity}.json`), record); + } + private async read(identity: string): Promise { + const text = await readOptional(join(this.directory, `${identity}.json`)); + return text === null ? null : RecordSchema.parse(JSON.parse(text)); + } +} +function initialRecord(input: CreationInput, fingerprint: string): Record { + return { + fingerprint, + inFlight: null, + snapshot: { + kind: input.kind, + idempotencyKey: input.key, + revision: 0, + phase: "accepted", + error: null, + workspaceId: input.workspaceId ?? (input.kind === "workspace" ? generateWorkspaceId() : null), + agentId: input.hasAgent ? (input.agentId ?? randomUUID()) : null, + }, + }; +} +function identityFor(kind: string, key: string): string { + return digest([kind, key]); +} +function digest(input: unknown): string { + return createHash("sha256") + .update( + JSON.stringify(input, (_key, value: unknown) => { + if (value && typeof value === "object" && !Array.isArray(value)) + return Object.fromEntries(Object.entries(value).sort(([a], [b]) => a.localeCompare(b))); + return value; + }), + ) + .digest("hex"); +} +async function readOptional(file: string): Promise { + try { + return await readFile(file, "utf8"); + } catch (error) { + if (error instanceof Error && "code" in error && error.code === "ENOENT") return null; + throw error; + } +} diff --git a/packages/server/src/server/session.test.ts b/packages/server/src/server/session.test.ts index e5a0572581..9c77dceebf 100644 --- a/packages/server/src/server/session.test.ts +++ b/packages/server/src/server/session.test.ts @@ -1,4 +1,7 @@ -import { createRequestReceiptsStub } from "./test-utils/session-stubs.js"; +import { + createRequestReceiptsStub, + createCreationServiceStub, +} from "./test-utils/session-stubs.js"; import { execSync } from "child_process"; import { existsSync, mkdtempSync, realpathSync, rmSync, symlinkSync, writeFileSync } from "fs"; import { tmpdir } from "os"; @@ -365,6 +368,7 @@ function createSessionForTest(options: SessionForTestOptions = {}): Session { const sessionOptions: SessionOptions = { requestReceipts: createRequestReceiptsStub(), + creationService: createCreationServiceStub(), clientId: options.clientId ?? "test-client", onMessage: (message) => messages.push(message), ...(options.targetedMessages diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index 34be0cc271..c81456a927 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -1,6 +1,10 @@ import type { BrowserToolsBroker } from "./browser-tools/broker.js"; import { BrowserAutomationHostCapabilitySchema } from "@getpaseo/protocol/browser-automation/capabilities"; import type { SessionEventSubscription } from "@getpaseo/protocol/messages"; +import { relative } from "node:path"; +import { isAbsolute } from "node:path"; +import { CreationService } from "./creation/index.js"; +import type { CreationSnapshot, AgentCreateRequest } from "@getpaseo/protocol/messages"; import type { RequestReceipts } from "./request-receipts/index.js"; import equal from "fast-deep-equal"; import { SessionDelivery, type OwnedSubscription } from "./session/owned-subscriptions/index.js"; @@ -460,6 +464,7 @@ export interface SessionOptions { agentManager: AgentManager; agentStorage: AgentStorage; requestReceipts: Pick; + creationService: Pick; projectRegistry: ProjectRegistry; workspaceRegistry: WorkspaceRegistry; directorySync?: DirectorySyncService; @@ -637,6 +642,16 @@ function resolveWorkspaceLabelService( return service ?? null; } +function hasInitialCreationPrompt( + request?: Pick, +): boolean { + return Boolean( + request && + buildAgentPrompt(request.initialPrompt?.trim() ?? "", request.images, request.attachments) + .length, + ); +} + function workspaceLabelErrorCode(error: unknown): string { if ( error instanceof WorkspaceLabelError || @@ -786,6 +801,9 @@ export class Session { "createAgent" | "createWorkspace" | "sendMessage" >; private readonly createAgentLifecycleDispatch: CreateAgentLifecycleDispatch; + private readonly creationService: Pick; + private readonly defaultCreationSource = {}; + private readonly creationSubscriptions = new Map void>>(); constructor(options: SessionOptions) { const { @@ -860,6 +878,7 @@ export class Session { this.pushNotifications = pushNotifications; this.paseoHome = paseoHome; this.requestReceipts = options.requestReceipts; + this.creationService = options.creationService; this.projectIcons = new ProjectIconReader(paseoHome); this.worktreesRoot = worktreesRoot; this.pluginRuntime = pluginRuntime; @@ -1242,6 +1261,8 @@ export class Session { .catch((err) => this.sessionLogger.error({ err }, "Failed to release source subscriptions")); this.clientSources.delete(source); this.refreshObservationProducers(); + for (const unsubscribe of this.creationSubscriptions.get(source)?.values() ?? []) unsubscribe(); + this.creationSubscriptions.delete(source); } private subscribeAgentTimelines(agentIds: string[]): OwnedSubscription { @@ -2263,6 +2284,7 @@ export class Session { this.dispatchAgentRelationshipMessage(msg) ?? this.dispatchAgentTimelineMessage(msg, source) ?? this.dispatchHubExecutionMessage(msg) ?? + this.dispatchCreationMessage(msg, source) ?? this.dispatchAgentLifecycleMessage(msg) ?? this.dispatchAgentConfigMessage(msg) ?? this.dispatchCheckoutMessage(msg) ?? @@ -2639,6 +2661,24 @@ export class Session { return undefined; } + private dispatchCreationMessage( + msg: SessionInboundMessage, + source?: object, + ): Promise | undefined { + switch (msg.type) { + case "agent.create.request": + return this.handleAgentCreation(msg, source); + case "creation.subscribe.request": + return this.handleCreationSubscription(msg, source); + case "workspace.create.request": + if (msg.subscribe || msg.agent || msg.workspaceId) + return this.handleWorkspaceCreation(msg, source); + return undefined; + default: + return undefined; + } + } + private dispatchAgentLifecycleMessage(msg: SessionInboundMessage): Promise | undefined { switch (msg.type) { case "fetch_agents_request": @@ -3822,6 +3862,201 @@ export class Session { /** * Handle create agent request */ + private creationObserver(snapshot: CreationSnapshot, source?: object): void { + if (this.isCleanedUp) return; + const message: SessionOutboundMessage = { + type: snapshot.kind === "workspace" ? "workspace.create.update" : "agent.create.update", + payload: snapshot, + }; + this.emitForSource(message, source); + } + + private async handleCreationSubscription( + request: Extract, + source?: object, + ): Promise { + const key = JSON.stringify([request.kind, request.idempotencyKey]); + const owner = source ?? this.defaultCreationSource; + const subscriptions = this.creationSubscriptions.get(owner) ?? new Map void>(); + subscriptions.get(key)?.(); + subscriptions.delete(key); + this.creationSubscriptions.set(owner, subscriptions); + const observation = await this.creationService.subscribe( + request.kind, + request.idempotencyKey, + (snapshot) => this.creationObserver(snapshot, source), + ); + if (request.subscribe === false || this.isCleanedUp) observation.unsubscribe(); + else subscriptions.set(key, observation.unsubscribe); + this.emitForSource( + { + type: "creation.subscribe.response", + payload: { requestId: request.requestId, snapshot: observation.snapshot, error: null }, + }, + source, + ); + } + + private creationResourceExists = async ( + kind: "workspace" | "agent", + id: string, + ): Promise => + kind === "workspace" + ? (await this.workspaceRegistry.get(id)) !== null + : this.agentManager.getAgent(id) != null || (await this.agentStorage.get(id)) !== null; + + private async handleAgentCreation(request: AgentCreateRequest, source?: object): Promise { + const { requestId, subscribe, type: _type, idempotencyKey, ...intent } = request; + try { + const creation = await this.creationService.create( + { + kind: "agent", + key: idempotencyKey ?? requestId, + request: intent, + workspaceId: request.workspaceId, + agentId: request.agentId, + hasAgent: true, + hasPrompt: hasInitialCreationPrompt(request), + exists: this.creationResourceExists, + createAgent: (id, _workspace, onReady) => + this.createSessionAgent({ ...request, type: "create_agent_request" }, id, onReady), + }, + subscribe ? (snapshot) => this.creationObserver(snapshot, source) : undefined, + ); + this.emitForSource( + { + type: "agent.create.response", + payload: { requestId, agent: creation.agent ?? null, error: creation.error, creation }, + }, + source, + ); + } catch (error) { + this.emitForSource( + { + type: "agent.create.response", + payload: { + requestId, + agent: null, + error: error instanceof Error ? error.message : String(error), + }, + }, + source, + ); + } + } + + private async handleWorkspaceCreation( + request: Extract, + source?: object, + ): Promise { + const { requestId, subscribe, type: _type, idempotencyKey, ...intent } = request; + try { + if ( + request.agent && + !this.authorization.allowsPermission("workspace.write") && + !this.authorization.allowsPermission("hub.execute") + ) { + throw new Error("Session is not authorized to create an agent"); + } + const agentInput = request.agent; + const creation = await this.creationService.create( + { + kind: "workspace", + key: idempotencyKey ?? requestId, + request: intent, + workspaceId: request.workspaceId, + agentId: agentInput?.agentId, + hasAgent: Boolean(agentInput), + hasPrompt: hasInitialCreationPrompt(agentInput), + exists: this.creationResourceExists, + provision: async (id) => { + const descriptor = await this.createRequestedWorkspace( + { + ...request, + firstAgentContext: agentInput + ? { prompt: agentInput.initialPrompt, attachments: agentInput.attachments } + : request.firstAgentContext, + }, + id, + ); + const workspace = await this.workspaceRegistry.get(descriptor.id); + return { + workspace: descriptor, + setupSkippedReason: workspace?.untrustedSource + ? formatWorkspaceAutomationBlockedMessage(workspace.untrustedSource) + : undefined, + }; + }, + createAgent: agentInput + ? async (id, workspace, onReady) => { + if (!workspace?.workspaceDirectory) + throw new Error("Created workspace has no directory"); + const sourceCwd = + request.source.kind === "directory" + ? request.source.path + : await resolveWorktreeSourceCwd(request.source, this.projectRegistry); + const relativeCwd = relative(resolve(sourceCwd), resolve(agentInput.config.cwd)); + if ( + relativeCwd === ".." || + relativeCwd.startsWith(`..${sep}`) || + isAbsolute(relativeCwd) + ) + throw new Error("Agent directory must be inside the workspace source"); + return this.createSessionAgent( + { + ...agentInput, + type: "create_agent_request", + requestId, + config: { + ...agentInput.config, + cwd: resolve(workspace.workspaceDirectory, relativeCwd), + }, + workspaceId: workspace.id, + }, + id, + onReady, + ); + } + : undefined, + }, + subscribe ? (snapshot) => this.creationObserver(snapshot, source) : undefined, + ); + this.emitForSource( + { + type: "workspace.create.response", + payload: { + requestId, + workspace: creation.workspace ?? null, + agent: creation.agent, + creation, + setupSkippedReason: creation.setupSkippedReason, + error: creation.error, + errorCode: creation.errorCode, + setupTerminalId: null, + }, + }, + source, + ); + } catch (error) { + this.emitForSource( + { + type: "workspace.create.response", + payload: { + requestId, + workspace: null, + error: error instanceof Error ? error.message : String(error), + errorCode: + error instanceof Error && "code" in error && typeof error.code === "string" + ? error.code + : undefined, + setupTerminalId: null, + }, + }, + source, + ); + } + } + private async handleCreateAgentRequest(msg: CreateAgentRequestMessage): Promise { try { let agent: AgentSnapshotPayload; @@ -3882,6 +4117,7 @@ export class Session { private async createSessionAgent( msg: CreateAgentRequestMessage, agentId?: string, + onAgentReady?: (agent: AgentSnapshotPayload) => Promise, ): Promise { const { config, @@ -3951,6 +4187,10 @@ export class Session { }, { kind: "session", + onAgentReady: async (agent) => { + createdAgentId = agent.id; + await onAgentReady?.(await this.buildAgentPayload(agent)); + }, agentId, config: resolvedIntent.config, workspaceId: resolvedIntent.intent.workspaceId, @@ -8250,6 +8490,9 @@ export class Session { this.sessionLogger.trace({}, "agent.session.lifecycle.cleanup"); this.isCleanedUp = true; await this.delivery.close(); + for (const subscriptions of this.creationSubscriptions.values()) + for (const unsubscribe of subscriptions.values()) unsubscribe(); + this.creationSubscriptions.clear(); if (this.unsubscribeAgentEvents) { this.unsubscribeAgentEvents(); diff --git a/packages/server/src/server/session.workspace-git-watch.test.ts b/packages/server/src/server/session.workspace-git-watch.test.ts index a78b2210b1..3d33904374 100644 --- a/packages/server/src/server/session.workspace-git-watch.test.ts +++ b/packages/server/src/server/session.workspace-git-watch.test.ts @@ -1,4 +1,7 @@ -import { createRequestReceiptsStub } from "./test-utils/session-stubs.js"; +import { + createRequestReceiptsStub, + createCreationServiceStub, +} from "./test-utils/session-stubs.js"; import { describe, expect, test, vi } from "vitest"; import path from "node:path"; import type pino from "pino"; @@ -186,6 +189,7 @@ function createSessionForWorkspaceGitWatchTests(options?: { const session = new Session({ requestReceipts: createRequestReceiptsStub(), + creationService: createCreationServiceStub(), clientId: "test-client", permissions: OWNER_PERMISSIONS, onMessage: (message) => emitted.push(message as { type: string; payload: unknown }), diff --git a/packages/server/src/server/session.workspace-resolution-invariants.test.ts b/packages/server/src/server/session.workspace-resolution-invariants.test.ts index 69714acd16..207e07956a 100644 --- a/packages/server/src/server/session.workspace-resolution-invariants.test.ts +++ b/packages/server/src/server/session.workspace-resolution-invariants.test.ts @@ -1,4 +1,7 @@ -import { createRequestReceiptsStub } from "./test-utils/session-stubs.js"; +import { + createRequestReceiptsStub, + createCreationServiceStub, +} from "./test-utils/session-stubs.js"; // Invariant tests for cwd → workspace resolution on the open_project_request path. // Each test encodes a default behavior we want from `findOrCreateWorkspaceForDirectory`. // Run to see which invariants the current code already satisfies (green) and which @@ -90,6 +93,7 @@ function createHarness(input: { const session = new Session({ requestReceipts: createRequestReceiptsStub(), + creationService: createCreationServiceStub(), clientId: "test", permissions: OWNER_PERMISSIONS, appVersion: null, diff --git a/packages/server/src/server/session.workspaces.test.ts b/packages/server/src/server/session.workspaces.test.ts index 61212705a8..f2c63ccca3 100644 --- a/packages/server/src/server/session.workspaces.test.ts +++ b/packages/server/src/server/session.workspaces.test.ts @@ -1,4 +1,7 @@ -import { createRequestReceiptsStub } from "./test-utils/session-stubs.js"; +import { + createRequestReceiptsStub, + createCreationServiceStub, +} from "./test-utils/session-stubs.js"; import { execFileSync } from "node:child_process"; import { existsSync, @@ -643,6 +646,7 @@ function createSessionForWorkspaceTests( const session = asTestSession( new Session({ requestReceipts: createRequestReceiptsStub(), + creationService: createCreationServiceStub(), clientId: "test-client", permissions: OWNER_PERMISSIONS, appVersion: options.appVersion ?? null, @@ -1004,6 +1008,7 @@ test("create_agent_request keeps requested child cwd when grouped under an exist const session = asTestSession( new Session({ requestReceipts: createRequestReceiptsStub(), + creationService: createCreationServiceStub(), clientId: "test-client", serverId: "test-server", permissions: OWNER_PERMISSIONS, @@ -1158,6 +1163,7 @@ test("create_agent_request launches from an exact subdirectory in a created work const emitted: SessionOutboundMessage[] = []; const session = new Session({ requestReceipts: createRequestReceiptsStub(), + creationService: createCreationServiceStub(), clientId: "test-client", permissions: OWNER_PERMISSIONS, appVersion: null, @@ -1296,6 +1302,7 @@ test("create_agent_request does not title an existing workspace from the agent p const session = asTestSession( new Session({ requestReceipts: createRequestReceiptsStub(), + creationService: createCreationServiceStub(), clientId: "test-client", permissions: OWNER_PERMISSIONS, appVersion: null, @@ -1566,6 +1573,7 @@ test("archive emits an authoritative agent_update upsert for subscribed clients" const session = asTestSession( new Session({ requestReceipts: createRequestReceiptsStub(), + creationService: createCreationServiceStub(), clientId: "test-client", permissions: OWNER_PERMISSIONS, onMessage: (message) => emitted.push(message), @@ -2049,6 +2057,7 @@ test("close_items_request archives agents and kills terminals in one batch", asy const session = asTestSession( new Session({ requestReceipts: createRequestReceiptsStub(), + creationService: createCreationServiceStub(), clientId: "test-client", permissions: OWNER_PERMISSIONS, onMessage: (message) => emitted.push(message), @@ -2218,6 +2227,7 @@ test("close_items_request archives stored agents that are not currently loaded", const session = asTestSession( new Session({ requestReceipts: createRequestReceiptsStub(), + creationService: createCreationServiceStub(), clientId: "test-client", permissions: OWNER_PERMISSIONS, onMessage: (message) => emitted.push(message), @@ -2378,6 +2388,7 @@ test("close_items_request continues after an archive failure", async () => { const session = asTestSession( new Session({ requestReceipts: createRequestReceiptsStub(), + creationService: createCreationServiceStub(), clientId: "test-client", permissions: OWNER_PERMISSIONS, onMessage: (message) => emitted.push(message), @@ -3650,6 +3661,7 @@ test("workspace update stream keeps persisted workspace visible after agents sto const session = asTestSession( new Session({ requestReceipts: createRequestReceiptsStub(), + creationService: createCreationServiceStub(), clientId: "test-client", permissions: OWNER_PERMISSIONS, onMessage: (message) => emitted.push(message), diff --git a/packages/server/src/server/snapshot-mutation-ownership.test.ts b/packages/server/src/server/snapshot-mutation-ownership.test.ts index bc6d01c203..63aa871d5e 100644 --- a/packages/server/src/server/snapshot-mutation-ownership.test.ts +++ b/packages/server/src/server/snapshot-mutation-ownership.test.ts @@ -1,4 +1,7 @@ -import { createRequestReceiptsStub } from "./test-utils/session-stubs.js"; +import { + createRequestReceiptsStub, + createCreationServiceStub, +} from "./test-utils/session-stubs.js"; import os from "node:os"; import path from "node:path"; import { mkdtempSync, rmSync } from "node:fs"; @@ -101,6 +104,7 @@ describe("snapshot mutation ownership boundary", () => { const session = asInternals( new Session({ requestReceipts: createRequestReceiptsStub(), + creationService: createCreationServiceStub(), clientId: "test-client", permissions: OWNER_PERMISSIONS, onMessage, diff --git a/packages/server/src/server/test-utils/fake-agent-client.ts b/packages/server/src/server/test-utils/fake-agent-client.ts index 323e9944c7..2a437f9bbf 100644 --- a/packages/server/src/server/test-utils/fake-agent-client.ts +++ b/packages/server/src/server/test-utils/fake-agent-client.ts @@ -62,6 +62,7 @@ interface FakeAgentSessionOptions { } export interface TestAgentClientOptions { + beforeCreateSession?: () => Promise; closeSession?: () => Promise; onStartTurn?: (prompt: AgentPromptInput) => void; supportsMcpServers?: boolean; @@ -1212,6 +1213,7 @@ class FakeAgentClient implements AgentClient { config: AgentSessionConfig, _launchContext?: AgentLaunchContext, ): Promise { + await this.options.beforeCreateSession?.(); return new FakeAgentSession({ providerName: this.provider, config: { ...config }, diff --git a/packages/server/src/server/test-utils/session-stubs.ts b/packages/server/src/server/test-utils/session-stubs.ts index 15879f4b9e..03acd72fee 100644 --- a/packages/server/src/server/test-utils/session-stubs.ts +++ b/packages/server/src/server/test-utils/session-stubs.ts @@ -308,3 +308,14 @@ export function createProviderSnapshot( }), }; } + +export function createCreationServiceStub(): SessionOptions["creationService"] { + return { + create: async () => { + throw new Error("Unexpected creation in legacy session fixture"); + }, + subscribe: async () => { + throw new Error("Unexpected creation subscription in legacy session fixture"); + }, + }; +} diff --git a/packages/server/src/server/websocket-server.ts b/packages/server/src/server/websocket-server.ts index 9760acf973..8f40f5791c 100644 --- a/packages/server/src/server/websocket-server.ts +++ b/packages/server/src/server/websocket-server.ts @@ -1,3 +1,6 @@ +import { stat } from "node:fs/promises"; +import type { CreationSnapshot } from "@getpaseo/protocol/messages"; +import { CreationService } from "./creation/index.js"; import { RequestReceipts } from "./request-receipts/index.js"; import { WebSocket, WebSocketServer } from "ws"; import type { IncomingMessage, Server as HTTPServer } from "http"; @@ -530,6 +533,7 @@ export class VoiceAssistantWebSocketServer { private readonly agentManager: AgentManager; private readonly agentStorage: AgentStorage; private readonly requestReceipts: RequestReceipts; + private readonly creationService: CreationService; private readonly projectRegistry: ProjectRegistry; private readonly workspaceRegistry: WorkspaceRegistry; private readonly workspaceLabelService: WorkspaceLabelService | null; @@ -585,6 +589,25 @@ export class VoiceAssistantWebSocketServer { private readonly pluginRuntime: SessionOptions["pluginRuntime"]; private readonly orchestrationSkills: SessionOptions["orchestrationSkills"]; + private async validateCompletedCreation(snapshot: CreationSnapshot): Promise { + if (snapshot.workspace && snapshot.kind === "workspace") { + const workspace = await this.workspaceRegistry.get(snapshot.workspace.id); + if (!workspace) throw new Error("Previously created workspace no longer exists"); + const directory = await stat(workspace.cwd).catch(() => null); + if (!directory?.isDirectory()) + throw Object.assign(new Error(`Directory not found: ${workspace.cwd}`), { + code: "directory_not_found", + }); + } + if ( + snapshot.agentId && + !this.agentManager.getAgent(snapshot.agentId) && + !(await this.agentStorage.get(snapshot.agentId)) + ) { + throw new Error("Previously created agent no longer exists"); + } + } + constructor( server: HTTPServer, logger: pino.Logger, @@ -650,6 +673,9 @@ export class VoiceAssistantWebSocketServer { this.agentManager = agentManager; this.agentStorage = agentStorage; this.requestReceipts = new RequestReceipts(join(paseoHome, "agent-requests")); + this.creationService = new CreationService(join(paseoHome, "creations"), (snapshot) => + this.validateCompletedCreation(snapshot), + ); this.projectRegistry = projectRegistry ?? createNoopProjectRegistry(); this.workspaceRegistry = workspaceRegistry ?? createNoopWorkspaceRegistry(); this.workspaceLabelService = workspaceLabelService ?? null; @@ -1426,6 +1452,7 @@ export class VoiceAssistantWebSocketServer { agentManager: this.agentManager, agentStorage: this.agentStorage, requestReceipts: this.requestReceipts, + creationService: this.creationService, projectRegistry: this.projectRegistry, workspaceRegistry: this.workspaceRegistry, workspaceLabelService: this.workspaceLabelService ?? undefined, @@ -1647,6 +1674,7 @@ export class VoiceAssistantWebSocketServer { ownedSubscriptions: true, agentRequestReceipts: true, workspaceRequestReceipts: true, + creationLifecycle: true, hubAgentRpc: true, // COMPAT(directorySync): added in v0.3.x, remove gate after 2027-02-12. directorySync: true, diff --git a/packages/server/src/server/wire-compat.test.ts b/packages/server/src/server/wire-compat.test.ts index 7fa159de43..0b2d899326 100644 --- a/packages/server/src/server/wire-compat.test.ts +++ b/packages/server/src/server/wire-compat.test.ts @@ -1,4 +1,7 @@ -import { createRequestReceiptsStub } from "./test-utils/session-stubs.js"; +import { + createRequestReceiptsStub, + createCreationServiceStub, +} from "./test-utils/session-stubs.js"; import pino from "pino"; import { z } from "zod"; import { describe, expect, test } from "vitest"; @@ -209,6 +212,7 @@ function createSessionForWireCompatTest(options?: { const session = new Session({ requestReceipts: createRequestReceiptsStub(), + creationService: createCreationServiceStub(), clientId: "wire-compat-client", permissions: OWNER_PERMISSIONS, clientCapabilities: options?.clientCapabilities ?? null,