Open History to see the last commit
1import * as NodeServices from "@effect/platform-node/NodeServices";
2import { describe, expect, it } from "@effect/vitest";
3import * as Deferred from "effect/Deferred";
4import * as Effect from "effect/Effect";
5import * as Fiber from "effect/Fiber";
6import * as Sink from "effect/Sink";
7import * as TestClock from "effect/testing/TestClock";
8import * as Result from "effect/Result";
9import * as Stream from "effect/Stream";
10import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner";
11import * as HostProcess from "@t3tools/shared/HostProcess";
12
13import {
14 makeAcpClientTerminals,
15 resolveEmbeddedTerminalContent,
16 type AcpClientTerminals,
17} from "./clientTerminals.ts";
18
19const withTerminals = <A, E>(use: (terminals: AcpClientTerminals) => Effect.Effect<A, E>) =>
20 Effect.gen(function* () {
21 const terminals = yield* makeAcpClientTerminals({
22 defaultCwd: process.cwd(),
23 });
24 return yield* use(terminals).pipe(Effect.ensuring(terminals.disposeAll));
25 }).pipe(Effect.provide(NodeServices.layer), Effect.scoped);
26
27describe("AcpClientTerminals", () => {
28 it.live("runs a command, buffers output, and reports the exit status", () =>
29 withTerminals((terminals) =>
30 Effect.gen(function* () {
31 const created = yield* terminals.create({
32 sessionId: "session",
33 command: process.execPath,
34 args: ["-e", "console.log('hello from acp'); process.exit(3);"],
35 });
36
37 const exit = yield* terminals.waitForExit({
38 sessionId: "session",
39 terminalId: created.terminalId,
40 });
41 expect(exit.exitCode).toBe(3);
42
43 const output = yield* terminals.output({
44 sessionId: "session",
45 terminalId: created.terminalId,
46 });
47 expect(output.output).toContain("hello from acp");
48 expect(output.truncated).toBe(false);
49 expect(output.exitStatus?.exitCode).toBe(3);
50 }),
51 ),
52 );
53
54 it.effect("preserves UTF-8 when stderr interleaves a split stdout character", () =>
55 Effect.gen(function* () {
56 const stdoutStarted = yield* Deferred.make<void>();
57 const stderrWritten = yield* Deferred.make<void>();
58 const child = ChildProcessSpawner.makeHandle({
59 pid: ChildProcessSpawner.ProcessId(1),
60 exitCode: Effect.succeed(ChildProcessSpawner.ExitCode(0)),
61 isRunning: Effect.succeed(false),
62 kill: () => Effect.void,
63 stdin: Sink.drain,
64 stdout: Stream.make(new Uint8Array([0xe2])).pipe(
65 Stream.concat(Stream.fromEffectDrain(Deferred.succeed(stdoutStarted, undefined))),
66 Stream.concat(Stream.fromEffectDrain(Deferred.await(stderrWritten))),
67 Stream.concat(Stream.make(new Uint8Array([0x82, 0xac]))),
68 ),
69 stderr: Stream.fromEffectDrain(Deferred.await(stdoutStarted)).pipe(
70 Stream.concat(Stream.make(new Uint8Array([0x58]))),
71 Stream.concat(Stream.fromEffectDrain(Deferred.succeed(stderrWritten, undefined))),
72 ),
73 all: Stream.empty,
74 getInputFd: () => Sink.drain,
75 getOutputFd: () => Stream.empty,
76 unref: Effect.succeed(Effect.void),
77 });
78 const terminals = yield* makeAcpClientTerminals({ defaultCwd: process.cwd() }).pipe(
79 Effect.provideService(
80 ChildProcessSpawner.ChildProcessSpawner,
81 ChildProcessSpawner.make(() => Effect.succeed(child)),
82 ),
83 );
84 yield* Effect.addFinalizer(() => terminals.disposeAll);
85 const terminal = yield* terminals.create({ sessionId: "session", command: "fixture" });
86 const request = { sessionId: "session", ...terminal };
87 yield* terminals.waitForExit(request);
88
89 const output = yield* terminals.output(request);
90 expect(output.output).toContain("€");
91 expect(output.output).toContain("X");
92 expect(output.output).not.toContain("\uFFFD");
93 expect(output.truncated).toBe(false);
94 expect(terminals.readOutputSnapshot("session", terminal.terminalId)?.output).toBe(
95 output.output,
96 );
97 }).pipe(Effect.scoped),
98 );
99
100 it.live("resolves terminal environment from the requesting ACP session", () =>
101 Effect.gen(function* () {
102 const terminals = yield* makeAcpClientTerminals({
103 defaultCwd: process.cwd(),
104 environmentForSession: (sessionId) => ({
105 T3_TEST_ACP_SESSION_TOKEN: sessionId === "session-a" ? "token-a" : "token-b",
106 }),
107 });
108 yield* Effect.addFinalizer(() => terminals.disposeAll);
109
110 for (const [sessionId, expected] of [
111 ["session-a", "token-a"],
112 ["session-b", "token-b"],
113 ] as const) {
114 const created = yield* terminals.create({
115 sessionId,
116 command: process.execPath,
117 args: ["-e", "process.stdout.write(process.env.T3_TEST_ACP_SESSION_TOKEN ?? '')"],
118 });
119 yield* terminals.waitForExit({ sessionId, terminalId: created.terminalId });
120 const output = yield* terminals.output({ sessionId, terminalId: created.terminalId });
121 expect(output.output).toBe(expected);
122 }
123 }).pipe(Effect.provide(NodeServices.layer), Effect.scoped),
124 );
125
126 it.live("runs Devin shell source with arguments, pipes, and session fallback variables", () =>
127 Effect.gen(function* () {
128 const terminals = yield* makeAcpClientTerminals({
129 defaultCwd: process.cwd(),
130 shellCommands: true,
131 environmentForSession: () => ({ T3_ACP_MCP_NODE: "mailbox-probe" }),
132 });
133 yield* Effect.addFinalizer(() => terminals.disposeAll);
134 const terminal = yield* terminals.create({
135 sessionId: "devin",
136 command: 'printf "%s\\n" "$T3_ACP_MCP_NODE" | tr a-z A-Z',
137 });
138 const exit = yield* terminals.waitForExit({ sessionId: "devin", ...terminal });
139 const output = yield* terminals.output({ sessionId: "devin", ...terminal });
140 expect(exit.exitCode).toBe(0);
141 expect(output.output.trim()).toBe("MAILBOX-PROBE");
142 }).pipe(Effect.provide(NodeServices.layer), Effect.scoped),
143 );
144
145 it.live.each([
146 { text: "a".repeat(64) + "TAIL", limit: 16 },
147 { text: "€".repeat(16) + "TAIL", limit: 6 },
148 ])("truncates buffered output from the beginning at the $limit byte limit", ({ text, limit }) =>
149 withTerminals((terminals) =>
150 Effect.gen(function* () {
151 const created = yield* terminals.create({
152 sessionId: "session",
153 command: process.execPath,
154 args: ["-e", `process.stdout.write(${JSON.stringify(text)});`],
155 outputByteLimit: limit,
156 });
157 yield* terminals.waitForExit({ sessionId: "session", terminalId: created.terminalId });
158
159 const output = yield* terminals.output({
160 sessionId: "session",
161 terminalId: created.terminalId,
162 });
163 expect(output.truncated).toBe(true);
164 expect(output.output.endsWith("TAIL")).toBe(true);
165 expect(Buffer.byteLength(output.output)).toBeLessThanOrEqual(limit);
166 expect(output.output).not.toContain("\uFFFD");
167 }),
168 ),
169 );
170
171 it.live("kills long-running commands and rejects released terminal handles", () =>
172 withTerminals((terminals) =>
173 Effect.gen(function* () {
174 const created = yield* terminals.create({
175 sessionId: "session",
176 command: process.execPath,
177 args: ["-e", "setInterval(() => {}, 1000);"],
178 });
179 yield* terminals.kill({ sessionId: "session", terminalId: created.terminalId });
180 const exit = yield* terminals.waitForExit({
181 sessionId: "session",
182 terminalId: created.terminalId,
183 });
184 expect(exit.exitCode === 0 ? null : exit.exitCode).not.toBe(0);
185 // Windows has no signals: a killed process only reports an exit code.
186 if (yield* HostProcess.isWindows) {
187 expect(exit.signal).toBeNull();
188 } else {
189 expect(exit.signal).toBe("SIGTERM");
190 }
191
192 yield* terminals.release({ sessionId: "session", terminalId: created.terminalId });
193 const rejected = yield* Effect.flip(
194 terminals.output({ sessionId: "session", terminalId: created.terminalId }),
195 );
196 expect(rejected.message).toContain("unknown terminal ID");
197
198 expect(terminals.readOutputSnapshot("session", created.terminalId)).toBeDefined();
199 }),
200 ),
201 );
202
203 it.effect("answers wait_for_exit when a background child keeps the output pipes open", () =>
204 Effect.gen(function* () {
205 // Models `sh -c "setsid server & echo done"`: sh exits, the orphan inherits stdout/stderr.
206 const orphanExited = yield* Deferred.make<void>();
207 const heldOpen = Stream.fromEffectDrain(Deferred.await(orphanExited));
208 const child = ChildProcessSpawner.makeHandle({
209 pid: ChildProcessSpawner.ProcessId(1),
210 exitCode: Effect.succeed(ChildProcessSpawner.ExitCode(0)),
211 isRunning: Effect.succeed(false),
212 kill: () => Effect.void,
213 stdin: Sink.drain,
214 stdout: Stream.make(new TextEncoder().encode("done\n")).pipe(Stream.concat(heldOpen)),
215 stderr: heldOpen,
216 all: Stream.empty,
217 getInputFd: () => Sink.drain,
218 getOutputFd: () => Stream.empty,
219 unref: Effect.succeed(Effect.void),
220 });
221 const terminals = yield* makeAcpClientTerminals({ defaultCwd: process.cwd() }).pipe(
222 Effect.provideService(
223 ChildProcessSpawner.ChildProcessSpawner,
224 ChildProcessSpawner.make(() => Effect.succeed(child)),
225 ),
226 );
227 yield* Effect.addFinalizer(() =>
228 Deferred.succeed(orphanExited, undefined).pipe(Effect.andThen(terminals.disposeAll)),
229 );
230 const terminal = yield* terminals.create({ sessionId: "session", command: "fixture" });
231 const request = { sessionId: "session", ...terminal };
232 const waiting = yield* terminals.waitForExit(request).pipe(Effect.forkChild);
233
234 yield* TestClock.adjust("5 seconds");
235 expect(waiting.pollUnsafe()).toBeDefined();
236 expect(yield* Fiber.join(waiting)).toEqual({ exitCode: 0, signal: null });
237 const output = yield* terminals.output(request);
238 expect(output.output).toBe("done\n");
239 expect(output.exitStatus).toEqual({ exitCode: 0, signal: null });
240 }).pipe(Effect.scoped),
241 );
242
243 it.effect.each([
244 { deadline: "default", forceKillAfter: undefined, deadlineMillis: 5000 },
245 { deadline: "configured", forceKillAfter: "250 millis" as const, deadlineMillis: 250 },
246 ])(
247 "waits for the $deadline deadline before force-killing a SIGTERM-ignoring terminal",
248 ({ forceKillAfter, deadlineMillis }) =>
249 Effect.gen(function* () {
250 const exited = yield* Deferred.make<ChildProcessSpawner.ExitCode>();
251 const child = ChildProcessSpawner.makeHandle({
252 pid: ChildProcessSpawner.ProcessId(1),
253 exitCode: Deferred.await(exited),
254 isRunning: Deferred.isDone(exited).pipe(Effect.map((done) => !done)),
255 // Model a process that only exits when the spawner escalates after its deadline.
256 kill: (options) =>
257 options?.forceKillAfter === undefined
258 ? Deferred.await(exited).pipe(Effect.asVoid)
259 : Effect.sleep(options.forceKillAfter).pipe(
260 Effect.andThen(Deferred.succeed(exited, ChildProcessSpawner.ExitCode(137))),
261 Effect.asVoid,
262 ),
263 stdin: Sink.drain,
264 stdout: Stream.empty,
265 stderr: Stream.empty,
266 all: Stream.empty,
267 getInputFd: () => Sink.drain,
268 getOutputFd: () => Stream.empty,
269 unref: Effect.succeed(Effect.void),
270 });
271 const terminals = yield* makeAcpClientTerminals({
272 defaultCwd: process.cwd(),
273 ...(forceKillAfter === undefined ? {} : { forceKillAfter }),
274 }).pipe(
275 Effect.provideService(
276 ChildProcessSpawner.ChildProcessSpawner,
277 ChildProcessSpawner.make(() => Effect.succeed(child)),
278 ),
279 );
280 yield* Effect.addFinalizer(() =>
281 Deferred.succeed(exited, ChildProcessSpawner.ExitCode(137)).pipe(
282 Effect.andThen(terminals.disposeAll),
283 ),
284 );
285 const terminal = yield* terminals.create({
286 sessionId: "session",
287 command: "ignores-SIGTERM",
288 });
289 const request = { sessionId: "session", ...terminal };
290 const killing = yield* terminals.kill(request).pipe(Effect.forkChild);
291 const waiting = yield* terminals.waitForExit(request).pipe(Effect.forkChild);
292
293 yield* TestClock.adjust(deadlineMillis - 1);
294 expect(yield* child.isRunning).toBe(true);
295 expect(killing.pollUnsafe()).toBeUndefined();
296 expect(waiting.pollUnsafe()).toBeUndefined();
297 yield* TestClock.adjust("1 millis");
298 expect(yield* child.isRunning).toBe(false);
299 yield* Fiber.join(killing);
300 expect(yield* Fiber.join(waiting)).toEqual({ exitCode: 137, signal: null });
301 }).pipe(Effect.scoped),
302 );
303
304 it.live("force-kills commands that ignore SIGTERM and resolves wait_for_exit", () =>
305 Effect.gen(function* () {
306 const spawner = yield* ChildProcessSpawner.ChildProcessSpawner;
307 const ready = yield* Deferred.make<void>();
308 const spawned = yield* Deferred.make<ChildProcessSpawner.ChildProcessHandle>();
309 let stdout = "";
310 const terminals = yield* makeAcpClientTerminals({
311 defaultCwd: process.cwd(),
312 forceKillAfter: "50 millis",
313 }).pipe(
314 Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, {
315 ...spawner,
316 spawn: (command) =>
317 spawner.spawn(command).pipe(
318 Effect.tap((child) => Deferred.succeed(spawned, child)),
319 Effect.map((child) => ({
320 ...child,
321 stdout: child.stdout.pipe(
322 Stream.tap((chunk) => {
323 stdout += Buffer.from(chunk).toString("utf8");
324 return stdout.includes("ready")
325 ? Deferred.succeed(ready, undefined)
326 : Effect.void;
327 }),
328 ),
329 })),
330 ),
331 }),
332 );
333 yield* Effect.addFinalizer(() => terminals.disposeAll);
334 const terminal = yield* terminals.create({
335 sessionId: "session",
336 command: process.execPath,
337 args: [
338 "-e",
339 "process.on('SIGTERM', () => {}); console.log('ready'); setInterval(() => {}, 1000);",
340 ],
341 });
342 const child = yield* Deferred.await(spawned);
343 // Reap only this test's captured child, including when the regression fails.
344 yield* Effect.addFinalizer(() => child.kill({ killSignal: "SIGKILL" }).pipe(Effect.ignore));
345 yield* Deferred.await(ready);
346
347 const request = { sessionId: "session", ...terminal };
348 yield* terminals.kill(request);
349 const exit = yield* terminals.waitForExit(request);
350 // Windows has no signals: the command is terminated without a signal name.
351 if (!(yield* HostProcess.isWindows)) {
352 expect(exit).toEqual({ exitCode: null, signal: "SIGKILL" });
353 }
354 expect(yield* child.isRunning).toBe(false);
355 }).pipe(Effect.provide(NodeServices.layer), Effect.scoped),
356 );
357
358 it.live("reserves terminal capacity across concurrent creates", () =>
359 withTerminals((terminals) =>
360 Effect.gen(function* () {
361 const results = yield* Effect.all(
362 Array.from({ length: 17 }, () =>
363 terminals
364 .create({
365 sessionId: "session",
366 command: process.execPath,
367 args: ["-e", "setInterval(() => {}, 1000);"],
368 })
369 .pipe(Effect.result),
370 ),
371 { concurrency: "unbounded" },
372 );
373
374 expect(results.filter(Result.isSuccess)).toHaveLength(16);
375 expect(results.filter(Result.isFailure)).toHaveLength(1);
376 }),
377 ),
378 );
379
380 it.live("bounds unreleased terminal handles without invalidating existing ids", () =>
381 withTerminals((terminals) =>
382 Effect.gen(function* () {
383 const terminalIds: string[] = [];
384 for (let index = 0; index < 64; index += 1) {
385 const created = yield* terminals.create({
386 sessionId: "session",
387 command: process.execPath,
388 args: ["-e", `process.stdout.write('${index}');`],
389 });
390 yield* terminals.waitForExit({
391 sessionId: "session",
392 terminalId: created.terminalId,
393 });
394 terminalIds.push(created.terminalId);
395 }
396
397 const rejected = yield* terminals
398 .create({
399 sessionId: "session",
400 command: process.execPath,
401 args: ["-e", ""],
402 })
403 .pipe(Effect.result);
404 expect(Result.isFailure(rejected)).toBe(true);
405 expect(terminals.readOutputSnapshot("session", terminalIds[0]!)).toBeDefined();
406 expect(terminals.readOutputSnapshot("session", terminalIds.at(-1)!)).toBeDefined();
407 }),
408 ),
409 );
410
411 it.live("bounds aggregate output retained across unreleased terminals", () =>
412 withTerminals((terminals) =>
413 Effect.gen(function* () {
414 const terminalIds: string[] = [];
415 for (let index = 0; index < 3; index += 1) {
416 const created = yield* terminals.create({
417 sessionId: "session",
418 command: process.execPath,
419 args: ["-e", "process.stdout.write('x'.repeat(8 * 1024 * 1024));"],
420 outputByteLimit: 8 * 1024 * 1024,
421 });
422 yield* terminals.waitForExit({
423 sessionId: "session",
424 terminalId: created.terminalId,
425 });
426 terminalIds.push(created.terminalId);
427 }
428
429 const snapshots = terminalIds.map((terminalId) =>
430 terminals.readOutputSnapshot("session", terminalId)!,
431 );
432 const retainedBytes = snapshots.reduce(
433 (total, snapshot) => total + Buffer.byteLength(snapshot.output),
434 0,
435 );
436 expect(retainedBytes).toBeLessThanOrEqual(16 * 1024 * 1024);
437 expect(snapshots.some((snapshot) => snapshot.truncated)).toBe(true);
438 }),
439 ),
440 );
441
442 it.live("rejects terminal handles from a different ACP session", () =>
443 withTerminals((terminals) =>
444 Effect.gen(function* () {
445 const created = yield* terminals.create({
446 sessionId: "owner-session",
447 command: process.execPath,
448 args: ["-e", "process.stdout.write('owned');"],
449 });
450 const foreignRequest = {
451 sessionId: "foreign-session",
452 terminalId: created.terminalId,
453 };
454 const results = yield* Effect.forEach(
455 [
456 terminals.output(foreignRequest),
457 terminals.waitForExit(foreignRequest),
458 terminals.kill(foreignRequest),
459 terminals.release(foreignRequest),
460 ],
461 (effect) => effect.pipe(Effect.result),
462 { concurrency: "unbounded" },
463 );
464
465 expect(results.every(Result.isFailure)).toBe(true);
466 expect(terminals.readOutputSnapshot("foreign-session", created.terminalId)).toBeUndefined();
467 expect(terminals.readOutputSnapshot("owner-session", created.terminalId)).toBeDefined();
468 }),
469 ),
470 );
471});
472
473describe("resolveEmbeddedTerminalContent", () => {
474 it("rewrites terminal content into text from the buffered snapshot", () => {
475 const notification = {
476 sessionId: "session",
477 update: {
478 sessionUpdate: "tool_call_update" as const,
479 toolCallId: "call-1",
480 content: [
481 { type: "content" as const, content: { type: "text" as const, text: "before" } },
482 { type: "terminal" as const, terminalId: "t3-term-1" },
483 { type: "terminal" as const, terminalId: "t3-term-unknown" },
484 ],
485 },
486 };
487
488 const resolved = resolveEmbeddedTerminalContent(notification, (sessionId, terminalId) =>
489 sessionId === "session" && terminalId === "t3-term-1"
490 ? { output: "compiled 3 files", truncated: false, exitStatus: undefined }
491 : undefined,
492 );
493
494 expect(resolved.update).toMatchObject({
495 content: [
496 { type: "content", content: { type: "text", text: "before" } },
497 { type: "content", content: { type: "text", text: "compiled 3 files" } },
498 { type: "content", content: { type: "text", text: "[terminal t3-term-unknown]" } },
499 ],
500 });
501 });
502});