Ignore remote stdin writes after child exit

Packages/macOS/CmuxFoundation/Sources/CmuxFoundation/ProcessPipe/ProcessPipeEndRead.swift

@@ -62,4 +62,5 @@ public struct ProcessPipeEndRead: Equatable, Sendable {
             }
         }
     }
+
 }

Packages/macOS/CmuxFoundation/Sources/CmuxFoundation/ProcessPipe/ProcessPipeStopAwareReader.swift

@@ -0,0 +1,143 @@
+import Darwin
+public import Foundation
+
+/// Drains a process pipe until EOF or a separate stop descriptor fires.
+public struct ProcessPipeStopAwareReader: Sendable {
+    // Darwin's FIONREAD macro is not imported into Swift because it expands
+    // through `_IOR`. This is `_IOR('f', 127, Int32)` from sys/filio.h.
+    private static let bytesAvailableIOControlRequest =
+        UInt(0x4000_0000)
+        | (UInt(MemoryLayout<Int32>.size) << 16)
+        | UInt(0x66_7f)
+
+    private let fileDescriptor: Int32
+    private let chunkSize: Int
+    private let stopFileDescriptor: Int32
+
+    /// Creates a stop-aware pipe reader.
+    ///
+    /// - Parameters:
+    ///   - fileDescriptor: The process-pipe descriptor to drain.
+    ///   - chunkSize: The maximum bytes read per operation.
+    ///   - stopFileDescriptor: A descriptor whose readiness ends the drain
+    ///     after bytes already buffered in `fileDescriptor` are preserved.
+    public init(
+        fileDescriptor: Int32,
+        chunkSize: Int = FileHandle.processPipeReadChunkSize,
+        stopFileDescriptor: Int32
+    ) {
+        self.fileDescriptor = fileDescriptor
+        self.chunkSize = chunkSize
+        self.stopFileDescriptor = stopFileDescriptor
+    }
+
+    /// Drains the pipe, preserving bytes buffered when the stop signal fires.
+    ///
+    /// - Returns: The captured bytes and any read failure.
+    public func readToEnd() -> ProcessPipeEndRead {
+        var data = Data()
+
+        while true {
+            var descriptors = [
+                pollfd(
+                    fd: fileDescriptor,
+                    events: Int16(POLLIN | POLLERR | POLLHUP),
+                    revents: 0
+                ),
+                pollfd(
+                    fd: stopFileDescriptor,
+                    events: Int16(POLLIN | POLLERR | POLLHUP),
+                    revents: 0
+                ),
+            ]
+            let pollResult = Darwin.poll(
+                &descriptors,
+                nfds_t(descriptors.count),
+                -1
+            )
+            if pollResult < 0 {
+                let code = errno
+                if code == EINTR {
+                    continue
+                }
+                return ProcessPipeEndRead(
+                    data: data,
+                    readError: ProcessPipeReadError(
+                        operation: "readDataToEndOfFile.poll",
+                        errnoCode: code
+                    )
+                )
+            }
+            if (descriptors[0].revents & Int16(POLLNVAL)) != 0 {
+                return ProcessPipeEndRead(
+                    data: data,
+                    readError: ProcessPipeReadError(
+                        operation: "readDataToEndOfFile.poll",
+                        errnoCode: EBADF
+                    )
+                )
+            }
+
+            let isStopping = descriptors[1].revents != 0
+            var bytesRemainingAtStop: Int?
+            if isStopping {
+                var availableBytes: Int32 = 0
+                guard Darwin.ioctl(
+                    fileDescriptor,
+                    Self.bytesAvailableIOControlRequest,
+                    &availableBytes
+                ) == 0 else {
+                    return ProcessPipeEndRead(
+                        data: data,
+                        readError: ProcessPipeReadError(
+                            operation: "readDataToEndOfFile.ioctl",
+                            errnoCode: errno
+                        )
+                    )
+                }
+                bytesRemainingAtStop = Int(availableBytes)
+                if availableBytes == 0 {
+                    return ProcessPipeEndRead(data: data, readError: nil)
+                }
+            }
+
+            repeat {
+                let maximumReadLength = bytesRemainingAtStop.map {
+                    min(chunkSize, $0)
+                } ?? chunkSize
+                switch ProcessPipeAvailableRead.readOnceIfReady(
+                    fileDescriptor: fileDescriptor,
+                    maxLength: maximumReadLength,
+                    operation: "readDataToEndOfFile"
+                ) {
+                case .success(.data(let chunk)):
+                    data.append(chunk)
+                    if let remaining = bytesRemainingAtStop.map({
+                        $0 - chunk.count
+                    }) {
+                        if remaining <= 0 {
+                            return ProcessPipeEndRead(
+                                data: data,
+                                readError: nil
+                            )
+                        }
+                        bytesRemainingAtStop = remaining
+                    }
+                case .success(.wouldBlock):
+                    if isStopping {
+                        return ProcessPipeEndRead(data: data, readError: nil)
+                    }
+                    break
+                case .success(.endOfFile):
+                    return ProcessPipeEndRead(data: data, readError: nil)
+                case .failure(let error):
+                    return ProcessPipeEndRead(data: data, readError: error)
+                }
+            } while isStopping
+
+            if isStopping {
+                return ProcessPipeEndRead(data: data, readError: nil)
+            }
+        }
+    }
+}

