diff --git a/CHANGELOG.md b/CHANGELOG.md index 8338212dc7eb..6222d835caa3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -21,6 +21,7 @@ Docs: https://docs.openclaw.ai ### Fixes +- **Gateway provider auth startup:** avoid default provider-auth warm sweeps while preserving on-demand auth checks and auth-failure rewarming, preventing startup discovery from starving gateway RPCs and channel timers. (#100667, #86752, #86512) Thanks @vincentkoc. - **Small-context compaction:** cap the effective reserve against the known model context window so small local models do not enter compaction from the first token. (#100621) Thanks @vincentkoc. - **Control UI terminal reset:** isolate each web terminal's Ghostty WASM runtime so closing a tab and reopening starts with an empty screen instead of reusing freed terminal cells. - **Plugin install diagnostics:** suppress the misleading hook-pack fallback after plugin install failures only when the hook manifest is absent, while preserving actionable malformed hook-pack errors. (#100554) Thanks @vincentkoc. diff --git a/apps/.i18n/native-source.json b/apps/.i18n/native-source.json index 4f10d97c3c4a..de6c5e6f7b10 100644 --- a/apps/.i18n/native-source.json +++ b/apps/.i18n/native-source.json @@ -18515,7 +18515,7 @@ }, { "kind": "conditional-branch", - "line": 172, + "line": 176, "path": "apps/macos/Sources/OpenClaw/RemotePortTunnel.swift", "source": "ssh tunnel exited before listening", "surface": "apple", @@ -18523,7 +18523,7 @@ }, { "kind": "conditional-branch", - "line": 188, + "line": 192, "path": "apps/macos/Sources/OpenClaw/RemotePortTunnel.swift", "source": "ssh tunnel did not open local port \\(localPort)", "surface": "apple", @@ -18531,7 +18531,7 @@ }, { "kind": "conditional-branch", - "line": 188, + "line": 192, "path": "apps/macos/Sources/OpenClaw/RemotePortTunnel.swift", "source": "ssh tunnel failed: \\(stderr)", "surface": "apple", diff --git a/apps/macos/Sources/OpenClaw/PortGuardian.swift b/apps/macos/Sources/OpenClaw/PortGuardian.swift index b9c995f6f450..383065a8d114 100644 --- a/apps/macos/Sources/OpenClaw/PortGuardian.swift +++ b/apps/macos/Sources/OpenClaw/PortGuardian.swift @@ -21,7 +21,10 @@ actor PortGuardian { let executablePath: String? } - private var records: [Record] = [] + /// Tunnels spawned by THIS process. Disk (`port-guard.json`) holds the union + /// across app instances; entries there that we do not own are reap candidates, + /// never adopted — adopting them would resurrect records a sibling removed. + private var ownRecords: [Int32: Record] = [:] private let logger = Logger(subsystem: "ai.openclaw", category: "portguard") #if DEBUG private var testingDescriptors: [Int: Descriptor] = [:] @@ -35,8 +38,8 @@ actor PortGuardian { self.appSupportDir.appendingPathComponent("port-guard.json", isDirectory: false) } - init() { - self.records = Self.loadRecords(from: Self.recordPath) + private nonisolated static var recordLockPath: URL { + self.appSupportDir.appendingPathComponent("port-guard.lock", isDirectory: false) } func sweep(mode: AppState.ConnectionMode) async { @@ -48,60 +51,56 @@ actor PortGuardian { self.logger.info("port sweep skipped (mode=unconfigured)") return } - let ports = [GatewayEnvironment.gatewayPort()] - for port in ports { - let listeners = await self.listeners(on: port) - guard !listeners.isEmpty else { continue } - for listener in listeners { - if Self.isExpected(listener, port: port, mode: mode) { - let message = """ - port \(port) already served by expected \(listener.command) - (pid \(listener.pid)) — keeping - """ - self.logger.info("\(message, privacy: .public)") - continue - } - if mode == .remote { - let message = """ - port \(port) held by \(listener.command) - (pid \(listener.pid)) in remote mode — not killing - """ - self.logger.warning(message) - continue - } - let killed = await self.kill(listener.pid) - if killed { - let message = """ - port \(port) was held by \(listener.command) - (pid \(listener.pid)); terminated - """ - self.logger.error("\(message, privacy: .public)") - } else { - self.logger.error("failed to terminate pid \(listener.pid) on port \(port, privacy: .public)") - } + let port = GatewayEnvironment.gatewayPort() + for listener in await self.listeners(on: port) { + if Self.isExpected(listener, port: port, mode: mode) { + let message = """ + port \(port) already served by expected \(listener.command) + (pid \(listener.pid)) — keeping + """ + self.logger.info("\(message, privacy: .public)") + continue + } + if mode == .remote { + let message = """ + port \(port) held by \(listener.command) + (pid \(listener.pid)) in remote mode — not killing + """ + self.logger.warning(message) + continue + } + if await Self.terminateProcess(listener.pid) { + let message = """ + port \(port) was held by \(listener.command) + (pid \(listener.pid)); terminated + """ + self.logger.error("\(message, privacy: .public)") + } else { + self.logger.error("failed to terminate pid \(listener.pid) on port \(port, privacy: .public)") } } self.logger.info("port sweep done") } - func record(port: Int, pid: Int32, command: String, mode: AppState.ConnectionMode) async { - try? FileManager().createDirectory(at: Self.appSupportDir, withIntermediateDirectories: true) - self.records.removeAll { $0.pid == pid } - self.records.append( - Record( - port: port, - pid: pid, - command: command, - mode: mode.rawValue, - timestamp: Date().timeIntervalSince1970)) - self.save() + func record(port: Int, pid: Int32, command: String, mode: AppState.ConnectionMode) { + let record = Record( + port: port, + pid: pid, + command: command, + mode: mode.rawValue, + timestamp: Date().timeIntervalSince1970) + self.ownRecords[pid] = record + Self.persistRecords { $0[pid] = record } } func removeRecord(pid: Int32) { - let before = self.records.count - self.records.removeAll { $0.pid == pid } - if self.records.count != before { - self.save() + guard let owned = self.ownRecords.removeValue(forKey: pid) else { return } + // Content-guarded like reap: a sibling instance may have re-recorded a + // recycled pid for its own new tunnel; that fresh record must survive. + Self.persistRecords { merged in + if merged[pid] == owned { + merged.removeValue(forKey: pid) + } } } @@ -127,32 +126,34 @@ actor PortGuardian { /// leaves the tunnel reparented to launchd, holding the remote connection and /// squatting the preferred local port so new tunnels drift to ephemeral ports. func reapOrphanedTunnels() async { - let disk = Self.loadRecords(from: Self.recordPath) let plan = Self.planTunnelReap( - memory: self.records, - disk: disk, + own: Array(self.ownRecords.values), + disk: Self.loadRecords(from: Self.recordPath), processInfo: Self.tunnelProcessInfo(pid:)) - var kept = plan.keep + var removals = plan.drop for record in plan.reap { - if await self.reapTunnelProcess(record.pid) { + if await Self.terminateProcess(record.pid) { + removals.append(record) let message = """ reaped orphaned ssh tunnel (pid \(record.pid), local port \(record.port)) """ self.logger.error("\(message, privacy: .public)") } else { - // Keep the record so the next sweep retries the kill. - kept.append(record) + // Leave the record in place so the next sweep retries the kill. self.logger.error("failed to reap orphaned tunnel pid \(record.pid, privacy: .public)") } } - // Persist whenever the kept records differ from either source by content, not - // just pid set: a reaped/dropped record may exist only in the file (crashed - // sibling), and a pid-colliding stale disk record must be overwritten too. - let byTime: (Record, Record) -> Bool = { ($0.timestamp, $0.pid) < ($1.timestamp, $1.pid) } - kept.sort(by: byTime) - if kept != self.records.sorted(by: byTime) || kept != disk.sorted(by: byTime) { - self.records = kept - self.save() + guard !removals.isEmpty else { return } + for record in removals where self.ownRecords[record.pid] == record { + self.ownRecords.removeValue(forKey: record.pid) + } + // Remove only records whose content still matches what was classified: a + // sibling instance may have re-recorded the same pid for a new tunnel in the + // meantime, and that fresh record must survive. + Self.persistRecords { merged in + for record in removals where merged[record.pid] == record { + merged.removeValue(forKey: record.pid) + } } } @@ -194,40 +195,47 @@ actor PortGuardian { } static func planTunnelReap( - memory: [Record], + own: [Record], disk: [Record], - processInfo: (Int32) -> TunnelProcessInfo?) -> (reap: [Record], keep: [Record]) + processInfo: (Int32) -> TunnelProcessInfo?) -> (reap: [Record], keep: [Record], drop: [Record]) { // All app instances share port-guard.json, so disk can hold records from a - // crashed sibling this instance never saw. In-memory wins on pid collisions. + // crashed sibling this instance never saw. Own records win pid collisions — + // they are fresher than anything a dead sibling left behind. var merged: [Int32: Record] = [:] for record in disk { merged[record.pid] = record } - for record in memory { + for record in own { merged[record.pid] = record } let ordered = merged.values.sorted { ($0.timestamp, $0.pid) < ($1.timestamp, $1.pid) } var reap: [Record] = [] var keep: [Record] = [] + var drop: [Record] = [] for record in ordered { switch self.classifyTunnelRecord(record, process: processInfo(record.pid)) { case .keep: keep.append(record) - case .drop: continue + case .drop: drop.append(record) case .reap: reap.append(record) } } - return (reap, keep) + return (reap, keep, drop) } - /// Only a confirmed exit may drop the record (later sweeps retry otherwise), and - /// callers pick a local port right after reaping — the port is only free once ssh - /// has actually exited, not when the signal was delivered. TERM first, then KILL. - private func reapTunnelProcess(_ pid: Int32) async -> Bool { - _ = await ShellExecutor.run(command: ["kill", "-TERM", "\(pid)"], cwd: nil, env: nil, timeout: 2) - if await Self.waitForProcessExit(pid: pid) { return true } - _ = await ShellExecutor.run(command: ["kill", "-KILL", "\(pid)"], cwd: nil, env: nil, timeout: 2) - return await Self.waitForProcessExit(pid: pid) + /// TERM first, then KILL, returning only once the process is confirmed gone: + /// sweep and reap callers rebind the freed port immediately, and reap must not + /// forget a still-running tunnel (the record is its only retry path). + private static func terminateProcess(_ pid: Int32) async -> Bool { + #if canImport(Darwin) + guard pid > 0 else { return false } + _ = Darwin.kill(pid, SIGTERM) + if await self.waitForProcessExit(pid: pid) { return true } + _ = Darwin.kill(pid, SIGKILL) + return await self.waitForProcessExit(pid: pid) + #else + return false + #endif } private static func waitForProcessExit(pid: Int32, timeout: TimeInterval = 1.0) async -> Bool { @@ -325,23 +333,17 @@ actor PortGuardian { if mode == .unconfigured { return [] } - let ports = [GatewayEnvironment.gatewayPort()] - var reports: [PortReport] = [] - - for port in ports { - let listeners = await self.listeners(on: port) - let tunnelHealthy = await self.probeGatewayHealthIfNeeded( - port: port, - mode: mode, - listeners: listeners) - reports.append(Self.buildReport( - port: port, - listeners: listeners, - mode: mode, - tunnelHealthy: tunnelHealthy)) - } - - return reports + let port = GatewayEnvironment.gatewayPort() + let listeners = await self.listeners(on: port) + let tunnelHealthy = await self.probeGatewayHealthIfNeeded( + port: port, + mode: mode, + listeners: listeners) + return [Self.buildReport( + port: port, + listeners: listeners, + mode: mode, + tunnelHealthy: tunnelHealthy)] } func probeGatewayHealth(port: Int, timeout: TimeInterval = 2.0) async -> Bool { @@ -518,13 +520,6 @@ actor PortGuardian { #endif } - private func kill(_ pid: Int32) async -> Bool { - let term = await ShellExecutor.run(command: ["kill", "-TERM", "\(pid)"], cwd: nil, env: nil, timeout: 2) - if term.ok { return true } - let sigkill = await ShellExecutor.run(command: ["kill", "-KILL", "\(pid)"], cwd: nil, env: nil, timeout: 2) - return sigkill.ok - } - private static func isExpected(_ listener: Listener, port: Int, mode: AppState.ConnectionMode) -> Bool { let cmd = listener.command.lowercased() let full = listener.fullCommand.lowercased() @@ -565,9 +560,41 @@ actor PortGuardian { return decoded } - private func save() { - guard let data = try? JSONEncoder().encode(self.records) else { return } - try? data.write(to: Self.recordPath, options: [.atomic]) + /// The single write path for port-guard.json: read-merge-write under a file + /// lock so concurrent app instances (prod + dev) never clobber each other's + /// records — a lost record means an unreapable orphan after a crash. + private static func persistRecords(_ mutate: (inout [Int32: Record]) -> Void) { + try? FileManager().createDirectory(at: self.appSupportDir, withIntermediateDirectories: true) + self.withRecordFileLock { + let disk = self.loadRecords(from: self.recordPath) + var merged: [Int32: Record] = [:] + for record in disk { + merged[record.pid] = record + } + mutate(&merged) + let ordered = merged.values.sorted { ($0.timestamp, $0.pid) < ($1.timestamp, $1.pid) } + guard ordered != disk else { return } + guard let data = try? JSONEncoder().encode(ordered) else { return } + try? data.write(to: self.recordPath, options: [.atomic]) + } + } + + /// flock on a sidecar (atomic writes swap the json inode, so locking the json + /// itself would not serialize anything). The kernel releases it if we die. + private static func withRecordFileLock(_ body: () -> Void) { + #if canImport(Darwin) + let fd = open(self.recordLockPath.path, O_CREAT | O_RDWR, 0o644) + guard fd >= 0 else { + body() + return + } + defer { close(fd) } + _ = flock(fd, LOCK_EX) + defer { _ = flock(fd, LOCK_UN) } + body() + #else + body() + #endif } } @@ -580,11 +607,7 @@ extension PortGuardian { self.testingDescriptors.removeValue(forKey: port) } } -} -#endif -#if DEBUG -extension PortGuardian { static func _testTunnelProcessInfo(pid: Int32) -> TunnelProcessInfo? { self.tunnelProcessInfo(pid: pid) } diff --git a/apps/macos/Sources/OpenClaw/RemotePortTunnel.swift b/apps/macos/Sources/OpenClaw/RemotePortTunnel.swift index 5b9e0e925e2e..aee361e30ba3 100644 --- a/apps/macos/Sources/OpenClaw/RemotePortTunnel.swift +++ b/apps/macos/Sources/OpenClaw/RemotePortTunnel.swift @@ -49,18 +49,23 @@ final class RemotePortTunnel: @unchecked Sendable { } deinit { - Self.cleanupStderr(self.stderrHandle) - let pid = self.process.processIdentifier - self.process.terminate() - Task { await PortGuardian.shared.removeRecord(pid: pid) } + self.teardown(waitUntilExit: false) } func terminate() { + self.teardown(waitUntilExit: true) + } + + /// deinit cannot block on waitUntilExit; explicit terminate() must, so callers + /// can rebind the tunnel's local port immediately after it returns. + private func teardown(waitUntilExit: Bool) { Self.cleanupStderr(self.stderrHandle) let pid = self.process.processIdentifier if self.process.isRunning { self.process.terminate() - self.process.waitUntilExit() + if waitUntilExit { + self.process.waitUntilExit() + } } Task { await PortGuardian.shared.removeRecord(pid: pid) } } @@ -147,14 +152,13 @@ final class RemotePortTunnel: @unchecked Sendable { stderrHandle: stderrHandle, stderrCapture: stderrCapture) - // Track tunnel so we can clean up stale listeners on restart. - Task { - await PortGuardian.shared.record( - port: Int(localPort), - pid: process.processIdentifier, - command: process.executableURL?.path ?? "ssh", - mode: CommandResolver.connectionSettings().mode) - } + // Record before returning: a crash inside this window would otherwise leave + // an untracked (unreapable) orphan. + await PortGuardian.shared.record( + port: Int(localPort), + pid: process.processIdentifier, + command: process.executableURL?.path ?? "ssh", + mode: CommandResolver.connectionSettings().mode) return RemotePortTunnel(process: process, localPort: localPort, stderrHandle: stderrHandle) } @@ -367,10 +371,6 @@ final class RemotePortTunnel: @unchecked Sendable { try? handle.close() } - private static func drainStderr(_ handle: FileHandle) -> String { - self.drainStderr(handle, captured: "") - } - private static func drainStderr(_ handle: FileHandle, captured: String) -> String { handle.readabilityHandler = nil defer { try? handle.close() } @@ -410,7 +410,7 @@ final class RemotePortTunnel: @unchecked Sendable { } static func _testDrainStderr(_ handle: FileHandle) -> String { - self.drainStderr(handle) + self.drainStderr(handle, captured: "") } #endif diff --git a/apps/macos/Tests/OpenClawIPCTests/LowCoverageHelperTests.swift b/apps/macos/Tests/OpenClawIPCTests/LowCoverageHelperTests.swift index 156133d6cc89..8b9fae39aa51 100644 --- a/apps/macos/Tests/OpenClawIPCTests/LowCoverageHelperTests.swift +++ b/apps/macos/Tests/OpenClawIPCTests/LowCoverageHelperTests.swift @@ -260,13 +260,13 @@ struct LowCoverageHelperTests { // pid 10: our live tunnel (parent alive). Disk-only records from a crashed // sibling instance: pid 20 orphaned, pid 30 already gone. - let memory = [record(pid: 10, port: 18790, timestamp: 300)] + let own = [record(pid: 10, port: 18790, timestamp: 300)] let disk = [ record(pid: 20, port: 18789, timestamp: 100), record(pid: 30, port: 18791, timestamp: 200), - record(pid: 10, port: 1, timestamp: 1), // superseded by the in-memory record + record(pid: 10, port: 1, timestamp: 1), // superseded by the own record ] - let plan = PortGuardian.planTunnelReap(memory: memory, disk: disk, processInfo: { pid in + let plan = PortGuardian.planTunnelReap(own: own, disk: disk, processInfo: { pid in switch pid { case 10: .init(parentPid: 987, startedAt: 299, fullCommand: tunnel(port: 18790)) case 20: .init(parentPid: 1, startedAt: 99, fullCommand: tunnel(port: 18789)) @@ -276,6 +276,9 @@ struct LowCoverageHelperTests { #expect(plan.reap.map(\.pid) == [20]) #expect(plan.keep.map(\.pid) == [10]) #expect(plan.keep.first?.port == 18790) + // The dead pid 30 is reported as a drop; the shadowed pid-10 disk record is + // superseded, not dropped, so a reap cycle cannot delete the fresh record. + #expect(plan.drop.map(\.pid) == [30]) } @Test func `port guardian classifies a real orphaned tunnel process for reaping`() async throws {