diff --git a/native/macos/MCPProxy/MCPProxy/API/APIClient.swift b/native/macos/MCPProxy/MCPProxy/API/APIClient.swift index 33e8a44d..2b7ed4bf 100644 --- a/native/macos/MCPProxy/MCPProxy/API/APIClient.swift +++ b/native/macos/MCPProxy/MCPProxy/API/APIClient.swift @@ -44,7 +44,15 @@ actor APIClient { /// Pass an empty string to force TCP-only mode. /// - baseURL: TCP base URL. Used as fallback or when socket is unavailable. /// - apiKey: Optional API key for authentication. - init(socketPath: String? = nil, baseURL: String = "http://127.0.0.1:8080", apiKey: String? = nil) { + /// - requestTimeout: Per-request timeout. The default is deliberately + /// generous; a liveness probe wants something much shorter so one slow + /// response cannot stall it (see `CoreProcessManager.probeTimeout`). + init( + socketPath: String? = nil, + baseURL: String = "http://127.0.0.1:8080", + apiKey: String? = nil, + requestTimeout: TimeInterval = 30 + ) { self.baseURL = baseURL self.apiKey = apiKey @@ -54,11 +62,11 @@ actor APIClient { // so it's safe to register even before the socket file exists. if let path = socketPath, path.isEmpty { // Explicitly requested TCP-only - self.session = SocketTransport.makeTCPSession() + self.session = SocketTransport.makeTCPSession(timeout: requestTimeout) } else { // Always use socket-backed session — SocketURLProtocol falls through // to standard networking if the socket file doesn't exist yet. - self.session = SocketTransport.makeURLSession(socketPath: socketPath) + self.session = SocketTransport.makeURLSession(socketPath: socketPath, timeout: requestTimeout) } } diff --git a/native/macos/MCPProxy/MCPProxy/API/SSEClient.swift b/native/macos/MCPProxy/MCPProxy/API/SSEClient.swift index d9cedc8a..293e563b 100644 --- a/native/macos/MCPProxy/MCPProxy/API/SSEClient.swift +++ b/native/macos/MCPProxy/MCPProxy/API/SSEClient.swift @@ -329,7 +329,10 @@ actor SSEClient { if useSocket { config.protocolClasses = [SocketURLProtocol.self] if let socketPath { - SocketURLProtocol.overrideSocketPath = socketPath + // Per-session, not a process-global (see SocketURLProtocol). + config.httpAdditionalHeaders = [ + SocketURLProtocol.routeHeader: SocketURLProtocol.makeRoute(to: socketPath) + ] } } diff --git a/native/macos/MCPProxy/MCPProxy/Core/CoreProcessManager.swift b/native/macos/MCPProxy/MCPProxy/Core/CoreProcessManager.swift index 4badda95..168578c8 100644 --- a/native/macos/MCPProxy/MCPProxy/Core/CoreProcessManager.swift +++ b/native/macos/MCPProxy/MCPProxy/Core/CoreProcessManager.swift @@ -58,7 +58,27 @@ actor CoreProcessManager { /// mislabel that tray-spawned core as `.externalAttached` — after which the /// tray would refuse to stop it and leak it on quit. private var superseded: Bool = false - private var retryCount: Int = 0 + + /// True while an operation that establishes a connection to a core is + /// running: an attach, or a reconnection. + /// + /// Every such operation SUSPENDS (probe, backoff sleep, connect, relaunch), + /// and there are four things that can start one — the liveness tick, the SSE + /// disconnect handler, the process-exit handler, and the attach watcher + /// racing a manual retry. Without a gate, two can run at once: two + /// `connectToCore()` calls replacing each other's clients and tasks, a late + /// failure from one overwriting the other's `.connected` with `.error`, or — + /// for a core we own — two `launchAndConnect()` calls that each spawn a core + /// and orphan one holding the BBolt lock. + /// + /// Acquire it ONLY through `beginConnectionWork()`, which is synchronous and + /// therefore atomic under actor isolation. A check followed by an `await` + /// followed by a set guards nothing: another invocation runs during the + /// suspension and passes the same check. + private var connectionWorkInFlight: Bool = false + /// Rungs of the relaunch ladder used so far. Readable for tests, which need + /// to see that a successful relaunch clears it. + private(set) var retryCount: Int = 0 private let maxRetries: Int = 3 private let notificationService: NotificationService private let reconnectionPolicy: ReconnectionPolicy @@ -81,20 +101,126 @@ actor CoreProcessManager { /// Socket path for the core process. private let socketPath: String + /// How often the periodic refresh tick runs. Injectable so tests can drive + /// the tick — and the liveness probe on it — without waiting 30 seconds. + private let refreshInterval: TimeInterval + + /// Per-probe timeout. A liveness probe must fail fast: it runs on the + /// refresh tick and everything behind it waits. The API client's own 30s + /// request timeout is right for a real call and far too long here — it + /// would let one slow sample stall the detector for a whole tick. + /// + /// 5s: three orders of magnitude above a healthy socket round-trip + /// (sub-millisecond) and the same order as the core's own advertised SSE + /// retry interval, so a core that is merely busy still gets to answer. + private let probeTimeout: TimeInterval + + /// Consecutive missed readiness checks required before the tray concludes a + /// still-listening core is lost. + /// + /// 2, not 1: readiness is a sample, and one miss is not evidence — a + /// saturated core, a GC pause, or a laptop resuming from sleep can all eat + /// one. It is not higher because each strike costs a full refresh interval, + /// and 2 already bounds worst-case detection at ~1 minute for this case. + /// The unambiguous case (socket gone) is not subject to the budget at all. + private let probeFailureBudget: Int = 2 + + /// Missed readiness checks since the last successful one. Reset on any + /// success, so the budget counts CONSECUTIVE misses — otherwise a healthy + /// core that blips once an hour would eventually be declared dead. + private var consecutiveProbeFailures: Int = 0 + + /// Short-timeout client used only for liveness probes. + private var probeClient: APIClient? + + /// Set by `shutdown()` before it does anything else, and never cleared: a + /// manager is single-use, like `superseded`. + /// + /// Setting `.idle` is not enough to stop work already in flight. A ladder + /// sitting in its backoff sleep will wake up and launch a core after the + /// user pressed Stop, leaving the tray showing "Stopped" while owning a + /// running process — worse than showing an error, because nothing invites + /// the user to look. Every step that can lead to a spawn checks this. + private var shutdownRequested: Bool = false + + /// How many core processes this manager has actually started. + /// + /// The authoritative count for the invariant "never two cores on one data + /// directory": it is incremented inside the one function that can create a + /// process, so it cannot be fooled by a launched process that dies before it + /// records itself. (An earlier version of these tests counted lines written + /// by the launched script and reported ZERO while four cores were being + /// spawned and reaped.) + private(set) var coresLaunched: Int = 0 + + /// Which launch a termination callback belongs to. A process we gave up on + /// and reaped must not, when its callback finally lands, drive a relaunch. + private var launchGeneration: Int = 0 + + /// Fires when a core has been holding the socket without answering for too + /// long, turning an indefinite wait into something the user can act on. + private var unresponsiveDeadlineTask: Task? + + /// What the socket probes of the CURRENT connection episode have shown. + /// + /// The wait path deliberately refuses to act on a single refused probe. That + /// leaves one question open at the deadline: is this a live core we must keep + /// our hands off, or a file a dead one left behind? Nobody else can answer + /// it — the tray no longer unlinks anything, and the core's own + /// `cleanupStaleSocket` (internal/server/listener_unix.go) does not run until + /// a core is launched — so a stale file would deadlock startup forever. + /// Accumulating the probes turns "one sample" into evidence. + private var socketEvidence = SocketEvidence() + + /// Whether the wait we are currently in is allowed to end in a spawn. Carried + /// from `start(maySpawn:)` so the escalation below cannot become a back door + /// around the launch policy (#410) sixty seconds after it was applied. + private var waitMaySpawn: Bool = false + + /// How long to wait for a core that holds the socket but will not answer + /// before saying so. Matches `waitForSocket`'s 60s budget for a core we + /// launched ourselves: a real core can be slow to start (Docker pulls, a + /// large config) and this must not undercut that, but waiting forever is + /// safe for the database and unusable for the person. + private let unresponsiveCoreTimeout: TimeInterval + + /// How long a core we launched has to create its socket before the attempt + /// is abandoned (and the process reaped). Injectable so a test can exercise + /// the ladder without waiting a real minute per rung. + private let socketWaitTimeout: TimeInterval + // MARK: - Initialization + /// - Parameters: + /// - socketPath: Path of the core's Unix socket. Defaults to + /// `~/.mcpproxy/mcpproxy.sock`. Injectable so a test (or a second app + /// instance) can be pointed at an isolated core instead of contending + /// with the user's live one — without it, nothing about attach mode is + /// testable, because attaching IS "probe this socket" (GH #926). + /// - refreshInterval: Period of the background refresh/liveness tick. + /// - probeTimeout: How long a single liveness probe may take before it + /// counts as a miss. See `probeTimeout` above. init( appState: AppState, notificationService: NotificationService, - reconnectionPolicy: ReconnectionPolicy = .default + reconnectionPolicy: ReconnectionPolicy = .default, + socketPath: String? = nil, + refreshInterval: TimeInterval = 30.0, + probeTimeout: TimeInterval = 5.0, + unresponsiveCoreTimeout: TimeInterval = 60.0, + socketWaitTimeout: TimeInterval = 60.0 ) { self.appState = appState self.notificationService = notificationService self.reconnectionPolicy = reconnectionPolicy + self.refreshInterval = refreshInterval + self.probeTimeout = probeTimeout + self.unresponsiveCoreTimeout = unresponsiveCoreTimeout + self.socketWaitTimeout = socketWaitTimeout // Compute socket path: ~/.mcpproxy/mcpproxy.sock let home = FileManager.default.homeDirectoryForCurrentUser.path - self.socketPath = "\(home)/.mcpproxy/mcpproxy.sock" + self.socketPath = socketPath ?? "\(home)/.mcpproxy/mcpproxy.sock" } // MARK: - Public API @@ -111,24 +237,90 @@ actor CoreProcessManager { /// goes idle and watches for one to appear, so a core the user starts /// later (CLI, launchd, brew services) is picked up without a restart. func start(maySpawn: Bool = true) async { - if await attachIfCoreIsRunning() { return } + // The gate covers the WHOLE of start, spawn included. `retry()` kills the + // managed process and then calls this; that process's termination + // handler is on its way to starting a reconnection at the same time. + // Without the gate around the spawn, both launch — two cores on one data + // directory. + guard !shutdownRequested else { return } + guard beginConnectionWork() else { + NSLog("[MCPProxy] start: another connection attempt is already running") + return + } + defer { endConnectionWork() } + + // A fresh attempt reasons from fresh probes: whatever the socket did + // during an earlier episode says nothing about this one. + beginSocketEvidence() + + switch await attachIfCoreIsRunningLocked() { + case .attached: + return + case .inFlight: + // Cannot happen while we hold the gate, but treat it as "someone + // else owns this" rather than as an invitation to spawn. + return + case .unusable(let reason): + // Something is, or may be, on the other end of that socket. Do NOT + // unlink it and do NOT start a competing core against the same data + // directory. Wait instead — the attach watch re-probes and attaches + // the moment it answers, and a deadline turns "waiting" into + // something the user can act on if it never does. + await waitForTheCoreThatIsAlreadyThere(reason: reason, maySpawn: maySpawn) + return + case .noCore: + break + } guard maySpawn else { await awaitExternalCore() return } - // Stale socket — remove it so our new core can create a fresh one. - if SocketTransport.isSocketAvailable(path: socketPath) { - try? FileManager.default.removeItem(atPath: socketPath) + // Nothing answered — but "nothing answered once" is not proof. Confirm + // over a short window before doing anything destructive: a live core + // will accept at least one connection, a stale socket file will refuse + // every one. + guard await confirmNothingIsListening() else { + await waitForTheCoreThatIsAlreadyThere( + reason: "something is listening on the socket", maySpawn: maySpawn + ) + return } + guard !shutdownRequested else { return } + + // Confirmed empty — and the tray does NOT unlink the socket file here. + // A socket file left over from a dead core never reaches this line: it + // probes `.refused`, which `attachIfCoreIsRunningLocked()` reports as + // `.unusable` above, sending us down the wait path. So the only file + // that could be here is one that appeared AFTER the probes — i.e. one + // something just bound — and unlinking that strands a live core. + // + // Cleaning up a genuinely stale file is the core's job anyway: it + // removes one it can prove nobody is listening on (see + // `cleanupStaleSocket` in internal/server/listener_unix.go) and refuses + // to start when the socket really is in use. // Launch our own core as a subprocess await MainActor.run { appState.isStopped = false appState.ownership = .trayManaged } - await launchAndConnect() + await launchWithRetries() + } + + /// A core we cannot use is holding the socket. Wait for it rather than + /// fighting it — but not forever (see `unresponsiveCoreTimeout`). + /// + /// - Parameter maySpawn: whether this wait is allowed to end in a launch if + /// the deadline expires having never seen anything accept a connection. + private func waitForTheCoreThatIsAlreadyThere(reason: String, maySpawn: Bool) async { + NSLog("[MCPProxy] Not taking over %@: %@ — waiting", socketPath, reason) + waitMaySpawn = maySpawn + await MainActor.run { appState.isStopped = false } + await transitionState(to: .waitingForCore) + startAttachWatch() + armUnresponsiveCoreDeadline(reason: reason) } /// Retire this manager: stop watching and never touch AppState again. Called @@ -136,18 +328,224 @@ actor CoreProcessManager { func supersede() { superseded = true cancelAttachWatch() + cancelUnresponsiveCoreDeadline() } - /// Attach to a core that is already up. Returns false when none is. - private func attachIfCoreIsRunning() async -> Bool { - guard !superseded else { return false } - guard SocketTransport.isSocketAvailable(path: socketPath) else { return false } - guard await probeExternalCore() else { return false } - guard !superseded else { return false } // a replacement took over while we probed - await attachToExternalCore() + /// Whether `shutdown()` has been called. Read by background loops so they + /// stop instead of waking up into a spawn. + var isShuttingDown: Bool { shutdownRequested } + + // MARK: - Private: Connection-work gate + + /// Claim the right to establish a connection. Synchronous on purpose: under + /// actor isolation a check-and-set with no `await` between them cannot be + /// interleaved, which is exactly what makes this a guard rather than a hint. + /// + /// Internal rather than private only so tests can take the gate before + /// calling `preflightLaunch()` — its first check is that the gate is held. + func beginConnectionWork() -> Bool { + guard !connectionWorkInFlight else { return false } + connectionWorkInFlight = true return true } + func endConnectionWork() { + connectionWorkInFlight = false + } + + // MARK: - Is a core alive at this socket, and how sure are we? + + /// One classification of the socket probe, shared by everything that asks. + /// Startup and the attach watcher consult `assessCore()` itself; the + /// liveness tick switches on `probeSocket()` directly because it counts + /// strikes per case, but it applies the SAME mapping — absent = act now, + /// localFailure = not evidence either way, refused = a strike, connectable = + /// go ask `/ready`. + /// + /// That mapping used to differ between the two: the tick classified errno + /// carefully while startup flattened everything that was not connectable + /// into "no core", so a full listen backlog or `EMFILE` at startup still led + /// to unlinking a live core's socket and spawning over it. + enum CoreLiveness: Equatable { + /// Socket connectable AND `/ready` answered. A core is definitely there. + case alive + /// Connectable, but no answer. Something IS listening — never take its + /// socket. + case listeningNoAnswer + /// We cannot tell: refused connection (stale socket file OR a live core + /// with a full listen queue) or a local failure (EMFILE/ENOMEM/EACCES). + /// Never act destructively on this without confirmation over time. + case indeterminate(String) + /// No socket file at all. A running core always owns its socket file. + case gone + + /// Whether it is categorically unsafe to take this socket over. + var somethingIsListening: Bool { + switch self { + case .alive, .listeningNoAnswer: return true + case .indeterminate, .gone: return false + } + } + } + + /// Everything the probes have said about this socket since the current + /// connection episode began. Deliberately conservative: it can only ever + /// conclude "nothing has EVER been there", never "something is there". + struct SocketEvidence: Equatable { + /// Probes that found a socket file and were refused by it. + private(set) var refusals: Int = 0 + + /// Whether any probe got a connection accepted. One is enough, and it is + /// never forgotten within the episode: a core that answered a minute ago + /// and refuses now (a full listen backlog looks exactly like a stale + /// file) is a core, and taking its socket is two writers on one database. + private(set) var somethingAccepted: Bool = false + + mutating func record(_ probe: SocketTransport.SocketProbe) { + switch probe { + case .connectable: + somethingAccepted = true + case .refused: + refusals += 1 + case .absent: + // No file to be wrong about. Not evidence of a live core, and + // not counted as a refusal either. + break + case .localFailure: + // Descriptor exhaustion, ENOMEM, EACCES: our problem, not the + // core's. It says nothing about the other end, so it can neither + // prove staleness nor disprove it. + break + } + } + + /// A socket FILE exists and not one connection has ever been accepted on + /// it. The only reading of that which is consistent with everything we + /// have seen is: the process that created it is gone. + var onlyEverRefused: Bool { refusals > 0 && !somethingAccepted } + } + + /// Fold one probe into the episode's evidence. Called from the two places + /// that probe on behalf of a decision — the liveness assessment and the + /// confirmation window — so the deadline reasons over every sample, not just + /// the last one. + private func record(probe: SocketTransport.SocketProbe) { + socketEvidence.record(probe) + } + + /// Forget what earlier episodes saw. A core that was alive before it died + /// must not leave `somethingAccepted` behind to veto the recovery of the + /// socket file it left on disk. + private func beginSocketEvidence() { + socketEvidence = SocketEvidence() + } + + /// Probe the socket, then (only if something answered the connect) ask + /// `/ready`. The single source of truth for core presence. + private func assessCore() async -> CoreLiveness { + let probe = SocketTransport.probeSocket(path: socketPath) + record(probe: probe) + switch probe { + case .absent: + return .gone + case .localFailure(let code): + return .indeterminate("liveness probe could not run (errno \(code))") + case .refused: + return .indeterminate("socket refused the connection") + case .connectable: + return await probeIsReady() ? .alive : .listeningNoAnswer + } + } + + /// Confirm over time that nothing is listening, before doing anything + /// destructive (unlinking the socket, spawning a core). + /// + /// A single failed connect cannot distinguish a stale socket file from a + /// live core whose listen queue is momentarily full — but a core that is + /// really there will accept at least one connection across several attempts, + /// and a stale file will refuse every one. This is the only honest way to + /// turn `.indeterminate` into a decision, and being wrong here means two + /// writers on one database. + /// + /// Internal rather than private so a test can start a listener between two + /// of its probes and pin that it notices. + /// + /// - Parameter afterEachProbe: called with the index of each probe once its + /// result is in. The seam exists for the test of the race this function is + /// here to win: to exercise it the listener must bind strictly BETWEEN two + /// probes, and a test that starts a task and sleeps cannot establish that + /// ordering — if the task runs late its first probe already sees the + /// listener and the race is never run. Awaiting here lets a test hold the + /// confirmation at a known probe while it binds. Nil in production. + func confirmNothingIsListening( + attempts: Int = 3, + interval: TimeInterval = 0.3, + afterEachProbe: (@Sendable (Int) async -> Void)? = nil + ) async -> Bool { + for attempt in 0.. 0 { + do { + try await Task.sleep(nanoseconds: UInt64(interval * 1_000_000_000)) + } catch { + return false // cancelled — assume the worst and do nothing + } + } + if shutdownRequested { return false } + let probe = SocketTransport.probeSocket(path: socketPath) + record(probe: probe) + await afterEachProbe?(attempt) + if probe == .connectable { + NSLog("[MCPProxy] Something answered on %@ — not taking the socket over", socketPath) + return false + } + } + return true + } + + /// What happened when we looked for a core to attach to. + /// + /// Three of these are emphatically NOT `noCore`, and conflating them is how + /// a tray ends up destroying a running core: + /// - `inFlight`: another task is already attaching. Spawning or going idle + /// would fight the attach under way. + /// - `unusable`: something is, or may be, on the other end of that socket. + /// Never unlink it and never spawn against it — that is a second writer on + /// one BBolt database. + enum AttachOutcome: Equatable { + case attached + case inFlight + case unusable(reason: String) + case noCore + } + + /// Attach to a core that is already up, taking the connection gate. + /// Used by the attach watcher, which is not already holding it. + private func attachIfCoreIsRunningGated() async -> AttachOutcome { + guard beginConnectionWork() else { return .inFlight } + defer { endConnectionWork() } + return await attachIfCoreIsRunningLocked() + } + + /// Attach to a core that is already up. Caller MUST hold the connection gate. + private func attachIfCoreIsRunningLocked() async -> AttachOutcome { + guard !superseded, !shutdownRequested else { return .noCore } + + switch await assessCore() { + case .alive: + guard !superseded, !shutdownRequested else { return .noCore } + await attachToExternalCore() + return .attached + case .listeningNoAnswer: + return .unusable(reason: "a core is listening but did not answer") + case .indeterminate(let why): + // Cannot tell whether a core is there. The caller must not treat + // this as an empty socket. + return .unusable(reason: why) + case .gone: + return .noCore + } + } + /// Idle mode (#410): no core, and we are not allowed to start one. Sit in the /// stopped state and poll for a core to attach to. private func awaitExternalCore() async { @@ -166,7 +564,10 @@ actor CoreProcessManager { while !Task.isCancelled { try? await Task.sleep(nanoseconds: UInt64(attachWatchInterval * 1_000_000_000)) guard !Task.isCancelled, let self else { return } - if await self.attachIfCoreIsRunning() { return } + guard await !self.isShuttingDown else { return } + // Keep polling while another task is mid-attach: if that attach + // fails, this watch is what picks the core up next time round. + if case .attached = await self.attachIfCoreIsRunningGated() { return } } } } @@ -183,26 +584,23 @@ actor CoreProcessManager { attachWatchTask = nil } - /// Probe an existing socket to see if a live core is behind it. - /// Returns true only if the core responds to an API call. - private func probeExternalCore() async -> Bool { - let probeClient = APIClient(socketPath: socketPath) - do { - let ready = try await probeClient.ready() - return ready - } catch { - return false - } - } - /// Gracefully shut down the core process and all connections. /// /// A core we merely ATTACHED to is left running: we disconnect from it and /// stop there. That rule is now enforced by the ownership check below rather /// than by the fact that `process` happens to be nil for an attached core. func shutdown() async { + // FIRST, before any suspension: stop work that is already in flight. + // A ladder asleep in its backoff, an attach watch mid-probe, or a + // reconnection waiting to retry will all otherwise wake up after this + // returns and launch a core the user just asked us to stop — leaving the + // tray showing "Stopped" while owning a running process. Never cleared: + // a manager is single-use after shutdown. + shutdownRequested = true + await transitionState(to: .shuttingDown) cancelAttachWatch() + cancelUnresponsiveCoreDeadline() // Disconnect SSE sseTask?.cancel() @@ -215,6 +613,8 @@ actor CoreProcessManager { } sseClient = nil apiClient = nil + probeClient = nil + consecutiveProbeFailures = 0 await MainActor.run { appState.apiClient = nil } let ownsCore = await MainActor.run { appState.ownership.shouldTerminateOnShutdown } @@ -227,25 +627,17 @@ actor CoreProcessManager { // Terminate the process if we own it if let process, process.isRunning { - // Send SIGTERM for graceful shutdown - process.interrupt() // sends SIGINT on macOS, which Go handles gracefully - - // Also send SIGTERM explicitly - kill(process.processIdentifier, SIGTERM) - - // Wait up to 10 seconds for graceful exit - let deadline = Date().addingTimeInterval(10.0) - while process.isRunning && Date() < deadline { - try? await Task.sleep(nanoseconds: 100_000_000) // 100ms - } + // SIGINT first: Go handles it gracefully. + process.interrupt() + } + await terminateManagedProcess(reason: "shutdown") - // Force kill if still running - if process.isRunning { - kill(process.processIdentifier, SIGKILL) - process.waitUntilExit() - } + // A launch may have been in flight when we set the flag: it checks the + // flag before running the binary, but if it got past that check we sweep + // it up here rather than leaving an orphan behind a "Stopped" icon. + if let stray = self.process, stray.isRunning { + await terminateManagedProcess(reason: "shutdown sweep") } - self.process = nil await transitionState(to: .idle) } @@ -261,12 +653,11 @@ actor CoreProcessManager { retryCount = 0 stderrBuffer = "" - // Clean up any existing process - if let process, process.isRunning { - kill(process.processIdentifier, SIGTERM) - process.waitUntilExit() - } - self.process = nil + // Clean up any existing process. Goes through the reaper so the exit + // callback for the core WE just killed is recognised as belonging to an + // abandoned launch — otherwise it lands as "the core crashed" and races + // the start() below for the right to launch a replacement. + await terminateManagedProcess(reason: "retry requested") let ownership = await MainActor.run { appState.ownership } let maySpawn = CoreLaunchPolicy.retryMaySpawn( @@ -288,6 +679,7 @@ actor CoreProcessManager { // healthy core was running. The watch loop exits on its own the moment // attachIfCoreIsRunning() reports success. attachWatchTask = nil + cancelUnresponsiveCoreDeadline() await MainActor.run { appState.ownership = .externalAttached // A core IS running, so we are not in the stopped state — even if we @@ -311,36 +703,238 @@ actor CoreProcessManager { // MARK: - Private: Launch and Connect - /// Full launch sequence: resolve binary, start process, wait for socket, connect. - private func launchAndConnect() async { - do { - await transitionState(to: .launching) + /// One launch attempt. THROWS on failure instead of publishing a terminal + /// error, so a caller running the retry ladder can decide whether there is + /// another rung. On success the ladder is cleared: a core that came back and + /// is running normally must not carry old strikes into its next crash. + private func launchAndConnectOnce() async throws { + await transitionState(to: .launching) - // Resolve the core binary - let binaryPath = try resolveBinary() - coreBinaryPath = binaryPath + // Resolve the core binary + let binaryPath = try resolveBinary() + coreBinaryPath = binaryPath - // Launch the process (core uses its own config API key) - try await launchCore(binaryPath: binaryPath) + // Launch the process (core uses its own config API key) + try await launchCore(binaryPath: binaryPath) - // Wait for the socket to become available - await transitionState(to: .waitingForCore) - try await waitForSocket(timeout: 60.0) + // Wait for the socket to become available + await transitionState(to: .waitingForCore) + try await waitForSocket(timeout: socketWaitTimeout) - // Connect API and SSE clients - try await connectToCore() + // Connect API and SSE clients + try await connectToCore() - await transitionState(to: .connected) - await refreshState() - startSSEStream() - startPeriodicRefresh() + await transitionState(to: .connected) + retryCount = 0 + await refreshState() + startSSEStream() + startPeriodicRefresh() + } - } catch let error as CoreError { - NSLog("[MCPProxy] launchAndConnect FAILED (CoreError): %@", error.userMessage) - await handleCoreError(error) - } catch { - NSLog("[MCPProxy] launchAndConnect FAILED: %@", error.localizedDescription) - await handleCoreError(.general(error.localizedDescription)) + /// Terminate and REAP a process we launched, and return only once it is + /// gone. Bumping the generation first means its termination callback — which + /// lands asynchronously — is recognised as belonging to a launch we have + /// already abandoned, so it cannot drive a relaunch. + private func terminateManagedProcess(reason: String) async { + guard let proc = process else { return } + launchGeneration += 1 + process = nil + + guard proc.isRunning else { return } + NSLog("[MCPProxy] Terminating PID %d (%@)", proc.processIdentifier, reason) + kill(proc.processIdentifier, SIGTERM) + + let deadline = Date().addingTimeInterval(5.0) + while proc.isRunning && Date() < deadline { + try? await Task.sleep(nanoseconds: 100_000_000) + } + if proc.isRunning { + NSLog("[MCPProxy] PID %d ignored SIGTERM — SIGKILL", proc.processIdentifier) + kill(proc.processIdentifier, SIGKILL) + proc.waitUntilExit() + } + } + + /// Start the clock on a core that holds the socket but will not answer. + /// + /// Refusing to fight it is right for the database and, on its own, unusable + /// for the person: `.waitingForCore` offers neither Stop nor Retry, so a + /// permanently silent listener would leave quitting the app as the only + /// move. The deadline turns it into an error the menu can act on — while the + /// attach watch keeps running underneath, so a core that does eventually + /// answer is still picked up without the user doing anything. + private func armUnresponsiveCoreDeadline(reason: String) { + unresponsiveDeadlineTask?.cancel() + let timeout = unresponsiveCoreTimeout + unresponsiveDeadlineTask = Task { [weak self] in + try? await Task.sleep(nanoseconds: UInt64(timeout * 1_000_000_000)) + guard !Task.isCancelled, let self else { return } + await self.reportUnresponsiveCore(reason: reason) + } + } + + private func cancelUnresponsiveCoreDeadline() { + unresponsiveDeadlineTask?.cancel() + unresponsiveDeadlineTask = nil + } + + /// Still waiting when the deadline expired. Either a core really is there and + /// silent — say so — or nothing ever was, and the wait is a deadlock we have + /// to break ourselves. + private func reportUnresponsiveCore(reason: String) async { + guard !superseded, !shutdownRequested else { return } + guard case .waitingForCore = await MainActor.run(body: { appState.coreState }) else { return } + + // This task IS the deadline; drop the handle before doing anything that + // might cancel it, or we would cancel ourselves mid-launch. + unresponsiveDeadlineTask = nil + + if socketEvidence.onlyEverRefused, await escalateToLaunchOverAStaleSocket(reason: reason) { + return + } + + NSLog("[MCPProxy] Giving up waiting for the core on %@ (%@)", socketPath, reason) + await transitionState(to: .error(.general( + "A core is holding \(socketPath) but is not responding (\(reason)). " + + "Quit that process, then use Retry." + ))) + } + + /// The deadline expired and not one connection has been accepted on this + /// socket in the whole window: a file is on disk and the process that made it + /// is gone. Launch a core rather than telling the user to quit a process that + /// does not exist — the error we would otherwise show is unactionable, and + /// Retry would repeat this same wait forever, because nothing else in the + /// system removes that file. + /// + /// Why this cannot resurrect the double-spawn hazard the tray-side unlink + /// had. It never unlinks anything: the dangerous step is done by the core we + /// spawn, which dials the socket immediately before removing it (a far + /// tighter window than anything the tray can manage) and fails with "socket + /// is in use by another process" when that dial succeeds — see + /// `cleanupStaleSocket` in internal/server/listener_unix.go. Everything in + /// front of that is unchanged: the connection gate is taken here, the + /// confirmation window runs again as a last look, and the launch goes through + /// `launchWithRetries` into the one choke point (`preflightLaunch`), behind + /// which bbolt's exclusive lock on the data directory is the final backstop. + /// And if a core has accepted so much as one connection during the window we + /// are not here at all: `onlyEverRefused` is false, and stays false for the + /// rest of the episode. + /// + /// Returns true when it has taken the episode over (nothing further to + /// report), false when the caller should fall through to the error. + private func escalateToLaunchOverAStaleSocket(reason: String) async -> Bool { + guard waitMaySpawn else { return false } + + // The attach watch takes the gate for each of its probes and holds it for + // microseconds. Wait for it rather than reporting an error we would have + // to retract — but do not wait indefinitely, or a wedged connection + // attempt would silence the deadline altogether. + guard await claimConnectionWork(waitingUpTo: 1.0) else { + NSLog("[MCPProxy] Deadline expired while a connection attempt is in flight — waiting again") + armUnresponsiveCoreDeadline(reason: reason) + return true + } + defer { endConnectionWork() } + + // A last look, with the gate held so no probe of ours is in flight: a + // core that is merely slow accepts at least one of these, and that answer + // ends the escalation. Same window the spawn path uses. + guard await confirmNothingIsListening() else { return false } + guard !superseded, !shutdownRequested else { return false } + + NSLog("[MCPProxy] Nothing has ever answered on %@ (%@) — treating it as a socket file " + + "left behind by a dead core and launching", socketPath, reason) + + // Called from the deadline task, never from inside the watch itself, so + // cancelling here cannot cancel an attach in progress. + cancelAttachWatch() + await MainActor.run { + appState.isStopped = false + appState.ownership = .trayManaged + } + await launchWithRetries() + return true + } + + /// Take the connection gate, waiting up to `seconds` for a short-lived holder + /// to finish. Polls rather than queues: the gate is a flag, not a lock, and + /// the only thing that ever holds it briefly is a probe. + private func claimConnectionWork(waitingUpTo seconds: TimeInterval) async -> Bool { + let deadline = Date().addingTimeInterval(seconds) + while true { + if beginConnectionWork() { return true } + guard Date() < deadline, !shutdownRequested else { return false } + do { + try await Task.sleep(nanoseconds: 50_000_000) + } catch { + return false + } + } + } + + /// Launch the core, climbing the retry ladder until it comes up or the + /// rungs run out. Caller MUST hold the connection gate. + /// + /// The ladder lives with the gate holder, not in the process-termination + /// callback. A core that dies during `launchAndConnectOnce()` fires that + /// callback while this task still holds the gate, so the callback stands + /// down — as it must, or it would launch a second core. If the ladder lived + /// there, `maxRetries` would silently collapse to a single attempt. + private func launchWithRetries() async { + while true { + guard !shutdownRequested else { return } + + // Never ladder INTO a socket somebody owns. A core that came up sick + // — listening but not answering — makes connectToCore() fail, and + // without this the reconnection path would happily relaunch on top + // of it. The rung check is here as well as in launchCore() because + // this one can do something useful about it: wait. + if await assessCore().somethingIsListening { + await waitForTheCoreThatIsAlreadyThere( + reason: "a core is already listening on the socket", + // We are inside the launch path, so spawning is permitted — + // but this wait can never escalate anyway: a probe just + // accepted, which is the one thing that rules it out. + maySpawn: true + ) + return + } + + do { + try await launchAndConnectOnce() + return // success; launchAndConnectOnce cleared the ladder + } catch { + let coreError = (error as? CoreError) ?? .general(error.localizedDescription) + + // A rung that gave up may have left a process RUNNING: a core + // that hung without ever becoming ready is still a core. Reap it + // before the next rung, or the ladder is a core factory. + await terminateManagedProcess(reason: "launch attempt failed") + + guard !shutdownRequested else { return } + guard coreError.isRetryable else { + // A config error or a port conflict will not fix itself. + await handleCoreError(coreError) + return + } + guard retryCount < maxRetries else { + await handleCoreError(.maxRetriesExceeded) + return + } + + retryCount += 1 + NSLog("[MCPProxy] Core launch failed (%@) — retry %d/%d", + coreError.userMessage, retryCount, maxRetries) + await transitionState(to: .reconnecting(attempt: retryCount)) + + let delay = reconnectionPolicy.delay(forAttempt: retryCount) + do { + try await Task.sleep(nanoseconds: UInt64(delay * 1_000_000_000)) + } catch { + return // cancelled + } + } } } @@ -431,8 +1025,44 @@ actor CoreProcessManager { // MARK: - Private: Process Launch + /// THE choke point's preflight. Every path that wants a core arrives at + /// `launchCore` — start(), each rung of the ladder, retry(), the + /// process-exit handler — so the invariant "never two cores on one data + /// directory" is enforced HERE, once, instead of being a property of six + /// call paths all staying disciplined. Three rounds of review have shown + /// that discipline is not a thing to rely on. + /// + /// Every check is synchronous, so no other task on this actor can interleave + /// between them and `proc.run()`. That is the whole of what actor isolation + /// buys, and it is worth being honest about the rest: an EXTERNAL process + /// can still bind the socket in the milliseconds after the last probe here. + /// The backstops for that race are core-side, not tray-side — bbolt takes an + /// exclusive lock on the data directory and the loser exits with code 3, and + /// a core whose socket is already in use refuses to start at all. + /// + /// Throws rather than returning a Bool so the reason reaches the caller's + /// error handling (and the user) unchanged. + func preflightLaunch() throws { + guard connectionWorkInFlight else { + throw CoreError.general("internal error: tried to launch a core without the connection gate") + } + guard !shutdownRequested else { + throw CoreError.general("shutting down") + } + if let existing = process, existing.isRunning { + throw CoreError.general( + "internal error: tried to launch a core while PID \(existing.processIdentifier) is still running" + ) + } + if SocketTransport.probeSocket(path: socketPath) == .connectable { + throw CoreError.general("A core is already running on \(socketPath)") + } + } + /// Launch the mcpproxy core process. private func launchCore(binaryPath: String) async throws { + try preflightLaunch() + let proc = Process() proc.executableURL = URL(fileURLWithPath: binaryPath) proc.arguments = ["serve"] @@ -477,11 +1107,15 @@ actor CoreProcessManager { // Set up process group for clean termination proc.qualityOfService = .userInitiated - // Handle unexpected termination + // Handle unexpected termination. The generation stamps WHICH launch this + // callback belongs to: a process we gave up on and reaped must not, when + // its callback finally lands, drive a fresh relaunch. + launchGeneration += 1 + let generation = launchGeneration proc.terminationHandler = { [weak self] terminatedProcess in let status = terminatedProcess.terminationStatus Task { - await self?.handleProcessExit(status: status) + await self?.handleProcessExit(status: status, generation: generation) } } @@ -491,6 +1125,7 @@ actor CoreProcessManager { throw CoreError.general("Failed to launch core: \(error.localizedDescription)") } + coresLaunched += 1 process = proc } @@ -585,6 +1220,10 @@ actor CoreProcessManager { } apiClient = client + // Separate client for liveness probes: same socket, much shorter + // timeout, and never shared with the UI so a probe can never be queued + // behind a slow user-facing request. + consecutiveProbeFailures = 0 await MainActor.run { appState.apiClient = client } // Create SSE client — uses TCP (not socket) so needs the API key @@ -725,25 +1364,140 @@ actor CoreProcessManager { startSSEStream() } else { // Socket gone; core likely crashed - await transitionState(to: .reconnecting(attempt: 1)) - await attemptReconnection() + await handleCoreLoss() } } // MARK: - Private: State Refresh - /// Start a periodic refresh task that polls servers every 30 seconds. + /// Start a periodic refresh task that polls the core every `refreshInterval`. + /// + /// The tick is also the tray's liveness detector for a core it does not own + /// (GH #926). See `coreIsAlive()`. private func startPeriodicRefresh() { refreshTask?.cancel() + let interval = refreshInterval refreshTask = Task { [weak self] in while !Task.isCancelled { - try? await Task.sleep(nanoseconds: 30_000_000_000) // 30s - guard !Task.isCancelled else { break } - await self?.refreshState() + try? await Task.sleep(nanoseconds: UInt64(interval * 1_000_000_000)) + guard !Task.isCancelled, let self else { break } + // Stop refreshing the moment the core is gone: handleCoreLoss() + // takes over (and restarts this task if it reconnects). + guard await self.coreIsAlive() else { break } + await self.refreshState() + } + } + } + + /// Liveness probe for the periodic tick — the ONLY thing that notices a + /// dead core we did not spawn (GH #926). + /// + /// A core the tray launched is watched by `Process.terminationHandler`. An + /// ATTACHED core has no Process and no PID, and the other two candidate + /// detectors are inert: `SSEClient` retries forever without ever finishing + /// its stream (so `handleSSEDisconnect()` never runs), and every refresh + /// helper swallows its errors by design. The result was a tray that showed + /// a healthy icon indefinitely after the core died. + /// + /// The probe is the same one `attachIfCoreIsRunning()` uses to decide a core + /// is there in the first place — socket, then a real API call — so "we think + /// we are connected" is checked against exactly the evidence that made us + /// connect. Returns true when the core answered (or when we are not in a + /// state where the question applies); returns false after handing off to + /// reconnection. + func coreIsAlive() async -> Bool { + // Only meaningful while we believe we are connected. Any other state is + // already being driven by launch/reconnect/shutdown logic. + guard case .connected = await MainActor.run(body: { appState.coreState }) else { return true } + + switch SocketTransport.probeSocket(path: socketPath) { + case .absent: + // No socket FILE. A running core always owns its socket file, so + // this one is unambiguous — act on it immediately. This is the + // reporter's case (`kill -9` plus a cleaned-up socket). + NSLog("[MCPProxy] Core socket %@ is gone — reconnecting", socketPath) + consecutiveProbeFailures = 0 + await handleCoreLoss() + return false + + case .localFailure(let code): + // Descriptor exhaustion, out of memory, permissions. Our problem, + // not the core's — it is not evidence either way, so it must not + // even count as a strike. + NSLog("[MCPProxy] Liveness probe could not run (errno %d) — not treating as core loss", code) + return true + + case .refused: + // AMBIGUOUS: a dead process leaves a stale socket file that refuses + // connections, and so does a live core whose listen backlog is + // momentarily full. Same evidence, opposite conclusions — so it gets + // the same tolerance as a missed readiness check rather than an + // immediate verdict. + return await recordProbeFailure(reason: "socket refused connections") + + case .connectable: + // Something is listening. Whether it is HEALTHY is a softer + // question: a saturated core can miss a readiness check and be + // perfectly fine a second later. Condemning it on one sample would + // put "reconnecting" in the menu bar every time the core got busy. + if await probeIsReady() { + consecutiveProbeFailures = 0 + return true } + return await recordProbeFailure(reason: "no answer from /ready") } } + /// Count one inconclusive probe. Returns true while the budget holds (treat + /// the core as alive), false once it is spent — having handed off to + /// reconnection. + private func recordProbeFailure(reason: String) async -> Bool { + consecutiveProbeFailures += 1 + guard consecutiveProbeFailures >= probeFailureBudget else { + NSLog("[MCPProxy] Core liveness check missed (%d/%d): %@", + consecutiveProbeFailures, probeFailureBudget, reason) + return true + } + + NSLog("[MCPProxy] Core failed %d consecutive liveness checks (%@) — reconnecting", + consecutiveProbeFailures, reason) + consecutiveProbeFailures = 0 + await handleCoreLoss() + return false + } + + /// One readiness call, bounded by `probeTimeout` rather than the API + /// client's generous request timeout. + private func probeIsReady() async -> Bool { + // ONE probe client for the manager's whole life: the same socket and the + // same short timeout are wanted before the attach (is a core there?) and + // after it (is it still there?). Creating one per probe would also mint a + // routing token per probe, and the attach watcher probes every 2s. + let client: APIClient + if let existing = probeClient { + client = existing + } else { + client = APIClient(socketPath: socketPath, requestTimeout: probeTimeout) + probeClient = client + } + return (try? await client.ready()) == true + } + + /// The core we were connected to is gone. Reuse the reconnection path, which + /// re-attaches to a core that comes back and refuses to spawn a replacement + /// for one we never owned. + func handleCoreLoss() async { + // Read state first (this suspends), THEN claim the gate — the claim must + // be the last thing before the work, with no suspension between claim + // and use. + guard case .connected = await MainActor.run(body: { appState.coreState }) else { return } + guard beginConnectionWork() else { return } + defer { endConnectionWork() } + + await transitionState(to: .reconnecting(attempt: 1)) + await attemptReconnection() + } + /// Fetch full state from the core and update appState. /// Spec 048: dropped the per-tick refreshServers() call. The server list /// is now SSE-driven (spec 047 servers.changed payload). MCPProxyApp @@ -879,7 +1633,16 @@ actor CoreProcessManager { // MARK: - Private: Process Exit Handling /// Handle the core process exiting. - private func handleProcessExit(status: Int32) async { + func handleProcessExit(status: Int32, generation: Int? = nil) async { + // A callback from a launch we already abandoned says nothing about the + // core we have now. Acting on it is how a reaped attempt turns into an + // extra spawn. + if let generation, generation != launchGeneration { + NSLog("[MCPProxy] Ignoring exit of superseded launch %d (current %d)", + generation, launchGeneration) + return + } + let stderr = stderrBuffer // If stopped by user, don't retry — this is intentional @@ -903,8 +1666,15 @@ actor CoreProcessManager { await notificationService.sendCoreError(error: error) if error.isRetryable && retryCount < maxRetries { - retryCount += 1 - await transitionState(to: .reconnecting(attempt: retryCount)) + // Honour the same gate as every other reconnection entry point. If + // one is already running it is ALREADY relaunching this core, and it + // owns the whole retry ladder — standing down here loses nothing. + guard beginConnectionWork() else { + NSLog("[MCPProxy] handleProcessExit: a reconnection is already in flight") + return + } + defer { endConnectionWork() } + await transitionState(to: .reconnecting(attempt: retryCount + 1)) await attemptReconnection() } else { await transitionState(to: .error(error)) @@ -914,43 +1684,60 @@ actor CoreProcessManager { // MARK: - Private: Reconnection /// Attempt to reconnect after a failure. + /// Run the retry ladder to completion. Caller MUST hold the connection gate. + /// + /// The ladder lives HERE, in the task that holds the gate, rather than in + /// the process-termination callback. It used to depend on each relaunched + /// core's exit callback starting the next attempt — but that callback now + /// (correctly) stands down when a reconnection is already running, which + /// silently reduced `maxRetries` to a single relaunch for a core that + /// crashes instantly. Owning the loop here keeps both properties: exactly + /// one launch at a time, AND every rung of the ladder. private func attemptReconnection() async { - let delay = reconnectionPolicy.delay(forAttempt: retryCount) - let delayNs = UInt64(delay * 1_000_000_000) - - do { - try await Task.sleep(nanoseconds: delayNs) - } catch { - return // Cancelled - } - - // If an external core came up, attach to it - if SocketTransport.isSocketAvailable(path: socketPath) { + // The core we were connected to answered plenty of probes; none of that + // is evidence about the socket we are about to reason over now. + beginSocketEvidence() + while true { + guard !shutdownRequested else { return } + let delay = reconnectionPolicy.delay(forAttempt: max(retryCount, 1)) do { - try await connectToCore() - await transitionState(to: .connected) - retryCount = 0 - await refreshState() - startSSEStream() - startPeriodicRefresh() - return + try await Task.sleep(nanoseconds: UInt64(delay * 1_000_000_000)) } catch { - // Fall through to relaunch + return // Cancelled } - } + guard !shutdownRequested else { return } - // If we own the core, relaunch it - let ownership = await MainActor.run { appState.ownership } - if ownership == .trayManaged { - if retryCount < maxRetries { - await launchAndConnect() - } else { - await transitionState(to: .error(.maxRetriesExceeded)) - await notificationService.sendCoreError(error: .maxRetriesExceeded) + // If a core is up on the socket — ours or someone else's — attach to + // it rather than starting another. + if SocketTransport.isSocketAvailable(path: socketPath) { + do { + try await connectToCore() + await transitionState(to: .connected) + retryCount = 0 + await refreshState() + startSSEStream() + startPeriodicRefresh() + return + } catch { + // Fall through to relaunch + } } - } else { - // External core is gone and we don't own it - await transitionState(to: .error(.general("External core process is no longer available"))) + + let ownership = await MainActor.run { appState.ownership } + guard ownership == .trayManaged else { + // External core is gone and we don't own it. Say so — and keep + // watching the socket, so a core that comes back (launchd, brew + // services, the user re-running `mcpproxy serve`) is picked up + // without needing a menu click. Same cheap poll idle mode uses. + await transitionState( + to: .error(.general("External core process is no longer available")) + ) + startAttachWatch() + return + } + + await launchWithRetries() + return } } diff --git a/native/macos/MCPProxy/MCPProxy/Core/SocketTransport.swift b/native/macos/MCPProxy/MCPProxy/Core/SocketTransport.swift index c534dc0f..c35f3b91 100644 --- a/native/macos/MCPProxy/MCPProxy/Core/SocketTransport.swift +++ b/native/macos/MCPProxy/MCPProxy/Core/SocketTransport.swift @@ -20,11 +20,68 @@ final class SocketURLProtocol: URLProtocol { NSHomeDirectory() + "/.mcpproxy/mcpproxy.sock" }() - /// Allow override for testing. - static var overrideSocketPath: String? + /// Header carrying an opaque route token identifying the socket a + /// particular session must use. + /// + /// The route travels PER SESSION, not in a global mutable path. It used to + /// live in a mutable static, which meant creating any second client — + /// another `CoreProcessManager`, a probe, a concurrent test — silently + /// redirected every existing client to the newest path. That is a + /// liveness-detector hazard, not just a test smell: a client could report a + /// different core's health as its own. + /// + /// The value is a TOKEN, not the path. `URLSessionConfiguration` + /// `httpAdditionalHeaders` are attached to every request the session makes, + /// including ones this protocol declines to intercept and ones following a + /// redirect — so whatever goes in here must be safe to disclose. A UUID is; + /// `/Users//.mcpproxy/mcpproxy.sock` is not. + static let routeHeader = "X-MCPProxy-Route" + + /// token -> socket path. Written once per session at creation and read on + /// every request. Not an alias for "the current path": each entry belongs to + /// exactly one session, which is the whole point. + private static let routes = RouteTable() + + /// Register a socket path and return the token that routes to it. + static func makeRoute(to socketPath: String) -> String { + routes.add(socketPath) + } + + /// Resolve a route token back to its socket path. + static func routes(for token: String) -> String? { + routes.path(for: token) + } + + /// Socket path this request must be routed over, or nil when the request + /// carries no route (a client constructed without an explicit path). + static func routedSocketPath(for request: URLRequest) -> String? { + guard let token = request.value(forHTTPHeaderField: routeHeader) else { return nil } + return routes.path(for: token) + } + + /// Socket path this request must use, falling back to the default. + static func effectiveSocketPath(for request: URLRequest) -> String { + routedSocketPath(for: request) ?? socketPath + } + + /// Thread-safe token table. + final class RouteTable: @unchecked Sendable { + private let lock = NSLock() + private var paths: [String: String] = [:] + + func add(_ path: String) -> String { + let token = UUID().uuidString + lock.lock() + paths[token] = path + lock.unlock() + return token + } - private static var effectiveSocketPath: String { - overrideSocketPath ?? socketPath + func path(for token: String) -> String? { + lock.lock() + defer { lock.unlock() } + return paths[token] + } } /// Active read task, retained for cancellation. @@ -43,8 +100,17 @@ final class SocketURLProtocol: URLProtocol { (host == "localhost" || host == "127.0.0.1") else { return false } - // Only intercept if the socket exists. - return FileManager.default.fileExists(atPath: effectiveSocketPath) + // A request that carries a route is PINNED to that socket: intercept it + // even when the socket is missing, and let it fail. Falling back to TCP + // there would silently send a client that was told "talk to this core" + // to whatever happens to be listening on 127.0.0.1:8080 — a different + // core's health reported as this one's, which is precisely the failure + // the routing exists to prevent. + if routedSocketPath(for: request) != nil { return true } + + // Unrouted request: legacy behaviour, intercept only if the default + // socket exists, otherwise let it go out over TCP. + return FileManager.default.fileExists(atPath: socketPath) } override class func canonicalRequest(for request: URLRequest) -> URLRequest { @@ -64,7 +130,7 @@ final class SocketURLProtocol: URLProtocol { // Build sockaddr_un var addr = sockaddr_un() addr.sun_family = sa_family_t(AF_UNIX) - let path = Self.effectiveSocketPath + let path = Self.effectiveSocketPath(for: request) let pathBytes = path.utf8CString guard pathBytes.count <= MemoryLayout.size(ofValue: addr.sun_path) else { Darwin.close(fd) @@ -183,6 +249,8 @@ final class SocketURLProtocol: URLProtocol { for (key, value) in allHeaders { let lowerKey = key.lowercased() if lowerKey == "host" { continue } // already added + // Transport routing hint — never goes on the wire. + if lowerKey == Self.routeHeader.lowercased() { continue } if lowerKey == "content-length" { hasContentLength = true } lines.append("\(key): \(value)") } @@ -480,14 +548,22 @@ enum SocketTransport { /// Create a `URLSession` configured to route traffic through the mcpproxy Unix socket. /// Falls back to standard networking if the socket is not available. - static func makeURLSession(socketPath: String? = nil) -> URLSession { - if let path = socketPath { - SocketURLProtocol.overrideSocketPath = path - } - + /// + /// - Parameters: + /// - socketPath: socket this session must use. Carried per-request in a + /// header rather than a process-global, so two clients can talk to two + /// different cores without redirecting each other. + /// - timeout: per-request timeout. The default is generous because the + /// core can be slow under load; liveness probes pass something short. + static func makeURLSession(socketPath: String? = nil, timeout: TimeInterval = 30) -> URLSession { let config = URLSessionConfiguration.default config.protocolClasses = [SocketURLProtocol.self] - config.timeoutIntervalForRequest = 30 + if let socketPath { + config.httpAdditionalHeaders = [ + SocketURLProtocol.routeHeader: SocketURLProtocol.makeRoute(to: socketPath) + ] + } + config.timeoutIntervalForRequest = timeout config.timeoutIntervalForResource = 300 config.httpShouldSetCookies = false config.httpCookieAcceptPolicy = .never @@ -495,27 +571,49 @@ enum SocketTransport { return URLSession(configuration: config) } - /// Create a standard TCP-based `URLSession` (no socket override). - static func makeTCPSession() -> URLSession { + /// Create a standard TCP-based `URLSession` (never routed over the socket). + static func makeTCPSession(timeout: TimeInterval = 30) -> URLSession { let config = URLSessionConfiguration.default - config.timeoutIntervalForRequest = 30 + config.timeoutIntervalForRequest = timeout config.timeoutIntervalForResource = 300 config.httpShouldSetCookies = false config.httpCookieAcceptPolicy = .never return URLSession(configuration: config) } + /// Why a socket probe failed. "Not connectable" is NOT the same as "the core + /// is dead", and the difference decides whether the tray may act on it. + enum SocketProbe: Equatable { + /// A listener accepted (or is accepting) the connection. + case connectable + /// No socket file at all. A running core always owns its socket file, + /// so this one is unambiguous. + case absent + /// The file is there but the connection was refused. Ambiguous: a dead + /// process leaves a stale file behind, AND a live core with a full + /// listen backlog refuses connections in exactly the same way. + case refused + /// The failure was on OUR side — descriptor exhaustion, out of memory, + /// permissions. Says nothing whatsoever about the core. + case localFailure(Int32) + } + /// Check whether the mcpproxy Unix socket file exists and is connectable. static func isSocketAvailable(path: String? = nil) -> Bool { + probeSocket(path: path) == .connectable + } + + /// Probe the socket and classify the outcome. + static func probeSocket(path: String? = nil) -> SocketProbe { let socketPath = path ?? SocketURLProtocol.socketPath guard FileManager.default.fileExists(atPath: socketPath) else { - return false + return .absent } // Attempt a quick connect to verify the socket is alive let fd = Darwin.socket(AF_UNIX, SOCK_STREAM, 0) - guard fd >= 0 else { return false } + guard fd >= 0 else { return .localFailure(errno) } defer { Darwin.close(fd) } // Set non-blocking for a quick probe @@ -540,6 +638,22 @@ enum SocketTransport { } // Non-blocking connect returns 0 on immediate success or EINPROGRESS - return result == 0 || errno == EINPROGRESS + if result == 0 { return .connectable } + let connectErrno = errno + if connectErrno == EINPROGRESS { return .connectable } + + switch connectErrno { + case ENOENT: + // Unlinked between the stat above and the connect. + return .absent + case EMFILE, ENFILE, ENOMEM, ENOBUFS, EACCES, EPERM: + // Our process ran out of something, or cannot reach the socket. + // Not evidence about the core. + return .localFailure(connectErrno) + default: + // ECONNREFUSED and friends: a dead process's stale socket looks + // exactly like a live core whose listen queue is full. + return .refused + } } } diff --git a/native/macos/MCPProxy/MCPProxy/Services/NotificationService.swift b/native/macos/MCPProxy/MCPProxy/Services/NotificationService.swift index 65a93d34..9d384eec 100644 --- a/native/macos/MCPProxy/MCPProxy/Services/NotificationService.swift +++ b/native/macos/MCPProxy/MCPProxy/Services/NotificationService.swift @@ -95,13 +95,34 @@ actor NotificationService { private var settleGate = ConnectionSettleGate() /// The shared notification center. - private let center = UNUserNotificationCenter.current() + /// + /// Resolved lazily on first use rather than at init: `current()` requires a + /// real application bundle and raises `NSInternalInconsistencyException` + /// otherwise, which made merely CONSTRUCTING this service — and therefore + /// anything that depends on it, such as `CoreProcessManager` — impossible to + /// unit-test. Delivery paths are unchanged; nothing in the app touches the + /// center before `setup()`. + private lazy var center = UNUserNotificationCenter.current() + + /// Whether this service may actually talk to `UNUserNotificationCenter`. + /// + /// `current()` requires a real application bundle and raises + /// `NSInternalInconsistencyException` otherwise, which kills a unit-test + /// process outright. Tests construct the service with delivery off so that + /// code paths which happen to send a notification (a core-exit handler, say) + /// are exercisable. Always on in the app. + private let deliveryEnabled: Bool + + init(deliveryEnabled: Bool = true) { + self.deliveryEnabled = deliveryEnabled + } // MARK: - Setup /// Request notification permission and register action categories. /// Call this once during app launch. func setup() async { + guard deliveryEnabled else { return } // Request authorization do { let granted = try await center.requestAuthorization(options: [.alert, .sound, .badge]) @@ -273,6 +294,7 @@ actor NotificationService { /// Schedule a notification for immediate delivery. private func deliver(content: UNMutableNotificationContent, identifier: String) async { + guard deliveryEnabled else { return } let trigger = UNTimeIntervalNotificationTrigger(timeInterval: 0.1, repeats: false) let request = UNNotificationRequest( identifier: identifier, diff --git a/native/macos/MCPProxy/MCPProxyTests/CoreLivenessTests.swift b/native/macos/MCPProxy/MCPProxyTests/CoreLivenessTests.swift new file mode 100644 index 00000000..2d6d6b57 --- /dev/null +++ b/native/macos/MCPProxy/MCPProxyTests/CoreLivenessTests.swift @@ -0,0 +1,1202 @@ +// CoreLivenessTests.swift +// MCPProxyTests +// +// GH #926 — "menu bar icon never detects externally managed core stopping". +// +// When the tray ATTACHES to a core it did not spawn, there is no `Process` and +// therefore no `terminationHandler` — which was the only thing in the app that +// ever noticed a dead core. The SSE client retries forever without ever +// finishing its stream, and every periodic refresh swallows its errors, so +// `coreState` could never leave `.connected`: the menu bar icon kept claiming +// the core was healthy indefinitely. +// +// These tests drive the real attach path against a stub core on a Unix socket, +// then take that core away — or, harder and more realistic, leave it listening +// while it stops answering. + +import XCTest +@testable import MCPProxy + +final class CoreLivenessTests: XCTestCase { + + private var stub: UnixSocketHTTPStub? + private var manager: CoreProcessManager? + + /// Async teardown so shutdown actually completes before the next test runs: + /// a fire-and-forget `Task { await shutdown() }` would race the next test's + /// stub over sockets and background tasks. + override func tearDown() async throws { + if let manager { + await manager.shutdown() + } + manager = nil + stub?.stop() + stub = nil + try await super.tearDown() + } + + /// Bring up a stub core and attach a manager to it. + /// `maySpawn: false` guarantees the test can never launch a real binary. + @discardableResult + private func attachToStubCore( + refreshInterval: TimeInterval = 60, + probeTimeout: TimeInterval = 5.0, + baseDelay: TimeInterval = 0.05, + ready: ReadyBehaviour = ReadyBehaviour() + ) async throws -> (CoreProcessManager, AppState, UnixSocketHTTPStub) { + let stub = UnixSocketHTTPStub.healthyCore(ready: ready) + try stub.start() + self.stub = stub + + let appState = await MainActor.run { AppState() } + let manager = CoreProcessManager( + appState: appState, + // Notification delivery needs a real app bundle; keep it off so the + // core-exit path is exercisable without killing the test process. + notificationService: NotificationService(deliveryEnabled: false), + // Keep reconnect backoff short so the test is fast; the production + // policy only changes how long the user waits, not the outcome. + reconnectionPolicy: ReconnectionPolicy( + baseDelay: baseDelay, maxDelay: baseDelay * 2, maxAttempts: 3, jitterFactor: 0.0 + ), + socketPath: stub.path, + refreshInterval: refreshInterval, + probeTimeout: probeTimeout + ) + self.manager = manager + await manager.start(maySpawn: false) + return (manager, appState, stub) + } + + private func coreState(_ appState: AppState) async -> CoreState { + await MainActor.run { appState.coreState } + } + + /// Run a shell pipeline and read back an integer (process counting). + private func shellCount(_ command: String) throws -> Int { + let proc = Process() + proc.executableURL = URL(fileURLWithPath: "/bin/sh") + proc.arguments = ["-c", command] + let pipe = Pipe() + proc.standardOutput = pipe + proc.standardError = FileHandle.nullDevice + try proc.run() + let data = pipe.fileHandleForReading.readDataToEndOfFile() + proc.waitUntilExit() + let text = String(data: data, encoding: .utf8)? + .trimmingCharacters(in: .whitespacesAndNewlines) ?? "0" + return Int(text) ?? 0 + } + + /// Poll until `predicate` holds or the deadline passes. Returns the last state. + private func waitForState( + _ appState: AppState, + timeout: TimeInterval, + until predicate: @escaping (CoreState) -> Bool + ) async -> CoreState { + let deadline = Date().addingTimeInterval(timeout) + var last = await coreState(appState) + while Date() < deadline { + if predicate(last) { return last } + try? await Task.sleep(nanoseconds: 50_000_000) // 50ms + last = await coreState(appState) + } + return last + } + + // MARK: - The seam itself + + /// Baseline: with an injected socket path the manager attaches to a core it + /// did not spawn, exactly as it does against `~/.mcpproxy/mcpproxy.sock`. + /// Everything below depends on this being a faithful attach. + func testAttachesToAnExternalCoreOverAnInjectedSocket() async throws { + let (_, appState, _) = try await attachToStubCore() + + let state = await coreState(appState) + XCTAssertEqual(state, .connected, + "the manager must attach to a live core on the injected socket") + let ownership = await MainActor.run { appState.ownership } + XCTAssertEqual(ownership, .externalAttached, + "a core we did not spawn must be marked external") + let isStopped = await MainActor.run { appState.isStopped } + XCTAssertFalse(isStopped) + } + + /// Two managers on two sockets must not redirect each other. The socket path + /// used to live in a mutable static, so constructing the second client + /// silently repointed the first — a liveness detector reporting another + /// core's health as its own. + func testTwoManagersOnDifferentSocketsDoNotRedirectEachOther() async throws { + let (_, firstState, firstStub) = try await attachToStubCore() + + let secondStub = UnixSocketHTTPStub.healthyCore() + try secondStub.start() + defer { secondStub.stop() } + + let secondState = await MainActor.run { AppState() } + let secondManager = CoreProcessManager( + appState: secondState, + notificationService: NotificationService(deliveryEnabled: false), + socketPath: secondStub.path, + refreshInterval: 60 + ) + await secondManager.start(maySpawn: false) + + let connected = await coreState(secondState) + XCTAssertEqual(connected, .connected) + XCTAssertGreaterThan(secondStub.requestCount(path: "/api/v1/info"), 0, + "the second manager must talk to its OWN socket") + + // The first manager's client must still reach the first stub, not the + // one created later. + let firstProbesBefore = firstStub.requestCount(path: "/ready") + let stillAlive = await self.manager?.coreIsAlive() + XCTAssertEqual(stillAlive, true) + XCTAssertEqual(firstStub.requestCount(path: "/ready"), firstProbesBefore + 1, + "the first manager's probe must still land on the first socket") + let firstStillConnected = await coreState(firstState) + XCTAssertEqual(firstStillConnected, .connected) + + await secondManager.shutdown() + } + + // MARK: - GH #926: the core disappears + + /// The regression under test: kill an externally-attached core and the tray + /// must stop reporting `.connected`. Before the fix this assertion failed — + /// the state stayed `.connected` for as long as anyone cared to watch + /// (the reporter measured 5.5 minutes). + func testDeadExternalCoreLeavesConnectedState() async throws { + let (_, appState, stub) = try await attachToStubCore(refreshInterval: 0.2) + let attached = await coreState(appState) + XCTAssertEqual(attached, .connected, "precondition: attached") + + // The core dies: socket closed and removed, as after `kill -9`. + stub.stop() + + let state = await waitForState(appState, timeout: 5.0) { $0 != .connected } + XCTAssertNotEqual(state, .connected, + "the tray must notice an externally-managed core that stopped") + + // And it must land somewhere the tray icon can render as "not fine" — + // never silently back in `.connected`. + let settled = await waitForState(appState, timeout: 5.0) { + if case .error = $0 { return true } + return false + } + XCTAssertEqual(settled, .error(.general("External core process is no longer available")), + "a core we do not own and cannot find must surface as an error") + } + + /// The tray must not spawn a replacement for a core it never owned, and must + /// pick the core back up by itself when it returns (launchd/brew restart). + func testReattachesWhenTheExternalCoreComesBack() async throws { + let (_, appState, stub) = try await attachToStubCore(refreshInterval: 0.2) + + stub.stop() + let lost = await waitForState(appState, timeout: 5.0) { + if case .error = $0 { return true } + return false + } + guard case .error = lost else { + return XCTFail("precondition: the tray must first notice the core is gone, got \(lost)") + } + + let ownership = await MainActor.run { appState.ownership } + XCTAssertEqual(ownership, .externalAttached, + "losing an external core must not make the tray claim ownership") + + // The core comes back on the same socket. + let revived = UnixSocketHTTPStub.healthyCore(at: stub.path) + try revived.start() + self.stub = revived + + let state = await waitForState(appState, timeout: 10.0) { $0 == .connected } + XCTAssertEqual(state, .connected, "the tray must re-attach when the core returns") + } + + // MARK: - GH #926: the core stops ANSWERING (socket still up) + + /// The half the probe actually exists for. The socket is connectable + /// throughout, so a socket-existence-only implementation sees nothing wrong + /// forever. Only a real `/ready` call notices. + /// + /// It also pins the hysteresis: ONE missed readiness check is not evidence + /// of a dead core — a busy core can miss a sample — so the tray must not + /// flip the icon on it. Two consecutive misses are. + func testOneReadinessMissIsToleratedAndTwoAreNot() async throws { + let ready = ReadyBehaviour() + let (manager, appState, stub) = try await attachToStubCore(ready: ready) + + ready.current = .json(#"{"success":false,"error":"busy"}"#, status: 503) + let probesBefore = stub.requestCount(path: "/ready") + + let afterFirstMiss = await manager.coreIsAlive() + XCTAssertTrue(afterFirstMiss, + "a single readiness miss must not condemn a core whose socket is still up") + let stillConnected = await coreState(appState) + XCTAssertEqual(stillConnected, .connected) + XCTAssertEqual(stub.requestCount(path: "/ready"), probesBefore + 1, + "the probe must actually call /ready — the socket alone proves nothing") + + let afterSecondMiss = await manager.coreIsAlive() + XCTAssertFalse(afterSecondMiss, + "two consecutive readiness misses must be treated as core loss") + let lost = await coreState(appState) + XCTAssertNotEqual(lost, .connected, + "a listening-but-unresponsive core must leave the connected state") + } + + /// A recovered core clears the strike count: misses must be CONSECUTIVE, + /// otherwise a healthy core that blips once an hour eventually gets killed. + func testAReadinessRecoveryResetsTheFailureCount() async throws { + let ready = ReadyBehaviour() + let (manager, appState, _) = try await attachToStubCore(ready: ready) + + ready.current = .json(#"{"success":false}"#, status: 503) + _ = await manager.coreIsAlive() // strike one + ready.current = .json(#"{"success":true}"#) + let recovered = await manager.coreIsAlive() + XCTAssertTrue(recovered, "the core answered again") + + ready.current = .json(#"{"success":false}"#, status: 503) + let afterIsolatedMiss = await manager.coreIsAlive() + XCTAssertTrue(afterIsolatedMiss, + "an isolated miss after a recovery must not be strike two") + let state = await coreState(appState) + XCTAssertEqual(state, .connected) + } + + /// A wedged core accepts the connection and never answers. The probe must + /// give up on its own schedule instead of inheriting the API client's 30s + /// request timeout — 30 seconds of blocking per sample makes the detector + /// useless and stalls the refresh loop behind it. + func testProbeGivesUpQuicklyOnAnUnresponsiveCore() async throws { + let ready = ReadyBehaviour() + let (manager, _, _) = try await attachToStubCore(probeTimeout: 0.3, ready: ready) + + ready.current = .hang(seconds: 5.0) + + let started = Date() + let alive = await manager.coreIsAlive() + let elapsed = Date().timeIntervalSince(started) + + XCTAssertTrue(alive, "first miss is tolerated (see hysteresis)") + XCTAssertLessThan(elapsed, 2.0, + "the probe must use its own short timeout, not the 30s request timeout") + } + + // MARK: - GH #926: one reconnection at a time + + /// Two detectors can see the same dead core at the same moment — the + /// liveness tick and the SSE disconnect handler, or the tick and the + /// process-exit handler. `attemptReconnection()` suspends, so two + /// overlapping runs can each connect (and, for a core we own, each LAUNCH), + /// overwrite `process`, and orphan one holding the BBolt lock. + /// + /// `connectToCore()` fetches `/api/v1/info` exactly once, so counting that + /// request counts reconnections. + func testConcurrentLossSignalsRunExactlyOneReconnection() async throws { + let (manager, _, stub) = try await attachToStubCore(baseDelay: 0.2) + let connectsAfterAttach = stub.requestCount(path: "/api/v1/info") + XCTAssertEqual(connectsAfterAttach, 1, "precondition: one connect for the attach") + + // Four detectors fire at once. + await withTaskGroup(of: Void.self) { group in + for _ in 0..<4 { + group.addTask { await manager.handleCoreLoss() } + } + } + + XCTAssertEqual(stub.requestCount(path: "/api/v1/info"), connectsAfterAttach + 1, + "concurrent loss signals must produce exactly ONE reconnection") + } + + /// The process-exit handler is the second reconnection entry point. It must + /// honour the same in-flight guard: a managed core dying while the liveness + /// tick is already reconnecting must not start a second attempt. + func testProcessExitDuringAReconnectionDoesNotStartASecond() async throws { + let (manager, _, stub) = try await attachToStubCore(baseDelay: 2.0) + let connectsAfterAttach = stub.requestCount(path: "/api/v1/info") + + // Start a reconnection and let it reach its backoff sleep (1s at + // baseDelay 2.0), so it is provably still in flight below. + let reconnecting = Task { await manager.handleCoreLoss() } + try await Task.sleep(nanoseconds: 300_000_000) + + await manager.handleProcessExit(status: 1) + await reconnecting.value + + // Give a second (wrongly started) reconnection time to land. + try await Task.sleep(nanoseconds: 1_500_000_000) + + XCTAssertEqual(stub.requestCount(path: "/api/v1/info"), connectsAfterAttach + 1, + "a process exit during an in-flight reconnection must not start a second one") + } + + // MARK: - Never destroy or duplicate a live core + + /// The most dangerous failure available here: a core that is UP but slow to + /// answer `/ready` must not have its socket unlinked and a second core + /// started against the same data directory. That is two writers on one + /// BBolt database, not a UI glitch. + /// + /// Hysteresis does not help at startup — there is no established connection + /// to be tolerant about — so the startup path has to reason from the socket + /// itself: something is listening, therefore hands off. + func testAnUnresponsiveCoreIsNeverUnlinkedOrReplaced() async throws { + // If the code under test does try to spawn, spawn something harmless. + let fakeCore = try FakeCoreBinary(behaviour: "exit 1") + let restoreEnv = fakeCore.install() + defer { restoreEnv() } + + let ready = ReadyBehaviour() + ready.current = .json(#"{"success":false,"error":"still starting"}"#, status: 503) + let stub = UnixSocketHTTPStub.healthyCore(ready: ready) + try stub.start() + self.stub = stub + + let appState = await MainActor.run { AppState() } + let manager = CoreProcessManager( + appState: appState, + notificationService: NotificationService(deliveryEnabled: false), + socketPath: stub.path, + refreshInterval: 0.2, + probeTimeout: 0.3 + ) + self.manager = manager + + await manager.start(maySpawn: true) + + XCTAssertTrue(FileManager.default.fileExists(atPath: stub.path), + "a socket a live core is listening on must never be unlinked") + XCTAssertTrue(SocketTransport.isSocketAvailable(path: stub.path), + "the live core's socket must still be connectable") + let launched = await manager.coresLaunched + XCTAssertEqual(launched, 0, + "a competing core must not be started against the same data directory") + XCTAssertNil(manager.managedProcess, + "no process may be launched while another core holds the socket") + + // And when it finishes starting up, we attach — no user action needed. + ready.current = .json(#"{"success":true}"#) + let state = await waitForState(appState, timeout: 10.0) { $0 == .connected } + XCTAssertEqual(state, .connected, + "once the core answers, the tray must attach to it") + } + + /// `retry()` kills the managed core and immediately calls `start()`, while + /// that core's termination handler is on its way to starting a reconnection. + /// The gate has to cover the SPAWN path, not just attach and reconnect, or + /// both launch. + func testStartStandsDownWhileAReconnectionIsAlreadyRunning() async throws { + let fakeCore = try FakeCoreBinary(behaviour: "sleep 30") + let restoreEnv = fakeCore.install() + defer { restoreEnv() } + + let appState = await MainActor.run { AppState() } + let manager = CoreProcessManager( + appState: appState, + notificationService: NotificationService(deliveryEnabled: false), + // 2s base delay: the reconnection is provably still in its backoff + // when start() runs below. + reconnectionPolicy: ReconnectionPolicy( + baseDelay: 2.0, maxDelay: 2.0, maxAttempts: 3, jitterFactor: 0.0 + ), + socketPath: "/tmp/mcpproxy-test-\(UUID().uuidString.prefix(8)).sock", + refreshInterval: 60 + ) + self.manager = manager + + let reconnecting = Task { await manager.handleProcessExit(status: 1) } + try await Task.sleep(nanoseconds: 300_000_000) + + await manager.start(maySpawn: true) + XCTAssertEqual(fakeCore.launchCount(), 0, + "start() must not launch a core while a reconnection is already running") + + reconnecting.cancel() + _ = await reconnecting.value + } + + /// Standing down on process exit must not eat the retry ladder. The ladder + /// belongs to whoever holds the gate: it relaunches, sees its replacement + /// fail, and climbs to the next rung itself instead of waiting for a + /// termination callback that is (correctly) refusing to start a second + /// reconnection. + func testTheRelaunchLadderRunsEveryRung() async throws { + let fakeCore = try FakeCoreBinary(behaviour: "exit 1") + let restoreEnv = fakeCore.install() + defer { restoreEnv() } + + let appState = await MainActor.run { AppState() } + let manager = CoreProcessManager( + appState: appState, + notificationService: NotificationService(deliveryEnabled: false), + reconnectionPolicy: ReconnectionPolicy( + baseDelay: 0.02, maxDelay: 0.05, maxAttempts: 3, jitterFactor: 0.0 + ), + socketPath: "/tmp/mcpproxy-test-\(UUID().uuidString.prefix(8)).sock", + refreshInterval: 60 + ) + self.manager = manager + + // One launch from start(), then the ladder: maxRetries relaunches. + await manager.start(maySpawn: true) + + let deadline = Date().addingTimeInterval(20) + while await manager.coresLaunched < 4 && Date() < deadline { + try await Task.sleep(nanoseconds: 100_000_000) + } + + let launched = await manager.coresLaunched + XCTAssertEqual(launched, 4, + "one initial launch plus three ladder rungs — no more, no fewer") + let state = await waitForState(appState, timeout: 5.0) { + $0 == .error(.maxRetriesExceeded) + } + XCTAssertEqual(state, .error(.maxRetriesExceeded), + "the ladder must run out honestly rather than stopping after one try") + } + + /// A core that crashes, is relaunched, and then runs normally must not carry + /// its old strikes: the next crash gets the full ladder again. + func testASuccessfulRelaunchClearsTheLadder() async throws { + let fakeCore = try FakeCoreBinary.failingThenHealthy(failures: 1) + let restoreEnv = fakeCore.install() + defer { restoreEnv() } + + let socketPath = "/tmp/mcpproxy-test-\(UUID().uuidString.prefix(8)).sock" + let appState = await MainActor.run { AppState() } + let manager = CoreProcessManager( + appState: appState, + notificationService: NotificationService(deliveryEnabled: false), + reconnectionPolicy: ReconnectionPolicy( + baseDelay: 0.02, maxDelay: 0.05, maxAttempts: 3, jitterFactor: 0.0 + ), + socketPath: socketPath, + refreshInterval: 60 + ) + self.manager = manager + + // Rung 1 fails; rung 2 launches a core that stays up. + let reconnecting = Task { await manager.handleProcessExit(status: 1) } + + // The second launch is the one that survives — give it its socket, as a + // real core would create. + let deadline = Date().addingTimeInterval(20) + while fakeCore.launchCount() < 2 && Date() < deadline { + try await Task.sleep(nanoseconds: 50_000_000) + } + XCTAssertGreaterThanOrEqual(fakeCore.launchCount(), 2, "precondition: rung 1 failed, rung 2 launched") + + let stub = UnixSocketHTTPStub.healthyCore(at: socketPath) + try stub.start() + self.stub = stub + + _ = await reconnecting.value + + let state = await coreState(appState) + XCTAssertEqual(state, .connected, "the relaunched core came up and we connected to it") + let remainingStrikes = await manager.retryCount + XCTAssertEqual(remainingStrikes, 0, + "a successful relaunch must clear the ladder, or a later crash gets no retries") + } + + /// The routing hint rides on `httpAdditionalHeaders`, which Foundation + /// attaches to EVERY request the session makes — including ones this + /// protocol declines to intercept, and ones following a redirect. So it must + /// carry an opaque token rather than the socket path (which contains the + /// user's home directory), and it must never reach the wire. + func testTheRouteHintNeverLeaksThePathOrTheHeader() async throws { + let (_, _, stub) = try await attachToStubCore() + + let head = stub.lastRequestHead() ?? "" + XCTAssertFalse(head.isEmpty, "precondition: the stub saw a request") + XCTAssertFalse(head.lowercased().contains(SocketURLProtocol.routeHeader.lowercased()), + "the transport routing header must be stripped before the wire") + XCTAssertFalse(head.contains(stub.path), + "the socket path must never appear in an outgoing request") + + let token = SocketURLProtocol.makeRoute(to: stub.path) + XCTAssertNotEqual(token, stub.path, + "the header value must be an opaque token, not the path itself") + XCTAssertEqual(SocketURLProtocol.routes(for: token), stub.path, + "and it must resolve back to the socket it was minted for") + } + + // MARK: - The property: never two cores on one data directory + + /// The errno classification must be honoured by STARTUP, not only by the + /// liveness tick. A socket that refuses connections — a full listen queue on + /// a live core looks exactly like this — must not lead to unlinking the + /// socket and spawning over whatever is behind it. + /// + /// This is about the FIRST probe: one refusal is not evidence, so startup + /// waits. What the tray does when refusals are still all it has at the end of + /// the whole window is a different question, decided by the deadline — + /// see `testADeadlineWithNoAnswerEverEscalatesToALaunch`. The timeout here is + /// long on purpose so the two cannot be confused: nothing in this test is + /// allowed to depend on the deadline firing. + func testAnIndeterminateSocketAtStartupNeitherUnlinksNorSpawns() async throws { + let fakeCore = try FakeCoreBinary(behaviour: "exit 1") + let restoreEnv = fakeCore.install() + defer { restoreEnv() } + + // A socket file that refuses connections. From out here this is + // indistinguishable from a live core with a full backlog. + let stub = UnixSocketHTTPStub.healthyCore() + try stub.start() + stub.stop(unlinkSocket: false) + defer { unlink(stub.path) } + XCTAssertEqual(SocketTransport.probeSocket(path: stub.path), .refused, + "precondition: the socket refuses connections") + + let appState = await MainActor.run { AppState() } + let manager = CoreProcessManager( + appState: appState, + notificationService: NotificationService(deliveryEnabled: false), + socketPath: stub.path, + refreshInterval: 60, + probeTimeout: 0.3, + unresponsiveCoreTimeout: 30.0 + ) + self.manager = manager + + await manager.start(maySpawn: true) + + // Several attach-watch probes' worth of patience: none of them may turn + // into a spawn either. + try await Task.sleep(nanoseconds: 2_500_000_000) + + let launched = await manager.coresLaunched + XCTAssertEqual(launched, 0, + "a socket we cannot classify must not be spawned over") + XCTAssertNil(manager.managedProcess) + XCTAssertTrue(FileManager.default.fileExists(atPath: stub.path), + "and the file must still be there — the tray unlinks nothing") + + // ...and the tray does not pretend to be connected while it waits. + let state = await coreState(appState) + XCTAssertEqual(state, .waitingForCore, + "an unusable socket must leave the tray visibly waiting, not connected") + } + + /// A rung that gives up on a launch must terminate and reap it. A core that + /// hung without ever becoming ready is still a core; leaving it running + /// while the next rung launches makes the ladder a core factory. + func testAbandonedLaunchAttemptsAreReapedBeforeTheNextRung() async throws { + // Stays alive but never creates a socket: waitForSocket gives up on it + // while the process is still running. + let fakeCore = try FakeCoreBinary(behaviour: "sleep 30") + let restoreEnv = fakeCore.install() + defer { restoreEnv() } + + let appState = await MainActor.run { AppState() } + let manager = CoreProcessManager( + appState: appState, + notificationService: NotificationService(deliveryEnabled: false), + reconnectionPolicy: ReconnectionPolicy( + baseDelay: 0.02, maxDelay: 0.05, maxAttempts: 3, jitterFactor: 0.0 + ), + socketPath: "/tmp/mcpproxy-test-\(UUID().uuidString.prefix(8)).sock", + refreshInterval: 60, + probeTimeout: 0.3, + unresponsiveCoreTimeout: 0.5, + // Give up on a socket that never appears after a second, so the + // ladder actually climbs during the test. + socketWaitTimeout: 1.0 + ) + self.manager = manager + + let launching = Task { await manager.start(maySpawn: true) } + + // Sample while the ladder is climbing: by now at least two rungs have + // been tried, and each abandoned attempt is still sleeping unless it was + // reaped. + var worstCase = 0 + let deadline = Date().addingTimeInterval(6) + while Date() < deadline { + worstCase = max(worstCase, try shellCount("pgrep -f '\(fakeCore.path)' | wc -l")) + if await manager.coresLaunched >= 3 { break } + try await Task.sleep(nanoseconds: 200_000_000) + } + + let climbed = await manager.coresLaunched + XCTAssertGreaterThanOrEqual(climbed, 2, + "precondition: the ladder climbed past its first rung") + XCTAssertLessThanOrEqual(worstCase, 1, + "at most one launched core may be alive at any moment — " + + "an abandoned attempt must be terminated and reaped") + + launching.cancel() + _ = await launching.value + let leftovers = try shellCount("pgrep -f '\(fakeCore.path)' | wc -l") + XCTAssertEqual(leftovers, 0, "no launched core may outlive the ladder") + } + + /// Stop must be able to cancel work already in flight. A ladder asleep in + /// its backoff will otherwise wake up and launch a core after the user + /// pressed Stop, leaving the tray showing "Stopped" while owning a process. + func testShutdownCancelsAnInFlightLaunchLadder() async throws { + let fakeCore = try FakeCoreBinary(behaviour: "exit 1") + let restoreEnv = fakeCore.install() + defer { restoreEnv() } + + let appState = await MainActor.run { AppState() } + let manager = CoreProcessManager( + appState: appState, + notificationService: NotificationService(deliveryEnabled: false), + // 2s backoff: the ladder is provably asleep when we shut down. + reconnectionPolicy: ReconnectionPolicy( + baseDelay: 2.0, maxDelay: 2.0, maxAttempts: 3, jitterFactor: 0.0 + ), + socketPath: "/tmp/mcpproxy-test-\(UUID().uuidString.prefix(8)).sock", + refreshInterval: 60 + ) + self.manager = manager + + let launching = Task { await manager.start(maySpawn: true) } + // Wait for the first launch to happen and fail, putting the ladder to sleep. + let deadline = Date().addingTimeInterval(10) + while await manager.coresLaunched < 1 && Date() < deadline { + try await Task.sleep(nanoseconds: 50_000_000) + } + let beforeShutdown = await manager.coresLaunched + XCTAssertEqual(beforeShutdown, 1, "precondition: one launch, ladder now in backoff") + + await manager.shutdown() + let afterShutdown = await manager.coresLaunched + + // Well past the 2s backoff the ladder was sleeping through. + try await Task.sleep(nanoseconds: 3_000_000_000) + let finalCount = await manager.coresLaunched + XCTAssertEqual(finalCount, afterShutdown, + "a ladder must not wake up and launch a core after shutdown") + + let state = await coreState(appState) + XCTAssertEqual(state, .idle) + XCTAssertNil(manager.managedProcess, + "the tray must not say Stopped while owning a running process") + + launching.cancel() + _ = await launching.value + } + + /// The choke point: every path that wants a core goes through `launchCore`, + /// so the invariant is enforced there rather than depending on six callers + /// staying disciplined. With a core listening on the socket, no path may + /// produce a second one. + func testNoPathSpawnsWhileACoreIsListening() async throws { + let fakeCore = try FakeCoreBinary(behaviour: "sleep 30") + let restoreEnv = fakeCore.install() + defer { restoreEnv() } + + // A core that is UP but sick: it holds the socket and never answers + // /ready. This is the dangerous shape — connectToCore() fails, so every + // recovery path is tempted to launch a replacement on top of it. + let ready = ReadyBehaviour() + ready.current = .json(#"{"success":false,"error":"wedged"}"#, status: 503) + let stub = UnixSocketHTTPStub.healthyCore(ready: ready) + try stub.start() + self.stub = stub + + let appState = await MainActor.run { AppState() } + // As if this core were one WE launched — the ownership under which the + // recovery paths are allowed to relaunch. + await MainActor.run { appState.ownership = .trayManaged } + + let manager = CoreProcessManager( + appState: appState, + notificationService: NotificationService(deliveryEnabled: false), + reconnectionPolicy: ReconnectionPolicy( + baseDelay: 0.02, maxDelay: 0.05, maxAttempts: 3, jitterFactor: 0.0 + ), + socketPath: stub.path, + refreshInterval: 60, + probeTimeout: 0.3, + unresponsiveCoreTimeout: 0.5 + ) + self.manager = manager + + // Every entry point that can reach a spawn. + await manager.start(maySpawn: true) + await manager.handleProcessExit(status: 1) + await manager.retry() + await manager.handleCoreLoss() + + let launched = await manager.coresLaunched + XCTAssertEqual(launched, 0, + "no entry point may spawn a core while one is listening on the socket") + XCTAssertNil(manager.managedProcess) + XCTAssertTrue(FileManager.default.fileExists(atPath: stub.path), + "and none of them may unlink its socket") + XCTAssertEqual(SocketTransport.probeSocket(path: stub.path), .connectable, + "the core that was there must still be there") + } + + // MARK: - A socket file nobody ever answered on + + /// The common case, and the one the tray must recover from without help: a + /// core was `kill -9`ed and its socket FILE is still on disk. Every probe + /// refuses, so the tray correctly refuses to spawn on the spot — but if it + /// waited forever the user would be stuck, because nothing else in the system + /// removes that file. The Go core's own `cleanupStaleSocket` only runs once a + /// core is launched, and the tray no longer unlinks anything. + /// + /// So the deadline that used to end in an error ("quit that process") — about + /// a process that does not exist — ends in a launch when, and only when, + /// nothing has accepted a single connection in the whole window. + func testADeadlineWithNoAnswerEverEscalatesToALaunch() async throws { + // If the code under test does launch (it must), launch something + // harmless rather than the real core. + let fakeCore = try FakeCoreBinary(behaviour: "sleep 30") + let restoreEnv = fakeCore.install() + defer { restoreEnv() } + + // Exactly what a killed core leaves behind: the file, and nothing on it. + let stub = UnixSocketHTTPStub.healthyCore() + try stub.start() + stub.stop(unlinkSocket: false) + defer { unlink(stub.path) } + XCTAssertEqual(SocketTransport.probeSocket(path: stub.path), .refused, + "precondition: the socket file is there and refuses every connection") + + let appState = await MainActor.run { AppState() } + let manager = CoreProcessManager( + appState: appState, + notificationService: NotificationService(deliveryEnabled: false), + reconnectionPolicy: ReconnectionPolicy( + baseDelay: 0.05, maxDelay: 0.1, maxAttempts: 3, jitterFactor: 0.0 + ), + socketPath: stub.path, + refreshInterval: 60, + probeTimeout: 0.3, + unresponsiveCoreTimeout: 0.5, + socketWaitTimeout: 1.0 + ) + self.manager = manager + + await manager.start(maySpawn: true) + + let duringTheWait = await manager.coresLaunched + XCTAssertEqual(duringTheWait, 0, + "one refused probe proves nothing — the tray must wait before it acts") + let waiting = await coreState(appState) + XCTAssertEqual(waiting, .waitingForCore, + "precondition: the refused socket sent us down the wait path") + + let deadline = Date().addingTimeInterval(10) + while await manager.coresLaunched < 1 && Date() < deadline { + try await Task.sleep(nanoseconds: 50_000_000) + } + let launched = await manager.coresLaunched + XCTAssertGreaterThanOrEqual(launched, 1, + "a socket file nothing has EVER answered on must be recovered " + + "by launching a core, not by telling the user to quit a " + + "process that does not exist") + } + + /// The invariant the whole change exists to protect, at the same deadline: a + /// core that IS there — it accepts connections, it just never answers + /// `/ready` — must never be launched over, however long it stays silent. Here + /// the error message ("quit that process") is true, so that is what happens. + func testADeadlineOnASilentButLiveCoreNeverEscalates() async throws { + let fakeCore = try FakeCoreBinary(behaviour: "sleep 30") + let restoreEnv = fakeCore.install() + defer { restoreEnv() } + + let ready = ReadyBehaviour() + ready.current = .json(#"{"success":false,"error":"wedged"}"#, status: 503) + let stub = UnixSocketHTTPStub.healthyCore(ready: ready) + try stub.start() + self.stub = stub + + let appState = await MainActor.run { AppState() } + let manager = CoreProcessManager( + appState: appState, + notificationService: NotificationService(deliveryEnabled: false), + reconnectionPolicy: ReconnectionPolicy( + baseDelay: 0.05, maxDelay: 0.1, maxAttempts: 3, jitterFactor: 0.0 + ), + socketPath: stub.path, + refreshInterval: 60, + probeTimeout: 0.3, + unresponsiveCoreTimeout: 0.5, + socketWaitTimeout: 1.0 + ) + self.manager = manager + + await manager.start(maySpawn: true) + + let state = await waitForState(appState, timeout: 5.0) { + if case .error = $0 { return true } + return false + } + guard case .error(let error) = state else { + return XCTFail("a core that holds the socket and never answers must become an " + + "actionable error, got \(state)") + } + XCTAssertTrue(error.userMessage.contains("not responding"), + "the message must still be the one about a process to quit, got: \(error.userMessage)") + + // Well past the deadline, and past several attach-watch probes. + try await Task.sleep(nanoseconds: 1_500_000_000) + let launched = await manager.coresLaunched + XCTAssertEqual(launched, 0, + "a listener that is genuinely there must never be spawned over") + XCTAssertNil(manager.managedProcess) + XCTAssertEqual(SocketTransport.probeSocket(path: stub.path), .connectable, + "and the core that was there must still be there, socket and all") + } + + /// The case the accumulated evidence exists for, and the one a last-moment + /// probe cannot catch: a core that WAS accepting connections and then stops — + /// which is what a full listen backlog looks like from out here, and is + /// indistinguishable, sample by sample, from a file a dead process left + /// behind. Every probe from now until the deadline refuses, so the + /// confirmation window at the deadline is satisfied; only the memory of the + /// connection that was accepted earlier in the window stands between that + /// core and a second writer on its data directory. + func testACoreThatAcceptedEarlierInTheWindowIsNeverEscalatedOver() async throws { + let fakeCore = try FakeCoreBinary(behaviour: "sleep 30") + let restoreEnv = fakeCore.install() + defer { restoreEnv() } + + // Accepts connections, never answers /ready: the wait path, from a core + // that is demonstrably there. + let ready = ReadyBehaviour() + ready.current = .json(#"{"success":false,"error":"still starting"}"#, status: 503) + let stub = UnixSocketHTTPStub.healthyCore(ready: ready) + try stub.start() + defer { unlink(stub.path) } + + let appState = await MainActor.run { AppState() } + let manager = CoreProcessManager( + appState: appState, + notificationService: NotificationService(deliveryEnabled: false), + reconnectionPolicy: ReconnectionPolicy( + baseDelay: 0.05, maxDelay: 0.1, maxAttempts: 3, jitterFactor: 0.0 + ), + socketPath: stub.path, + refreshInterval: 60, + probeTimeout: 0.3, + unresponsiveCoreTimeout: 1.5, + socketWaitTimeout: 1.0 + ) + self.manager = manager + + await manager.start(maySpawn: true) + let waiting = await coreState(appState) + XCTAssertEqual(waiting, .waitingForCore, + "precondition: a probe was accepted, so the tray is waiting for that core") + + // Now it stops accepting, keeping its socket file — a saturated backlog. + // Every probe from here on is a refusal, so nothing sampled AT the + // deadline can tell this apart from a stale file. + stub.stop(unlinkSocket: false) + XCTAssertEqual(SocketTransport.probeSocket(path: stub.path), .refused, + "precondition: from now on every probe refuses") + + // Well past the deadline, and past the escalation's own confirmation. + try await Task.sleep(nanoseconds: 3_000_000_000) + + let launched = await manager.coresLaunched + XCTAssertEqual(launched, 0, + "a core that accepted a connection during the window must never be " + + "launched over, however many refusals follow it") + XCTAssertNil(manager.managedProcess) + } + + // MARK: - What a failed socket connect does and does not prove + + /// `isSocketAvailable` says false for three different situations, and only + /// one of them means "the core is gone". + func testSocketProbeDistinguishesAbsentFromRefused() throws { + let missing = "/tmp/mcpproxy-test-\(UUID().uuidString.prefix(8)).sock" + XCTAssertEqual(SocketTransport.probeSocket(path: missing), .absent, + "no socket file at all: a running core always owns its socket") + + let stub = UnixSocketHTTPStub.healthyCore() + try stub.start() + self.stub = stub + XCTAssertEqual(SocketTransport.probeSocket(path: stub.path), .connectable) + + // Leave the file behind with nothing listening — what `kill -9` leaves, + // and also what a full listen queue looks like from out here. + stub.stop(unlinkSocket: false) + XCTAssertEqual(SocketTransport.probeSocket(path: stub.path), .refused) + unlink(stub.path) + } + + /// A refused connection is ambiguous — a stale socket from a killed process + /// and a live core with a full listen backlog are indistinguishable — so it + /// gets the same tolerance as a missed readiness check, not an instant + /// verdict. + func testARefusedSocketIsToleratedOnceLikeAReadinessMiss() async throws { + let (manager, appState, stub) = try await attachToStubCore() + + stub.stop(unlinkSocket: false) + defer { unlink(stub.path) } + XCTAssertEqual(SocketTransport.probeSocket(path: stub.path), .refused, + "precondition: the socket file is still there, refusing connections") + + let afterFirst = await manager.coreIsAlive() + XCTAssertTrue(afterFirst, + "one refused connection could be a full listen queue on a healthy core") + let stillConnected = await coreState(appState) + XCTAssertEqual(stillConnected, .connected) + + let afterSecond = await manager.coreIsAlive() + XCTAssertFalse(afterSecond, "two in a row is enough") + } + + /// The attach path is the third entry point. A manual retry racing the idle + /// attach watcher used to run two `attachToExternalCore()` calls, each + /// replacing the other's clients and tasks — and a late failure from one + /// could overwrite the other's successful `.connected` with `.error`. + func testConcurrentAttachesAttachOnce() async throws { + let stub = UnixSocketHTTPStub.healthyCore() + try stub.start() + self.stub = stub + + let appState = await MainActor.run { AppState() } + let manager = CoreProcessManager( + appState: appState, + notificationService: NotificationService(deliveryEnabled: false), + socketPath: stub.path, + refreshInterval: 60 + ) + self.manager = manager + + await withTaskGroup(of: Void.self) { group in + for _ in 0..<4 { + group.addTask { await manager.start(maySpawn: false) } + } + } + + XCTAssertEqual(stub.requestCount(path: "/api/v1/info"), 1, + "concurrent attaches must produce exactly ONE attachment") + let state = await coreState(appState) + XCTAssertEqual(state, .connected, + "a concurrent attach must not leave the tray idle or errored") + } + + // MARK: - The choke point's preflight, tested directly + + /// The four checks in front of `proc.run()` are what keeps two cores off one + /// data directory. The tests above reach them through whole entry points + /// (start/retry/handleProcessExit), which proves the paths but leaves the + /// checks themselves only indirectly covered — a reordering or a dropped + /// guard could still pass them. These call the preflight directly. + private func makeManager( + socketPath: String, + appState: AppState + ) -> CoreProcessManager { + let manager = CoreProcessManager( + appState: appState, + notificationService: NotificationService(deliveryEnabled: false), + socketPath: socketPath, + refreshInterval: 60, + probeTimeout: 0.3 + ) + self.manager = manager + return manager + } + + /// Run `body` with the connection gate held, exactly as every real caller of + /// `launchCore` does. + private func withConnectionGate( + _ manager: CoreProcessManager, + _ body: (CoreProcessManager) async -> Void + ) async { + let took = await manager.beginConnectionWork() + XCTAssertTrue(took, "precondition: the gate was free") + await body(manager) + await manager.endConnectionWork() + } + + /// The check that matters most: something is listening, so there is nothing + /// to launch — whatever the caller believes about its own state. + func testPreflightRefusesToLaunchWhileSomethingIsListening() async throws { + let stub = UnixSocketHTTPStub.healthyCore() + try stub.start() + self.stub = stub + + let appState = await MainActor.run { AppState() } + let manager = makeManager(socketPath: stub.path, appState: appState) + + await withConnectionGate(manager) { manager in + do { + try await manager.preflightLaunch() + XCTFail("a connectable socket must never be launched over") + } catch { + XCTAssertEqual(error as? CoreError, + .general("A core is already running on \(stub.path)")) + } + } + } + + /// A core we already own is still a core. The process check comes before the + /// socket check on purpose: a managed core that has not yet created its + /// socket would otherwise pass the preflight and get a sibling. + func testPreflightRefusesToLaunchWhileAManagedProcessIsRunning() async throws { + let fakeCore = try FakeCoreBinary(behaviour: "sleep 30") + let restoreEnv = fakeCore.install() + defer { restoreEnv() } + + let socketPath = "/tmp/mcpproxy-test-\(UUID().uuidString.prefix(8)).sock" + let appState = await MainActor.run { AppState() } + let manager = makeManager(socketPath: socketPath, appState: appState) + + // Let the tray spawn its core, then give that core its socket — as a + // real one would — so the launch completes and the process survives. + let starting = Task { await manager.start(maySpawn: true) } + let deadline = Date().addingTimeInterval(10) + while fakeCore.launchCount() < 1 && Date() < deadline { + try await Task.sleep(nanoseconds: 50_000_000) + } + let stub = UnixSocketHTTPStub.healthyCore(at: socketPath) + try stub.start() + self.stub = stub + await starting.value + + let running = try XCTUnwrap(manager.managedProcess, "precondition: the tray owns a core") + XCTAssertTrue(running.isRunning, "precondition: that core is still up") + + await withConnectionGate(manager) { manager in + do { + try await manager.preflightLaunch() + XCTFail("a running managed core must never get a sibling") + } catch { + XCTAssertEqual( + error as? CoreError, + .general("internal error: tried to launch a core while " + + "PID \(running.processIdentifier) is still running") + ) + } + } + } + + /// The other half of the same coin, and the reason the tray no longer + /// unlinks anything: a socket FILE that nothing accepts on must not block a + /// launch. The core we spawn cleans that file up itself. + func testPreflightAllowsALaunchOverASocketFileNobodyAnswers() async throws { + let stub = UnixSocketHTTPStub.healthyCore() + try stub.start() + stub.stop(unlinkSocket: false) + defer { unlink(stub.path) } + XCTAssertTrue(FileManager.default.fileExists(atPath: stub.path), + "precondition: the socket file is still on disk") + XCTAssertEqual(SocketTransport.probeSocket(path: stub.path), .refused, + "precondition: nothing accepts on it") + + let appState = await MainActor.run { AppState() } + let manager = makeManager(socketPath: stub.path, appState: appState) + + await withConnectionGate(manager) { manager in + do { + try await manager.preflightLaunch() + } catch { + XCTFail("a stale socket file must not be mistaken for a live core — threw \(error)") + } + } + } + + /// Without the gate there is no serialisation, so the preflight's own + /// answer would be stale before `proc.run()`. It refuses instead. + func testPreflightRefusesToLaunchWithoutTheConnectionGate() async throws { + let socketPath = "/tmp/mcpproxy-test-\(UUID().uuidString.prefix(8)).sock" + let appState = await MainActor.run { AppState() } + let manager = makeManager(socketPath: socketPath, appState: appState) + + do { + try await manager.preflightLaunch() + XCTFail("a launch outside the connection gate must be refused") + } catch { + XCTAssertEqual( + error as? CoreError, + .general("internal error: tried to launch a core without the connection gate") + ) + } + } + + /// The confirmation window exists for exactly this: a core that binds the + /// socket while we are still deciding whether the socket is empty. One + /// sample would have missed it and spawned a second writer. + /// + /// The ordering is established, not hoped for. An earlier version of this + /// test started an unsynchronised task and slept: if that task had begun + /// late its very FIRST probe would have seen the listener, and the test would + /// have passed without ever running the race it names. Here the probe seam + /// holds the confirmation at probe 0 — which has already returned "nothing + /// there" — until the listener is bound, so the only way to pass is to notice + /// a core that appeared BETWEEN two probes. + func testConfirmationNoticesACoreThatAppearsBetweenProbes() async throws { + let socketPath = "/tmp/mcpproxy-test-\(UUID().uuidString.prefix(8)).sock" + let appState = await MainActor.run { AppState() } + let manager = makeManager(socketPath: socketPath, appState: appState) + XCTAssertEqual(SocketTransport.probeSocket(path: socketPath), .absent, + "precondition: the first probe sees nothing at all") + + let firstProbeDone = Latch() + let listenerIsUp = Latch() + let probesSeen = ProbeCounter() + + let confirming = Task { + await manager.confirmNothingIsListening(attempts: 4, interval: 0.05) { index in + await probesSeen.count() + guard index == 0 else { return } + await firstProbeDone.open() + await listenerIsUp.wait() + } + } + + await firstProbeDone.wait() + let beforeTheListener = await probesSeen.total + XCTAssertEqual(beforeTheListener, 1, + "precondition: exactly one probe has run, and it found an empty socket") + + let stub = UnixSocketHTTPStub.healthyCore(at: socketPath) + try stub.start() + self.stub = stub + await listenerIsUp.open() + + let empty = await confirming.value + XCTAssertFalse(empty, + "a core that binds mid-confirmation must be seen — the whole point " + + "of confirming over time rather than on one sample") + let afterTheListener = await probesSeen.total + XCTAssertGreaterThanOrEqual(afterTheListener, 2, + "and it must be seen by a LATER probe, not by the first one") + } +} + +// MARK: - Test-only synchronisation + +/// A one-shot gate: `wait()` suspends until someone calls `open()`, and returns +/// immediately ever after. Enough to pin an ordering between the code under test +/// and the test itself without a single sleep. +private actor Latch { + private var isOpen = false + private var waiting: [CheckedContinuation] = [] + + func open() { + guard !isOpen else { return } + isOpen = true + let resuming = waiting + waiting = [] + resuming.forEach { $0.resume() } + } + + func wait() async { + guard !isOpen else { return } + await withCheckedContinuation { waiting.append($0) } + } +} + +/// How many probes the confirmation window has actually run. +private actor ProbeCounter { + private(set) var total = 0 + func count() { total += 1 } +} diff --git a/native/macos/MCPProxy/MCPProxyTests/Support/FakeCoreBinary.swift b/native/macos/MCPProxy/MCPProxyTests/Support/FakeCoreBinary.swift new file mode 100644 index 00000000..783c7dd0 --- /dev/null +++ b/native/macos/MCPProxy/MCPProxyTests/Support/FakeCoreBinary.swift @@ -0,0 +1,79 @@ +// FakeCoreBinary.swift +// MCPProxyTests/Support +// +// A stand-in for the `mcpproxy` binary the tray spawns. +// +// The spawn path — resolve binary, launch, wait for socket, relaunch on failure +// — cannot be tested with an external stub core: `maySpawn: false` never +// reaches it. But letting a test spawn the REAL core would put a second writer +// on the developer's BBolt database, which is the exact accident these tests +// exist to prevent. So the tests point MCPPROXY_CORE_PATH at a shell script +// that records each launch and then behaves as instructed. + +import Foundation + +/// A script that stands in for the core binary and counts its own launches. +struct FakeCoreBinary { + + /// Path to pass via `MCPPROXY_CORE_PATH`. + let path: String + + /// File the script appends a line to on every launch. + let launchLogPath: String + + private let directory: String + + /// - Parameter behaviour: shell run after the launch is recorded. Use + /// `exit 1` for a core that dies immediately, `sleep 30` for one that + /// stays up. + init(behaviour: String) throws { + let dir = "/tmp/mcpproxy-fakecore-\(UUID().uuidString.prefix(8))" + try FileManager.default.createDirectory(atPath: dir, withIntermediateDirectories: true) + directory = dir + path = "\(dir)/mcpproxy" + launchLogPath = "\(dir)/launches.log" + + let script = """ + #!/bin/sh + echo launch >> "\(launchLogPath)" + \(behaviour) + """ + try script.write(toFile: path, atomically: true, encoding: .utf8) + try FileManager.default.setAttributes([.posixPermissions: 0o755], ofItemAtPath: path) + } + + /// A core that fails its first `failures` launches and then stays up. + /// Used to prove a successful relaunch clears the retry ladder. + static func failingThenHealthy(failures: Int) throws -> FakeCoreBinary { + let counterName = "attempts" + return try FakeCoreBinary(behaviour: """ + COUNT_FILE="$(dirname "$0")/\(counterName)" + n=$(cat "$COUNT_FILE" 2>/dev/null || echo 0) + n=$((n + 1)) + echo "$n" > "$COUNT_FILE" + if [ "$n" -le \(failures) ]; then + exit 1 + fi + sleep 30 + """) + } + + /// How many times the script has been launched. + func launchCount() -> Int { + guard let contents = try? String(contentsOfFile: launchLogPath, encoding: .utf8) else { + return 0 + } + return contents.split(separator: "\n").count + } + + /// Point the tray's binary resolution at this script for the duration of a + /// test. Returns a closure that restores the environment. + func install() -> () -> Void { + setenv("MCPPROXY_CORE_PATH", path, 1) + let dir = directory + return { + unsetenv("MCPPROXY_CORE_PATH") + try? FileManager.default.removeItem(atPath: dir) + } + } +} diff --git a/native/macos/MCPProxy/MCPProxyTests/Support/UnixSocketHTTPStub.swift b/native/macos/MCPProxy/MCPProxyTests/Support/UnixSocketHTTPStub.swift new file mode 100644 index 00000000..908eb5f0 --- /dev/null +++ b/native/macos/MCPProxy/MCPProxyTests/Support/UnixSocketHTTPStub.swift @@ -0,0 +1,373 @@ +// UnixSocketHTTPStub.swift +// MCPProxyTests/Support +// +// A minimal HTTP/1.1 server bound to a Unix domain socket, used to stand in for +// an externally-managed mcpproxy core. +// +// Why this exists: `CoreProcessManager` discovers a running core purely by +// probing a Unix socket (`SocketTransport.isSocketAvailable` + `GET /ready`). +// To exercise the attach path — and, crucially, what happens when that core +// DISAPPEARS or goes UNRESPONSIVE (GH #926) — a test needs a socket it can +// create, wedge, and destroy at will. Nothing else in the app can fake that: +// the transport is a real `URLProtocol` speaking real HTTP over a real fd. +// +// Deliberately tiny: one request per connection, `Connection: close`, no +// keep-alive, no chunking. That is all `SocketURLProtocol` ever asks for. + +import Foundation +#if canImport(Darwin) +import Darwin +#endif + +/// What the stub does with a request. +enum StubResponse { + /// Answer with this status and body. + case reply(status: Int, body: String) + /// Accept the request and never answer, for `seconds` — a core that is up + /// and listening but not responding. The client must time out on its own. + case hang(seconds: TimeInterval) + + static func json(_ body: String, status: Int = 200) -> StubResponse { + .reply(status: status, body: body) + } + + static let notFound = StubResponse.reply( + status: 404, body: #"{"success":false,"error":"not found"}"# + ) +} + +/// An HTTP server on a Unix domain socket. +/// +/// `start()`/`stop()` are called from the test thread; the accept loop runs on +/// its own thread. `stop()` closes the listening socket AND any accepted +/// connection (a stub wedged in `.hang` is blocked on one), then waits for the +/// thread to actually exit, so a stopped stub leaves no descriptor or thread +/// behind to race the next test. +final class UnixSocketHTTPStub { + + /// Path of the socket file. Kept short: `sun_path` is 104 bytes. + let path: String + + /// Maps (method, path) to a response. Anything unmatched gets a 404, which + /// the manager's refresh helpers treat as non-fatal — exactly as they do + /// against a real core that lacks an endpoint. + private let responder: @Sendable (_ method: String, _ path: String) -> StubResponse + + private let stateLock = NSLock() + private var listenFD: Int32 = -1 + private var clientFD: Int32 = -1 + private var stopped = true + private var threadRunning = false + private var requestCounts: [String: Int] = [:] + private var lastHead: String? + private let exited = DispatchSemaphore(value: 0) + + /// - Parameter path: socket path to bind, or `nil` for a fresh temporary one. + /// Pass an existing stub's path to simulate the same core coming back up. + init( + at path: String? = nil, + responder: @escaping @Sendable (_ method: String, _ path: String) -> StubResponse + ) { + // A path under NSTemporaryDirectory() can exceed sun_path's 104 bytes on + // some machines; /tmp keeps it comfortably short and is unlink-able. + self.path = path ?? "/tmp/mcpproxy-test-\(UUID().uuidString.prefix(8)).sock" + self.responder = responder + } + + deinit { + stop() + } + + // MARK: - Lifecycle + + /// Bind, listen, and start accepting. Throws if the socket cannot be created. + func start() throws { + unlink(path) + + let fd = Darwin.socket(AF_UNIX, SOCK_STREAM, 0) + guard fd >= 0 else { throw StubError.socketFailed(errno) } + var on: Int32 = 1 + setsockopt(fd, SOL_SOCKET, SO_NOSIGPIPE, &on, socklen_t(MemoryLayout.size)) + + var addr = sockaddr_un() + addr.sun_family = sa_family_t(AF_UNIX) + let pathBytes = path.utf8CString + guard pathBytes.count <= MemoryLayout.size(ofValue: addr.sun_path) else { + Darwin.close(fd) + throw StubError.pathTooLong(path) + } + withUnsafeMutablePointer(to: &addr.sun_path) { sunPathPtr in + sunPathPtr.withMemoryRebound(to: CChar.self, capacity: pathBytes.count) { dest in + for i in 0...size)) + } + } + guard bindResult == 0 else { + let err = errno + Darwin.close(fd) + throw StubError.bindFailed(err) + } + + guard Darwin.listen(fd, 16) == 0 else { + let err = errno + Darwin.close(fd) + throw StubError.listenFailed(err) + } + + stateLock.lock() + listenFD = fd + stopped = false + threadRunning = true + stateLock.unlock() + + let thread = Thread { [weak self] in + self?.acceptLoop(fd: fd) + self?.exited.signal() + } + thread.name = "UnixSocketHTTPStub" + thread.qualityOfService = .userInitiated + thread.start() + } + + /// Stop serving and remove the socket file — i.e. "the core died". + /// + /// After this returns, `SocketTransport.isSocketAvailable(path:)` is false + /// and the accept thread has exited. + /// + /// - Parameter unlinkSocket: pass `false` to leave the socket FILE behind + /// with nothing listening — what `kill -9` leaves on disk. Connections to + /// it are refused, which is indistinguishable from a live core whose + /// listen queue is full. + func stop(unlinkSocket: Bool = true) { + stopServing(unlinkSocket: unlinkSocket) + } + + private func stopServing(unlinkSocket: Bool) { + stateLock.lock() + let alreadyStopped = stopped + stopped = true + let listen = listenFD + let client = clientFD + listenFD = -1 + clientFD = -1 + let wasRunning = threadRunning + threadRunning = false + stateLock.unlock() + + guard !alreadyStopped else { return } + + // Closing both descriptors unblocks accept() and any pending read(). + if listen >= 0 { Darwin.close(listen) } + if client >= 0 { Darwin.close(client) } + if unlinkSocket { unlink(path) } + + if wasRunning { + // Bounded: a wedged handler polls `stopped` on a 20ms cadence. + _ = exited.wait(timeout: .now() + 2.0) + } + } + + // MARK: - Observability + + /// How many requests this stub has received for `path`. + /// + /// Used to count connect attempts: `connectToCore()` fetches + /// `/api/v1/info` exactly once, so that count IS the number of + /// (re)connections — the observable form of "did we attach twice?". + func requestCount(path: String) -> Int { + stateLock.lock() + defer { stateLock.unlock() } + return requestCounts[path] ?? 0 + } + + private func recordRequest(path: String) { + stateLock.lock() + requestCounts[path, default: 0] += 1 + stateLock.unlock() + } + + /// Raw request headers of the most recent request, exactly as they arrived + /// on the wire. Used to assert that transport-routing headers never leave + /// the process. + func lastRequestHead() -> String? { + stateLock.lock() + defer { stateLock.unlock() } + return lastHead + } + + private func recordHead(_ head: String) { + stateLock.lock() + lastHead = head + stateLock.unlock() + } + + private var isStopped: Bool { + stateLock.lock() + defer { stateLock.unlock() } + return stopped + } + + // MARK: - Accept loop + + private func acceptLoop(fd: Int32) { + while !isStopped { + let client = Darwin.accept(fd, nil, nil) + if client < 0 { + if isStopped { return } + if errno == EINTR { continue } + return + } + var on: Int32 = 1 + setsockopt(client, SOL_SOCKET, SO_NOSIGPIPE, &on, socklen_t(MemoryLayout.size)) + + stateLock.lock() + let stoppedNow = stopped + if !stoppedNow { clientFD = client } + stateLock.unlock() + if stoppedNow { + Darwin.close(client) + return + } + + handle(client: client) + closeClient(client) + } + } + + /// Close the accepted connection unless `stop()` already claimed it. + private func closeClient(_ fd: Int32) { + stateLock.lock() + let owned = clientFD == fd + if owned { clientFD = -1 } + stateLock.unlock() + if owned { Darwin.close(fd) } + } + + private func handle(client: Int32) { + guard let head = readRequestHead(fd: client) else { return } + let requestLine = head.components(separatedBy: "\r\n").first ?? "" + let parts = requestLine.split(separator: " ") + let method = parts.count > 0 ? String(parts[0]) : "GET" + let rawPath = parts.count > 1 ? String(parts[1]) : "/" + let pathOnly = rawPath.components(separatedBy: "?").first ?? rawPath + recordRequest(path: pathOnly) + recordHead(head) + + switch responder(method, pathOnly) { + case .hang(let seconds): + // Hold the connection open without answering. Poll `stopped` so + // teardown is never blocked for the full duration. + let deadline = Date().addingTimeInterval(seconds) + while Date() < deadline && !isStopped { + Thread.sleep(forTimeInterval: 0.02) + } + case .reply(let status, let body): + let bodyData = Data(body.utf8) + var headers = "HTTP/1.1 \(status) \(Self.reason(status))\r\n" + headers += "Content-Type: application/json\r\n" + headers += "Content-Length: \(bodyData.count)\r\n" + headers += "Connection: close\r\n\r\n" + var out = Data(headers.utf8) + out.append(bodyData) + write(fd: client, data: out) + } + } + + /// Read until the end of the request headers. Bodies are ignored — the tray + /// only ever GETs on this path. + private func readRequestHead(fd: Int32) -> String? { + var data = Data() + let bufSize = 4096 + let buf = UnsafeMutableRawPointer.allocate(byteCount: bufSize, alignment: 1) + defer { buf.deallocate() } + let separator = Data([0x0D, 0x0A, 0x0D, 0x0A]) + + while data.range(of: separator) == nil { + let n = Darwin.read(fd, buf, bufSize) + if n <= 0 { break } + data.append(buf.assumingMemoryBound(to: UInt8.self), count: n) + if data.count > 64 * 1024 { break } + } + return String(data: data, encoding: .utf8) + } + + private func write(fd: Int32, data: Data) { + data.withUnsafeBytes { raw in + guard let base = raw.baseAddress else { return } + var written = 0 + while written < data.count { + let n = Darwin.write(fd, base.advanced(by: written), data.count - written) + if n <= 0 { return } + written += n + } + } + } + + private static func reason(_ status: Int) -> String { + switch status { + case 200: return "OK" + case 404: return "Not Found" + case 500: return "Internal Server Error" + case 503: return "Service Unavailable" + default: return "Status" + } + } + + enum StubError: Error { + case socketFailed(Int32) + case bindFailed(Int32) + case listenFailed(Int32) + case pathTooLong(String) + } +} + +// MARK: - A stand-in for a core + +/// Lets a test change how `/ready` behaves while the stub keeps running — a +/// core that is still listening but has stopped answering. +final class ReadyBehaviour: @unchecked Sendable { + private let lock = NSLock() + private var value: StubResponse = .json(#"{"success":true}"#) + + var current: StubResponse { + get { lock.lock(); defer { lock.unlock() }; return value } + set { lock.lock(); value = newValue; lock.unlock() } + } +} + +extension UnixSocketHTTPStub { + + /// A stub that answers the two calls `CoreProcessManager.connectToCore()` + /// requires — `/ready` and `/api/v1/info` — and 404s everything else. + /// + /// `web_ui_url` points at 127.0.0.1:1, a port nothing can be listening on, + /// so the SSE client (which is deliberately TCP-only) fails fast and retries + /// in the background instead of reaching a real core on :8080. + static func healthyCore( + at socketPath: String? = nil, + ready: ReadyBehaviour = ReadyBehaviour() + ) -> UnixSocketHTTPStub { + UnixSocketHTTPStub(at: socketPath) { _, path in + switch path { + case "/ready": + return ready.current + case "/api/v1/info": + return .json(""" + {"success":true,"data":{ + "version":"0.0.0-test", + "web_ui_url":"http://127.0.0.1:1/ui/?apikey=test-api-key", + "listen_addr":"127.0.0.1:1", + "endpoints":{"http":"http://127.0.0.1:1","socket":"unix"} + }} + """) + default: + return .notFound + } + } + } +}