Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
106 changes: 106 additions & 0 deletions cockpit/Sources/CockpitCore/Transport.swift
Original file line number Diff line number Diff line change
Expand Up @@ -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<Route, Error>?
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<Void, Never>?

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<Route, Error> = 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<T: Sendable>(_ 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)
Expand Down
18 changes: 12 additions & 6 deletions cockpit/Sources/cockpit/MCP.swift
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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 }
Expand Down
32 changes: 6 additions & 26 deletions cockpit/Sources/cockpit/main.swift
Original file line number Diff line number Diff line change
Expand Up @@ -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 <name> --scopes read,act)
holds a scoped token (./fleetctl.sh api-token <name> --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:
Expand Down Expand Up @@ -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 {
Expand Down Expand Up @@ -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":
Expand Down Expand Up @@ -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).
Expand Down
Loading
Loading