Prevent RPC pipe writes from aborting after child exit

CHANGELOG.md

@@ -5,6 +5,7 @@
 - Fixed custom menu bar line breaks so provider icons stay above stacked usage percentages in the menu bar and layout preview (#3089).
 - Fixed the native blue selection highlight reappearing on provider cards after cached provider switches: cross-class cached rows now replace the item shell so the highlight override survives (#2998, #3091, #3093). Thanks @kiranmagic7!
 - Menu bar conditionals can now test every comparable block, not just usage percentages: time to reset, run-out estimate, pace, credit balance, today/30-day cost, and the direct primary/secondary/tertiary lanes, with a used/remaining select wherever both readings exist (so "session > 50% used **and** session resets in < 2h" or "balance remaining >= 5" are expressible). Ships an "Auto % / Resets in" default that shows the percentage while the lane has headroom and the reset countdown once it is spent (#3076).
+- 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!

Sources/CodexBarCore/Host/Process/RPCChildProcessTeardown.swift

@@ -1,14 +1,48 @@
+#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`.
     ///
     /// Closes the child's stdin first (codex app-server and grok agent stdio exit on EOF),
     /// 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)
     }
 }

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] {

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<Data>
@@ -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] {

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."),

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