diff --git a/apps/macos/Sources/OpenClaw/NodeMode/MacNodeCodexThreadCatalog.swift b/apps/macos/Sources/OpenClaw/NodeMode/MacNodeCodexThreadCatalog.swift index 6837d5dca34a..f367144d87ea 100644 --- a/apps/macos/Sources/OpenClaw/NodeMode/MacNodeCodexThreadCatalog.swift +++ b/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( + _ 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? - 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) { - 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) { - 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) - } - } -} diff --git a/apps/macos/Sources/OpenClaw/NodeMode/MacNodeCodexThreadCatalogClient.swift b/apps/macos/Sources/OpenClaw/NodeMode/MacNodeCodexThreadCatalogClient.swift new file mode 100644 index 000000000000..a881b714ec96 --- /dev/null +++ b/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? + var timer: DispatchSourceTimer? + + init( + token: UUID, + invocation: MacNodeCodexThreadCatalog.ResolvedInvocation, + method: String, + requestParamsData: Data, + maxLineBytes: Int, + continuation: CheckedContinuation) + { + 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, + 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) { + 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) + } + } +} diff --git a/apps/macos/Sources/OpenClaw/NodeMode/MacNodeRuntime.swift b/apps/macos/Sources/OpenClaw/NodeMode/MacNodeRuntime.swift index 18ab62e7378a..5a4806896699 100644 --- a/apps/macos/Sources/OpenClaw/NodeMode/MacNodeRuntime.swift +++ b/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) } diff --git a/apps/macos/Tests/OpenClawIPCTests/MacNodeCodexThreadCatalogTests.swift b/apps/macos/Tests/OpenClawIPCTests/MacNodeCodexThreadCatalogTests.swift index f6fa5f8b11bd..0f568ccd0123 100644 --- a/apps/macos/Tests/OpenClawIPCTests/MacNodeCodexThreadCatalogTests.swift +++ b/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) }