From 108b396114a64cb72c6dd7f902e37d206fdaa3e8 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Thu, 20 Aug 2026 00:11:05 -0700 Subject: [PATCH 1/2] fix: prevent RPC pipe writes from aborting after child exit --- CHANGELOG.md | 1 + .../Process/RPCChildProcessTeardown.swift | 38 +++++++- .../Providers/Grok/GrokRPCClient.swift | 24 +++-- Sources/CodexBarCore/UsageFetcher.swift | 24 +++-- .../RPCChildProcessTeardownTests.swift | 95 +++++++++++++++++++ 5 files changed, 160 insertions(+), 22 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ca4b2a3ecb..fc9983becd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,7 @@ ## 0.54.1 — Unreleased +- Fixed Codex and Grok app crashes when an RPC timeout or child exit races a write to closed subprocess stdin (#3087). - Added conditional tokens to the menu bar layout editor: named, reusable if/then/else rules (1–4 AND/OR clauses over Session/Weekly/Scoped/Auto thresholds) that swap or hide tokens based on live usage, downgrade-safe and localized across all 23 catalogs (#3076). Thanks @wdmitchelluk! - Fixed inconsistent German localization of "About" ("Um" → "Über") (#3077). Thanks @dwt! - Localized provider usage details in Simplified Chinese: DeepSeek detailed usage/balance, z.ai/GLM quota details, token charts, and the 5-hour reset text (#3084). Thanks @haixing23! diff --git a/Sources/CodexBarCore/Host/Process/RPCChildProcessTeardown.swift b/Sources/CodexBarCore/Host/Process/RPCChildProcessTeardown.swift index 4310393241..f860574079 100644 --- a/Sources/CodexBarCore/Host/Process/RPCChildProcessTeardown.swift +++ b/Sources/CodexBarCore/Host/Process/RPCChildProcessTeardown.swift @@ -1,5 +1,39 @@ +#if canImport(Darwin) +import Darwin +#endif import Foundation +package final class RPCChildProcessInput: @unchecked Sendable { + package let pipe = Pipe() + + private let lock = NSLock() + private var isClosed = false + + package init() { + #if canImport(Darwin) + // Keep broken pipes catchable instead of terminating the app with SIGPIPE. + _ = fcntl(self.pipe.fileHandleForWriting.fileDescriptor, F_SETNOSIGPIPE, 1) + #endif + } + + package func write(_ data: Data) throws { + try self.lock.withLock { + guard !self.isClosed else { + throw CocoaError(.fileWriteUnknown) + } + try self.pipe.fileHandleForWriting.write(contentsOf: data) + } + } + + package func close() { + self.lock.withLock { + guard !self.isClosed else { return } + self.isClosed = true + try? self.pipe.fileHandleForWriting.close() + } + } +} + package enum RPCChildProcessTeardown { /// Tears down a JSON-RPC child spawned via Foundation `Process`. /// @@ -7,8 +41,8 @@ package enum RPCChildProcessTeardown { /// then escalates SIGTERM -> bounded wait -> SIGKILL across the child's process tree via /// `SubprocessRunner.terminateProcess`, so children that ignore SIGTERM cannot leak /// (#2789). Foundation reaps the child once it exits, so no explicit waitpid is needed here. - package static func terminate(process: Process, stdinPipe: Pipe) { - try? stdinPipe.fileHandleForWriting.close() + package static func terminate(process: Process, stdin: RPCChildProcessInput) { + stdin.close() SubprocessRunner.terminateProcess(process, processGroup: nil) } } diff --git a/Sources/CodexBarCore/Providers/Grok/GrokRPCClient.swift b/Sources/CodexBarCore/Providers/Grok/GrokRPCClient.swift index 20bf173e8a..2478866612 100644 --- a/Sources/CodexBarCore/Providers/Grok/GrokRPCClient.swift +++ b/Sources/CodexBarCore/Providers/Grok/GrokRPCClient.swift @@ -9,7 +9,7 @@ final class GrokRPCClient: @unchecked Sendable { private static let log = CodexBarLog.logger(LogCategories.provider(.grok)) private let process = Process() - private let stdinPipe = Pipe() + private let stdin = RPCChildProcessInput() private let stdoutPipe = Pipe() private let stderrPipe = Pipe() private let initializeTimeoutSeconds: TimeInterval @@ -47,7 +47,7 @@ final class GrokRPCClient: @unchecked Sendable { self.process.environment = env self.process.executableURL = URL(fileURLWithPath: "/usr/bin/env") self.process.arguments = [resolvedExec] + arguments - self.process.standardInput = self.stdinPipe + self.process.standardInput = self.stdin.pipe self.process.standardOutput = self.stdoutPipe self.process.standardError = self.stderrPipe @@ -63,7 +63,7 @@ final class GrokRPCClient: @unchecked Sendable { let stdoutLineContinuation = self.stdoutLineContinuation let stdoutBuffer = BoundedLineBuffer() let process = self.process - let stdinPipe = self.stdinPipe + let stdin = self.stdin stdoutHandle.readabilityHandler = { handle in let data = handle.availableData if data.isEmpty { @@ -76,7 +76,7 @@ final class GrokRPCClient: @unchecked Sendable { Self.log.warning("Grok RPC line exceeded memory limit; terminating process") handle.readabilityHandler = nil DispatchQueue.global(qos: .userInitiated).async { - RPCChildProcessTeardown.terminate(process: process, stdinPipe: stdinPipe) + RPCChildProcessTeardown.terminate(process: process, stdin: stdin) } stdoutLineContinuation.finish() return @@ -128,7 +128,7 @@ final class GrokRPCClient: @unchecked Sendable { func shutdown() { Self.log.debug("Grok RPC stopping") - RPCChildProcessTeardown.terminate(process: self.process, stdinPipe: self.stdinPipe) + RPCChildProcessTeardown.terminate(process: self.process, stdin: self.stdin) } // MARK: - JSON-RPC plumbing (mirrors CodexRPCClient) @@ -195,9 +195,9 @@ final class GrokRPCClient: @unchecked Sendable { // Dispatch off the timeout task so the bounded TERM-to-KILL wait cannot delay the timeout // error or let the stdout-EOF failure win the race; `shutdown()` remains the synchronous backstop. let process = self.process - let stdinPipe = self.stdinPipe + let stdin = self.stdin DispatchQueue.global(qos: .userInitiated).async { - RPCChildProcessTeardown.terminate(process: process, stdinPipe: stdinPipe) + RPCChildProcessTeardown.terminate(process: process, stdin: stdin) } } @@ -221,12 +221,16 @@ final class GrokRPCClient: @unchecked Sendable { // the on-the-wire shape the grok agent expects. let unescaped = String(data: raw, encoding: .utf8)? .replacingOccurrences(of: "\\/", with: "/") - let data = unescaped.flatMap { $0.data(using: .utf8) } ?? raw + var data = unescaped.flatMap { $0.data(using: .utf8) } ?? raw if let preview = String(data: data.prefix(200), encoding: .utf8) { Self.log.debug("grok rpc -> \(preview)") } - self.stdinPipe.fileHandleForWriting.write(data) - self.stdinPipe.fileHandleForWriting.write(Data([0x0A])) + data.append(0x0A) + do { + try self.stdin.write(data) + } catch { + throw GrokRPCError.requestFailed("grok agent stdin closed: \(error.localizedDescription)") + } } private func readNextMessage() async throws -> [String: Any] { diff --git a/Sources/CodexBarCore/UsageFetcher.swift b/Sources/CodexBarCore/UsageFetcher.swift index 1d509fae56..25507b38bd 100644 --- a/Sources/CodexBarCore/UsageFetcher.swift +++ b/Sources/CodexBarCore/UsageFetcher.swift @@ -847,7 +847,7 @@ private final class CodexRPCClient: @unchecked Sendable { // Provider-specific by design: Codex RPC owns its dedicated subprocess log category. private static let log = CodexBarLog.logger(LogCategories.provider(.codex, scope: "rpc")) private let process = Process() - private let stdinPipe = Pipe() + private let stdin = RPCChildProcessInput() private let stdoutPipe = Pipe() private let stderrPipe = Pipe() private let stdoutLineStream: AsyncStream @@ -889,7 +889,7 @@ private final class CodexRPCClient: @unchecked Sendable { self.process.environment = env self.process.executableURL = URL(fileURLWithPath: "/usr/bin/env") self.process.arguments = [resolvedExec] + arguments - self.process.standardInput = self.stdinPipe + self.process.standardInput = self.stdin.pipe self.process.standardOutput = self.stdoutPipe self.process.standardError = self.stderrPipe @@ -912,7 +912,7 @@ private final class CodexRPCClient: @unchecked Sendable { let stdoutLineContinuation = self.stdoutLineContinuation let stdoutBuffer = BoundedLineBuffer() let process = self.process - let stdinPipe = self.stdinPipe + let stdin = self.stdin stdoutHandle.readabilityHandler = { handle in let data = handle.availableData if data.isEmpty { @@ -926,7 +926,7 @@ private final class CodexRPCClient: @unchecked Sendable { Self.log.warning("Codex RPC line exceeded memory limit; terminating process") handle.readabilityHandler = nil DispatchQueue.global(qos: .userInitiated).async { - RPCChildProcessTeardown.terminate(process: process, stdinPipe: stdinPipe) + RPCChildProcessTeardown.terminate(process: process, stdin: stdin) } stdoutLineContinuation.finish() return @@ -973,7 +973,7 @@ private final class CodexRPCClient: @unchecked Sendable { func shutdown() { Self.log.debug("Codex RPC stopping") - RPCChildProcessTeardown.terminate(process: self.process, stdinPipe: self.stdinPipe) + RPCChildProcessTeardown.terminate(process: self.process, stdin: self.stdin) } // MARK: - JSON-RPC helpers @@ -1052,9 +1052,9 @@ private final class CodexRPCClient: @unchecked Sendable { // Dispatch off the timeout task so the bounded TERM-to-KILL wait cannot delay the timeout // error or let the stdout-EOF failure win the race; `shutdown()` remains the synchronous backstop. let process = self.process - let stdinPipe = self.stdinPipe + let stdin = self.stdin DispatchQueue.global(qos: .userInitiated).async { - RPCChildProcessTeardown.terminate(process: process, stdinPipe: stdinPipe) + RPCChildProcessTeardown.terminate(process: process, stdin: stdin) } } @@ -1070,9 +1070,13 @@ private final class CodexRPCClient: @unchecked Sendable { } private func sendPayload(_ payload: [String: Any]) throws { - let data = try JSONSerialization.data(withJSONObject: payload) - self.stdinPipe.fileHandleForWriting.write(data) - self.stdinPipe.fileHandleForWriting.write(Data([0x0A])) + var data = try JSONSerialization.data(withJSONObject: payload) + data.append(0x0A) + do { + try self.stdin.write(data) + } catch { + throw RPCWireError.requestFailed("codex app-server stdin closed: \(error.localizedDescription)") + } } private func readNextMessage() async throws -> [String: Any] { diff --git a/Tests/CodexBarTests/RPCChildProcessTeardownTests.swift b/Tests/CodexBarTests/RPCChildProcessTeardownTests.swift index 6f7b62a6a1..1fbbecc72b 100644 --- a/Tests/CodexBarTests/RPCChildProcessTeardownTests.swift +++ b/Tests/CodexBarTests/RPCChildProcessTeardownTests.swift @@ -10,6 +10,101 @@ import Glibc @Suite(.serialized) struct RPCChildProcessTeardownTests { + @Test + func `RPC stdin writes after child teardown fail without aborting`() throws { + let stdin = RPCChildProcessInput() + let process = Process() + process.executableURL = URL(fileURLWithPath: "/bin/cat") + process.standardInput = stdin.pipe + process.standardOutput = Pipe() + process.standardError = Pipe() + try process.run() + + RPCChildProcessTeardown.terminate(process: process, stdin: stdin) + + #expect(throws: (any Error).self) { + try stdin.write(Data("{\"id\":1}\n".utf8)) + } + stdin.close() + } + + @Test + func `RPC stdin writes to an unexpectedly exited child throw without aborting`() throws { + let stdin = RPCChildProcessInput() + let process = Process() + process.executableURL = URL(fileURLWithPath: "/bin/cat") + process.standardInput = stdin.pipe + process.standardOutput = Pipe() + process.standardError = Pipe() + try process.run() + + process.terminate() + process.waitUntilExit() + defer { stdin.close() } + + #expect(throws: (any Error).self) { + try stdin.write(Data("{\"id\":1}\n".utf8)) + } + } + + @Test + func `Codex RPC reports a normal failure when its child closes stdin`() async throws { + let scriptURL = FileManager.default.temporaryDirectory + .appendingPathComponent("codex-closed-stdin-\(UUID().uuidString)") + defer { try? FileManager.default.removeItem(at: scriptURL) } + + let script = """ + #!/usr/bin/python3 -S + import os + import sys + + sys.stdin.readline() + os.close(0) + print('{"id":1,"result":{}}', flush=True) + os._exit(0) + """ + try script.write(to: scriptURL, atomically: true, encoding: .utf8) + try FileManager.default.setAttributes([.posixPermissions: 0o755], ofItemAtPath: scriptURL.path) + + let fetcher = UsageFetcher( + environment: ["CODEX_CLI_PATH": scriptURL.path], + initializeTimeoutSeconds: 5, + requestTimeoutSeconds: 2) + + let error = await #expect(throws: RPCWireError.self) { + _ = try await fetcher.loadLatestCLIAccountSnapshot() + } + guard case let .requestFailed(message) = error else { + Issue.record("Expected a normal RPC request failure, got \(String(describing: error))") + return + } + #expect(message.contains("stdin closed")) + } + + @Test + func `Grok RPC requests after child shutdown fail without aborting`() async throws { + let client = try GrokRPCClient( + executable: "/bin/cat", + arguments: [], + environment: [ + "PATH": "/usr/bin:/bin", + "GROK_CLI_PATH": "/bin/cat", + ], + initializeTimeoutSeconds: 5, + requestTimeoutSeconds: 2) + + client.shutdown() + + let error = await #expect(throws: GrokRPCError.self) { + try await client.initialize() + } + guard case let .requestFailed(message) = error else { + Issue.record("Expected a normal Grok request failure, got \(String(describing: error))") + return + } + #expect(message.contains("stdin closed")) + } + @Test func `Codex RPC shutdown kills an app-server child that ignores SIGTERM`() async throws { let temporaryDirectory = FileManager.default.temporaryDirectory From 20b530a59ed737b605c0cfdaa4b07f7c5b15b9f0 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Thu, 20 Aug 2026 03:11:24 -0700 Subject: [PATCH 2/2] test: repin UsageFetcher codex identity anchor after pipe-write refactor --- Tests/CodexBarTests/ProviderArchitectureGatekeeperTests.swift | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Tests/CodexBarTests/ProviderArchitectureGatekeeperTests.swift b/Tests/CodexBarTests/ProviderArchitectureGatekeeperTests.swift index d275c1428f..cca24b122b 100644 --- a/Tests/CodexBarTests/ProviderArchitectureGatekeeperTests.swift +++ b/Tests/CodexBarTests/ProviderArchitectureGatekeeperTests.swift @@ -1515,7 +1515,7 @@ struct ProviderArchitectureGatekeeperTests { reason: "This tagged diagnostic payload encodes MiniMax details under the matching wire key."), SuppressedProviderReference( path: "Sources/CodexBarCore/UsageFetcher.swift", - line: 1496, + line: 1500, anchor: "providerID: .codex,", expectedProviderIDs: ["codex"], reason: "This provider-specific core branch passes its already-selected identity to a shared helper."),