From e170a66713247dd3b2faea647626eb1947e4c3e0 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Sat, 1 Aug 2026 05:52:19 +0200 Subject: [PATCH] fix(server): preserve queued agent transitions Project each queued live snapshot before awaiting stored metadata so later lifecycle state cannot overwrite an earlier transition. --- packages/server/src/server/session.ts | 19 +--- .../src/server/session.workspaces.test.ts | 98 +++++++++++++------ 2 files changed, 71 insertions(+), 46 deletions(-) diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index b338dd9a7c..3f569f3d70 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -91,11 +91,7 @@ import { setAgentModeCommand, updateAgentCommand, } from "./agent/lifecycle-command.js"; -import { - buildStoredAgentPayload, - resolveStoredAgentPayloadUpdatedAt, - toAgentPayload, -} from "./agent/agent-projections.js"; +import { buildStoredAgentPayload, toAgentPayload } from "./agent/agent-projections.js"; import { appendTimelineItemIfAgentKnown, emitLiveTimelineItemIfAgentKnown, @@ -1672,18 +1668,9 @@ export class Session { } private async buildAgentPayload(agent: ManagedAgent): Promise { + const payload = toAgentPayload(agent); const storedRecord = await this.agentStorage.get(agent.id); - const title = storedRecord?.title ?? null; - const currentAgent = this.agentManager.getAgent(agent.id) ?? agent; - const payload = toAgentPayload(currentAgent, { title }); - const storedUpdatedAt = storedRecord ? resolveStoredAgentPayloadUpdatedAt(storedRecord) : null; - if (storedUpdatedAt) { - const liveUpdatedAt = Date.parse(payload.updatedAt); - const persistedUpdatedAt = Date.parse(storedUpdatedAt); - if (Number.isNaN(liveUpdatedAt) || persistedUpdatedAt > liveUpdatedAt) { - payload.updatedAt = storedUpdatedAt; - } - } + payload.title = storedRecord?.title ?? null; payload.archivedAt = storedRecord?.archivedAt ?? null; return payload; } diff --git a/packages/server/src/server/session.workspaces.test.ts b/packages/server/src/server/session.workspaces.test.ts index 6b2bf10b28..dbffc83104 100644 --- a/packages/server/src/server/session.workspaces.test.ts +++ b/packages/server/src/server/session.workspaces.test.ts @@ -311,6 +311,7 @@ function makeManagedAgent(input: { workspaceId?: string; lifecycle: AgentSnapshotPayload["status"]; updatedAt: string; + activeTurn?: { turnId: string; startedAt: string }; }) { const now = new Date(input.updatedAt); const snapshot = makeAgent({ @@ -346,6 +347,8 @@ function makeManagedAgent(input: { unsubscribeSession: null, session: null, activeForegroundTurnId: input.lifecycle === "running" ? "turn-1" : null, + activeTurnId: input.activeTurn?.turnId ?? null, + activeTurnStartedAt: input.activeTurn ? new Date(input.activeTurn.startedAt) : null, }; } @@ -735,47 +738,64 @@ function createSessionForWorkspaceTests( return session; } -test("agent updates pair a stored revision with current live state", async () => { - const idle = makeManagedAgent({ - id: "agent-coherent", - cwd: REPO_CWD, - workspaceId: "ws-repo-running", - lifecycle: "idle", - updatedAt: "2026-07-31T10:00:00.000Z", - }) as unknown as ManagedAgent; +test("agent updates preserve queued live transitions across stored metadata reads", async () => { const running = makeManagedAgent({ id: "agent-coherent", cwd: REPO_CWD, workspaceId: "ws-repo-running", lifecycle: "running", updatedAt: "2026-07-31T10:00:01.000Z", + activeTurn: { + turnId: "turn-running", + startedAt: "2026-07-31T10:00:00.500Z", + }, }) as unknown as ManagedAgent; - const stored = makeStoredAgent({ + const idle = makeManagedAgent({ id: "agent-coherent", cwd: REPO_CWD, + workspaceId: "ws-repo-running", + 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", + }; const storageReadStarted = deferred(); - const storageRead = deferred(); - const emittedAgentUpdate = - deferred["payload"]>(); - let current = idle; + const firstStorageRead = deferred(); + const emittedAgentUpdates: Extract< + SessionOutboundMessage, + { type: "agent_update" } + >["payload"][] = []; + const twoUpdatesEmitted = deferred(); + let storageReadCount = 0; let forwardAgentEvent: ((event: AgentManagerEvent) => void) | null = null; const session = createSessionForWorkspaceTests({ onMessage: (message) => { - if (message.type === "agent_update") emittedAgentUpdate.resolve(message.payload); + if (message.type !== "agent_update") return; + emittedAgentUpdates.push(message.payload); + if (emittedAgentUpdates.length === 2) twoUpdatesEmitted.resolve(); }, agentManager: { subscribe: (listener: (event: AgentManagerEvent) => void) => { forwardAgentEvent = listener; return () => {}; }, - getAgent: () => current, + getAgent: () => idle, }, agentStorage: { get: async () => { - storageReadStarted.resolve(); - return storageRead.promise; + storageReadCount += 1; + if (storageReadCount === 1) { + storageReadStarted.resolve(); + return firstStorageRead.promise; + } + return stored; }, }, }); @@ -788,22 +808,40 @@ test("agent updates pair a stored revision with current live state", async () => createdAt: "2026-07-31T10:00:00.000Z", updatedAt: "2026-07-31T10:00:00.000Z", }); - activateAgentUpdatesSubscription(session, "sub-coherent"); + activateAgentUpdatesSubscription(session, "sub-coherent", { includeArchived: true }); if (!forwardAgentEvent) throw new Error("Agent event listener was not installed"); - forwardAgentEvent({ type: "agent_state", agent: idle }); + forwardAgentEvent({ type: "agent_state", agent: running }); await storageReadStarted.promise; - current = running; - storageRead.resolve(stored); + forwardAgentEvent({ type: "agent_state", agent: idle }); + firstStorageRead.resolve(stored); + await twoUpdatesEmitted.promise; - await expect(emittedAgentUpdate.promise).resolves.toMatchObject({ - kind: "upsert", - agent: { - id: "agent-coherent", - status: "running", - updatedAt: "2026-07-31T10:00:02.000Z", - }, - }); + expect(emittedAgentUpdates).toEqual([ + expect.objectContaining({ + kind: "upsert", + agent: expect.objectContaining({ + status: "running", + activeTurn: { + turnId: "turn-running", + startedAt: "2026-07-31T10:00:00.500Z", + }, + updatedAt: "2026-07-31T10:00:01.000Z", + title: "Stored title", + archivedAt: "2026-07-31T10:00:04.000Z", + }), + }), + expect.objectContaining({ + kind: "upsert", + agent: expect.objectContaining({ + status: "idle", + activeTurn: null, + updatedAt: "2026-07-31T10:00:02.000Z", + title: "Stored title", + archivedAt: "2026-07-31T10:00:04.000Z", + }), + }), + ]); }); test("client heartbeat clears attention for the focused terminal", async () => {