diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index 42c676b50b..55b46aa9f8 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -682,6 +682,9 @@ interface ClientActivity { appVisibilityChangedAt: Date; } +// The agent refreshes in flight on this daemon, by agent id; see handleRefreshAgentRequest. +const pendingAgentRefreshes = new Map>(); + export class Session { private readonly connectedHostSession: ConnectedHostSession | null; private routed = false; @@ -4758,38 +4761,23 @@ export class Session { this.sessionLogger.info({ agentId }, `Refreshing agent ${agentId} from persistence`); try { - await this.restoreOwningWorkspaceForLegacyAgentRefresh(agentId); - await unarchiveAgentState(this.agentStorage, this.agentManager, agentId); - let snapshot: ManagedAgent; - const existing = this.agentManager.getAgent(agentId); - if (existing) { - await this.interruptAgentIfRunning(agentId); - snapshot = await this.agentManager.reloadAgentSession(agentId, undefined, { - rehydrateFromDisk: true, - }); - } else { - const record = await this.agentStorage.get(agentId); - if (!record) { - throw new Error(`Agent not found: ${agentId}`); - } - const registeredProviderIds = this.providerSnapshotManager.listRegisteredProviderIds(); - if (!isStoredAgentProviderAvailable(record, registeredProviderIds)) { - throw new Error(`Agent ${agentId} references unavailable provider '${record.provider}'`); - } - if (!toAgentPersistenceHandle(registeredProviderIds, record.persistence)) { - throw new Error(`Agent ${agentId} cannot be refreshed because it lacks persistence`); - } - // Share the loader's per-agent in-flight operation with timeline fetches. - // Unarchiving publishes the record before provider resume finishes, so - // the agent pane can otherwise race this request and resume it twice. - snapshot = await ensureAgentLoaded(agentId, { - agentManager: this.agentManager, - agentStorage: this.agentStorage, - broadcastTimeline: true, - logger: this.sessionLogger, - }); + // A refresh takes the agent out of the manager while it resumes the provider session, so a + // second refresh arriving meanwhile, from any connection, found no agent and failed with + // "Unknown agent". A client that gives up waiting and retries makes exactly that second + // request. It waits for the refresh in flight and answers with its outcome instead. + let refresh = pendingAgentRefreshes.get(agentId); + if (!refresh) { + const started = this.refreshAgentFromPersistence(agentId); + refresh = started; + pendingAgentRefreshes.set(agentId, started); + void started + .catch(() => undefined) + .finally(() => { + if (pendingAgentRefreshes.get(agentId) === started) + pendingAgentRefreshes.delete(agentId); + }); } - await this.agentManager.hydrateTimelineFromProvider(agentId, { broadcast: true }); + const snapshot = await refresh; await this.agentUpdates.forwardLiveAgent(snapshot); const timelineSize = this.agentManager.getTimeline(agentId).length; if (requestId) { @@ -4829,6 +4817,42 @@ export class Session { } } + private async refreshAgentFromPersistence(agentId: string): Promise { + await this.restoreOwningWorkspaceForLegacyAgentRefresh(agentId); + await unarchiveAgentState(this.agentStorage, this.agentManager, agentId); + let snapshot: ManagedAgent; + const existing = this.agentManager.getAgent(agentId); + if (existing) { + await this.interruptAgentIfRunning(agentId); + snapshot = await this.agentManager.reloadAgentSession(agentId, undefined, { + rehydrateFromDisk: true, + }); + } else { + const record = await this.agentStorage.get(agentId); + if (!record) { + throw new Error(`Agent not found: ${agentId}`); + } + const registeredProviderIds = this.providerSnapshotManager.listRegisteredProviderIds(); + if (!isStoredAgentProviderAvailable(record, registeredProviderIds)) { + throw new Error(`Agent ${agentId} references unavailable provider '${record.provider}'`); + } + if (!toAgentPersistenceHandle(registeredProviderIds, record.persistence)) { + throw new Error(`Agent ${agentId} cannot be refreshed because it lacks persistence`); + } + // Share the loader's per-agent in-flight operation with timeline fetches. + // Unarchiving publishes the record before provider resume finishes, so + // the agent pane can otherwise race this request and resume it twice. + snapshot = await ensureAgentLoaded(agentId, { + agentManager: this.agentManager, + agentStorage: this.agentStorage, + broadcastTimeline: true, + logger: this.sessionLogger, + }); + } + await this.agentManager.hydrateTimelineFromProvider(agentId, { broadcast: true }); + return snapshot; + } + private async handleCancelAgentRequest(agentId: string, requestId?: string): Promise { this.sessionLogger.info({ agentId }, `Cancel request received for agent ${agentId}`); diff --git a/packages/server/src/server/session.workspaces.test.ts b/packages/server/src/server/session.workspaces.test.ts index 9e73c9cad4..f875dcba66 100644 --- a/packages/server/src/server/session.workspaces.test.ts +++ b/packages/server/src/server/session.workspaces.test.ts @@ -5426,6 +5426,77 @@ test("refresh_agent_request leaves workspace archival independent when its direc expect(findByType(emitted, "rpc_error")).toBeUndefined(); }); +test("refresh_agent_request arriving while a refresh of the same agent resumes waits for it", async () => { + // A client gives up on a slow refresh after its request timeout and asks again. The refresh in + // flight has taken the agent out of the manager, so a second one used to fail with "Unknown + // agent", and every retry after it did the same while the first was still resuming. + const emitted: SessionOutboundMessage[] = []; + const session = createSessionForWorkspaceTests({ + onMessage: (message) => { + if (isSessionOutboundMessage(message)) emitted.push(message); + }, + }); + const agentId = "agent-refreshed-twice"; + const cwd = path.resolve("/tmp/paseo-refresh-twice"); + session.filesystem.isDirectory = async () => true; + session.agentStorage.get = async (id: string) => + id === agentId + ? makeStoredAgent({ id: agentId, cwd, updatedAt: "2026-03-10T00:00:00.000Z" }) + : null; + session.agentStorage.upsert = async () => {}; + const managed = makeManagedAgent({ + id: agentId, + cwd, + lifecycle: "idle", + updatedAt: "2026-03-10T00:00:00.000Z", + }); + let loaded: ManagedAgent | null = managed; + let releaseResume!: () => void; + const resumed = new Promise((resolve) => { + releaseResume = resolve; + }); + let reloads = 0; + session.agentManager.getAgent = () => loaded; + session.interruptAgentIfRunning = async () => undefined; + session.agentManager.reloadAgentSession = async () => { + reloads += 1; + loaded = null; + await resumed; + loaded = managed; + return managed; + }; + session.agentManager.hydrateTimelineFromProvider = async () => { + if (!loaded) throw new Error(`Unknown agent '${agentId}'`); + }; + session.agentManager.getTimeline = () => []; + session.agentUpdates.forwardLiveAgent = async () => undefined; + + const first = session.handleMessage({ + type: "refresh_agent_request", + agentId, + requestId: "req-first", + }); + await waitForImmediate(); + const retry = session.handleMessage({ + type: "refresh_agent_request", + agentId, + requestId: "req-retry", + }); + await waitForImmediate(); + releaseResume(); + await Promise.all([first, retry]); + + expect(reloads).toBe(1); + expect(findByType(emitted, "rpc_error")).toBeUndefined(); + const refreshed = emitted.filter( + (m) => m.type === "status" && (m.payload as { status?: string }).status === "agent_refreshed", + ); + expect(refreshed.map((m) => (m.payload as { requestId?: string }).requestId).sort()).toEqual([ + "req-first", + "req-retry", + ]); +}); + test("refresh_agent_request leaves workspace archival independent when its directory is missing", async () => { const emitted: SessionOutboundMessage[] = []; const session = createSessionForWorkspaceTests({