puter/paseo
1
0
Fork 0
mirror of https://github.com/getpaseo/paseo.git synced 2026-11-06 08:24:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(creation): reuse connection-owned progress and recovery

Browse Source
This commit is contained in:
Mohamed Boudra
2026-09-14 10:38:49 +02:00
parent 81aea18e72
commit 424a4adc86
14 changed files with 444 additions and 200 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/architecture.md
+5 -2
View File
@@ -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
packages/app/e2e/browser/creation-idempotency.spec.ts
+28 -16
View File
@@ -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<ReturnType<typeof createCreationScenario>>) =>
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);
});
}
packages/app/e2e/support/creation-fixtures.ts
+49
View File
@@ -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<void>; fail(): Promise<void> };
creation: Awaited<ReturnType<typeof createCreationScenario>>;
promptRetry: ReturnType<typeof createPromptRetryScenario>;
agentRetries: Awaited<ReturnType<typeof retryNextAgentCreation>>;
delayedCreation: Awaited<ReturnType<typeof delayBrowserAgentCreatedStatus>>;
}>({
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 {
packages/app/e2e/support/helpers/creation.ts
+21 -13
View File
@@ -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<string>();
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<string>();
const results: Array<{ status: string; agentId?: string }> = [];
let repeated = false;
await page.routeWebSocket(daemonWsRoutePattern(), (browser) => {
const server = browser.connectToServer();
browser.onMessage((frame) => {
const envelope = JSON.parse(frame.toString()) as { message?: SessionInboundMessage };
const request = envelope.message;
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();
packages/app/e2e/support/helpers/daemon-client-loader.ts
+7
View File
@@ -64,3 +64,10 @@ function loadAppVersion(): string {
}
return packageJson.version;
}
export async function loadProtocolSchemas(): Promise<typeof import("@getpaseo/protocol/messages")> {
const moduleUrl = pathToFileURL(
path.resolve(__dirname, "../../../../../packages/protocol/dist/messages.js"),
).href;
return import(moduleUrl);
}
packages/app/e2e/support/helpers/new-workspace-navigation.ts
+7 -2
View File
@@ -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({
packages/app/e2e/support/helpers/new-workspace.ts
+53 -78
View File
@@ -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<string, unknown> | 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<string, unknown>;
}
function getStringField(input: Record<string, unknown>, 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<void>;
waitForDelayedCreatedStatus(): Promise<void>;
expectSingleWorkspaceIntent(): void;
fail(error: string): void;
}
export async function delayBrowserAgentCreatedStatus(
page: Page,
): Promise<AgentCreatedDelayControl> {
const frames = await loadSessionMessageReaders();
const daemonPortPattern = daemonWsRoutePattern();
const createRequestIds = new Set<string>();
const creationRequests: Record<string, unknown>[] = [];
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<void>((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<string, unknown>)
: 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<void>;
}
/**
* 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<WorkspaceCreatedDelayControl> {
const frames = await loadSessionMessageReaders();
const daemonPortPattern = daemonWsRoutePattern();
const agentRequests: CreateAgentRequestMessage[] = [];
const agentRequests: AgentCreationIntent[] = [];
const creationKeys = new Set<string>();
const createRequestIds = new Set<string>();
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<string, unknown>)
: 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;
}
packages/client/src/creation/index.test.ts
+35 -2
View File
@@ -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();
});
packages/client/src/creation/index.ts
+47 -41
View File
@@ -19,10 +19,14 @@ export interface CreationResult {
}
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>;
observe: (
kind: Kind,
key: string,
next: (snapshot: CreationSnapshot | null) => void,
error: (error: unknown) => void,
) => () => void;
legacyAgent: (input: CreateAgentRequestOptions) => Promise<AgentSnapshotPayload>;
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<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);
});
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);
}
packages/client/src/daemon-client.test.ts
+83
View File
@@ -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"]);
});
packages/client/src/daemon-client.ts
+19 -9
View File
@@ -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);
}
packages/protocol/src/messages.ts
+2 -1
View File
@@ -4788,7 +4788,7 @@ export type WorkspaceInitialAgent = z.infer<typeof WorkspaceInitialAgentSchema>;
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(),
}),
packages/server/src/server/creation/creation.e2e.test.ts
+22 -3
View File
@@ -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<T>() {
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<void>();
const ready = deferred<CreationSnapshot>();
const disconnected = deferred<void>();
const cleaned = deferred<void>();
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();
packages/server/src/server/session.ts
+66 -33
View File
@@ -802,8 +802,6 @@ export class Session {
>;
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 {
@@ -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<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,
);
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<void> {
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<void> {
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();