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.
2026-08-27 12:56:01 -06:00 · 2026-08-26 13:25:46 -07:00
parent 38966daa33
commit 71a7283792
5 changed files with 255 additions and 257 deletions
@@ -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<pid_t, StartFailure>?
+        private var startWaiters: [UUID: CheckedContinuation<pid_t, any Error>] = [:]
        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<pid_t, StartFailure>) {
+            let waiters = self.lock.withLock {
+                guard self.startResult == nil else { return [CheckedContinuation<pid_t, any Error>]() }
+                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<pid_t, any Error>? = 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<Result<pid_t, StartFailure>>
    private let wakeContinuation: AsyncStream<WakeReason>.Continuation
    let completionTask: Task<TerminationStatus?, Never>

@@ -67,12 +100,10 @@ final class ManagedProcess: @unchecked Sendable {

    private init(
        state: State,
-        startEvents: AsyncStream<Result<pid_t, StartFailure>>,
        wakeContinuation: AsyncStream<WakeReason>.Continuation,
        completionTask: Task<TerminationStatus?, Never>)
    {
        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<pid_t, StartFailure>.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) {
@@ -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
@@ -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<RemotePortTunnel, Error>)?
+    private var retirementInFlight: (token: UUID, task: Task<Void, Never>)?
    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)
@@ -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
@@ -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<Void>.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 {