authorgravatar for 130821636+darjss@users.noreply.github.comAviddaram <130821636+darjss@users.noreply.github.com> 2026-10-08 09:22:00+00:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-08 02:22:00-07:00
log10e29a26c289140b8f2312045c1ce0c278c7fe1a
treedf481e47fcad390679f84d7d934cbccba34dea25
parent1f31906c8e3003b2a9d7118c4da163544fa86586
signature Signed by PGP key B5690EEEBB952194

perf(server): ACP tool updates no longer persist a snapshot per streamed chunk (#16682)

Fixes #16428.

3 files changed, 250 insertions(+), 1 deletions(-)

apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts+178
......@@ -998,6 +998,184 @@ describe("AcpAdapterV2", () => {
998998 }).pipe(Effect.provide(layerTest), Effect.scoped),
999999 );
10001000
1001 it.effect("persists a bounded subset of streamed root and child tool updates", () =>
1002 Effect.gen(function* () {
1003 const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner;
1004 const path = yield* Path.Path;
1005 const mockAgentPath = yield* path.fromFileUrl(
1006 new URL("../../../scripts/acp-mock-agent.ts", import.meta.url),
1007 );
1008 type Runtime = AcpSessionRuntime.AcpSessionRuntime["Service"];
1009 let handler: Parameters<Runtime["handleSessionUpdate"]>[0] | undefined;
1010 // Devin streams a file write as it generates it: each update resends the
1011 // whole file so far, a few characters longer, with no status.
1012 const file = `${"x = 1\n".repeat(800)}END_OF_FILE\n`;
1013 const chunks = 200;
1014 const streamedWrite = (
1015 toolCallId: string,
1016 parentAgentId?: string,
1017 ): Array<EffectAcpSchema.SessionUpdate> => {
1018 const meta =
1019 parentAgentId === undefined ? {} : { "cognition.ai/subagent_context": { parentAgentId } };
1020 const write = (text: string, status?: "completed") => ({
1021 sessionUpdate: "tool_call_update" as const,
1022 toolCallId,
1023 title: "Writing ./audit.py",
1024 ...(status === undefined ? {} : { status }),
1025 content: [
1026 { type: "diff" as const, path: "/repo/audit.py", oldText: null, newText: text },
1027 ],
1028 rawInput: { file_path: "/repo/audit.py", content: text },
1029 _meta: meta,
1030 });
1031 return [
1032 {
1033 sessionUpdate: "tool_call",
1034 toolCallId,
1035 title: "Writing …",
1036 kind: "edit",
1037 status: "pending",
1038 _meta: meta,
1039 },
1040 ...Array.from({ length: chunks }, (_, index) =>
1041 write(file.slice(0, Math.ceil((file.length * (index + 1)) / chunks))),
1042 ),
1043 write(file, "completed"),
1044 ];
1045 };
1046 const instanceId = ProviderInstanceId.make("devin-streamed-write");
1047 const adapter = makeAcpAdapterV2({
1048 instanceId,
1049 crypto: yield* Crypto.Crypto,
1050 fileSystem: yield* FileSystem.FileSystem,
1051 idAllocator: yield* IdAllocator.IdAllocatorV2,
1052 serverConfig: yield* ServerConfig.ServerConfig,
1053 selfInvocation: yield* resolveSelfInvocation(),
1054 flavor: {
1055 driver: ACP_TEST_DRIVER,
1056 capabilities: AcpProviderCapabilitiesV2,
1057 normalizeSessionUpdate: normalizeDevinSessionUpdate,
1058 normalizeToolCall: normalizeDevinToolCall,
1059 extractSubagentUpdate: extractDevinSubagentUpdate,
1060 makeRuntime: makeMockRuntime({
1061 childProcessSpawner,
1062 mockAgentPath,
1063 wrapRuntime: (runtime) => ({
1064 ...runtime,
1065 handleSessionUpdate: (next) =>
1066 Effect.sync(() => {
1067 handler = next;
1068 }).pipe(Effect.andThen(runtime.handleSessionUpdate(next))),
1069 prompt: () =>
1070 Effect.gen(function* () {
1071 const updates = [
1072 {
1073 sessionUpdate: "tool_call_update",
1074 toolCallId: "child-a",
1075 status: "in_progress",
1076 _meta: {
1077 "cognition.ai/subagent_started": {
1078 agentId: "child-a",
1079 title: "Write audit script",
1080 task: "Write audit.py.",
1081 },
1082 },
1083 },
1084 ...streamedWrite("root-write"),
1085 ...streamedWrite("child-write", "child-a"),
1086 // Same-length output replacements are still live output.
1087 ...Array.from({ length: 5 }, (_, index) => ({
1088 sessionUpdate: "tool_call_update" as const,
1089 toolCallId: "ticker",
1090 title: "Watch ticks",
1091 kind: "execute" as const,
1092 status: "in_progress" as const,
1093 rawOutput: `tick ${index + 1}`,
1094 })),
1095 ...Array.from({ length: 5 }, (_, index) => ({
1096 sessionUpdate: "tool_call_update" as const,
1097 toolCallId: "text-ticker",
1098 title: "Watch ticks",
1099 kind: "execute" as const,
1100 status: "in_progress" as const,
1101 rawOutput: { type: "Text", text: `tock ${index + 1}` },
1102 })),
1103 {
1104 sessionUpdate: "tool_call_update",
1105 toolCallId: "child-a",
1106 status: "completed",
1107 _meta: {
1108 "cognition.ai/subagent_completed": {
1109 agentId: "child-a",
1110 success: true,
1111 summary: "Wrote audit.py",
1112 },
1113 },
1114 },
1115 ] satisfies Array<EffectAcpSchema.SessionUpdate>;
1116 for (const update of updates)
1117 yield* handler!({ sessionId: "mock-session-1", update });
1118 return { stopReason: "end_turn" as const };
1119 }),
1120 }),
1121 }),
1122 },
1123 });
1124 const threadId = ThreadId.make("devin-streamed-write");
1125 const modelSelection = { instanceId, model: "default" };
1126 const runtimePolicy = ProviderAdapterV2RuntimePolicy.make({
1127 runtimeMode: "full-access",
1128 interactionMode: "default",
1129 cwd: process.cwd(),
1130 });
1131 const runtime = yield* adapter.openSession({
1132 threadId,
1133 providerSessionId: ProviderSessionId.make("devin-streamed-write-session"),
1134 modelSelection,
1135 runtimePolicy,
1136 });
1137 const providerThread = yield* runtime.ensureThread({
1138 threadId,
1139 modelSelection,
1140 runtimePolicy,
1141 });
1142 yield* runtime.startTurn(
1143 makeTurnInput({
1144 threadId,
1145 providerThread,
1146 instanceId,
1147 runtimePolicy,
1148 now: yield* DateTime.now,
1149 }),
1150 );
1151 const items = Array.from(
1152 yield* runtime.events.pipe(
1153 Stream.takeUntil((event) => event.type === "turn.terminal"),
1154 Stream.runCollect,
1155 ),
1156 ).flatMap((event) => (event.type === "turn_item.updated" ? [event.turnItem] : []));
1157 for (const toolCallId of ["root-write", "child-write"]) {
1158 const writes = items.filter((item) => item.nativeItemRef?.nativeId?.endsWith(toolCallId));
1159 assert.isAtLeast(writes.length, chunks / 10, toolCallId);
1160 assert.isAtMost(writes.length, chunks / 5, toolCallId);
1161 assert.equal(writes.at(-1)?.status, "completed", toolCallId);
1162 assert.include(JSON.stringify(writes.at(-1)), "END_OF_FILE", toolCallId);
1163 }
1164 for (const [toolCallId, word] of [
1165 ["ticker", "tick"],
1166 ["text-ticker", "tock"],
1167 ] as const) {
1168 const ticks = items.filter((item) => item.nativeItemRef?.nativeId?.endsWith(toolCallId));
1169 for (let tick = 1; tick <= 5; tick++) {
1170 assert.isTrue(
1171 ticks.some((item) => JSON.stringify(item).includes(`${word} ${tick}`)),
1172 `${word} ${tick} must persist`,
1173 );
1174 }
1175 }
1176 }).pipe(Effect.provide(layerTest), Effect.scoped),
1177 );
1178
10011179 it.effect("keeps Devin parent paragraphs intact while projecting native child work", () =>
10021180 Effect.gen(function* () {
10031181 const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner;
apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.ts+49-1
......@@ -73,6 +73,7 @@ import {
7373 mergeToolCallState,
7474 parsePermissionRequest,
7575 parseSessionUpdateEvent,
76 toolCallVisibleOutputChanged,
7677 type AcpPlanUpdate,
7778 type AcpAgentTerminalState,
7879 type AcpSessionModeState,
......@@ -1162,6 +1163,8 @@ interface ActiveAcpTurn {
11621163 contextUsage: ThreadTokenUsageSnapshot | null;
11631164 nativeMetadata: OrchestrationV2ProviderThreadNativeMetadata | null;
11641165 readonly tools: Map<string, AcpToolCallState>;
1166 /** Streamed tool updates skipped since the last persisted one; see `shouldPersistToolUpdate`. */
1167 readonly toolUpdatesSkipped: Map<string, number>;
11651168 readonly toolStartedAt: Map<string, DateTime.Utc>;
11661169 readonly subagents: Map<string, ActiveAcpSubagent>;
11671170 readonly subagentsBySessionId: Map<string, ActiveAcpSubagent>;
......@@ -1460,6 +1463,34 @@ interface SnapshotMessageState {
14601463 loadingIndex: number;
14611464}
14621465
1466const TOOL_UPDATE_PERSIST_EVERY = 10;
1467
1468/**
1469 * Some agents stream a tool's arguments (a file write's diff, `rawInput`) and
1470 * resend the whole call each time. Persist every 10th of those. Status, title,
1471 * and output the user watches live always persist, as does the agent's own
1472 * completed/failed when a flavor normalizes it to a non-terminal status.
1473 */
1474function shouldPersistToolUpdate(
1475 context: ActiveAcpTurn,
1476 key: string,
1477 previous: AcpToolCallState | undefined,
1478 next: AcpToolCallState,
1479 reportedStatus: AcpToolCallState["status"],
1480): boolean {
1481 const skipped = context.toolUpdatesSkipped.get(key) ?? 0;
1482 const persist =
1483 reportedStatus === "completed" ||
1484 reportedStatus === "failed" ||
1485 previous === undefined ||
1486 previous.status !== next.status ||
1487 previous.title !== next.title ||
1488 toolCallVisibleOutputChanged(previous, next) ||
1489 skipped + 1 >= TOOL_UPDATE_PERSIST_EVERY;
1490 context.toolUpdatesSkipped.set(key, persist ? 0 : skipped + 1);
1491 return persist;
1492}
1493
14631494export function makeAcpAdapterV2(
14641495 options: AcpAdapterV2Options,
14651496): ProviderAdapter.ProviderAdapterV2Shape {
......@@ -3196,6 +3227,19 @@ export function makeAcpAdapterV2(
31963227 return;
31973228 }
31983229 }
3230 if (
3231 projectedStatus === undefined &&
3232 !shouldPersistToolUpdate(
3233 context,
3234 toolCall.toolCallId,
3235 previous,
3236 toolCall,
3237 merged.status,
3238 )
3239 ) {
3240 yield* rearmDeferredFinalize(context);
3241 return;
3242 }
31993243 const status = projectedStatus ?? toolStatus(toolCall.status);
32003244 const now = yield* DateTime.now;
32013245 const nativeItemId = `${context.nativeThreadId}:tool:${toolCall.toolCallId}`;
......@@ -4369,8 +4413,11 @@ export function makeAcpAdapterV2(
43694413 continue;
43704414 }
43714415 const key = `${nativeTaskId}:tool:${toolCall.toolCallId}`;
4372 const merged = mergeToolCallState(context.tools.get(key), toolCall);
4416 const previous = context.tools.get(key);
4417 const merged = mergeToolCallState(previous, toolCall);
43734418 context.tools.set(key, merged);
4419 if (!shouldPersistToolUpdate(context, key, previous, merged, merged.status))
4420 continue;
43744421 // Terminals are remembered under the raw session id: the child's
43754422 // own session, or the root one when the flavor routes child
43764423 // updates out of it (Devin).
......@@ -6953,6 +7000,7 @@ export function makeAcpAdapterV2(
69537000 contextUsage: rememberedContextUsage ?? turnInput.providerThread.contextUsage ?? null,
69547001 nativeMetadata: initialNativeMetadata,
69557002 tools: new Map(),
7003 toolUpdatesSkipped: new Map(),
69567004 toolStartedAt: new Map(),
69577005 subagents: new Map(),
69587006 subagentsBySessionId: new Map(),
apps/server/src/provider/acp/AcpRuntimeModel.ts+23
......@@ -1000,6 +1000,29 @@ export function toolCallProgressLength(state: AcpToolCallState): number {
10001000 return Math.max(state.detail?.length ?? 0, contentChars, rawOutputChars);
10011001}
10021002
1003function toolCallContentTexts(state: AcpToolCallState): string {
1004 const content = state.data.content;
1005 if (!Array.isArray(content)) return "";
1006 return content
1007 .map((entry) =>
1008 isRecord(entry) ? (toolCallContentText(entry as EffectAcpSchema.ToolCallContent) ?? "") : "",
1009 )
1010 .join("\u0000");
1011}
1012
1013// The output a user watches: detail, text content, and any `rawOutput`. A
1014// streamed diff or `rawInput` is not part of it.
1015export function toolCallVisibleOutputChanged(
1016 previous: AcpToolCallState,
1017 next: AcpToolCallState,
1018): boolean {
1019 return (
1020 previous.detail !== next.detail ||
1021 toolCallContentTexts(previous) !== toolCallContentTexts(next) ||
1022 JSON.stringify(previous.data.rawOutput) !== JSON.stringify(next.data.rawOutput)
1023 );
1024}
1025
10031026export function decideToolCallUpdateEmission(
10041027 input: AcpToolCallEmitDecisionInput,
10051028): AcpToolCallEmitDecision {