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.
This commit is contained in:
Mohamed Boudra
2026-08-01 06:01:31 +02:00
parent e170a66713
commit b6db940501
4 changed files with 51 additions and 27 deletions
+7 -4
View File
@@ -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<AgentSnapshotPayload> {
const payload = toAgentPayload(agent);
const storedRecord = await this.agentStorage.get(agent.id);
private async enrichAgentPayload(payload: AgentSnapshotPayload): Promise<AgentSnapshotPayload> {
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<AgentSnapshotPayload> {
return this.enrichAgentPayload(toAgentPayload(agent));
}
private buildStoredAgentPayload(
record: StoredAgentRecord,
registeredProviderIds = new Set(this.providerSnapshotManager.listRegisteredProviderIds()),
@@ -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<void>();
const firstStorageRead = deferred<StoredAgentRecord | null>();
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",
}),
}),
]);
@@ -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;
@@ -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<SessionOutboundMessage, { type: "agent_update" }>["payload"];
type AgentUpdatesFilter = NonNullable<
@@ -52,7 +52,7 @@ export interface AgentUpdatesService {
export interface AgentUpdatesServiceDeps {
emit(message: SessionOutboundMessage): void;
buildAgentPayload(agent: ManagedAgent): Promise<AgentSnapshotPayload>;
enrichAgentPayload(payload: AgentSnapshotPayload): Promise<AgentSnapshotPayload>;
buildStoredAgentPayload(record: StoredAgentRecord): AgentSnapshotPayload;
isProviderVisibleToClient(provider: string): boolean;
buildProjectPlacementForWorkspaceId(workspaceId: string): Promise<ProjectPlacementPayload | null>;
@@ -258,10 +258,10 @@ export function createAgentUpdatesService(deps: AgentUpdatesServiceDeps): AgentU
return payload;
}
async function emitLiveAgentUpdate(agent: ManagedAgent): Promise<void> {
async function emitLiveAgentUpdate(payload: AgentSnapshotPayload): Promise<void> {
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<void> {
return enqueueAgentUpdate(agent.id, () => emitLiveAgentUpdate(agent));
const payload = toAgentPayload(agent);
return enqueueAgentUpdate(payload.id, () => emitLiveAgentUpdate(payload));
}
function removeAgent(agentId: string): Promise<void> {