fix(next): the remote-localhost proxy's trust scan runs one at a time off the cooperative pool

Packages/macOS/CmuxNext/Sources/CmuxNextRemoteLocalhost/PeerProcess.swift

@@ -1,4 +1,6 @@
+import CmuxNextCompat
 import Darwin
+import Dispatch
 import Network
 
 /// Who is on the other end of a loopback TCP connection to the proxy.
@@ -116,3 +118,103 @@ enum PeerProcess {
         return nil
     }
 }
+
+/// The proxy's trust scans. Each is a run of blocking `proc_pid*` calls (one per socket of the app and of
+/// every Chromium helper), so they run one at a time on a queue of their own: a flood of local connections
+/// neither occupies the cooperative pool nor runs many scans at once. A caller's wait is bounded and
+/// cancellation-aware: after `timeout`, or when its task is cancelled, it answers false (not trusted) at
+/// once and its queued scan is skipped. At most `limit` scans are queued or running; past that a caller is
+/// untrusted at once, so a stalled scan cannot grow the queue.
+final class PeerScan: Sendable {
+    /// The longest a caller waits for its scan.
+    static let timeout: Duration = .seconds(2)
+    static let pendingLimit = 32
+
+    private let queue = DispatchQueue(label: "com.cmuxterm.next.remote-localhost.peer-scan", qos: .userInitiated)
+    private let pending = Mutex(0)
+    private let limit: Int
+
+    init(limit: Int = PeerScan.pendingLimit) { self.limit = limit }
+
+    /// ``PeerProcess/isTrusted(peerHost:peerPort:localHost:localPort:candidates:)`` through ``run(timeout:_:)``.
+    func isTrusted(peerHost: NWEndpoint.Host, peerPort: UInt16, localHost: NWEndpoint.Host, localPort: UInt16) async -> Bool {
+        await run {
+            PeerProcess.isTrusted(peerHost: peerHost, peerPort: peerPort, localHost: localHost, localPort: localPort)
+        }
+    }
+
+    func run(timeout: Duration = PeerScan.timeout, _ scan: @escaping @Sendable () -> Bool) async -> Bool {
+        let admitted = pending.withLock { count -> Bool in
+            guard count < limit else { return false }
+            count += 1
+            return true
+        }
+        guard admitted else { return false }
+        let request = ScanRequest()
+        // task-owner: one timer per scan; cancelled below once the scan answers
+        let timer = Task {
+            // wakeup-allow: one-shot bound on a scan's wait, cancelled when the scan answers
+            do { try await Task.sleep(for: timeout) } catch { return }
+            request.cancel()
+        }
+        defer { timer.cancel() }
+        return await withTaskCancellationHandler {
+            await withCheckedContinuation { continuation in
+                request.arm(continuation)
+                queue.async {
+                    defer { self.pending.withLock { $0 -= 1 } }
+                    guard !request.isCancelled else { return }
+                    request.finish(scan())
+                }
+            }
+        } onCancel: {
+            request.cancel()
+        }
+    }
+}
+
+/// One scan's answer, delivered exactly once: the scan's result, or false when the wait was cancelled
+/// or timed out first.
+fileprivate final class ScanRequest: Sendable {
+    private struct State {
+        var continuation: CheckedContinuation<Bool, Never>?
+        var result: Bool?
+        var cancelled = false
+    }
+
+    private let state = Mutex(State())
+
+    var isCancelled: Bool { state.withLock { $0.cancelled } }
+
+    func arm(_ continuation: CheckedContinuation<Bool, Never>) {
+        let settled = state.withLock { state -> Bool? in
+            if let result = state.result { return result }
+            state.continuation = continuation
+            return nil
+        }
+        if let settled { continuation.resume(returning: settled) }
+    }
+
+    func finish(_ trusted: Bool) {
+        let waiting = state.withLock { Self.settle(&$0, trusted) }
+        waiting?.resume(returning: trusted)
+    }
+
+    /// Marks the request cancelled and answers false in one critical section, so a scan that finishes
+    /// at the same time sees either "cancelled and settled" or "not yet cancelled", never a mix.
+    func cancel() {
+        let waiting = state.withLock { state -> CheckedContinuation<Bool, Never>? in
+            state.cancelled = true
+            return Self.settle(&state, false)
+        }
+        waiting?.resume(returning: false)
+    }
+
+    /// Records the answer once and hands back the waiter to resume, if it is armed already.
+    private static func settle(_ state: inout State, _ value: Bool) -> CheckedContinuation<Bool, Never>? {
+        guard state.result == nil else { return nil }
+        state.result = value
+        defer { state.continuation = nil }
+        return state.continuation
+    }
+}

