import CryptoKit import Foundation import OpenClawChatUI import OpenClawKit import OpenClawProtocol import OSLog private let gatewayConnectionLogger = Logger(subsystem: "ai.openclaw", category: "gateway.connection") /// Single, shared Gateway websocket connection for the whole app. /// /// This owns exactly one `GatewayChannelActor` and reuses it across all callers /// (ControlChannel, debug actions, SwiftUI WebChat, etc.). actor GatewayConnection { static let shared = GatewayConnection( endpointProvider: GatewayConnection.defaultEndpointProvider) nonisolated static let operatorClientCaps = [OpenClawGatewayClientCapability.agentKind, OpenClawGatewayClientCapability.inlineWidgets] typealias Config = (url: URL, token: String?, password: String?) struct EndpointSnapshot { let config: Config let tls: GatewayTLSRoute? let routeAuthority: UInt64? let deviceAuthGatewayID: String? let revision: UInt64? init( config: Config, tls: GatewayTLSRoute? = nil, routeAuthority: UInt64?, deviceAuthGatewayID: String? = nil, revision: UInt64? = nil) { self.config = config self.tls = tls self.routeAuthority = routeAuthority self.deviceAuthGatewayID = deviceAuthGatewayID self.revision = revision } } typealias EndpointProvider = @Sendable () async throws -> EndpointSnapshot typealias SessionProvider = @Sendable (GatewayTLSRoute?) -> WebSocketSessionBox? struct Route: Equatable, Sendable { fileprivate let generation: UInt64 fileprivate let authority: UInt64? let url: URL fileprivate let token: String? fileprivate let password: String? let tls: GatewayTLSRoute? fileprivate let deviceAuthGatewayID: String? let activationOwnershipFingerprint: String? fileprivate func matches(_ endpoint: EndpointSnapshot) -> Bool { self.authority == endpoint.routeAuthority && self.url == endpoint.config.url && self.token == endpoint.config.token && self.password == endpoint.config.password && GatewayTLSRoute.hasSameConnectionIdentity(self.tls, endpoint.tls) && self.deviceAuthGatewayID == endpoint.deviceAuthGatewayID } } /// One connected Gateway server, not merely an endpoint configuration. /// A reconnect at the same URL creates a different lease. struct ServerLease: Sendable { // Managed-image HTTP reuses this captured route from its focused extension file. // Carrying the snapshot forward prevents endpoint or TLS rediscovery after suspension. let route: Route fileprivate let socketGeneration: UInt64 fileprivate let client: GatewayChannelActor } enum Method: String { case agent case status case setHeartbeats = "set-heartbeats" case systemEvent = "system-event" case health case channelsStatus = "channels.status" case configGet = "config.get" case configSet = "config.set" case configPatch = "config.patch" case configSchema = "config.schema" case configSchemaLookup = "config.schema.lookup" case wizardStart = "wizard.start" case wizardNext = "wizard.next" case wizardCancel = "wizard.cancel" case wizardStatus = "wizard.status" case talkConfig = "talk.config" case talkMode = "talk.mode" case talkSpeak = "talk.speak" case webLoginStart = "web.login.start" case webLoginWait = "web.login.wait" case channelsLogout = "channels.logout" case modelsList = "models.list" case agentsList = "agents.list" case agentIdentityGet = "agent.identity.get" case chatHistory = "chat.history" case sessionsPreview = "sessions.preview" case chatSend = "chat.send" case skillsStatus = "skills.status" case skillsSearch = "skills.search" case skillsDetail = "skills.detail" case skillsInstall = "skills.install" case skillsUpdate = "skills.update" case voicewakeGet = "voicewake.get" case voicewakeSet = "voicewake.set" case nodePairApprove = "node.pair.approve" case nodePairReject = "node.pair.reject" case devicePairList = "device.pair.list" case devicePairApprove = "device.pair.approve" case devicePairReject = "device.pair.reject" case execApprovalResolve = "exec.approval.resolve" case approvalResolve = "approval.resolve" case cronList = "cron.list" case cronRuns = "cron.runs" case cronRun = "cron.run" case cronRemove = "cron.remove" case cronUpdate = "cron.update" case cronAdd = "cron.add" case cronStatus = "cron.status" } private let endpointProvider: EndpointProvider private let supportsSharedEndpointRecovery: Bool private let activationBindingKeyProvider: @Sendable () -> SymmetricKey? private let includeDeviceIdentity: Bool private let sessionProvider: SessionProvider private let clientShutdown: @Sendable (GatewayChannelActor) async -> Void private let decoder = JSONDecoder() private struct ConfiguredConnection { let client: GatewayChannelActor let endpoint: EndpointSnapshot let tlsMetadataProvider: (any GatewayTLSRouteMetadataProviding)? let shutdownGeneration: UInt64 let activationBindingKey: SymmetricKey? func matches(endpoint: EndpointSnapshot, shutdownGeneration: UInt64? = nil) -> Bool { self.endpoint.config.url == endpoint.config.url && self.endpoint.config.token == endpoint.config.token && self.endpoint.config.password == endpoint.config.password && GatewayTLSRoute.hasSameConnectionIdentity(self.endpoint.tls, endpoint.tls) && self.endpoint.deviceAuthGatewayID == endpoint.deviceAuthGatewayID && self.endpoint.routeAuthority == endpoint.routeAuthority && shutdownGeneration.map { self.shutdownGeneration == $0 } ?? true } func matches(route: Route) -> Bool { route.matches(self.endpoint) } } private var configuredConnection: ConfiguredConnection? private var highestEndpointRevision: UInt64? private var routeGeneration: UInt64 = 0 /// Unbound operations capture this before their first suspension. Shutdown /// advances it so delayed config and retry work cannot recreate a route. private var shutdownGeneration: UInt64 = 0 // Callback work keeps the physical socket epoch that decoded it. Retiring // that epoch prevents delayed pushes from entering a replacement socket. private var activeSocketGeneration: UInt64? private var lastRetiredSocketGeneration: UInt64? private var subscribers: [UUID: AsyncStream.Continuation] = [:] private var lastSnapshot: HelloOk? var canvasPluginSurfaceURL: String? struct CanvasPluginSurfaceRefresh { let id: UUID let task: Task } var canvasPluginSurfaceRefresh: CanvasPluginSurfaceRefresh? init( endpointProvider: @escaping EndpointProvider = GatewayConnection.defaultEndpointProvider, supportsSharedEndpointRecovery: Bool = true, activationBindingKeyProvider: @escaping @Sendable () -> SymmetricKey? = GatewayConnection.defaultActivationBindingKey, sessionBox: WebSocketSessionBox? = nil, sessionProvider: SessionProvider? = nil, clientShutdown: @escaping @Sendable (GatewayChannelActor) async -> Void = { client in await client.shutdown() }) { self.endpointProvider = endpointProvider self.supportsSharedEndpointRecovery = supportsSharedEndpointRecovery self.activationBindingKeyProvider = activationBindingKeyProvider self.includeDeviceIdentity = true self.sessionProvider = Self.resolveSessionProvider( sessionBox: sessionBox, sessionProvider: sessionProvider) self.clientShutdown = clientShutdown } #if DEBUG private static let testingActivationBindingKey = SymmetricKey(size: .bits256) init( configProvider: @escaping @Sendable () async throws -> Config, activationBindingKeyProvider: @escaping @Sendable () -> SymmetricKey? = { GatewayConnection.testingActivationBindingKey }, sessionBox: WebSocketSessionBox? = nil, sessionProvider: SessionProvider? = nil, clientShutdown: @escaping @Sendable (GatewayChannelActor) async -> Void = { client in await client.shutdown() }) { self.endpointProvider = { try await EndpointSnapshot(config: configProvider(), routeAuthority: nil) } self.supportsSharedEndpointRecovery = false self.activationBindingKeyProvider = activationBindingKeyProvider // Mock WebSocket routes do not exercise device authentication and must not // depend on the process-global persisted identity store. self.includeDeviceIdentity = false self.sessionProvider = Self.resolveSessionProvider( sessionBox: sessionBox, sessionProvider: sessionProvider) self.clientShutdown = clientShutdown } init( testEndpointProvider: @escaping EndpointProvider, sessionBox: WebSocketSessionBox? = nil) { self.endpointProvider = testEndpointProvider self.supportsSharedEndpointRecovery = false self.activationBindingKeyProvider = { GatewayConnection.testingActivationBindingKey } // Mock WebSocket routes do not exercise device authentication and must not // depend on the process-global persisted identity store. self.includeDeviceIdentity = false self.sessionProvider = Self.resolveSessionProvider( sessionBox: sessionBox, sessionProvider: nil) self.clientShutdown = { client in await client.shutdown() } } #endif private static func resolveSessionProvider( sessionBox: WebSocketSessionBox?, sessionProvider: SessionProvider?) -> SessionProvider { if let sessionProvider { return sessionProvider } if let sessionBox { return { _ in sessionBox } } return { route in route.map { WebSocketSessionBox(session: GatewayTLSPinningSession(params: $0.params)) } } } // MARK: - Low-level request func request( method: String, params: [String: AnyCodable]?, timeoutMs: Double? = nil, retryTransportFailures: Bool = true) async throws -> Data { try await self.request( method: method, params: params, timeoutMs: timeoutMs, retryTransportFailures: retryTransportFailures, allowTLSRepair: true) } private func request( method: String, params: [String: AnyCodable]?, timeoutMs: Double?, retryTransportFailures: Bool, allowTLSRepair: Bool) async throws -> Data { let shutdownGeneration = shutdownGeneration let endpoint = try await currentEndpoint() let cfg = endpoint.config let client = try await configure( endpoint: endpoint, shutdownGeneration: shutdownGeneration) do { return try await client.request(method: method, params: params, timeoutMs: timeoutMs) } catch { if allowTLSRepair, let tlsError = error as? GatewayTLSValidationError, await GatewayTLSRepairCoordinator.shared.repair( route: endpoint.tls, url: cfg.url, failure: tlsError.failure) { return try await self.request( method: method, params: params, timeoutMs: timeoutMs, retryTransportFailures: retryTransportFailures, allowTLSRepair: false) } if !retryTransportFailures || error is GatewayResponseError || error is GatewayDecodingError { throw error } try requireCurrentShutdownGeneration(shutdownGeneration) // Profile-bound windows own a fixed endpoint. Shared recovery reads global // connection-mode state and may legitimately retarget only the primary app route. guard self.supportsSharedEndpointRecovery else { throw error } // Auto-recover in local mode by spawning/attaching a gateway and retrying a few times. // Canvas interactions should "just work" even if the local gateway isn't running yet. let mode = await MainActor.run { AppStateStore.shared.connectionMode } try requireCurrentShutdownGeneration(shutdownGeneration) switch mode { case .local: await MainActor.run { GatewayProcessManager.shared.setActive(true) } try requireCurrentShutdownGeneration(shutdownGeneration) let lastError: Error do { return try await self.retryRequest( client: client, method: method, params: params, timeoutMs: timeoutMs, after: error, shutdownGeneration: shutdownGeneration) } catch { lastError = error } let nsError = lastError as NSError if nsError.domain == URLError.errorDomain, let fallback = await GatewayEndpointStore.shared.maybeFallbackToTailnet(from: cfg.url) { try acceptEndpointRevision(fallback) let fallbackClient = try await configure( endpoint: fallback, shutdownGeneration: shutdownGeneration) return try await self.retryRequest( client: fallbackClient, method: method, params: params, timeoutMs: timeoutMs, after: lastError, shutdownGeneration: shutdownGeneration) } throw lastError case .remote: let nsError = error as NSError guard nsError.domain == URLError.errorDomain else { throw error } var lastError: Error = error await RemoteTunnelManager.shared.stopAll() try requireCurrentShutdownGeneration(shutdownGeneration) do { _ = try await GatewayEndpointStore.shared.ensureRemoteControlTunnel() try requireCurrentShutdownGeneration(shutdownGeneration) } catch { try requireCurrentShutdownGeneration(shutdownGeneration) lastError = error } for delayMs in [150, 400, 900] { try await Task.sleep(nanoseconds: UInt64(delayMs) * 1_000_000) try requireCurrentShutdownGeneration(shutdownGeneration) do { let endpoint = try await currentEndpoint() let client = try await configure( endpoint: endpoint, shutdownGeneration: shutdownGeneration) return try await client.request(method: method, params: params, timeoutMs: timeoutMs) } catch { try requireCurrentShutdownGeneration(shutdownGeneration) lastError = error } } try requireCurrentShutdownGeneration(shutdownGeneration) throw lastError case .unconfigured: throw error } } } private func retryRequest( client: GatewayChannelActor, method: String, params: [String: AnyCodable]?, timeoutMs: Double?, after initialError: Error, shutdownGeneration: UInt64) async throws -> Data { var lastError = initialError for delayMs in [150, 400, 900] { try await Task.sleep(nanoseconds: UInt64(delayMs) * 1_000_000) try requireCurrentShutdownGeneration(shutdownGeneration) do { return try await client.request(method: method, params: params, timeoutMs: timeoutMs) } catch { try requireCurrentShutdownGeneration(shutdownGeneration) lastError = error } } throw lastError } /// Route-bound requests never reconfigure or retry on another endpoint. /// Durable outbox rows must remain owned by the gateway that created them. func request( method: String, params: [String: AnyCodable]?, timeoutMs: Double? = nil, ifCurrentRoute route: Route, distinguishPreDispatchRouteChange: Bool = false) async throws -> Data { let endpoint = try await currentEndpoint() guard self.routeMatchesCurrentState(route, endpoint: endpoint), let client = self.configuredConnection?.client else { if distinguishPreDispatchRouteChange { throw OpenClawChatTransportSendError.notDispatched } throw CancellationError() } let data = try await client.request(method: method, params: params, timeoutMs: timeoutMs) guard await self.isCurrentRoute(route), self.configuredConnection?.client === client else { throw CancellationError() } return data } /// Server-bound requests never reconfigure, reconnect, or cross onto a /// replacement socket at the same endpoint. func request( method: String, params: [String: AnyCodable]?, timeoutMs: Double? = nil, ifCurrentServerLease lease: ServerLease) async throws -> Data { guard await isCurrentServerLease(lease) else { throw OpenClawChatTransportSendError.notDispatched } do { let data = try await lease.client.request( method: method, params: params, timeoutMs: timeoutMs, ifCurrentConnectionGeneration: lease.socketGeneration) guard await self.isCurrentServerLease(lease) else { throw OpenClawChatTransportSendError.notDispatched } return data } catch is CancellationError { if Task.isCancelled { throw CancellationError() } throw OpenClawChatTransportSendError.notDispatched } } } extension GatewayConnection { enum WizardCancellationOutcome: Equatable { case cancelled case absent case unresolved } /// Cancel on the socket that created the wizard, or a replacement socket on /// the same route. Absence stays distinct so callers can reconcile a commit. @discardableResult func cancelWizardSession( _ sessionID: String, on lease: ServerLease) async -> WizardCancellationOutcome { let initial = await sendWizardCancellation(sessionID, on: lease) if initial != .unresolved { return initial } guard let replacement = try? await acquireServerLease( ifSameRouteAs: lease, timeoutMs: 5000) else { return .unresolved } return await self.sendWizardCancellation(sessionID, on: replacement) } private func sendWizardCancellation( _ sessionID: String, on lease: ServerLease) async -> WizardCancellationOutcome { do { let data = try await lease.client.request( method: "wizard.cancel", params: ["sessionId": AnyCodable(sessionID)], timeoutMs: 10000, ifCurrentConnectionGeneration: lease.socketGeneration) let status = try decoder.decode(WizardCancellationStatus.self, from: data) return switch status.status { case "cancelled": .cancelled case "running": .unresolved default: .absent } } catch { return Self.wizardCancellationOutcome(after: error) } } static func wizardCancellationOutcome(after error: Error) -> WizardCancellationOutcome { guard let response = error as? GatewayResponseError else { return .unresolved } let sessionIsAbsent = response.method == "wizard.cancel" && response.code == "INVALID_REQUEST" && response.message == "wizard not found" return sessionIsAbsent ? .absent : .unresolved } private struct WizardCancellationStatus: Decodable { let status: String } func requestRaw( method: Method, params: [String: AnyCodable]? = nil, timeoutMs: Double? = nil) async throws -> Data { try await self.request(method: method.rawValue, params: params, timeoutMs: timeoutMs) } func request( _ request: OpenClawChatGatewayRequest, retryTransportFailures: Bool = true) async throws -> Data { try await self.request( method: request.method, params: request.params, timeoutMs: request.timeoutMs, retryTransportFailures: retryTransportFailures) } func request( _ request: OpenClawChatGatewayRequest, ifCurrentRoute route: Route, distinguishPreDispatchRouteChange: Bool = false) async throws -> Data { try await self.request( method: request.method, params: request.params, timeoutMs: request.timeoutMs, ifCurrentRoute: route, distinguishPreDispatchRouteChange: distinguishPreDispatchRouteChange) } func requestRaw( method: String, params: [String: AnyCodable]? = nil, timeoutMs: Double? = nil) async throws -> Data { try await self.request(method: method, params: params, timeoutMs: timeoutMs) } func requestDecoded( method: Method, params: [String: AnyCodable]? = nil, timeoutMs: Double? = nil) async throws -> T { let data = try await requestRaw(method: method, params: params, timeoutMs: timeoutMs) do { return try self.decoder.decode(T.self, from: data) } catch { throw GatewayDecodingError(method: method.rawValue, message: error.localizedDescription) } } func requestDecoded( method: Method, params: [String: AnyCodable]? = nil, timeoutMs: Double? = nil, ifCurrentRoute route: Route) async throws -> T { let data = try await self.request( method: method.rawValue, params: params, timeoutMs: timeoutMs, ifCurrentRoute: route, distinguishPreDispatchRouteChange: true) guard await self.isCurrentRoute(route) else { throw GatewayRouteChangedAfterDispatchError(method: method.rawValue) } do { return try self.decoder.decode(T.self, from: data) } catch { throw GatewayDecodingError(method: method.rawValue, message: error.localizedDescription) } } func requestVoid( method: Method, params: [String: AnyCodable]? = nil, timeoutMs: Double? = nil) async throws { _ = try await self.requestRaw(method: method, params: params, timeoutMs: timeoutMs) } /// Ensure the underlying socket is configured (and replaced if config changed). func refresh() async throws { let shutdownGeneration = shutdownGeneration let endpoint = try await currentEndpoint() _ = try await self.configure( endpoint: endpoint, shutdownGeneration: shutdownGeneration) } func captureRoute() async -> Route? { try? await self.captureRequiredRoute() } func captureRequiredRoute() async throws -> Route { let shutdownGeneration = shutdownGeneration let endpoint = try await currentEndpoint() let cfg = endpoint.config _ = try await self.configure( endpoint: endpoint, shutdownGeneration: shutdownGeneration) return Route( generation: self.routeGeneration, authority: endpoint.routeAuthority, url: cfg.url, token: cfg.token, password: cfg.password, tls: endpoint.tls, deviceAuthGatewayID: endpoint.deviceAuthGatewayID, activationOwnershipFingerprint: Self.activationOwnershipFingerprint( config: cfg, key: self.configuredConnection?.activationBindingKey)) } /// Connect and bind subsequent work to the hello snapshot's physical /// socket. The read-only health preflight intentionally uses the ordinary /// recovery path so a fresh local Gateway can start and a remote tunnel can /// recover before onboarding freezes the successful physical connection. func acquireServerLease() async throws -> ServerLease { try await self.acquireServerLease(timeoutMs: 15000, retryTransportFailures: true) } /// Captures the currently connected physical socket without probing or /// reconnecting. Queued mutations use this so waiting cannot retarget them. func captureServerLease() async -> ServerLease? { guard let route = await self.captureRoute(), let client = self.configuredConnection?.client, let socketGeneration = self.activeSocketGeneration else { return nil } let lease = ServerLease(route: route, socketGeneration: socketGeneration, client: client) guard await self.isCurrentServerLease(lease) else { return nil } return lease } private func acquireServerLease( timeoutMs: Double, retryTransportFailures: Bool) async throws -> ServerLease { let shutdownGeneration = self.shutdownGeneration _ = try await self.request( method: Method.health.rawValue, params: nil, timeoutMs: timeoutMs, retryTransportFailures: retryTransportFailures) try self.requireCurrentShutdownGeneration(shutdownGeneration) let endpoint = try await currentEndpoint() let cfg = endpoint.config guard let client = configuredClient( endpoint: endpoint, shutdownGeneration: shutdownGeneration) else { throw OpenClawChatTransportSendError.notDispatched } guard let socketGeneration = await client.currentConnectionGeneration() else { throw OpenClawChatTransportSendError.notDispatched } guard let authBinding = await client.authBinding( ifCurrentConnectionGeneration: socketGeneration) else { throw OpenClawChatTransportSendError.notDispatched } let lease = ServerLease( route: Route( generation: routeGeneration, authority: endpoint.routeAuthority, url: cfg.url, token: cfg.token, password: cfg.password, tls: endpoint.tls, deviceAuthGatewayID: endpoint.deviceAuthGatewayID, activationOwnershipFingerprint: Self.activationOwnershipFingerprint( config: cfg, authBinding: authBinding, key: self.configuredConnection?.activationBindingKey)), socketGeneration: socketGeneration, client: client) guard await self.isCurrentServerLease(lease) else { throw OpenClawChatTransportSendError.notDispatched } return lease } /// Reconnect one physical socket without crossing the route that owned the /// original request. Endpoint or credential changes must start fresh work. func acquireServerLease( ifSameRouteAs previous: ServerLease, timeoutMs: Double) async throws -> ServerLease { guard await self.isCurrentRoute(previous.route) else { throw OpenClawChatTransportSendError.notDispatched } // Restart reconciliation owns its overall deadline. One short, non-retrying // health request keeps a failed attempt from consuming the whole budget. let replacement = try await acquireServerLease( timeoutMs: timeoutMs, retryTransportFailures: false) guard replacement.route.generation == previous.route.generation, replacement.route.url == previous.route.url, replacement.route.token == previous.route.token, replacement.route.password == previous.route.password, GatewayTLSRoute.hasSameConnectionIdentity(replacement.route.tls, previous.route.tls), replacement.route.deviceAuthGatewayID == previous.route.deviceAuthGatewayID else { throw OpenClawChatTransportSendError.notDispatched } return replacement } func isCurrentRoute(_ route: Route) async -> Bool { guard let endpoint = try? await currentEndpoint() else { return false } return self.routeMatchesCurrentState(route, endpoint: endpoint) } func supportsServerCapability( _ capability: GatewayServerCapability, ifCurrentRoute route: Route) async -> Bool? { guard let endpoint = try? await currentEndpoint() else { return nil } guard self.routeMatchesCurrentState(route, endpoint: endpoint), let snapshot = lastSnapshot else { return nil } return snapshot.supportsServerCapability(capability) } func supportsServerCapability( _ capability: GatewayServerCapability, ifCurrentServerLease lease: ServerLease) async -> Bool? { guard await self.isCurrentServerLease(lease), self.serverLeaseMatchesCurrentState(lease), let snapshot = lastSnapshot else { return nil } return snapshot.supportsServerCapability(capability) } func supportsServerMethod( _ method: String, ifCurrentServerLease lease: ServerLease) async -> Bool? { guard await self.isCurrentServerLease(lease), self.serverLeaseMatchesCurrentState(lease), let snapshot = lastSnapshot else { return nil } let methods = snapshot.features["methods"]?.value as? [AnyCodable] ?? [] return methods.contains { ($0.value as? String) == method } } func isCurrentServerLease(_ lease: ServerLease) async -> Bool { guard let endpoint = try? await currentEndpoint(), serverLeaseMatchesCurrentState(lease), lease.route.matches(endpoint), await lease.client.currentConnectionGeneration() == lease.socketGeneration, serverLeaseMatchesCurrentState(lease) else { return false } return true } func activationOwnershipFingerprint( ifCurrentServerLease lease: ServerLease) async -> String? { guard await self.isCurrentServerLease(lease) else { return nil } return lease.route.activationOwnershipFingerprint } private func serverLeaseMatchesCurrentState(_ lease: ServerLease) -> Bool { self.routeMatchesConfiguredConnection(lease.route) && self.configuredConnection?.client === lease.client && self.activeSocketGeneration == lease.socketGeneration && self.lastSnapshot != nil } private func routeMatchesCurrentState(_ route: Route, endpoint: EndpointSnapshot) -> Bool { route.matches(endpoint) && self.routeMatchesConfiguredConnection(route) } private func routeMatchesConfiguredConnection(_ route: Route) -> Bool { route.generation == self.routeGeneration && self.configuredConnection?.matches(route: route) == true } func sessionRoutingIdentity( ifCurrentRoute route: Route) async throws -> OpenClawChatSessionRoutingIdentity { let data = try await request( OpenClawChatGatewayRequests.agentsList(), ifCurrentRoute: route) return try OpenClawChatGatewayPayloadCodec.decodeSessionRoutingIdentity(data) } func configuredGatewayURL() -> URL? { self.configuredConnection?.endpoint.config.url } func configuredTLSFingerprintSHA256() -> String? { self.configuredConnection?.tlsMetadataProvider?.effectiveTLSFingerprintSHA256 } func authSource() async -> GatewayAuthSource? { guard let connection = self.configuredConnection else { return nil } return await connection.client.authSource() } func shutdown() async { self.shutdownGeneration &+= 1 let client = self.retireConfiguredConnection() if let client { await self.clientShutdown(client) } } private func configure( endpoint: EndpointSnapshot, shutdownGeneration: UInt64) async throws -> GatewayChannelActor { try self.requireCurrentShutdownGeneration(shutdownGeneration) let config = endpoint.config if let client = configuredClient( endpoint: endpoint, shutdownGeneration: shutdownGeneration) { return client } let previousClient = self.retireConfiguredConnection() let configuredRouteGeneration = self.routeGeneration if let previousClient { await self.clientShutdown(previousClient) } try self.requireCurrentShutdownGeneration(shutdownGeneration) if self.routeGeneration != configuredRouteGeneration { if let client = configuredClient( endpoint: endpoint, shutdownGeneration: shutdownGeneration) { return client } throw CancellationError() } let activationBindingKey = self.activationBindingKeyProvider() let sessionBox = self.sessionProvider(endpoint.tls) let client = GatewayChannelActor( url: config.url, token: config.token, password: config.password, authBindingKey: activationBindingKey, session: sessionBox, connectSnapshotAdmissionHandler: { [weak self] snapshot, socketGeneration in await self?.admitConnectSnapshot( snapshot, routeGeneration: configuredRouteGeneration, socketGeneration: socketGeneration) }, pushHandler: { [weak self] push, socketGeneration in await self?.handle( push: push, routeGeneration: configuredRouteGeneration, socketGeneration: socketGeneration) }, connectOptions: GatewayConnectOptions( role: "operator", scopes: GatewayChannelActor.defaultOperatorConnectScopes, caps: Self.operatorClientCaps, commands: [], permissions: [:], clientId: "openclaw-macos", clientMode: "ui", clientDisplayName: InstanceIdentity.displayName, includeDeviceIdentity: self.includeDeviceIdentity, allowStoredDeviceAuth: endpoint.deviceAuthGatewayID != nil, deviceAuthGatewayID: endpoint.deviceAuthGatewayID), disconnectHandler: { [weak self] _, socketGeneration in await self?.handleDisconnect( routeGeneration: configuredRouteGeneration, socketGeneration: socketGeneration) }) self.configuredConnection = ConfiguredConnection( client: client, endpoint: endpoint, tlsMetadataProvider: sessionBox?.session as? GatewayTLSRouteMetadataProviding, shutdownGeneration: shutdownGeneration, activationBindingKey: activationBindingKey) return client } private func configuredClient( endpoint: EndpointSnapshot, shutdownGeneration: UInt64) -> GatewayChannelActor? { guard let connection = self.configuredConnection, connection.matches( endpoint: endpoint, shutdownGeneration: shutdownGeneration) else { return nil } return connection.client } /// Invalidate every route-owned fact before shutdown can suspend; otherwise /// reentrant work could continue on a client whose replacement is in flight. private func retireConfiguredConnection() -> GatewayChannelActor? { self.routeGeneration &+= 1 self.resetSocketGeneration() self.lastSnapshot = nil self.resetCanvasPluginSurfaceState() let client = self.configuredConnection?.client self.configuredConnection = nil return client } private func requireCurrentShutdownGeneration(_ shutdownGeneration: UInt64) throws { guard self.shutdownGeneration == shutdownGeneration else { throw CancellationError() } } private func handle( push: GatewayPush, routeGeneration: UInt64, socketGeneration: UInt64) { guard routeGeneration == self.routeGeneration, admitSocketGeneration(socketGeneration) else { return } broadcast(push) } /// Short connect-path admission only. Subscriber delivery stays on the /// ordinary push task so connect never waits on downstream UI work. private func admitConnectSnapshot( _ snapshot: HelloOk, routeGeneration: UInt64, socketGeneration: UInt64) { guard routeGeneration == self.routeGeneration, admitSocketGeneration(socketGeneration) else { return } self.lastSnapshot = snapshot self.installCanvasPluginSurfaceURL(from: snapshot) } private func handleDisconnect(routeGeneration: UInt64, socketGeneration: UInt64) { guard routeGeneration == self.routeGeneration, retireSocketGeneration(socketGeneration) else { return } self.lastSnapshot = nil self.resetCanvasPluginSurfaceState() } } extension GatewayConnection { private func admitSocketGeneration(_ socketGeneration: UInt64) -> Bool { if let lastRetiredSocketGeneration, socketGeneration <= lastRetiredSocketGeneration { return false } if let activeSocketGeneration { return socketGeneration == activeSocketGeneration } activeSocketGeneration = socketGeneration return true } private func retireSocketGeneration(_ socketGeneration: UInt64) -> Bool { if let lastRetiredSocketGeneration, socketGeneration <= lastRetiredSocketGeneration { return false } if let activeSocketGeneration, socketGeneration != activeSocketGeneration { return false } activeSocketGeneration = nil lastRetiredSocketGeneration = socketGeneration return true } private func resetSocketGeneration() { self.activeSocketGeneration = nil self.lastRetiredSocketGeneration = nil } #if DEBUG func _test_routeGeneration() -> UInt64 { self.routeGeneration } func _test_configuredURL() -> URL? { self.configuredConnection?.endpoint.config.url } func _test_handlePush( _ push: GatewayPush, routeGeneration: UInt64? = nil, socketGeneration: UInt64) { self.handle( push: push, routeGeneration: routeGeneration ?? self.routeGeneration, socketGeneration: socketGeneration) } func _test_handleDisconnect( routeGeneration: UInt64? = nil, socketGeneration: UInt64) { self.handleDisconnect( routeGeneration: routeGeneration ?? self.routeGeneration, socketGeneration: socketGeneration) } #endif private static func defaultEndpointProvider() async throws -> EndpointSnapshot { try await GatewayEndpointStore.shared.requireEndpoint() } private func currentEndpoint() async throws -> EndpointSnapshot { let endpoint = try await endpointProvider() try acceptEndpointRevision(endpoint) return endpoint } private func acceptEndpointRevision(_ endpoint: EndpointSnapshot) throws { guard let revision = endpoint.revision else { return } if let highestEndpointRevision, revision < highestEndpointRevision { throw CancellationError() } if highestEndpointRevision.map({ revision > $0 }) ?? true { highestEndpointRevision = revision } } static func defaultActivationBindingKey() -> SymmetricKey? { GatewayActivationBindingKeyStore.loadOrCreate() } private static func activationOwnershipFingerprint( config: Config, authBinding: GatewayAuthBinding? = nil, key: SymmetricKey?) -> String? { guard let key else { return nil } // The durable record is already keyed by the stable Gateway route identity. // Bind only auth here so an SSH tunnel's ephemeral local URL can rebind safely. var values = [config.token ?? "", config.password ?? ""] if authBinding?.source == .deviceToken { guard let credentialFingerprint = authBinding?.credentialFingerprint else { return nil } values.append(contentsOf: [GatewayAuthSource.deviceToken.rawValue, credentialFingerprint]) } let framed = values.map { "\($0.utf8.count):\($0)" }.joined(separator: "|") let tag = HMAC.authenticationCode(for: Data(framed.utf8), using: key) return tag.map { String(format: "%02x", $0) }.joined() } } // MARK: - Snapshot cache and subscriptions extension GatewayConnection { func controlUiAutoAuthToken(config: Config) async -> String? { guard let endpoint = try? await currentEndpoint(), endpoint.config.url == config.url, endpoint.config.token == config.token, endpoint.config.password == config.password, let client = self.configuredConnection?.client, let socketGeneration = activeSocketGeneration, controlUiRouteIsLive( endpoint: endpoint, client: client, socketGeneration: socketGeneration), await client.currentConnectionGeneration() == socketGeneration, controlUiRouteIsLive( endpoint: endpoint, client: client, socketGeneration: socketGeneration), let authBinding = await client.authBinding( ifCurrentConnectionGeneration: socketGeneration), controlUiRouteIsLive( endpoint: endpoint, client: client, socketGeneration: socketGeneration) else { return nil } switch authBinding.source { case .sharedToken: return config.token?.trimmingCharacters(in: .whitespacesAndNewlines).nonEmpty case .deviceToken: guard let gatewayID = configuredConnection?.endpoint.deviceAuthGatewayID? .trimmingCharacters(in: .whitespacesAndNewlines).nonEmpty else { return nil } if let deviceToken = lastSnapshot?.auth["deviceToken"]?.value as? String, let token = deviceToken.trimmingCharacters(in: .whitespacesAndNewlines).nonEmpty { return token } guard let identity = DeviceIdentityStore.loadOrCreatePersisted() else { return nil } return DeviceAuthStore.loadToken( deviceId: identity.deviceId, role: "operator", gatewayID: gatewayID)?.token .trimmingCharacters(in: .whitespacesAndNewlines).nonEmpty case .bootstrapToken, .password, .none: return nil } } private func controlUiRouteIsLive( endpoint: EndpointSnapshot, client: GatewayChannelActor, socketGeneration: UInt64) -> Bool { self.configuredConnection?.matches( endpoint: endpoint, shutdownGeneration: self.shutdownGeneration) == true && self.configuredConnection?.client === client && self.activeSocketGeneration == socketGeneration && self.lastSnapshot != nil } private func sessionDefaultString(_ defaults: [String: OpenClawProtocol.AnyCodable]?, key: String) -> String { let raw = defaults?[key]?.value as? String return (raw ?? "").trimmingCharacters(in: CharacterSet.whitespacesAndNewlines) } func cachedMainSessionKey() -> String? { guard let snapshot = lastSnapshot else { return nil } let trimmed = self.sessionDefaultString(snapshot.snapshot.sessiondefaults, key: "mainSessionKey") return trimmed.isEmpty ? nil : trimmed } func cachedGatewayVersion() -> String? { guard let snapshot = lastSnapshot else { return nil } let raw = snapshot.server["version"]?.value as? String let trimmed = raw?.trimmingCharacters(in: CharacterSet.whitespacesAndNewlines) ?? "" return trimmed.isEmpty ? nil : trimmed } func cachedGatewayVersion(ifCurrentServerLease lease: ServerLease) async -> String? { guard await self.isCurrentServerLease(lease) else { return nil } return self.cachedGatewayVersion() } func snapshotPaths() -> (configPath: String?, stateDir: String?) { guard let snapshot = lastSnapshot else { return (nil, nil) } let configPath = snapshot.snapshot.configpath?.trimmingCharacters(in: .whitespacesAndNewlines) let stateDir = snapshot.snapshot.statedir?.trimmingCharacters(in: .whitespacesAndNewlines) return ( configPath?.isEmpty == false ? configPath : nil, stateDir?.isEmpty == false ? stateDir : nil) } func subscribe(bufferingNewest: Int = 100) -> AsyncStream { let id = UUID() let snapshot = self.lastSnapshot let connection = self return AsyncStream(bufferingPolicy: .bufferingNewest(bufferingNewest)) { continuation in if let snapshot { continuation.yield(.snapshot(snapshot)) } self.subscribers[id] = continuation continuation.onTermination = { @Sendable _ in Task { await connection.removeSubscriber(id) } } } } private func removeSubscriber(_ id: UUID) { self.subscribers[id] = nil } private func broadcast(_ push: GatewayPush) { if case let .snapshot(snapshot) = push { self.lastSnapshot = snapshot if self.canvasPluginSurfaceURL == nil { self.installCanvasPluginSurfaceURL(from: snapshot) } if let mainSessionKey = cachedMainSessionKey() { Task { @MainActor in WorkActivityStore.shared.setMainSessionKey(mainSessionKey) } } } for (_, continuation) in self.subscribers { continuation.yield(push) } } private func canonicalizeSessionKey(_ raw: String) -> String { let trimmed = raw.trimmingCharacters(in: CharacterSet.whitespacesAndNewlines) guard !trimmed.isEmpty else { return trimmed } guard let defaults = lastSnapshot?.snapshot.sessiondefaults else { return trimmed } let mainSessionKey = self.sessionDefaultString(defaults, key: "mainSessionKey") guard !mainSessionKey.isEmpty else { return trimmed } let mainKey = self.sessionDefaultString(defaults, key: "mainKey") let defaultAgentId = self.sessionDefaultString(defaults, key: "defaultAgentId") let isMainAlias = trimmed == "main" || (!mainKey.isEmpty && trimmed == mainKey) || trimmed == mainSessionKey || (!defaultAgentId.isEmpty && (trimmed == "agent:\(defaultAgentId):main" || (mainKey.isEmpty == false && trimmed == "agent:\(defaultAgentId):\(mainKey)"))) return isMainAlias ? mainSessionKey : trimmed } } // MARK: - Typed gateway API extension GatewayConnection { struct ConfigGetSnapshot: Decodable { struct SnapshotConfig: Decodable { struct Session: Decodable { let mainKey: String? let scope: String? } let session: Session? } let config: SnapshotConfig? } static func mainSessionKey(fromConfigGetData data: Data) throws -> String { let snapshot = try JSONDecoder().decode(ConfigGetSnapshot.self, from: data) let scope = snapshot.config?.session?.scope?.trimmingCharacters(in: .whitespacesAndNewlines) if scope == "global" { return "global" } return "main" } func mainSessionKey(timeoutMs: Double = 15000) async -> String { if let cached = cachedMainSessionKey() { return cached } return await self.refreshMainSessionKey(timeoutMs: timeoutMs) } func refreshMainSessionKey(timeoutMs: Double = 15000) async -> String { do { let data = try await requestRaw(method: "config.get", params: nil, timeoutMs: timeoutMs) return try Self.mainSessionKey(fromConfigGetData: data) } catch { return "main" } } func status() async -> (ok: Bool, error: String?) { do { _ = try await self.requestRaw(method: .status) return (true, nil) } catch { return (false, error.localizedDescription) } } func setHeartbeatsEnabled(_ enabled: Bool) async -> Bool { do { try await self.requestVoid(method: .setHeartbeats, params: ["enabled": AnyCodable(enabled)]) return true } catch { gatewayConnectionLogger.error("setHeartbeatsEnabled failed \(error.localizedDescription, privacy: .public)") return false } } func sendAgent(_ invocation: GatewayAgentInvocation) async -> (ok: Bool, error: String?) { let trimmed = invocation.message.trimmingCharacters(in: .whitespacesAndNewlines) guard !trimmed.isEmpty else { return (false, "message empty") } let sessionKey = self.canonicalizeSessionKey(invocation.sessionKey) var params: [String: AnyCodable] = [ "message": AnyCodable(trimmed), "sessionKey": AnyCodable(sessionKey), "deliver": AnyCodable(invocation.deliver), "to": AnyCodable(invocation.to ?? ""), "channel": AnyCodable(invocation.channel.rawValue), "idempotencyKey": AnyCodable(invocation.idempotencyKey), ] if let thinking = invocation.thinking?.trimmingCharacters(in: .whitespacesAndNewlines), !thinking.isEmpty { params["thinking"] = AnyCodable(thinking) } if let timeout = invocation.timeoutSeconds { params["timeout"] = AnyCodable(timeout) } if let trigger = invocation.voiceWakeTrigger { params["voiceWakeTrigger"] = AnyCodable( trigger.trimmingCharacters(in: .whitespacesAndNewlines)) } do { try await self.requestVoid(method: .agent, params: params) return (true, nil) } catch { return (false, error.localizedDescription) } } // MARK: - Health func healthOK(timeoutMs: Int = 8000) async throws -> Bool { let data = try await requestRaw(method: .health, timeoutMs: Double(timeoutMs)) return (try? self.decoder.decode(OpenClawGatewayHealthOK.self, from: data))?.ok ?? true } // MARK: - Skills func skillsStatus() async throws -> SkillsStatusReport { try await self.requestDecoded(method: .skillsStatus) } func skillsInstall( name: String, installId: String, dangerouslyForceUnsafeInstall: Bool? = nil, timeoutMs: Int? = nil) async throws -> SkillInstallResult { var params: [String: AnyCodable] = [ "name": AnyCodable(name), "installId": AnyCodable(installId), ] if let dangerouslyForceUnsafeInstall { params["dangerouslyForceUnsafeInstall"] = AnyCodable(dangerouslyForceUnsafeInstall) } if let timeoutMs { params["timeoutMs"] = AnyCodable(timeoutMs) } return try await self.requestDecoded(method: .skillsInstall, params: params) } func skillsUpdate( skillKey: String, enabled: Bool? = nil, apiKey: String? = nil, env: [String: String]? = nil) async throws -> SkillUpdateResult { var params: [String: AnyCodable] = [ "skillKey": AnyCodable(skillKey), ] if let enabled { params["enabled"] = AnyCodable(enabled) } if let apiKey { params["apiKey"] = AnyCodable(apiKey) } if let env, !env.isEmpty { params["env"] = AnyCodable(env) } return try await self.requestDecoded(method: .skillsUpdate, params: params) } // MARK: - Sessions func sessionsPreview( keys: [String], limit: Int? = nil, maxChars: Int? = nil, timeoutMs: Int? = nil) async throws -> OpenClawSessionsPreviewPayload { let resolvedKeys = keys .map { self.canonicalizeSessionKey($0) } .filter { !$0.isEmpty } if resolvedKeys.isEmpty { return OpenClawSessionsPreviewPayload(ts: 0, previews: []) } var params: [String: AnyCodable] = ["keys": AnyCodable(resolvedKeys)] if let limit { params["limit"] = AnyCodable(limit) } if let maxChars { params["maxChars"] = AnyCodable(maxChars) } let timeout = timeoutMs.map { Double($0) } return try await self.requestDecoded( method: .sessionsPreview, params: params, timeoutMs: timeout) } // MARK: - Chat func agentIdentity(sessionKey: String, timeoutMs: Double = 10000) async throws -> AgentIdentityResult { // Identity and chat.send must resolve aliases to the same canonical session target. let resolvedKey = self.canonicalizeSessionKey(sessionKey) return try await self.requestDecoded( method: .agentIdentityGet, params: ["sessionKey": AnyCodable(resolvedKey)], timeoutMs: timeoutMs) } func chatHistory( sessionKey: String, agentID: String? = nil, limit: Int? = nil, maxChars: Int? = nil, timeoutMs: Int? = nil, ifCurrentRoute route: Route? = nil) async throws -> OpenClawChatHistoryPayload { let resolvedKey = self.canonicalizeSessionKey(sessionKey) let request = OpenClawChatGatewayRequests.history( sessionKey: resolvedKey, agentID: agentID, limit: limit, maxChars: maxChars, timeoutMs: timeoutMs) if let route { let data = try await self.request( request, ifCurrentRoute: route) return try self.decoder.decode(OpenClawChatHistoryPayload.self, from: data) } let data = try await self.request(request) return try self.decoder.decode(OpenClawChatHistoryPayload.self, from: data) } func chatSend( sessionKey: String, agentID: String? = nil, expectedSessionRoutingContract: String? = nil, message: String, thinking: String?, idempotencyKey: String, attachments: [OpenClawChatAttachmentPayload], runTimeoutMs: Int? = nil, requestTimeoutMs: Int = 30000, ifCurrentRoute route: Route? = nil, distinguishPreDispatchRouteChange: Bool = false) async throws -> OpenClawChatSendResponse { let resolvedKey = self.canonicalizeSessionKey(sessionKey) let request = OpenClawChatGatewayRequests.sendMessage( sessionKey: resolvedKey, agentID: agentID, expectedSessionRoutingContract: expectedSessionRoutingContract, message: message, thinking: thinking, idempotencyKey: idempotencyKey, attachments: attachments, runTimeoutMs: runTimeoutMs, requestTimeoutMs: requestTimeoutMs) if let route { let data = try await self.request( request, ifCurrentRoute: route, distinguishPreDispatchRouteChange: distinguishPreDispatchRouteChange) return try self.decoder.decode(OpenClawChatSendResponse.self, from: data) } let data = try await self.request(request) return try self.decoder.decode(OpenClawChatSendResponse.self, from: data) } func talkMode(enabled: Bool, phase: String? = nil) async { var params: [String: AnyCodable] = ["enabled": AnyCodable(enabled)] if let phase { params["phase"] = AnyCodable(phase) } try? await self.requestVoid(method: .talkMode, params: params) } // MARK: - VoiceWake func voiceWakeGetTriggers() async throws -> [String] { struct VoiceWakePayload: Decodable { let triggers: [String] } let payload: VoiceWakePayload = try await requestDecoded(method: .voicewakeGet) return payload.triggers } func voiceWakeSetTriggers(_ triggers: [String]) async { do { try await self.requestVoid( method: .voicewakeSet, params: ["triggers": AnyCodable(triggers)], timeoutMs: 10000) } catch { // Best-effort only. } } // MARK: - Node pairing func nodePairApprove(requestId: String) async throws { try await self.requestVoid( method: .nodePairApprove, params: ["requestId": AnyCodable(requestId)], timeoutMs: 10000) } func nodePairReject(requestId: String) async throws { try await self.requestVoid( method: .nodePairReject, params: ["requestId": AnyCodable(requestId)], timeoutMs: 10000) } // MARK: - Device pairing func devicePairApprove(requestId: String) async throws { try await self.requestVoid( method: .devicePairApprove, params: ["requestId": AnyCodable(requestId)], timeoutMs: 10000) } func devicePairReject(requestId: String) async throws { try await self.requestVoid( method: .devicePairReject, params: ["requestId": AnyCodable(requestId)], timeoutMs: 10000) } // MARK: - Cron struct CronSchedulerStatus: Decodable { let enabled: Bool let storePath: String let sqlitePath: String? let jobs: Int let nextWakeAtMs: Int? } func cronStatus() async throws -> CronSchedulerStatus { try await self.requestDecoded(method: .cronStatus) } func cronList(includeDisabled: Bool = true) async throws -> [CronJob] { let data = try await requestRaw( method: .cronList, params: ["includeDisabled": AnyCodable(includeDisabled)]) return try Self.decodeCronListResponse(data) } func cronRuns(jobId: String, limit: Int = 200) async throws -> [CronRunLogEntry] { let data = try await requestRaw( method: .cronRuns, params: ["id": AnyCodable(jobId), "limit": AnyCodable(limit)]) return try Self.decodeCronRunsResponse(data) } func cronRun(jobId: String, force: Bool = true) async throws { try await self.requestVoid( method: .cronRun, params: [ "id": AnyCodable(jobId), "mode": AnyCodable(force ? "force" : "due"), ], timeoutMs: 20000) } func cronRemove(jobId: String) async throws { try await self.requestVoid(method: .cronRemove, params: ["id": AnyCodable(jobId)]) } func cronUpdate(jobId: String, patch: [String: AnyCodable]) async throws { try await self.requestVoid( method: .cronUpdate, params: ["id": AnyCodable(jobId), "patch": AnyCodable(patch)]) } func cronAdd(payload: [String: AnyCodable]) async throws { try await self.requestVoid(method: .cronAdd, params: payload) } }