diff --git a/packages/provider-acp/src/server/clientTerminals.test.ts b/packages/provider-acp/src/server/clientTerminals.test.ts index 5955690dc6b70..525840bece427 100644 --- a/packages/provider-acp/src/server/clientTerminals.test.ts +++ b/packages/provider-acp/src/server/clientTerminals.test.ts @@ -3,6 +3,8 @@ import { describe, expect, it } from "@effect/vitest"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Fiber from "effect/Fiber"; +import * as FileSystem from "effect/FileSystem"; +import * as Path from "effect/Path"; import * as Sink from "effect/Sink"; import * as TestClock from "effect/testing/TestClock"; import * as Result from "effect/Result"; @@ -200,6 +202,41 @@ describe("AcpClientTerminals", () => { ), ); + it.live("keeps output a running command writes while release stops it", () => + Effect.gen(function* () { + // Windows terminates without delivering SIGTERM to a handler. + if (yield* HostProcess.isWindows) return; + const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const handlerInstalled = yield* Deferred.make(); + const terminals = yield* makeAcpClientTerminals({ defaultCwd: process.cwd() }).pipe( + Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, { + ...spawner, + spawn: (command) => + spawner.spawn(command).pipe( + Effect.map((child) => ({ + ...child, + stdout: child.stdout.pipe( + Stream.tap(() => Deferred.succeed(handlerInstalled, undefined)), + ), + })), + ), + }), + ); + yield* Effect.addFinalizer(() => terminals.disposeAll); + const created = yield* terminals.create({ + sessionId: "session", + command: process.execPath, + args: [ + "-e", + "process.on('SIGTERM', () => { process.stdout.write('bye'); process.exit(0); }); console.log('started'); setInterval(() => {}, 1000);", + ], + }); + yield* Deferred.await(handlerInstalled); + yield* terminals.release({ sessionId: "session", terminalId: created.terminalId }); + expect(terminals.readOutputSnapshot("session", created.terminalId)?.output).toContain("bye"); + }).pipe(Effect.provide(NodeServices.layer), Effect.scoped), + ); + it.effect.each([ { deadline: "default", forceKillAfter: undefined, deadlineMillis: 5000 }, { deadline: "configured", forceKillAfter: "250 millis" as const, deadlineMillis: 250 }, @@ -214,12 +251,15 @@ describe("AcpClientTerminals", () => { isRunning: Deferred.isDone(exited).pipe(Effect.map((done) => !done)), // Model a process that only exits when the spawner escalates after its deadline. kill: (options) => - options?.forceKillAfter === undefined - ? Deferred.await(exited).pipe(Effect.asVoid) - : Effect.sleep(options.forceKillAfter).pipe( - Effect.andThen(Deferred.succeed(exited, ChildProcessSpawner.ExitCode(137))), - Effect.asVoid, - ), + Effect.gen(function* () { + if (yield* Deferred.isDone(exited)) return; + if (options?.forceKillAfter === undefined) { + yield* Deferred.await(exited); + return; + } + yield* Effect.sleep(options.forceKillAfter); + yield* Deferred.succeed(exited, ChildProcessSpawner.ExitCode(137)); + }), stdin: Sink.drain, stdout: Stream.empty, stderr: Stream.empty, @@ -428,6 +468,104 @@ describe("AcpClientTerminals", () => { }), ), ); + + // A daemonized grandchild holds the pipe write-ends past the direct child's + // exit, so the output pumps never see EOF and release would wedge behind + // them. Real agent terminals hit this with dev servers and ssh -f. + it.live.each([true, false])( + "completes release with held pipes (wait before release: %s)", + (waitBeforeRelease) => + Effect.gen(function* () { + const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const spawned = yield* Deferred.make(); + const stdoutStopped = yield* Deferred.make(); + const stderrStopped = yield* Deferred.make(); + const terminals = yield* makeAcpClientTerminals({ defaultCwd: process.cwd() }).pipe( + Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, { + ...spawner, + spawn: (command) => + spawner.spawn(command).pipe( + Effect.tap((child) => Deferred.succeed(spawned, child)), + Effect.map((child) => ({ + ...child, + stdout: child.stdout.pipe( + Stream.ensuring(Deferred.succeed(stdoutStopped, undefined)), + ), + stderr: child.stderr.pipe( + Stream.ensuring(Deferred.succeed(stderrStopped, undefined)), + ), + })), + ), + }), + ); + yield* Effect.addFinalizer(() => terminals.disposeAll); + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const directory = yield* fs.makeTempDirectoryScoped({ prefix: "t3-acp-daemon-" }); + const pidFile = path.join(directory, "daemon.pid"); + const daemonScript = path.join(directory, "daemon.js"); + yield* fs.writeFileString( + daemonScript, + [ + "const cp = require('child_process');", + "const holder = cp.spawn(process.execPath, ['-e', 'setInterval(() => { console.log(String(process.pid)); }, 250)'], {", + " detached: true,", + " windowsHide: true,", + " stdio: ['ignore', process.stdout, process.stderr],", + "});", + "holder.unref();", + "require('fs').writeFileSync(process.env.DAEMON_PID_FILE, String(holder.pid));", + "console.log('daemonized');", + ].join("\n"), + ); + // Kill the holder last so a failed assertion cannot wedge the suite on + // the same held pipes that wedge the real caller. + const cleanup = Effect.gen(function* () { + const text = yield* fs.readFileString(pidFile).pipe(Effect.orElseSucceed(() => "")); + const pid = Number.parseInt(text, 10); + if (Number.isInteger(pid)) { + yield* Effect.sync(() => { + try { + process.kill(pid); + } catch { + // Already gone. + } + }); + } + }); + const created = yield* terminals.create({ + sessionId: "session", + command: process.execPath, + args: [daemonScript], + env: [{ name: "DAEMON_PID_FILE", value: pidFile }], + }); + yield* Effect.gen(function* () { + if (waitBeforeRelease) { + const exit = yield* terminals.waitForExit({ + sessionId: "session", + terminalId: created.terminalId, + }); + expect(exit.exitCode).toBe(0); + expect(yield* Deferred.isDone(stdoutStopped)).toBe(false); + expect(yield* Deferred.isDone(stderrStopped)).toBe(false); + } else { + const child = yield* Deferred.await(spawned); + yield* child.exitCode; + } + yield* terminals.release({ + sessionId: "session", + terminalId: created.terminalId, + }); + expect(yield* Deferred.isDone(stdoutStopped)).toBe(true); + expect(yield* Deferred.isDone(stderrStopped)).toBe(true); + expect(terminals.readOutputSnapshot("session", created.terminalId)?.output).toContain( + "daemonized", + ); + const pid = Number.parseInt(yield* fs.readFileString(pidFile), 10); + expect(() => process.kill(pid, 0)).not.toThrow(); + }).pipe(Effect.ensuring(cleanup)); + }).pipe(Effect.provide(NodeServices.layer), Effect.scoped), + ); }); describe("resolveEmbeddedTerminalContent", () => { diff --git a/packages/provider-acp/src/server/clientTerminals.ts b/packages/provider-acp/src/server/clientTerminals.ts index a62698f496351..2f20ad3a4548e 100644 --- a/packages/provider-acp/src/server/clientTerminals.ts +++ b/packages/provider-acp/src/server/clientTerminals.ts @@ -19,6 +19,9 @@ const MAX_RETAINED_TERMINALS = 32; const MAX_RETAINED_OUTPUT_BYTES = 16 * 1024 * 1024; const DEFAULT_OUTPUT_BYTE_LIMIT = 4 * 1024 * 1024; const MAX_OUTPUT_BYTE_LIMIT = 8 * 1024 * 1024; +// A daemonized grandchild can hold the pipe write-ends past the direct child's +// exit; the drain wait is bounded so release cannot wedge behind it. +const TERMINAL_OUTPUT_DRAIN_GRACE = Duration.seconds(1); const terminalRequestError = (message: string, cause?: unknown) => EffectAcpErrors.AcpRequestError.internalError( @@ -289,8 +292,8 @@ export const makeAcpClientTerminals = Effect.fn("makeAcpClientTerminals")(functi ), Effect.ignore, ); - const stdoutPump = yield* Effect.forkDetach(pump(child.stdout)); - const stderrPump = yield* Effect.forkDetach(pump(child.stderr)); + const stdoutPump = yield* pump(child.stdout).pipe(Effect.forkIn(terminalScope)); + const stderrPump = yield* pump(child.stderr).pipe(Effect.forkIn(terminalScope)); yield* Effect.forkDetach( child.exitCode.pipe( Effect.match({ @@ -298,9 +301,10 @@ export const makeAcpClientTerminals = Effect.fn("makeAcpClientTerminals")(functi onSuccess: (code) => ({ exitCode: Number(code), signal: null }), }), Effect.flatMap((status) => - Effect.all([Fiber.join(stdoutPump), Fiber.join(stderrPump)], { + Effect.all([Fiber.await(stdoutPump), Fiber.await(stderrPump)], { discard: true, }).pipe( + Effect.timeoutOption(TERMINAL_OUTPUT_DRAIN_GRACE), Effect.andThen( Effect.sync(() => { record.exitStatus = status; @@ -336,17 +340,22 @@ export const makeAcpClientTerminals = Effect.fn("makeAcpClientTerminals")(functi Effect.as({}), ); + // Kill and drain before closing the scope: closing it first would interrupt + // the output pumps before the command's shutdown output is read. + const releaseTerminal = (terminal: ManagedAcpTerminal) => + Effect.sync(() => { + terminal.released = true; + }).pipe( + Effect.andThen(terminal.kill), + Effect.andThen(Deferred.await(terminal.exit)), + Effect.andThen(terminal.dispose), + ); + const release: AcpClientTerminals["release"] = (request) => Effect.uninterruptible( requireTerminal(request.sessionId, request.terminalId, "terminal/release").pipe( Effect.flatMap((terminal) => - Effect.sync(() => { - terminal.released = true; - }).pipe( - Effect.andThen(terminal.dispose), - Effect.andThen(Deferred.await(terminal.exit)), - Effect.andThen(Effect.sync(pruneReleasedSnapshots)), - ), + releaseTerminal(terminal).pipe(Effect.andThen(Effect.sync(pruneReleasedSnapshots))), ), Effect.as({}), ), @@ -367,14 +376,10 @@ export const makeAcpClientTerminals = Effect.fn("makeAcpClientTerminals")(functi const disposeAll: AcpClientTerminals["disposeAll"] = Effect.uninterruptible( Effect.suspend(() => - Effect.forEach( - Array.from(terminals.values()), - (terminal) => - Effect.sync(() => { - terminal.released = true; - }).pipe(Effect.andThen(terminal.dispose), Effect.andThen(Deferred.await(terminal.exit))), - { concurrency: "unbounded", discard: true }, - ).pipe(Effect.andThen(Effect.sync(pruneReleasedSnapshots))), + Effect.forEach(Array.from(terminals.values()), releaseTerminal, { + concurrency: "unbounded", + discard: true, + }).pipe(Effect.andThen(Effect.sync(pruneReleasedSnapshots))), ), );