improve(macos): reuse the Codex App Server client (#116325)

* perf(macos): reuse Codex app server client

* refactor(macos): split Codex app server client

* fix(macos): enforce Codex frame limits per message

* refactor(macos): remove obsolete Codex wrappers

* test(macos): lock Codex wall-clock deadlines

apps/macos/Sources/OpenClaw/NodeMode/MacNodeCodexThreadCatalog.swift

@@ -1,5 +1,4 @@
 import CoreFoundation
-import Darwin
 import Foundation
 
 enum MacNodeCodexThreadCatalogContract {
@@ -11,7 +10,7 @@ enum MacNodeCodexThreadCatalogContract {
 }
 
 enum MacNodeCodexThreadCatalog {
-    struct ResolvedInvocation: Equatable {
+    struct ResolvedInvocation: Equatable, Sendable {
         var executable: String
         var arguments: [String]
         var cwd: URL?
@@ -140,6 +139,7 @@ enum MacNodeCodexThreadCatalog {
         .appendingPathComponent("Applications/Codex Beta.app/Contents/Resources/codex")
         .path
     static let defaultTimeoutSeconds: Double = 60
+    static let defaultIdleTimeoutSeconds: Double = 30
     private static let maxSessionIdLength = 256
     private static let maxSessionNameLength = 500
     private static let maxCwdLength = 4096
@@ -173,24 +173,20 @@ enum MacNodeCodexThreadCatalog {
         var archived: Bool
     }
 
-    static func list(paramsJSON: String?) async throws -> String {
-        try await self.list(paramsJSON: paramsJSON) {
-            OpenClawConfigFile.loadDict()
+    static func list(
+        paramsJSON: String?,
+        loadRoot: () -> [String: Any]) async throws -> String
+    {
+        let client = CodexAppServerThreadClient()
+        return try await self.withEphemeralClient(client) {
+            try await self.list(paramsJSON: paramsJSON, loadRoot: loadRoot, client: client)
         }
     }
 
-    static func turns(paramsJSON: String?) async throws -> String {
-        let params = try decodeTurnParams(paramsJSON)
-        let root = OpenClawConfigFile.loadDict()
-        guard self.shouldAdvertise(root: root) else {
-            throw CatalogError.catalogDisabled
-        }
-        return try await self.turns(params: params, invocation: resolveInvocation(root: root))
-    }
-
     static func list(
         paramsJSON: String?,
-        loadRoot: () -> [String: Any]) async throws -> String
+        loadRoot: () -> [String: Any],
+        client: CodexAppServerThreadClient) async throws -> String
     {
         let params = try self.decodeParams(paramsJSON)
         // Keep authorization and spawn selection on one config snapshot. A second read could
@@ -200,7 +196,23 @@ enum MacNodeCodexThreadCatalog {
             throw CatalogError.catalogDisabled
         }
         let invocation = try self.resolveInvocation(root: root)
-        return try await self.list(params: params, invocation: invocation)
+        return try await self.list(params: params, invocation: invocation, client: client)
+    }
+
+    static func turns(
+        paramsJSON: String?,
+        loadRoot: () -> [String: Any],
+        client: CodexAppServerThreadClient) async throws -> String
+    {
+        let params = try decodeTurnParams(paramsJSON)
+        let root = loadRoot()
+        guard self.shouldAdvertise(root: root) else {
+            throw CatalogError.catalogDisabled
+        }
+        return try await self.turns(
+            params: params,
+            invocation: resolveInvocation(root: root),
+            client: client)
     }
 
     static func turns(
@@ -213,20 +225,25 @@ enum MacNodeCodexThreadCatalog {
         maxLineBytes: Int = 20 * 1024 * 1024) async throws -> String
     {
         let params = try decodeTurnParams(paramsJSON)
-        return try await self.turns(
-            params: params,
-            invocation: ResolvedInvocation(
-                executable: executable,
-                arguments: arguments ?? self.defaultArguments,
-                cwd: cwd,
-                clearEnv: clearEnv),
-            timeoutSeconds: timeoutSeconds,
-            maxLineBytes: maxLineBytes)
+        let client = CodexAppServerThreadClient()
+        return try await self.withEphemeralClient(client) {
+            try await self.turns(
+                params: params,
+                invocation: ResolvedInvocation(
+                    executable: executable,
+                    arguments: arguments ?? self.defaultArguments,
+                    cwd: cwd,
+                    clearEnv: clearEnv),
+                client: client,
+                timeoutSeconds: timeoutSeconds,
+                maxLineBytes: maxLineBytes)
+        }
     }
 
     private static func turns(
         params: TurnParams,
         invocation: ResolvedInvocation,
+        client: CodexAppServerThreadClient,
         timeoutSeconds: Double = MacNodeCodexThreadCatalog.defaultTimeoutSeconds,
         maxLineBytes: Int = 20 * 1024 * 1024) async throws -> String
     {
@@ -234,6 +251,7 @@ enum MacNodeCodexThreadCatalog {
         try await self.requireCatalogThread(
             params.threadId,
             invocation: invocation,
+            client: client,
             deadline: deadline)
         var requestParams: [String: Any] = [
             "threadId": params.threadId,
@@ -244,14 +262,13 @@ enum MacNodeCodexThreadCatalog {
         if let cursor = params.cursor {
             requestParams["cursor"] = cursor
         }
-        let session = try CodexAppServerThreadRequestSession(
+        let resultData = try await client.request(
             invocation: invocation,
             method: "thread/turns/list",
             requestParams: requestParams,
             timeoutSeconds: max(0.01, deadline.timeIntervalSinceNow),
             maxLineBytes: maxLineBytes)
-        let output = try await session.run()
-        guard let payload = String(data: output.resultData, encoding: .utf8) else {
+        guard let payload = String(data: resultData, encoding: .utf8) else {
             throw CatalogError.appServerUnavailable
         }
         return payload
@@ -260,6 +277,7 @@ enum MacNodeCodexThreadCatalog {
     private static func requireCatalogThread(
         _ threadId: String,
         invocation: ResolvedInvocation,
+        client: CodexAppServerThreadClient,
         deadline: Date) async throws
     {
         var cursor: String?
@@ -270,6 +288,7 @@ enum MacNodeCodexThreadCatalog {
             let payload = try await list(
                 params: ListParams(cursor: cursor, limit: 100),
                 invocation: invocation,
+                client: client,
                 timeoutSeconds: remainingTimeout)
             guard let response = try JSONSerialization.jsonObject(with: Data(payload.utf8)) as? [String: Any],
                   let sessions = response["sessions"] as? [[String: Any]]
@@ -317,32 +336,36 @@ enum MacNodeCodexThreadCatalog {
         maxLineBytes: Int = 5 * 1024 * 1024) async throws -> String
     {
         let params = try self.decodeParams(paramsJSON)
-        return try await self.list(
-            params: params,
-            invocation: ResolvedInvocation(
-                executable: executable,
-                arguments: arguments ?? self.defaultArguments,
-                cwd: cwd,
-                clearEnv: clearEnv),
-            timeoutSeconds: timeoutSeconds,
-            maxLineBytes: maxLineBytes)
+        let client = CodexAppServerThreadClient()
+        return try await self.withEphemeralClient(client) {
+            try await self.list(
+                params: params,
+                invocation: ResolvedInvocation(
+                    executable: executable,
+                    arguments: arguments ?? self.defaultArguments,
+                    cwd: cwd,
+                    clearEnv: clearEnv),
+                client: client,
+                timeoutSeconds: timeoutSeconds,
+                maxLineBytes: maxLineBytes)
+        }
     }
 
     private static func list(
         params: ListParams,
         invocation: ResolvedInvocation,
+        client: CodexAppServerThreadClient,
         timeoutSeconds: Double = MacNodeCodexThreadCatalog.defaultTimeoutSeconds,
         maxLineBytes: Int = 5 * 1024 * 1024) async throws -> String
     {
         guard params.searchTerm != nil else {
-            let session = try CodexAppServerThreadRequestSession(
+            let resultData = try await client.request(
                 invocation: invocation,
                 method: "thread/list",
                 requestParams: self.appServerParams(params),
                 timeoutSeconds: timeoutSeconds,
                 maxLineBytes: maxLineBytes)
-            let output = try await session.run()
-            return try self.normalize(listResultData: output.resultData)
+            return try self.normalize(listResultData: resultData)
         }
 
         // Native search also inspects transcript-derived previews. Scan a bounded
@@ -363,15 +386,14 @@ enum MacNodeCodexThreadCatalog {
             var pageParams = params
             pageParams.cursor = cursor
             pageParams.limit = remainingLimit
-            let session = try CodexAppServerThreadRequestSession(
+            let resultData = try await client.request(
                 invocation: invocation,
                 method: "thread/list",
                 requestParams: self.appServerParams(pageParams),
                 timeoutSeconds: remainingTimeout,
                 maxLineBytes: maxLineBytes)
-            let output = try await session.run()
             let page = try self.normalizedResponse(
-                listResultData: output.resultData,
+                listResultData: resultData,
                 searchTerm: params.searchTerm)
             if pageIndex == 0 {
                 backwardsCursor = page.backwardsCursor
@@ -401,6 +423,20 @@ enum MacNodeCodexThreadCatalog {
             nextCursor: nextCursor,
             backwardsCursor: backwardsCursor))
     }
+
+    private static func withEphemeralClient<T>(
+        _ client: CodexAppServerThreadClient,
+        operation: () async throws -> T) async throws -> T
+    {
+        do {
+            let result = try await operation()
+            await client.shutdown()
+            return result
+        } catch {
+            await client.shutdown()
+            throw error
+        }
+    }
 }
 
 extension MacNodeCodexThreadCatalog {
@@ -1207,296 +1243,3 @@ extension MacNodeCodexThreadCatalog {
             overflow: .truncate)
     }
 }
-
-private final class CodexAppServerThreadRequestSession: @unchecked Sendable {
-    struct Output {
-        var resultData: Data
-    }
-
-    private enum Phase {
-        case initialize
-        case request
-    }
-
-    private let process = Process()
-    private let stdinPipe = Pipe()
-    private let stdoutPipe = Pipe()
-    private let stderrPipe = Pipe()
-    private let queue = DispatchQueue(label: "ai.openclaw.codex-thread-catalog")
-    private let requestData: Data
-    private let timeoutSeconds: Double
-    private let maxLineBytes: Int
-    private var continuation: CheckedContinuation<Output, Error>?
-    private var timer: DispatchSourceTimer?
-    private var stdoutBuffer = Data()
-    private var phase = Phase.initialize
-    private var finished = false
-    private var launched = false
-
-    private struct ReadChunk {
-        var data: Data
-        var reachedEOF: Bool
-    }
-
-    init(
-        invocation: MacNodeCodexThreadCatalog.ResolvedInvocation,
-        method: String,
-        requestParams: [String: Any],
-        timeoutSeconds: Double,
-        maxLineBytes: Int) throws
-    {
-        self.process.executableURL = URL(fileURLWithPath: invocation.executable)
-        self.process.arguments = invocation.arguments
-        self.process.currentDirectoryURL = invocation.cwd
-        var environment = ProcessInfo.processInfo.environment
-        environment["PATH"] = CommandResolver.preferredPaths().joined(separator: ":")
-        for key in invocation.clearEnv {
-            environment.removeValue(forKey: key)
-        }
-        self.process.environment = environment
-        self.process.standardInput = self.stdinPipe
-        self.process.standardOutput = self.stdoutPipe
-        self.process.standardError = self.stderrPipe
-        self.timeoutSeconds = max(0.01, timeoutSeconds)
-        self.maxLineBytes = max(1, maxLineBytes)
-        self.requestData = try Self.jsonData([
-            "id": 2,
-            "method": method,
-            "params": requestParams,
-        ])
-    }
-
-    func run() async throws -> Output {
-        try await withTaskCancellationHandler {
-            try await withCheckedThrowingContinuation { continuation in
-                self.queue.async {
-                    self.start(continuation)
-                }
-            }
-        } onCancel: {
-            self.queue.async {
-                self.finish(.failure(CancellationError()))
-            }
-        }
-    }
-
-    private func start(_ continuation: CheckedContinuation<Output, Error>) {
-        guard !self.finished else {
-            continuation.resume(throwing: CancellationError())
-            return
-        }
-        self.continuation = continuation
-        // DispatchSource readability callbacks may be followed by future drain
-        // loops. Keep both pipes non-blocking so an open App Server cannot stall
-        // the catalog handshake after emitting one JSON-RPC frame.
-        Self.setNonBlocking(self.stdoutPipe.fileHandleForReading)
-        Self.setNonBlocking(self.stderrPipe.fileHandleForReading)
-        self.stdoutPipe.fileHandleForReading.readabilityHandler = { [weak self] handle in
-            guard let session = self else { return }
-            session.queue.async { [session] in
-                session.drainStdout(from: handle)
-            }
-        }
-        // Drain stderr so the child cannot block. App Server stderr is deliberately
-        // not forwarded over the Gateway because it may contain local paths.
-        self.stderrPipe.fileHandleForReading.readabilityHandler = { handle in
-            if Self.drainAvailable(from: handle) {
-                handle.readabilityHandler = nil
-            }
-        }
-        self.process.terminationHandler = { [weak self] _ in
-            guard let session = self else { return }
-            session.queue.async { [session] in
-                // A short-lived App Server can exit before its readability callback
-                // is admitted. Drain its final frame before projecting termination.
-                session.drainStdout(from: session.stdoutPipe.fileHandleForReading)
-                guard !session.finished else { return }
-                session.finish(.failure(MacNodeCodexThreadCatalog.CatalogError.appServerUnavailable))
-            }
-        }
-
-        let timer = DispatchSource.makeTimerSource(queue: self.queue)
-        timer.schedule(deadline: .now() + self.timeoutSeconds)
-        timer.setEventHandler { [weak self] in
-            self?.finish(.failure(MacNodeCodexThreadCatalog.CatalogError.timedOut))
-        }
-        self.timer = timer
-        timer.resume()
-
-        do {
-            try self.process.run()
-            self.launched = true
-            try self.write(Self.initializeRequestData())
-        } catch {
-            self.finish(.failure(MacNodeCodexThreadCatalog.CatalogError.appServerUnavailable))
-        }
-    }
-
-    private func drainStdout(from handle: FileHandle) {
-        guard !self.finished else { return }
-        let chunk = Self.readAvailable(from: handle, maxBytes: self.maxLineBytes)
-        if chunk.reachedEOF {
-            handle.readabilityHandler = nil
-        }
-        guard !chunk.data.isEmpty else { return }
-        self.consumeStdout(chunk.data)
-    }
-
-    private func consumeStdout(_ data: Data) {
-        guard !self.finished else { return }
-        self.stdoutBuffer.append(data)
-        guard self.stdoutBuffer.count <= self.maxLineBytes else {
-            self.finish(.failure(MacNodeCodexThreadCatalog.CatalogError.responseTooLarge))
-            return
-        }
-
-        while let newline = self.stdoutBuffer.firstIndex(of: 0x0A) {
-            let line = self.stdoutBuffer.prefix(upTo: newline)
-            self.stdoutBuffer.removeSubrange(...newline)
-            guard !line.isEmpty else { continue }
-            self.handleLine(Data(line))
-            if self.finished {
-                return
-            }
-        }
-    }
-
-    private func handleLine(_ data: Data) {
-        guard let message = try? JSONSerialization.jsonObject(with: data) as? [String: Any],
-              let id = (message["id"] as? NSNumber)?.intValue
-        else { return }
-
-        if message["error"] is [String: Any] {
-            self.finish(.failure(MacNodeCodexThreadCatalog.CatalogError.appServerUnavailable))
-            return
-        }
-
-        switch (self.phase, id) {
-        case (.initialize, 1):
-            guard message["result"] is [String: Any] else {
-                self.finish(.failure(MacNodeCodexThreadCatalog.CatalogError.appServerUnavailable))
-                return
-            }
-            self.phase = .request
-            do {
-                try self.write(Self.initializedNotificationData())
-                try self.write(self.requestData)
-            } catch {
-                self.finish(.failure(MacNodeCodexThreadCatalog.CatalogError.appServerUnavailable))
-            }
-        case (.request, 2):
-            guard let result = message["result"] as? [String: Any],
-                  let resultData = try? Self.jsonData(result)
-            else {
-                self.finish(.failure(MacNodeCodexThreadCatalog.CatalogError.appServerUnavailable))
-                return
-            }
-            self.finish(.success(Output(resultData: resultData)))
-        default:
-            break
-        }
-    }
-
-    private func write(_ data: Data) throws {
-        var frame = data
-        frame.append(0x0A)
-        try self.stdinPipe.fileHandleForWriting.write(contentsOf: frame)
-    }
-
-    private func finish(_ result: Result<Output, Error>) {
-        guard !self.finished else { return }
-        self.finished = true
-        self.timer?.cancel()
-        self.timer = nil
-        self.stdoutPipe.fileHandleForReading.readabilityHandler = nil
-        self.stderrPipe.fileHandleForReading.readabilityHandler = nil
-        try? self.stdinPipe.fileHandleForWriting.close()
-        if self.launched, self.process.isRunning {
-            self.process.terminate()
-        }
-        guard let continuation = self.continuation else { return }
-        self.continuation = nil
-        continuation.resume(with: result)
-    }
-
-    private static func initializeRequestData() throws -> Data {
-        try self.jsonData([
-            "id": 1,
-            "method": "initialize",
-            "params": [
-                "clientInfo": [
-                    "name": "openclaw_macos",
-                    "title": "OpenClaw macOS Node",
-                    "version": GatewayEnvironment.appVersionString() ?? "unknown",
-                ],
-                "capabilities": ["experimentalApi": true],
-            ],
-        ])
-    }
-
-    private static func initializedNotificationData() throws -> Data {
-        try self.jsonData(["method": "initialized"])
-    }
-
-    private static func jsonData(_ object: Any) throws -> Data {
-        try JSONSerialization.data(withJSONObject: object)
-    }
-
-    private static func readAvailable(from handle: FileHandle, maxBytes: Int) -> ReadChunk {
-        // FileHandle.read(upToCount:) can wait for EOF despite a readability callback.
-        // The descriptor is non-blocking, so drain one complete JSONL frame (or
-        // the response cap plus one byte) without waiting for the App Server to exit.
-        var data = Data()
-        let captureLimit = maxBytes == Int.max ? Int.max : maxBytes + 1
-        var buffer = [UInt8](repeating: 0, count: 64 * 1024)
-        while true {
-            let count = buffer.withUnsafeMutableBytes { bytes in
-                Darwin.read(handle.fileDescriptor, bytes.baseAddress, bytes.count)
-            }
-            if count > 0 {
-                let remaining = max(0, captureLimit - data.count)
-                data.append(contentsOf: buffer.prefix(min(count, remaining)))
-                if data.count > maxBytes {
-                    return ReadChunk(data: data, reachedEOF: false)
-                }
-                continue
-            }
-            if count == 0 {
-                return ReadChunk(data: data, reachedEOF: true)
-            }
-            if errno == EINTR { continue }
-            if errno == EAGAIN || errno == EWOULDBLOCK {
-                return ReadChunk(data: data, reachedEOF: false)
-            }
-            return ReadChunk(data: data, reachedEOF: true)
-        }
-    }
-
-    private static func drainAvailable(from handle: FileHandle) -> Bool {
-        var buffer = [UInt8](repeating: 0, count: 64 * 1024)
-        while true {
-            let count = buffer.withUnsafeMutableBytes { bytes in
-                Darwin.read(handle.fileDescriptor, bytes.baseAddress, bytes.count)
-            }
-            if count > 0 {
-                continue
-            }
-            if count == 0 {
-                return true
-            }
-            if errno == EINTR { continue }
-            if errno == EAGAIN || errno == EWOULDBLOCK {
-                return false
-            }
-            return true
-        }
-    }
-
-    private static func setNonBlocking(_ handle: FileHandle) {
-        let descriptor = handle.fileDescriptor
-        let flags = Darwin.fcntl(descriptor, F_GETFL)
-        if flags >= 0 {
-            _ = Darwin.fcntl(descriptor, F_SETFL, flags | O_NONBLOCK)
-        }
-    }
-}

apps/macos/Sources/OpenClaw/NodeMode/MacNodeCodexThreadCatalogClient.swift

@@ -0,0 +1,597 @@
+import Darwin
+import Foundation
+
+final class MacNodeCodexThreadCatalogClient: @unchecked Sendable {
+    private let loadRoot: () -> [String: Any]
+    private let appServerClient: CodexAppServerThreadClient
+
+    init(
+        idleTimeoutSeconds: Double = MacNodeCodexThreadCatalog.defaultIdleTimeoutSeconds,
+        loadRoot: @escaping () -> [String: Any] = { OpenClawConfigFile.loadDict() })
+    {
+        self.loadRoot = loadRoot
+        self.appServerClient = CodexAppServerThreadClient(
+            idleTimeoutSeconds: idleTimeoutSeconds)
+    }
+
+    func list(paramsJSON: String?) async throws -> String {
+        try await MacNodeCodexThreadCatalog.list(
+            paramsJSON: paramsJSON,
+            loadRoot: self.loadRoot,
+            client: self.appServerClient)
+    }
+
+    func turns(paramsJSON: String?) async throws -> String {
+        try await MacNodeCodexThreadCatalog.turns(
+            paramsJSON: paramsJSON,
+            loadRoot: self.loadRoot,
+            client: self.appServerClient)
+    }
+
+    func shutdown() async {
+        await self.appServerClient.shutdown()
+    }
+}
+
+final class CodexAppServerThreadClient: @unchecked Sendable {
+    private final class CancellationState: @unchecked Sendable {
+        private let lock = NSLock()
+        private var cancelled = false
+
+        func cancel() {
+            self.lock.lock()
+            self.cancelled = true
+            self.lock.unlock()
+        }
+
+        func isCancelled() -> Bool {
+            self.lock.lock()
+            defer { self.lock.unlock() }
+            return self.cancelled
+        }
+    }
+
+    private final class PendingRequest: @unchecked Sendable {
+        let token: UUID
+        let invocation: MacNodeCodexThreadCatalog.ResolvedInvocation
+        let method: String
+        let requestParamsData: Data
+        let maxLineBytes: Int
+        var requestID: Int?
+        var requestData: Data?
+        var continuation: CheckedContinuation<Data, Error>?
+        var timer: DispatchSourceTimer?
+
+        init(
+            token: UUID,
+            invocation: MacNodeCodexThreadCatalog.ResolvedInvocation,
+            method: String,
+            requestParamsData: Data,
+            maxLineBytes: Int,
+            continuation: CheckedContinuation<Data, Error>)
+        {
+            self.token = token
+            self.invocation = invocation
+            self.method = method
+            self.requestParamsData = requestParamsData
+            self.maxLineBytes = maxLineBytes
+            self.continuation = continuation
+        }
+    }
+
+    private final class Connection: @unchecked Sendable {
+        let generation = UUID()
+        let invocation: MacNodeCodexThreadCatalog.ResolvedInvocation
+        let initializeRequestID: Int
+        let process = Process()
+        let stdinPipe = Pipe()
+        let stdoutPipe = Pipe()
+        let stderrPipe = Pipe()
+        var stdoutBuffer = Data()
+        var initialized = false
+        var exited = false
+        var stdoutReachedEOF = false
+
+        init(
+            invocation: MacNodeCodexThreadCatalog.ResolvedInvocation,
+            initializeRequestID: Int)
+        {
+            self.invocation = invocation
+            self.initializeRequestID = initializeRequestID
+        }
+    }
+
+    private static let maxQueuedRequests = 64
+    private static let maxStdoutDrainBytes = 256 * 1024
+
+    private let queue = DispatchQueue(label: "ai.openclaw.codex-thread-catalog")
+    private let idleTimeoutSeconds: Double
+    private let idleReadLimit: Int
+    private var nextRequestID = 1
+    private var pending: [PendingRequest] = []
+    private var active: PendingRequest?
+    private var connection: Connection?
+    private var idleTimer: DispatchSourceTimer?
+
+    init(
+        idleTimeoutSeconds: Double = MacNodeCodexThreadCatalog.defaultIdleTimeoutSeconds,
+        idleReadLimit: Int = 20 * 1024 * 1024)
+    {
+        self.idleTimeoutSeconds = max(0.01, idleTimeoutSeconds)
+        self.idleReadLimit = max(1, idleReadLimit)
+    }
+
+    deinit {
+        self.cancelIdleTimer()
+        self.stopConnection()
+    }
+
+    func request(
+        invocation: MacNodeCodexThreadCatalog.ResolvedInvocation,
+        method: String,
+        requestParams: [String: Any],
+        timeoutSeconds: Double,
+        maxLineBytes: Int) async throws -> Data
+    {
+        try Task.checkCancellation()
+        let requestParamsData = try Self.jsonData(requestParams)
+        let token = UUID()
+        let cancellationState = CancellationState()
+        let result: Data = try await withTaskCancellationHandler {
+            try await withCheckedThrowingContinuation { continuation in
+                self.queue.async {
+                    guard !cancellationState.isCancelled() else {
+                        continuation.resume(throwing: CancellationError())
+                        return
+                    }
+                    guard self.pending.count + (self.active == nil ? 0 : 1) <
+                        Self.maxQueuedRequests
+                    else {
+                        continuation.resume(
+                            throwing: MacNodeCodexThreadCatalog.CatalogError.appServerUnavailable)
+                        return
+                    }
+                    let request = PendingRequest(
+                        token: token,
+                        invocation: invocation,
+                        method: method,
+                        requestParamsData: requestParamsData,
+                        maxLineBytes: max(1, maxLineBytes),
+                        continuation: continuation)
+                    // Callers pass the operation's remaining wall-clock deadline.
+                    // Queue wait therefore counts, matching the former one-shot session.
+                    request.timer = self.makeRequestTimer(
+                        token: token,
+                        timeoutSeconds: timeoutSeconds)
+                    self.pending.append(request)
+                    self.cancelIdleTimer()
+                    self.startNextIfNeeded()
+                }
+            }
+        } onCancel: {
+            cancellationState.cancel()
+            self.queue.async {
+                self.cancel(token: token)
+            }
+        }
+        try Task.checkCancellation()
+        return result
+    }
+
+    func shutdown() async {
+        await withCheckedContinuation { continuation in
+            self.queue.async {
+                self.cancelIdleTimer()
+                if let active = self.active {
+                    self.active = nil
+                    self.complete(active, with: .failure(CancellationError()))
+                }
+                let pending = self.pending
+                self.pending.removeAll()
+                for request in pending {
+                    self.complete(request, with: .failure(CancellationError()))
+                }
+                self.stopConnection()
+                continuation.resume()
+            }
+        }
+    }
+
+    private func takeRequestIDOnQueue() -> Int {
+        let requestID = self.nextRequestID
+        self.nextRequestID = requestID == Int.max ? 1 : requestID + 1
+        return requestID
+    }
+
+    private func makeRequestTimer(token: UUID, timeoutSeconds: Double) -> DispatchSourceTimer {
+        let timer = DispatchSource.makeTimerSource(queue: self.queue)
+        timer.schedule(deadline: .now() + max(0.01, timeoutSeconds))
+        timer.setEventHandler { [weak self] in
+            self?.timeout(token: token)
+        }
+        timer.resume()
+        return timer
+    }
+
+    private func startNextIfNeeded() {
+        guard self.active == nil, !self.pending.isEmpty else {
+            if self.active == nil, self.pending.isEmpty {
+                self.scheduleIdleShutdown()
+            }
+            return
+        }
+        let request = self.pending.removeFirst()
+        self.active = request
+        if let connection = self.connection,
+           connection.exited ||
+           !connection.process.isRunning ||
+           connection.invocation != request.invocation
+        {
+            self.stopConnection()
+        }
+        guard let connection = self.connection else {
+            self.startConnection(for: request)
+            return
+        }
+        guard connection.initialized else { return }
+        self.sendActiveRequest(over: connection)
+    }
+
+    private func startConnection(for request: PendingRequest) {
+        let connection = Connection(
+            invocation: request.invocation,
+            initializeRequestID: self.takeRequestIDOnQueue())
+        self.connection = connection
+        let process = connection.process
+        process.executableURL = URL(fileURLWithPath: request.invocation.executable)
+        process.arguments = request.invocation.arguments
+        process.currentDirectoryURL = request.invocation.cwd
+        var environment = ProcessInfo.processInfo.environment
+        environment["PATH"] = CommandResolver.preferredPaths().joined(separator: ":")
+        for key in request.invocation.clearEnv {
+            environment.removeValue(forKey: key)
+        }
+        process.environment = environment
+        process.standardInput = connection.stdinPipe
+        process.standardOutput = connection.stdoutPipe
+        process.standardError = connection.stderrPipe
+
+        // DispatchSource readability callbacks may be followed by future drain
+        // loops. Keep both pipes non-blocking so an open App Server cannot stall
+        // the catalog handshake after emitting one JSON-RPC frame.
+        Self.setNonBlocking(connection.stdoutPipe.fileHandleForReading)
+        Self.setNonBlocking(connection.stderrPipe.fileHandleForReading)
+        let generation = connection.generation
+        connection.stdoutPipe.fileHandleForReading.readabilityHandler = { [weak self] handle in
+            guard let self else { return }
+            self.queue.async {
+                self.drainStdout(from: handle, generation: generation)
+            }
+        }
+        connection.stderrPipe.fileHandleForReading.readabilityHandler = { handle in
+            if Self.drainAvailable(from: handle) {
+                handle.readabilityHandler = nil
+            }
+        }
+        process.terminationHandler = { [weak self] _ in
+            guard let self else { return }
+            self.queue.async {
+                self.handleTermination(generation: generation)
+            }
+        }
+
+        do {
+            try process.run()
+            try self.write(
+                Self.initializeRequestData(id: connection.initializeRequestID),
+                over: connection)
+        } catch {
+            self.finishActive(
+                .failure(MacNodeCodexThreadCatalog.CatalogError.appServerUnavailable),
+                restartConnection: true)
+        }
+    }
+
+    private func sendActiveRequest(over connection: Connection) {
+        guard let active = self.active else { return }
+        do {
+            if active.requestData == nil {
+                let requestID = self.takeRequestIDOnQueue()
+                let requestParams = try JSONSerialization.jsonObject(
+                    with: active.requestParamsData)
+                active.requestID = requestID
+                active.requestData = try Self.jsonData([
+                    "id": requestID,
+                    "method": active.method,
+                    "params": requestParams,
+                ])
+            }
+            guard let requestData = active.requestData else {
+                throw MacNodeCodexThreadCatalog.CatalogError.appServerUnavailable
+            }
+            try self.write(requestData, over: connection)
+        } catch {
+            self.finishActive(
+                .failure(MacNodeCodexThreadCatalog.CatalogError.appServerUnavailable),
+                restartConnection: true)
+        }
+    }
+
+    private func drainStdout(from handle: FileHandle, generation: UUID) {
+        guard let connection = self.connection, connection.generation == generation else { return }
+        var drainedBytes = 0
+        var buffer = [UInt8](repeating: 0, count: 64 * 1024)
+        while true {
+            // Yield after a bounded batch so request timers, cancellation, and
+            // shutdown work queued behind a noisy App Server can still run.
+            if drainedBytes >= Self.maxStdoutDrainBytes {
+                self.queue.async { [weak self] in
+                    self?.drainStdout(from: handle, generation: generation)
+                }
+                return
+            }
+            let count = buffer.withUnsafeMutableBytes { bytes in
+                Darwin.read(handle.fileDescriptor, bytes.baseAddress, bytes.count)
+            }
+            if count > 0 {
+                drainedBytes += count
+                self.consumeStdout(
+                    Data(buffer.prefix(count)),
+                    connection: connection)
+                guard self.connection?.generation == generation else { return }
+                continue
+            }
+            if count == 0 {
+                connection.stdoutReachedEOF = true
+                handle.readabilityHandler = nil
+                self.finishTerminatedConnectionIfNeeded(generation: generation)
+                return
+            }
+            if errno == EINTR { continue }
+            if errno == EAGAIN || errno == EWOULDBLOCK {
+                if connection.exited {
+                    self.queue.asyncAfter(deadline: .now() + .milliseconds(1)) { [weak self] in
+                        self?.drainStdout(from: handle, generation: generation)
+                    }
+                }
+                return
+            }
+            connection.stdoutReachedEOF = true
+            handle.readabilityHandler = nil
+            self.finishTerminatedConnectionIfNeeded(generation: generation)
+            return
+        }
+    }
+
+    private func consumeStdout(
+        _ data: Data,
+        connection: Connection)
+    {
+        connection.stdoutBuffer.append(data)
+
+        while let newline = connection.stdoutBuffer.firstIndex(of: 0x0A) {
+            let line = connection.stdoutBuffer.prefix(upTo: newline)
+            let maxLineBytes = self.active?.maxLineBytes ?? self.idleReadLimit
+            guard line.count <= maxLineBytes else {
+                self.rejectOversizedFrame()
+                return
+            }
+            connection.stdoutBuffer.removeSubrange(...newline)
+            guard !line.isEmpty else { continue }
+            self.handleLine(Data(line), connection: connection)
+            guard self.connection?.generation == connection.generation else { return }
+        }
+        let maxLineBytes = self.active?.maxLineBytes ?? self.idleReadLimit
+        guard connection.stdoutBuffer.count <= maxLineBytes else {
+            self.rejectOversizedFrame()
+            return
+        }
+    }
+
+    private func rejectOversizedFrame() {
+        if self.active == nil {
+            self.stopConnection()
+            self.startNextIfNeeded()
+        } else {
+            self.finishActive(
+                .failure(MacNodeCodexThreadCatalog.CatalogError.responseTooLarge),
+                restartConnection: true)
+        }
+    }
+
+    private func handleLine(_ data: Data, connection: Connection) {
+        guard let message = try? JSONSerialization.jsonObject(with: data) as? [String: Any],
+              let id = (message["id"] as? NSNumber)?.intValue
+        else { return }
+
+        if id == connection.initializeRequestID {
+            guard message["error"] == nil, message["result"] is [String: Any] else {
+                self.finishActive(
+                    .failure(MacNodeCodexThreadCatalog.CatalogError.appServerUnavailable),
+                    restartConnection: true)
+                return
+            }
+            connection.initialized = true
+            do {
+                try self.write(Self.initializedNotificationData(), over: connection)
+                self.sendActiveRequest(over: connection)
+            } catch {
+                self.finishActive(
+                    .failure(MacNodeCodexThreadCatalog.CatalogError.appServerUnavailable),
+                    restartConnection: true)
+            }
+            return
+        }
+
+        guard let active = self.active, id == active.requestID else { return }
+        guard message["error"] == nil,
+              let result = message["result"] as? [String: Any],
+              let resultData = try? Self.jsonData(result)
+        else {
+            self.finishActive(
+                .failure(MacNodeCodexThreadCatalog.CatalogError.appServerUnavailable),
+                restartConnection: message["error"] == nil)
+            return
+        }
+        self.finishActive(.success(resultData), restartConnection: false)
+    }
+
+    private func handleTermination(generation: UUID) {
+        guard let connection = self.connection, connection.generation == generation else { return }
+        connection.exited = true
+        // A short-lived server can exit before its readability callback runs.
+        // Drain its final frame before projecting termination onto the request.
+        self.drainStdout(
+            from: connection.stdoutPipe.fileHandleForReading,
+            generation: generation)
+        self.finishTerminatedConnectionIfNeeded(generation: generation)
+    }
+
+    private func finishTerminatedConnectionIfNeeded(generation: UUID) {
+        guard let connection = self.connection,
+              connection.generation == generation,
+              connection.exited,
+              connection.stdoutReachedEOF
+        else { return }
+        if self.active != nil {
+            self.finishActive(
+                .failure(MacNodeCodexThreadCatalog.CatalogError.appServerUnavailable),
+                restartConnection: true)
+        } else {
+            self.stopConnection()
+            self.startNextIfNeeded()
+        }
+    }
+
+    private func timeout(token: UUID) {
+        if let index = self.pending.firstIndex(where: { $0.token == token }) {
+            let request = self.pending.remove(at: index)
+            self.complete(
+                request,
+                with: .failure(MacNodeCodexThreadCatalog.CatalogError.timedOut))
+            return
+        }
+        guard self.active?.token == token else { return }
+        self.finishActive(
+            .failure(MacNodeCodexThreadCatalog.CatalogError.timedOut),
+            restartConnection: true)
+    }
+
+    private func cancel(token: UUID) {
+        if let index = self.pending.firstIndex(where: { $0.token == token }) {
+            let request = self.pending.remove(at: index)
+            self.complete(request, with: .failure(CancellationError()))
+            return
+        }
+        guard self.active?.token == token else { return }
+        self.finishActive(.failure(CancellationError()), restartConnection: true)
+    }
+
+    private func finishActive(
+        _ result: Result<Data, Error>,
+        restartConnection: Bool)
+    {
+        guard let active = self.active else { return }
+        self.active = nil
+        self.complete(active, with: result)
+        if restartConnection {
+            self.stopConnection()
+        }
+        self.startNextIfNeeded()
+    }
+
+    private func complete(_ request: PendingRequest, with result: Result<Data, Error>) {
+        request.timer?.cancel()
+        request.timer = nil
+        guard let continuation = request.continuation else { return }
+        request.continuation = nil
+        continuation.resume(with: result)
+    }
+
+    private func scheduleIdleShutdown() {
+        guard self.connection != nil, self.idleTimer == nil else { return }
+        let timer = DispatchSource.makeTimerSource(queue: self.queue)
+        timer.schedule(deadline: .now() + self.idleTimeoutSeconds)
+        timer.setEventHandler { [weak self] in
+            guard let self, self.active == nil, self.pending.isEmpty else { return }
+            self.cancelIdleTimer()
+            self.stopConnection()
+        }
+        self.idleTimer = timer
+        timer.resume()
+    }
+
+    private func cancelIdleTimer() {
+        self.idleTimer?.cancel()
+        self.idleTimer = nil
+    }
+
+    private func stopConnection() {
+        guard let connection = self.connection else { return }
+        self.connection = nil
+        connection.stdoutPipe.fileHandleForReading.readabilityHandler = nil
+        connection.stderrPipe.fileHandleForReading.readabilityHandler = nil
+        connection.process.terminationHandler = nil
+        try? connection.stdinPipe.fileHandleForWriting.close()
+        if connection.process.isRunning {
+            connection.process.terminate()
+        }
+    }
+
+    private func write(_ data: Data, over connection: Connection) throws {
+        var frame = data
+        frame.append(0x0A)
+        try connection.stdinPipe.fileHandleForWriting.write(contentsOf: frame)
+    }
+
+    private static func initializeRequestData(id: Int) throws -> Data {
+        try self.jsonData([
+            "id": id,
+            "method": "initialize",
+            "params": [
+                "clientInfo": [
+                    "name": "openclaw_macos",
+                    "title": "OpenClaw macOS Node",
+                    "version": GatewayEnvironment.appVersionString() ?? "unknown",
+                ],
+                "capabilities": ["experimentalApi": true],
+            ],
+        ])
+    }
+
+    private static func initializedNotificationData() throws -> Data {
+        try self.jsonData(["method": "initialized"])
+    }
+
+    private static func jsonData(_ object: Any) throws -> Data {
+        try JSONSerialization.data(withJSONObject: object)
+    }
+
+    private static func drainAvailable(from handle: FileHandle) -> Bool {
+        var buffer = [UInt8](repeating: 0, count: 64 * 1024)
+        while true {
+            let count = buffer.withUnsafeMutableBytes { bytes in
+                Darwin.read(handle.fileDescriptor, bytes.baseAddress, bytes.count)
+            }
+            if count > 0 {
+                continue
+            }
+            if count == 0 {
+                return true
+            }
+            if errno == EINTR { continue }
+            if errno == EAGAIN || errno == EWOULDBLOCK {
+                return false
+            }
+            return true
+        }
+    }
+
+    private static func setNonBlocking(_ handle: FileHandle) {
+        let descriptor = handle.fileDescriptor
+        let flags = Darwin.fcntl(descriptor, F_GETFL)
+        if flags >= 0 {
+            _ = Darwin.fcntl(descriptor, F_SETFL, flags | O_NONBLOCK)
+        }
+    }
+}

apps/macos/Sources/OpenClaw/NodeMode/MacNodeRuntime.swift

@@ -119,8 +119,9 @@ actor MacNodeRuntime {
     private let computerControlEnabled: @Sendable () -> Bool
     private let canvasHostedSurfaceResolver: MacNodeCanvasHostedSurfaceResolver
     private let codexThreadCatalogEnabled: @Sendable () -> Bool
-    private let codexThreadListRequest: @Sendable (String?) async throws -> String
-    private let codexThreadTurnsRequest: @Sendable (String?) async throws -> String
+    private let codexThreadCatalogClient: MacNodeCodexThreadCatalogClient
+    private let codexThreadListRequest: (@Sendable (String?) async throws -> String)?
+    private let codexThreadTurnsRequest: (@Sendable (String?) async throws -> String)?
     private let claudeSessionCatalogEnabled: @Sendable () -> Bool
     private let claudeSessionListRequest: @Sendable (String?) async throws -> String
     private let claudeSessionReadRequest: @Sendable (String?) async throws -> String
@@ -148,12 +149,9 @@ actor MacNodeRuntime {
         codexThreadCatalogEnabled: @escaping @Sendable () -> Bool = {
             MacNodeCodexThreadCatalog.shouldAdvertise()
         },
-        codexThreadListRequest: @escaping @Sendable (String?) async throws -> String = { paramsJSON in
-            try await MacNodeCodexThreadCatalog.list(paramsJSON: paramsJSON)
-        },
-        codexThreadTurnsRequest: @escaping @Sendable (String?) async throws -> String = { paramsJSON in
-            try await MacNodeCodexThreadCatalog.turns(paramsJSON: paramsJSON)
-        },
+        codexThreadCatalogClient: MacNodeCodexThreadCatalogClient = MacNodeCodexThreadCatalogClient(),
+        codexThreadListRequest: (@Sendable (String?) async throws -> String)? = nil,
+        codexThreadTurnsRequest: (@Sendable (String?) async throws -> String)? = nil,
         claudeSessionCatalogEnabled: @escaping @Sendable () -> Bool = {
             MacNodeClaudeSessionCatalog.shouldAdvertise()
         },
@@ -171,6 +169,7 @@ actor MacNodeRuntime {
             currentSurfaceURL: canvasSurfaceUrl,
             refreshSurfaceURL: refreshCanvasSurfaceUrl)
         self.codexThreadCatalogEnabled = codexThreadCatalogEnabled
+        self.codexThreadCatalogClient = codexThreadCatalogClient
         self.codexThreadListRequest = codexThreadListRequest
         self.codexThreadTurnsRequest = codexThreadTurnsRequest
         self.claudeSessionCatalogEnabled = claudeSessionCatalogEnabled
@@ -259,10 +258,19 @@ actor MacNodeRuntime {
                 code: .unavailable,
                 message: "UNAVAILABLE: Codex session catalog is disabled")
         }
-        let request = req.command == MacNodeCodexThreadCatalogContract.listCommand
-            ? self.codexThreadListRequest
-            : self.codexThreadTurnsRequest
-        let payload = try await request(req.paramsJSON)
+        let payload: String = if req.command == MacNodeCodexThreadCatalogContract.listCommand {
+            if let request = self.codexThreadListRequest {
+                try await request(req.paramsJSON)
+            } else {
+                try await self.codexThreadCatalogClient.list(paramsJSON: req.paramsJSON)
+            }
+        } else {
+            if let request = self.codexThreadTurnsRequest {
+                try await request(req.paramsJSON)
+            } else {
+                try await self.codexThreadCatalogClient.turns(paramsJSON: req.paramsJSON)
+            }
+        }
         return BridgeInvokeResponse(id: req.id, ok: true, payloadJSON: payload)
     }
 

apps/macos/Tests/OpenClawIPCTests/MacNodeCodexThreadCatalogTests.swift

@@ -1,5 +1,6 @@
 import Darwin
 import Foundation
+import OpenClawKit
 import Testing
 @testable import OpenClaw
 
@@ -24,7 +25,11 @@ struct MacNodeCodexThreadCatalogTests {
             capture: URL(fileURLWithPath: executable.path + ".requests"))
     }
 
-    private func listResponseJSON(names: [String], nextCursor: String?) throws -> String {
+    private func listResponseJSON(
+        id: Int = 2,
+        names: [String],
+        nextCursor: String?) throws -> String
+    {
         let encodedNextCursor: Any = if let nextCursor {
             nextCursor
         } else {
@@ -38,7 +43,7 @@ struct MacNodeCodexThreadCatalogTests {
             ]
         }
         let data = try JSONSerialization.data(withJSONObject: [
-            "id": 2,
+            "id": id,
             "result": [
                 "data": threads,
                 "nextCursor": encodedNextCursor,
@@ -48,6 +53,37 @@ struct MacNodeCodexThreadCatalogTests {
         return try #require(String(data: data, encoding: .utf8))
     }
 
+    private func waitForFile(_ url: URL, timeout: Duration = .seconds(2)) async -> Bool {
+        let clock = ContinuousClock()
+        let deadline = clock.now.advanced(by: timeout)
+        while !FileManager.default.fileExists(atPath: url.path), clock.now < deadline {
+            try? await Task.sleep(for: .milliseconds(10))
+        }
+        return FileManager.default.fileExists(atPath: url.path)
+    }
+
+    private func readTrimmed(_ url: URL) throws -> String {
+        try String(contentsOf: url, encoding: .utf8)
+            .trimmingCharacters(in: .whitespacesAndNewlines)
+    }
+
+    private func requestEmptyList(
+        client: CodexAppServerThreadClient,
+        executable: URL,
+        timeoutSeconds: Double = 2,
+        maxLineBytes: Int = 1024 * 1024) async throws -> Data
+    {
+        try await client.request(
+            invocation: MacNodeCodexThreadCatalog.ResolvedInvocation(
+                executable: executable.path,
+                arguments: [],
+                cwd: nil),
+            method: "thread/list",
+            requestParams: ["limit": 1],
+            timeoutSeconds: timeoutSeconds,
+            maxLineBytes: maxLineBytes)
+    }
+
     @Test func `normalizes App Server metadata and drops sensitive thread fields`() throws {
         let raw: [String: Any] = [
             "data": [[
@@ -764,6 +800,74 @@ struct MacNodeCodexThreadCatalogTests {
         #expect(listParams["useStateDbOnly"] as? Bool == false)
     }
 
+    @Test func `Mac node runtime reuses its owned App Server across invokes`() async throws {
+        let fake = try makeFakeCodex(#"""
+        #!/bin/sh
+        count=0
+        [ ! -f "${0}.processes" ] || count=$(cat "${0}.processes")
+        printf '%s\n' "$((count + 1))" > "${0}.processes"
+        IFS= read -r initialize || exit 2
+        printf '%s\n' "$initialize" >> "${0}.requests"
+        id=$(printf '%s\n' "$initialize" | /usr/bin/sed -E 's/.*"id":([0-9]+).*/\1/')
+        printf '{"id":%s,"result":{}}\n' "$id"
+        IFS= read -r initialized || exit 3
+        printf '%s\n' "$initialized" >> "${0}.requests"
+        while IFS= read -r request; do
+          printf '%s\n' "$request" >> "${0}.requests"
+          id=$(printf '%s\n' "$request" | /usr/bin/sed -E 's/.*"id":([0-9]+).*/\1/')
+          printf '{"id":%s,"result":{"data":[],"nextCursor":null,"backwardsCursor":null}}\n' "$id"
+        done
+        """#)
+        defer { try? FileManager.default.removeItem(at: fake.directory) }
+        let root: [String: Any] = [
+            "plugins": [
+                "entries": [
+                    "codex": [
+                        "enabled": true,
+                        "config": [
+                            "supervision": ["enabled": true],
+                            "appServer": [
+                                "transport": "stdio",
+                                "homeScope": "user",
+                                "command": fake.executable.path,
+                            ],
+                        ],
+                    ],
+                ],
+            ],
+        ]
+        let client = MacNodeCodexThreadCatalogClient(
+            idleTimeoutSeconds: 10,
+            loadRoot: { root })
+        let runtime = MacNodeRuntime(
+            codexThreadCatalogEnabled: { true },
+            codexThreadCatalogClient: client)
+
+        let first = await runtime.handleInvoke(BridgeInvokeRequest(
+            id: "first",
+            command: MacNodeCodexThreadCatalogContract.listCommand))
+        let second = await runtime.handleInvoke(BridgeInvokeRequest(
+            id: "second",
+            command: MacNodeCodexThreadCatalogContract.listCommand))
+
+        #expect(first.ok)
+        #expect(second.ok)
+        #expect(try self.readTrimmed(
+            URL(fileURLWithPath: fake.executable.path + ".processes")) == "1")
+        let captured = try String(contentsOf: fake.capture, encoding: .utf8)
+            .split(whereSeparator: \.isNewline)
+            .map { try #require(
+                JSONSerialization.jsonObject(with: Data($0.utf8)) as? [String: Any]) }
+        #expect(captured.map { $0["method"] as? String } == [
+            "initialize",
+            "initialized",
+            "thread/list",
+            "thread/list",
+        ])
+        #expect(captured.compactMap { ($0["id"] as? NSNumber)?.intValue } == [1, 2, 3])
+        await client.shutdown()
+    }
+
     @Test func `reads one paginated transcript turn page from App Server`() async throws {
         let fake = try makeFakeCodex(#"""
         #!/bin/sh
@@ -778,17 +882,12 @@ struct MacNodeCodexThreadCatalogTests {
         printf '%s\n' '{"id":1,"result":{}}'
         IFS= read -r initialized || exit 3
         printf '%s\n' "$initialized" >> "$capture"
-        IFS= read -r request || exit 4
-        printf '%s\n' "$request" >> "$capture"
-        case "$count" in
-          1)
-            printf '%s\n' '{"id":2,"result":{"data":[{"id":"thread-1","name":"Task","status":{"type":"notLoaded"}}],"nextCursor":null,"backwardsCursor":null}}'
-            ;;
-          2)
-            printf '%s\n' '{"id":2,"result":{"data":[{"id":"turn-1","items":[{"id":"item-1","type":"agentMessage","text":"full answer"}]}],"nextCursor":"turns-2","backwardsCursor":null}}'
-            ;;
-          *) exit 5 ;;
-        esac
+        IFS= read -r list || exit 4
+        printf '%s\n' "$list" >> "$capture"
+        printf '%s\n' '{"id":2,"result":{"data":[{"id":"thread-1","name":"Task","status":{"type":"notLoaded"}}],"nextCursor":null,"backwardsCursor":null}}'
+        IFS= read -r turns || exit 5
+        printf '%s\n' "$turns" >> "$capture"
+        printf '%s\n' '{"id":3,"result":{"data":[{"id":"turn-1","items":[{"id":"item-1","type":"agentMessage","text":"full answer"}]}],"nextCursor":"turns-2","backwardsCursor":null}}'
         sleep 1
         """#)
         defer { try? FileManager.default.removeItem(at: fake.directory) }
@@ -805,13 +904,14 @@ struct MacNodeCodexThreadCatalogTests {
         let captured = try String(contentsOf: fake.capture, encoding: .utf8)
             .split(whereSeparator: \.isNewline)
             .map { try JSONSerialization.jsonObject(with: Data($0.utf8)) as? [String: Any] }
-        #expect(captured.count == 6)
+        #expect(captured.count == 4)
         let initializeParams = try #require(captured[0]?["params"] as? [String: Any])
         let capabilities = try #require(initializeParams["capabilities"] as? [String: Any])
         #expect(capabilities["experimentalApi"] as? Bool == true)
         #expect(captured[2]?["method"] as? String == "thread/list")
-        #expect(captured[5]?["method"] as? String == "thread/turns/list")
-        let params = try #require(captured[5]?["params"] as? [String: Any])
+        #expect(captured[3]?["method"] as? String == "thread/turns/list")
+        #expect(captured.compactMap { ($0?["id"] as? NSNumber)?.intValue } == [1, 2, 3])
+        let params = try #require(captured[3]?["params"] as? [String: Any])
         #expect(params["threadId"] as? String == "thread-1")
         #expect(params["cursor"] as? String == "turns-1")
         #expect(params["limit"] as? Int == 25)
@@ -821,32 +921,33 @@ struct MacNodeCodexThreadCatalogTests {
 
     @Test func `title search fills one result page across bounded native pages`() async throws {
         let first = try listResponseJSON(
+            id: 2,
             names: ["Target one", "Other one", "Other two"],
             nextCursor: "cursor-1")
         let second = try listResponseJSON(
+            id: 3,
             names: ["Other three", "Other four"],
             nextCursor: "cursor-2")
         let third = try listResponseJSON(
+            id: 4,
             names: ["Target two", "Target three"],
             nextCursor: "cursor-3")
         let fake = try makeFakeCodex(#"""
         #!/bin/sh
-        counter="${0}.counter"
-        count=0
-        [ ! -f "$counter" ] || count=$(cat "$counter")
-        count=$((count + 1))
-        printf '%s\n' "$count" > "$counter"
         IFS= read -r initialize || exit 2
         printf '%s\n' '{"id":1,"result":{}}'
         IFS= read -r initialized || exit 3
-        IFS= read -r list || exit 4
-        printf '%s\n' "$list" >> "${0}.requests"
-        case "$count" in
+        count=0
+        while IFS= read -r list; do
+          count=$((count + 1))
+          printf '%s\n' "$list" >> "${0}.requests"
+          case "$count" in
           1) printf '%s\n' '\#(first)' ;;
           2) printf '%s\n' '\#(second)' ;;
-          3) printf '%s\n' '\#(third)' ;;
+          3) printf '%s\n' '\#(third)'; exit 0 ;;
           *) exit 9 ;;
-        esac
+          esac
+        done
         """#)
         defer { try? FileManager.default.removeItem(at: fake.directory) }
 
@@ -881,27 +982,28 @@ struct MacNodeCodexThreadCatalogTests {
     @Test func `title search scans at most four pages and returns the continuation cursor`() async throws {
         let names = (0..<40).map { "Other \($0)" }
         let responses = try (1...4).map { page in
-            try self.listResponseJSON(names: names, nextCursor: "cursor-\(page)")
+            try self.listResponseJSON(
+                id: page + 1,
+                names: names,
+                nextCursor: "cursor-\(page)")
         }
         let fake = try makeFakeCodex(#"""
         #!/bin/sh
-        counter="${0}.counter"
-        count=0
-        [ ! -f "$counter" ] || count=$(cat "$counter")
-        count=$((count + 1))
-        printf '%s\n' "$count" > "$counter"
         IFS= read -r initialize || exit 2
         printf '%s\n' '{"id":1,"result":{}}'
         IFS= read -r initialized || exit 3
-        IFS= read -r list || exit 4
-        printf '%s\n' "$list" >> "${0}.requests"
-        case "$count" in
+        count=0
+        while IFS= read -r list; do
+          count=$((count + 1))
+          printf '%s\n' "$list" >> "${0}.requests"
+          case "$count" in
           1) printf '%s\n' '\#(responses[0])' ;;
           2) printf '%s\n' '\#(responses[1])' ;;
           3) printf '%s\n' '\#(responses[2])' ;;
-          4) printf '%s\n' '\#(responses[3])' ;;
+          4) printf '%s\n' '\#(responses[3])'; exit 0 ;;
           *) exit 9 ;;
-        esac
+          esac
+        done
         """#)
         defer { try? FileManager.default.removeItem(at: fake.directory) }
 
@@ -925,25 +1027,23 @@ struct MacNodeCodexThreadCatalogTests {
     }
 
     @Test func `title search stops a native cursor cycle`() async throws {
-        let first = try listResponseJSON(names: ["Other one"], nextCursor: "same")
-        let second = try listResponseJSON(names: ["Other two"], nextCursor: "same")
+        let first = try listResponseJSON(id: 2, names: ["Other one"], nextCursor: "same")
+        let second = try listResponseJSON(id: 3, names: ["Other two"], nextCursor: "same")
         let fake = try makeFakeCodex(#"""
         #!/bin/sh
-        counter="${0}.counter"
-        count=0
-        [ ! -f "$counter" ] || count=$(cat "$counter")
-        count=$((count + 1))
-        printf '%s\n' "$count" > "$counter"
         IFS= read -r initialize || exit 2
         printf '%s\n' '{"id":1,"result":{}}'
         IFS= read -r initialized || exit 3
-        IFS= read -r list || exit 4
-        printf '%s\n' "$list" >> "${0}.requests"
-        case "$count" in
+        count=0
+        while IFS= read -r list; do
+          count=$((count + 1))
+          printf '%s\n' "$list" >> "${0}.requests"
+          case "$count" in
           1) printf '%s\n' '\#(first)' ;;
-          2) printf '%s\n' '\#(second)' ;;
+          2) printf '%s\n' '\#(second)'; exit 0 ;;
           *) exit 9 ;;
-        esac
+          esac
+        done
         """#)
         defer { try? FileManager.default.removeItem(at: fake.directory) }
 
@@ -957,6 +1057,299 @@ struct MacNodeCodexThreadCatalogTests {
             .split(whereSeparator: \.isNewline)
         #expect(requests.count == 2)
     }
+}
+
+extension MacNodeCodexThreadCatalogTests {
+    @Test func `restarts the lifecycle client when the resolved invocation changes`() async throws {
+        let first = try makeFakeCodex(#"""
+        #!/bin/sh
+        count=0
+        [ ! -f "${0}.processes" ] || count=$(cat "${0}.processes")
+        printf '%s\n' "$((count + 1))" > "${0}.processes"
+        trap 'touch "${0}.terminated"; exit 0' TERM
+        IFS= read -r initialize || exit 2
+        id=$(printf '%s\n' "$initialize" | /usr/bin/sed -E 's/.*"id":([0-9]+).*/\1/')
+        printf '{"id":%s,"result":{}}\n' "$id"
+        IFS= read -r initialized || exit 3
+        while IFS= read -r request; do
+          id=$(printf '%s\n' "$request" | /usr/bin/sed -E 's/.*"id":([0-9]+).*/\1/')
+          printf '{"id":%s,"result":{"data":[]}}\n' "$id"
+        done
+        """#)
+        let second = try makeFakeCodex(#"""
+        #!/bin/sh
+        count=0
+        [ ! -f "${0}.processes" ] || count=$(cat "${0}.processes")
+        printf '%s\n' "$((count + 1))" > "${0}.processes"
+        IFS= read -r initialize || exit 2
+        id=$(printf '%s\n' "$initialize" | /usr/bin/sed -E 's/.*"id":([0-9]+).*/\1/')
+        printf '{"id":%s,"result":{}}\n' "$id"
+        IFS= read -r initialized || exit 3
+        while IFS= read -r request; do
+          id=$(printf '%s\n' "$request" | /usr/bin/sed -E 's/.*"id":([0-9]+).*/\1/')
+          printf '{"id":%s,"result":{"data":[]}}\n' "$id"
+        done
+        """#)
+        defer {
+            try? FileManager.default.removeItem(at: first.directory)
+            try? FileManager.default.removeItem(at: second.directory)
+        }
+        let client = CodexAppServerThreadClient(idleTimeoutSeconds: 10)
+
+        _ = try await self.requestEmptyList(client: client, executable: first.executable)
+        _ = try await self.requestEmptyList(client: client, executable: second.executable)
+
+        #expect(try self.readTrimmed(
+            URL(fileURLWithPath: first.executable.path + ".processes")) == "1")
+        #expect(try self.readTrimmed(
+            URL(fileURLWithPath: second.executable.path + ".processes")) == "1")
+        #expect(await self.waitForFile(
+            URL(fileURLWithPath: first.executable.path + ".terminated")))
+        await client.shutdown()
+    }
+
+    @Test func `restarts the lifecycle client after the child exits`() async throws {
+        let fake = try makeFakeCodex(#"""
+        #!/bin/sh
+        count=0
+        [ ! -f "${0}.processes" ] || count=$(cat "${0}.processes")
+        printf '%s\n' "$((count + 1))" > "${0}.processes"
+        IFS= read -r initialize || exit 2
+        id=$(printf '%s\n' "$initialize" | /usr/bin/sed -E 's/.*"id":([0-9]+).*/\1/')
+        printf '{"id":%s,"result":{}}\n' "$id"
+        IFS= read -r initialized || exit 3
+        IFS= read -r request || exit 4
+        id=$(printf '%s\n' "$request" | /usr/bin/sed -E 's/.*"id":([0-9]+).*/\1/')
+        printf '{"id":%s,"result":{"data":[]}}\n' "$id"
+        """#)
+        defer { try? FileManager.default.removeItem(at: fake.directory) }
+        let client = CodexAppServerThreadClient(idleTimeoutSeconds: 10)
+
+        _ = try await self.requestEmptyList(client: client, executable: fake.executable)
+        try await Task.sleep(for: .milliseconds(50))
+        _ = try await self.requestEmptyList(client: client, executable: fake.executable)
+
+        #expect(try self.readTrimmed(
+            URL(fileURLWithPath: fake.executable.path + ".processes")) == "2")
+        await client.shutdown()
+    }
+
+    @Test func `shuts down an idle lifecycle client`() async throws {
+        let fake = try makeFakeCodex(#"""
+        #!/bin/sh
+        trap 'touch "${0}.terminated"; exit 0' TERM
+        IFS= read -r initialize || exit 2
+        id=$(printf '%s\n' "$initialize" | /usr/bin/sed -E 's/.*"id":([0-9]+).*/\1/')
+        printf '{"id":%s,"result":{}}\n' "$id"
+        IFS= read -r initialized || exit 3
+        while IFS= read -r request; do
+          id=$(printf '%s\n' "$request" | /usr/bin/sed -E 's/.*"id":([0-9]+).*/\1/')
+          printf '{"id":%s,"result":{"data":[]}}\n' "$id"
+        done
+        """#)
+        defer { try? FileManager.default.removeItem(at: fake.directory) }
+        let client = CodexAppServerThreadClient(idleTimeoutSeconds: 0.05)
+
+        _ = try await self.requestEmptyList(client: client, executable: fake.executable)
+
+        #expect(await self.waitForFile(
+            URL(fileURLWithPath: fake.executable.path + ".terminated")))
+        await client.shutdown()
+    }
+
+    @Test func `client deinit terminates its owned child`() async throws {
+        let fake = try makeFakeCodex(#"""
+        #!/bin/sh
+        trap 'touch "${0}.terminated"; exit 0' TERM
+        IFS= read -r initialize || exit 2
+        id=$(printf '%s\n' "$initialize" | /usr/bin/sed -E 's/.*"id":([0-9]+).*/\1/')
+        printf '{"id":%s,"result":{}}\n' "$id"
+        IFS= read -r initialized || exit 3
+        while IFS= read -r request; do
+          id=$(printf '%s\n' "$request" | /usr/bin/sed -E 's/.*"id":([0-9]+).*/\1/')
+          printf '{"id":%s,"result":{"data":[]}}\n' "$id"
+        done
+        """#)
+        defer { try? FileManager.default.removeItem(at: fake.directory) }
+        var client: CodexAppServerThreadClient? = CodexAppServerThreadClient(
+            idleTimeoutSeconds: 10)
+
+        _ = try await self.requestEmptyList(
+            client: #require(client),
+            executable: fake.executable)
+        client = nil
+
+        #expect(await self.waitForFile(
+            URL(fileURLWithPath: fake.executable.path + ".terminated")))
+    }
+
+    @Test func `oversized idle output resets the lifecycle client`() async throws {
+        let fake = try makeFakeCodex(#"""
+        #!/bin/sh
+        count=0
+        [ ! -f "${0}.processes" ] || count=$(cat "${0}.processes")
+        count=$((count + 1))
+        printf '%s\n' "$count" > "${0}.processes"
+        IFS= read -r initialize || exit 2
+        id=$(printf '%s\n' "$initialize" | /usr/bin/sed -E 's/.*"id":([0-9]+).*/\1/')
+        printf '{"id":%s,"result":{}}\n' "$id"
+        IFS= read -r initialized || exit 3
+        while IFS= read -r request; do
+          id=$(printf '%s\n' "$request" | /usr/bin/sed -E 's/.*"id":([0-9]+).*/\1/')
+          printf '{"id":%s,"result":{"data":[]}}\n' "$id"
+          if [ "$count" = 1 ]; then
+            sleep 0.1
+            printf '%512s\n' x
+          fi
+        done
+        """#)
+        defer { try? FileManager.default.removeItem(at: fake.directory) }
+        let client = CodexAppServerThreadClient(
+            idleTimeoutSeconds: 10,
+            idleReadLimit: 128)
+
+        _ = try await self.requestEmptyList(
+            client: client,
+            executable: fake.executable,
+            maxLineBytes: 128)
+        try await Task.sleep(for: .milliseconds(300))
+        _ = try await self.requestEmptyList(
+            client: client,
+            executable: fake.executable,
+            maxLineBytes: 128)
+
+        #expect(try self.readTrimmed(
+            URL(fileURLWithPath: fake.executable.path + ".processes")) == "2")
+        await client.shutdown()
+    }
+
+    @Test func `timeout restarts the client without dropping the next request`() async throws {
+        let fake = try makeFakeCodex(#"""
+        #!/bin/sh
+        count=0
+        [ ! -f "${0}.processes" ] || count=$(cat "${0}.processes")
+        count=$((count + 1))
+        printf '%s\n' "$count" > "${0}.processes"
+        IFS= read -r initialize || exit 2
+        id=$(printf '%s\n' "$initialize" | /usr/bin/sed -E 's/.*"id":([0-9]+).*/\1/')
+        printf '{"id":%s,"result":{}}\n' "$id"
+        IFS= read -r initialized || exit 3
+        IFS= read -r request || exit 4
+        if [ "$count" = 1 ]; then
+          touch "${0}.request-started"
+          sleep 5
+          exit 0
+        fi
+        id=$(printf '%s\n' "$request" | /usr/bin/sed -E 's/.*"id":([0-9]+).*/\1/')
+        printf '{"id":%s,"result":{"data":[]}}\n' "$id"
+        """#)
+        defer { try? FileManager.default.removeItem(at: fake.directory) }
+        let client = CodexAppServerThreadClient(idleTimeoutSeconds: 10)
+
+        let first = Task {
+            try await self.requestEmptyList(
+                client: client,
+                executable: fake.executable,
+                timeoutSeconds: 0.5)
+        }
+        #expect(await self.waitForFile(
+            URL(fileURLWithPath: fake.executable.path + ".request-started")))
+        let second = Task {
+            try await self.requestEmptyList(client: client, executable: fake.executable)
+        }
+
+        await #expect(throws: MacNodeCodexThreadCatalog.CatalogError.timedOut) {
+            try await first.value
+        }
+        let result = try await second.value
+        #expect(try (JSONSerialization.jsonObject(with: result) as? [String: Any])?["data"] != nil)
+        #expect(try self.readTrimmed(
+            URL(fileURLWithPath: fake.executable.path + ".processes")) == "2")
+        await client.shutdown()
+    }
+
+    @Test func `queued request consumes its wall-clock deadline`() async throws {
+        let fake = try makeFakeCodex(#"""
+        #!/bin/sh
+        IFS= read -r initialize || exit 2
+        id=$(printf '%s\n' "$initialize" | /usr/bin/sed -E 's/.*"id":([0-9]+).*/\1/')
+        printf '{"id":%s,"result":{}}\n' "$id"
+        IFS= read -r initialized || exit 3
+        IFS= read -r request || exit 4
+        touch "${0}.request-started"
+        sleep 5
+        """#)
+        defer { try? FileManager.default.removeItem(at: fake.directory) }
+        let client = CodexAppServerThreadClient(idleTimeoutSeconds: 10)
+
+        let first = Task {
+            try await self.requestEmptyList(
+                client: client,
+                executable: fake.executable,
+                timeoutSeconds: 0.5)
+        }
+        #expect(await self.waitForFile(
+            URL(fileURLWithPath: fake.executable.path + ".request-started")))
+        let second = Task {
+            try await self.requestEmptyList(
+                client: client,
+                executable: fake.executable,
+                timeoutSeconds: 0.05)
+        }
+
+        await #expect(throws: MacNodeCodexThreadCatalog.CatalogError.timedOut) {
+            try await second.value
+        }
+        await #expect(throws: MacNodeCodexThreadCatalog.CatalogError.timedOut) {
+            try await first.value
+        }
+        await client.shutdown()
+    }
+
+    @Test func `cancellation restarts the client without dropping the next request`() async throws {
+        let fake = try makeFakeCodex(#"""
+        #!/bin/sh
+        count=0
+        [ ! -f "${0}.processes" ] || count=$(cat "${0}.processes")
+        count=$((count + 1))
+        printf '%s\n' "$count" > "${0}.processes"
+        IFS= read -r initialize || exit 2
+        id=$(printf '%s\n' "$initialize" | /usr/bin/sed -E 's/.*"id":([0-9]+).*/\1/')
+        printf '{"id":%s,"result":{}}\n' "$id"
+        IFS= read -r initialized || exit 3
+        IFS= read -r request || exit 4
+        if [ "$count" = 1 ]; then
+          touch "${0}.request-started"
+          sleep 5
+          exit 0
+        fi
+        id=$(printf '%s\n' "$request" | /usr/bin/sed -E 's/.*"id":([0-9]+).*/\1/')
+        printf '{"id":%s,"result":{"data":[]}}\n' "$id"
+        """#)
+        defer { try? FileManager.default.removeItem(at: fake.directory) }
+        let client = CodexAppServerThreadClient(idleTimeoutSeconds: 10)
+        let first = Task {
+            try await self.requestEmptyList(
+                client: client,
+                executable: fake.executable,
+                timeoutSeconds: 5)
+        }
+        #expect(await self.waitForFile(
+            URL(fileURLWithPath: fake.executable.path + ".request-started")))
+        let second = Task {
+            try await self.requestEmptyList(client: client, executable: fake.executable)
+        }
+
+        first.cancel()
+        await #expect(throws: CancellationError.self) {
+            try await first.value
+        }
+        let result = try await second.value
+        #expect(try (JSONSerialization.jsonObject(with: result) as? [String: Any])?["data"] != nil)
+        #expect(try self.readTrimmed(
+            URL(fileURLWithPath: fake.executable.path + ".processes")) == "2")
+        await client.shutdown()
+    }
 
     @Test func `drains App Server frames larger than one pipe read while server stays open`() async throws {
         let threads: [[String: Any]] = (0..<50).map { index in
@@ -995,6 +1388,106 @@ struct MacNodeCodexThreadCatalogTests {
         #expect((decoded["sessions"] as? [Any])?.count == 50)
     }
 
+    @Test func `drains a large final frame before handling process exit`() async throws {
+        let threads: [[String: Any]] = (0..<100).map { index in
+            [
+                "id": "thread-\(index)",
+                "name": "Final catalog \(index)",
+                "cwd": "/workspace/\(String(repeating: "x", count: 3500))",
+                "status": ["type": "notLoaded"],
+            ]
+        }
+        let responseData = try JSONSerialization.data(withJSONObject: [
+            "id": 2,
+            "result": ["data": threads],
+        ])
+        let response = try #require(String(data: responseData, encoding: .utf8))
+        #expect(response.utf8.count > 256 * 1024)
+        let fake = try makeFakeCodex("""
+        #!/bin/sh
+        IFS= read -r initialize || exit 2
+        printf '%s\n' '{"id":1,"result":{}}'
+        IFS= read -r initialized || exit 3
+        IFS= read -r list || exit 4
+        printf '%s\n' '\(response)'
+        """)
+        defer { try? FileManager.default.removeItem(at: fake.directory) }
+
+        let payload = try await MacNodeCodexThreadCatalog.list(
+            paramsJSON: #"{"limit":100}"#,
+            executable: fake.executable.path,
+            timeoutSeconds: 10)
+        let decoded = try #require(
+            JSONSerialization.jsonObject(with: Data(payload.utf8)) as? [String: Any])
+        #expect((decoded["sessions"] as? [Any])?.count == 100)
+    }
+
+    @Test func `accepts coalesced JSONL frames within the per-frame limit`() async throws {
+        let fake = try makeFakeCodex(#"""
+        #!/bin/sh
+        IFS= read -r initialize || exit 2
+        printf '%s\n' '{"id":1,"result":{}}'
+        IFS= read -r initialized || exit 3
+        IFS= read -r list || exit 4
+        printf '%s\n%s\n' \
+          '{"method":"thread/started","params":{"padding":"xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"}}' \
+          '{"id":2,"result":{"data":[]}}'
+        sleep 1
+        """#)
+        defer { try? FileManager.default.removeItem(at: fake.directory) }
+        let client = CodexAppServerThreadClient(idleTimeoutSeconds: 10)
+
+        let result = try await self.requestEmptyList(
+            client: client,
+            executable: fake.executable,
+            maxLineBytes: 128)
+
+        #expect(try (JSONSerialization.jsonObject(with: result) as? [String: Any])?["data"] != nil)
+        await client.shutdown()
+    }
+
+    @Test func `uses the active request frame limit after advancing the queue`() async throws {
+        let fake = try makeFakeCodex(#"""
+        #!/bin/sh
+        IFS= read -r initialize || exit 2
+        printf '%s\n' '{"id":1,"result":{}}'
+        IFS= read -r initialized || exit 3
+        IFS= read -r first || exit 4
+        first_id=$(printf '%s\n' "$first" | /usr/bin/sed -E 's/.*"id":([0-9]+).*/\1/')
+        touch "${0}.first-started"
+        sleep 0.1
+        printf '{"id":%s,"result":{"data":[]}}\n' "$first_id"
+        IFS= read -r second || exit 5
+        second_id=$(printf '%s\n' "$second" | /usr/bin/sed -E 's/.*"id":([0-9]+).*/\1/')
+        padding=xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
+        printf '{"id":%s,"result":{"data":[],"padding":"%s"}}\n' "$second_id" "$padding"
+        sleep 1
+        """#)
+        defer { try? FileManager.default.removeItem(at: fake.directory) }
+        let client = CodexAppServerThreadClient(idleTimeoutSeconds: 10)
+
+        let first = Task {
+            try await self.requestEmptyList(
+                client: client,
+                executable: fake.executable,
+                maxLineBytes: 1024)
+        }
+        #expect(await self.waitForFile(
+            URL(fileURLWithPath: fake.executable.path + ".first-started")))
+        let second = Task {
+            try await self.requestEmptyList(
+                client: client,
+                executable: fake.executable,
+                maxLineBytes: 64)
+        }
+
+        _ = try await first.value
+        await #expect(throws: MacNodeCodexThreadCatalog.CatalogError.responseTooLarge) {
+            try await second.value
+        }
+        await client.shutdown()
+    }
+
     @Test func `default deadline allows cold large catalog scans`() {
         #expect(MacNodeCodexThreadCatalog.defaultTimeoutSeconds == 60)
     }