fix(command-mode): drain process output concurrently to prevent TerminalService pipe deadlock

Sources/Fluid/Services/TerminalService.swift

@@ -3,6 +3,10 @@ import Foundation
 /// Simple terminal command execution service
 /// All responses are JSON-parsable for easy AI processing
 final class TerminalService {
+    nonisolated static let maxCapturedOutputBytes = 1 * 1024 * 1024
+
+    nonisolated private static let pipeReadChunkBytes = 32 * 1024
+
     // MARK: - JSON Response Types
 
     struct CommandResult: Codable {
@@ -108,12 +112,27 @@ final class TerminalService {
                 }
             }
 
+            // Drain stdout and stderr concurrently while the process is still
+            // running. A child that writes more than the pipe buffer (~64KB)
+            // blocks on write() until its output is read, so reading only after
+            // waitUntilExit() would deadlock until the timeout fired and return
+            // truncated output. Background reads keep both pipes drained so the
+            // process can run to completion.
+            let outputHandle = outputPipe.fileHandleForReading
+            let errorHandle = errorPipe.fileHandleForReading
+            async let pendingOutput = Task.detached {
+                Self.readCapturedOutput(from: outputHandle, streamName: "stdout")
+            }.value
+            async let pendingError = Task.detached {
+                Self.readCapturedOutput(from: errorHandle, streamName: "stderr")
+            }.value
+
+            let outputData = await pendingOutput
+            let errorData = await pendingError
+
             process.waitUntilExit()
             timeoutTask.cancel()
 
-            let outputData = outputPipe.fileHandleForReading.readDataToEndOfFile()
-            let errorData = errorPipe.fileHandleForReading.readDataToEndOfFile()
-
             let output = String(data: outputData, encoding: .utf8)?.trimmingCharacters(in: .whitespacesAndNewlines) ?? ""
             let errorOutput = String(data: errorData, encoding: .utf8)?.trimmingCharacters(in: .whitespacesAndNewlines)
 
@@ -168,4 +187,58 @@ final class TerminalService {
         // If serialization fails for any reason, return minimal safe JSON
         return #"{"success":false,"output":"<json-serialization-failed>","exitCode":-1}"#
     }
+
+    nonisolated private static func readCapturedOutput(from handle: FileHandle, streamName: String) -> Data {
+        var capturedData = Data()
+        var didTruncate = false
+
+        while true {
+            let chunk: Data
+            do {
+                guard let nextChunk = try handle.read(upToCount: Self.pipeReadChunkBytes),
+                      !nextChunk.isEmpty
+                else {
+                    break
+                }
+                chunk = nextChunk
+            } catch {
+                break
+            }
+
+            if capturedData.count < Self.maxCapturedOutputBytes {
+                let remainingBytes = Self.maxCapturedOutputBytes - capturedData.count
+                if chunk.count <= remainingBytes {
+                    capturedData.append(chunk)
+                } else {
+                    capturedData.append(contentsOf: chunk.prefix(remainingBytes))
+                    didTruncate = true
+                }
+            } else {
+                didTruncate = true
+            }
+        }
+
+        if didTruncate {
+            Self.trimTrailingIncompleteUTF8Sequence(in: &capturedData)
+            capturedData.append(Self.truncationNotice(for: streamName))
+        }
+
+        return capturedData
+    }
+
+    nonisolated private static func trimTrailingIncompleteUTF8Sequence(in data: inout Data) {
+        for _ in 0 ..< 4 {
+            if String(data: data, encoding: .utf8) != nil {
+                return
+            }
+            if data.isEmpty {
+                return
+            }
+            data.removeLast()
+        }
+    }
+
+    nonisolated private static func truncationNotice(for streamName: String) -> Data {
+        Data("\n[\(streamName) truncated after \(Self.maxCapturedOutputBytes) bytes]".utf8)
+    }
 }

Tests/FluidDictationIntegrationTests/DictationE2ETests.swift

@@ -691,6 +691,140 @@ final class DictationE2ETests: XCTestCase {
         )
     }
 
