PTY: fix resource leaks and reader-isolate races (T-76)

NativePty.close() now awaits the reader-isolate spawn, kills the
child first to drive EOF on the master fd, awaits the isolate's
EOF acknowledgement, and only then closes the fd. Previously the
fd-close racing with the polling isolate left a window where the
fd number could be reused and the isolate would briefly target the
wrong file.

Both NativePty and PtySession now surface reader-isolate spawn
errors via the output stream's addError instead of silently
swallowing them.

PtySession.spawn closes the master fd on any post-receive failure,
closes parentSock in finally (was leaking on every spawn), and
kills the ptyc process if recvFd fails.

PtySession._recvFdAsync uses try/finally to close the ReceivePort
and kill the spawn isolate even when Isolate.spawn itself throws.

Co-Authored-By: Claude <noreply@anthropic.com>

lib/src/pty/native_pty.dart

@@ -90,6 +90,13 @@ class NativePty {
   final _out = StreamController<Uint8List>.broadcast();
   bool _dead = false;
 
+  /// Tracks the reader isolate's spawn — close() awaits this before
+  /// tearing down so we never race a still-spawning isolate.
+  Future<void>? _readerReady;
+  Isolate? _readerIsolate;
+  ReceivePort? _readerPort;
+  Completer<void>? _readerExited;
+
   NativePty._(this._fd, this.pid);
 
   /// Byte stream of data produced by the child.
@@ -222,13 +229,32 @@ class NativePty {
 
   // -- I/O ------------------------------------------------------------------
 
-  void _spawnReader() async {
+  void _spawnReader() {
+    _readerReady = _spawnReaderAsync();
+  }
+
+  Future<void> _spawnReaderAsync() async {
     final rp = ReceivePort();
-    await Isolate.spawn(_readLoop, (rp.sendPort, _fd));
+    _readerPort = rp;
+    _readerExited = Completer<void>();
+    try {
+      _readerIsolate = await Isolate.spawn(_readLoop, (rp.sendPort, _fd));
+    } catch (e) {
+      // Surface the spawn failure instead of leaving the PTY in a
+      // half-alive state where output never flows but isClosed=false.
+      _dead = true;
+      if (!_out.isClosed) _out.addError(PtyException('reader-spawn', '$e'));
+      rp.close();
+      _readerPort = null;
+      if (!_readerExited!.isCompleted) _readerExited!.complete();
+      return;
+    }
     rp.listen((msg) {
       if (msg == null) {
         if (!_out.isClosed) _out.close();
         rp.close();
+        _readerPort = null;
+        if (!_readerExited!.isCompleted) _readerExited!.complete();
         _reap();
       } else {
         if (!_out.isClosed) _out.add(msg as Uint8List);
@@ -330,12 +356,39 @@ class NativePty {
   }
 
   /// Kill the child and release resources.
+  ///
+  /// Order matters: kill the child first so its slave PTY closes,
+  /// causing the master fd to return EOF. The reader isolate sees
+  /// EOF and exits cleanly. Only then do we close the master fd —
+  /// closing it before the isolate exits creates a window where the
+  /// fd number could be reused and the isolate would briefly poll
+  /// the wrong file.
   Future<void> close() async {
     if (_dead) return;
     _dead = true;
-    _nativeClose(_fd);
+
+    // Make sure the reader is fully spawned before we tear it down —
+    // otherwise close() racing with start() leaves an orphan isolate.
+    await _readerReady;
+
     _nativeKill(pid, _kSighup);
     _nativeKill(pid, 9);
+
+    // Wait for the isolate to send `null` (EOF) — confirms it has
+    // exited its poll loop and won't touch the fd again.
+    if (_readerExited != null) {
+      await _readerExited!.future.timeout(
+        const Duration(milliseconds: 500),
+        onTimeout: () {},
+      );
+    }
+
+    _nativeClose(_fd);
+    _readerIsolate?.kill(priority: Isolate.immediate);
+    _readerIsolate = null;
+    _readerPort?.close();
+    _readerPort = null;
+
     final s = calloc<ffi.Int32>();
     _waitpid(pid, s, 0);
     calloc.free(s);

lib/src/pty/session.dart

@@ -132,31 +132,44 @@ class PtySession {
       // Receive the master fd over the parent side of the socketpair.
       // recvFd blocks until ptyc sends — run in a child isolate so the
       // calling isolate's event loop stays responsive.
-      final masterFd = await _recvFdAsync(parentSock);
-
-      // Apply initial winsize (ptyc already did this, but doing it
-      // again from Dart confirms the wire + gives a place to call it
-      // when resize() lands).
-      libc.setWinsize(masterFd, cols, rows);
-
-      // Drain ptyc's stdout to parse the success envelope. We don't
-      // strictly need it — the fd arriving is proof-of-life — but
-      // draining avoids a PIPE accumulating.
-      final stdoutLine = await proc.stdout.transform(const Utf8Decoder()).transform(const LineSplitter()).first.timeout(const Duration(seconds: 5));
-      final pid = _extractPid(stdoutLine);
-
-      final code = await proc.exitCode;
-      if (code != 0) {
-        final stderr = await proc.stderr.transform(const Utf8Decoder()).join();
-        libc.close(masterFd);
-        throw PtyException('ptyc', 'ptyc exited with code $code: $stderr');
+      final int masterFd;
+      try {
+        masterFd = await _recvFdAsync(parentSock);
+      } catch (_) {
+        proc.kill();
+        rethrow;
       }
 
-      return PtySession._(pid: pid, masterFd: masterFd);
+      // Once we own masterFd, every error path below must close it
+      // before rethrowing. Wrap the rest of the spawn in its own
+      // try/catch so the cleanup is centralized.
+      try {
+        libc.setWinsize(masterFd, cols, rows);
+
+        final stdoutLine = await proc.stdout
+            .transform(const Utf8Decoder())
+            .transform(const LineSplitter())
+            .first
+            .timeout(const Duration(seconds: 5));
+        final pid = _extractPid(stdoutLine);
+
+        final code = await proc.exitCode;
+        if (code != 0) {
+          final stderr = await proc.stderr.transform(const Utf8Decoder()).join();
+          libc.close(masterFd);
+          throw PtyException('ptyc', 'ptyc exited with code $code: $stderr');
+        }
+
+        return PtySession._(pid: pid, masterFd: masterFd);
+      } catch (_) {
+        libc.close(masterFd);
+        rethrow;
+      }
     } finally {
       // parent keeps its own fd until the session is closed; ptyc-side
       // fd is released either way (ptyc has exited by now).
       if (childSock >= 0) libc.close(childSock);
+      if (parentSock >= 0) libc.close(parentSock);
       pkg_ffi.calloc.free(sv);
     }
   }
@@ -254,12 +267,16 @@ class PtySession {
   /// stall the calling isolate's event loop.
   static Future<int> _recvFdAsync(int socketFd) async {
     final port = ReceivePort();
-    final iso = await Isolate.spawn(_recvFdEntry, _RecvFdArgs(socketFd, port.sendPort));
-    final result = await port.first;
-    iso.kill(priority: Isolate.immediate);
-    port.close();
-    if (result is int) return result;
-    throw PtyException('recvFd', '$result');
+    Isolate? iso;
+    try {
+      iso = await Isolate.spawn(_recvFdEntry, _RecvFdArgs(socketFd, port.sendPort));
+      final result = await port.first;
+      if (result is int) return result;
+      throw PtyException('recvFd', '$result');
+    } finally {
+      iso?.kill(priority: Isolate.immediate);
+      port.close();
+    }
   }
 
   static void _recvFdEntry(_RecvFdArgs args) {
@@ -288,7 +305,19 @@ class PtySession {
     Isolate.spawn<_ReaderArgs>(
       _readerEntrypoint,
       _ReaderArgs(fd: _masterFd, sendPort: port.sendPort),
-    ).then((iso) => _readerIsolate = iso);
+    ).then(
+      (iso) => _readerIsolate = iso,
+      onError: (Object e) {
+        // Spawn failure leaves the session unable to ever produce
+        // output. Surface the error and mark the controller closed
+        // so consumers don't hang waiting on the stream.
+        if (!_outputCtrl.isClosed) {
+          _outputCtrl.addError(PtyException('reader-spawn', '$e'));
+          _outputCtrl.close();
+        }
+        if (!_readerExited.isCompleted) _readerExited.complete();
+      },
+    );
   }
 
   // -- request builder ------------------------------------------------------