From b6db940501eb76fa4a48530218f554792fe0f45e Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Sat, 1 Aug 2026 06:01:31 +0200 Subject: [PATCH] fix(server): snapshot queued agent updates Capture each live agent payload before it enters the per-agent queue so shared nested state cannot change while an earlier storage read is pending. --- packages/server/src/server/session.ts | 11 ++++--- .../src/server/session.workspaces.test.ts | 25 ++++++++------- .../agent-updates-service.test.ts | 31 +++++++++++++++---- .../agent-updates/agent-updates-service.ts | 11 ++++--- 4 files changed, 51 insertions(+), 27 deletions(-) diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index 3f569f3d70..129d7037de 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -901,7 +901,7 @@ export class Session { }); this.agentUpdates = createAgentUpdatesService({ emit: (message) => this.emit(message), - buildAgentPayload: (agent) => this.buildAgentPayload(agent), + enrichAgentPayload: (payload) => this.enrichAgentPayload(payload), buildStoredAgentPayload: (record) => this.buildStoredAgentPayload(record), isProviderVisibleToClient: (provider) => this.isProviderVisibleToClient(provider), buildProjectPlacementForWorkspaceId: (workspaceId) => @@ -1667,14 +1667,17 @@ export class Session { }; } - private async buildAgentPayload(agent: ManagedAgent): Promise { - const payload = toAgentPayload(agent); - const storedRecord = await this.agentStorage.get(agent.id); + private async enrichAgentPayload(payload: AgentSnapshotPayload): Promise { + const storedRecord = await this.agentStorage.get(payload.id); payload.title = storedRecord?.title ?? null; payload.archivedAt = storedRecord?.archivedAt ?? null; return payload; } + private buildAgentPayload(agent: ManagedAgent): Promise { + return this.enrichAgentPayload(toAgentPayload(agent)); + } + private buildStoredAgentPayload( record: StoredAgentRecord, registeredProviderIds = new Set(this.providerSnapshotManager.listRegisteredProviderIds()), diff --git a/packages/server/src/server/session.workspaces.test.ts b/packages/server/src/server/session.workspaces.test.ts index dbffc83104..de33dfe2a8 100644 --- a/packages/server/src/server/session.workspaces.test.ts +++ b/packages/server/src/server/session.workspaces.test.ts @@ -757,15 +757,14 @@ test("agent updates preserve queued live transitions across stored metadata read lifecycle: "idle", updatedAt: "2026-07-31T10:00:02.000Z", }) as unknown as ManagedAgent; - const stored = { - ...makeStoredAgent({ - id: "agent-coherent", - cwd: REPO_CWD, - updatedAt: "2026-07-31T10:00:03.000Z", - }), - title: "Stored title", - archivedAt: "2026-07-31T10:00:04.000Z", - }; + running.config.model = "running-model"; + idle.config = running.config; + const stored = makeStoredAgent({ + id: "agent-coherent", + cwd: REPO_CWD, + updatedAt: "2026-07-31T10:00:03.000Z", + }); + stored.title = "Stored title"; const storageReadStarted = deferred(); const firstStorageRead = deferred(); const emittedAgentUpdates: Extract< @@ -808,12 +807,14 @@ test("agent updates preserve queued live transitions across stored metadata read createdAt: "2026-07-31T10:00:00.000Z", updatedAt: "2026-07-31T10:00:00.000Z", }); - activateAgentUpdatesSubscription(session, "sub-coherent", { includeArchived: true }); + activateAgentUpdatesSubscription(session, "sub-coherent"); if (!forwardAgentEvent) throw new Error("Agent event listener was not installed"); forwardAgentEvent({ type: "agent_state", agent: running }); await storageReadStarted.promise; + idle.config.model = "idle-model"; forwardAgentEvent({ type: "agent_state", agent: idle }); + idle.config.model = "mutated-after-queue"; firstStorageRead.resolve(stored); await twoUpdatesEmitted.promise; @@ -827,8 +828,8 @@ test("agent updates preserve queued live transitions across stored metadata read startedAt: "2026-07-31T10:00:00.500Z", }, updatedAt: "2026-07-31T10:00:01.000Z", + model: "running-model", title: "Stored title", - archivedAt: "2026-07-31T10:00:04.000Z", }), }), expect.objectContaining({ @@ -837,8 +838,8 @@ test("agent updates preserve queued live transitions across stored metadata read status: "idle", activeTurn: null, updatedAt: "2026-07-31T10:00:02.000Z", + model: "idle-model", title: "Stored title", - archivedAt: "2026-07-31T10:00:04.000Z", }), }), ]); diff --git a/packages/server/src/server/session/agent-updates/agent-updates-service.test.ts b/packages/server/src/server/session/agent-updates/agent-updates-service.test.ts index e3ae732cb9..01aba3f1f5 100644 --- a/packages/server/src/server/session/agent-updates/agent-updates-service.test.ts +++ b/packages/server/src/server/session/agent-updates/agent-updates-service.test.ts @@ -102,7 +102,7 @@ function buildHarness() { const service = createAgentUpdatesService({ emit: (message) => emitted.push(message), - buildAgentPayload: async (agent) => { + enrichAgentPayload: async (payload) => { const queuedPayload = queuedPayloadBuilds.shift(); if (queuedPayload) { return queuedPayload; @@ -110,11 +110,11 @@ function buildHarness() { if (buildAgentPayloadError) { throw buildAgentPayloadError; } - const payload = payloadById.get(agent.id); - if (!payload) { - throw new Error(`no payload registered for ${agent.id}`); + const registeredPayload = payloadById.get(payload.id); + if (!registeredPayload) { + throw new Error(`no payload registered for ${payload.id}`); } - return payload; + return registeredPayload; }, buildStoredAgentPayload: (record) => { const payload = payloadById.get(record.id); @@ -167,7 +167,26 @@ function buildHarness() { ); }, managed(id: string): ManagedAgent { - return { id } as unknown as ManagedAgent; + const payload = payloadById.get(id) ?? makeAgentPayload({ id }); + return { + ...payload, + lifecycle: payload.status, + config: { + provider: payload.provider, + cwd: payload.cwd, + }, + createdAt: new Date(payload.createdAt), + updatedAt: new Date(payload.updatedAt), + lastUserMessageAt: null, + pendingPermissions: new Map(), + attention: { + requiresAttention: payload.requiresAttention ?? false, + attentionReason: null, + attentionTimestamp: new Date(payload.updatedAt), + }, + activeTurnId: null, + activeTurnStartedAt: null, + } as unknown as ManagedAgent; }, stored(id: string): StoredAgentRecord { return { id } as unknown as StoredAgentRecord; diff --git a/packages/server/src/server/session/agent-updates/agent-updates-service.ts b/packages/server/src/server/session/agent-updates/agent-updates-service.ts index 45d256c0aa..ac4f3c8d7a 100644 --- a/packages/server/src/server/session/agent-updates/agent-updates-service.ts +++ b/packages/server/src/server/session/agent-updates/agent-updates-service.ts @@ -7,7 +7,7 @@ import type { } from "../../messages.js"; import type { ManagedAgent } from "../../agent/agent-manager.js"; import type { StoredAgentRecord } from "../../agent/agent-storage.js"; -import { resolveEffectiveThinkingOptionId } from "../../agent/agent-projections.js"; +import { resolveEffectiveThinkingOptionId, toAgentPayload } from "../../agent/agent-projections.js"; type AgentUpdatePayload = Extract["payload"]; type AgentUpdatesFilter = NonNullable< @@ -52,7 +52,7 @@ export interface AgentUpdatesService { export interface AgentUpdatesServiceDeps { emit(message: SessionOutboundMessage): void; - buildAgentPayload(agent: ManagedAgent): Promise; + enrichAgentPayload(payload: AgentSnapshotPayload): Promise; buildStoredAgentPayload(record: StoredAgentRecord): AgentSnapshotPayload; isProviderVisibleToClient(provider: string): boolean; buildProjectPlacementForWorkspaceId(workspaceId: string): Promise; @@ -258,10 +258,10 @@ export function createAgentUpdatesService(deps: AgentUpdatesServiceDeps): AgentU return payload; } - async function emitLiveAgentUpdate(agent: ManagedAgent): Promise { + async function emitLiveAgentUpdate(payload: AgentSnapshotPayload): Promise { try { const sub = subscription; - const payload = await deps.buildAgentPayload(agent); + payload = await deps.enrichAgentPayload(payload); if (sub) { const project = payload.workspaceId ? await deps.buildProjectPlacementForWorkspaceId(payload.workspaceId) @@ -319,7 +319,8 @@ export function createAgentUpdatesService(deps: AgentUpdatesServiceDeps): AgentU } function forwardLiveAgent(agent: ManagedAgent): Promise { - return enqueueAgentUpdate(agent.id, () => emitLiveAgentUpdate(agent)); + const payload = toAgentPayload(agent); + return enqueueAgentUpdate(payload.id, () => emitLiveAgentUpdate(payload)); } function removeAgent(agentId: string): Promise {