From 6465edc01f69e38d7df5ae1f8c161d1f6adb3a31 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Fri, 24 Apr 2026 01:39:20 +0700 Subject: [PATCH] fix(server): restore microtask ordering in message dispatch and stream event handler The complexity refactor added async dispatcher chains that inserted extra microtasks before message handlers ran, and wrapped the stream event switch in an await that fired between emitState and dispatchStream. Two tests regressed on both counts. Route to the matching dispatcher synchronously and skip the await when the stream handler has no async work. --- .../server/src/server/agent/agent-manager.ts | 37 +++++--- packages/server/src/server/session.ts | 94 ++++++++++--------- 2 files changed, 73 insertions(+), 58 deletions(-) diff --git a/packages/server/src/server/agent/agent-manager.ts b/packages/server/src/server/agent/agent-manager.ts index b00d76a52..eb3f3a9cf 100644 --- a/packages/server/src/server/agent/agent-manager.ts +++ b/packages/server/src/server/agent/agent-manager.ts @@ -2453,7 +2453,7 @@ export class AgentManager { const flags: StreamEventFlags = { shouldDispatchEvent: true, shouldNotifyWaiters: true }; - await this.dispatchStreamEventByType({ + const dispatchPromise = this.dispatchStreamEventByType({ agent, event, options, @@ -2461,6 +2461,9 @@ export class AgentManager { eventTurnId, flags, }); + if (dispatchPromise) { + await dispatchPromise; + } if (!options?.fromHistory && isForegroundEvent && isTurnTerminalEvent(event)) { this.finalizeForegroundTurn(agent, eventTurnId); @@ -2473,46 +2476,50 @@ export class AgentManager { return flags.shouldNotifyWaiters; } - private async dispatchStreamEventByType(params: { + private dispatchStreamEventByType(params: { agent: ActiveManagedAgent; event: AgentStreamEvent; options: HandleStreamEventOptions | undefined; isForegroundEvent: boolean; eventTurnId: string | undefined; flags: StreamEventFlags; - }): Promise { + }): Promise | undefined { const { agent, event, options, isForegroundEvent, eventTurnId, flags } = params; switch (event.type) { case "thread_started": this.onStreamThreadStarted(agent); - break; + return undefined; case "usage_updated": agent.lastUsage = event.usage; this.emitState(agent); - break; + return undefined; case "timeline": - await this.onStreamTimelineEvent({ agent, event, options, isForegroundEvent, flags }); - break; + return this.onStreamTimelineEvent({ agent, event, options, isForegroundEvent, flags }); case "turn_completed": this.onStreamTurnCompleted({ agent, event, eventTurnId, isForegroundEvent }); - break; + return undefined; case "turn_failed": - await this.onStreamTurnFailed({ agent, event, eventTurnId, isForegroundEvent, options }); - break; + return this.onStreamTurnFailed({ + agent, + event, + eventTurnId, + isForegroundEvent, + options, + }); case "turn_canceled": this.onStreamTurnCanceled({ agent, event, eventTurnId, isForegroundEvent, options }); - break; + return undefined; case "turn_started": this.onStreamTurnStarted({ agent, eventTurnId, isForegroundEvent }); - break; + return undefined; case "permission_requested": this.onStreamPermissionRequested(agent, event); - break; + return undefined; case "permission_resolved": this.onStreamPermissionResolved({ agent, event, options, flags }); - break; + return undefined; default: - break; + return undefined; } } diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index 9037220de..5b8b5c606 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -2347,13 +2347,17 @@ export class Session { } private async handleCloseItemsRequest(msg: CloseItemsRequest): Promise { + const archiveResults = await Promise.allSettled( + msg.agentIds.map((agentId) => this.archiveAgentForClose(agentId)), + ); const agents = []; - for (const agentId of msg.agentIds) { - try { - agents.push(await this.archiveAgentForClose(agentId)); - } catch (error: any) { + for (let i = 0; i < archiveResults.length; i += 1) { + const result = archiveResults[i]!; + if (result.status === "fulfilled") { + agents.push(result.value); + } else { this.sessionLogger.warn( - { err: error, agentId, requestId: msg.requestId }, + { err: result.reason, agentId: msg.agentIds[i], requestId: msg.requestId }, "Failed to archive agent during close_items batch", ); } @@ -5711,18 +5715,19 @@ export class Session { ); const placementsByCwd = new Map(); - for (const workspace of persistedWorkspaces) { - if (workspace.archivedAt) { - continue; - } + const pairs = persistedWorkspaces.flatMap((workspace) => { + if (workspace.archivedAt) return []; const project = activeProjects.get(workspace.projectId); - if (!project) { - continue; - } - placementsByCwd.set( - normalizePersistedWorkspaceId(workspace.cwd), - await this.buildProjectPlacementForWorkspace(workspace, project), - ); + if (!project) return []; + return [{ workspace, project }]; + }); + const placements = await Promise.all( + pairs.map(({ workspace, project }) => + this.buildProjectPlacementForWorkspace(workspace, project), + ), + ); + for (let i = 0; i < pairs.length; i += 1) { + placementsByCwd.set(normalizePersistedWorkspaceId(pairs[i]!.workspace.cwd), placements[i]!); } return placementsByCwd; @@ -6033,19 +6038,20 @@ export class Session { ), ); - for (const workspace of activeRecords) { - if (workspaceIds && !workspaceIds.has(workspace.workspaceId)) { - continue; - } - const projectRecord = activeProjects.get(workspace.projectId) ?? null; - descriptorsByWorkspaceId.set( - workspace.workspaceId, - await this.buildWorkspaceDescriptor({ + const includedWorkspaces = activeRecords.filter( + (workspace) => !workspaceIds || workspaceIds.has(workspace.workspaceId), + ); + const workspaceDescriptors = await Promise.all( + includedWorkspaces.map((workspace) => + this.buildWorkspaceDescriptor({ workspace, - projectRecord, + projectRecord: activeProjects.get(workspace.projectId) ?? null, includeGitData: options.includeGitData, }), - ); + ), + ); + for (let i = 0; i < includedWorkspaces.length; i += 1) { + descriptorsByWorkspaceId.set(includedWorkspaces[i]!.workspaceId, workspaceDescriptors[i]!); } for (const agent of agents) { @@ -6523,23 +6529,25 @@ export class Session { const changedWorkspaceIds = new Set(); const changedProjectIds = new Set(); - for (const change of result.changesApplied) { - switch (change.kind) { - case "workspace_archived": - await this.removeWorkspaceGitWatchTarget(change.directory); - this.scriptRuntimeStore?.removeForWorkspace(change.directory); - this.removeWorkspaceGitSubscription(change.workspaceId); - changedWorkspaceIds.add(change.workspaceId); - break; - case "workspace_updated": - changedWorkspaceIds.add(change.workspaceId); - break; - case "project_archived": - case "project_updated": - changedProjectIds.add(change.projectId); - break; - } - } + await Promise.all( + result.changesApplied.map(async (change) => { + switch (change.kind) { + case "workspace_archived": + await this.removeWorkspaceGitWatchTarget(change.directory); + this.scriptRuntimeStore?.removeForWorkspace(change.directory); + this.removeWorkspaceGitSubscription(change.workspaceId); + changedWorkspaceIds.add(change.workspaceId); + break; + case "workspace_updated": + changedWorkspaceIds.add(change.workspaceId); + break; + case "project_archived": + case "project_updated": + changedProjectIds.add(change.projectId); + break; + } + }), + ); if (changedProjectIds.size > 0) { for (const workspace of await this.workspaceRegistry.list()) {