+    func testTerminalServiceExecute_largeStdoutIsNotTruncatedAndReturnsQuickly() async {
+        // Regression for a pipe-buffer deadlock: execute() used to call
+        // waitUntilExit() before draining stdout, so a command writing more than
+        // the ~64KB pipe buffer blocked on write() until the 30s timeout fired,
+        // returning truncated output with success == false. Draining stdout and
+        // stderr concurrently lets the command run to completion. The output
+        // is large enough to exceed the pipe buffer but small enough to stay
+        // under TerminalService's captured-output cap.
+        let service = TerminalService()
+        let lineCount = 100_000
+
+        let result = await service.execute(command: "seq 1 \(lineCount)")
+
+        XCTAssertTrue(result.success, "Expected success, got exitCode \(result.exitCode) error \(result.error ?? "nil")")
+        XCTAssertEqual(result.exitCode, 0)
+        XCTAssertGreaterThan(result.output.utf8.count, 64 * 1024, "Output should exceed the 64KB pipe buffer")
+        XCTAssertTrue(result.output.hasSuffix("\(lineCount)"), "Last line should be complete, output was truncated")
+        XCTAssertFalse(result.output.contains("truncated after"))
+        XCTAssertEqual(result.output.split(separator: "\n").count, lineCount)
+        XCTAssertLessThan(result.executionTimeMs, 15000, "Should return promptly, not after the ~30s timeout")
+    }
+
+    func testTerminalServiceExecute_largeStderrIsNotTruncated() async throws {
+        // The same deadlock applied to stderr; both pipes must be drained concurrently.
+        let service = TerminalService()
+        let lineCount = 100_000
+
+        let result = await service.execute(command: "seq 1 \(lineCount) 1>&2")
+
+        XCTAssertTrue(result.success, "Expected success, got exitCode \(result.exitCode)")
+        let stderr = try XCTUnwrap(result.error)
+        XCTAssertGreaterThan(stderr.utf8.count, 64 * 1024, "Stderr should exceed the 64KB pipe buffer")
+        XCTAssertTrue(stderr.hasSuffix("\(lineCount)"), "Last stderr line should be complete, output was truncated")
+        XCTAssertFalse(stderr.contains("truncated after"))
+        XCTAssertLessThan(result.executionTimeMs, 15000, "Should return promptly, not after the ~30s timeout")
+    }
+
+    func testTerminalServiceExecute_largeStdoutCaptureIsBoundedAndMarked() async {
+        let service = TerminalService()
+        let requestedBytes = TerminalService.maxCapturedOutputBytes + (128 * 1024)
+
+        let result = await service.execute(command: "yes stdout | head -c \(requestedBytes)")
+
+        XCTAssertTrue(result.success, "Expected success, got exitCode \(result.exitCode) error \(result.error ?? "nil")")
+        XCTAssertEqual(result.exitCode, 0)
+        XCTAssertTrue(
+            result.output.contains("[stdout truncated after \(TerminalService.maxCapturedOutputBytes) bytes]"),
+            "Expected stdout truncation notice"
+        )
+        XCTAssertLessThan(
+            result.output.utf8.count,
+            TerminalService.maxCapturedOutputBytes + 128,
+            "Captured stdout should stay close to the configured cap"
+        )
+        XCTAssertLessThan(result.executionTimeMs, 15000, "Should drain without waiting for timeout")
+    }
+
+    func testTerminalServiceExecute_largeStderrCaptureIsBoundedAndMarked() async throws {
+        let service = TerminalService()
+        let requestedBytes = TerminalService.maxCapturedOutputBytes + (128 * 1024)
+
+        let result = await service.execute(command: "yes stderr | head -c \(requestedBytes) 1>&2")
+
+        XCTAssertTrue(result.success, "Expected success, got exitCode \(result.exitCode)")
+        let stderr = try XCTUnwrap(result.error)
+        XCTAssertTrue(
+            stderr.contains("[stderr truncated after \(TerminalService.maxCapturedOutputBytes) bytes]"),
+            "Expected stderr truncation notice"
+        )
+        XCTAssertLessThan(
+            stderr.utf8.count,
+            TerminalService.maxCapturedOutputBytes + 128,
+            "Captured stderr should stay close to the configured cap"
+        )
+        XCTAssertLessThan(result.executionTimeMs, 15000, "Should drain without waiting for timeout")
+    }
+
+    func testTerminalServiceExecute_multibyteStdoutSplitAtCapKeepsTruncationNotice() async {
+        let service = TerminalService()
+        let requestedBytes = TerminalService.maxCapturedOutputBytes + 64
+
+        let result = await service.execute(command: "yes é | head -c \(requestedBytes)")
+
+        XCTAssertTrue(result.success, "Expected success, got exitCode \(result.exitCode) error \(result.error ?? "nil")")
+        XCTAssertTrue(
+            result.output.contains("[stdout truncated after \(TerminalService.maxCapturedOutputBytes) bytes]"),
+            "Truncation notice should survive even when the cap splits a UTF-8 scalar"
+        )
+        XCTAssertLessThan(
+            result.output.utf8.count,
+            TerminalService.maxCapturedOutputBytes + 128,
+            "Captured stdout should stay close to the configured cap"
+        )
+    }
+
+    func testTerminalServiceExecute_unboundedStdoutTimesOutWithBoundedCapture() async {
+        let service = TerminalService()
+
+        let result = await service.execute(command: "yes stdout", timeout: 1)
+
+        XCTAssertFalse(result.success, "Unbounded command should be terminated by timeout")
+        XCTAssertNotEqual(result.exitCode, 0)
+        XCTAssertTrue(
+            result.output.contains("[stdout truncated after \(TerminalService.maxCapturedOutputBytes) bytes]"),
+            "Expected stdout truncation notice"
+        )
+        XCTAssertLessThan(
+            result.output.utf8.count,
+            TerminalService.maxCapturedOutputBytes + 128,
+            "Captured stdout should stay close to the configured cap"
+        )
+        XCTAssertLessThan(result.executionTimeMs, 5000, "Should return shortly after timeout")
+    }
+
+    func testTerminalServiceExecute_unboundedStderrTimesOutWithBoundedCapture() async throws {
+        let service = TerminalService()
+
+        let result = await service.execute(command: "yes stderr 1>&2", timeout: 1)
+
+        XCTAssertFalse(result.success, "Unbounded command should be terminated by timeout")
+        XCTAssertNotEqual(result.exitCode, 0)
+        let stderr = try XCTUnwrap(result.error)
+        XCTAssertTrue(
+            stderr.contains("[stderr truncated after \(TerminalService.maxCapturedOutputBytes) bytes]"),
+            "Expected stderr truncation notice"
+        )
+        XCTAssertLessThan(
+            stderr.utf8.count,
+            TerminalService.maxCapturedOutputBytes + 128,
+            "Captured stderr should stay close to the configured cap"
+        )
+        XCTAssertLessThan(result.executionTimeMs, 5000, "Should return shortly after timeout")
+    }
+
     private static func modelDirectoryForRun() -> URL {
         // Use a stable path on CI so GitHub Actions cache can speed up runs.
         if ProcessInfo.processInfo.environment["GITHUB_ACTIONS"] == "true" ||