fix(acp): terminal release no longer wedges behind a daemonized grandchild

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<void>();
+      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<ChildProcessSpawner.ChildProcessHandle>();
+        const stdoutStopped = yield* Deferred.make<void>();
+        const stderrStopped = yield* Deferred.make<void>();
+        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", () => {

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,18 +292,19 @@ 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({
                 onFailure: (error) => ({ exitCode: null, signal: terminalExitSignal(error) }),
                 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))),
     ),
   );