diff --git a/packages/provider-acp/src/server/clientTerminals.test.ts b/packages/provider-acp/src/server/clientTerminals.test.ts index 791bbc22a6a03..5d95b1a17327b 100644 --- a/packages/provider-acp/src/server/clientTerminals.test.ts +++ b/packages/provider-acp/src/server/clientTerminals.test.ts @@ -9,6 +9,8 @@ import * as Result from "effect/Result"; import * as Stream from "effect/Stream"; import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner"; import * as HostProcess from "@t3tools/shared/HostProcess"; +import * as NodeFS from "effect/FileSystem"; +import * as NodePath from "effect/Path"; import { makeAcpClientTerminals, @@ -378,6 +380,99 @@ 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* NodeFS.FileSystem; + const path = yield* NodePath.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.try(() => process.kill(pid)).pipe(Effect.ignore); + } + }); + const created = yield* terminals.create({ + sessionId: "session", + command: process.execPath, + args: [daemonScript], + env: [{ name: "DAEMON_PID_FILE", value: pidFile }], + }); + const outcome = 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(); + return true; + }).pipe(Effect.result, Effect.ensuring(cleanup)); + expect(Result.isSuccess(outcome) && outcome.success === true).toBe(true); + }).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 2c3278d3c144d..86ce033d4ce2b 100644 --- a/packages/provider-acp/src/server/clientTerminals.ts +++ b/packages/provider-acp/src/server/clientTerminals.ts @@ -18,6 +18,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( @@ -285,8 +288,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({ @@ -294,9 +297,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;