Packages/macOS/CmuxFoundation/Sources/CmuxFoundation/ProcessPipe/ProcessPipeStopSignal.swift

@@ -0,0 +1,52 @@
+import Darwin
+import Foundation
+
+/// A one-shot, pollable signal for stopping process-pipe I/O.
+///
+/// Closing the signal's write end leaves its read end permanently hung up, so
+/// every reader polling the same descriptor wakes. Retain the signal until all
+/// polling operations have returned; the read descriptor closes at deinit.
+public final class ProcessPipeStopSignal: Sendable {
+    /// The stable descriptor callers include in their `poll(2)` descriptor set.
+    ///
+    /// The signal owns this descriptor. Callers must not close it and must stop
+    /// polling before releasing the signal.
+    public let readFileDescriptor: Int32
+
+    private let writeFileDescriptor: Int32
+    private let didSignal = AtomicBooleanGate(false)
+
+    /// Creates a close-on-exec pipe used exclusively as a stop signal.
+    ///
+    /// - Throws: A ``POSIXError`` when the pipe or close-on-exec flags cannot
+    ///   be created.
+    public init() throws {
+        var descriptors: [Int32] = [-1, -1]
+        guard Darwin.pipe(&descriptors) == 0 else {
+            throw POSIXError(POSIXErrorCode(rawValue: errno) ?? .EIO)
+        }
+
+        guard fcntl(descriptors[0], F_SETFD, FD_CLOEXEC) == 0,
+              fcntl(descriptors[1], F_SETFD, FD_CLOEXEC) == 0 else {
+            let code = errno
+            Darwin.close(descriptors[0])
+            Darwin.close(descriptors[1])
+            throw POSIXError(POSIXErrorCode(rawValue: code) ?? .EIO)
+        }
+        readFileDescriptor = descriptors[0]
+        writeFileDescriptor = descriptors[1]
+    }
+
+    /// Wakes every poller. Repeated calls are safe and have no effect.
+    public func signal() {
+        guard didSignal.compareExchange(expected: false, desired: true) else {
+            return
+        }
+        Darwin.close(writeFileDescriptor)
+    }
+
+    deinit {
+        signal()
+        Darwin.close(readFileDescriptor)
+    }
+}

Packages/macOS/CmuxFoundation/Tests/CmuxFoundationTests/FileHandleProcessPipeReadingTests.swift

@@ -64,6 +64,82 @@ struct FileHandleProcessPipeReadingTests {
         #expect(result.readError == readError)
     }
 
