From 1774ce3c4c0fc0915e9ab48b6a40d91b6c3a10d7 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Fri, 17 Jul 2026 13:44:50 +0200 Subject: [PATCH] fix superseded directory sync races --- packages/app/src/contexts/session-context.tsx | 9 ++- packages/app/src/runtime/host-runtime.test.ts | 60 ++++++++++++++++++- packages/app/src/runtime/host-runtime.ts | 7 ++- .../src/utils/agent-directory-sync.test.ts | 11 +++- .../app/src/utils/agent-directory-sync.ts | 6 ++ .../buffered-directory-transaction.test.ts | 23 +++++++ .../utils/buffered-directory-transaction.ts | 12 ++++ 7 files changed, 123 insertions(+), 5 deletions(-) create mode 100644 packages/app/src/utils/buffered-directory-transaction.test.ts create mode 100644 packages/app/src/utils/buffered-directory-transaction.ts diff --git a/packages/app/src/contexts/session-context.tsx b/packages/app/src/contexts/session-context.tsx index 645b2ccd2..8edb1f711 100644 --- a/packages/app/src/contexts/session-context.tsx +++ b/packages/app/src/contexts/session-context.tsx @@ -78,6 +78,8 @@ import { } from "@/workspace/legacy-daemon-workspaces"; import { useProviderSubagentStore } from "@/subagents/provider-store"; import { revalidateSessionAfterResume } from "@/contexts/session-resume-revalidation"; +import { inheritBufferedDirectoryDeltas } from "@/utils/buffered-directory-transaction"; +import { shouldApplyTimelineAgentSnapshot } from "@/utils/agent-directory-sync"; // Re-export types from session-store and draft-store for backward compatibility export type { DraftInput } from "@/stores/draft-store"; @@ -592,7 +594,10 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider id: Symbol("workspace hydration"), client, workspaces: new Map(), - deltas: [], + deltas: inheritBufferedDirectoryDeltas({ + client, + previous: workspaceHydrationRef.current, + }), }; workspaceHydrationRef.current = transaction; let snapshot: WorkspaceHydrationSnapshot | null; @@ -978,7 +983,7 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider return next; }); - if (payload.agent) { + if (payload.agent && shouldApplyTimelineAgentSnapshot(serverId, agentId)) { const normalized = normalizeAgentSnapshot(payload.agent, serverId); applyAuthoritativeAgentSnapshot( applyLegacyDaemonWorkspaceOwnership({ diff --git a/packages/app/src/runtime/host-runtime.test.ts b/packages/app/src/runtime/host-runtime.test.ts index 7a19bc67e..b01bd5fda 100644 --- a/packages/app/src/runtime/host-runtime.test.ts +++ b/packages/app/src/runtime/host-runtime.test.ts @@ -229,16 +229,22 @@ function makeFetchAgentsPayload(input: { class Deferred { readonly promise: Promise; private resolvePromise!: (value: T) => void; + private rejectPromise!: (error: Error) => void; constructor() { - this.promise = new Promise((resolve) => { + this.promise = new Promise((resolve, reject) => { this.resolvePromise = resolve; + this.rejectPromise = reject; }); } resolve(value: T): void { this.resolvePromise(value); } + + reject(error: Error): void { + this.rejectPromise(error); + } } async function waitForDirectoryReady(store: HostRuntimeStore, serverId: string): Promise { @@ -1967,6 +1973,58 @@ describe("HostRuntimeStore", () => { useSessionStore.getState().clearSession(host.serverId); }); + it("replays inherited deltas when a superseding refresh fails", async () => { + const host = makeHost({ serverId: "srv_overlap_failure" }); + const fakeClient = new FakeDaemonClient(); + fakeClient.setConnectionState({ status: "connected" }); + fakeClient.fetchAgentsResponses.push(makeFetchAgentsPayload({ entries: [] })); + const store = new HostRuntimeStore({ + deps: { + createClient: () => fakeClient as unknown as DaemonClient, + connectToDaemon: async () => ({ + client: fakeClient as unknown as DaemonClient, + serverId: host.serverId, + hostname: null, + }), + getClientId: async () => "cid_overlap_failure", + }, + }); + useSessionStore + .getState() + .initializeSession(host.serverId, fakeClient as unknown as DaemonClient, 1); + store.syncHosts([host]); + await fakeClient.waitForFetches(1); + await waitForDirectoryReady(store, host.serverId); + + const olderPage = new Deferred>>(); + fakeClient.fetchAgentsResponses.push(olderPage.promise); + const olderRefresh = store.refreshAgentDirectory({ serverId: host.serverId }); + await fakeClient.waitForFetches(2); + const liveEntry = makeFetchAgentsEntry({ + id: "live-delta", + cwd: "/repo", + updatedAt: "2026-07-17T10:00:00.000Z", + title: "preserved", + }); + fakeClient.agentUpdate({ kind: "upsert", agent: liveEntry.agent, project: liveEntry.project }); + + const newerPage = new Deferred>>(); + fakeClient.fetchAgentsResponses.push(newerPage.promise); + const newerRefresh = store.refreshAgentDirectory({ serverId: host.serverId }); + await fakeClient.waitForFetches(3); + newerPage.reject(new Error("newer refresh failed")); + await expect(newerRefresh).rejects.toThrow("newer refresh failed"); + + expect( + useSessionStore.getState().sessions[host.serverId]?.agents.get("live-delta")?.title, + ).toBe("preserved"); + + olderPage.resolve(makeFetchAgentsPayload({ entries: [] })); + await expect(olderRefresh).rejects.toThrow(); + store.syncHosts([]); + useSessionStore.getState().clearSession(host.serverId); + }); + it("rejects a refresh when the session generation changes before commit", async () => { const host = makeHost({ serverId: "srv_stale_generation" }); const fakeClient = new FakeDaemonClient(); diff --git a/packages/app/src/runtime/host-runtime.ts b/packages/app/src/runtime/host-runtime.ts index 16568cd2e..70d39b782 100644 --- a/packages/app/src/runtime/host-runtime.ts +++ b/packages/app/src/runtime/host-runtime.ts @@ -67,6 +67,7 @@ import { splitComposerAttachmentsForSubmit, } from "@/composer/attachments/submit"; import { encodeImages } from "@/utils/encode-images"; +import { inheritBufferedDirectoryDeltas } from "@/utils/buffered-directory-transaction"; export type HostRuntimeConnectionStatus = "idle" | "connecting" | "online" | "offline" | "error"; export type HostRegistryStatus = "loading" | "ready"; @@ -2245,12 +2246,16 @@ export class HostRuntimeStore { throw new Error(`Host ${input.serverId} is not connected`); } + const previousTransaction = this.agentDirectoryTransactions.get(input.serverId); const transaction: AgentDirectoryTransaction = { id: Symbol("agent directory refresh"), client, clientGeneration: snapshot.clientGeneration, entries: [], - deltas: [], + deltas: + previousTransaction?.clientGeneration === snapshot.clientGeneration + ? inheritBufferedDirectoryDeltas({ client, previous: previousTransaction }) + : [], }; this.agentDirectoryTransactions.set(input.serverId, transaction); diff --git a/packages/app/src/utils/agent-directory-sync.test.ts b/packages/app/src/utils/agent-directory-sync.test.ts index 6b1c0910e..b357652e7 100644 --- a/packages/app/src/utils/agent-directory-sync.test.ts +++ b/packages/app/src/utils/agent-directory-sync.test.ts @@ -7,7 +7,11 @@ import { useSessionStore } from "@/stores/session-store"; import { normalizeAgentSnapshot } from "@/utils/agent-snapshots"; import { isAgentArchiving, setAgentArchiving } from "@/hooks/use-archive-agent"; import { queryClient } from "@/data/query-client"; -import { applyAgentDirectoryDelta, replaceFetchedAgentDirectory } from "./agent-directory-sync"; +import { + applyAgentDirectoryDelta, + replaceFetchedAgentDirectory, + shouldApplyTimelineAgentSnapshot, +} from "./agent-directory-sync"; function createAgentPayload( input: Partial> & { @@ -141,6 +145,7 @@ describe("replaceFetchedAgentDirectory", () => { serverId, new Map([["permission", { key: "permission", agentId, request: null as never }]]), ); + store.setInitializingAgents(serverId, new Map([[agentId, true]])); setAgentArchiving({ queryClient, serverId, agentId, isArchiving: true }); applyAgentDirectoryDelta({ serverId, delta: { kind: "remove", agentId } }); @@ -152,6 +157,8 @@ describe("replaceFetchedAgentDirectory", () => { queued: session?.queuedMessages.has(agentId), cursor: session?.agentTimelineCursor.has(agentId), permissions: session?.pendingPermissions.size, + initializing: session?.initializingAgents.has(agentId), + acceptsStaleTimelineSnapshot: shouldApplyTimelineAgentSnapshot(serverId, agentId), archivePending: isAgentArchiving({ queryClient, serverId, agentId }), }).toEqual({ agents: false, @@ -159,6 +166,8 @@ describe("replaceFetchedAgentDirectory", () => { queued: false, cursor: false, permissions: 0, + initializing: false, + acceptsStaleTimelineSnapshot: false, archivePending: false, }); diff --git a/packages/app/src/utils/agent-directory-sync.ts b/packages/app/src/utils/agent-directory-sync.ts index dfbff645d..fd785ba97 100644 --- a/packages/app/src/utils/agent-directory-sync.ts +++ b/packages/app/src/utils/agent-directory-sync.ts @@ -24,6 +24,11 @@ export function applyAgentDirectoryDelta(input: { serverId: string; delta: Agent return upsertAgentDirectoryReplica(input.serverId, input.delta); } +export function shouldApplyTimelineAgentSnapshot(serverId: string, agentId: string): boolean { + const session = useSessionStore.getState().sessions[serverId]; + return session?.agents.has(agentId) === true || session?.initializingAgents.get(agentId) === true; +} + type AgentUpsertDelta = Extract; function upsertAgentDirectoryReplica( @@ -100,6 +105,7 @@ function removeAgentDirectoryReplica(serverId: string, agentId: string): void { store.setAgentDetails(serverId, removeKey); store.setQueuedMessages(serverId, removeKey); store.setAgentTimelineCursor(serverId, removeKey); + store.setInitializingAgents(serverId, removeKey); store.setPendingPermissions(serverId, (current) => { const next = new Map(current); for (const [key, pending] of next) { diff --git a/packages/app/src/utils/buffered-directory-transaction.test.ts b/packages/app/src/utils/buffered-directory-transaction.test.ts new file mode 100644 index 000000000..ec06ca9d5 --- /dev/null +++ b/packages/app/src/utils/buffered-directory-transaction.test.ts @@ -0,0 +1,23 @@ +import { describe, expect, it } from "vitest"; +import { inheritBufferedDirectoryDeltas } from "./buffered-directory-transaction"; + +describe("inheritBufferedDirectoryDeltas", () => { + it("hands buffered updates to a superseding transaction for the same client", () => { + const client = {}; + const previous = { client, deltas: ["first"] }; + + const inherited = inheritBufferedDirectoryDeltas({ client, previous }); + previous.deltas.push("later"); + + expect(inherited).toEqual(["first"]); + }); + + it("does not carry updates across clients", () => { + expect( + inheritBufferedDirectoryDeltas({ + client: {}, + previous: { client: {}, deltas: ["stale"] }, + }), + ).toEqual([]); + }); +}); diff --git a/packages/app/src/utils/buffered-directory-transaction.ts b/packages/app/src/utils/buffered-directory-transaction.ts new file mode 100644 index 000000000..05592fafa --- /dev/null +++ b/packages/app/src/utils/buffered-directory-transaction.ts @@ -0,0 +1,12 @@ +export interface BufferedDirectoryTransaction { + client: TClient; + deltas: TDelta[]; +} + +export function inheritBufferedDirectoryDeltas(input: { + client: TClient; + previous: BufferedDirectoryTransaction | null | undefined; +}): TDelta[] { + if (input.previous?.client !== input.client) return []; + return [...input.previous.deltas]; +}