Packages/macOS/CmuxNext/Sources/CmuxNextRemoteLocalhost/ProxyConnection.swift

@@ -23,10 +23,11 @@ final class ProxyConnection: Sendable {
         self.machineListener = machineListener
     }
 
-    /// A machine listener's route when the peer is this app or its helpers.
-    private func trustedRoute() -> RemoteLocalhostProxy.Route? {
+    /// A machine listener's route when the peer is this app or its helpers. The process scan runs off
+    /// the cooperative pool (``PeerScan``).
+    private func trustedRoute() async -> RemoteLocalhostProxy.Route? {
         guard machineListener, case .hostPort(let peerHost, let peerPort) = client.endpoint,
-              PeerProcess.isTrusted(
+              await proxy.peerScan.isTrusted(
                   peerHost: peerHost,
                   peerPort: peerPort.rawValue,
                   localHost: .ipv4(.loopback),
@@ -52,8 +53,10 @@ final class ProxyConnection: Sendable {
             deadline.cancel()
             return
         }
+        // The deadline stays armed through the trust scan (itself bounded by its own timeout).
+        let trusted = await trustedRoute()
         deadline.cancel()
-        guard let route = trustedRoute() ?? proxy.route(forAuthorization: head.proxyAuthorization) else {
+        guard let route = trusted ?? proxy.route(forAuthorization: head.proxyAuthorization) else {
             proxy.count(\.unauthorized)
             proxy.note("407 \(head.kind == .connect ? "CONNECT" : head.method) \(head.host):\(head.port) credential=\(head.proxyAuthorization == nil ? "none" : "wrong")")
             try? await client.sendAll(ProxyResponses.authenticationRequired)

Packages/macOS/CmuxNext/Sources/CmuxNextRemoteLocalhost/RemoteLocalhostProxy.swift

@@ -86,6 +86,8 @@ public final class RemoteLocalhostProxy: Sendable {
     private let state = Mutex(State())
     private let secret: String
     let queue = DispatchQueue(label: "com.cmuxterm.next.remote-localhost", qos: .userInitiated)
+    /// The trust scans of this proxy's machine listeners (``PeerScan``).
+    let peerScan = PeerScan()
     let logger = Logger(subsystem: "com.cmuxterm.app.next", category: "remote-localhost")
     /// Open connections, capped so a flood cannot exhaust descriptors.
     static let maxConnections = 256

Packages/macOS/CmuxNext/Tests/CmuxNextAppTests/PeerProcessScanTests.swift

@@ -0,0 +1,162 @@
+import Foundation
+import Network
+import Testing
+@testable import CmuxNextRemoteLocalhost
+
+/// The proxy's trust check is a run of blocking `proc_pid*` calls. It runs one at a time on its own queue, so
+/// a flood of local connections neither holds cooperative-pool threads nor runs many scans at once, and a wait
+/// for it is bounded: a stalled scan cannot hold the connections queued behind it.
+nonisolated struct PeerProcessScanTests {
+    private final class Gauge: @unchecked Sendable {
+        private let lock = NSLock()
+        private var running = 0
+        private var _peak = 0
+        private var _started = 0
+        private let release = DispatchSemaphore(value: 0)
+        private let stalledStarted = DispatchSemaphore(value: 0)
+
+        var peak: Int { lock.withLock { _peak } }
+        var started: Int { lock.withLock { _started } }
+
+        func slowScan() -> Bool {
+            lock.withLock {
+                running += 1
+                _started += 1
+                _peak = max(_peak, running)
+            }
+            usleep(20_000)
+            lock.withLock { running -= 1 }
+            return true
+        }
+
+        /// A scan that does not finish until `unblock()`.
+        func stalledScan() -> Bool {
+            lock.withLock { _started += 1 }
+            stalledStarted.signal()
+            release.wait()
+            return true
+        }
+
+        func unblock() { release.signal() }
+
+        /// Returns once a stalled scan is running on the queue (without holding a cooperative thread).
+        func waitUntilStalled() async {
+            await withCheckedContinuation { continuation in
+                DispatchQueue.global().async {
+                    self.stalledStarted.wait()
+                    continuation.resume()
+                }
+            }
+        }
+    }
+
+    private static func check(_ scan: PeerScan, timeout: Duration = .seconds(5),
+                              _ work: @escaping @Sendable () -> Bool) async -> Bool {
+        await scan.run(timeout: timeout, work)
+    }
+
+    @Test func scansRunOneAtATime() async {
+        let gauge = Gauge()
+        let scan = PeerScan(limit: 100)
+        await withTaskGroup(of: Bool.self) { group in
+            for _ in 0..<32 { group.addTask { await Self.check(scan) { gauge.slowScan() } } }
+            for await trusted in group { #expect(trusted) }
+        }
+        #expect(gauge.started == 32)
+        #expect(gauge.peak == 1, "\(gauge.peak) scans ran at once")
+    }
+
+    /// With 64 scans pending, the cooperative pool still runs other work: a canary task finishes long before
+    /// the scans (64 x 20 ms) do.
+    @Test func pendingScansDoNotStarveTheCooperativePool() async {
+        let gauge = Gauge()
+        let scan = PeerScan(limit: 100)
+        let started = ContinuousClock.now
+        var canaryDuration = Duration.zero
+        await withTaskGroup(of: Duration?.self) { group in
+            for _ in 0..<64 { group.addTask { _ = await Self.check(scan) { gauge.slowScan() }; return nil } }
+            group.addTask {
+                try? await Task.sleep(for: .milliseconds(50))
+                return ContinuousClock.now - started
+            }
+            for await finished in group { if let finished { canaryDuration = finished } }
+        }
+        #expect(canaryDuration < .milliseconds(600), "the canary waited \(canaryDuration) behind the scans")
+        #expect(gauge.started == 64)
+    }
+
+    /// One scan stalls. The callers behind it answer "not trusted" when their wait runs out, and their scans
+    /// are skipped, so they hold no connection slot and add no work once the stall ends.
+    @Test func aStalledScanDoesNotHoldTheCallersBehindIt() async {
+        let gauge = Gauge()
+        let scan = PeerScan()
+        let stalled = Task { await Self.check(scan, timeout: .seconds(30)) { gauge.stalledScan() } }
+        await gauge.waitUntilStalled()
+        let started = ContinuousClock.now
+        let answers = await withTaskGroup(of: Bool.self, returning: [Bool].self) { group in
+            for _ in 0..<5 { group.addTask { await Self.check(scan, timeout: .milliseconds(150)) { gauge.slowScan() } } }
+            var all: [Bool] = []
+            for await answer in group { all.append(answer) }
+            return all
+        }
+        #expect(answers == [Bool](repeating: false, count: 5))
+        #expect(ContinuousClock.now - started < .seconds(2), "the callers waited for the stalled scan")
+        gauge.unblock()
+        #expect(await stalled.value)
+        // The queue is serial: when this sentinel answers, everything queued before it has run or been skipped.
+        let sentinel = await Self.check(scan) { true }
+        #expect(sentinel)
+        #expect(gauge.started == 1, "\(gauge.started) scans started; the timed-out ones must be skipped")
+    }
+
+    @Test func aCancelledCallerAnswersAtOnceAndItsScanIsSkipped() async {
+        let gauge = Gauge()
+        let scan = PeerScan()
+        let blocker = Task { await Self.check(scan, timeout: .seconds(30)) { gauge.stalledScan() } }
+        await gauge.waitUntilStalled()
+        let waiter = Task { await Self.check(scan, timeout: .seconds(30)) { gauge.slowScan() } }
+        // Cancelling before the waiter reaches the queue is fine too: it answers false either way.
+        waiter.cancel()
+        #expect(await waiter.value == false)
+        gauge.unblock()
+        #expect(await blocker.value)
+        let sentinel = await Self.check(scan) { true }
+        #expect(sentinel)
+        #expect(gauge.started == 1, "the cancelled caller's scan ran")
+    }
+
+    /// Past the pending limit a caller is untrusted at once instead of queueing another block behind a stall.
+    @Test func callersPastThePendingLimitAnswerAtOnce() async {
+        let gauge = Gauge()
+        let scan = PeerScan()
+        let stalled = Task { await Self.check(scan, timeout: .seconds(30)) { gauge.stalledScan() } }
+        await gauge.waitUntilStalled()
+        let extra = 10
+        let waiting = PeerScan.pendingLimit - 1
+        let started = ContinuousClock.now
+        await withTaskGroup(of: Bool.self) { group in
+            for _ in 0..<(waiting + extra) { group.addTask { await Self.check(scan, timeout: .seconds(30)) { gauge.slowScan() } } }
+            // The ones refused return at once; the admitted ones wait on the stall (30 s timeout).
+            var refused = 0
+            for await answer in group {
+                #expect(!answer)
+                refused += 1
+                if refused == extra { break }
+            }
+            #expect(ContinuousClock.now - started < .seconds(5), "the refused callers waited for the stall")
+            group.cancelAll()
+            for await _ in group {}
+        }
+        gauge.unblock()
+        #expect(await stalled.value)
+        let sentinel = await Self.check(scan) { true }
+        #expect(sentinel)
+        #expect(gauge.started == 1, "queued scans ran")
+    }
+
+    @Test func aPeerThatIsNotThisProcessIsNotTrusted() async {
+        let trusted = await PeerScan().isTrusted(peerHost: .ipv4(.loopback), peerPort: 1, localHost: .ipv4(.loopback),
+                                                         localPort: 2)
+        #expect(!trusted)
+    }
+}

plans/cmux-next/threading-model.md

@@ -57,6 +57,7 @@ and changes both with the code).
 | **Main-thread watchdog** thread `cmux-next main-thread watchdog` | Stall/hang detection and stack samples | Its own kqueue (`MainThreadWatchdogAlarm`: a one-shot timer the main thread arms when its run loop wakes and deletes before it sleeps, and a user event for stop and task starts); a main turn shorter than the timer wakes it never | user-interactive | Rings of 128 hangs and 128 over-budget entries |
 | **Daemon connection** reader thread `cmux-tui reader <path>`; writer queue `com.cmuxterm.next.daemon.write`; `actor DaemonConnection`; `EventInbox` pump task | Socket read and line split; outbox drain; handshake and reconnect; off-main event buffer | Blocking `read(2)` (reader); `write` (writer) | user-interactive (reader), default (writer) | Line over 64 MiB closes the connection; outbox 64 MiB, past it `write` fails `.backlog`; inbox 4096 events, field events coalesced, overflow collapses to one resync; main drains 32 per slice under the visible budget |
 | **Cloud link credentials** `cmux.cloud.link-credential` (CloudLinkLauncher) | Listener lifecycle; Mutex protects peer generation and owned request tasks | Bind and peer inspection on the serial queue; socket reads/writes on utility dispatch workers, never the cooperative pool | utility | At most 32 requests including Worker calls; excess connections close. One 64 KiB line, 5 s read deadline and 5 s write timeout. Stop cancels owned tasks and invalidates the generation; each request closes its descriptor |
+| **Remote-localhost peer scan** `com.cmuxterm.next.remote-localhost.peer-scan` (`PeerProcess.isTrustedOffPool`) | Whether a machine listener's peer is this app or a helper | `proc_pidinfo`/`proc_pidfdinfo` over the app's and its children's sockets, one scan at a time | user-initiated | A caller waits at most 2 s, or less if cancelled, then counts as untrusted and its queued scan is skipped; at most 32 scans pending, past that a caller is untrusted at once |
 | **Terminal attach** (`TerminalAttachment` reader → `TerminalEventQueue`) | One terminal's output stream | The reader waits above 8 MiB queued, resumes below 2 MiB | user-interactive | The daemon's attach mailbox overflows and ends the stream |
 | **Terminal output lane** `com.cmuxterm.next.terminal.output`; local PTY `com.cmuxterm.next.localpty.read` / `.write` | Feeding `ghostty_surface_process_output`; local PTY I/O | PTY read/write | user-interactive | 2 MiB high water: the consumer task suspends |
 | **Ghostty renderer and IO threads** (per surface, inside GhosttyKit) | Rendering; termio | Renderer mailbox (64, latest-value-wins, never waits); termio queue 64 | focused user-interactive, visible user-initiated, hidden utility | cmux runs surfaces in manual IO mode: no Ghostty read thread |
@@ -123,7 +124,7 @@ matches the label's literal text; `<x>` matches an interpolation). Its owner row
 - Settings and files: `com.cmuxterm.next.settings.watcher`, `cmux.files.fsevents`, `cmux.files.tree-fsevents`,
   `com.cmuxterm.home.cache` (the `HomeCache` actor's executor, user-initiated: its file writes block this queue,
   never a pool thread, so the forced quit's last draft write runs with the pool full; CmuxHomeCore, shared with iOS)
-- Remote and mobile: `com.cmuxterm.next.remote-localhost`, `com.cmuxterm.next.cloud-browser-proxy`,
+- Remote and mobile: `com.cmuxterm.next.remote-localhost`, `com.cmuxterm.next.remote-localhost.peer-scan`, `com.cmuxterm.next.cloud-browser-proxy`,
   `com.cmuxterm.next.cloud-hub-loopback` (Open Desktop's SOCKS connections to the app's WireGuard hub),
   `cmux.cloud.machine-io.stdin` (one Worker machine exec or file call's request into the bundled cmux's stdin,
   one thread per call: its pipe write blocks only it; it ends when the child has read the request or closed its