From 10e29a26c289140b8f2312045c1ce0c278c7fe1a Mon Sep 17 00:00:00 2001 From: Aviddaram <130821636+darjss@users.noreply.github.com> Date: Thu, 8 Oct 2026 09:22:00 +0000 Subject: [PATCH] perf(server): ACP tool updates no longer persist a snapshot per streamed chunk (#16682) Fixes #16428. --- .../Adapters/AcpAdapterV2.test.ts | 178 ++++++++++++++++++ .../orchestration-v2/Adapters/AcpAdapterV2.ts | 50 ++++- .../src/provider/acp/AcpRuntimeModel.ts | 23 +++ 3 files changed, 250 insertions(+), 1 deletion(-) diff --git a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts index 0724c3c0a57f5088579d619957e4bde68de95d89..dfefe387a58a39bece9ea0a6e8b6d30386a6cbc1 100644 --- a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts @@ -998,6 +998,184 @@ describe("AcpAdapterV2", () => { }).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; + const path = yield* Path.Path; + const mockAgentPath = yield* path.fromFileUrl( + new URL("../../../scripts/acp-mock-agent.ts", import.meta.url), + ); + type Runtime = AcpSessionRuntime.AcpSessionRuntime["Service"]; + let handler: Parameters[0] | undefined; + // Devin streams a file write as it generates it: each update resends the + // whole file so far, a few characters longer, with no status. + const file = `${"x = 1\n".repeat(800)}END_OF_FILE\n`; + const chunks = 200; + const streamedWrite = ( + toolCallId: string, + parentAgentId?: string, + ): Array => { + const meta = + parentAgentId === undefined ? {} : { "cognition.ai/subagent_context": { parentAgentId } }; + const write = (text: string, status?: "completed") => ({ + sessionUpdate: "tool_call_update" as const, + toolCallId, + title: "Writing ./audit.py", + ...(status === undefined ? {} : { status }), + content: [ + { type: "diff" as const, path: "/repo/audit.py", oldText: null, newText: text }, + ], + rawInput: { file_path: "/repo/audit.py", content: text }, + _meta: meta, + }); + return [ + { + sessionUpdate: "tool_call", + toolCallId, + title: "Writing …", + kind: "edit", + status: "pending", + _meta: meta, + }, + ...Array.from({ length: chunks }, (_, index) => + write(file.slice(0, Math.ceil((file.length * (index + 1)) / chunks))), + ), + write(file, "completed"), + ]; + }; + const instanceId = ProviderInstanceId.make("devin-streamed-write"); + const adapter = makeAcpAdapterV2({ + instanceId, + crypto: yield* Crypto.Crypto, + fileSystem: yield* FileSystem.FileSystem, + idAllocator: yield* IdAllocator.IdAllocatorV2, + serverConfig: yield* ServerConfig.ServerConfig, + selfInvocation: yield* resolveSelfInvocation(), + flavor: { + driver: ACP_TEST_DRIVER, + capabilities: AcpProviderCapabilitiesV2, + normalizeSessionUpdate: normalizeDevinSessionUpdate, + normalizeToolCall: normalizeDevinToolCall, + extractSubagentUpdate: extractDevinSubagentUpdate, + makeRuntime: makeMockRuntime({ + childProcessSpawner, + mockAgentPath, + wrapRuntime: (runtime) => ({ + ...runtime, + handleSessionUpdate: (next) => + Effect.sync(() => { + handler = next; + }).pipe(Effect.andThen(runtime.handleSessionUpdate(next))), + prompt: () => + Effect.gen(function* () { + const updates = [ + { + sessionUpdate: "tool_call_update", + toolCallId: "child-a", + status: "in_progress", + _meta: { + "cognition.ai/subagent_started": { + agentId: "child-a", + title: "Write audit script", + task: "Write audit.py.", + }, + }, + }, + ...streamedWrite("root-write"), + ...streamedWrite("child-write", "child-a"), + // Same-length output replacements are still live output. + ...Array.from({ length: 5 }, (_, index) => ({ + sessionUpdate: "tool_call_update" as const, + toolCallId: "ticker", + title: "Watch ticks", + kind: "execute" as const, + status: "in_progress" as const, + rawOutput: `tick ${index + 1}`, + })), + ...Array.from({ length: 5 }, (_, index) => ({ + sessionUpdate: "tool_call_update" as const, + toolCallId: "text-ticker", + title: "Watch ticks", + kind: "execute" as const, + status: "in_progress" as const, + rawOutput: { type: "Text", text: `tock ${index + 1}` }, + })), + { + sessionUpdate: "tool_call_update", + toolCallId: "child-a", + status: "completed", + _meta: { + "cognition.ai/subagent_completed": { + agentId: "child-a", + success: true, + summary: "Wrote audit.py", + }, + }, + }, + ] satisfies Array; + for (const update of updates) + yield* handler!({ sessionId: "mock-session-1", update }); + return { stopReason: "end_turn" as const }; + }), + }), + }), + }, + }); + const threadId = ThreadId.make("devin-streamed-write"); + const modelSelection = { instanceId, model: "default" }; + const runtimePolicy = ProviderAdapterV2RuntimePolicy.make({ + runtimeMode: "full-access", + interactionMode: "default", + cwd: process.cwd(), + }); + const runtime = yield* adapter.openSession({ + threadId, + providerSessionId: ProviderSessionId.make("devin-streamed-write-session"), + modelSelection, + runtimePolicy, + }); + const providerThread = yield* runtime.ensureThread({ + threadId, + modelSelection, + runtimePolicy, + }); + yield* runtime.startTurn( + makeTurnInput({ + threadId, + providerThread, + instanceId, + runtimePolicy, + now: yield* DateTime.now, + }), + ); + const items = Array.from( + yield* runtime.events.pipe( + Stream.takeUntil((event) => event.type === "turn.terminal"), + Stream.runCollect, + ), + ).flatMap((event) => (event.type === "turn_item.updated" ? [event.turnItem] : [])); + for (const toolCallId of ["root-write", "child-write"]) { + const writes = items.filter((item) => item.nativeItemRef?.nativeId?.endsWith(toolCallId)); + assert.isAtLeast(writes.length, chunks / 10, toolCallId); + assert.isAtMost(writes.length, chunks / 5, toolCallId); + assert.equal(writes.at(-1)?.status, "completed", toolCallId); + assert.include(JSON.stringify(writes.at(-1)), "END_OF_FILE", toolCallId); + } + for (const [toolCallId, word] of [ + ["ticker", "tick"], + ["text-ticker", "tock"], + ] as const) { + const ticks = items.filter((item) => item.nativeItemRef?.nativeId?.endsWith(toolCallId)); + for (let tick = 1; tick <= 5; tick++) { + assert.isTrue( + ticks.some((item) => JSON.stringify(item).includes(`${word} ${tick}`)), + `${word} ${tick} must persist`, + ); + } + } + }).pipe(Effect.provide(layerTest), Effect.scoped), + ); + it.effect("keeps Devin parent paragraphs intact while projecting native child work", () => Effect.gen(function* () { const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; diff --git a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.ts index 179455cb64a8b8ed5f0a4bdd1834d4a9d1433a9f..7fa7b4874d8fe0fcd2ef3854887b5718a33c52bd 100644 --- a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.ts @@ -73,6 +73,7 @@ import { mergeToolCallState, parsePermissionRequest, parseSessionUpdateEvent, + toolCallVisibleOutputChanged, type AcpPlanUpdate, type AcpAgentTerminalState, type AcpSessionModeState, @@ -1162,6 +1163,8 @@ interface ActiveAcpTurn { contextUsage: ThreadTokenUsageSnapshot | null; nativeMetadata: OrchestrationV2ProviderThreadNativeMetadata | null; readonly tools: Map; + /** Streamed tool updates skipped since the last persisted one; see `shouldPersistToolUpdate`. */ + readonly toolUpdatesSkipped: Map; readonly toolStartedAt: Map; readonly subagents: Map; readonly subagentsBySessionId: Map; @@ -1460,6 +1463,34 @@ interface SnapshotMessageState { loadingIndex: number; } +const TOOL_UPDATE_PERSIST_EVERY = 10; + +/** + * Some agents stream a tool's arguments (a file write's diff, `rawInput`) and + * resend the whole call each time. Persist every 10th of those. Status, title, + * and output the user watches live always persist, as does the agent's own + * completed/failed when a flavor normalizes it to a non-terminal status. + */ +function shouldPersistToolUpdate( + context: ActiveAcpTurn, + key: string, + previous: AcpToolCallState | undefined, + next: AcpToolCallState, + reportedStatus: AcpToolCallState["status"], +): boolean { + const skipped = context.toolUpdatesSkipped.get(key) ?? 0; + const persist = + reportedStatus === "completed" || + reportedStatus === "failed" || + previous === undefined || + previous.status !== next.status || + previous.title !== next.title || + toolCallVisibleOutputChanged(previous, next) || + skipped + 1 >= TOOL_UPDATE_PERSIST_EVERY; + context.toolUpdatesSkipped.set(key, persist ? 0 : skipped + 1); + return persist; +} + export function makeAcpAdapterV2( options: AcpAdapterV2Options, ): ProviderAdapter.ProviderAdapterV2Shape { @@ -3196,6 +3227,19 @@ export function makeAcpAdapterV2( return; } } + if ( + projectedStatus === undefined && + !shouldPersistToolUpdate( + context, + toolCall.toolCallId, + previous, + toolCall, + merged.status, + ) + ) { + yield* rearmDeferredFinalize(context); + return; + } const status = projectedStatus ?? toolStatus(toolCall.status); const now = yield* DateTime.now; const nativeItemId = `${context.nativeThreadId}:tool:${toolCall.toolCallId}`; @@ -4369,8 +4413,11 @@ export function makeAcpAdapterV2( continue; } const key = `${nativeTaskId}:tool:${toolCall.toolCallId}`; - const merged = mergeToolCallState(context.tools.get(key), toolCall); + const previous = context.tools.get(key); + const merged = mergeToolCallState(previous, toolCall); context.tools.set(key, merged); + if (!shouldPersistToolUpdate(context, key, previous, merged, merged.status)) + continue; // Terminals are remembered under the raw session id: the child's // own session, or the root one when the flavor routes child // updates out of it (Devin). @@ -6953,6 +7000,7 @@ export function makeAcpAdapterV2( contextUsage: rememberedContextUsage ?? turnInput.providerThread.contextUsage ?? null, nativeMetadata: initialNativeMetadata, tools: new Map(), + toolUpdatesSkipped: new Map(), toolStartedAt: new Map(), subagents: new Map(), subagentsBySessionId: new Map(), diff --git a/apps/server/src/provider/acp/AcpRuntimeModel.ts b/apps/server/src/provider/acp/AcpRuntimeModel.ts index 694f761ff8f752f7d0ada345a3ff560bb56732fb..f4172d17089b1ac66d059bd43c469e684d6adf85 100644 --- a/apps/server/src/provider/acp/AcpRuntimeModel.ts +++ b/apps/server/src/provider/acp/AcpRuntimeModel.ts @@ -1000,6 +1000,29 @@ export function toolCallProgressLength(state: AcpToolCallState): number { return Math.max(state.detail?.length ?? 0, contentChars, rawOutputChars); } +function toolCallContentTexts(state: AcpToolCallState): string { + const content = state.data.content; + if (!Array.isArray(content)) return ""; + return content + .map((entry) => + isRecord(entry) ? (toolCallContentText(entry as EffectAcpSchema.ToolCallContent) ?? "") : "", + ) + .join("\u0000"); +} + +// The output a user watches: detail, text content, and any `rawOutput`. A +// streamed diff or `rawInput` is not part of it. +export function toolCallVisibleOutputChanged( + previous: AcpToolCallState, + next: AcpToolCallState, +): boolean { + return ( + previous.detail !== next.detail || + toolCallContentTexts(previous) !== toolCallContentTexts(next) || + JSON.stringify(previous.data.rawOutput) !== JSON.stringify(next.data.rawOutput) + ); +} + export function decideToolCallUpdateEmission( input: AcpToolCallEmitDecisionInput, ): AcpToolCallEmitDecision { -- 2.54.0