diff --git a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts index 1a0c6caf1f92..bbeb4ac3d41d 100644 --- a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts @@ -13,7 +13,7 @@ import { decodeAcpReplayTranscript, makeAcpReplayRuntime } from "./AcpAdapterV2. import * as NodePath from "node:path"; import * as NodeServices from "@effect/platform-node/NodeServices"; -import { assert, describe, it } from "@effect/vitest"; +import { assert, describe, it, vi } from "@effect/vitest"; import { CheckpointId, EnvironmentId, @@ -1216,6 +1216,110 @@ describe("AcpAdapterV2", () => { }).pipe(Effect.provide(layerTest), Effect.scoped), ); + it.effect("replays history in order, with stable earlier snapshots and linear work", () => + Effect.gen(function* () { + const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const path = yield* Path.Path; + const selfInvocation = yield* resolveSelfInvocation(); + const mockAgentPath = yield* path.fromFileUrl( + new URL("../../../scripts/acp-mock-agent.ts", import.meta.url), + ); + type RuntimeService = AcpSessionRuntime.AcpSessionRuntime["Service"]; + let sessionUpdateHandler: Parameters[0] | undefined; + const instanceId = ProviderInstanceId.make("acp-test-v2-replay-order"); + const adapter = yield* makeAcpAdapterV2({ + instanceId, + flavor: { + driver: ACP_TEST_DRIVER, + capabilities: AcpProviderCapabilitiesV2, + makeRuntime: makeMockRuntime({ + childProcessSpawner, + mockAgentPath, + wrapRuntime: (runtime) => ({ + ...runtime, + handleSessionUpdate: (handler) => + Effect.sync(() => { + sessionUpdateHandler = handler; + }).pipe(Effect.andThen(runtime.handleSessionUpdate(handler))), + }), + }), + }, + selfInvocation, + }); + const threadId = ThreadId.make("thread-acp-v2-replay-order"); + const runtimePolicy = ProviderAdapter.ProviderAdapterV2RuntimePolicy.make({ + runtimeMode: "full-access", + interactionMode: "default", + cwd: process.cwd(), + }); + const modelSelection = { instanceId, model: "default" } as const; + const runtime = yield* adapter.openSession({ + threadId, + providerSessionId: ProviderSessionId.make("provider-session-acp-v2-replay-order"), + modelSelection, + runtimePolicy, + }); + const providerThread = yield* runtime.ensureThread({ + threadId, + modelSelection, + runtimePolicy, + }); + assert.isDefined(sessionUpdateHandler); + const replay = ( + sessionUpdate: "user_message_chunk" | "agent_message_chunk", + messageId: string, + text: string, + ) => + sessionUpdateHandler!({ + sessionId: "mock-session-1", + update: { sessionUpdate, messageId, content: { type: "text", text } }, + }); + + yield* replay("user_message_chunk", "u1", "hel"); + yield* replay("user_message_chunk", "u1", "lo"); + yield* replay("agent_message_chunk", "a1", "hi"); + yield* replay("user_message_chunk", "u2", "next"); + yield* replay("agent_message_chunk", "a1", " there"); + const first = yield* runtime.readThreadSnapshot({ providerThread }); + yield* replay("agent_message_chunk", "a2", "done"); + yield* replay("user_message_chunk", "u2", " one"); + const second = yield* runtime.readThreadSnapshot({ providerThread }); + + const texts = (snapshot: typeof first) => + snapshot.messages.map((message) => `${message.role}:${message.text}`); + assert.deepEqual(texts(first), ["user:hello", "assistant:hi there", "user:next"]); + assert.deepEqual(texts(second), [ + "user:hello", + "assistant:hi there", + "user:next one", + "assistant:done", + ]); + + // Copying the history Map for each chunk calls Map.prototype.set once per + // stored message, so a long replay would make about n^2 / 2 writes instead + // of one per chunk. + const longHistory = 500; + const mapWrites = yield* Effect.acquireUseRelease( + Effect.sync(() => vi.spyOn(Map.prototype, "set")), + (mapSet) => + Effect.forEach( + Array.from({ length: longHistory }, (_, index) => index), + (index) => + replay( + index % 2 === 0 ? "user_message_chunk" : "agent_message_chunk", + `h${index}`, + "x", + ), + { discard: true }, + ).pipe(Effect.map(() => mapSet.mock.calls.length)), + (mapSet) => Effect.sync(() => mapSet.mockRestore()), + ); + const long = yield* runtime.readThreadSnapshot({ providerThread }); + assert.equal(long.messages.length, 4 + longHistory); + assert.isBelow(mapWrites, longHistory * 2); + }).pipe(Effect.provide(layerTest), Effect.scoped), + ); + it.effect("persists a bounded subset of streamed root and child tool updates", () => Effect.gen(function* () { const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; diff --git a/packages/provider-acp/src/server/adapter.ts b/packages/provider-acp/src/server/adapter.ts index 01a729cd7c9e..280bd4603c1e 100644 --- a/packages/provider-acp/src/server/adapter.ts +++ b/packages/provider-acp/src/server/adapter.ts @@ -1329,6 +1329,7 @@ type PendingRuntimeRequest = { } ); +/** Writers update `order` and `messages` in place; readers copy them out. */ interface SnapshotMessageState { readonly order: Array; readonly messages: Map; @@ -2294,14 +2295,9 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( const rememberSnapshotMessage = (message: OrchestrationV2ConversationMessage) => Ref.update(snapshot, (current) => { const key = String(message.id); - const exists = current.messages.has(key); - const messages = new Map(current.messages); - messages.set(key, message); - return { - ...current, - order: exists ? current.order : [...current.order, key], - messages, - }; + if (!current.messages.has(key)) current.order.push(key); + current.messages.set(key, message); + return current; }); const textStreamFor = ( @@ -4076,8 +4072,8 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( const messageId = providerMessageId(nativeItemId); const key = String(messageId); const previous = current.messages.get(key); - const messages = new Map(current.messages); - messages.set(key, { + if (previous === undefined) current.order.push(key); + current.messages.set(key, { createdBy: previous?.createdBy ?? (role === "user" ? "user" : "agent"), creationSource: previous?.creationSource ?? "provider", id: messageId, @@ -4092,8 +4088,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( updatedAt: now, }); return { - order: current.order.includes(key) ? current.order : [...current.order, key], - messages, + ...current, loadingRole: role, loadingMessageId: sourceMessageId ?? (startsNew ? null : current.loadingMessageId), loadingIndex, @@ -8298,6 +8293,10 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( yield* Ref.set(activeInteractionMode, null); } const state = yield* Ref.get(snapshot); + const messages = state.order.flatMap((key) => { + const message = state.messages.get(key); + return message === undefined ? [] : [message]; + }); const now = yield* DateTime.now; return { providerThread: { @@ -8307,10 +8306,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( updatedAt: now, }, providerTurns: [...(yield* Ref.get(providerTurns)).values()], - messages: state.order.flatMap((key) => { - const message = state.messages.get(key); - return message === undefined ? [] : [message]; - }), + messages, runtimeRequests: [], providerPayload: { protocol: ACP_PROTOCOL, sessionId }, };