1import * as Channel from "effect/Channel";
2import * as Deferred from "effect/Deferred";
3import * as Duration from "effect/Duration";
4import * as Effect from "effect/Effect";
5import * as Exit from "effect/Exit";
6import * as Fiber from "effect/Fiber";
7import * as Scope from "effect/Scope";
8import * as Semaphore from "effect/Semaphore";
9import * as Stream from "effect/Stream";
10import * as ChildProcess from "effect/process/ChildProcess";
11import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner";
12import * as EffectAcpErrors from "effect-acp/errors";
13import type * as EffectAcpSchema from "effect-acp/compat";
14import * as NodeBuffer from "node:buffer";
15
16const MAX_LIVE_TERMINALS = 16;
17const MAX_UNRELEASED_TERMINALS = 64;
18const MAX_RETAINED_TERMINALS = 32;
19const MAX_RETAINED_OUTPUT_BYTES = 16 * 1024 * 1024;
20const DEFAULT_OUTPUT_BYTE_LIMIT = 4 * 1024 * 1024;
21const MAX_OUTPUT_BYTE_LIMIT = 8 * 1024 * 1024;
22const OUTPUT_DRAIN_GRACE = "1 second";
23
24const terminalRequestError = (message: string, cause?: unknown) =>
25 EffectAcpErrors.AcpRequestError.internalError(
26 message,
27 undefined,
28 cause === undefined ? {} : { cause },
29 );
30
31interface OutputBufferState {
32 chunks: Array<Uint8Array>;
33 bytes: number;
34 truncated: boolean;
35 readonly limit: number;
36}
37
38function trimOutputStart(state: OutputBufferState, bytes: number): void {
39 let remaining = Math.min(bytes, state.bytes);
40 while (remaining > 0) {
41 const head = state.chunks[0]!;
42 if (head.byteLength <= remaining) {
43 state.chunks.shift();
44 state.bytes -= head.byteLength;
45 remaining -= head.byteLength;
46 } else {
47 state.chunks[0] = head.subarray(remaining);
48 state.bytes -= remaining;
49 remaining = 0;
50 }
51 state.truncated = true;
52 }
53}
54
55function appendOutput(state: OutputBufferState, chunk: Uint8Array): void {
56 state.chunks.push(chunk);
57 state.bytes += chunk.byteLength;
58 // The spec requires truncation from the beginning of the output.
59 if (state.bytes > state.limit) trimOutputStart(state, state.bytes - state.limit);
60}
61
62function readOutput(state: OutputBufferState): string {
63 let combined: Uint8Array = NodeBuffer.Buffer.concat(state.chunks, state.bytes);
64 if (state.truncated) {
65 // The spec requires truncation at a UTF-8 character boundary.
66 let start = 0;
67 while (start < combined.byteLength && (combined[start]! & 0b1100_0000) === 0b1000_0000) {
68 start += 1;
69 }
70 combined = combined.subarray(start);
71 }
72 return NodeBuffer.Buffer.from(combined).toString("utf8");
73}
74
75function terminalExitSignal(error: unknown): string | null {
76 const cause = error instanceof Error ? error.cause : undefined;
77 const message =
78 cause instanceof Error ? cause.message : error instanceof Error ? error.message : "";
79 return message.match(/signal: '([^']+)'/u)?.[1] ?? null;
80}
81
82export interface AcpTerminalOutputSnapshot {
83 readonly output: string;
84 readonly truncated: boolean;
85 readonly exitStatus: EffectAcpSchema.TerminalExitStatus | undefined;
86}
87
88interface ManagedAcpTerminal {
89 readonly terminalId: string;
90 readonly sessionId: string;
91 readonly commandLine: string;
92 readonly buffer: OutputBufferState;
93 readonly exit: Deferred.Deferred<EffectAcpSchema.WaitForTerminalExitResponse>;
94 readonly kill: Effect.Effect<void>;
95 readonly dispose: Effect.Effect<void>;
96 exitStatus: EffectAcpSchema.TerminalExitStatus | undefined;
97 released: boolean;
98}
99
100export interface AcpClientTerminals {
101 readonly create: (
102 request: EffectAcpSchema.CreateTerminalRequest,
103 ) => Effect.Effect<EffectAcpSchema.CreateTerminalResponse, EffectAcpErrors.AcpRequestError>;
104 readonly output: (
105 request: EffectAcpSchema.TerminalOutputRequest,
106 ) => Effect.Effect<EffectAcpSchema.TerminalOutputResponse, EffectAcpErrors.AcpRequestError>;
107 readonly waitForExit: (
108 request: EffectAcpSchema.WaitForTerminalExitRequest,
109 ) => Effect.Effect<EffectAcpSchema.WaitForTerminalExitResponse, EffectAcpErrors.AcpRequestError>;
110 readonly kill: (
111 request: EffectAcpSchema.KillTerminalRequest,
112 ) => Effect.Effect<EffectAcpSchema.KillTerminalResponse, EffectAcpErrors.AcpRequestError>;
113 readonly release: (
114 request: EffectAcpSchema.ReleaseTerminalRequest,
115 ) => Effect.Effect<EffectAcpSchema.ReleaseTerminalResponse, EffectAcpErrors.AcpRequestError>;
116 /** Buffered output for embedded tool-call content; survives release. */
117 readonly readOutputSnapshot: (
118 sessionId: string,
119 terminalId: string,
120 ) => AcpTerminalOutputSnapshot | undefined;
121 readonly readCommandLine: (terminalId: string) => string | undefined;
122 readonly disposeAll: Effect.Effect<void>;
123}
124
125export interface AcpClientTerminalsOptions {
126 /** Devin sends shell source in command, rather than an executable plus args. */
127 readonly shellCommands?: boolean | undefined;
128 readonly defaultCwd: string;
129 readonly environment?: NodeJS.ProcessEnv | undefined;
130 readonly environmentForSession?:
131 | ((sessionId: string) => NodeJS.ProcessEnv | undefined)
132 | undefined;
133 readonly forceKillAfter?: Duration.Input | undefined;
134}
135
136function acpTerminalCommand(input: {
137 readonly request: EffectAcpSchema.CreateTerminalRequest;
138 readonly shellCommands?: boolean | undefined;
139 readonly defaultCwd: string;
140 readonly environment?: NodeJS.ProcessEnv | undefined;
141 readonly forceKillAfter: Duration.Input;
142}) {
143 return ChildProcess.make(input.request.command, input.request.args ?? [], {
144 cwd: input.request.cwd ?? input.defaultCwd,
145 ...(input.environment === undefined ? {} : { env: input.environment }),
146 shell: input.shellCommands === true && (input.request.args?.length ?? 0) === 0,
147 forceKillAfter: input.forceKillAfter,
148 });
149}
150
151export const makeAcpClientTerminals = Effect.fn("makeAcpClientTerminals")(function* (
152 options: AcpClientTerminalsOptions,
153): Effect.fn.Return<AcpClientTerminals, never, ChildProcessSpawner.ChildProcessSpawner> {
154 const spawner = yield* ChildProcessSpawner.ChildProcessSpawner;
155 const forceKillAfter = options.forceKillAfter ?? "5 seconds";
156 const creationPermit = yield* Semaphore.make(1);
157 const terminals = new Map<string, ManagedAcpTerminal>();
158 let nextTerminalNumber = 1;
159
160 const enforceTotalOutputLimit = (): void => {
161 let excess =
162 Array.from(terminals.values()).reduce((total, terminal) => total + terminal.buffer.bytes, 0) -
163 MAX_RETAINED_OUTPUT_BYTES;
164 if (excess <= 0) return;
165 for (const terminal of terminals.values()) {
166 const before = terminal.buffer.bytes;
167 trimOutputStart(terminal.buffer, excess);
168 excess -= before - terminal.buffer.bytes;
169 if (excess <= 0) return;
170 }
171 };
172
173 const pruneReleasedSnapshots = (): void => {
174 const released = Array.from(terminals.values()).filter((terminal) => terminal.released);
175 let retainedBytes = released.reduce((total, terminal) => total + terminal.buffer.bytes, 0);
176 while (released.length > MAX_RETAINED_TERMINALS || retainedBytes > MAX_RETAINED_OUTPUT_BYTES) {
177 const terminal = released.shift();
178 if (terminal === undefined) break;
179 retainedBytes -= terminal.buffer.bytes;
180 terminals.delete(terminal.terminalId);
181 }
182 };
183
184 const requireTerminal = (
185 sessionId: string,
186 terminalId: string,
187 operation: string,
188 ): Effect.Effect<ManagedAcpTerminal, EffectAcpErrors.AcpRequestError> => {
189 const terminal = terminals.get(terminalId);
190 if (terminal === undefined || terminal.sessionId !== sessionId || terminal.released) {
191 return Effect.fail(
192 terminalRequestError(`ACP ${operation} received an unknown terminal ID '${terminalId}'.`),
193 );
194 }
195 return Effect.succeed(terminal);
196 };
197
198 const create: AcpClientTerminals["create"] = (request) =>
199 creationPermit.withPermit(
200 Effect.uninterruptibleMask((restore) =>
201 Effect.gen(function* () {
202 const liveCount = Array.from(terminals.values()).filter(
203 (terminal) => !terminal.released && terminal.exitStatus === undefined,
204 ).length;
205 const unreleasedCount = Array.from(terminals.values()).filter(
206 (terminal) => !terminal.released,
207 ).length;
208 if (unreleasedCount >= MAX_UNRELEASED_TERMINALS) {
209 return yield* terminalRequestError(
210 `ACP terminal/create exceeded the limit of ${MAX_UNRELEASED_TERMINALS} unreleased terminals.`,
211 );
212 }
213 if (liveCount >= MAX_LIVE_TERMINALS) {
214 return yield* terminalRequestError(
215 `ACP terminal/create exceeded the limit of ${MAX_LIVE_TERMINALS} concurrent terminals.`,
216 );
217 }
218 const requestEnvironment = Object.fromEntries(
219 (request.env ?? []).map((variable) => [variable.name, variable.value] as const),
220 );
221 const sessionEnvironment = options.environmentForSession?.(request.sessionId);
222 const environment =
223 options.environment === undefined &&
224 sessionEnvironment === undefined &&
225 (request.env?.length ?? 0) === 0
226 ? undefined
227 : { ...options.environment, ...sessionEnvironment, ...requestEnvironment };
228 const terminalScope = yield* Scope.make();
229 const child = yield* restore(
230 spawner
231 .spawn(
232 acpTerminalCommand({
233 request,
234 shellCommands: options.shellCommands,
235 defaultCwd: options.defaultCwd,
236 environment,
237 forceKillAfter,
238 }),
239 )
240 .pipe(
241 Effect.provideService(Scope.Scope, terminalScope),
242 Effect.mapError((cause) =>
243 terminalRequestError(
244 `Could not start terminal command '${request.command}'.`,
245 cause,
246 ),
247 ),
248 ),
249 ).pipe(
250 Effect.onExit((exit) =>
251 Exit.isFailure(exit) ? Scope.close(terminalScope, Exit.void) : Effect.void,
252 ),
253 );
254
255 const terminalId = `t3-term-${nextTerminalNumber}`;
256 nextTerminalNumber += 1;
257 const record: ManagedAcpTerminal = {
258 terminalId,
259 sessionId: request.sessionId,
260 commandLine: [request.command, ...(request.args ?? [])].join(" "),
261 buffer: {
262 chunks: [],
263 bytes: 0,
264 truncated: false,
265 limit: Math.min(
266 request.outputByteLimit ?? DEFAULT_OUTPUT_BYTE_LIMIT,
267 MAX_OUTPUT_BYTE_LIMIT,
268 ),
269 },
270 exit: yield* Deferred.make<EffectAcpSchema.WaitForTerminalExitResponse>(),
271 // The spawner only applies the command's forceKillAfter on scope close;
272 // an explicit kill must pass it or a SIGTERM-ignoring command never exits.
273 kill: child.kill({ forceKillAfter }).pipe(Effect.ignore),
274 dispose: Scope.close(terminalScope, Exit.void).pipe(Effect.ignore),
275 exitStatus: undefined,
276 released: false,
277 };
278 terminals.set(terminalId, record);
279
280 const pump = (stream: typeof child.stdout) =>
281 stream.pipe(
282 // Keep each pipe's UTF-8 decoder independent and preserve literal BOMs.
283 Stream.pipeThroughChannel(Channel.decodeText("utf-8", { ignoreBOM: true })),
284 Stream.encodeText,
285 Stream.runForEach((chunk) =>
286 Effect.sync(() => {
287 appendOutput(record.buffer, chunk);
288 enforceTotalOutputLimit();
289 }),
290 ),
291 Effect.ignore,
292 );
293 const stdoutPump = yield* Effect.forkDetach(pump(child.stdout));
294 const stderrPump = yield* Effect.forkDetach(pump(child.stderr));
295 yield* Effect.forkDetach(
296 child.exitCode.pipe(
297 Effect.match({
298 onFailure: (error) => ({ exitCode: null, signal: terminalExitSignal(error) }),
299 onSuccess: (code) => ({ exitCode: Number(code), signal: null }),
300 }),
301 Effect.flatMap((status) =>
302 Effect.all([Fiber.join(stdoutPump), Fiber.join(stderrPump)], {
303 discard: true,
304 }).pipe(
305 // A background child can inherit the pipes and hold them open indefinitely.
306 Effect.timeoutOption(OUTPUT_DRAIN_GRACE),
307 Effect.andThen(
308 Effect.sync(() => {
309 record.exitStatus = status;
310 }),
311 ),
312 Effect.andThen(Deferred.succeed(record.exit, status)),
313 ),
314 ),
315 ),
316 );
317 return { terminalId };
318 }),
319 ),
320 );
321
322 const output: AcpClientTerminals["output"] = (request) =>
323 requireTerminal(request.sessionId, request.terminalId, "terminal/output").pipe(
324 Effect.map((terminal) => ({
325 output: readOutput(terminal.buffer),
326 truncated: terminal.buffer.truncated,
327 ...(terminal.exitStatus === undefined ? {} : { exitStatus: terminal.exitStatus }),
328 })),
329 );
330
331 const waitForExit: AcpClientTerminals["waitForExit"] = (request) =>
332 requireTerminal(request.sessionId, request.terminalId, "terminal/wait_for_exit").pipe(
333 Effect.flatMap((terminal) => Deferred.await(terminal.exit)),
334 );
335
336 const kill: AcpClientTerminals["kill"] = (request) =>
337 requireTerminal(request.sessionId, request.terminalId, "terminal/kill").pipe(
338 Effect.flatMap((terminal) => terminal.kill),
339 Effect.as({}),
340 );
341
342 const release: AcpClientTerminals["release"] = (request) =>
343 Effect.uninterruptible(
344 requireTerminal(request.sessionId, request.terminalId, "terminal/release").pipe(
345 Effect.flatMap((terminal) =>
346 Effect.sync(() => {
347 terminal.released = true;
348 }).pipe(
349 Effect.andThen(terminal.dispose),
350 Effect.andThen(Deferred.await(terminal.exit)),
351 Effect.andThen(Effect.sync(pruneReleasedSnapshots)),
352 ),
353 ),
354 Effect.as({}),
355 ),
356 );
357
358 const readOutputSnapshot: AcpClientTerminals["readOutputSnapshot"] = (sessionId, terminalId) => {
359 const terminal = terminals.get(terminalId);
360 if (terminal === undefined || terminal.sessionId !== sessionId) return undefined;
361 return {
362 output: readOutput(terminal.buffer),
363 truncated: terminal.buffer.truncated,
364 exitStatus: terminal.exitStatus,
365 };
366 };
367
368 const readCommandLine: AcpClientTerminals["readCommandLine"] = (terminalId) =>
369 terminals.get(terminalId)?.commandLine;
370
371 const disposeAll: AcpClientTerminals["disposeAll"] = Effect.uninterruptible(
372 Effect.suspend(() =>
373 Effect.forEach(
374 Array.from(terminals.values()),
375 (terminal) =>
376 Effect.sync(() => {
377 terminal.released = true;
378 }).pipe(Effect.andThen(terminal.dispose), Effect.andThen(Deferred.await(terminal.exit))),
379 { concurrency: "unbounded", discard: true },
380 ).pipe(Effect.andThen(Effect.sync(pruneReleasedSnapshots))),
381 ),
382 );
383
384 const logged =
385 <Request, Response>(
386 operation: string,
387 handler: (request: Request) => Effect.Effect<Response, EffectAcpErrors.AcpRequestError>,
388 ) =>
389 (request: Request) =>
390 handler(request).pipe(
391 Effect.tapError((error) =>
392 Effect.logWarning("ACP client terminal operation failed", {
393 operation,
394 detail: error.message,
395 }),
396 ),
397 );
398
399 return {
400 create: logged("terminal/create", create),
401 output: logged("terminal/output", output),
402 waitForExit: logged("terminal/wait_for_exit", waitForExit),
403 kill: logged("terminal/kill", kill),
404 release: logged("terminal/release", release),
405 readOutputSnapshot,
406 readCommandLine,
407 disposeAll,
408 };
409});
410
411export function resolveEmbeddedTerminalContent(
412 notification: EffectAcpSchema.SessionNotification,
413 readOutputSnapshot: AcpClientTerminals["readOutputSnapshot"],
414): EffectAcpSchema.SessionNotification {
415 const update = notification.update;
416 if (update.sessionUpdate !== "tool_call" && update.sessionUpdate !== "tool_call_update") {
417 return notification;
418 }
419 const content = update.content;
420 if (!content || !content.some((entry) => entry.type === "terminal")) {
421 return notification;
422 }
423 const mapped = content.map((entry): EffectAcpSchema.ToolCallContent => {
424 if (entry.type !== "terminal") return entry;
425 const snapshot = readOutputSnapshot(notification.sessionId, entry.terminalId);
426 return {
427 type: "content",
428 content: {
429 type: "text",
430 text: snapshot === undefined ? `[terminal ${entry.terminalId}]` : snapshot.output,
431 },
432 };
433 });
434 return { ...notification, update: { ...update, content: mapped } };
435}