fix(creation): keep initial prompts in daemon-owned creation requests

This commit is contained in:
Mohamed Boudra
2026-09-14 10:38:49 +02:00
parent fe8524ad96
commit 81aea18e72
31 changed files with 2140 additions and 260 deletions
+24
View File
@@ -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`:
@@ -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);
});
+26 -15
View File
@@ -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<typeof gate.getClientRequests>[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);
},
};
@@ -559,6 +559,8 @@ export interface AgentCreatedDelayControl {
release(): void;
waitForCreateRequest(): Promise<void>;
waitForDelayedCreatedStatus(): Promise<void>;
expectSingleWorkspaceIntent(): void;
fail(error: string): void;
}
export async function delayBrowserAgentCreatedStatus(
@@ -566,8 +568,10 @@ export async function delayBrowserAgentCreatedStatus(
): Promise<AgentCreatedDelayControl> {
const daemonPortPattern = daemonWsRoutePattern();
const createRequestIds = new Set<string>();
const creationRequests: Record<string, unknown>[] = [];
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<void>((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<string, unknown>)
: 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,
};
-24
View File
@@ -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<DaemonClient, "createAgent" | "sendMessage">,
options: Omit<CreateAgentRequestOptions, "outputSchema"> & {
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;
}
@@ -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,
],
);
@@ -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 });
+151 -145
View File
@@ -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<AgentSnapshotPayload>;
retry: (input: CreateAgentRequestOptions) => Promise<AgentSnapshotPayload>;
};
serverId: string;
draftKey: string;
clearDraft: (lifecycle: "sent" | "abandoned") => void;
@@ -795,30 +797,6 @@ interface WorkspaceDraftSubmissionConfig {
target: WorkspaceTabTarget;
}
async function createAndMergeWorkspace(input: {
client: NonNullable<ReturnType<typeof useHostRuntimeClient>>;
createInput: Parameters<
NonNullable<ReturnType<typeof useHostRuntimeClient>>["createPaseoWorktree"]
>[0];
mergeWorkspaces: (
serverId: string,
workspaces: ReturnType<typeof normalizeWorkspaceDescriptor>[],
) => void;
serverId: string;
createFailedMessage: string;
}): Promise<ReturnType<typeof normalizeWorkspaceDescriptor>> {
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<ReturnType<typeof normalizeWorkspaceDescriptor>>;
serverId: string;
draftKey: string;
@@ -945,7 +932,18 @@ function buildComposerInitialValues(input: {
return undefined;
}
async function runCreateChatAgent(input: CreateChatAgentInput): Promise<SubmitOutcome> {
const pendingWorkspaceSubmissions = new Map<string, Promise<SubmitOutcome>>();
function runCreateChatAgent(input: CreateChatAgentInput): Promise<SubmitOutcome> {
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<SubmitOutcome> {
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<SubmitOu
format: attachmentSubmitFormat,
});
const workspaceNamingAttachments = getWorkspaceNamingAttachments(reviewAttachments);
const ensuredWorkspace = await ensureWorkspace({
cwd,
prompt: text,
attachments: workspaceNamingAttachments,
withInitialAgent: true,
const wirePayload = splitComposerAttachmentsForSubmit(attachments, {
format: attachmentSubmitFormat,
});
const initialSetup = buildWorkspaceDraftSetupForCreatedWorkspace({
forkDraftSetup: input.forkDraftSetup,
workspaceDirectory: ensuredWorkspace.workspaceDirectory,
provider,
composerState,
});
return await submitWorkspaceDraft({
clearConsumedDraft,
serverId,
clearDraft,
draftKey: input.draftKey,
draftId: input.draftId,
draftContextScopeKey: input.draftContextScopeKey,
initialSetup,
workspaceId: ensuredWorkspace.id,
workspaceDirectory: ensuredWorkspace.workspaceDirectory,
text,
attachments,
provider,
composerState,
supportsForgeSearch: input.supportsForgeSearch,
resolveClient: input.resolveClient,
isStillOnCreateScreen: input.isStillOnCreateScreen,
const images = await encodeImages(wirePayload.images);
let resolveAgent!: (agent: AgentSnapshotPayload) => void;
let rejectAgent!: (error: unknown) => void;
const result = new Promise<AgentSnapshotPayload>((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<CreateWorkspaceRequestOptions["agent"]> = {
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<AgentSnapshotPayload> => {
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<SubmitOutcome> {
function submitWorkspaceDraft(input: SubmitDraftInput): SubmitOutcome {
const {
serverId,
clearDraft,
@@ -1082,33 +1147,6 @@ async function submitWorkspaceDraft(input: SubmitDraftInput): Promise<SubmitOutc
composerState,
initialSetup,
});
// Creation blocks on a slow daemon RPC. If the user moved on while it ran, the destination
// screen's draft tab will never mount to issue create_agent, so this path does it instead.
if (!input.isStillOnCreateScreen()) {
await createWorkspaceAgentInBackground({
clearConsumedDraft: input.clearConsumedDraft,
createAgent: () =>
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<SubmitOutc
...(submission.thinkingOptionId ? { thinkingOptionId: submission.thinkingOptionId } : {}),
...(submission.featureValues ? { featureValues: submission.featureValues } : {}),
allowEmptyText: true,
agentCreation: input.agentCreation,
});
clearDraft("sent");
navigateToWorkspace({
@@ -2002,42 +2041,15 @@ export function NewWorkspaceScreen({
setProjectPickerOpen(nextOpen);
}, []);
const buildCreateWorktreeInput = useCallback(
(input: {
cwd: string;
prompt: string;
attachments: AgentAttachment[];
checkoutRequest: PickerCheckoutRequest | undefined;
}): CreatePaseoWorktreeInput => {
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,
@@ -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<never>(() => {});
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();
});
@@ -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<AgentSnapshotPayload>;
retry: (input: CreateAgentRequestOptions) => Promise<AgentSnapshotPayload>;
};
serverId: string;
workspaceId: string;
draftId: string;
@@ -28,6 +35,7 @@ export interface PendingWorkspaceDraftSetup {
interface WorkspaceDraftSubmissionState {
pendingByDraftId: Record<string, PendingWorkspaceDraftSubmission>;
setupByDraftId: Record<string, PendingWorkspaceDraftSetup>;
creationByDraftId: Record<string, NonNullable<PendingWorkspaceDraftSubmission["agentCreation"]>>;
setPending: (submission: PendingWorkspaceDraftSubmission) => void;
setDraftSetup: (input: {
draftId: string;
@@ -61,8 +69,12 @@ export const useWorkspaceDraftSubmissionStore = create<WorkspaceDraftSubmissionS
(set, get) => ({
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<WorkspaceDraftSubmissionS
const normalizedDraftId = normalizeDraftId(draftId);
if (!normalizedDraftId) return;
set((state) => {
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) => {
+201
View File
@@ -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<string, unknown> }> = [];
const legacy: Array<{ kind: string; input: unknown }> = [];
let resolve!: (result: CreationResult) => void;
const result = new Promise<CreationResult>((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();
});
+270
View File
@@ -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<string, unknown>) => Promise<CreationResult>;
subscribe: (kind: Kind, key: string, enabled?: boolean) => Promise<CreationSnapshot | null>;
legacyAgent: (input: CreateAgentRequestOptions) => Promise<AgentSnapshotPayload>;
legacyWorkspace: (
input: CreateWorkspaceRequestOptions,
) => Promise<WorkspaceCreateResponse["payload"]>;
sendMessage: (
id: string,
text: string,
options: {
messageId: string;
images?: CreateAgentRequestOptions["images"];
attachments?: CreateAgentRequestOptions["attachments"];
},
) => Promise<unknown>;
}
interface Operation {
kind: Kind;
key: string;
fingerprint: string;
input: Record<string, unknown>;
observers: Set<Observer>;
snapshot?: CreationSnapshot;
result: Promise<CreationResult>;
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<string, Operation>();
constructor(private readonly deps: Dependencies) {}
createAgent(input: CreateAgentRequestOptions): Promise<CreationResult> {
const { onEvent, ...request } = input;
return this.start("agent", request, onEvent, () => this.legacyAgent(request));
}
createWorkspace(input: CreateWorkspaceRequestOptions): Promise<CreationResult> {
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<CreationResult> {
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<string, unknown>,
observer: Observer | undefined,
legacy: (operation: Operation) => Promise<CreationResult>,
): Promise<CreationResult> {
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<CreationResult>((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<void> {
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<void> {
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)}`;
}
+83
View File
@@ -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");
});
+98 -6
View File
@@ -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<Omit<AgentSessionConfig, "provider" | "cwd">>;
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<string, string>;
}
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<AgentSnapshotPayload> {
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<AgentSnapshotPayload> {
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<WorkspaceCreatePayload> {
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<WorkspaceCreatePayload> {
// 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,
+43 -22
View File
@@ -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<WorkspaceCreateRequest, "type" | "requestId"> & {
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");
}
+94 -1
View File
@@ -1681,7 +1681,7 @@ export type CreateAgentWorktreeTarget = z.infer<typeof CreateAgentWorktreeTarget
export const CreateAgentRequestMessageSchema = z.object({
type: z.literal("create_agent_request"),
// A creation key requires initialPrompt to be sent separately with a stable messageId.
// Legacy create_agent_request uses a separate initial-message receipt when keyed.
idempotencyKey: z.string().min(1).max(512).optional(),
config: AgentSessionConfigSchema,
env: z.record(z.string(), z.string()).optional(),
@@ -1702,6 +1702,33 @@ export const CreateAgentRequestMessageSchema = z.object({
requestId: z.string(),
});
/** Resource IDs are optional on creation; supplied IDs retain their identity on replay. */
export const AgentCreateRequestSchema = CreateAgentRequestMessageSchema.extend({
type: z.literal("agent.create.request"),
agentId: z.uuid().optional(),
attachments: z.array(AgentAttachmentSchema).optional(),
subscribe: z.boolean().optional(),
});
export const WorkspaceInitialAgentSchema = AgentCreateRequestSchema.omit({
type: true,
requestId: true,
idempotencyKey: true,
subscribe: true,
workspaceId: true,
worktree: true,
worktreeName: true,
git: true,
});
export const CreationSubscribeRequestSchema = z.object({
type: z.literal("creation.subscribe.request"),
requestId: z.string(),
kind: z.enum(["workspace", "agent"]),
idempotencyKey: z.string().min(1).max(512),
subscribe: z.boolean().optional(),
});
export const ListProviderModelsRequestMessageSchema = z.object({
type: z.literal("list_provider_models_request"),
provider: AgentProviderSchema,
@@ -2562,6 +2589,12 @@ export const ArchiveWorkspaceRequestSchema = z.object({
// between an existing local directory and a newly created paseo worktree.
export const WorkspaceCreateRequestSchema = z.object({
type: z.literal("workspace.create.request"),
workspaceId: z
.string()
.regex(/^wks_[a-f0-9]{16}$/)
.optional(),
agent: WorkspaceInitialAgentSchema.optional(),
subscribe: z.boolean().optional(),
requestId: z.string(),
idempotencyKey: z.string().min(1).max(512).optional(),
// Optional user-set title applied to the created workspace.
@@ -3255,6 +3288,8 @@ export const SessionInboundMessageSchema = z.discriminatedUnion("type", [
ProjectGithubCloneRequestSchema,
ArchiveWorkspaceRequestSchema,
WorkspaceCreateRequestSchema,
AgentCreateRequestSchema,
CreationSubscribeRequestSchema,
WorkspaceClearAttentionRequestSchema,
WorkspaceMarkUnreadRequestSchema,
FileExplorerRequestSchema,
@@ -3483,6 +3518,7 @@ export const ServerInfoStatusPayloadSchema = z
agentRequestReceipts: z.boolean().optional(),
// COMPAT(workspaceRequestReceipts): added in v0.8.0; remove gate after 2027-03-07.
workspaceRequestReceipts: z.boolean().optional(),
creationLifecycle: z.boolean().optional(),
// COMPAT(hubAgentRpc): added in v0.8.0; remove gate after 2027-03-05.
hubAgentRpc: z.boolean().optional(),
providersSnapshot: z.boolean().optional(),
@@ -4724,9 +4760,62 @@ export const ClearAgentAttentionResponseMessageSchema = z.object({
}),
});
export const CreationSnapshotSchema = z.object({
kind: z.enum(["workspace", "agent"]),
idempotencyKey: z.string(),
revision: z.number().int().nonnegative(),
phase: z.enum([
"accepted",
"workspace_ready",
"agent_ready",
"prompt_started",
"completed",
"failed",
]),
workspaceId: z.string().nullable(),
agentId: z.string().nullable(),
workspace: WorkspaceDescriptorPayloadSchema.optional(),
agent: AgentSnapshotPayloadSchema.optional(),
error: z.string().nullable(),
setupSkippedReason: z.string().optional(),
errorCode: z.string().optional(),
failedStage: z.enum(["workspace", "agent", "prompt"]).optional(),
outcomeUnknown: z.boolean().optional(),
});
export type CreationSnapshot = z.infer<typeof CreationSnapshotSchema>;
export type AgentCreateRequest = z.infer<typeof AgentCreateRequestSchema>;
export type WorkspaceInitialAgent = z.infer<typeof WorkspaceInitialAgentSchema>;
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,
@@ -55,6 +55,7 @@ export type EnsureWorkspaceForCreate = (
export interface CreateAgentFromSessionInput {
kind: "session";
onAgentReady?: (agent: ManagedAgent) => Promise<void>;
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;
@@ -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<InboundOperation, PermissionRequirement>;
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",
@@ -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<T>() {
let resolve!: (value: T) => void;
const promise = new Promise<T>((done) => {
resolve = done;
});
return { promise, resolve };
}
test("the daemon continues the combined intent after its initiating client closes", async () => {
const provider = deferred<void>();
const ready = deferred<CreationSnapshot>();
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);
@@ -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<void>((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);
});
@@ -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<boolean>;
provision?: (
workspaceId: string,
) => Promise<{ workspace: WorkspaceDescriptorPayload; setupSkippedReason?: string }>;
createAgent?: (
agentId: string,
workspace: WorkspaceDescriptorPayload | undefined,
onReady: (agent: AgentSnapshotPayload) => Promise<void>,
) => Promise<AgentSnapshotPayload>;
}
const RecordSchema = z.object({
fingerprint: z.string(),
snapshot: CreationSnapshotSchema,
inFlight: z.enum(["workspace", "agent", "prompt"]).nullable(),
});
type Record = z.infer<typeof RecordSchema>;
/** Owns creation across socket lifetimes. Resource owners perform work; this module commits its milestones. */
export class CreationService {
private admission: Promise<unknown> = Promise.resolve();
private readonly active = new Map<string, Promise<CreationSnapshot>>();
private readonly observers = new Map<string, Set<Observer>>();
constructor(
private readonly directory: string,
private readonly validateCompleted: (
snapshot: CreationSnapshot,
) => Promise<void> = async () => {},
) {}
async create(input: CreationInput, observer?: Observer): Promise<CreationSnapshot> {
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<CreationSnapshot> {
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<void> {
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<void> {
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<Observer>();
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<CreationSnapshot>,
): Promise<void> {
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<void> {
return writeJsonFileAtomic(join(this.directory, `${identity}.json`), record);
}
private async read(identity: string): Promise<Record | null> {
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<string | null> {
try {
return await readFile(file, "utf8");
} catch (error) {
if (error instanceof Error && "code" in error && error.code === "ENOENT") return null;
throw error;
}
}
+5 -1
View File
@@ -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
+243
View File
@@ -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<RequestReceipts, "createAgent" | "createWorkspace" | "sendMessage">;
creationService: Pick<CreationService, "create" | "subscribe">;
projectRegistry: ProjectRegistry;
workspaceRegistry: WorkspaceRegistry;
directorySync?: DirectorySyncService;
@@ -637,6 +642,16 @@ function resolveWorkspaceLabelService(
return service ?? null;
}
function hasInitialCreationPrompt(
request?: Pick<AgentCreateRequest, "initialPrompt" | "images" | "attachments">,
): 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<CreationService, "create" | "subscribe">;
private readonly defaultCreationSource = {};
private readonly creationSubscriptions = new Map<object, Map<string, () => 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<void> | 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<void> | 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<SessionInboundMessage, { type: "creation.subscribe.request" }>,
source?: object,
): Promise<void> {
const key = JSON.stringify([request.kind, request.idempotencyKey]);
const owner = source ?? this.defaultCreationSource;
const subscriptions = this.creationSubscriptions.get(owner) ?? new Map<string, () => 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<boolean> =>
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<void> {
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<SessionInboundMessage, { type: "workspace.create.request" }>,
source?: object,
): Promise<void> {
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<void> {
try {
let agent: AgentSnapshotPayload;
@@ -3882,6 +4117,7 @@ export class Session {
private async createSessionAgent(
msg: CreateAgentRequestMessage,
agentId?: string,
onAgentReady?: (agent: AgentSnapshotPayload) => Promise<void>,
): Promise<AgentSnapshotPayload> {
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();
@@ -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 }),
@@ -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,
@@ -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),
@@ -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<SessionInternals>(
new Session({
requestReceipts: createRequestReceiptsStub(),
creationService: createCreationServiceStub(),
clientId: "test-client",
permissions: OWNER_PERMISSIONS,
onMessage,
@@ -62,6 +62,7 @@ interface FakeAgentSessionOptions {
}
export interface TestAgentClientOptions {
beforeCreateSession?: () => Promise<void>;
closeSession?: () => Promise<void>;
onStartTurn?: (prompt: AgentPromptInput) => void;
supportsMcpServers?: boolean;
@@ -1212,6 +1213,7 @@ class FakeAgentClient implements AgentClient {
config: AgentSessionConfig,
_launchContext?: AgentLaunchContext,
): Promise<AgentSession> {
await this.options.beforeCreateSession?.();
return new FakeAgentSession({
providerName: this.provider,
config: { ...config },
@@ -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");
},
};
}
@@ -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<void> {
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,
@@ -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,