+    @Test("A stopped end read drains buffered bytes without waiting for EOF")
+    func stoppedEndReadDrainsBufferedBytesWithoutEOF() throws {
+        let pipe = Pipe()
+        let stopSignal = try ProcessPipeStopSignal()
+        defer {
+            try? pipe.fileHandleForWriting.close()
+            try? pipe.fileHandleForReading.close()
+        }
+        try pipe.fileHandleForWriting.write(contentsOf: Data("buffered".utf8))
+        stopSignal.signal()
+
+        let result = ProcessPipeStopAwareReader(
+            fileDescriptor: pipe.fileHandleForReading.fileDescriptor,
+            stopFileDescriptor: stopSignal.readFileDescriptor
+        ).readToEnd()
+
+        #expect(result.data == Data("buffered".utf8))
+        #expect(result.readError == nil)
+    }
+
+    @Test("A stopped end read drains every byte present at the stop snapshot")
+    func stoppedEndReadDrainsSnapshotLargerThanSixteenChunks() throws {
+        let pipe = Pipe()
+        let stopSignal = try ProcessPipeStopSignal()
+        let payload = Data((0..<128).map { UInt8($0) })
+        defer {
+            try? pipe.fileHandleForWriting.close()
+            try? pipe.fileHandleForReading.close()
+        }
+        try pipe.fileHandleForWriting.write(contentsOf: payload)
+        stopSignal.signal()
+
+        let result = ProcessPipeStopAwareReader(
+            fileDescriptor: pipe.fileHandleForReading.fileDescriptor,
+            chunkSize: 1,
+            stopFileDescriptor: stopSignal.readFileDescriptor
+        ).readToEnd()
+
+        #expect(result.data == payload)
+        #expect(result.readError == nil)
+    }
+
+    @Test("One stop signal wakes every blocked pipe reader")
+    func stopSignalBroadcastsToBlockedReaders() throws {
+        let firstPipe = Pipe()
+        let secondPipe = Pipe()
+        let stopSignal = try ProcessPipeStopSignal()
+        defer {
+            try? firstPipe.fileHandleForWriting.close()
+            try? firstPipe.fileHandleForReading.close()
+            try? secondPipe.fileHandleForWriting.close()
+            try? secondPipe.fileHandleForReading.close()
+        }
+
+        stopSignal.signal()
+        let firstFinished = DispatchSemaphore(value: 0)
+        let secondFinished = DispatchSemaphore(value: 0)
+        DispatchQueue.global(qos: .utility).async {
+            _ = ProcessPipeStopAwareReader(
+                fileDescriptor: firstPipe.fileHandleForReading.fileDescriptor,
+                stopFileDescriptor: stopSignal.readFileDescriptor
+            ).readToEnd()
+            firstFinished.signal()
+        }
+        DispatchQueue.global(qos: .utility).async {
+            _ = ProcessPipeStopAwareReader(
+                fileDescriptor: secondPipe.fileHandleForReading.fileDescriptor,
+                stopFileDescriptor: stopSignal.readFileDescriptor
+            ).readToEnd()
+            secondFinished.signal()
+        }
+
+        #expect(firstFinished.wait(timeout: .now() + 1) == .success)
+        #expect(secondFinished.wait(timeout: .now() + 1) == .success)
+    }
+
     @Test("readDataToEndOfFileOrEmpty drains a closed pipe")
     func drainsClosedPipe() throws {
         let pipe = Pipe()

Packages/macOS/CmuxRemoteSession/Sources/CmuxRemoteSession/Process/RemoteProcessCompletionWaitOutcome.swift

@@ -0,0 +1,6 @@
+enum RemoteProcessCompletionWaitOutcome: Equatable, Sendable {
+    case processExited
+    case stdinWriteFailed
+    case timedOut
+    case waitFailed(Int32)
+}

Packages/macOS/CmuxRemoteSession/Sources/CmuxRemoteSession/Process/RemoteProcessCompletionWaiter.swift

@@ -0,0 +1,57 @@
+internal import Darwin
+internal import Foundation
+
+struct RemoteProcessCompletionWaiter: Sendable {
+    let processExitFileDescriptor: Int32
+    let stdinFailureFileDescriptor: Int32
+
+    func wait(
+        until deadline: DispatchTime
+    ) -> RemoteProcessCompletionWaitOutcome {
+        while true {
+            let now = DispatchTime.now().uptimeNanoseconds
+            guard now < deadline.uptimeNanoseconds else {
+                return .timedOut
+            }
+            let remainingNanoseconds = deadline.uptimeNanoseconds - now
+            let wholeMilliseconds = remainingNanoseconds / 1_000_000
+            let roundedMilliseconds = wholeMilliseconds
+                + (remainingNanoseconds % 1_000_000 == 0 ? 0 : 1)
+            let timeoutMilliseconds = Int32(
+                min(roundedMilliseconds, UInt64(Int32.max))
+            )
+            var descriptors = [
+                pollfd(
+                    fd: processExitFileDescriptor,
+                    events: Int16(POLLIN | POLLERR | POLLHUP),
+                    revents: 0
+                ),
+                pollfd(
+                    fd: stdinFailureFileDescriptor,
+                    events: Int16(POLLIN | POLLERR | POLLHUP),
+                    revents: 0
+                ),
+            ]
+            let result = Darwin.poll(
+                &descriptors,
+                nfds_t(descriptors.count),
+                timeoutMilliseconds
+            )
+            if result > 0 {
+                if descriptors[1].revents != 0 {
+                    return .stdinWriteFailed
+                }
+                if descriptors[0].revents != 0 {
+                    return .processExited
+                }
+                continue
+            }
+            if result == 0 {
+                return .timedOut
+            }
+            if errno != EINTR {
+                return .waitFailed(errno)
+            }
+        }
+    }
+}

Packages/macOS/CmuxRemoteSession/Sources/CmuxRemoteSession/Process/RemoteProcessExitSignal.swift

@@ -0,0 +1,34 @@
+internal import CmuxFoundation
+internal import Foundation
+
+/// One-shot process-exit notification with deadline-aware blocking waits.
+///
+/// Foundation can report `isRunning == false` just before its termination
+/// callback runs. `recordExit()` is therefore idempotent so the blocking owner
+/// may publish that observed exit without racing the callback.
+final class RemoteProcessExitSignal: Sendable {
+    private let didExit = AtomicBooleanGate(false)
+    private let group = DispatchGroup()
+    private let pollSignal: ProcessPipeStopSignal
+
+    init() throws {
+        pollSignal = try ProcessPipeStopSignal()
+        group.enter()
+    }
+
+    var readFileDescriptor: Int32 {
+        pollSignal.readFileDescriptor
+    }
+
+    func recordExit() {
+        guard didExit.compareExchange(expected: false, desired: true) else {
+            return
+        }
+        pollSignal.signal()
+        group.leave()
+    }
+
+    func wait(until deadline: DispatchTime) -> Bool {
+        group.wait(timeout: deadline) == .success
+    }
+}

Packages/macOS/CmuxRemoteSession/Sources/CmuxRemoteSession/Process/RemoteProcessPipeCaptureState.swift

@@ -0,0 +1,10 @@
+internal import Foundation
+
+/// Process-local capture storage whose mutations are confined to the runner's
+/// serial capture queue and read only after its capture group completes.
+final class RemoteProcessPipeCaptureState: @unchecked Sendable {
+    var stdoutData = Data()
+    var stderrData = Data()
+    var stdoutReadError: (any Error)?
+    var stderrReadError: (any Error)?
+}