From 6e83dd8450f8a95a91334da0972b6874e6895a6a Mon Sep 17 00:00:00 2001 From: Taylor Addison Date: Thu, 22 Oct 2026 14:40:07 -0400 Subject: [PATCH] Close cockpit mcp's tunnel after 90 s idle instead of holding it all session Each `cockpit mcp` (one per open Claude Code session) held an ssh tunnel to ultra for the session's whole life, each one of the 42 sshd slots launchd allows; on 2026-10-22 idle sessions filled 40 and proxied MCP servers failed to connect. The tunnel now opens when a message needs the daemon and closes after --idle seconds (default 90, 0 = hold) with no call in flight; a long wait_for_checks keeps it open until it returns. Concurrent calls share one open. RouteCache moves into CockpitCore as OnDemandRoute, which `cockpit wait` also uses (without the idle close). Co-Authored-By: Claude Opus 5.5 --- cockpit/Sources/CockpitCore/Transport.swift | 106 ++++++++ cockpit/Sources/cockpit/MCP.swift | 18 +- cockpit/Sources/cockpit/main.swift | 32 +-- .../CockpitCoreTests/TransportTests.swift | 247 ++++++++++++++++++ docs/cockpit/agents.md | 9 + docs/cockpit/cli.md | 4 +- 6 files changed, 383 insertions(+), 33 deletions(-) create mode 100644 cockpit/Tests/CockpitCoreTests/TransportTests.swift diff --git a/cockpit/Sources/CockpitCore/Transport.swift b/cockpit/Sources/CockpitCore/Transport.swift index d725c96..33e68db 100644 --- a/cockpit/Sources/CockpitCore/Transport.swift +++ b/cockpit/Sources/CockpitCore/Transport.swift @@ -157,6 +157,112 @@ public final class TunnelTransport: Transport, @unchecked Sendable { deinit { close() } } +/// A transport's route, opened when a call needs it and — with `idle` set — +/// closed again once no call has used it for that long. +/// +/// `cockpit mcp` runs for as long as its Claude Code session, idle nearly all +/// of that time. Holding a tunnel the whole while cost the host one of its 42 +/// ssh slots per open session (launchd's cap on sshd instances); on 2026-10-22 +/// ultra reached 40 and every proxied MCP server failed to connect. With an +/// idle close a session holds a slot only while it is asking the fleet. +/// +/// Concurrent callers share one open: a second call that arrives while the +/// tunnel is coming up waits for it rather than starting another (and +/// `TunnelTransport.open` closes whatever it had before it starts). +public final class OnDemandRoute: @unchecked Sendable { + private let transport: Transport + private let idle: TimeInterval? + private let lock = NSLock() + private var route: Route? + private var opening: Task? + private var users = 0 + /// Bumped whenever the route is taken or dropped, so an idle close that + /// was scheduled before then does nothing. + private var epoch = 0 + private var closer: Task? + + public init(_ transport: Transport, idle: TimeInterval? = nil) { + self.transport = transport + self.idle = idle + } + + /// The route's base URL, opening (or reopening, once the tunnel process + /// has died) as needed. + public func base() async throws -> URL { + let pending: Task = lock.withLock { + if let route, transport.isAlive { return Task { route } } + route = nil + if let opening { return opening } + let t = Task { [transport] in try await transport.open() } + opening = t + return t + } + do { + let r = try await pending.value + lock.withLock { + if opening == pending { opening = nil; route = r } + } + return r.baseURL + } catch { + lock.withLock { if opening == pending { opening = nil } } + throw error + } + } + + /// Runs `body` with the route held open; the idle clock starts when the + /// last call in flight returns. + public func use(_ body: @Sendable (URL) async throws -> T) async throws -> T { + acquire() + defer { release() } + return try await body(try await base()) + } + + /// The route answered nothing: drop it, so the next use opens a new tunnel + /// instead of reconnecting to a forward whose far end is gone. + public func invalidate() { + lock.withLock { route = nil; epoch += 1 } + transport.close() + } + + /// Closes the route for good (the caller is exiting). + public func close() { + lock.withLock { closer?.cancel(); closer = nil } + invalidate() + } + + /// True while a route is held and its tunnel is up. + public var isOpen: Bool { lock.withLock { route != nil } && transport.isAlive } + + private func acquire() { + lock.withLock { + users += 1 + epoch += 1 + closer?.cancel() + closer = nil + } + } + + private func release() { + lock.withLock { + users -= 1 + guard users == 0, let idle else { return } + let mine = epoch + closer = Task { [weak self] in + try? await Task.sleep(nanoseconds: UInt64(idle * 1_000_000_000)) + guard !Task.isCancelled, let self else { return } + // Closed under the lock, so a call that starts now opens + // its tunnel after this one is gone, not before. + self.lock.withLock { + guard self.users == 0, self.epoch == mine else { return } + self.route = nil + self.closer = nil + self.transport.close() + } + } + } + } +} + /// Asks the kernel for an unused loopback port. func freeLocalPort() -> Int? { let fd = socket(AF_INET, SOCK_STREAM, 0) diff --git a/cockpit/Sources/cockpit/MCP.swift b/cockpit/Sources/cockpit/MCP.swift index d953dc0..21eb541 100644 --- a/cockpit/Sources/cockpit/MCP.swift +++ b/cockpit/Sources/cockpit/MCP.swift @@ -10,12 +10,18 @@ import CockpitCore // read and every action is refused. The token `cockpit pair` saves holds // every scope, and an agent holding it could confirm its own previews of // anything, switching a host off included. +// +// The tunnel opens when a message needs the daemon and closes after `idle` +// seconds with none in flight: one of these runs for every open Claude Code +// session, and each held tunnel was one of the host's 42 ssh slots. enum MCPBridge { static let out = NSLock() - static func run(_ link: Transport) async { - let routes = RouteCache(link) + static let idle: TimeInterval = 90 + + static func run(_ link: Transport, idle: TimeInterval?) async { + let routes = OnDemandRoute(link, idle: idle) let token = ProcessInfo.processInfo.environment["FLEET_API_TOKEN"].flatMap { $0.isEmpty ? nil : $0 } await withTaskGroup(of: Void.self) { group in // wait_for_checks holds a call for minutes; a ping or a second @@ -25,14 +31,14 @@ enum MCPBridge { group.addTask { await forward(data, routes: routes, token: token) } } } - link.close() + routes.close() } - static func forward(_ data: Data, routes: RouteCache, token: String?) async { + static func forward(_ data: Data, routes: OnDemandRoute, token: String?) async { let id = (try? JSONSerialization.jsonObject(with: data) as? [String: Any])?["id"] do { - let client = FleetClient(base: try await routes.base(), token: token) - if let reply = try await client.mcp(data) { write(reply) } + let reply = try await routes.use { base in try await FleetClient(base: base, token: token).mcp(data) } + if let reply { write(reply) } } catch { routes.invalidate() guard let id else { return } diff --git a/cockpit/Sources/cockpit/main.swift b/cockpit/Sources/cockpit/main.swift index f3f3df6..eab4a56 100644 --- a/cockpit/Sources/cockpit/main.swift +++ b/cockpit/Sources/cockpit/main.swift @@ -30,7 +30,9 @@ commands: anything but a read needs --yes mcp the dashboard's MCP tools over stdio (fleet_status, diagnose_repo, wait_for_checks, list_queue, …). Read-only unless FLEET_API_TOKEN - holds a scoped token (./fleetctl.sh api-token --scopes read,act) + holds a scoped token (./fleetctl.sh api-token --scopes read,act). + The tunnel opens when a call needs it and closes after --idle seconds + with none (default 90; 0 holds it for the whole session) fixtures list the bundled fixture names options: @@ -149,29 +151,6 @@ func obtain(_ o: Options) async -> Source { } } -/// The dashboard route for a wait: opened on first use, reopened once the -/// tunnel process has died. -final class RouteCache: @unchecked Sendable { - private let transport: Transport - private var route: Route? - private let lock = NSLock() - init(_ t: Transport) { transport = t } - - func base() async throws -> URL { - if let r = lock.withLock({ route }), transport.isAlive { return r.baseURL } - let r = try await transport.open() - lock.withLock { route = r } - return r.baseURL - } - - /// The route answered nothing: drop it, so the next use opens a new tunnel - /// instead of reconnecting to a forward whose far end is gone. - func invalidate() { - lock.withLock { route = nil } - transport.close() - } -} - let defaultAliases = TransportConfig.defaultAliases func transport(_ o: Options) -> Transport { @@ -248,7 +227,8 @@ let opts = parse(Array(CommandLine.arguments.dropFirst())) switch opts.command { case "mcp": - await MCPBridge.run(transport(opts)) + let idle = opts.flags["idle"].flatMap(TimeInterval.init) ?? MCPBridge.idle + await MCPBridge.run(transport(opts), idle: idle > 0 ? idle : nil) case "fixtures": for n in Fixtures.names { print(n) } case "status": @@ -359,7 +339,7 @@ case "wait": }() if opts.fixture != nil { fail("wait asks the live dashboard; fixtures cover status, queue and brief", code: 64) } let link = transport(opts) - let routes = RouteCache(link) + let routes = OnDemandRoute(link) // The last word on the deadline, outside Swift concurrency altogether: if // anything below still has not answered `timeout` + 30 s from now, say so // and exit (2026-10-09: waits ran 1.5 h past --timeout). diff --git a/cockpit/Tests/CockpitCoreTests/TransportTests.swift b/cockpit/Tests/CockpitCoreTests/TransportTests.swift new file mode 100644 index 0000000..98f719a --- /dev/null +++ b/cockpit/Tests/CockpitCoreTests/TransportTests.swift @@ -0,0 +1,247 @@ +import Foundation +import Testing +@testable import CockpitCore + +/// A transport that counts its opens and closes, routed to a stub daemon. +final class CountingTransport: Transport, @unchecked Sendable { + let base: URL + let openDelay: TimeInterval + private let lock = NSLock() + private var alive = false + private(set) var opens = 0 + private(set) var closes = 0 + + init(base: URL, openDelay: TimeInterval = 0) { + self.base = base + self.openDelay = openDelay + } + + var isAlive: Bool { lock.withLock { alive } } + + func open() async throws -> Route { + lock.withLock { opens += 1 } + if openDelay > 0 { try await Task.sleep(nanoseconds: UInt64(openDelay * 1_000_000_000)) } + lock.withLock { alive = true } + return Route(baseURL: base, label: "stub") + } + + func close() { lock.withLock { if alive { closes += 1 }; alive = false } } + + /// The tunnel process died on its own (sleep, network change, sshd restart). + func die() { lock.withLock { alive = false } } + + var counts: (opens: Int, closes: Int) { lock.withLock { (opens, closes) } } +} + +private func waitUntil(_ seconds: TimeInterval = 5, _ cond: () -> Bool) async throws { + let deadline = Date().addingTimeInterval(seconds) + while !cond(), Date() < deadline { try await Task.sleep(nanoseconds: 20_000_000) } +} + +@Suite("on-demand route: open, idle close, reopen", .serialized) +struct OnDemandRouteTests { + private func server() async throws -> StubServer { + let s = try StubServer(routes: ["/api/health": .json(200, "{\"ok\":true}")]) + try await s.start() + return s + } + + @Test func opensOnFirstUseNotBefore() async throws { + let s = try await server() + defer { s.stop() } + let t = CountingTransport(base: s.base) + let r = OnDemandRoute(t, idle: 0.3) + #expect(t.counts.opens == 0) + #expect(!r.isOpen) + let ok = try await r.use { try await Probe.health($0) } + #expect(ok) + #expect(t.counts.opens == 1) + #expect(r.isOpen) + r.close() + } + + @Test func closesAfterIdleAndReopensOnTheNextCall() async throws { + let s = try await server() + defer { s.stop() } + let t = CountingTransport(base: s.base) + let r = OnDemandRoute(t, idle: 0.3) + _ = try await r.use { try await Probe.health($0) } + // Calls within the idle window reuse the tunnel. + _ = try await r.use { try await Probe.health($0) } + #expect(t.counts.opens == 1) + + try await waitUntil { !r.isOpen } + #expect(!r.isOpen) + #expect(t.counts.closes == 1) + + _ = try await r.use { try await Probe.health($0) } + #expect(t.counts.opens == 2) + #expect(r.isOpen) + r.close() + #expect(t.counts.closes == 2) + } + + @Test func aCallInFlightKeepsTheTunnelPastTheIdleWindow() async throws { + let s = try await server() + defer { s.stop() } + let t = CountingTransport(base: s.base) + let r = OnDemandRoute(t, idle: 0.2) + // wait_for_checks holds one call for minutes; the tunnel must stay up + // under it even while other, short calls come and go. + let long = Task { try await r.use { _ in try await Task.sleep(nanoseconds: 800_000_000) } } + try await waitUntil { r.isOpen } + _ = try await r.use { try await Probe.health($0) } + try await Task.sleep(nanoseconds: 500_000_000) + #expect(r.isOpen) + #expect(t.counts.closes == 0) + try await long.value + try await waitUntil { !r.isOpen } + #expect(t.counts.closes == 1) + #expect(t.counts.opens == 1) + } + + @Test func aNewCallCancelsAPendingIdleClose() async throws { + let s = try await server() + defer { s.stop() } + let t = CountingTransport(base: s.base) + let r = OnDemandRoute(t, idle: 1) + _ = try await r.use { try await Probe.health($0) } + try await Task.sleep(nanoseconds: 600_000_000) + _ = try await r.use { try await Probe.health($0) } + // 1.2 s after the first call, 0.6 s after the second: still open. + try await Task.sleep(nanoseconds: 600_000_000) + #expect(r.isOpen) + #expect(t.counts.closes == 0) + try await waitUntil { !r.isOpen } + #expect(t.counts.closes == 1) + } + + @Test func concurrentCallsShareOneOpen() async throws { + let s = try await server() + defer { s.stop() } + let t = CountingTransport(base: s.base, openDelay: 0.2) + let r = OnDemandRoute(t, idle: 1) + try await withThrowingTaskGroup(of: Bool.self) { g in + for _ in 0..<5 { g.addTask { try await r.use { try await Probe.health($0) } } } + for try await ok in g { #expect(ok) } + } + #expect(t.counts.opens == 1) + r.close() + } + + @Test func aDeadTunnelIsReopened() async throws { + let s = try await server() + defer { s.stop() } + let t = CountingTransport(base: s.base) + let r = OnDemandRoute(t, idle: 5) + _ = try await r.use { try await Probe.health($0) } + t.die() + _ = try await r.use { try await Probe.health($0) } + #expect(t.counts.opens == 2) + r.close() + } + + @Test func invalidateDropsTheRoute() async throws { + let s = try await server() + defer { s.stop() } + let t = CountingTransport(base: s.base) + let r = OnDemandRoute(t, idle: 5) + _ = try await r.base() + r.invalidate() + #expect(!r.isOpen) + _ = try await r.base() + #expect(t.counts.opens == 2) + r.close() + } + + @Test func withoutIdleTheRouteIsHeld() async throws { + let s = try await server() + defer { s.stop() } + let t = CountingTransport(base: s.base) + let r = OnDemandRoute(t) + _ = try await r.use { try await Probe.health($0) } + try await Task.sleep(nanoseconds: 300_000_000) + #expect(r.isOpen) + r.close() + } +} + +/// A stand-in for /usr/bin/ssh: forwards `-L 127.0.0.1:P:127.0.0.1:R` locally, +/// writes its pid to `.pid`, and exits when stdin closes — as the real +/// tunnel's remote `cat` does. +private let fakeSSH = """ +import os, socket, sys, threading +args = sys.argv[1:] +fwd = args[args.index("-L") + 1].split(":") +lport, rport = int(fwd[1]), int(fwd[3]) +alias = args[args.index("-L") + 2] +open(alias + ".pid", "w").write(str(os.getpid())) +def pipe(a, b): + try: + while True: + d = a.recv(65536) + if not d: break + b.sendall(d) + except OSError: pass + for s in (a, b): + try: s.shutdown(socket.SHUT_RDWR) + except OSError: pass +def serve(): + ls = socket.socket(); ls.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + ls.bind(("127.0.0.1", lport)); ls.listen(16) + while True: + c, _ = ls.accept() + r = socket.create_connection(("127.0.0.1", rport)) + threading.Thread(target=pipe, args=(c, r), daemon=True).start() + threading.Thread(target=pipe, args=(r, c), daemon=True).start() +threading.Thread(target=serve, daemon=True).start() +sys.stdin.buffer.read() +os._exit(0) +""" + +private let python3 = ["/opt/homebrew/bin/python3", "/usr/local/bin/python3", "/usr/bin/python3"] + .first { FileManager.default.isExecutableFile(atPath: $0) } + +@Suite("tunnel process: opened on demand, gone after idle", .serialized) +struct TunnelLifecycleTests { + @Test(.enabled(if: python3 != nil, "needs python3 for the fake ssh")) + func theTunnelProcessExitsAfterIdleAndComesBack() async throws { + let dir = FileManager.default.temporaryDirectory.appendingPathComponent("cockpit-tunnel-\(UUID().uuidString)") + try FileManager.default.createDirectory(at: dir, withIntermediateDirectories: true) + defer { try? FileManager.default.removeItem(at: dir) } + let ssh = dir.appendingPathComponent("ssh") + try "#!\(python3!)\n\(fakeSSH)".write(to: ssh, atomically: true, encoding: .utf8) + try FileManager.default.setAttributes([.posixPermissions: 0o755], ofItemAtPath: ssh.path) + + let s = try StubServer(routes: ["/api/health": .json(200, "{\"ok\":true}")]) + try await s.start() + defer { s.stop() } + + let alias = dir.appendingPathComponent("host").path + let pidFile = alias + ".pid" + func pid() throws -> pid_t { pid_t(try String(contentsOfFile: pidFile, encoding: .utf8))! } + func running(_ p: pid_t) -> Bool { kill(p, 0) == 0 } + + let t = TunnelTransport(aliases: [alias], remotePort: s.port, sshPath: ssh.path) + let r = OnDemandRoute(t, idle: 0.5) + #expect(!FileManager.default.fileExists(atPath: pidFile)) + + #expect(try await r.use { try await Probe.health($0) }) + let first = try pid() + #expect(running(first)) + + // Idle: the process (and with it the host's session) goes away. + try await waitUntil { !running(first) } + #expect(!running(first)) + #expect(!r.isOpen) + + // The next call brings up a new tunnel and still answers. + #expect(try await r.use { try await Probe.health($0) }) + let second = try pid() + #expect(second != first) + #expect(running(second)) + r.close() + try await waitUntil { !running(second) } + #expect(!running(second)) + } +} diff --git a/docs/cockpit/agents.md b/docs/cockpit/agents.md index 8ede0e5..59152ca 100644 --- a/docs/cockpit/agents.md +++ b/docs/cockpit/agents.md @@ -23,6 +23,15 @@ claude mcp add --scope user fleet-cockpit -- ~/.local/bin/cockpit mcp claude mcp get fleet-cockpit # Status: ✔ Connected ``` +The tunnel isn't held for the session's life. It opens when a call needs the +daemon and closes after 90 seconds with no call in flight (a `wait_for_checks` +that holds a call for minutes keeps it open until it returns). One `cockpit mcp` +runs for every open Claude Code session, and each tunnel takes one of the host's +SSH slots: launchd caps sshd at 42 at once, and on 2026-10-22 idle sessions' +tunnels filled 40 of them. The first call after a quiet spell waits a second or +two for SSH. `--idle ` changes the window; `--idle 0` holds the tunnel +for the whole session, as before. + Any MCP client that can launch a stdio server can use the same command. On the host itself, or through an SSH forward of port 7878, a client can talk to `http://localhost:7878/mcp` directly instead. diff --git a/docs/cockpit/cli.md b/docs/cockpit/cli.md index 75621be..04f8af2 100644 --- a/docs/cockpit/cli.md +++ b/docs/cockpit/cli.md @@ -409,7 +409,9 @@ success, `1` on failure or refusal. ## `cockpit mcp` -The dashboard's MCP tools over stdio. See [Agent sessions](agents.md). +The dashboard's MCP tools over stdio. The tunnel opens when a call needs it and +closes after `--idle` seconds with none in flight (default `90`; `0` holds it +for the whole session). See [Agent sessions](agents.md#the-mcp-server). ## `cockpit fixtures`