From 0d867826170b4d5e21a7439bcd145d62aced5446 Mon Sep 17 00:00:00 2001 From: rishabhx29 Date: Tue, 3 Nov 2026 23:58:25 +0530 Subject: [PATCH 1/3] fix(acp): terminal release no longer wedges behind a daemonized grandchild The exit watcher joined the stdout and stderr pumps before reporting the exit status, and release waited on that report uninterruptibly. A daemonized grandchild (a dev server, ssh -f) inherits the pipe write-ends and keeps them open past the direct child's exit, so the pumps never see EOF and both waitForExit and release wedge forever. Bound the drain wait with a 1-second grace: the direct child's exit status is reported and release proceeds while the detached holder keeps running. Real-environment verification: a terminal running a script that spawns a detached pipe-holding holder wedged waitForExit and release on main (holder confirmed alive, no return within 10s); with the bound, exit 0 is reported and release completes while the holder lives. The committed regression test reproduces this with real processes on the live clock. --- .../src/server/clientTerminals.test.ts | 63 +++++++++++++++++++ .../src/server/clientTerminals.ts | 4 ++ 2 files changed, 67 insertions(+) diff --git a/packages/provider-acp/src/server/clientTerminals.test.ts b/packages/provider-acp/src/server/clientTerminals.test.ts index 791bbc22a6a03..6486089d05dc7 100644 --- a/packages/provider-acp/src/server/clientTerminals.test.ts +++ b/packages/provider-acp/src/server/clientTerminals.test.ts @@ -3,12 +3,15 @@ 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 Option from "effect/Option"; import * as Sink from "effect/Sink"; import * as TestClock from "effect/testing/TestClock"; 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 +381,66 @@ 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("completes release when a daemonized grandchild holds the pipes", () => + withTerminals((terminals) => + Effect.gen(function* () { + 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: ["-e", daemonScript], + env: [{ name: "DAEMON_PID_FILE", value: pidFile }], + }); + const outcome = yield* Effect.gen(function* () { + yield* terminals.waitForExit({ + sessionId: "session", + terminalId: created.terminalId, + }); + yield* terminals.release({ + sessionId: "session", + terminalId: created.terminalId, + }); + return true; + }).pipe( + Effect.timeoutOption("10 seconds"), + Effect.map(Option.match({ onSome: () => true, onNone: () => false })), + Effect.result, + Effect.ensuring(cleanup), + ); + expect(Result.isSuccess(outcome) && outcome.success === true).toBe(true); + }), + ), + ); }); describe("resolveEmbeddedTerminalContent", () => { diff --git a/packages/provider-acp/src/server/clientTerminals.ts b/packages/provider-acp/src/server/clientTerminals.ts index 2c3278d3c144d..f82891f2f85e9 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( @@ -297,6 +300,7 @@ export const makeAcpClientTerminals = Effect.fn("makeAcpClientTerminals")(functi Effect.all([Fiber.join(stdoutPump), Fiber.join(stderrPump)], { discard: true, }).pipe( + Effect.timeoutOption(TERMINAL_OUTPUT_DRAIN_GRACE), Effect.andThen( Effect.sync(() => { record.exitStatus = status; From 2107feed2dd0c52ee5077a7b1147fc8d6d5e5289 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Tue, 3 Nov 2026 19:58:44 +0000 Subject: [PATCH 2/3] fix(acp): stop terminal output pumps after drain grace --- .../src/server/clientTerminals.test.ts | 125 ++++++++++-------- .../src/server/clientTerminals.ts | 5 + 2 files changed, 78 insertions(+), 52 deletions(-) diff --git a/packages/provider-acp/src/server/clientTerminals.test.ts b/packages/provider-acp/src/server/clientTerminals.test.ts index 6486089d05dc7..77ed9db55eea3 100644 --- a/packages/provider-acp/src/server/clientTerminals.test.ts +++ b/packages/provider-acp/src/server/clientTerminals.test.ts @@ -3,7 +3,6 @@ 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 Option from "effect/Option"; import * as Sink from "effect/Sink"; import * as TestClock from "effect/testing/TestClock"; import * as Result from "effect/Result"; @@ -385,61 +384,83 @@ describe("AcpClientTerminals", () => { // 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("completes release when a daemonized grandchild holds the pipes", () => - withTerminals((terminals) => - Effect.gen(function* () { - 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); - } + Effect.gen(function* () { + const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; + 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.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* () { + const exit = yield* terminals.waitForExit({ + sessionId: "session", + terminalId: created.terminalId, }); - const created = yield* terminals.create({ + expect(exit.exitCode).toBe(0); + expect(yield* Deferred.isDone(stdoutStopped)).toBe(true); + expect(yield* Deferred.isDone(stderrStopped)).toBe(true); + yield* terminals.release({ sessionId: "session", - command: process.execPath, - args: ["-e", daemonScript], - env: [{ name: "DAEMON_PID_FILE", value: pidFile }], + terminalId: created.terminalId, }); - const outcome = yield* Effect.gen(function* () { - yield* terminals.waitForExit({ - sessionId: "session", - terminalId: created.terminalId, - }); - yield* terminals.release({ - sessionId: "session", - terminalId: created.terminalId, - }); - return true; - }).pipe( - Effect.timeoutOption("10 seconds"), - Effect.map(Option.match({ onSome: () => true, onNone: () => false })), - Effect.result, - Effect.ensuring(cleanup), + expect(terminals.readOutputSnapshot("session", created.terminalId)?.output).toContain( + "daemonized", ); - expect(Result.isSuccess(outcome) && outcome.success === true).toBe(true); - }), - ), + 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), ); }); diff --git a/packages/provider-acp/src/server/clientTerminals.ts b/packages/provider-acp/src/server/clientTerminals.ts index f82891f2f85e9..8137a5015a5c9 100644 --- a/packages/provider-acp/src/server/clientTerminals.ts +++ b/packages/provider-acp/src/server/clientTerminals.ts @@ -301,6 +301,11 @@ export const makeAcpClientTerminals = Effect.fn("makeAcpClientTerminals")(functi discard: true, }).pipe( Effect.timeoutOption(TERMINAL_OUTPUT_DRAIN_GRACE), + Effect.ensuring( + Effect.all([Fiber.interrupt(stdoutPump), Fiber.interrupt(stderrPump)], { + discard: true, + }), + ), Effect.andThen( Effect.sync(() => { record.exitStatus = status; From e6f7fbbbfe703da055cecc15859bf332dd6ee2f6 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Tue, 3 Nov 2026 20:14:04 +0000 Subject: [PATCH 3/3] fix(acp): retain terminal output pumps until release --- .../src/server/clientTerminals.test.ts | 161 ++++++++++-------- .../src/server/clientTerminals.ts | 11 +- 2 files changed, 89 insertions(+), 83 deletions(-) diff --git a/packages/provider-acp/src/server/clientTerminals.test.ts b/packages/provider-acp/src/server/clientTerminals.test.ts index 77ed9db55eea3..5d95b1a17327b 100644 --- a/packages/provider-acp/src/server/clientTerminals.test.ts +++ b/packages/provider-acp/src/server/clientTerminals.test.ts @@ -383,84 +383,95 @@ 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("completes release when a daemonized grandchild holds the pipes", () => - Effect.gen(function* () { - const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; - 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.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* () { - const exit = yield* terminals.waitForExit({ - sessionId: "session", - terminalId: created.terminalId, + 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); + } }); - expect(exit.exitCode).toBe(0); - expect(yield* Deferred.isDone(stdoutStopped)).toBe(true); - expect(yield* Deferred.isDone(stderrStopped)).toBe(true); - yield* terminals.release({ + const created = yield* terminals.create({ sessionId: "session", - terminalId: created.terminalId, + command: process.execPath, + args: [daemonScript], + env: [{ name: "DAEMON_PID_FILE", value: pidFile }], }); - 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), + 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), ); }); diff --git a/packages/provider-acp/src/server/clientTerminals.ts b/packages/provider-acp/src/server/clientTerminals.ts index 8137a5015a5c9..86ce033d4ce2b 100644 --- a/packages/provider-acp/src/server/clientTerminals.ts +++ b/packages/provider-acp/src/server/clientTerminals.ts @@ -288,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({ @@ -297,15 +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.ensuring( - Effect.all([Fiber.interrupt(stdoutPump), Fiber.interrupt(stderrPump)], { - discard: true, - }), - ), Effect.andThen( Effect.sync(() => { record.exitStatus = status;