diff --git a/packages/adapter-utils/src/acpx-engine/ephemeral-session-environment.test.ts b/packages/adapter-utils/src/acpx-engine/ephemeral-session-environment.test.ts index 40d13e7b403..ef0e834aa8b 100644 --- a/packages/adapter-utils/src/acpx-engine/ephemeral-session-environment.test.ts +++ b/packages/adapter-utils/src/acpx-engine/ephemeral-session-environment.test.ts @@ -162,4 +162,45 @@ describe("ephemeral ACPX session environment", () => { const input = freeze(record()); await expect(wrapper.save(input)).rejects.toBe(failure); expect(input.acpx?.session_options?.env).toEqual(currentEnvironment); }); + + it("never overlaps two saves of one record, even across wrappers, and keeps their order", async () => { + let active = 0, maxActive = 0; + const order: string[] = []; + const failure = new Error("fixture rename failure"); + const backing = stubStore(undefined); + backing.save.mockImplementation(async (saved) => { + active += 1; maxActive = Math.max(maxActive, active); + await new Promise(resolve => setTimeout(resolve, 5)); + order.push(saved.lastUsedAt); + active -= 1; + if (saved.lastUsedAt === "checkpoint") throw failure; + }); + const turnStore = createEphemeralSessionEnvironmentStore(backing, currentEnvironment); + const closeStore = createEphemeralSessionEnvironmentStore(backing, currentEnvironment); + const results = await Promise.allSettled([ + turnStore.save({ ...record(), lastUsedAt: "checkpoint" }), + turnStore.save({ ...record(), lastUsedAt: "turn-end" }), + closeStore.save({ ...record(), lastUsedAt: "close" }), + ]); + expect(maxActive).toBe(1); + expect(order).toEqual(["checkpoint", "turn-end", "close"]); + expect(results.map(result => result.status)).toEqual(["rejected", "fulfilled", "fulfilled"]); + expect((results[0] as PromiseRejectedResult).reason).toBe(failure); + }); + + it("does not serialize saves of different records", async () => { + let active = 0, maxActive = 0; + const backing = stubStore(undefined); + backing.save.mockImplementation(async () => { + active += 1; maxActive = Math.max(maxActive, active); + await new Promise(resolve => setTimeout(resolve, 5)); + active -= 1; + }); + const wrapper = createEphemeralSessionEnvironmentStore(backing, currentEnvironment); + await Promise.all([ + wrapper.save({ ...record(), acpxRecordId: "fixture-record-a" }), + wrapper.save({ ...record(), acpxRecordId: "fixture-record-b" }), + ]); + expect(maxActive).toBe(2); + }); }); diff --git a/packages/adapter-utils/src/acpx-engine/ephemeral-session-environment.ts b/packages/adapter-utils/src/acpx-engine/ephemeral-session-environment.ts index 3bd59942ec9..a4d068f9f6c 100644 --- a/packages/adapter-utils/src/acpx-engine/ephemeral-session-environment.ts +++ b/packages/adapter-utils/src/acpx-engine/ephemeral-session-environment.ts @@ -1,5 +1,22 @@ import type { AcpSessionStore } from "acpx/runtime"; +// ACPX saves one record from several writers (live checkpoint, turn end, +// finalize, close) with no lock. Each save renames a temp file onto the record +// file, and on Windows two overlapping renames onto one file fail with EPERM. +// Chain every save of one record so they never overlap. +const pendingSaves = new Map>(); + +function saveInOrder(recordId: string, save: () => Promise): Promise { + const previous = pendingSaves.get(recordId) ?? Promise.resolve(); + const current = previous.catch(() => undefined).then(save); + const settled = current.catch(() => undefined); + pendingSaves.set(recordId, settled); + void settled.then(() => { + if (pendingSaves.get(recordId) === settled) pendingSaves.delete(recordId); + }); + return current; +} + /** Keep the current run's environment in memory, outside persisted conversation state. */ export function createEphemeralSessionEnvironmentStore( persistedStore: AcpSessionStore, @@ -26,7 +43,7 @@ export function createEphemeralSessionEnvironmentStore( const { env: _environment, ...sessionOptions } = record.acpx.session_options; persisted.acpx = { ...record.acpx, session_options: sessionOptions }; } - return persistedStore.save(persisted); + return saveInOrder(record.acpxRecordId, () => persistedStore.save(persisted)); }, }; }