From 71a7283792e4fd57e108acd4e356d7a9c6474db3 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Wed, 26 Aug 2026 13:25:46 -0700 Subject: [PATCH] fix(macos): own process startup and tunnel retirement (#130332) Replace the cancellable one-shot startup stream with a replayable result. Join process cleanup and durable SSH receipt retirement before admitting a replacement tunnel, and keep stale callers fenced across actor suspension. --- .../Sources/OpenClaw/ManagedProcess.swift | 57 ++- .../Sources/OpenClaw/RemotePortTunnel.swift | 11 +- .../OpenClaw/RemoteTunnelManager.swift | 386 +++++++----------- .../OpenClaw/TalkMLXSpeechSynthesizer.swift | 2 + .../ManagedProcessTests.swift | 56 ++- 5 files changed, 255 insertions(+), 257 deletions(-) diff --git a/apps/macos/Sources/OpenClaw/ManagedProcess.swift b/apps/macos/Sources/OpenClaw/ManagedProcess.swift index 899268025c71..befa4726563e 100644 --- a/apps/macos/Sources/OpenClaw/ManagedProcess.swift +++ b/apps/macos/Sources/OpenClaw/ManagedProcess.swift @@ -18,6 +18,8 @@ final class ManagedProcess: @unchecked Sendable { private final class State: @unchecked Sendable { private let lock = NSLock() private var childHandles: [FileHandle] + private var startResult: Result? + private var startWaiters: [UUID: CheckedContinuation] = [:] private var abortiveTerminationRequested = false private var finished = false @@ -37,6 +39,38 @@ final class ManagedProcess: @unchecked Sendable { self.lock.withLock { self.abortiveTerminationRequested = true } } + func publishStart(_ result: Result) { + let waiters = self.lock.withLock { + guard self.startResult == nil else { return [CheckedContinuation]() } + self.startResult = result + defer { self.startWaiters.removeAll() } + return Array(self.startWaiters.values) + } + for waiter in waiters { + waiter.resume(with: result.mapError { $0 as any Error }) + } + } + + func waitUntilStarted() async throws -> pid_t { + let id = UUID() + let pid = try await withTaskCancellationHandler { + try await withCheckedThrowingContinuation { continuation in + let result: Result? = self.lock.withLock { + if Task.isCancelled { return .failure(CancellationError()) } + if let result = self.startResult { return result.mapError { $0 as any Error } } + self.startWaiters[id] = continuation + return nil + } + if let result { continuation.resume(with: result) } + } + } onCancel: { + let waiter = self.lock.withLock { self.startWaiters.removeValue(forKey: id) } + waiter?.resume(throwing: CancellationError()) + } + try Task.checkCancellation() + return pid + } + func closeChildHandles() { let handles = self.lock.withLock { defer { self.childHandles.removeAll() } @@ -57,7 +91,6 @@ final class ManagedProcess: @unchecked Sendable { } private let state: State - private let startEvents: AsyncStream> private let wakeContinuation: AsyncStream.Continuation let completionTask: Task @@ -67,12 +100,10 @@ final class ManagedProcess: @unchecked Sendable { private init( state: State, - startEvents: AsyncStream>, wakeContinuation: AsyncStream.Continuation, completionTask: Task) { self.state = state - self.startEvents = startEvents self.wakeContinuation = wakeContinuation self.completionTask = completionTask } @@ -92,7 +123,6 @@ final class ManagedProcess: @unchecked Sendable { .send(signal: .terminate, toProcessGroup: true, allowedDurationToNextStep: .milliseconds(250)), ] let state = State(childHandles: childHandles) - let (startEvents, startContinuation) = AsyncStream.makeStream(of: Result.self) let (wakeEvents, wakeContinuation) = AsyncStream.makeStream( of: WakeReason.self, bufferingPolicy: .bufferingNewest(1)) @@ -106,15 +136,22 @@ final class ManagedProcess: @unchecked Sendable { { execution in let pid = pid_t(execution.processIdentifier.value) state.closeChildHandles() - startContinuation.yield(.success(pid)) + // Startup is a replayable lifecycle fact; cancelling one waiter + // must not consume it or cancel the independently owned child. + state.publishStart(.success(pid)) let exitSource = DispatchSource.makeProcessSource( identifier: pid, eventMask: .exit, queue: .global(qos: .userInitiated)) - exitSource.setEventHandler { wakeContinuation.yield(.exited) } + let didExit: @Sendable () -> Void = { + // Stop reuse at leader exit; completion still joins descendant cleanup. + state.finish() + wakeContinuation.yield(.exited) + } + exitSource.setEventHandler(handler: didExit) exitSource.resume() - if State.hasExited(pid) { wakeContinuation.yield(.exited) } + if State.hasExited(pid) { didExit() } var wakeIterator = wakeEvents.makeAsyncIterator() let wakeReason = await wakeIterator.next() ?? .terminate(gracefully: true) exitSource.cancel() @@ -152,14 +189,13 @@ final class ManagedProcess: @unchecked Sendable { } catch { state.closeChildHandles() let message = (error as? SubprocessError)?.description ?? error.localizedDescription - startContinuation.yield(.failure(StartFailure(message: message))) + state.publishStart(.failure(StartFailure(message: message))) state.finish() return nil } } return ManagedProcess( state: state, - startEvents: startEvents, wakeContinuation: wakeContinuation, completionTask: task) } @@ -189,8 +225,7 @@ final class ManagedProcess: @unchecked Sendable { } func waitUntilStarted() async throws -> pid_t { - var iterator = self.startEvents.makeAsyncIterator() - return try await iterator.next()!.get() + try await self.state.waitUntilStarted() } func requestTermination(gracefully: Bool = true) { diff --git a/apps/macos/Sources/OpenClaw/RemotePortTunnel.swift b/apps/macos/Sources/OpenClaw/RemotePortTunnel.swift index 8b0f3059764f..4f2035bf7773 100644 --- a/apps/macos/Sources/OpenClaw/RemotePortTunnel.swift +++ b/apps/macos/Sources/OpenClaw/RemotePortTunnel.swift @@ -74,10 +74,6 @@ final class RemotePortTunnel: @unchecked Sendable { deinit { Self.cleanupStderr(self.stderrHandle) let receipt = self.guardianReceipt - guard self.process.isRunning else { - Task { await PortGuardian.shared.removeRecord(receipt) } - return - } // deinit cannot wait. Leave the receipt durable until a later sweep proves // the child exited; deleting it after TERM alone can orphan a resistant SSH. Task { await PortGuardian.shared.relinquishRecord(receipt) } @@ -87,8 +83,8 @@ final class RemotePortTunnel: @unchecked Sendable { func terminate() async { await self.process.terminate() Self.cleanupStderr(self.stderrHandle) - let receipt = self.guardianReceipt - Task { await PortGuardian.shared.removeRecord(receipt) } + // Finish retiring this receipt before a replacement spawn reserves the ledger. + await PortGuardian.shared.removeRecord(self.guardianReceipt) } static func configuration(remotePort: Int) throws -> Configuration { @@ -195,6 +191,9 @@ final class RemotePortTunnel: @unchecked Sendable { do { processIdentifier = try await process.waitUntilStarted() } catch { + // Cancellation abandons the waiter, not the detached spawn. Reap the + // child before releasing its reservation or closing inherited handles. + await process.terminate(gracefully: false) await PortGuardian.shared.cancelTunnelSpawn(spawnPreparation) Self.cleanupStderr(stderrHandle) throw error diff --git a/apps/macos/Sources/OpenClaw/RemoteTunnelManager.swift b/apps/macos/Sources/OpenClaw/RemoteTunnelManager.swift index 300798088efa..d23478bf2e75 100644 --- a/apps/macos/Sources/OpenClaw/RemoteTunnelManager.swift +++ b/apps/macos/Sources/OpenClaw/RemoteTunnelManager.swift @@ -10,25 +10,13 @@ actor RemoteTunnelManager { let generation: UInt64 } - private enum CreateJoinResult { + private enum RouteLookupResult { case none - case replacedMismatchedCreate(UInt64) + case retired(UInt64) case staleConfiguration case route(Route) } - private enum ActiveRouteLookupResult { - case none - case retiredActive(UInt64) - case staleConfiguration - case route(Route) - } - - private struct EnsureStepResolution { - let route: Route? - let lifecycleGeneration: UInt64 - } - private struct ActiveTunnel { let tunnel: RemotePortTunnel let configuration: RemotePortTunnel.Configuration @@ -42,31 +30,28 @@ actor RemoteTunnelManager { configuration: RemotePortTunnel.Configuration, lifecycleGeneration: UInt64, task: Task)? + private var retirementInFlight: (token: UUID, task: Task)? private var tunnelGeneration: UInt64 = 0 private var lifecycleGeneration: UInt64 = 0 - private var restartInFlight = false private var lastRestartAt: Date? private let restartBackoffSeconds: TimeInterval = 2.0 func controlTunnelRouteIfRunning() async -> Route? { + guard self.retirementInFlight == nil else { return nil } guard let configuration = try? RemotePortTunnel.configuration( remotePort: GatewayEnvironment.gatewayPort()) else { - self.lifecycleGeneration &+= 1 - self.createInFlight?.task.cancel() - self.createInFlight = nil - let tunnel = self.controlTunnel?.tunnel - self.controlTunnel = nil - if tunnel != nil { - self.tunnelGeneration &+= 1 - } - await tunnel?.terminate() + self.beginRetirement() + await self.waitForRetirement() return nil } - switch await self.lookupControlTunnelRoute(configuration: configuration) { + switch await self.lookupControlTunnelRoute( + configuration: configuration, + lifecycleGeneration: self.lifecycleGeneration) + { case let .route(route): return route - case .none, .retiredActive, .staleConfiguration: + case .none, .retired, .staleConfiguration: return nil } } @@ -77,40 +62,29 @@ actor RemoteTunnelManager { private func lookupControlTunnelRoute( configuration: RemotePortTunnel.Configuration, - requireCurrentConfiguration: Bool = false) async -> ActiveRouteLookupResult + lifecycleGeneration: UInt64) async -> RouteLookupResult { - if requireCurrentConfiguration { - guard let currentConfiguration = try? RemotePortTunnel.configuration( - remotePort: GatewayEnvironment.gatewayPort()), - Self.isCurrentConfiguration( - requested: configuration, - current: currentConfiguration) - else { - return .staleConfiguration - } - } - if self.restartInFlight { - self.logger.info("control tunnel restart in flight; skipping reuse check") - return .none + await self.waitForRetirement() + guard self.lifecycleGeneration == lifecycleGeneration else { return .none } + guard let currentConfiguration = try? RemotePortTunnel.configuration( + remotePort: GatewayEnvironment.gatewayPort()), + Self.isCurrentConfiguration(requested: configuration, current: currentConfiguration) + else { + return .staleConfiguration } if let active = controlTunnel { guard Self.canReuse(active.configuration, for: configuration) else { self.logger.info("configured SSH route changed; replacing control tunnel") - self.lifecycleGeneration &+= 1 - let replacementGeneration = self.lifecycleGeneration - self.controlTunnel = nil - self.tunnelGeneration &+= 1 - await active.tunnel.terminate() - return .retiredActive(replacementGeneration) + let replacementGeneration = self.beginRetirement() + await self.waitForRetirement() + return .retired(replacementGeneration) } guard active.tunnel.isRunning, let local = active.tunnel.localPort else { - self.lifecycleGeneration &+= 1 - let replacementGeneration = self.lifecycleGeneration - self.controlTunnel = nil - self.tunnelGeneration &+= 1 - return .retiredActive(replacementGeneration) + let replacementGeneration = self.beginRetirement() + await self.waitForRetirement() + return .retired(replacementGeneration) } let pid = active.tunnel.processIdentifier let isListening = await PortGuardian.shared.isListening(port: Int(local), pid: pid) @@ -121,19 +95,19 @@ actor RemoteTunnelManager { current.configuration == active.configuration, current.route == active.route else { return .none } + if (try? RemotePortTunnel.configuration(remotePort: GatewayEnvironment.gatewayPort())) != configuration { + return .staleConfiguration + } if isListening { self.logger.info("reusing active SSH tunnel localPort=\(local, privacy: .public)") return .route(current.route) } self.logger.error( "active SSH tunnel on port \(local, privacy: .public) is not listening; restarting") - self.lifecycleGeneration &+= 1 - let replacementGeneration = self.lifecycleGeneration - self.controlTunnel = nil - self.tunnelGeneration &+= 1 - self.beginRestart() - await active.tunnel.terminate() - return .retiredActive(replacementGeneration) + let replacementGeneration = self.beginRetirement() + self.lastRestartAt = Date() + await self.waitForRetirement() + return .retired(replacementGeneration) } return .none } @@ -152,59 +126,32 @@ actor RemoteTunnelManager { requested == current } - private func resolveActiveLookup( - _ result: ActiveRouteLookupResult, - lifecycleGeneration: UInt64) async throws -> EnsureStepResolution + private func resolveLookup( + _ result: RouteLookupResult, + lifecycleGeneration: UInt64) async throws -> Route? { + try Task.checkCancellation() switch result { case let .route(route): - return EnsureStepResolution(route: route, lifecycleGeneration: lifecycleGeneration) - case let .retiredActive(replacementGeneration): + guard self.lifecycleGeneration == lifecycleGeneration else { throw CancellationError() } + return route + case let .retired(replacementGeneration): guard self.lifecycleGeneration == replacementGeneration else { throw CancellationError() } - return EnsureStepResolution(route: nil, lifecycleGeneration: replacementGeneration) + // Another caller may have installed the replacement during retirement. + return try await self.ensureControlTunnelRoute(lifecycleGeneration: replacementGeneration) case .staleConfiguration: - try Task.checkCancellation() guard self.lifecycleGeneration == lifecycleGeneration else { throw CancellationError() } - let route = try await self.ensureControlTunnelRoute( + return try await self.ensureControlTunnelRoute( lifecycleGeneration: lifecycleGeneration) - return EnsureStepResolution(route: route, lifecycleGeneration: lifecycleGeneration) case .none: guard self.lifecycleGeneration == lifecycleGeneration else { throw CancellationError() } - return EnsureStepResolution(route: nil, lifecycleGeneration: lifecycleGeneration) - } - } - - private func resolveCreateJoin( - _ result: CreateJoinResult, - lifecycleGeneration: UInt64) async throws -> EnsureStepResolution - { - switch result { - case let .route(route): - return EnsureStepResolution(route: route, lifecycleGeneration: lifecycleGeneration) - case let .replacedMismatchedCreate(replacementGeneration): - guard self.lifecycleGeneration == replacementGeneration else { - throw CancellationError() - } - return EnsureStepResolution(route: nil, lifecycleGeneration: replacementGeneration) - case .staleConfiguration: - try Task.checkCancellation() - guard self.lifecycleGeneration == lifecycleGeneration else { - throw CancellationError() - } - let route = try await self.ensureControlTunnelRoute( - lifecycleGeneration: lifecycleGeneration) - return EnsureStepResolution(route: route, lifecycleGeneration: lifecycleGeneration) - case .none: - guard self.lifecycleGeneration == lifecycleGeneration else { - throw CancellationError() - } - return EnsureStepResolution(route: nil, lifecycleGeneration: lifecycleGeneration) + return nil } } @@ -220,115 +167,83 @@ actor RemoteTunnelManager { } private func ensureControlTunnelRoute( - lifecycleGeneration initialLifecycleGeneration: UInt64) async throws -> Route + lifecycleGeneration: UInt64) async throws -> Route { - var lifecycleGeneration = initialLifecycleGeneration - try Task.checkCancellation() - guard self.lifecycleGeneration == lifecycleGeneration else { - throw CancellationError() - } - - let configuration = try RemotePortTunnel.configuration( - remotePort: GatewayEnvironment.gatewayPort()) - let identitySet = !configuration.identity.isEmpty - self.logger.info( - "ensure SSH tunnel target=\(configuration.target.host, privacy: .public) " + - "identitySet=\(identitySet, privacy: .public)") - - var resolution = try await self.resolveActiveLookup( - self.lookupControlTunnelRoute( - configuration: configuration, - requireCurrentConfiguration: true), - lifecycleGeneration: lifecycleGeneration) - if let route = resolution.route { - return route - } - lifecycleGeneration = resolution.lifecycleGeneration - - var joinResult = try await self.joinCreateInFlight(configuration: configuration) - resolution = try await self.resolveCreateJoin( - joinResult, - lifecycleGeneration: lifecycleGeneration) - if let route = resolution.route { - return route - } - lifecycleGeneration = resolution.lifecycleGeneration - - try await self.waitForRestartBackoffIfNeeded() - try Task.checkCancellation() - guard self.lifecycleGeneration == lifecycleGeneration else { - throw CancellationError() - } - - // The backoff suspends this actor. Another caller may have installed or - // started the canonical tunnel while we slept, so join it instead of - // launching a duplicate SSH process. - resolution = try await self.resolveActiveLookup( - self.lookupControlTunnelRoute( - configuration: configuration, - requireCurrentConfiguration: true), - lifecycleGeneration: lifecycleGeneration) - if let route = resolution.route { - return route - } - lifecycleGeneration = resolution.lifecycleGeneration - - joinResult = try await self.joinCreateInFlight(configuration: configuration) - resolution = try await self.resolveCreateJoin( - joinResult, - lifecycleGeneration: lifecycleGeneration) - if let route = resolution.route { - return route - } - lifecycleGeneration = resolution.lifecycleGeneration - try Task.checkCancellation() - guard self.lifecycleGeneration == lifecycleGeneration else { - throw CancellationError() - } - - let currentConfiguration = try RemotePortTunnel.configuration( - remotePort: GatewayEnvironment.gatewayPort()) - guard currentConfiguration == configuration else { + var waitedForBackoff = false + while true { + try Task.checkCancellation() guard self.lifecycleGeneration == lifecycleGeneration else { throw CancellationError() } - return try await self.ensureControlTunnelRoute( + let configuration = try RemotePortTunnel.configuration( + remotePort: GatewayEnvironment.gatewayPort()) + if let route = try await self.resolveLookup( + self.lookupControlTunnelRoute( + configuration: configuration, + lifecycleGeneration: lifecycleGeneration), lifecycleGeneration: lifecycleGeneration) - } - - let desiredPort = UInt16(GatewayEnvironment.gatewayPort()) - let token = UUID() - let task = Task { - try await RemotePortTunnel.create( - configuration: configuration, - preferredLocalPort: desiredPort, - allowRandomLocalPort: true) - } - self.createInFlight = ( - token: token, - configuration: configuration, - lifecycleGeneration: lifecycleGeneration, - task: task) - let tunnel: RemotePortTunnel - do { - tunnel = try await task.value - } catch { - if self.createInFlight?.token == token { - self.createInFlight = nil + { + return route } - throw error + if let route = try await self.resolveLookup( + self.joinCreateInFlight( + configuration: configuration, + lifecycleGeneration: lifecycleGeneration), + lifecycleGeneration: lifecycleGeneration) + { + return route + } + if !waitedForBackoff { + try await self.waitForRestartBackoffIfNeeded() + waitedForBackoff = true + continue + } + + // Every suspension can admit another owner. Check all slots and the + // current configuration in the same actor turn that claims creation. + try Task.checkCancellation() + guard self.lifecycleGeneration == lifecycleGeneration else { throw CancellationError() } + let currentConfiguration = try RemotePortTunnel.configuration( + remotePort: GatewayEnvironment.gatewayPort()) + guard self.retirementInFlight == nil, self.controlTunnel == nil, + self.createInFlight == nil, currentConfiguration == configuration + else { continue } + + let desiredPort = UInt16(GatewayEnvironment.gatewayPort()) + let token = UUID() + let task = Task { + try await RemotePortTunnel.create( + configuration: configuration, + preferredLocalPort: desiredPort, + allowRandomLocalPort: true) + } + self.createInFlight = ( + token: token, + configuration: configuration, + lifecycleGeneration: lifecycleGeneration, + task: task) + let tunnel: RemotePortTunnel + do { + tunnel = try await task.value + } catch { + if self.createInFlight?.token == token { self.createInFlight = nil } + throw error + } + return try await self.installCreatedTunnel( + tunnel, + token: token, + configuration: configuration, + lifecycleGeneration: lifecycleGeneration, + fallbackPort: desiredPort) } - return try await self.installCreatedTunnel( - tunnel, - token: token, - configuration: configuration, - lifecycleGeneration: lifecycleGeneration, - fallbackPort: desiredPort) } private func joinCreateInFlight( - configuration: RemotePortTunnel.Configuration) async throws -> CreateJoinResult + configuration: RemotePortTunnel.Configuration, + lifecycleGeneration: UInt64) async throws -> RouteLookupResult { + await self.waitForRetirement() + guard self.lifecycleGeneration == lifecycleGeneration else { throw CancellationError() } guard let create = createInFlight else { return .none } guard create.configuration == configuration else { let currentConfiguration = try RemotePortTunnel.configuration( @@ -342,11 +257,9 @@ actor RemoteTunnelManager { // A suspended create owns the prior SSH route. It must not become // the loopback endpoint for the replacement Gateway. - self.lifecycleGeneration &+= 1 - let replacementGeneration = self.lifecycleGeneration - create.task.cancel() - self.createInFlight = nil - return .replacedMismatchedCreate(replacementGeneration) + let replacementGeneration = self.beginRetirement() + await self.waitForRetirement() + return .retired(replacementGeneration) } self.logger.info("control tunnel create in flight; joining") @@ -367,6 +280,37 @@ actor RemoteTunnelManager { fallbackPort: UInt16(GatewayEnvironment.gatewayPort()))) } + @discardableResult + private func beginRetirement() -> UInt64 { + self.lifecycleGeneration &+= 1 + let active = self.controlTunnel?.tunnel + let create = self.createInFlight?.task + guard active != nil || create != nil else { return self.lifecycleGeneration } + self.controlTunnel = nil + self.createInFlight = nil + self.tunnelGeneration &+= 1 + create?.cancel() + + // Publish cleanup ownership before suspending the actor. Reentrant ensures + // must join this barrier before reserving the ledger for a replacement. + let previous = self.retirementInFlight?.task + self.retirementInFlight = (UUID(), Task { + await previous?.value + if let tunnel = try? await create?.value { await tunnel.terminate() } + await active?.terminate() + }) + return self.lifecycleGeneration + } + + private func waitForRetirement() async { + while let retirement = self.retirementInFlight { + await retirement.task.value + if self.retirementInFlight?.token == retirement.token { + self.retirementInFlight = nil + } + } + } + private func installCreatedTunnel( _ tunnel: RemotePortTunnel, token: UUID, @@ -375,14 +319,14 @@ actor RemoteTunnelManager { fallbackPort: UInt16) async throws -> Route { guard self.lifecycleGeneration == lifecycleGeneration else { - await tunnel.terminate() + await self.waitForRetirement() throw CancellationError() } if let active = controlTunnel, active.tunnel === tunnel { return active.route } guard self.createInFlight?.token == token else { - await tunnel.terminate() + await self.waitForRetirement() throw CancellationError() } let currentConfiguration: RemotePortTunnel.Configuration @@ -390,16 +334,13 @@ actor RemoteTunnelManager { currentConfiguration = try RemotePortTunnel.configuration( remotePort: GatewayEnvironment.gatewayPort()) } catch { - self.lifecycleGeneration &+= 1 - self.createInFlight = nil - await tunnel.terminate() + self.beginRetirement() + await self.waitForRetirement() throw error } guard currentConfiguration == configuration else { - self.lifecycleGeneration &+= 1 - let replacementGeneration = self.lifecycleGeneration - self.createInFlight = nil - await tunnel.terminate() + let replacementGeneration = self.beginRetirement() + await self.waitForRetirement() try Task.checkCancellation() guard self.lifecycleGeneration == replacementGeneration else { throw CancellationError() @@ -415,7 +356,6 @@ actor RemoteTunnelManager { tunnel: tunnel, configuration: configuration, route: route) - self.endRestart() self.logger.info( "ssh tunnel ready localPort=\(resolvedPort, privacy: .public) " + "generation=\(route.generation, privacy: .public)") @@ -425,13 +365,8 @@ actor RemoteTunnelManager { func stopAll() async { // Invalidate every captured route before terminating processes. Delayed // health checks and create completions cannot resurrect this epoch. - self.lifecycleGeneration &+= 1 - self.tunnelGeneration &+= 1 - self.createInFlight?.task.cancel() - self.createInFlight = nil - let tunnel = self.controlTunnel?.tunnel - self.controlTunnel = nil - await tunnel?.terminate() + self.beginRetirement() + await self.waitForRetirement() } #if DEBUG @@ -457,25 +392,6 @@ actor RemoteTunnelManager { } #endif - private func beginRestart() { - guard !self.restartInFlight else { return } - self.restartInFlight = true - self.lastRestartAt = Date() - self.logger.info("control tunnel restart started") - Task { [weak self] in - guard let self else { return } - try? await Task.sleep(nanoseconds: UInt64(self.restartBackoffSeconds * 1_000_000_000)) - await self.endRestart() - } - } - - private func endRestart() { - if self.restartInFlight { - self.restartInFlight = false - self.logger.info("control tunnel restart finished") - } - } - private func waitForRestartBackoffIfNeeded() async throws { guard let last = lastRestartAt else { return } let elapsed = Date().timeIntervalSince(last) diff --git a/apps/macos/Sources/OpenClaw/TalkMLXSpeechSynthesizer.swift b/apps/macos/Sources/OpenClaw/TalkMLXSpeechSynthesizer.swift index ef6de3353bba..bace8977154e 100644 --- a/apps/macos/Sources/OpenClaw/TalkMLXSpeechSynthesizer.swift +++ b/apps/macos/Sources/OpenClaw/TalkMLXSpeechSynthesizer.swift @@ -679,6 +679,8 @@ private actor ProcessMLXTTSTransport: MLXTTSTransport { do { _ = try await process.waitUntilStarted() } catch { + // The detached launch can still spawn; reap it before closing inherited pipes. + await process.terminate(gracefully: false) output.readabilityHandler = nil continuation.finish() throw error diff --git a/apps/macos/Tests/OpenClawIPCTests/ManagedProcessTests.swift b/apps/macos/Tests/OpenClawIPCTests/ManagedProcessTests.swift index 1f8ddd38ac95..19e8b2f55a97 100644 --- a/apps/macos/Tests/OpenClawIPCTests/ManagedProcessTests.swift +++ b/apps/macos/Tests/OpenClawIPCTests/ManagedProcessTests.swift @@ -5,15 +5,61 @@ import Testing #if canImport(Darwin) struct ManagedProcessTests { - @Test func `launch failures preserve the executable description`() async { - let executable = "/tmp/openclaw-missing-process-\(UUID().uuidString)" + @Test func `cancelled startup wait throws without crashing or consuming the process`() async throws { + let process = try await self.start(executable: "/bin/sh", arguments: ["-c", "sleep 30"]) + defer { process.requestTermination(gracefully: false) } + let (gate, gateContinuation) = AsyncStream.makeStream() + defer { gateContinuation.finish() } + + let waiter = Task { + var iterator = gate.makeAsyncIterator() + _ = await iterator.next() + return try await process.waitUntilStarted() + } + waiter.cancel() do { - _ = try await self.start(executable: executable) - Issue.record("expected the missing executable to fail") + _ = try await waiter.value + Issue.record("expected the cancelled startup waiter to throw") } catch { - #expect(error.localizedDescription.contains(executable)) + #expect(error is CancellationError) } + + #expect(process.isRunning) + await process.terminate(gracefully: false) + #expect(!process.isRunning) + } + + @Test func `startup result is replayed to concurrent observers`() async throws { + let process = try await self.start(executable: "/bin/sh", arguments: ["-c", "sleep 30"]) + defer { process.requestTermination(gracefully: false) } + + async let first = process.waitUntilStarted() + async let second = process.waitUntilStarted() + let processIdentifiers = try await [first, second] + + #expect(processIdentifiers[0] == processIdentifiers[1]) + #expect(processIdentifiers[0] > 0) + await process.terminate(gracefully: false) + } + + @Test func `launch failures preserve and replay the executable description`() async { + let executable = "/tmp/openclaw-missing-process-\(UUID().uuidString)" + let process = ManagedProcess.launch( + configuration: Subprocess.Configuration(executable: .path(.init(executable))), + input: .none, + output: .discarded, + error: .discarded) + + for _ in 0..<2 { + do { + _ = try await process.waitUntilStarted() + Issue.record("expected the missing executable to fail") + } catch { + #expect(error.localizedDescription.contains(executable)) + } + } + await process.terminate(gracefully: false) } @Test func `abortive termination interrupts the stdin grace period`() async throws {