From 3101ebd99205816acb95489746b8dd97eaff42df Mon Sep 17 00:00:00 2001 From: postoso Date: Fri, 26 Jun 2026 17:25:23 -0400 Subject: [PATCH 1/2] fix(command-mode): drain process output concurrently to prevent TerminalService pipe deadlock --- Sources/Fluid/Services/TerminalService.swift | 17 ++++++++-- .../DictationE2ETests.swift | 33 +++++++++++++++++++ 2 files changed, 47 insertions(+), 3 deletions(-) diff --git a/Sources/Fluid/Services/TerminalService.swift b/Sources/Fluid/Services/TerminalService.swift index 47bf79d4..1baaef26 100644 --- a/Sources/Fluid/Services/TerminalService.swift +++ b/Sources/Fluid/Services/TerminalService.swift @@ -108,12 +108,23 @@ 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 { outputHandle.readDataToEndOfFile() }.value + async let pendingError = Task.detached { errorHandle.readDataToEndOfFile() }.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) diff --git a/Tests/FluidDictationIntegrationTests/DictationE2ETests.swift b/Tests/FluidDictationIntegrationTests/DictationE2ETests.swift index 3f97b50a..8b3216fa 100644 --- a/Tests/FluidDictationIntegrationTests/DictationE2ETests.swift +++ b/Tests/FluidDictationIntegrationTests/DictationE2ETests.swift @@ -691,6 +691,39 @@ 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. + let service = TerminalService() + let lineCount = 200_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") + 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 = 200_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") + XCTAssertLessThan(result.executionTimeMs, 15000, "Should return promptly, not after the ~30s 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" || From 85f4a1b2a60db9e3d2ca06dce7dbbfb605059346 Mon Sep 17 00:00:00 2001 From: postoso Date: Mon, 6 Jul 2026 07:28:19 -0400 Subject: [PATCH 2/2] fix(command-mode): bound retained terminal output --- Sources/Fluid/Services/TerminalService.swift | 66 ++++++++++- .../DictationE2ETests.swift | 107 +++++++++++++++++- 2 files changed, 168 insertions(+), 5 deletions(-) diff --git a/Sources/Fluid/Services/TerminalService.swift b/Sources/Fluid/Services/TerminalService.swift index 1baaef26..1c7aa447 100644 --- a/Sources/Fluid/Services/TerminalService.swift +++ b/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 { @@ -116,8 +120,12 @@ final class TerminalService { // process can run to completion. let outputHandle = outputPipe.fileHandleForReading let errorHandle = errorPipe.fileHandleForReading - async let pendingOutput = Task.detached { outputHandle.readDataToEndOfFile() }.value - async let pendingError = Task.detached { errorHandle.readDataToEndOfFile() }.value + 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 @@ -179,4 +187,58 @@ final class TerminalService { // If serialization fails for any reason, return minimal safe JSON return #"{"success":false,"output":"","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) + } } diff --git a/Tests/FluidDictationIntegrationTests/DictationE2ETests.swift b/Tests/FluidDictationIntegrationTests/DictationE2ETests.swift index 8b3216fa..95d21967 100644 --- a/Tests/FluidDictationIntegrationTests/DictationE2ETests.swift +++ b/Tests/FluidDictationIntegrationTests/DictationE2ETests.swift @@ -696,9 +696,11 @@ final class DictationE2ETests: XCTestCase { // 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. + // 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 = 200_000 + let lineCount = 100_000 let result = await service.execute(command: "seq 1 \(lineCount)") @@ -706,6 +708,7 @@ final class DictationE2ETests: XCTestCase { 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") } @@ -713,7 +716,7 @@ final class DictationE2ETests: XCTestCase { func testTerminalServiceExecute_largeStderrIsNotTruncated() async throws { // The same deadlock applied to stderr; both pipes must be drained concurrently. let service = TerminalService() - let lineCount = 200_000 + let lineCount = 100_000 let result = await service.execute(command: "seq 1 \(lineCount) 1>&2") @@ -721,9 +724,107 @@ final class DictationE2ETests: XCTestCase { 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" ||