From 2b4d14672cdaa074b76f33fd88d5108472e5a8e7 Mon Sep 17 00:00:00 2001 From: "paseo-bot[bot]" <266920839+paseo-bot[bot]@users.noreply.github.com> Date: Wed, 12 Aug 2026 08:49:00 +0000 Subject: [PATCH] fix(server): deduplicate concurrent agent creation Concurrent create requests for one client message each ran the creation lifecycle. Share the in-flight operation and reply to every requester with the resulting agent so retries cannot start duplicate provider sessions. --- packages/server/src/server/session.ts | 123 +++++++++++++----- .../src/server/session.workspaces.test.ts | 48 +++++-- 2 files changed, 124 insertions(+), 47 deletions(-) diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index 1fb8228a8f..c35a4e5edb 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -366,6 +366,10 @@ type FetchAgentsResponsePageInfo = FetchAgentsResponsePayload["pageInfo"]; type AgentUpdatesFilter = FetchAgentsRequestFilter; type CreateAgentRequestMessage = Extract; +interface CreatedAgentRequest { + agent: AgentSnapshotPayload; +} + interface ResolvedSessionCreateAgentIntent { config: AgentSessionConfig; intent: CreateAgentIntent; @@ -676,6 +680,10 @@ export class Session { private readonly hubExecutionController: HubExecutionController | null; private readonly workspaceScripts: WorkspaceScriptsService; private readonly createAgentLifecycleDispatch: CreateAgentLifecycleDispatch; + private readonly inFlightAgentCreationsByClientMessageId = new Map< + string, + Promise + >(); constructor(options: SessionOptions) { const { @@ -3102,10 +3110,86 @@ export class Session { * Handle create agent request */ private async handleCreateAgentRequest(msg: CreateAgentRequestMessage): Promise { + const clientMessageId = msg.clientMessageId?.trim(); + const existingCreation = clientMessageId + ? this.inFlightAgentCreationsByClientMessageId.get(clientMessageId) + : undefined; + if (existingCreation) { + try { + const created = await existingCreation; + this.emitCreatedAgentResponse(msg.requestId, created.agent); + } catch (error) { + this.emitFailedAgentCreateResponse(msg.requestId, error); + } + return; + } + + const creation = this.createAgentForRequest(msg); + if (clientMessageId) { + this.inFlightAgentCreationsByClientMessageId.set(clientMessageId, creation); + } + try { + const created = await creation; + this.emitCreatedAgentResponse(msg.requestId, created.agent); + } catch (error) { + const wireError = toWorktreeWireError(error); + this.sessionLogger.error({ err: error }, "Failed to create agent"); + this.emitFailedAgentCreateResponse(msg.requestId, wireError); + this.emit({ + type: "activity_log", + payload: { + id: uuidv4(), + timestamp: new Date(), + type: "error", + content: `Failed to create agent: ${wireError.message}`, + }, + }); + } finally { + if ( + clientMessageId && + this.inFlightAgentCreationsByClientMessageId.get(clientMessageId) === creation + ) { + this.inFlightAgentCreationsByClientMessageId.delete(clientMessageId); + } + } + } + + private emitCreatedAgentResponse( + requestId: string | undefined, + agent: AgentSnapshotPayload, + ): void { + if (!requestId) return; + this.emit({ + type: "status", + payload: { + status: "agent_created", + agentId: agent.id, + requestId, + agent, + }, + }); + } + + private emitFailedAgentCreateResponse(requestId: string | undefined, error: unknown): void { + if (!requestId) return; + const wireError = toWorktreeWireError(error); + this.emit({ + type: "status", + payload: { + status: "agent_create_failed", + requestId, + error: wireError.message, + errorCode: wireError.code, + }, + }); + } + + private async createAgentForRequest( + msg: CreateAgentRequestMessage, + ): Promise { const { config, worktreeName, - requestId, initialPrompt, clientMessageId, outputSchema, @@ -3205,50 +3289,19 @@ export class Session { agentId: snapshot.id, createdWorktree, }); - if (requestId) { - const agentPayload = await this.buildAgentPayload(liveSnapshot); - this.emit({ - type: "status", - payload: { - status: "agent_created", - agentId: liveSnapshot.id, - requestId, - agent: agentPayload, - }, - }); - } + const agentPayload = await this.buildAgentPayload(liveSnapshot); this.sessionLogger.info( { agentId: snapshot.id, provider: snapshot.provider }, `Created agent ${snapshot.id} (${snapshot.provider})`, ); + return { agent: agentPayload }; } catch (error) { await this.createAgentLifecycleDispatch.cleanupCreatedWorktreeAfterFailedAgentCreate({ createdWorktree: createdWorktreeForCleanup, createdAgentId, }); - const wireError = toWorktreeWireError(error); - this.sessionLogger.error({ err: error }, "Failed to create agent"); - if (requestId) { - this.emit({ - type: "status", - payload: { - status: "agent_create_failed", - requestId, - error: wireError.message, - errorCode: wireError.code, - }, - }); - } - this.emit({ - type: "activity_log", - payload: { - id: uuidv4(), - timestamp: new Date(), - type: "error", - content: `Failed to create agent: ${wireError.message}`, - }, - }); + throw error; } } diff --git a/packages/server/src/server/session.workspaces.test.ts b/packages/server/src/server/session.workspaces.test.ts index 572680f9aa..6fbd706435 100644 --- a/packages/server/src/server/session.workspaces.test.ts +++ b/packages/server/src/server/session.workspaces.test.ts @@ -872,7 +872,7 @@ test("client heartbeat clears attention for the focused terminal", async () => { }); }); -test("create_agent_request keeps requested child cwd when grouped under an existing parent workspace", async () => { +test("create_agent_request deduplicates concurrent draft submission retries", async () => { const workdir = mkdtempSync(path.join(tmpdir(), "paseo-create-agent-cwd-")); try { const parent = path.join(workdir, "parent"); @@ -981,14 +981,26 @@ test("create_agent_request keeps requested child cwd when grouped under an exist terminalManager: null, }), ); - await session.handleMessage({ - type: "create_agent_request", - requestId: "req-create-child", - config: { provider: "codex", cwd: child }, - attachments: [], - }); + await Promise.all([ + session.handleMessage({ + type: "create_agent_request", + requestId: "req-create-child-1", + clientMessageId: "draft-submission-1", + config: { provider: "codex", cwd: child }, + attachments: [], + }), + session.handleMessage({ + type: "create_agent_request", + requestId: "req-create-child-2", + clientMessageId: "draft-submission-1", + config: { provider: "codex", cwd: child }, + attachments: [], + }), + ]); - const [createdAgent] = agentManager.listAgents(); + const createdAgents = agentManager.listAgents(); + expect(createdAgents).toHaveLength(1); + const [createdAgent] = createdAgents; expect(createdAgent?.cwd).toBe(child); const createdWorkspace = await workspaceRegistry.get(createdAgent!.workspaceId!); expect(createdWorkspace).not.toBeNull(); @@ -1007,10 +1019,22 @@ test("create_agent_request keeps requested child cwd when grouped under an exist projectKey: createdWorkspace!.projectId, checkout: { cwd: child }, }); - expect(findByType(emitted, "status")?.payload).toMatchObject({ - status: "agent_created", - agent: { cwd: child }, - }); + expect( + emitted.filter((message) => message.type === "status").map((message) => message.payload), + ).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + status: "agent_created", + requestId: "req-create-child-1", + agent: expect.objectContaining({ cwd: child }), + }), + expect.objectContaining({ + status: "agent_created", + requestId: "req-create-child-2", + agent: expect.objectContaining({ cwd: child }), + }), + ]), + ); } finally { rmSync(workdir, { recursive: true, force: true }); }