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