From 424a4adc86c0e3b36f6ed50ae490caed871c0620 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Mon, 14 Sep 2026 10:38:25 +0200 Subject: [PATCH] fix(creation): reuse connection-owned progress and recovery --- docs/architecture.md | 7 +- .../e2e/browser/creation-idempotency.spec.ts | 44 +++--- packages/app/e2e/support/creation-fixtures.ts | 49 +++++++ packages/app/e2e/support/helpers/creation.ts | 34 +++-- .../support/helpers/daemon-client-loader.ts | 7 + .../helpers/new-workspace-navigation.ts | 9 +- .../app/e2e/support/helpers/new-workspace.ts | 131 +++++++----------- packages/client/src/creation/index.test.ts | 37 ++++- packages/client/src/creation/index.ts | 88 ++++++------ packages/client/src/daemon-client.test.ts | 83 +++++++++++ packages/client/src/daemon-client.ts | 28 ++-- packages/protocol/src/messages.ts | 3 +- .../src/server/creation/creation.e2e.test.ts | 25 +++- packages/server/src/server/session.ts | 99 ++++++++----- 14 files changed, 444 insertions(+), 200 deletions(-) diff --git a/docs/architecture.md b/docs/architecture.md index 160b5cf35c..7febb9006a 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -340,8 +340,11 @@ sequence after the client process disappears. Explicitly requested IDs or receip 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. +Creation executes through the existing Session capabilities. Connection-owned delivery +controls observation only: detaching a socket or cleaning up its Session does not cancel +accepted creation. Updates require an explicit subscription and go only to that socket; +reconnect uses the shared subscription owner. Legacy consumers, including Hub, keep their +existing response contract. ## Agent lifecycle diff --git a/packages/app/e2e/browser/creation-idempotency.spec.ts b/packages/app/e2e/browser/creation-idempotency.spec.ts index d3fb5f47c9..39e73dbded 100644 --- a/packages/app/e2e/browser/creation-idempotency.spec.ts +++ b/packages/app/e2e/browser/creation-idempotency.spec.ts @@ -1,3 +1,4 @@ +import type { createCreationScenario } from "../support/helpers/creation"; import { expect } from "../support/fixtures"; import { test } from "../support/creation-fixtures"; @@ -89,30 +90,41 @@ test("separate drafts can intentionally create two agents in the same workspace" await creation.expectAgentCount(2); }); -test("new workspace navigation and optimistic prompt precede agent completion", async ({ +test("new workspace navigation and optimistic prompt precede agent initialization", async ({ creation, + startup, delayedCreation, }) => { await creation.openWorkspaceForm("worktree"); await creation.submitPrompt("Show this prompt while the agent is starting.", "Create"); - await delayedCreation.waitForDelayedCreatedStatus(); - await creation.expectWorkspaceReadyBeforeAgentCompletion(); + await creation.expectAgentStillStarting(); delayedCreation.expectSingleWorkspaceIntent(); delayedCreation.release(); + await startup.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); -}); +for (const scenario of [ + { suffix: "", prepare: async () => {} }, + { + suffix: " after remount", + prepare: async (creation: Awaited>) => + creation.evictAndReturnToDraft(), + }, +]) { + test(`retrying failed agent initialization preserves its workspace${scenario.suffix}`, async ({ + creation, + startup, + }) => { + await creation.openWorkspaceForm("local"); + await creation.submitPrompt("Retry this workspace and agent together.", "Create"); + await creation.expectAgentStillStarting(); + await startup.fail(); + await creation.expectStartupFailure(); + await creation.expectOneCreatedWorkspace(); + await scenario.prepare(creation); + await creation.submitPrompt("Retry this workspace and agent together."); + await creation.expectAgentCount(1); + }); +} diff --git a/packages/app/e2e/support/creation-fixtures.ts b/packages/app/e2e/support/creation-fixtures.ts index e409909055..67a9528b05 100644 --- a/packages/app/e2e/support/creation-fixtures.ts +++ b/packages/app/e2e/support/creation-fixtures.ts @@ -1,3 +1,7 @@ +import { writeFile, rm } from "node:fs/promises"; +import path from "node:path"; +import { copyPluginExample } from "./helpers/plugin-fixture"; +import { connectNewWorkspaceDaemonClient } from "./helpers/new-workspace"; import { test as base } from "./fixtures"; import { createCreationScenario, @@ -8,11 +12,56 @@ import { installDaemonWebSocketGate } from "./helpers/daemon-websocket-gate"; import { delayBrowserAgentCreatedStatus } from "./helpers/new-workspace"; export const test = base.extend<{ + startup: { release(): Promise; fail(): Promise }; creation: Awaited>; promptRetry: ReturnType; agentRetries: Awaited>; delayedCreation: Awaited>; }>({ + startup: async ({ page }, provide) => { + void page; + const client = await connectNewWorkspaceDaemonClient(); + const plugin = await copyPluginExample("lifecycle-logger"); + const gate = path.join(plugin.directory, "startup.txt"); + const previous = await client.getDaemonConfig(); + try { + await writeFile(gate, "hold"); + await writeFile( + path.join(plugin.directory, "index.server.ts"), + ` +import { readFile, rm } from "node:fs/promises"; +export default function contribute(server) { + return server.before("agent.session_open", async ({ request }) => { + if (request.reason !== "create") return request; + const gate = ${JSON.stringify(gate)}; + let command = await readFile(gate, "utf8").catch(() => "release"); + while (command === "hold") { + await new Promise(resolve => setTimeout(resolve, 20)); + command = await readFile(gate, "utf8").catch(() => "release"); + } + if (command === "fail") { + await rm(gate, { force: true }); + throw new Error("Creation startup failed for test"); + } + return request; + }); +} +`, + ); + await client.patchDaemonConfig({ pluginsEnabled: true }); + await client.installDirectoryPlugin(plugin.directory); + await provide({ + release: () => rm(gate, { force: true }), + fail: () => writeFile(gate, "fail"), + }); + } finally { + await rm(gate, { force: true }); + await client.removePlugin("lifecycle-logger"); + await client.patchDaemonConfig({ pluginsEnabled: previous.config.pluginsEnabled }); + await plugin.cleanup(); + await client.close(); + } + }, creation: async ({ page }, provide) => { const creation = await createCreationScenario(page); try { diff --git a/packages/app/e2e/support/helpers/creation.ts b/packages/app/e2e/support/helpers/creation.ts index 4231edf118..c3dfedca44 100644 --- a/packages/app/e2e/support/helpers/creation.ts +++ b/packages/app/e2e/support/helpers/creation.ts @@ -1,11 +1,14 @@ import { expect, type Page } from "@playwright/test"; -import type { SessionInboundMessage, SessionOutboundMessage } from "@getpaseo/protocol/messages"; import { daemonWsRoutePattern } from "./daemon-port"; import { gotoAppShell } from "./app"; import { gotoWorkspace } from "./launcher"; import { fillComposerDraft } from "./composer"; import { createAgentTabFromMenu } from "./workspace-tabs"; -import { openNewWorkspaceComposer, selectWorkspaceIsolation } from "./new-workspace"; +import { + openNewWorkspaceComposer, + selectWorkspaceIsolation, + loadSessionMessageReaders, +} from "./new-workspace"; import { seedWorkspace } from "./seed-client"; import { waitForSidebarHydration, @@ -28,12 +31,12 @@ export async function pressSubmitBeforeTheNextRender(page: Page, name: string): }); } -export function observeCreationRequests(page: Page) { +export async function observeCreationRequests(page: Page) { + const frames = await loadSessionMessageReaders(); const pending = new Set(); page.on("websocket", (socket) => { socket.on("framesent", ({ payload }) => { - const envelope = JSON.parse(payload.toString()) as { message?: SessionInboundMessage }; - const request = envelope.message; + const request = frames.client(payload); if ( request?.type === "workspace.create.request" || request?.type === "agent.create.request" @@ -42,8 +45,7 @@ export function observeCreationRequests(page: Page) { } }); socket.on("framereceived", ({ payload }) => { - const envelope = JSON.parse(payload.toString()) as { message?: SessionOutboundMessage }; - const response = envelope.message; + const response = frames.server(payload); if ( response?.type === "workspace.create.response" || response?.type === "agent.create.response" || @@ -64,29 +66,28 @@ export function observeCreationRequests(page: Page) { } export async function retryNextAgentCreation(page: Page) { + const frames = await loadSessionMessageReaders(); const retryIds = new Set(); const results: Array<{ status: string; agentId?: string }> = []; let repeated = false; await page.routeWebSocket(daemonWsRoutePattern(), (browser) => { const server = browser.connectToServer(); browser.onMessage((frame) => { - const envelope = JSON.parse(frame.toString()) as { message?: SessionInboundMessage }; - const request = envelope.message; + const request = frames.client(frame); 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}`; retryIds.add(requestId); // Keep the app's operation key and payload; only RPC correlation changes. - server.send(JSON.stringify({ ...envelope, message: { ...request, requestId } })); + server.send(JSON.stringify({ type: "session", message: { ...request, requestId } })); } return; } server.send(frame); }); server.onMessage((frame) => { - const envelope = JSON.parse(frame.toString()) as { message?: SessionOutboundMessage }; - const response = envelope.message; + const response = frames.server(frame); if ( response?.type === "agent.create.response" && retryIds.delete(response.payload.requestId) @@ -113,7 +114,7 @@ export async function retryNextAgentCreation(page: Page) { } export async function createCreationScenario(page: Page) { - const requests = observeCreationRequests(page); + const requests = await observeCreationRequests(page); const project = await seedWorkspace({ repoPrefix: "creation-idempotency-" }); let workspaceId = project.workspaceId; return { @@ -152,6 +153,13 @@ export async function createCreationScenario(page: Page) { .filter({ visible: true }), ).toBeVisible(); }, + async expectAgentStillStarting() { + await this.expectWorkspaceReadyBeforeAgentCompletion(); + expect((await project.client.fetchAgents()).entries).toHaveLength(0); + }, + async expectStartupFailure() { + await expect(page.getByText(/Creation startup failed for test/).first()).toBeVisible(); + }, async expectOneCreatedWorkspace() { await expect(page).toHaveURL(/\/workspace\//); await requests.settled(); diff --git a/packages/app/e2e/support/helpers/daemon-client-loader.ts b/packages/app/e2e/support/helpers/daemon-client-loader.ts index b36bbae32a..eee8cdaffb 100644 --- a/packages/app/e2e/support/helpers/daemon-client-loader.ts +++ b/packages/app/e2e/support/helpers/daemon-client-loader.ts @@ -64,3 +64,10 @@ function loadAppVersion(): string { } return packageJson.version; } + +export async function loadProtocolSchemas(): Promise { + const moduleUrl = pathToFileURL( + path.resolve(__dirname, "../../../../../packages/protocol/dist/messages.js"), + ).href; + return import(moduleUrl); +} diff --git a/packages/app/e2e/support/helpers/new-workspace-navigation.ts b/packages/app/e2e/support/helpers/new-workspace-navigation.ts index 6e6220c581..1b3106bdf7 100644 --- a/packages/app/e2e/support/helpers/new-workspace-navigation.ts +++ b/packages/app/e2e/support/helpers/new-workspace-navigation.ts @@ -134,15 +134,20 @@ export async function verifyDelayedWorkspaceCreation( ); if (!upload) throw new Error("The create request did not contain the submitted file"); expect(await readFile(upload.path)).toEqual(CONTEXT.buffer); + const agents = await client.fetchAgents(); + expect( + agents.entries.find((entry) => entry.agent.workspaceId === created.id)?.agent, + ).toMatchObject({ + cwd: created.workspaceDirectory, + }); expect(delay.agentRequests[0]).toMatchObject({ - workspaceId: created.id, initialPrompt: PROMPT, config: { provider: "mock", model: "ten-second-stream", modeId: "load-test", thinkingOptionId: "low", - cwd: created.workspaceDirectory, + cwd: repo.path, }, attachments: [ expect.objectContaining({ diff --git a/packages/app/e2e/support/helpers/new-workspace.ts b/packages/app/e2e/support/helpers/new-workspace.ts index 69d50c1d10..c6695a268d 100644 --- a/packages/app/e2e/support/helpers/new-workspace.ts +++ b/packages/app/e2e/support/helpers/new-workspace.ts @@ -1,8 +1,8 @@ import { expect, type BrowserContext, type Page } from "@playwright/test"; -import type { CreateAgentRequestMessage } from "@getpaseo/protocol/messages"; +import type { CreateAgentRequestMessage, SessionInboundMessage } from "@getpaseo/protocol/messages"; import type { DaemonClient as InternalDaemonClient } from "@getpaseo/client/internal/daemon-client"; import { decodeWorkspaceIdFromPathSegment } from "@/utils/host-routes"; -import { connectDaemonClient } from "./daemon-client-loader"; +import { connectDaemonClient, loadProtocolSchemas } from "./daemon-client-loader"; import { daemonWsRoutePattern } from "./daemon-port"; import { projectEquivalenceViewKey } from "./project-view-key"; import { expectWorkspaceHeader } from "./workspace-ui"; @@ -535,24 +535,19 @@ function parseWebSocketJson(message: WebSocketMessage): unknown { } } -function getSessionMessage(message: WebSocketMessage): Record | null { - const envelope = parseWebSocketJson(message); - if (!envelope || typeof envelope !== "object") { - return null; - } - const maybeEnvelope = envelope as { type?: unknown; message?: unknown }; - if (maybeEnvelope.type !== "session" || !maybeEnvelope.message) { - return null; - } - if (typeof maybeEnvelope.message !== "object") { - return null; - } - return maybeEnvelope.message as Record; -} - -function getStringField(input: Record, key: string): string | null { - const value = input[key]; - return typeof value === "string" ? value : null; +export async function loadSessionMessageReaders() { + // Use the same ESM module instance as the dynamically loaded daemon client. + const { WSInboundMessageSchema, WSOutboundMessageSchema } = await loadProtocolSchemas(); + return { + client(message: WebSocketMessage) { + const parsed = WSInboundMessageSchema.safeParse(parseWebSocketJson(message)); + return parsed.success && parsed.data.type === "session" ? parsed.data.message : null; + }, + server(message: WebSocketMessage) { + const parsed = WSOutboundMessageSchema.safeParse(parseWebSocketJson(message)); + return parsed.success && parsed.data.type === "session" ? parsed.data.message : null; + }, + }; } export interface AgentCreatedDelayControl { @@ -560,18 +555,17 @@ export interface AgentCreatedDelayControl { waitForCreateRequest(): Promise; waitForDelayedCreatedStatus(): Promise; expectSingleWorkspaceIntent(): void; - fail(error: string): void; } export async function delayBrowserAgentCreatedStatus( page: Page, ): Promise { + const frames = await loadSessionMessageReaders(); const daemonPortPattern = daemonWsRoutePattern(); const createRequestIds = new Set(); - const creationRequests: Record[] = []; + const creationRequests: SessionInboundMessage[] = []; const delayedForwards: Array<() => void> = []; let releaseRequested = false; - let responseError: string | null = null; let resolveCreateRequest: (() => void) | null = null; let resolveDelayedCreatedStatus: (() => void) | null = null; const createRequestSeen = new Promise((resolve) => { @@ -585,7 +579,7 @@ export async function delayBrowserAgentCreatedStatus( const server = ws.connectToServer(); ws.onMessage((message) => { - const sessionMessage = getSessionMessage(message); + const sessionMessage = frames.client(message); if (sessionMessage?.type === "send_agent_message_request") creationRequests.push(sessionMessage); if ( @@ -594,46 +588,29 @@ export async function delayBrowserAgentCreatedStatus( sessionMessage?.type === "workspace.create.request" ) { creationRequests.push(sessionMessage); - const requestId = getStringField(sessionMessage, "requestId"); - if (requestId) { - createRequestIds.add(requestId); - resolveCreateRequest?.(); - } + createRequestIds.add(sessionMessage.requestId); + resolveCreateRequest?.(); } server.send(message); }); server.onMessage((message) => { - const sessionMessage = getSessionMessage(message); - const payload = - sessionMessage && typeof sessionMessage.payload === "object" - ? (sessionMessage.payload as Record) - : null; - const requestId = payload ? getStringField(payload, "requestId") : null; - + const sessionMessage = frames.server(message); + const isCreated = + (sessionMessage?.type === "status" && sessionMessage.payload.status === "agent_created") || + sessionMessage?.type === "agent.create.response" || + sessionMessage?.type === "workspace.create.response"; if ( - (payload?.status === "agent_created" || - sessionMessage?.type === "agent.create.response" || - sessionMessage?.type === "workspace.create.response") && - requestId && - createRequestIds.has(requestId) + isCreated && + typeof sessionMessage.payload.requestId === "string" && + createRequestIds.has(sessionMessage.payload.requestId) ) { resolveDelayedCreatedStatus?.(); if (releaseRequested) { ws.send(message); return; } - 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)); - }); + delayedForwards.push(() => ws.send(message)); return; } @@ -648,12 +625,7 @@ 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({ @@ -666,21 +638,28 @@ export async function delayBrowserAgentCreatedStatus( }; } +type AgentCreationIntent = Pick< + CreateAgentRequestMessage, + "config" | "initialPrompt" | "attachments" +>; + export interface WorkspaceCreatedDelayControl { - agentRequests: readonly CreateAgentRequestMessage[]; + agentRequests: readonly AgentCreationIntent[]; release(): void; waitForCreateRequest(): Promise; } /** - * Holds `workspace.create.response` so a test can act as the user does while the daemon is still + * Holds workspace readiness events and responses so a test can act while the daemon is still * running `git worktree add` — most importantly, navigate somewhere else. */ export async function delayBrowserWorkspaceCreatedResponse( page: Page, ): Promise { + const frames = await loadSessionMessageReaders(); const daemonPortPattern = daemonWsRoutePattern(); - const agentRequests: CreateAgentRequestMessage[] = []; + const agentRequests: AgentCreationIntent[] = []; + const creationKeys = new Set(); const createRequestIds = new Set(); const delayedForwards: Array<() => void> = []; let releaseRequested = false; @@ -693,29 +672,25 @@ export async function delayBrowserWorkspaceCreatedResponse( const server = ws.connectToServer(); ws.onMessage((message) => { - const sessionMessage = getSessionMessage(message); - if (sessionMessage?.type === "create_agent_request") - agentRequests.push(sessionMessage as CreateAgentRequestMessage); + const sessionMessage = frames.client(message); + if (sessionMessage?.type === "create_agent_request") agentRequests.push(sessionMessage); if (sessionMessage?.type === "workspace.create.request") { - const requestId = getStringField(sessionMessage, "requestId"); - if (requestId) { - createRequestIds.add(requestId); - resolveCreateRequest?.(); - } + createRequestIds.add(sessionMessage.requestId); + if (sessionMessage.idempotencyKey) creationKeys.add(sessionMessage.idempotencyKey); + if (sessionMessage.agent) agentRequests.push(sessionMessage.agent); + resolveCreateRequest?.(); } server.send(message); }); server.onMessage((message) => { - const sessionMessage = getSessionMessage(message); - const payload = - sessionMessage?.type === "workspace.create.response" && - typeof sessionMessage.payload === "object" - ? (sessionMessage.payload as Record) - : null; - const requestId = payload ? getStringField(payload, "requestId") : null; - - if (requestId && createRequestIds.has(requestId) && !releaseRequested) { + const sessionMessage = frames.server(message); + const isCreation = + (sessionMessage?.type === "workspace.create.response" && + createRequestIds.has(sessionMessage.payload.requestId)) || + (sessionMessage?.type === "workspace.create.update" && + creationKeys.has(sessionMessage.payload.idempotencyKey)); + if (isCreation && !releaseRequested) { delayedForwards.push(() => ws.send(message)); return; } diff --git a/packages/client/src/creation/index.test.ts b/packages/client/src/creation/index.test.ts index 38d1548474..40828d4bfb 100644 --- a/packages/client/src/creation/index.test.ts +++ b/packages/client/src/creation/index.test.ts @@ -59,13 +59,12 @@ function fixture(modern: boolean) { }); const client = new CreationClient({ supports: () => modern, - connected: () => true, requestId: () => "generated-key", request: async (kind, input) => { requests.push({ kind, input }); return result; }, - subscribe: async () => null, + observe: () => () => {}, legacyWorkspace: async (input) => { legacy.push({ kind: "workspace", input }); return { requestId: "legacy-workspace", workspace, error: null, setupTerminalId: null }; @@ -199,3 +198,37 @@ test("keyed creation with an empty prompt works on legacy daemons without sendin ]); f.client.close(); }); + +test.each([ + ["/repo/project", "/repo/project/../outside"], + ["C:\\repo\\project", "C:\\repo\\project\\..\\outside"], +])("legacy creation rejects escaping %s before creating a workspace", async (source, cwd) => { + const f = fixture(false); + await expect( + f.client.createWorkspace({ + ...f.input, + source: { kind: "directory", path: source }, + agent: { ...f.input.agent, config: { provider: "codex", cwd } }, + }), + ).rejects.toThrow("inside the workspace source"); + expect(f.legacy).toEqual([]); + f.client.close(); +}); + +test.each([ + { source: { kind: "directory" as const, path: "/project" }, cwd: "/project/src" }, + { source: { kind: "directory" as const, path: "C:\\project" }, cwd: "C:\\project\\src" }, + { source: { kind: "worktree" as const, projectId: "project" }, cwd: "/project/src" }, +])("legacy creation preserves the agent subdirectory for $source", async ({ source, cwd }) => { + const f = fixture(false); + await f.client.createWorkspace({ + ...f.input, + source, + agent: { config: { provider: "codex", cwd } }, + }); + expect(f.legacy[1]).toMatchObject({ + kind: "agent", + input: { workspaceId: workspace.id, config: { cwd: "/project/worktree/src" } }, + }); + f.client.close(); +}); diff --git a/packages/client/src/creation/index.ts b/packages/client/src/creation/index.ts index 07d8247359..4ae64c34ff 100644 --- a/packages/client/src/creation/index.ts +++ b/packages/client/src/creation/index.ts @@ -19,10 +19,14 @@ export interface CreationResult { } interface Dependencies { supports: () => boolean; - connected: () => boolean; requestId: () => string; request: (kind: Kind, input: Record) => Promise; - subscribe: (kind: Kind, key: string, enabled?: boolean) => Promise; + observe: ( + kind: Kind, + key: string, + next: (snapshot: CreationSnapshot | null) => void, + error: (error: unknown) => void, + ) => () => void; legacyAgent: (input: CreateAgentRequestOptions) => Promise; legacyWorkspace: ( input: CreateWorkspaceRequestOptions, @@ -50,6 +54,7 @@ interface Operation { settled: boolean; modern: boolean; recovering: boolean; + stopObserving?: () => void; } /** Client-owned observation and compatibility. UI callbacks never advance creation. */ @@ -67,6 +72,10 @@ export class CreationClient { if (request.workspaceId || request.agent?.agentId) throw new Error("Update the host to use caller-selected creation IDs."); const { agent, ...workspaceInput } = request; + const sourceCwd = + request.source.kind === "directory" ? request.source.path : request.source.cwd; + const relativeCwd = + agent && sourceCwd ? relativeDirectory(agent.config!.cwd, sourceCwd) : undefined; const workspace = await this.deps.legacyWorkspace(workspaceInput); if (workspace.error || !workspace.workspace) return workspace; this.receive({ @@ -81,11 +90,9 @@ export class CreationClient { 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 cwd = `${workspace.workspace.workspaceDirectory!.replace(/[\\/]+$/, "")}${ + relativeCwd ?? relativeDirectory(agent.config!.cwd, workspace.workspace.projectRootPath) + }`; const created = await this.legacyAgent({ ...agent, config: { ...agent.config!, cwd }, @@ -180,16 +187,22 @@ export class CreationClient { if (result.creation) this.receive(result.creation); this.finish(operation, result); } catch (error) { - if (operation.settled || !this.deps.connected()) return; + if ( + operation.settled || + (error instanceof Error && "code" in error && error.code === "DAEMON_CONNECTION_LOST") + ) + 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)) + 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.snapshot?.revision !== snapshot.revision) { + 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, @@ -201,32 +214,22 @@ export class CreationClient { }); } } - private async recover(operation: Operation): Promise { - operation.recovering = true; - const snapshot = await this.deps.subscribe(operation.kind, operation.key); - if (!snapshot) { - await this.submit(operation); - return; - } - this.receive(snapshot); - if (snapshot.phase === "completed" || snapshot.phase === "failed") { - this.finish(operation, { - agent: snapshot.agent, - workspace: snapshot.workspace, - creation: snapshot, - setupSkippedReason: snapshot.setupSkippedReason, - error: snapshot.error, - errorCode: snapshot.errorCode, - }); - } - } - reconnect(): void { for (const operation of this.operations.values()) { - if (operation.settled || !operation.modern) continue; - void this.recover(operation).catch((error) => { - if (this.deps.connected()) this.fail(operation, error); - }); + if (operation.settled || !operation.modern || operation.recovering) continue; + operation.recovering = true; + operation.stopObserving = this.deps.observe( + operation.kind, + operation.key, + (snapshot) => { + if (!snapshot) void this.submit(operation); + else { + // Re-subscription is authoritative even if its revision matches the last live event. + this.receive(snapshot); + } + }, + (error) => this.fail(operation, error), + ); } } close(): void { @@ -237,8 +240,7 @@ export class CreationClient { 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.stopObserving?.(); operation.resolve(result); operation.observers.clear(); this.operations.delete(JSON.stringify([operation.kind, operation.key])); @@ -246,6 +248,7 @@ export class CreationClient { private fail(operation: Operation, error: unknown): void { if (operation.settled) return; operation.settled = true; + operation.stopObserving?.(); operation.reject(error instanceof Error ? error : new Error(String(error))); operation.observers.clear(); this.operations.delete(JSON.stringify([operation.kind, operation.key])); @@ -259,12 +262,15 @@ export class CreationClient { } } -function remapDirectory(cwd: string, source: string, target: string): string { - const normalize = (value: string) => value.replace(/\\/g, "/").replace(/\/$/, ""); +function relativeDirectory(cwd: string, source: string): string { + const normalize = (value: string) => value.replace(/\\/g, "/").replace(/\/+$/, ""); const normalized = normalize(cwd), root = normalize(source); - if (normalized === root) return target; + // Parent traversal must never survive mapping into the newly created checkout. + if ([normalized, root].some((path) => path.split("/").includes(".."))) + throw new Error("Agent directory must be inside the workspace source"); + if (normalized === root) return ""; if (!normalized.startsWith(`${root}/`)) throw new Error("Agent directory must be inside the workspace source"); - return `${normalize(target)}${normalized.slice(root.length)}`; + return normalized.slice(root.length); } diff --git a/packages/client/src/daemon-client.test.ts b/packages/client/src/daemon-client.test.ts index 5ed4264957..8e6eddf438 100644 --- a/packages/client/src/daemon-client.test.ts +++ b/packages/client/src/daemon-client.test.ts @@ -6405,3 +6405,86 @@ test("creation lifecycle acknowledgement reaches the observer before the final r ); await expect(creation).rejects.toThrow("provider unavailable"); }); + +test("creation reconnect observation uses connection-owned subscriptions and releases on completion", async () => { + const transport = createMockTransport(); + const client = new DaemonClient({ + url: "ws://test", + clientId: "creation-reconnect", + transportFactory: () => transport.transport, + reconnect: { enabled: false }, + }); + clients.push(client); + const features = { creationLifecycle: true, ownedSubscriptions: true }; + const connect = client.connect(); + transport.triggerOpen({ features }); + await connect; + const phases: string[] = []; + const creation = client.createWorkspace({ + idempotencyKey: "reconnect-one", + source: { kind: "directory", path: "/project" }, + onEvent: (snapshot) => phases.push(snapshot.phase), + }); + void creation.catch(() => undefined); + const snapshot = { + kind: "workspace", + idempotencyKey: "reconnect-one", + phase: "accepted", + revision: 0, + workspaceId: "wks_0123456789abcdef", + agentId: null, + error: null, + }; + for (const subscriptionId of ["first-connection", "second-connection"]) { + transport.triggerClose(); + const reconnected = client.connect(); + transport.triggerOpen({ features }); + await reconnected; + await expect.poll(() => transport.sent.length).toBe(1); + const request = parseSentFrame(transport.sent[0]); + expect(request).toMatchObject({ + type: "creation.subscribe.request", + idempotencyKey: "reconnect-one", + }); + transport.triggerMessage( + wrapSessionMessage({ + type: "creation.subscribe.response", + payload: { + requestId: request.requestId, + subscriptionId, + snapshot, + error: null, + }, + }), + ); + } + transport.triggerMessage( + wrapSessionMessage({ + type: "workspace.create.update", + payload: { + ...snapshot, + subscriptionId: "second-connection", + revision: 1, + phase: "failed", + error: "Provisioning failed", + }, + }), + ); + expect(await creation).toMatchObject({ error: "Provisioning failed" }); + await expect.poll(() => transport.sent.length).toBe(2); + const release = parseSentFrame(transport.sent.at(-1)); + expect(release).toMatchObject({ + type: "subscription.release.request", + subscriptionId: "second-connection", + }); + transport.triggerMessage( + wrapSessionMessage({ + type: "subscription.release.response", + payload: { + requestId: release.requestId, + subscriptionId: "second-connection", + }, + }), + ); + expect(phases).toEqual(["accepted", "failed"]); +}); diff --git a/packages/client/src/daemon-client.ts b/packages/client/src/daemon-client.ts index 78a386dba1..24d3d67657 100644 --- a/packages/client/src/daemon-client.ts +++ b/packages/client/src/daemon-client.ts @@ -2677,7 +2677,6 @@ export class DaemonClient { private readonly creations = new CreationClient({ supports: () => this.lastServerInfoMessage?.features?.creationLifecycle === true, - connected: () => this.connectionState.status === "connected", requestId: () => this.createRequestId(), request: (kind, input) => kind === "workspace" @@ -2693,13 +2692,23 @@ export class DaemonClient { 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", + observe: (kind, idempotencyKey, next, error) => { + const observation = this.observe("creation.subscribe.response", { + type: "creation.subscribe.request", + kind, + idempotencyKey, }); - if (result.error) throw new Error(result.error); - return result.snapshot; + observation.subscribe({ + snapshot: (result) => next(result.snapshot), + update: (message) => { + if (message.type === "workspace.create.update" || message.type === "agent.create.update") + next(message.payload); + }, + error, + }); + return () => { + void observation.release().catch(error); + }; }, legacyAgent: (input) => this.createLegacyAgent(input), legacyWorkspace: (input) => this.createLegacyWorkspace(input, input.requestId), @@ -6396,8 +6405,9 @@ export class DaemonClient { if ( consumerMessage.type === "workspace.create.update" || consumerMessage.type === "agent.create.update" - ) - this.creations.receive(consumerMessage.payload); + ) { + if (!consumerMessage.payload.subscriptionId) this.creations.receive(consumerMessage.payload); + } this.resolveWaiters(consumerMessage); this.owned.receive(consumerMessage); } diff --git a/packages/protocol/src/messages.ts b/packages/protocol/src/messages.ts index d1e2963985..a1198edcd4 100644 --- a/packages/protocol/src/messages.ts +++ b/packages/protocol/src/messages.ts @@ -4788,7 +4788,7 @@ export type WorkspaceInitialAgent = z.infer; export const WorkspaceCreationUpdateSchema = z.object({ type: z.literal("workspace.create.update"), - payload: CreationSnapshotSchema, + payload: CreationSnapshotSchema.extend({ subscriptionId: z.string().optional() }), }); export const AgentCreationUpdateSchema = WorkspaceCreationUpdateSchema.extend({ type: z.literal("agent.create.update"), @@ -4797,6 +4797,7 @@ export const CreationSubscribeResponseSchema = z.object({ type: z.literal("creation.subscribe.response"), payload: z.object({ requestId: z.string(), + subscriptionId: z.string().optional(), snapshot: CreationSnapshotSchema.nullable(), error: z.string().nullable(), }), diff --git a/packages/server/src/server/creation/creation.e2e.test.ts b/packages/server/src/server/creation/creation.e2e.test.ts index 13cb5b1109..0d6e55547c 100644 --- a/packages/server/src/server/creation/creation.e2e.test.ts +++ b/packages/server/src/server/creation/creation.e2e.test.ts @@ -2,7 +2,8 @@ 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 { expect, test, vi } from "vitest"; +import pino from "pino"; 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"; @@ -16,13 +17,25 @@ function deferred() { return { promise, resolve }; } -test("the daemon continues the combined intent after its initiating client closes", async () => { +test("creation progresses before agent readiness and continues after the disconnected Session is cleaned up", async () => { const provider = deferred(); const ready = deferred(); + const disconnected = deferred(); + const cleaned = deferred(); const directory = await mkdtemp(join(tmpdir(), "creation-wire-")); let agents = 0; let prompts = 0; const daemon = await createTestPaseoDaemon({ + logger: pino( + { level: "trace" }, + { + write(line) { + const { msg } = JSON.parse(line); + if (msg === "Client disconnected; waiting for reconnect") disconnected.resolve(); + if (msg === "agent.session.lifecycle.cleanup") cleaned.resolve(); + }, + }, + ), agentClients: createTestAgentClients({ beforeCreateSession: async () => { agents++; @@ -41,7 +54,7 @@ test("the daemon continues the combined intent after its initiating client close const observer = new DaemonClient({ url: `ws://127.0.0.1:${daemon.port}/ws`, appVersion: "0.7.2", - clientId: "creation-subscriber", + clientId: "creation-independent-observer", }); const observerMessages: SessionOutboundMessage[] = []; const unsubscribe = observer.subscribeRawMessages((message) => observerMessages.push(message)); @@ -75,7 +88,12 @@ test("the daemon continues the combined intent after its initiating client close expect( observerMessages.filter((message) => message.type === "workspace.create.update"), ).toHaveLength(0); + vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] }); await client.close(); + await disconnected.promise; + await vi.advanceTimersByTimeAsync(90_000); + await cleaned.promise; + vi.useRealTimers(); await expect(first).rejects.toThrow("closed"); // A disconnected form cannot prevent the provider or initial prompt from starting. provider.resolve(); @@ -94,6 +112,7 @@ test("the daemon continues the combined intent after its initiating client close expect(conflict.error).toBe("workspace_id_conflict"); expect(agents).toBe(1); } finally { + vi.useRealTimers(); provider.resolve(); unsubscribe(); await client.close(); diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index c81456a927..6a41036e65 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -802,8 +802,6 @@ export class Session { >; private readonly createAgentLifecycleDispatch: CreateAgentLifecycleDispatch; private readonly creationService: Pick; - private readonly defaultCreationSource = {}; - private readonly creationSubscriptions = new Map void>>(); constructor(options: SessionOptions) { const { @@ -1261,8 +1259,6 @@ 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 { @@ -3862,39 +3858,61 @@ export class Session { /** * Handle create agent request */ - private creationObserver(snapshot: CreationSnapshot, source?: object): void { - if (this.isCleanedUp) return; - const message: SessionOutboundMessage = { + private creationUpdate(snapshot: CreationSnapshot): SessionOutboundMessage { + return { type: snapshot.kind === "workspace" ? "workspace.create.update" : "agent.create.update", payload: snapshot, }; - this.emitForSource(message, source); } private async handleCreationSubscription( request: Extract, source?: object, ): Promise { - const key = JSON.stringify([request.kind, request.idempotencyKey]); - const owner = source ?? this.defaultCreationSource; - const subscriptions = this.creationSubscriptions.get(owner) ?? new Map void>(); - subscriptions.get(key)?.(); - subscriptions.delete(key); - this.creationSubscriptions.set(owner, subscriptions); - const observation = await this.creationService.subscribe( - request.kind, - request.idempotencyKey, - (snapshot) => this.creationObserver(snapshot, source), - ); - if (request.subscribe === false || this.isCleanedUp) observation.unsubscribe(); - else subscriptions.set(key, observation.unsubscribe); - this.emitForSource( - { - type: "creation.subscribe.response", - payload: { requestId: request.requestId, snapshot: observation.snapshot, error: null }, - }, - source, - ); + let unsubscribe = () => {}; + const owner = + request.subscribe === false + ? undefined + : this.delivery.begin( + "creation", + undefined, + () => unsubscribe(), + JSON.stringify([request.kind, request.idempotencyKey]), + ); + try { + const observation = await this.creationService.subscribe( + request.kind, + request.idempotencyKey, + (snapshot) => owner?.emit(this.creationUpdate(snapshot)), + ); + unsubscribe = observation.unsubscribe; + if (!owner || owner.signal.aborted) unsubscribe(); + this.emitForSource( + { + type: "creation.subscribe.response", + payload: { + requestId: request.requestId, + subscriptionId: owner?.responseId, + snapshot: observation.snapshot, + error: null, + }, + }, + source, + ); + } catch (error) { + await owner?.release(); + this.emitForSource( + { + type: "creation.subscribe.response", + payload: { + requestId: request.requestId, + snapshot: null, + error: error instanceof Error ? error.message : String(error), + }, + }, + source, + ); + } } private creationResourceExists = async ( @@ -3907,6 +3925,13 @@ export class Session { private async handleAgentCreation(request: AgentCreateRequest, source?: object): Promise { const { requestId, subscribe, type: _type, idempotencyKey, ...intent } = request; + const progress = subscribe + ? this.delivery.operation( + (message) => + message.type === "workspace.create.update" || message.type === "agent.create.update", + () => {}, + ) + : undefined; try { const creation = await this.creationService.create( { @@ -3921,7 +3946,7 @@ export class Session { createAgent: (id, _workspace, onReady) => this.createSessionAgent({ ...request, type: "create_agent_request" }, id, onReady), }, - subscribe ? (snapshot) => this.creationObserver(snapshot, source) : undefined, + progress ? (snapshot) => progress.emit(this.creationUpdate(snapshot)) : undefined, ); this.emitForSource( { @@ -3942,6 +3967,8 @@ export class Session { }, source, ); + } finally { + await progress?.release(); } } @@ -3950,6 +3977,13 @@ export class Session { source?: object, ): Promise { const { requestId, subscribe, type: _type, idempotencyKey, ...intent } = request; + const progress = subscribe + ? this.delivery.operation( + (message) => + message.type === "workspace.create.update" || message.type === "agent.create.update", + () => {}, + ) + : undefined; try { if ( request.agent && @@ -4019,7 +4053,7 @@ export class Session { } : undefined, }, - subscribe ? (snapshot) => this.creationObserver(snapshot, source) : undefined, + progress ? (snapshot) => progress.emit(this.creationUpdate(snapshot)) : undefined, ); this.emitForSource( { @@ -4054,6 +4088,8 @@ export class Session { }, source, ); + } finally { + await progress?.release(); } } @@ -8490,9 +8526,6 @@ 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();