diff --git a/packages/gateway-client/package.json b/packages/gateway-client/package.json index b5bbda7ceee1..7647e58b7ca2 100644 --- a/packages/gateway-client/package.json +++ b/packages/gateway-client/package.json @@ -14,6 +14,11 @@ "import": "./dist/index.mjs", "default": "./dist/index.mjs" }, + "./browser": { + "types": "./dist/browser.d.mts", + "import": "./dist/browser.mjs", + "default": "./dist/browser.mjs" + }, "./readiness": { "types": "./dist/readiness.d.mts", "import": "./dist/readiness.mjs", @@ -26,12 +31,12 @@ } }, "scripts": { - "build": "tsdown src/index.ts src/readiness.ts src/timeouts.ts --no-config --platform node --format esm --dts --out-dir dist --clean" + "build": "tsdown src/index.ts src/browser.ts src/readiness.ts src/timeouts.ts --no-config --platform node --format esm --dts --out-dir dist --clean" }, "dependencies": { "@openclaw/gateway-protocol": "workspace:*", + "@openclaw/net-policy": "workspace:*", "@openclaw/retry": "workspace:*", - "ipaddr.js": "2.4.0", "ws": "8.21.0" } } diff --git a/packages/gateway-client/src/browser.ts b/packages/gateway-client/src/browser.ts new file mode 100644 index 000000000000..aec69fbd6b1a --- /dev/null +++ b/packages/gateway-client/src/browser.ts @@ -0,0 +1,10 @@ +// Browser-safe gateway client surface. Keep Node transport/TLS dependencies out +// of this entry so browser consumers share the wire engine without polyfills. +export * from "./device-auth.js"; +export * from "./protocol-client.js"; +export * from "./reconnect-policy.js"; +export * from "@openclaw/gateway-protocol/client-info"; +export * from "@openclaw/gateway-protocol/connect-error-details"; +export * from "@openclaw/gateway-protocol/startup-unavailable"; +export * from "@openclaw/gateway-protocol/version"; +export type { ConnectParams, ErrorShape, EventFrame, HelloOk } from "@openclaw/gateway-protocol"; diff --git a/packages/gateway-client/src/client.ts b/packages/gateway-client/src/client.ts index e5986f7a23ac..33ec7e2b874f 100644 --- a/packages/gateway-client/src/client.ts +++ b/packages/gateway-client/src/client.ts @@ -11,24 +11,32 @@ import { formatConnectErrorMessage, readConnectErrorDetailCode, readConnectErrorRecoveryAdvice, - readPairingConnectErrorDetails, type ConnectErrorRecoveryAdvice, } from "@openclaw/gateway-protocol/connect-error-details"; -import { - type ConnectParams, - type ErrorShape, - type EventFrame, - type HelloOk, - isGatewayEventFrame, - isGatewayResponseFrame, - type RequestFrame, +import type { + ConnectParams, + ErrorShape, + EventFrame, + HelloOk, } from "@openclaw/gateway-protocol/frame-guards"; import { resolveGatewayStartupRetryAfterMs } from "@openclaw/gateway-protocol/startup-unavailable"; import { MIN_CLIENT_PROTOCOL_VERSION, PROTOCOL_VERSION } from "@openclaw/gateway-protocol/version"; -import { RetrySupervisor, sleepWithAbort, type BackoffPolicy } from "@openclaw/retry"; -import ipaddr from "ipaddr.js"; +import { + isLoopbackIpAddress, + normalizeIpAddress, + parseCanonicalIpAddress, + type ParsedIpAddress, +} from "@openclaw/net-policy/ip"; import { WebSocket, type ClientOptions, type CertMeta } from "ws"; import { buildDeviceAuthPayloadV3 } from "./device-auth.js"; +import { + GatewayProtocolClient, + GatewayProtocolRequestError, + type GatewayProtocolCloseContext, + type GatewayProtocolSocket, + type GatewayProtocolSocketHandlers, +} from "./protocol-client.js"; +import { shouldPauseGatewayReconnect } from "./reconnect-policy.js"; import { resolveConnectChallengeTimeoutMs, resolveSafeTimeoutDelayMs } from "./timeouts.js"; export type DeviceIdentity = { @@ -131,8 +139,6 @@ function parseHostForAddressChecks( }; } -type ParsedIpAddress = ipaddr.IPv4 | ipaddr.IPv6; - const PRIVATE_OR_LOOPBACK_IPV4_RANGES = new Set([ "loopback", "private", @@ -147,24 +153,9 @@ const PRIVATE_OR_LOOPBACK_IPV6_RANGES = new Set([ "deprecatedSiteLocal", ]); -function parseGatewayIpAddress(host: string): ParsedIpAddress | null { - const normalized = host.toLowerCase(); - if (ipaddr.IPv4.isValid(normalized) && !ipaddr.IPv4.isValidFourPartDecimal(normalized)) { - return null; - } - if (!ipaddr.isValid(normalized)) { - return null; - } - const parsed = ipaddr.parse(normalized); - // WHATWG URL canonicalization can turn ::ffff:127.0.0.1 into ::ffff:7f00:1. - // Normalize mapped forms so IPv4 loopback/private policy stays identical. - if (parsed.kind() === "ipv6") { - const ipv6 = parsed as ipaddr.IPv6; - if (ipv6.isIPv4MappedAddress()) { - return ipv6.toIPv4Address(); - } - } - return parsed; +function parseGatewayIpAddress(host: string): ParsedIpAddress | undefined { + const normalized = normalizeIpAddress(host); + return normalized ? parseCanonicalIpAddress(normalized) : undefined; } function isPrivateOrLoopbackIpAddress(address: ParsedIpAddress): boolean { @@ -181,11 +172,7 @@ function isLoopbackHost(host: string): boolean { if (parsed.isLocalhost) { return true; } - const address = parseGatewayIpAddress(parsed.unbracketedHost); - if (!address) { - return false; - } - return address.range() === "loopback"; + return isLoopbackIpAddress(parsed.unbracketedHost); } function isPrivateOrLoopbackHost(host: string): boolean { @@ -233,7 +220,7 @@ function isSecureWebSocketUrl(rawUrl: string, options?: { allowPrivateWs?: boole ? url.hostname.slice(1, -1) : url.hostname; return ( - isPrivateOrLoopbackHost(url.hostname) || parseGatewayIpAddress(hostForIpCheck) === null + isPrivateOrLoopbackHost(url.hostname) || parseGatewayIpAddress(hostForIpCheck) === undefined ); } return false; @@ -242,16 +229,6 @@ function isSecureWebSocketUrl(rawUrl: string, options?: { allowPrivateWs?: boole } } -type Pending = { - resolve: (value: unknown) => void; - reject: (err: unknown) => void; - expectFinal: boolean; - timeout: NodeJS.Timeout | null; - cleanup?: () => void; - onAccepted?: (payload: unknown) => void; - acceptedNotified?: boolean; -}; - export type GatewayClientRequestOptions = { expectFinal?: boolean; timeoutMs?: number | null; @@ -308,19 +285,18 @@ export type GatewayClientCloseInfo = { transientPreHelloCleanClose: boolean; }; -export class GatewayClientRequestError extends Error { +export class GatewayClientRequestError extends GatewayProtocolRequestError { readonly gatewayCode: string; - readonly details?: unknown; - readonly retryable: boolean; - readonly retryAfterMs?: number; + override readonly retryable: boolean; constructor(error: Partial) { - super(formatConnectErrorMessage({ message: error.message, details: error.details })); + super({ + ...error, + message: formatConnectErrorMessage({ message: error.message, details: error.details }), + }); this.name = "GatewayClientRequestError"; this.gatewayCode = error.code ?? "UNAVAILABLE"; - this.details = error.details; this.retryable = error.retryable === true; - this.retryAfterMs = error.retryAfterMs; } } @@ -331,6 +307,8 @@ class GatewayClientTransientPreHelloCloseError extends Error { } } +class GatewayClientTransportPolicyError extends Error {} + const GATEWAY_CONNECT_ASSEMBLY_ERROR = Symbol("gateway.connectAssemblyError"); type GatewayConnectAssemblyError = Error & { @@ -458,12 +436,6 @@ export function resolveGatewayClientConnectChallengeTimeoutMs( const FORCE_STOP_TERMINATE_GRACE_MS = 250; const STOP_AND_WAIT_TIMEOUT_MS = 1_000; const MAX_SUPPRESSED_TRANSIENT_PRE_HELLO_CLEAN_CLOSES = 1; -const GATEWAY_RECONNECT_POLICY: BackoffPolicy = { - initialMs: 1_000, - maxMs: 30_000, - factor: 2, - jitter: 0, -}; type PendingStop = { ws: WebSocket; @@ -473,31 +445,22 @@ type PendingStop = { }; export class GatewayClient { + private readonly protocol: GatewayProtocolClient; private ws: WebSocket | null = null; private opts: GatewayClientOptions; private deps: Required; - private pending = new Map(); - private readonly reconnectSupervisor = new RetrySupervisor(GATEWAY_RECONNECT_POLICY); - private closed = false; - private lastSeq: number | null = null; - private connectNonce: string | null = null; - private connectSent = false; - private connectTimer: NodeJS.Timeout | null = null; + private stopped = false; private pendingDeviceTokenRetry = false; private deviceTokenRetryBudgetUsed = false; private approvalRuntimeTokenCompatibilityDisabled = false; private approvalRuntimeTokenRetryBudgetUsed = false; - private pendingConnectErrorDetailCode: string | null = null; - private pendingConnectErrorDetails: unknown = null; // Track last tick to detect silent stalls. private lastTick: number | null = null; private tickIntervalMs = 30_000; private tickTimer: NodeJS.Timeout | null = null; private readonly requestTimeoutMs: number; private pendingStop: PendingStop | null = null; - private socketOpened = false; private transportValidated = false; - private helloOkReceived = false; private suppressedTransientPreHelloCleanCloses = 0; constructor(opts: GatewayClientOptions) { @@ -537,6 +500,66 @@ export class GatewayClient { typeof opts.requestTimeoutMs === "number" && Number.isFinite(opts.requestTimeoutMs) ? resolveSafeTimeoutDelayMs(opts.requestTimeoutMs, { minMs: 0 }) : 30_000; + this.protocol = new GatewayProtocolClient({ + createSocket: (handlers) => this.createSocket(handlers), + createRequestId: randomUUID, + createRequestError: (error) => new GatewayClientRequestError(error), + createRequestTimeoutError: (method) => new Error(`gateway request timeout for ${method}`), + createRequestAbortError: createGatewayRequestAbortError, + buildConnectPlan: ({ nonce }) => { + if (!nonce) { + throw new Error("gateway connect challenge missing nonce"); + } + return this.assembleConnectParams({ role: this.opts.role ?? "operator", nonce }); + }, + buildConnectParams: (assembled) => assembled.params, + onConnectPlanError: (error) => { + this.stopped = true; + const marked = markGatewayConnectAssemblyError(error); + const msg = `gateway connect failed: ${formatGatewayClientErrorForLog(error)}`; + if (this.opts.mode === GATEWAY_CLIENT_MODES.PROBE || isGatewayClientStoppedError(error)) { + this.logDebug(msg); + } else { + this.logError(msg); + } + return { closeCode: 1008, closeReason: "connect failed", stop: true, error: marked }; + }, + onConnectHello: (hello, context) => this.handleConnectHello(hello, context.plan), + onHello: (hello) => this.opts.onHelloOk?.(hello), + onConnectFailure: (error, context) => this.handleConnectRequestFailure(error, context.plan), + resolveClose: (context) => this.resolveClose(context), + onClose: (context, decision) => { + if (this.tickTimer) { + clearInterval(this.tickTimer); + this.tickTimer = null; + } + if (decision.notify) { + this.opts.onClose?.(context.code, context.reason, this.closeInfo(context)); + } + }, + notifyStoppedClose: true, + onConnectError: (error) => this.notifyConnectError(error), + onParseError: (error) => + this.logDebug(`gateway client parse error: ${formatGatewayClientErrorForLog(error)}`), + onEvent: (event) => this.opts.onEvent?.(event), + onGap: (info) => this.opts.onGap?.(info), + onActivity: () => { + this.lastTick = Date.now(); + }, + onCallbackError: (label, error) => + this.logDebug( + `gateway client ${label === "hello" ? "hello-ok" : label === "gap" ? "event" : label} handler error: ${formatGatewayClientErrorForLog(error)}`, + ), + handshake: { + mode: "require-challenge", + timeoutMs: resolveGatewayClientConnectChallengeTimeoutMs(this.opts), + timeoutMessage: (elapsedMs) => + `gateway connect challenge timeout (waited ${elapsedMs}ms, limit ${resolveGatewayClientConnectChallengeTimeoutMs(this.opts)}ms)`, + }, + reconnect: { initialMs: 1_000, multiplier: 2, maxMs: 30_000 }, + requestTimeoutMs: this.requestTimeoutMs, + rethrowSocketFactoryError: (error) => error instanceof GatewayClientTransportPolicyError, + }); } getConnectionMetadata(): GatewayClientConnectionMetadata { @@ -549,17 +572,16 @@ export class GatewayClient { } start() { - if (this.closed) { + if (this.stopped) { return; } - this.reconnectSupervisor.cancel(); - this.clearConnectChallengeTimeout(); - this.connectNonce = null; - this.connectSent = false; + this.protocol.start(); + } + + private createSocket(handlers: GatewayProtocolSocketHandlers): GatewayProtocolSocket { const url = this.opts.url ?? DEFAULT_GATEWAY_CLIENT_URL; if (this.opts.tlsFingerprint && !url.startsWith("wss://")) { - this.notifyConnectError(new Error("gateway tls fingerprint requires wss:// gateway url")); - return; + throw new Error("gateway tls fingerprint requires wss:// gateway url"); } const allowPrivateWs = @@ -574,7 +596,7 @@ export class GatewayClient { } catch { // Use raw URL if parsing fails } - const error = new Error( + throw new Error( `SECURITY ERROR: Cannot connect to "${displayHost}" over plaintext ws://. ` + "Both credentials and chat data would be exposed to network interception. " + "Use wss:// for remote URLs. Safe defaults: keep gateway.bind=loopback and connect via SSH tunnel " + @@ -584,8 +606,6 @@ export class GatewayClient { : "Break-glass (trusted private networks only): set OPENCLAW_ALLOW_INSECURE_PRIVATE_WS=1. ") + "Run `openclaw doctor --fix` for guidance.", ); - this.notifyConnectError(error); - return; } // Allow node screen snapshots and other large responses. this.deps.beforeConnect(); @@ -619,115 +639,53 @@ export class GatewayClient { let ws: WebSocket; // Managed proxies can intercept local traffic; the host owns the bypass // lifecycle and must remove it immediately after the socket is created. - const unregisterGatewayLoopbackBypass = this.deps.registerGatewayLoopbackBypass(url); + let unregisterGatewayLoopbackBypass: (() => void) | undefined; + try { + unregisterGatewayLoopbackBypass = this.deps.registerGatewayLoopbackBypass(url); + } catch (error) { + throw new GatewayClientTransportPolicyError( + error instanceof Error ? error.message : String(error), + ); + } try { ws = new WebSocket(url, wsOptions as ClientOptions); } catch (error) { - this.notifyConnectError(error instanceof Error ? error : new Error(String(error))); - return; + throw error instanceof Error ? error : new Error(String(error)); } finally { unregisterGatewayLoopbackBypass?.(); } this.ws = ws; - this.socketOpened = false; this.transportValidated = false; - this.helloOkReceived = false; - this.connectNonce = null; - this.connectSent = false; - this.clearConnectChallengeTimeout(); - ws.on("open", () => { - this.socketOpened = true; + handlers.open(); if (url.startsWith("wss://") && this.opts.tlsFingerprint) { const tlsError = this.validateTlsFingerprint(); if (tlsError) { - this.notifyConnectError(tlsError); - this.ws?.close(1008, tlsError.message); + handlers.error(tlsError); + ws.close(1008, tlsError.message); return; } } this.transportValidated = true; - this.beginPreauthHandshake(); }); - ws.on("message", (data) => this.handleMessage(rawDataToString(data))); + ws.on("message", (data) => handlers.message(rawDataToString(data))); ws.on("close", (code, reason) => { const reasonText = rawDataToString(reason); - const closeInfo: GatewayClientCloseInfo = { - phase: this.helloOkReceived ? "post-hello" : "pre-hello", - socketOpened: this.socketOpened, - transportValidated: this.transportValidated, - transientPreHelloCleanClose: !this.helloOkReceived && code === 1000 && reasonText === "", - }; - const connectErrorDetailCode = this.pendingConnectErrorDetailCode; - const connectErrorDetails = this.pendingConnectErrorDetails; - this.pendingConnectErrorDetailCode = null; - this.pendingConnectErrorDetails = null; if (this.ws === ws) { this.ws = null; } - this.socketOpened = false; - this.transportValidated = false; this.resolvePendingStop(ws); - if (this.reconnectSupervisor.nextDelayOverrideMs !== undefined) { - this.scheduleReconnect(); - return; - } - if ( - closeInfo.transientPreHelloCleanClose && - this.suppressedTransientPreHelloCleanCloses < - MAX_SUPPRESSED_TRANSIENT_PRE_HELLO_CLEAN_CLOSES - ) { - this.suppressedTransientPreHelloCleanCloses += 1; - this.flushPendingErrors(new GatewayClientTransientPreHelloCloseError()); - this.scheduleReconnect(); - this.notifyClose(code, reasonText, closeInfo); - return; - } - // Clear persisted device auth state only when device-token auth was active. - // Shared token/password failures can return the same close reason but should - // not erase a valid cached device token. - if ( - code === 1008 && - normalizeLowercaseStringOrEmpty(reasonText).includes("device token mismatch") && - !this.opts.token && - !this.opts.password && - this.opts.deviceIdentity - ) { - const deviceId = this.opts.deviceIdentity.deviceId; - const role = this.opts.role ?? "operator"; - try { - this.deps.clearDeviceAuthToken({ deviceId, role, env: this.opts.env }); - this.logDebug(`cleared stale device-auth token for device ${deviceId}`); - } catch (err) { - this.logDebug( - `failed clearing stale device-auth token for device ${deviceId}: ${String(err)}`, - ); - } - } - this.flushPendingErrors(new Error(`gateway closed (${code}): ${reasonText}`)); - if ( - this.shouldPauseReconnectAfterAuthFailure({ - detailCode: connectErrorDetailCode, - details: connectErrorDetails, - }) - ) { - this.notifyReconnectPaused({ - code, - reason: reasonText, - detailCode: connectErrorDetailCode, - }); - this.notifyClose(code, reasonText, closeInfo); - return; - } - this.scheduleReconnect(); - this.notifyClose(code, reasonText, closeInfo); + handlers.close(code, reasonText); }); ws.on("error", (err) => { this.logDebug(`gateway client error: ${formatGatewayClientErrorForLog(err)}`); - if (!this.connectSent) { - this.notifyConnectError(err instanceof Error ? err : new Error(String(err))); - } + handlers.error(err instanceof Error ? err : new Error(String(err))); }); + return { + isOpen: () => ws.readyState === WebSocket.OPEN, + send: (data) => ws.send(data), + close: (code, reason) => ws.close(code, reason), + }; } stop() { @@ -764,19 +722,14 @@ export class GatewayClient { } private beginStop(): Promise | null { - this.closed = true; + this.stopped = true; this.pendingDeviceTokenRetry = false; this.deviceTokenRetryBudgetUsed = false; - this.pendingConnectErrorDetailCode = null; - this.pendingConnectErrorDetails = null; - this.reconnectSupervisor.reset(); if (this.tickTimer) { clearInterval(this.tickTimer); this.tickTimer = null; } - this.clearConnectChallengeTimeout(); if (this.pendingStop) { - this.flushPendingErrors(new Error("gateway client stopped")); return this.pendingStop.promise; } const ws = this.ws; @@ -786,16 +739,21 @@ export class GatewayClient { const forceTerminateTimer = setTimeout(() => { try { ws.terminate(); - } catch {} - this.resolvePendingStop(ws); + } finally { + this.resolvePendingStop(ws); + } }, FORCE_STOP_TERMINATE_GRACE_MS); forceTerminateTimer.unref?.(); pendingStop.terminateTimer = forceTerminateTimer; - ws.close(); - this.flushPendingErrors(new Error("gateway client stopped")); + if (this.protocol.connecting) { + const error = new Error("gateway client stopped"); + this.notifyConnectError(error); + this.logDebug(`gateway connect failed: ${formatGatewayClientErrorForLog(error)}`); + } + this.protocol.stop(); return pendingStop.promise; } - this.flushPendingErrors(new Error("gateway client stopped")); + this.protocol.stop(); return null; } @@ -835,144 +793,6 @@ export class GatewayClient { this.deps.logError(this.deps.redactForLog(message)); } - private sendConnect() { - if (this.connectSent) { - return; - } - const nonce = normalizeOptionalString(this.connectNonce) ?? ""; - if (!nonce) { - this.notifyConnectError(new Error("gateway connect challenge missing nonce")); - this.ws?.close(1008, "connect challenge missing nonce"); - return; - } - const role = this.opts.role ?? "operator"; - let assembled: AssembledConnect; - try { - // Build the full connect frame before marking connectSent so synchronous - // signing/storage failures surface as connect-assembly errors, not RPCs. - assembled = this.assembleConnectParams({ role, nonce }); - } catch (err) { - this.handleConnectFailure(err); - return; - } - - this.connectSent = true; - this.clearConnectChallengeTimeout(); - - void this.request("connect", assembled.params) - .then((helloOk) => { - this.helloOkReceived = true; - this.pendingDeviceTokenRetry = false; - this.deviceTokenRetryBudgetUsed = false; - this.pendingConnectErrorDetailCode = null; - this.pendingConnectErrorDetails = null; - this.suppressedTransientPreHelloCleanCloses = 0; - const authInfo = helloOk?.auth; - if (authInfo?.deviceToken && this.opts.deviceIdentity) { - this.deps.storeDeviceAuthToken({ - deviceId: this.opts.deviceIdentity.deviceId, - role: authInfo.role ?? role, - token: authInfo.deviceToken, - scopes: authInfo.scopes ?? [], - env: this.opts.env, - }); - } - this.reconnectSupervisor.reset(); - this.tickIntervalMs = - typeof helloOk.policy?.tickIntervalMs === "number" - ? helloOk.policy.tickIntervalMs - : 30_000; - this.lastTick = Date.now(); - this.startTickWatch(); - this.notifyHelloOk(helloOk); - }) - .catch((err: unknown) => { - if (err instanceof GatewayClientTransientPreHelloCloseError) { - return; - } - this.pendingConnectErrorDetailCode = - err instanceof GatewayClientRequestError ? readConnectErrorDetailCode(err.details) : null; - this.pendingConnectErrorDetails = - err instanceof GatewayClientRequestError ? err.details : null; - const shouldRetryWithDeviceToken = this.shouldRetryWithStoredDeviceToken({ - error: err, - explicitGatewayToken: normalizeOptionalString(this.opts.token), - resolvedDeviceToken: assembled.resolvedDeviceToken, - storedToken: assembled.storedToken, - }); - if ( - this.opts.deviceIdentity && - assembled.usingStoredDeviceToken && - err instanceof GatewayClientRequestError && - readConnectErrorDetailCode(err.details) === - ConnectErrorDetailCodes.AUTH_DEVICE_TOKEN_MISMATCH - ) { - const deviceId = this.opts.deviceIdentity.deviceId; - try { - this.deps.clearDeviceAuthToken({ deviceId, role, env: this.opts.env }); - this.logDebug(`cleared stale device-auth token for device ${deviceId}`); - } catch (clearErr) { - this.logDebug( - `failed clearing stale device-auth token for device ${deviceId}: ${String(clearErr)}`, - ); - } - } - if (shouldRetryWithDeviceToken) { - this.pendingDeviceTokenRetry = true; - this.deviceTokenRetryBudgetUsed = true; - this.reconnectSupervisor.reset(250); - } - const startupRetryAfterMs = resolveGatewayStartupRetryAfterMs(err); - if (startupRetryAfterMs !== null) { - // Startup Retry-After is a floor for this wait, not a failed connect - // attempt. Preserve the exponential sequence for the next failure. - this.reconnectSupervisor.nextDelayOverrideMs = startupRetryAfterMs; - this.logDebug(`gateway connect failed: ${formatGatewayClientErrorForLog(err)}`); - this.ws?.close(1013, "gateway starting"); - return; - } - if ( - this.shouldFailClosedForUnsupportedAgentRuntimeIdentity({ - error: err, - authAgentRuntimeIdentityToken: assembled.authAgentRuntimeIdentityToken, - }) - ) { - const unsupportedIdentityError = new Error( - "gateway rejected required agent runtime identity auth field; refusing to retry without it", - ); - this.notifyConnectError(unsupportedIdentityError); - this.logError(`gateway connect failed: ${unsupportedIdentityError.message}`); - // This identity scopes model-mediated cron calls. Retrying without it - // would turn an old/new mismatch into an unscoped operator call. - this.closed = true; - this.reconnectSupervisor.cancel(); - this.ws?.close(1008, "connect failed"); - return; - } - if ( - this.shouldRetryWithoutApprovalRuntimeToken({ - error: err, - authApprovalRuntimeToken: assembled.authApprovalRuntimeToken, - }) - ) { - this.approvalRuntimeTokenCompatibilityDisabled = true; - this.approvalRuntimeTokenRetryBudgetUsed = true; - this.reconnectSupervisor.reset(250); - this.logDebug("gateway rejected approval runtime auth field; retrying without it"); - this.ws?.close(1008, "connect retry"); - return; - } - this.notifyConnectError(err instanceof Error ? err : new Error(String(err))); - const msg = `gateway connect failed: ${formatGatewayClientErrorForLog(err)}`; - if (this.opts.mode === GATEWAY_CLIENT_MODES.PROBE || isGatewayClientStoppedError(err)) { - this.logDebug(msg); - } else { - this.logError(msg); - } - this.ws?.close(1008, "connect failed"); - }); - } - private assembleConnectParams(params: { role: string; nonce: string }): AssembledConnect { const { role, nonce } = params; // Auth selection is intentionally centralized: retry decisions depend on @@ -1095,18 +915,194 @@ export class GatewayClient { }; } - private handleConnectFailure(err: unknown) { - const error = err instanceof Error ? err : new Error(String(err)); - this.clearConnectChallengeTimeout(); - this.closed = true; - this.notifyConnectError(markGatewayConnectAssemblyError(error)); - const msg = `gateway connect failed: ${formatGatewayClientErrorForLog(error)}`; - if (this.opts.mode === GATEWAY_CLIENT_MODES.PROBE || isGatewayClientStoppedError(error)) { - this.logDebug(msg); - } else { - this.logError(msg); + private handleConnectHello(helloOk: HelloOk, assembled: AssembledConnect): void { + this.pendingDeviceTokenRetry = false; + this.deviceTokenRetryBudgetUsed = false; + this.suppressedTransientPreHelloCleanCloses = 0; + const role = this.opts.role ?? "operator"; + const authInfo = helloOk.auth; + if (authInfo?.deviceToken && this.opts.deviceIdentity) { + this.deps.storeDeviceAuthToken({ + deviceId: this.opts.deviceIdentity.deviceId, + role: authInfo.role ?? role, + token: authInfo.deviceToken, + scopes: authInfo.scopes ?? [], + env: this.opts.env, + }); + } + this.tickIntervalMs = + typeof helloOk.policy?.tickIntervalMs === "number" ? helloOk.policy.tickIntervalMs : 30_000; + this.lastTick = Date.now(); + this.startTickWatch(); + void assembled; + } + + private handleConnectRequestFailure( + error: GatewayProtocolRequestError, + assembled: AssembledConnect, + ) { + const role = this.opts.role ?? "operator"; + const shouldRetryWithDeviceToken = this.shouldRetryWithStoredDeviceToken({ + error, + explicitGatewayToken: normalizeOptionalString(this.opts.token), + resolvedDeviceToken: assembled.resolvedDeviceToken, + storedToken: assembled.storedToken, + }); + if ( + this.opts.deviceIdentity && + assembled.usingStoredDeviceToken && + error instanceof GatewayClientRequestError && + readConnectErrorDetailCode(error.details) === + ConnectErrorDetailCodes.AUTH_DEVICE_TOKEN_MISMATCH + ) { + const deviceId = this.opts.deviceIdentity.deviceId; + try { + this.deps.clearDeviceAuthToken({ deviceId, role, env: this.opts.env }); + this.logDebug(`cleared stale device-auth token for device ${deviceId}`); + } catch (clearError) { + this.logDebug( + `failed clearing stale device-auth token for device ${deviceId}: ${String(clearError)}`, + ); + } + } + if (shouldRetryWithDeviceToken) { + this.pendingDeviceTokenRetry = true; + this.deviceTokenRetryBudgetUsed = true; + this.protocol.resetReconnectBackoff(250); + } + const startupRetryAfterMs = resolveGatewayStartupRetryAfterMs(error); + if (startupRetryAfterMs !== null) { + this.logDebug(`gateway connect failed: ${formatGatewayClientErrorForLog(error)}`); + return { + closeCode: 1013, + closeReason: "gateway starting", + reconnectDelayMs: startupRetryAfterMs, + }; + } + if ( + this.shouldFailClosedForUnsupportedAgentRuntimeIdentity({ + error, + authAgentRuntimeIdentityToken: assembled.authAgentRuntimeIdentityToken, + }) + ) { + const unsupportedIdentityError = new Error( + "gateway rejected required agent runtime identity auth field; refusing to retry without it", + ); + this.stopped = true; + this.notifyConnectError(unsupportedIdentityError); + this.logError(`gateway connect failed: ${unsupportedIdentityError.message}`); + return { closeCode: 1008, closeReason: "connect failed", stop: true }; + } + if ( + this.shouldRetryWithoutApprovalRuntimeToken({ + error, + authApprovalRuntimeToken: assembled.authApprovalRuntimeToken, + }) + ) { + this.approvalRuntimeTokenCompatibilityDisabled = true; + this.approvalRuntimeTokenRetryBudgetUsed = true; + this.protocol.resetReconnectBackoff(250); + this.logDebug("gateway rejected approval runtime auth field; retrying without it"); + return { closeCode: 1008, closeReason: "connect retry" }; + } + this.notifyConnectError(error); + const message = `gateway connect failed: ${formatGatewayClientErrorForLog(error)}`; + if (this.opts.mode === GATEWAY_CLIENT_MODES.PROBE || isGatewayClientStoppedError(error)) { + this.logDebug(message); + } else { + this.logError(message); + } + return { + closeCode: 1008, + closeReason: "connect failed", + }; + } + + private resolveClose(context: GatewayProtocolCloseContext) { + const info = this.closeInfo(context); + const detailCode = + context.connectFailure?.error instanceof GatewayClientRequestError + ? readConnectErrorDetailCode(context.connectFailure.error.details) + : null; + const details = + context.connectFailure?.error instanceof GatewayClientRequestError + ? context.connectFailure.error.details + : undefined; + if (context.code === 1013 && context.connectFailure?.reconnectDelayMs !== undefined) { + return { + retry: true, + notify: false, + reconnectDelayMs: context.connectFailure.reconnectDelayMs, + }; + } + if ( + info.transientPreHelloCleanClose && + this.suppressedTransientPreHelloCleanCloses < MAX_SUPPRESSED_TRANSIENT_PRE_HELLO_CLEAN_CLOSES + ) { + this.suppressedTransientPreHelloCleanCloses += 1; + return { + retry: true, + notify: true, + pendingError: new GatewayClientTransientPreHelloCloseError(), + }; + } + if ( + info.transientPreHelloCleanClose || + (context.connectRequestSent && !context.helloReceived && !context.connectFailure) + ) { + const error = new Error(`gateway closed (${context.code}): ${context.reason}`); + this.notifyConnectError(error); + this.logError(`gateway connect failed: ${formatGatewayClientErrorForLog(error)}`); + } + this.clearStaleDeviceTokenForClose(context.code, context.reason); + if ( + shouldPauseGatewayReconnect({ + details, + deviceTokenRetryPending: this.pendingDeviceTokenRetry, + tokenMismatchIsTerminal: true, + clientVersionMismatchIsTerminal: true, + }) + ) { + this.notifyReconnectPaused({ code: context.code, reason: context.reason, detailCode }); + return { retry: false, notify: true }; + } + return { + retry: true, + notify: true, + reconnectDelayMs: context.connectFailure?.reconnectDelayMs, + }; + } + + private closeInfo(context: GatewayProtocolCloseContext): GatewayClientCloseInfo { + return { + phase: context.helloReceived ? "post-hello" : "pre-hello", + socketOpened: context.socketOpened, + transportValidated: this.transportValidated, + transientPreHelloCleanClose: + !context.helloReceived && context.code === 1000 && context.reason === "", + }; + } + + private clearStaleDeviceTokenForClose(code: number, reason: string): void { + if ( + code !== 1008 || + !normalizeLowercaseStringOrEmpty(reason).includes("device token mismatch") || + this.opts.token || + this.opts.password || + !this.opts.deviceIdentity + ) { + return; + } + const deviceId = this.opts.deviceIdentity.deviceId; + const role = this.opts.role ?? "operator"; + try { + this.deps.clearDeviceAuthToken({ deviceId, role, env: this.opts.env }); + this.logDebug(`cleared stale device-auth token for device ${deviceId}`); + } catch (error) { + this.logDebug( + `failed clearing stale device-auth token for device ${deviceId}: ${String(error)}`, + ); } - this.ws?.close(1008, "connect failed"); } private notifyConnectError(error: Error) { @@ -1119,16 +1115,6 @@ export class GatewayClient { } } - private notifyHelloOk(helloOk: HelloOk): void { - try { - this.opts.onHelloOk?.(helloOk); - } catch (err) { - this.logDebug( - `gateway client hello-ok handler error: ${formatGatewayClientErrorForLog(err)}`, - ); - } - } - private notifyReconnectPaused(info: GatewayReconnectPausedInfo): void { try { this.opts.onReconnectPaused?.(info); @@ -1139,18 +1125,6 @@ export class GatewayClient { } } - private notifyClose(code: number, reason: string, info?: GatewayClientCloseInfo): void { - try { - if (info === undefined) { - this.opts.onClose?.(code, reason); - return; - } - this.opts.onClose?.(code, reason, info); - } catch (err) { - this.logDebug(`gateway client close handler error: ${formatGatewayClientErrorForLog(err)}`); - } - } - private resolveConnectScopes(params: { usingStoredDeviceToken?: boolean; storedScopes?: string[]; @@ -1189,43 +1163,6 @@ export class GatewayClient { }; } - private shouldPauseReconnectAfterAuthFailure(params: { - detailCode: string | null; - details?: unknown; - }): boolean { - const { detailCode, details } = params; - if (!detailCode) { - return false; - } - const pairingDetails = readPairingConnectErrorDetails(details); - if ( - detailCode === ConnectErrorDetailCodes.PAIRING_REQUIRED && - (pairingDetails?.pauseReconnect === false || - pairingDetails?.recommendedNextStep === "wait_then_retry") - ) { - return false; - } - if ( - detailCode === ConnectErrorDetailCodes.AUTH_TOKEN_MISSING || - detailCode === ConnectErrorDetailCodes.AUTH_BOOTSTRAP_TOKEN_INVALID || - detailCode === ConnectErrorDetailCodes.AUTH_PASSWORD_MISSING || - detailCode === ConnectErrorDetailCodes.AUTH_PASSWORD_MISMATCH || - detailCode === ConnectErrorDetailCodes.AUTH_RATE_LIMITED || - detailCode === ConnectErrorDetailCodes.AUTH_DEVICE_TOKEN_MISMATCH || - detailCode === ConnectErrorDetailCodes.AUTH_SCOPE_MISMATCH || - detailCode === ConnectErrorDetailCodes.PAIRING_REQUIRED || - detailCode === ConnectErrorDetailCodes.CONTROL_UI_DEVICE_IDENTITY_REQUIRED || - detailCode === ConnectErrorDetailCodes.DEVICE_IDENTITY_REQUIRED || - detailCode === ConnectErrorDetailCodes.CLIENT_VERSION_MISMATCH - ) { - return true; - } - if (detailCode === ConnectErrorDetailCodes.AUTH_TOKEN_MISMATCH) { - return !this.pendingDeviceTokenRetry; - } - return false; - } - private shouldRetryWithStoredDeviceToken(params: { error: unknown; explicitGatewayToken?: string; @@ -1369,152 +1306,6 @@ export class GatewayClient { }; } - private handleMessage(raw: string) { - let parsed: unknown; - try { - parsed = JSON.parse(raw); - } catch (err) { - this.logDebug(`gateway client parse error: ${formatGatewayClientErrorForLog(err)}`); - return; - } - if (isGatewayEventFrame(parsed)) { - this.lastTick = Date.now(); - const evt = parsed; - if (evt.event === "connect.challenge") { - const payload = evt.payload as { nonce?: unknown } | undefined; - const nonce = payload && typeof payload.nonce === "string" ? payload.nonce : null; - if (!nonce || nonce.trim().length === 0) { - this.notifyConnectError(new Error("gateway connect challenge missing nonce")); - this.ws?.close(1008, "connect challenge missing nonce"); - return; - } - this.connectNonce = nonce.trim(); - if (this.socketOpened) { - this.sendConnect(); - } - return; - } - try { - const seq = typeof evt.seq === "number" ? evt.seq : null; - if (seq !== null) { - if (this.lastSeq !== null && seq > this.lastSeq + 1) { - this.opts.onGap?.({ expected: this.lastSeq + 1, received: seq }); - } - this.lastSeq = seq; - } - if (evt.event === "tick") { - this.lastTick = Date.now(); - } - this.opts.onEvent?.(evt); - } catch (err) { - this.logDebug(`gateway client event handler error: ${formatGatewayClientErrorForLog(err)}`); - } - return; - } - if (isGatewayResponseFrame(parsed)) { - this.lastTick = Date.now(); - const pending = this.pending.get(parsed.id); - if (!pending) { - return; - } - // If the payload is an ack with status accepted, keep waiting for final. - const payload = parsed.payload as { status?: unknown } | undefined; - const status = payload?.status; - if (pending.expectFinal && status === "accepted") { - if (!pending.acceptedNotified) { - pending.acceptedNotified = true; - try { - pending.onAccepted?.(parsed.payload); - } catch (err) { - this.logDebug( - `gateway client accepted callback error: ${formatGatewayClientErrorForLog(err)}`, - ); - } - } - return; - } - this.pending.delete(parsed.id); - pending.cleanup?.(); - if (parsed.ok) { - pending.resolve(parsed.payload); - } else { - pending.reject( - new GatewayClientRequestError({ - code: parsed.error?.code, - message: parsed.error?.message ?? "unknown error", - details: parsed.error?.details, - retryable: parsed.error?.retryable, - retryAfterMs: parsed.error?.retryAfterMs, - }), - ); - } - } - } - - private beginPreauthHandshake() { - if (this.connectSent) { - return; - } - if (this.connectNonce && !this.connectSent) { - this.armConnectChallengeTimeout(); - this.sendConnect(); - return; - } - this.armConnectChallengeTimeout(); - } - - private clearConnectChallengeTimeout() { - if (this.connectTimer) { - clearTimeout(this.connectTimer); - this.connectTimer = null; - } - } - - private armConnectChallengeTimeout() { - const connectChallengeTimeoutMs = resolveGatewayClientConnectChallengeTimeoutMs(this.opts); - const armedAt = Date.now(); - this.clearConnectChallengeTimeout(); - this.connectTimer = setTimeout(() => { - if (this.connectSent || this.ws?.readyState !== WebSocket.OPEN) { - return; - } - const elapsedMs = Date.now() - armedAt; - this.notifyConnectError( - new Error( - `gateway connect challenge timeout (waited ${elapsedMs}ms, limit ${connectChallengeTimeoutMs}ms)`, - ), - ); - this.ws?.close(1008, "connect challenge timeout"); - }, connectChallengeTimeoutMs); - } - - private scheduleReconnect() { - if (this.closed) { - return; - } - if (this.tickTimer) { - clearInterval(this.tickTimer); - this.tickTimer = null; - } - const retry = this.reconnectSupervisor.next(); - if (!retry) { - return; - } - // Ignore cancelled sleeps only; reconnect start failures must remain observable. - void sleepWithAbort(retry.delayMs, retry.signal).then( - () => this.start(), - () => {}, - ); - } - - private flushPendingErrors(err: Error) { - for (const [, p] of this.pending) { - p.cleanup?.(); - p.reject(err); - } - this.pending.clear(); - } - private startTickWatch() { if (this.tickTimer) { clearInterval(this.tickTimer); @@ -1526,15 +1317,14 @@ export class GatewayClient { : 1000; const interval = resolveSafeTimeoutDelayMs(Math.max(this.tickIntervalMs, minInterval)); this.tickTimer = setInterval(() => { - if (this.closed) { + if (this.stopped) { return; } if (!this.lastTick) { return; } const allPendingRequestsHaveTimeouts = - this.pending.size > 0 && - [...this.pending.values()].every((pending) => pending.timeout !== null); + this.protocol.hasPendingRequests && !this.protocol.hasUnboundedPendingRequests; // Finite requests own their deadline. One unbounded request keeps the // transport watchdog active so a dead socket cannot strand it forever. if (allPendingRequestsHaveTimeouts) { @@ -1549,7 +1339,7 @@ export class GatewayClient { ? Math.max(1, rawTimeoutMs) : this.tickIntervalMs * 2; if (gap > timeoutMs) { - this.ws?.close(4000, "tick timeout"); + this.protocol.closeSocket(4000, "tick timeout"); } }, interval); } @@ -1586,17 +1376,6 @@ export class GatewayClient { params?: unknown, opts?: GatewayClientRequestOptions, ): Promise { - if (!this.ws || this.ws.readyState !== WebSocket.OPEN) { - throw new Error("gateway not connected"); - } - if (opts?.signal?.aborted) { - throw createGatewayRequestAbortError(method); - } - if (typeof method !== "string" || method.length === 0) { - throw new Error("invalid request frame: method must be a non-empty string"); - } - const id = randomUUID(); - const frame: RequestFrame = { type: "req", id, method, params }; const expectFinal = opts?.expectFinal === true; const timeoutMs = opts?.timeoutMs === null @@ -1606,50 +1385,12 @@ export class GatewayClient { : expectFinal ? null : this.requestTimeoutMs; - const signal = opts?.signal; - const p = new Promise((resolve, reject) => { - const timeout = - timeoutMs === null - ? null - : setTimeout(() => { - const pending = this.pending.get(id); - this.pending.delete(id); - pending?.cleanup?.(); - reject(new Error(`gateway request timeout for ${method}`)); - }, timeoutMs); - const cleanup = () => { - if (timeout) { - clearTimeout(timeout); - } - if (signal && abortHandler) { - signal.removeEventListener("abort", abortHandler); - } - }; - const abortHandler: (() => void) | undefined = () => { - const pending = this.pending.get(id); - this.pending.delete(id); - pending?.cleanup?.(); - reject(createGatewayRequestAbortError(method)); - }; - this.pending.set(id, { - resolve: (value) => resolve(value as T), - reject, - expectFinal, - timeout, - cleanup, - onAccepted: opts?.onAccepted, - }); - signal?.addEventListener("abort", abortHandler, { once: true }); + return this.protocol.request(method, params, { + expectFinal, + timeoutMs, + signal: opts?.signal, + onAccepted: opts?.onAccepted, }); - try { - this.ws.send(JSON.stringify(frame)); - } catch (error) { - const pending = this.pending.get(id); - this.pending.delete(id); - pending?.cleanup?.(); - throw error; - } - return p; } } diff --git a/packages/gateway-client/src/client.watchdog.test.ts b/packages/gateway-client/src/client.watchdog.test.ts index 6609a706795f..72df38c23078 100644 --- a/packages/gateway-client/src/client.watchdog.test.ts +++ b/packages/gateway-client/src/client.watchdog.test.ts @@ -4,6 +4,7 @@ import { createServer } from "node:net"; import { afterEach, describe, expect, test, vi } from "vitest"; import { WebSocket, WebSocketServer } from "ws"; import { GatewayClient, resolveGatewayClientConnectChallengeTimeoutMs } from "./client.js"; +import type { GatewayProtocolSocket } from "./protocol-client.js"; import { DEFAULT_PREAUTH_HANDSHAKE_TIMEOUT_MS, MAX_SAFE_TIMEOUT_DELAY_MS, @@ -45,20 +46,44 @@ function createOpenGatewayClient(requestTimeoutMs: number): { requestTimeoutMs, }); const send = vi.fn(); - ( - client as unknown as { - ws: WebSocket | { readyState: number; send: () => void; close: () => void }; - } - ).ws = { - readyState: WebSocket.OPEN, - send, - close: vi.fn(), - }; + installSyntheticSocket(client, send, vi.fn()); return { client, send }; } function getPendingCount(client: GatewayClient): number { - return (client as unknown as { pending: Map }).pending.size; + return protocolHarness(client).requests.pending.size; +} + +type ProtocolHarness = { + socket: GatewayProtocolSocket | null; + stopped: boolean; + generation: number; + reconnectSupervisor: { reset(initialMs?: number): void }; + requests: { pending: Map }; + handleMessage: (socket: GatewayProtocolSocket, generation: number, raw: string) => void; +}; + +function protocolHarness(client: GatewayClient): ProtocolHarness { + return (client as unknown as { protocol: ProtocolHarness }).protocol; +} + +function installSyntheticSocket( + client: GatewayClient, + send: (data: string) => unknown, + close: (code?: number, reason?: string) => unknown, +): void { + const socket: GatewayProtocolSocket = { + isOpen: () => true, + send: (data) => send(data), + close: (code, reason) => close(code, reason), + }; + Object.assign(protocolHarness(client), { socket, stopped: false, generation: 1 }); + (client as unknown as { ws: unknown }).ws = { + readyState: WebSocket.OPEN, + send, + close, + terminate: vi.fn(), + }; } function trackSettlement(promise: Promise): () => boolean { @@ -86,13 +111,8 @@ function createWatchedGatewayClient(): { }); const close = vi.fn(); const send = vi.fn(); - Object.assign(client as unknown as { ws: unknown; tickIntervalMs: number; lastTick: number }, { - ws: { - readyState: WebSocket.OPEN, - send, - close, - terminate: vi.fn(), - }, + installSyntheticSocket(client, send, close); + Object.assign(client as unknown as { tickIntervalMs: number; lastTick: number }, { tickIntervalMs: 5, lastTick: Date.now(), }); @@ -101,9 +121,11 @@ function createWatchedGatewayClient(): { } function handleGatewayMessage(client: GatewayClient, payload: Record): void { - (client as unknown as { handleMessage: (raw: string) => void }).handleMessage( - JSON.stringify(payload), - ); + const protocol = protocolHarness(client); + if (!protocol.socket) { + throw new Error("synthetic protocol socket missing"); + } + protocol.handleMessage(protocol.socket, protocol.generation, JSON.stringify(payload)); } async function stopSyntheticClient(client: GatewayClient): Promise { @@ -116,6 +138,8 @@ describe("GatewayClient", () => { let httpsServer: ReturnType | null = null; afterEach(async () => { + vi.useRealTimers(); + vi.restoreAllMocks(); if (wss) { for (const client of wss.clients) { client.terminate(); @@ -296,11 +320,7 @@ describe("GatewayClient", () => { helloCount += 1; if (helloCount === 1) { // Keep the real reconnect lifecycle fast without changing production defaults. - ( - client as unknown as { - reconnectSupervisor: { reset(initialMs?: number): void }; - } - ).reconnectSupervisor.reset(10); + protocolHarness(client).reconnectSupervisor.reset(10); resolveFirstHello(); return; } @@ -339,25 +359,21 @@ describe("GatewayClient", () => { test("lets finite pending requests own their timeout when ticks are missing", async () => { vi.useFakeTimers(); - try { - const { client, close } = createWatchedGatewayClient(); - const request = client.request("status", undefined, { timeoutMs: 100 }); - const requestExpectation = expect(request).rejects.toThrow( - "gateway request timeout for status", - ); - await vi.advanceTimersByTimeAsync(20); + const { client, close } = createWatchedGatewayClient(); + const request = client.request("status", undefined, { timeoutMs: 100 }); + const requestExpectation = expect(request).rejects.toThrow( + "gateway request timeout for status", + ); + await vi.advanceTimersByTimeAsync(20); - expect(close).not.toHaveBeenCalled(); + expect(close).not.toHaveBeenCalled(); - await vi.advanceTimersByTimeAsync(80); - await requestExpectation; - await vi.advanceTimersByTimeAsync(5); + await vi.advanceTimersByTimeAsync(80); + await requestExpectation; + await vi.advanceTimersByTimeAsync(5); - expect(close).toHaveBeenCalledWith(4000, "tick timeout"); - await stopSyntheticClient(client); - } finally { - vi.useRealTimers(); - } + expect(close).toHaveBeenCalledWith(4000, "tick timeout"); + await stopSyntheticClient(client); }); test.each([ @@ -373,218 +389,123 @@ describe("GatewayClient", () => { }, ])("keeps the watchdog active for $label request", async ({ method, options }) => { vi.useFakeTimers(); - try { - const { client, close } = createWatchedGatewayClient(); - const request = client.request(method, undefined, options); - const requestExpectation = expect(request).rejects.toThrow("gateway client stopped"); + const { client, close } = createWatchedGatewayClient(); + const request = client.request(method, undefined, options); + const requestExpectation = expect(request).rejects.toThrow("gateway client stopped"); - await vi.advanceTimersByTimeAsync(20); + await vi.advanceTimersByTimeAsync(20); - expect(close).toHaveBeenCalledWith(4000, "tick timeout"); - await stopSyntheticClient(client); - await requestExpectation; - } finally { - vi.useRealTimers(); - } + expect(close).toHaveBeenCalledWith(4000, "tick timeout"); + await stopSyntheticClient(client); + await requestExpectation; }); test("keeps the watchdog active for mixed finite and unbounded requests", async () => { vi.useFakeTimers(); - try { - const { client, close } = createWatchedGatewayClient(); - const requests = [ - client.request("status", undefined, { timeoutMs: 100 }), - client.request("chat.send", undefined, { expectFinal: true }), - ]; - const settlements = Promise.allSettled(requests); + const { client, close } = createWatchedGatewayClient(); + const requests = [ + client.request("status", undefined, { timeoutMs: 100 }), + client.request("chat.send", undefined, { expectFinal: true }), + ]; + const settlements = Promise.allSettled(requests); - await vi.advanceTimersByTimeAsync(20); + await vi.advanceTimersByTimeAsync(20); - expect(close).toHaveBeenCalledWith(4000, "tick timeout"); - await stopSyntheticClient(client); - await expect(settlements).resolves.toEqual([ - expect.objectContaining({ status: "rejected" }), - expect.objectContaining({ status: "rejected" }), - ]); - } finally { - vi.useRealTimers(); - } + expect(close).toHaveBeenCalledWith(4000, "tick timeout"); + await stopSyntheticClient(client); + await expect(settlements).resolves.toEqual([ + expect.objectContaining({ status: "rejected" }), + expect.objectContaining({ status: "rejected" }), + ]); }); test("keeps an unbounded request alive while inbound ticks continue", async () => { vi.useFakeTimers(); - try { - const { client, close, send } = createWatchedGatewayClient(); - const request = client.request<{ status: string }>("chat.send", undefined, { - expectFinal: true, - }); - const requestFrame = JSON.parse(String(send.mock.calls[0]?.[0])) as { id: string }; + const { client, close, send } = createWatchedGatewayClient(); + const request = client.request<{ status: string }>("chat.send", undefined, { + expectFinal: true, + }); + const requestFrame = JSON.parse(String(send.mock.calls[0]?.[0])) as { id: string }; - for (let seq = 1; seq <= 4; seq += 1) { - await vi.advanceTimersByTimeAsync(5); - handleGatewayMessage(client, { type: "event", event: "tick", seq, payload: {} }); - } - - expect(close).not.toHaveBeenCalled(); - handleGatewayMessage(client, { - type: "res", - id: requestFrame.id, - ok: true, - payload: { status: "ok" }, - }); - await expect(request).resolves.toEqual({ status: "ok" }); - await stopSyntheticClient(client); - } finally { - vi.useRealTimers(); + for (let seq = 1; seq <= 4; seq += 1) { + await vi.advanceTimersByTimeAsync(5); + handleGatewayMessage(client, { type: "event", event: "tick", seq, payload: {} }); } + + expect(close).not.toHaveBeenCalled(); + handleGatewayMessage(client, { + type: "res", + id: requestFrame.id, + ok: true, + payload: { status: "ok" }, + }); + await expect(request).resolves.toEqual({ status: "ok" }); + await stopSyntheticClient(client); }); test("honors explicit tick watchdog timeout threshold", async () => { vi.useFakeTimers(); - try { - const client = new GatewayClient({ - tickWatchMinIntervalMs: 5, - tickWatchTimeoutMs: 50, - }); - const close = vi.fn(); - Object.assign( - client as unknown as { ws: unknown; tickIntervalMs: number; lastTick: number }, - { - ws: { - readyState: WebSocket.OPEN, - send: vi.fn(), - close, - }, - tickIntervalMs: 5, - lastTick: Date.now(), - }, - ); + const client = new GatewayClient({ + tickWatchMinIntervalMs: 5, + tickWatchTimeoutMs: 50, + }); + const close = vi.fn(); + installSyntheticSocket(client, vi.fn(), close); + Object.assign(client as unknown as { tickIntervalMs: number; lastTick: number }, { + tickIntervalMs: 5, + lastTick: Date.now(), + }); - ( - client as unknown as { - startTickWatch: () => void; - } - ).startTickWatch(); - await vi.advanceTimersByTimeAsync(20); - expect(close).not.toHaveBeenCalled(); + ( + client as unknown as { + startTickWatch: () => void; + } + ).startTickWatch(); + await vi.advanceTimersByTimeAsync(20); + expect(close).not.toHaveBeenCalled(); - await vi.advanceTimersByTimeAsync(35); - expect(close).toHaveBeenCalledWith(4000, "tick timeout"); - } finally { - vi.useRealTimers(); - } + await vi.advanceTimersByTimeAsync(35); + expect(close).toHaveBeenCalledWith(4000, "tick timeout"); }); test("clamps oversized tick watchdog intervals before scheduling", () => { vi.useFakeTimers(); - try { - const setIntervalSpy = vi.spyOn(globalThis, "setInterval"); - const client = new GatewayClient({ - tickWatchMinIntervalMs: 5, - }); - Object.assign( - client as unknown as { ws: unknown; tickIntervalMs: number; lastTick: number }, - { - ws: { - readyState: WebSocket.OPEN, - send: vi.fn(), - close: vi.fn(), - }, - tickIntervalMs: Number.MAX_SAFE_INTEGER, - lastTick: Date.now(), - }, - ); + const setIntervalSpy = vi.spyOn(globalThis, "setInterval"); + const client = new GatewayClient({ + tickWatchMinIntervalMs: 5, + }); + Object.assign(client as unknown as { ws: unknown; tickIntervalMs: number; lastTick: number }, { + ws: { + readyState: WebSocket.OPEN, + send: vi.fn(), + close: vi.fn(), + }, + tickIntervalMs: Number.MAX_SAFE_INTEGER, + lastTick: Date.now(), + }); - ( - client as unknown as { - startTickWatch: () => void; - } - ).startTickWatch(); + ( + client as unknown as { + startTickWatch: () => void; + } + ).startTickWatch(); - expect(setIntervalSpy).toHaveBeenCalledWith(expect.any(Function), MAX_SAFE_TIMEOUT_DELAY_MS); - client.stop(); - } finally { - vi.useRealTimers(); - } - }); - - test("times out unresolved requests and clears pending state", async () => { - vi.useFakeTimers(); - try { - const { client, send } = createOpenGatewayClient(25); - - const requestPromise = client.request("status"); - const requestExpectation = expect(requestPromise).rejects.toThrow( - "gateway request timeout for status", - ); - expect(send).toHaveBeenCalledTimes(1); - expect(getPendingCount(client)).toBe(1); - - await vi.advanceTimersByTimeAsync(25); - - await requestExpectation; - expect(getPendingCount(client)).toBe(0); - } finally { - vi.useRealTimers(); - } + expect(setIntervalSpy).toHaveBeenCalledWith(expect.any(Function), MAX_SAFE_TIMEOUT_DELAY_MS); + client.stop(); }); test("cleans pending request state when websocket send throws", async () => { - const client = new GatewayClient({ - requestTimeoutMs: 25, + const { client, send } = createOpenGatewayClient(25); + send.mockImplementationOnce(() => { + throw new Error("synthetic send failure"); }); - const sendError = new Error("synthetic send failure"); - ( - client as unknown as { - ws: WebSocket | { readyState: number; send: () => void; close: () => void }; - } - ).ws = { - readyState: WebSocket.OPEN, - send: vi.fn(() => { - throw sendError; - }), - close: vi.fn(), - }; await expect(client.request("status")).rejects.toThrow("synthetic send failure"); expect(getPendingCount(client)).toBe(0); }); - test("does not auto-timeout expectFinal requests", async () => { - vi.useFakeTimers(); - try { - const { client, send } = createOpenGatewayClient(25); - - const requestPromise = client.request("chat.send", undefined, { expectFinal: true }); - const isSettled = trackSettlement(requestPromise); - expect(send).toHaveBeenCalledTimes(1); - - await vi.advanceTimersByTimeAsync(25); - - expect(isSettled()).toBe(false); - expect(getPendingCount(client)).toBe(1); - - client.stop(); - await expect(requestPromise).rejects.toThrow("gateway client stopped"); - } finally { - vi.useRealTimers(); - } - }); - test("notifies accepted expectFinal requests while continuing to wait for final", async () => { - const client = new GatewayClient({ - requestTimeoutMs: 25, - }); - const send = vi.fn(); - ( - client as unknown as { - ws: WebSocket | { readyState: number; send: (data: string) => void; close: () => void }; - } - ).ws = { - readyState: WebSocket.OPEN, - send, - close: vi.fn(), - }; + const { client, send } = createOpenGatewayClient(25); const onAccepted = vi.fn(); const requestPromise = client.request<{ status: string }>("agent", undefined, { @@ -593,53 +514,29 @@ describe("GatewayClient", () => { }); const frame = JSON.parse(String(send.mock.calls[0]?.[0])) as { id: string }; - ( - client as unknown as { - handleMessage: (raw: string) => void; - } - ).handleMessage( - JSON.stringify({ - type: "res", - id: frame.id, - ok: true, - payload: { status: "accepted", runId: "run-1" }, - }), - ); + handleGatewayMessage(client, { + type: "res", + id: frame.id, + ok: true, + payload: { status: "accepted", runId: "run-1" }, + }); expect(onAccepted).toHaveBeenCalledWith({ status: "accepted", runId: "run-1" }); - expect((client as unknown as { pending: Map }).pending.size).toBe(1); + expect(getPendingCount(client)).toBe(1); - ( - client as unknown as { - handleMessage: (raw: string) => void; - } - ).handleMessage( - JSON.stringify({ - type: "res", - id: frame.id, - ok: true, - payload: { status: "ok" }, - }), - ); + handleGatewayMessage(client, { + type: "res", + id: frame.id, + ok: true, + payload: { status: "ok" }, + }); await expect(requestPromise).resolves.toEqual({ status: "ok" }); - expect((client as unknown as { pending: Map }).pending.size).toBe(0); + expect(getPendingCount(client)).toBe(0); }); test("aborts in-flight requests from caller AbortSignal", async () => { - const client = new GatewayClient({ - requestTimeoutMs: 25, - }); - const send = vi.fn(); - ( - client as unknown as { - ws: WebSocket | { readyState: number; send: () => void; close: () => void }; - } - ).ws = { - readyState: WebSocket.OPEN, - send, - close: vi.fn(), - }; + const { client, send } = createOpenGatewayClient(25); const controller = new AbortController(); const requestPromise = client.request("status", undefined, { @@ -647,20 +544,24 @@ describe("GatewayClient", () => { timeoutMs: null, }); expect(send).toHaveBeenCalledTimes(1); - expect((client as unknown as { pending: Map }).pending.size).toBe(1); + expect(getPendingCount(client)).toBe(1); controller.abort(); await expect(requestPromise).rejects.toThrow("gateway request aborted for status"); - expect((client as unknown as { pending: Map }).pending.size).toBe(0); + expect(getPendingCount(client)).toBe(0); }); - test("clamps oversized explicit request timeouts before scheduling", async () => { - vi.useFakeTimers(); - try { - const { client } = createOpenGatewayClient(25); + test.each([ + { defaultTimeoutMs: 25, options: { timeoutMs: 2_592_010_000 } }, + { defaultTimeoutMs: 2_592_010_000, options: undefined }, + ])( + "clamps oversized request timeouts before scheduling", + async ({ defaultTimeoutMs, options }) => { + vi.useFakeTimers(); + const { client } = createOpenGatewayClient(defaultTimeoutMs); - const requestPromise = client.request("status", undefined, { timeoutMs: 2_592_010_000 }); + const requestPromise = client.request("status", undefined, options); const isSettled = trackSettlement(requestPromise); await vi.advanceTimersByTimeAsync(1); @@ -670,56 +571,29 @@ describe("GatewayClient", () => { client.stop(); await expect(requestPromise).rejects.toThrow("gateway client stopped"); - } finally { - vi.useRealTimers(); - } - }); - - test("clamps oversized default request timeouts before scheduling", async () => { - vi.useFakeTimers(); - try { - const { client } = createOpenGatewayClient(2_592_010_000); - - const requestPromise = client.request("status"); - const isSettled = trackSettlement(requestPromise); - - await vi.advanceTimersByTimeAsync(1); - - expect(isSettled()).toBe(false); - expect(getPendingCount(client)).toBe(1); - - client.stop(); - await expect(requestPromise).rejects.toThrow("gateway client stopped"); - } finally { - vi.useRealTimers(); - } - }); + }, + ); test("clamps oversized stopAndWait timeouts before scheduling", async () => { vi.useFakeTimers(); - try { - const client = new GatewayClient({}); - const ws = { - readyState: WebSocket.OPEN, - close: vi.fn(), - terminate: vi.fn(), - }; - (client as unknown as { ws: unknown }).ws = ws; - const setTimeoutSpy = vi.spyOn(globalThis, "setTimeout"); + const client = new GatewayClient({}); + const ws = { + readyState: WebSocket.OPEN, + close: vi.fn(), + terminate: vi.fn(), + }; + (client as unknown as { ws: unknown }).ws = ws; + const setTimeoutSpy = vi.spyOn(globalThis, "setTimeout"); - const stopPromise = client.stopAndWait({ timeoutMs: Number.MAX_SAFE_INTEGER }); + const stopPromise = client.stopAndWait({ timeoutMs: Number.MAX_SAFE_INTEGER }); - await vi.advanceTimersByTimeAsync(1); - expect(ws.terminate).not.toHaveBeenCalled(); - expect(setTimeoutSpy).toHaveBeenCalledWith(expect.any(Function), MAX_SAFE_TIMEOUT_DELAY_MS); + await vi.advanceTimersByTimeAsync(1); + expect(ws.terminate).not.toHaveBeenCalled(); + expect(setTimeoutSpy).toHaveBeenCalledWith(expect.any(Function), MAX_SAFE_TIMEOUT_DELAY_MS); - await vi.advanceTimersByTimeAsync(249); - await expect(stopPromise).resolves.toBeUndefined(); - expect(ws.terminate).toHaveBeenCalledTimes(1); - } finally { - vi.useRealTimers(); - vi.restoreAllMocks(); - } + await vi.advanceTimersByTimeAsync(249); + await expect(stopPromise).resolves.toBeUndefined(); + expect(ws.terminate).toHaveBeenCalledTimes(1); }); test("rejects mismatched tls fingerprint", async () => { diff --git a/packages/gateway-client/src/protocol-client-types.ts b/packages/gateway-client/src/protocol-client-types.ts new file mode 100644 index 000000000000..459399a232d1 --- /dev/null +++ b/packages/gateway-client/src/protocol-client-types.ts @@ -0,0 +1,129 @@ +import type { ErrorShape, EventFrame, HelloOk } from "@openclaw/gateway-protocol"; +export type GatewayProtocolSocket = { + isOpen: () => boolean; + send: (data: string) => void; + close: (code?: number, reason?: string) => void; +}; +export type GatewayProtocolSocketHandlers = { + open: () => void; + message: (data: string) => void; + close: (code: number, reason: string) => void; + error: (error: Error) => void; +}; +export type GatewayProtocolRequestOptions = { + timeoutMs?: number | null; + expectFinal?: boolean; + onAccepted?: (payload: unknown) => void; + signal?: AbortSignal; +}; +type GatewayProtocolConnectContext = { + generation: number; + nonce: string | null; + plan: TPlan; +}; +export type GatewayProtocolCloseContext = { + code: number; + reason: string; + generation: number; + socketOpened: boolean; + helloReceived: boolean; + connectRequestSent: boolean; + connectFailure?: { error: Error; reconnectDelayMs?: number }; +}; +type GatewayProtocolConnectDecision = { + closeCode: number; + closeReason: string; + reconnectDelayMs?: number; + stop?: boolean; + error?: Error; +}; +type GatewayProtocolCloseDecision = { + retry: boolean; + notify: boolean; + reconnectDelayMs?: number; + pendingError?: Error; +}; +export type GatewayProtocolTiming = { + phase: + | "socket-open" + | "challenge" + | "fallback" + | "device-identity-ready" + | "connect-plan-ready" + | "request-sent" + | "hello" + | "failed"; + generation: number; + durationMs: number; + phaseDurationMs: number; + hasChallenge: boolean; + usedFallback: boolean; + plan?: TPlan; + detail?: unknown; +}; +export type GatewayProtocolRequestTiming = { + id: string; + method: string; + ok: boolean; + durationMs: number; + startedAtMs: number; + endedAtMs: number; + errorCode?: string; +}; +export type GatewayProtocolClientOptions = { + createSocket: (handlers: GatewayProtocolSocketHandlers) => GatewayProtocolSocket; + createRequestId: () => string; + createRequestError?: (error: Partial) => GatewayProtocolRequestError; + createRequestTimeoutError?: (method: string, timeoutMs: number) => Error; + createRequestAbortError?: (method: string) => Error; + buildConnectPlan: (params: { + nonce: string | null; + generation: number; + }) => TPlan | Promise; + buildConnectParams: (plan: TPlan) => unknown; + onConnectPlanError?: (error: Error) => GatewayProtocolConnectDecision; + onConnectHello?: (hello: HelloOk, context: GatewayProtocolConnectContext) => void; + onHello?: (hello: HelloOk) => void; + onConnectFailure?: ( + error: GatewayProtocolRequestError, + context: GatewayProtocolConnectContext, + ) => GatewayProtocolConnectDecision; + resolveClose: (context: GatewayProtocolCloseContext) => GatewayProtocolCloseDecision; + onClose?: (context: GatewayProtocolCloseContext, decision: GatewayProtocolCloseDecision) => void; + notifyStoppedClose?: boolean; + onConnectError?: (error: Error) => void; + onSocketFactoryError?: (error: Error) => void; + onParseError?: (error: unknown) => void; + onEvent?: (event: EventFrame) => void; + onGap?: (info: { expected: number; received: number }) => void; + onActivity?: () => void; + onTiming?: (timing: GatewayProtocolTiming) => void; + onRequestTiming?: (timing: GatewayProtocolRequestTiming) => void; + onCallbackError?: (label: string, error: unknown) => void; + handshake: + | { mode: "fallback"; timeoutMs: number } + | { + mode: "require-challenge"; + timeoutMs: number; + timeoutMessage?: (elapsedMs: number) => string; + }; + reconnect: { initialMs: number; multiplier: number; maxMs: number }; + requestTimeoutMs?: number; + nowMs?: () => number; + rethrowSocketFactoryError?: (error: Error) => boolean; +}; +export class GatewayProtocolRequestError extends Error { + readonly code: string; + readonly details?: unknown; + readonly retryable?: boolean; + readonly retryAfterMs?: number; + + constructor(error: Partial) { + super(error.message ?? "request failed"); + this.name = "GatewayProtocolRequestError"; + this.code = error.code ?? "UNAVAILABLE"; + this.details = error.details; + this.retryable = error.retryable; + this.retryAfterMs = error.retryAfterMs; + } +} diff --git a/packages/gateway-client/src/protocol-client.ts b/packages/gateway-client/src/protocol-client.ts new file mode 100644 index 000000000000..6fc2c30c474e --- /dev/null +++ b/packages/gateway-client/src/protocol-client.ts @@ -0,0 +1,462 @@ +import type { EventFrame, HelloOk } from "@openclaw/gateway-protocol"; +import { + isGatewayEventFrame, + isGatewayResponseFrame, +} from "@openclaw/gateway-protocol/frame-guards"; +import { RetrySupervisor, sleepWithAbort } from "@openclaw/retry"; +import { + GatewayProtocolRequestError, + type GatewayProtocolClientOptions, + type GatewayProtocolCloseContext, + type GatewayProtocolRequestOptions, + type GatewayProtocolSocket, + type GatewayProtocolTiming, +} from "./protocol-client-types.js"; +import { GatewayProtocolRequests } from "./protocol-requests.js"; + +export * from "./protocol-client-types.js"; + +type ConnectTimingState = { + generation: number; + startedAtMs: number; + lastAtMs: number; + hasChallenge: boolean; + usedFallback: boolean; +}; +type CloseSnapshot = Omit; + +/** + * Browser-safe gateway wire client. Environment adapters own transport and auth + * policy; this class owns the single socket/handshake/reconnect/frame state machine. + */ +export class GatewayProtocolClient { + private socket: GatewayProtocolSocket | null = null; + private readonly requests: GatewayProtocolRequests; + private listeners = new Set<(event: EventFrame) => void>(); + private stopped = true; + private generation = 0; + private lastSeq: number | null = null; + private connectNonce: string | null = null; + private connectSent = false; + private connectRequestSent = false; + private handshakeTimer: ReturnType | null = null; + private readonly reconnectSupervisor: RetrySupervisor; + private socketOpened = false; + private helloReceived = false; + private connectFailure: GatewayProtocolCloseContext["connectFailure"]; + private connectTiming: ConnectTimingState | null = null; + private stoppedSocket?: { socket: GatewayProtocolSocket; context: CloseSnapshot }; + + constructor(private readonly opts: GatewayProtocolClientOptions) { + this.reconnectSupervisor = new RetrySupervisor({ + initialMs: opts.reconnect.initialMs, + maxMs: opts.reconnect.maxMs, + factor: opts.reconnect.multiplier, + jitter: 0, + }); + this.requests = new GatewayProtocolRequests(opts); + } + + get connected(): boolean { + return this.socket?.isOpen() ?? false; + } + + get hasPendingRequests(): boolean { + return this.requests.hasPending; + } + + get connecting(): boolean { + return this.connectSent && !this.helloReceived; + } + + get hasUnboundedPendingRequests(): boolean { + return this.requests.hasUnboundedPending; + } + + start(): void { + this.stopped = false; + this.reconnectSupervisor.cancel(); + this.connect(); + } + + stop(): void { + this.stopped = true; + this.clearHandshakeTimer(); + this.reconnectSupervisor.reset(); + const socket = this.socket; + if (socket && this.opts.notifyStoppedClose) { + // Node callers observe the transport's final close during explicit stop; + // browser callers intentionally suppress it. + this.stoppedSocket = { socket, context: this.closeContext() }; + } + this.socket = null; + this.connectFailure = undefined; + this.connectTiming = null; + this.requests.flush(new Error("gateway client stopped")); + if (!socket) { + return; + } + socket.close(); + } + + request( + method: string, + params?: unknown, + options?: GatewayProtocolRequestOptions, + ): Promise { + const socket = this.socket; + if (!socket?.isOpen()) { + return Promise.reject(new Error("gateway not connected")); + } + if (typeof method !== "string" || method.length === 0) { + return Promise.reject(new Error("invalid request frame: method must be a non-empty string")); + } + return this.requests.request(socket, method, params, options); + } + + addEventListener(listener: (event: EventFrame) => void): () => void { + this.listeners.add(listener); + return () => this.listeners.delete(listener); + } + + closeSocket(code?: number, reason?: string): void { + this.socket?.close(code, reason); + } + + resetReconnectBackoff(initialMs: number): void { + this.reconnectSupervisor.reset(initialMs); + } + + recordTiming( + phase: GatewayProtocolTiming["phase"], + generation: number, + plan?: TPlan, + detail?: unknown, + ): void { + const now = this.nowMs(); + const state = this.connectTiming; + if (!state || state.generation !== generation) { + return; + } + state.hasChallenge ||= phase === "challenge"; + state.usedFallback ||= phase === "fallback"; + this.invoke("connect timing", () => + this.opts.onTiming?.({ + phase, + generation, + durationMs: Math.max(0, now - state.startedAtMs), + phaseDurationMs: Math.max(0, now - state.lastAtMs), + hasChallenge: state.hasChallenge, + usedFallback: state.usedFallback, + plan, + detail, + }), + ); + state.lastAtMs = now; + if (phase === "hello" || phase === "failed") { + this.connectTiming = null; + } + } + + private connect(): void { + if (this.stopped) { + return; + } + const generation = this.generation + 1; + this.connectNonce = null; + this.connectSent = false; + this.connectRequestSent = false; + this.socketOpened = false; + this.helloReceived = false; + this.connectFailure = undefined; + let socket: GatewayProtocolSocket; + try { + socket = this.opts.createSocket({ + open: () => this.handleOpen(socket, generation), + message: (data) => this.handleMessage(socket, generation, data), + close: (code, reason) => this.handleClose(socket, generation, code, reason), + error: (error) => this.handleSocketError(socket, generation, error), + }); + } catch (error) { + const normalized = error instanceof Error ? error : new Error(String(error)); + this.opts.onSocketFactoryError?.(normalized); + this.opts.onConnectError?.(normalized); + if (this.opts.rethrowSocketFactoryError?.(normalized)) { + throw normalized; + } + return; + } + this.generation = generation; + this.socket = socket; + const now = this.nowMs(); + this.connectTiming = { + generation, + startedAtMs: now, + lastAtMs: now, + hasChallenge: false, + usedFallback: false, + }; + } + + private handleOpen(socket: GatewayProtocolSocket, generation: number): void { + if (!this.isActive(socket, generation)) { + return; + } + this.socketOpened = true; + this.recordTiming("socket-open", generation); + if (this.connectNonce) { + this.sendConnect(socket, generation); + return; + } + this.armHandshakeTimer(socket, generation); + } + + private armHandshakeTimer(socket: GatewayProtocolSocket, generation: number): void { + this.clearHandshakeTimer(); + const armedAt = Date.now(); + this.handshakeTimer = setTimeout(() => { + this.handshakeTimer = null; + if (!this.isActive(socket, generation) || this.connectSent || !socket.isOpen()) { + return; + } + if (this.opts.handshake.mode === "fallback") { + this.recordTiming("fallback", generation); + this.sendConnect(socket, generation); + return; + } + const elapsedMs = Date.now() - armedAt; + const error = new Error( + this.opts.handshake.timeoutMessage?.(elapsedMs) ?? + `gateway connect challenge timeout after ${elapsedMs}ms`, + ); + this.opts.onConnectError?.(error); + socket.close(1008, "connect challenge timeout"); + }, this.opts.handshake.timeoutMs); + this.handshakeTimer.unref?.(); + } + + private sendConnect(socket: GatewayProtocolSocket, generation: number): void { + if (!this.isActive(socket, generation) || !socket.isOpen() || this.connectSent) { + return; + } + this.connectSent = true; + this.clearHandshakeTimer(); + let planOrPromise: TPlan | Promise; + try { + planOrPromise = this.opts.buildConnectPlan({ nonce: this.connectNonce, generation }); + } catch (error) { + this.handleConnectPlanError(socket, generation, error); + return; + } + if (planOrPromise instanceof Promise) { + void planOrPromise + .then((plan) => this.sendConnectPlan(socket, generation, plan)) + .catch((error: unknown) => this.handleConnectPlanError(socket, generation, error)); + return; + } + this.sendConnectPlan(socket, generation, planOrPromise); + } + + private handleConnectPlanError( + socket: GatewayProtocolSocket, + generation: number, + error: unknown, + ): void { + if (!this.isActive(socket, generation)) { + return; + } + const normalized = error instanceof Error ? error : new Error(String(error)); + const outcome = this.opts.onConnectPlanError?.(normalized) ?? { + closeCode: 1008, + closeReason: "connect failed", + }; + this.opts.onConnectError?.(outcome.error ?? normalized); + if (outcome.stop) { + this.stopped = true; + } + socket.close(outcome.closeCode, outcome.closeReason); + } + + private sendConnectPlan(socket: GatewayProtocolSocket, generation: number, plan: TPlan): void { + if (!this.isActive(socket, generation) || !socket.isOpen()) { + return; + } + const context = { generation, nonce: this.connectNonce, plan }; + this.recordTiming("connect-plan-ready", generation, plan); + this.recordTiming("request-sent", generation, plan); + this.connectRequestSent = true; + void this.requests + .request(socket, "connect", this.opts.buildConnectParams(plan)) + .then((hello) => { + if (!this.isActive(socket, generation)) { + return; + } + this.helloReceived = true; + this.connectFailure = undefined; + this.reconnectSupervisor.reset(); + this.recordTiming("hello", generation, plan); + this.opts.onConnectHello?.(hello, context); + this.invoke("hello", () => this.opts.onHello?.(hello)); + }) + .catch((error: unknown) => { + if (!this.isActive(socket, generation)) { + return; + } + const requestError = + error instanceof GatewayProtocolRequestError + ? error + : new GatewayProtocolRequestError({ message: String(error) }); + const outcome = this.opts.onConnectFailure?.(requestError, context) ?? { + closeCode: 1008, + closeReason: "connect failed", + }; + this.connectFailure = { error: requestError, reconnectDelayMs: outcome.reconnectDelayMs }; + if (outcome.stop) { + this.stopped = true; + } + socket.close(outcome.closeCode, outcome.closeReason); + }); + } + + private handleMessage(socket: GatewayProtocolSocket, generation: number, raw: string): void { + if (!this.isActive(socket, generation)) { + return; + } + let parsed: unknown; + try { + parsed = JSON.parse(raw); + } catch (error) { + this.opts.onParseError?.(error); + return; + } + if (isGatewayEventFrame(parsed)) { + this.opts.onActivity?.(); + if (parsed.event === "connect.challenge") { + const payload = parsed.payload as { nonce?: unknown } | undefined; + const nonce = typeof payload?.nonce === "string" ? payload.nonce.trim() : ""; + if (!nonce) { + if (this.opts.handshake.mode === "require-challenge") { + const error = new Error("gateway connect challenge missing nonce"); + this.opts.onConnectError?.(error); + socket.close(1008, "connect challenge missing nonce"); + } + return; + } + this.connectNonce = nonce; + this.recordTiming("challenge", generation); + this.sendConnect(socket, generation); + return; + } + const seq = typeof parsed.seq === "number" ? parsed.seq : null; + if (seq !== null) { + if (this.lastSeq !== null && seq > this.lastSeq + 1) { + const expected = this.lastSeq + 1; + this.invoke("gap", () => this.opts.onGap?.({ expected, received: seq })); + } + this.lastSeq = seq; + } + this.invoke("event", () => this.opts.onEvent?.(parsed)); + for (const listener of this.listeners) { + this.invoke("event listener", () => listener(parsed)); + } + return; + } + if (!isGatewayResponseFrame(parsed)) { + return; + } + this.opts.onActivity?.(); + this.requests.handleResponse(parsed); + } + + private handleClose( + socket: GatewayProtocolSocket, + generation: number, + code: number, + reason: string, + ): void { + if (this.socket !== socket) { + if (this.stoppedSocket?.socket === socket) { + const context = { ...this.stoppedSocket.context, code, reason }; + this.stoppedSocket = undefined; + this.invoke("close", () => this.opts.onClose?.(context, { retry: false, notify: true })); + } + return; + } + this.socket = null; + this.clearHandshakeTimer(); + const context: GatewayProtocolCloseContext = { + ...this.closeContext(), + code, + reason, + generation, + }; + this.connectFailure = undefined; + const decision = this.opts.resolveClose(context); + this.requests.flush( + decision.pendingError ?? + context.connectFailure?.error ?? + new Error(`gateway closed (${code}): ${reason}`), + ); + this.invoke("close", () => this.opts.onClose?.(context, decision)); + if (decision.retry && !this.stopped) { + this.scheduleReconnect(decision.reconnectDelayMs ?? context.connectFailure?.reconnectDelayMs); + } + } + + private handleSocketError(socket: GatewayProtocolSocket, generation: number, error: Error): void { + if (!this.isActive(socket, generation) || this.connectSent) { + return; + } + this.opts.onConnectError?.(error); + } + + private scheduleReconnect(overrideMs?: number): void { + if (overrideMs !== undefined) { + // Retry-After is a floor for this wait, not a failed attempt. Preserve + // the exponential sequence for the next transport failure. + this.reconnectSupervisor.nextDelayOverrideMs = overrideMs; + } + const retry = this.reconnectSupervisor.next(); + if (!retry) { + return; + } + // Ignore cancelled sleeps only; reconnect start failures stay observable. + void sleepWithAbort(retry.delayMs, retry.signal).then( + () => this.connect(), + () => {}, + ); + } + + private closeContext(): CloseSnapshot { + return { + generation: this.generation, + socketOpened: this.socketOpened, + helloReceived: this.helloReceived, + connectRequestSent: this.connectRequestSent, + connectFailure: this.connectFailure, + }; + } + + private isActive(socket: GatewayProtocolSocket, generation: number): boolean { + return !this.stopped && this.socket === socket && this.generation === generation; + } + + private nowMs(): number { + return this.opts.nowMs?.() ?? Date.now(); + } + + private clearHandshakeTimer(): void { + if (this.handshakeTimer) { + clearTimeout(this.handshakeTimer); + this.handshakeTimer = null; + } + } + + private invoke(label: string, callback: () => void): void { + try { + callback(); + } catch (error) { + this.opts.onCallbackError?.(label, error); + } + } +} diff --git a/packages/gateway-client/src/protocol-requests.ts b/packages/gateway-client/src/protocol-requests.ts new file mode 100644 index 000000000000..760631b3be84 --- /dev/null +++ b/packages/gateway-client/src/protocol-requests.ts @@ -0,0 +1,164 @@ +import type { ResponseFrame } from "@openclaw/gateway-protocol"; +import { + GatewayProtocolRequestError, + type GatewayProtocolClientOptions, + type GatewayProtocolRequestOptions, + type GatewayProtocolSocket, +} from "./protocol-client-types.js"; + +type Pending = { + resolve: (value: unknown) => void; + reject: (error: Error) => void; + expectFinal: boolean; + acceptedNotified: boolean; + onAccepted?: (payload: unknown) => void; + cleanup?: () => void; + unbounded: boolean; + method: string; + startedAtMs: number; +}; + +export class GatewayProtocolRequests { + private readonly pending = new Map(); + + constructor(private readonly opts: GatewayProtocolClientOptions) {} + + get hasPending(): boolean { + return this.pending.size > 0; + } + + get hasUnboundedPending(): boolean { + return [...this.pending.values()].some((pending) => pending.unbounded); + } + + request( + socket: GatewayProtocolSocket, + method: string, + params?: unknown, + options?: GatewayProtocolRequestOptions, + ): Promise { + const id = this.opts.createRequestId(); + const frame = { type: "req", id, method, params }; + const timeoutMs = + options?.timeoutMs === null ? undefined : (options?.timeoutMs ?? this.opts.requestTimeoutMs); + return new Promise((resolve, reject) => { + let timeout: ReturnType | undefined; + const pending: Pending = { + resolve: (value) => resolve(value as T), + reject, + expectFinal: options?.expectFinal === true, + acceptedNotified: false, + onAccepted: options?.onAccepted, + unbounded: timeoutMs === undefined, + method, + startedAtMs: this.opts.nowMs?.() ?? Date.now(), + }; + const onAbort = () => { + this.pending.delete(id); + if (timeout) { + clearTimeout(timeout); + } + this.finishTiming(id, pending, false, "CLIENT_ABORTED"); + reject( + this.opts.createRequestAbortError?.(method) ?? + new Error(`gateway request aborted for ${method}`), + ); + }; + const cleanup = () => { + if (timeout) { + clearTimeout(timeout); + } + options?.signal?.removeEventListener("abort", onAbort); + }; + if (options?.signal?.aborted) { + reject( + this.opts.createRequestAbortError?.(method) ?? + new Error(`gateway request aborted for ${method}`), + ); + return; + } + pending.cleanup = cleanup; + if (timeoutMs !== undefined && timeoutMs >= 0) { + timeout = setTimeout(() => { + this.pending.delete(id); + options?.signal?.removeEventListener("abort", onAbort); + this.finishTiming(id, pending, false, "CLIENT_TIMEOUT"); + reject( + this.opts.createRequestTimeoutError?.(method, timeoutMs) ?? + new Error(`gateway request timed out after ${timeoutMs}ms: ${method}`), + ); + }, timeoutMs); + timeout.unref?.(); + } + options?.signal?.addEventListener("abort", onAbort, { once: true }); + this.pending.set(id, pending); + try { + socket.send(JSON.stringify(frame)); + } catch (error) { + this.pending.delete(id); + cleanup(); + this.finishTiming(id, pending, false, "CLIENT_SEND_ERROR"); + reject(error instanceof Error ? error : new Error(String(error))); + } + }); + } + + handleResponse(frame: ResponseFrame): void { + const pending = this.pending.get(frame.id); + if (!pending) { + return; + } + const status = (frame.payload as { status?: unknown } | undefined)?.status; + if (pending.expectFinal && status === "accepted") { + if (!pending.acceptedNotified) { + pending.acceptedNotified = true; + this.invoke("accepted", () => pending.onAccepted?.(frame.payload)); + } + return; + } + this.pending.delete(frame.id); + pending.cleanup?.(); + if (frame.ok) { + this.finishTiming(frame.id, pending, true); + pending.resolve(frame.payload); + return; + } + this.finishTiming(frame.id, pending, false, frame.error?.code); + pending.reject( + this.opts.createRequestError?.(frame.error ?? {}) ?? + new GatewayProtocolRequestError(frame.error ?? {}), + ); + } + + flush(error: Error): void { + for (const [id, pending] of this.pending) { + this.finishTiming(id, pending, false, "CLIENT_CLOSED"); + pending.cleanup?.(); + pending.reject(error); + } + this.pending.clear(); + } + + private finishTiming(id: string, pending: Pending, ok: boolean, errorCode?: string): void { + const endedAtMs = this.opts.nowMs?.() ?? Date.now(); + this.invoke("request timing", () => + this.opts.onRequestTiming?.({ + id, + method: pending.method, + ok, + durationMs: Math.max(0, endedAtMs - pending.startedAtMs), + startedAtMs: pending.startedAtMs, + endedAtMs, + errorCode, + }), + ); + } + + private invoke(label: string, callback: () => void): void { + try { + callback(); + } catch (error) { + this.opts.onCallbackError?.(label, error); + } + } +} diff --git a/packages/gateway-client/src/reconnect-policy.ts b/packages/gateway-client/src/reconnect-policy.ts new file mode 100644 index 000000000000..2a9072ce1b50 --- /dev/null +++ b/packages/gateway-client/src/reconnect-policy.ts @@ -0,0 +1,48 @@ +import { + ConnectErrorDetailCodes, + readConnectErrorDetailCode, + readPairingConnectErrorDetails, +} from "@openclaw/gateway-protocol/connect-error-details"; + +const NON_RECOVERABLE_AUTH_ERRORS = new Set([ + ConnectErrorDetailCodes.AUTH_TOKEN_MISSING, + ConnectErrorDetailCodes.AUTH_BOOTSTRAP_TOKEN_INVALID, + ConnectErrorDetailCodes.AUTH_PASSWORD_MISSING, + ConnectErrorDetailCodes.AUTH_PASSWORD_MISMATCH, + ConnectErrorDetailCodes.AUTH_RATE_LIMITED, + ConnectErrorDetailCodes.AUTH_DEVICE_TOKEN_MISMATCH, + ConnectErrorDetailCodes.AUTH_SCOPE_MISMATCH, + ConnectErrorDetailCodes.PAIRING_REQUIRED, + ConnectErrorDetailCodes.CONTROL_UI_DEVICE_IDENTITY_REQUIRED, + ConnectErrorDetailCodes.DEVICE_IDENTITY_REQUIRED, +]); + +export function shouldPauseGatewayReconnect(params: { + details?: unknown; + deviceTokenRetryPending?: boolean; + tokenMismatchIsTerminal?: boolean; + protocolMismatchIsTerminal?: boolean; + clientVersionMismatchIsTerminal?: boolean; +}): boolean { + const code = readConnectErrorDetailCode(params.details); + if (!code) { + return false; + } + const pairing = readPairingConnectErrorDetails(params.details); + if ( + code === ConnectErrorDetailCodes.PAIRING_REQUIRED && + (pairing?.pauseReconnect === false || pairing?.recommendedNextStep === "wait_then_retry") + ) { + return false; + } + if (code === ConnectErrorDetailCodes.AUTH_TOKEN_MISMATCH) { + return params.tokenMismatchIsTerminal === true && !params.deviceTokenRetryPending; + } + return ( + NON_RECOVERABLE_AUTH_ERRORS.has(code) || + (params.protocolMismatchIsTerminal === true && + code === ConnectErrorDetailCodes.PROTOCOL_MISMATCH) || + (params.clientVersionMismatchIsTerminal === true && + code === ConnectErrorDetailCodes.CLIENT_VERSION_MISMATCH) + ); +} diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 0360ff2170b1..84a9b2c5030c 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -1991,12 +1991,12 @@ importers: '@openclaw/gateway-protocol': specifier: workspace:* version: link:../gateway-protocol + '@openclaw/net-policy': + specifier: workspace:* + version: link:../net-policy '@openclaw/retry': specifier: workspace:* version: link:../retry - ipaddr.js: - specifier: 2.4.0 - version: 2.4.0 ws: specifier: 8.21.0 version: 8.21.0 @@ -2097,6 +2097,9 @@ importers: ui: dependencies: + '@openclaw/gateway-client': + specifier: workspace:* + version: link:../packages/gateway-client '@codemirror/commands': specifier: 6.10.4 version: 6.10.4 diff --git a/src/gateway/client.test.ts b/src/gateway/client.test.ts index d8886ec51996..f3ce8c85ed5a 100644 --- a/src/gateway/client.test.ts +++ b/src/gateway/client.test.ts @@ -276,6 +276,10 @@ beforeAll(async () => { await loadGatewayClientModule(); }); +afterEach(() => { + vi.useRealTimers(); +}); + describe("GatewayClient security checks", () => { const envSnapshot = captureEnv([ "OPENCLAW_ALLOW_INSECURE_PRIVATE_WS", @@ -822,25 +826,21 @@ describe("GatewayClient close handling", () => { it("keeps a managed reconnect timer after gateway restart closes", async () => { vi.useFakeTimers(); - try { - const client = new GatewayClient({ - url: "ws://127.0.0.1:18789", - }); + const client = new GatewayClient({ + url: "ws://127.0.0.1:18789", + }); - client.start(); - getLatestWs().emitClose(1012, "service restart"); + client.start(); + getLatestWs().emitClose(1012, "service restart"); - expect(wsInstances).toHaveLength(1); - await vi.advanceTimersByTimeAsync(999); - expect(wsInstances).toHaveLength(1); + expect(wsInstances).toHaveLength(1); + await vi.advanceTimersByTimeAsync(999); + expect(wsInstances).toHaveLength(1); - await vi.advanceTimersByTimeAsync(1); + await vi.advanceTimersByTimeAsync(1); - expect(wsInstances).toHaveLength(2); - client.stop(); - } finally { - vi.useRealTimers(); - } + expect(wsInstances).toHaveLength(2); + client.stop(); }); it("reconnects quietly after one clean pre-hello close with a pending connect", async () => { @@ -987,99 +987,86 @@ describe("GatewayClient close handling", () => { it("clears pending reconnect timers on stop", async () => { vi.useFakeTimers(); - try { - const client = new GatewayClient({ - url: "ws://127.0.0.1:18789", - }); + const client = new GatewayClient({ + url: "ws://127.0.0.1:18789", + }); - client.start(); - getLatestWs().emitClose(1012, "service restart"); - client.stop(); + client.start(); + getLatestWs().emitClose(1012, "service restart"); + client.stop(); - await vi.advanceTimersByTimeAsync(30_000); + await vi.advanceTimersByTimeAsync(30_000); - expect(wsInstances).toHaveLength(1); - } finally { - vi.useRealTimers(); - } + expect(wsInstances).toHaveLength(1); }); it("force-terminates a lingering socket after stop", async () => { vi.useFakeTimers(); - try { - const client = new GatewayClient({ - url: "ws://127.0.0.1:18789", - }); + const client = new GatewayClient({ + url: "ws://127.0.0.1:18789", + }); - client.start(); - const ws = getLatestWs(); - ws.autoCloseOnClose = false; + client.start(); + const ws = getLatestWs(); + ws.autoCloseOnClose = false; - client.stop(); + client.stop(); - expect(ws.closeCalls).toBe(1); - expect(ws.terminateCalls).toBe(0); + expect(ws.closeCalls).toBe(1); + expect(ws.terminateCalls).toBe(0); - await vi.advanceTimersByTimeAsync(250); + await vi.advanceTimersByTimeAsync(250); - expect(ws.terminateCalls).toBe(1); - } finally { - vi.useRealTimers(); - } + expect(ws.terminateCalls).toBe(1); }); it("does not force-terminate a socket that closes during stop", async () => { vi.useFakeTimers(); - try { - const client = new GatewayClient({ - url: "ws://127.0.0.1:18789", - }); + const onClose = vi.fn(); + const client = new GatewayClient({ + url: "ws://127.0.0.1:18789", + onClose, + }); - client.start(); - const ws = getLatestWs(); + client.start(); + const ws = getLatestWs(); - client.stop(); + client.stop(); - expect(ws.closeCalls).toBe(1); - await vi.advanceTimersByTimeAsync(250); + expect(ws.closeCalls).toBe(1); + expect(onClose).toHaveBeenCalledOnce(); + await vi.advanceTimersByTimeAsync(250); - expect(ws.terminateCalls).toBe(0); - } finally { - vi.useRealTimers(); - } + expect(ws.terminateCalls).toBe(0); }); it("waits for a lingering socket to terminate in stopAndWait", async () => { vi.useFakeTimers(); - try { - const client = new GatewayClient({ - url: "ws://127.0.0.1:18789", - }); + const client = new GatewayClient({ + url: "ws://127.0.0.1:18789", + }); - client.start(); - const ws = getLatestWs(); - ws.autoCloseOnClose = false; + client.start(); + const ws = getLatestWs(); + ws.autoCloseOnClose = false; - let settled = false; - const stopPromise = client.stopAndWait().then(() => { - settled = true; - }); + let settled = false; + const stopPromise = client.stopAndWait().then(() => { + settled = true; + }); - expect(ws.closeCalls).toBe(1); - expect(settled).toBe(false); + expect(ws.closeCalls).toBe(1); + expect(settled).toBe(false); - await vi.advanceTimersByTimeAsync(249); - expect(ws.terminateCalls).toBe(0); - expect(settled).toBe(false); + await vi.advanceTimersByTimeAsync(249); + expect(ws.terminateCalls).toBe(0); + expect(settled).toBe(false); - await vi.advanceTimersByTimeAsync(1); - await stopPromise; + await vi.advanceTimersByTimeAsync(1); + await stopPromise; - expect(ws.terminateCalls).toBe(1); - expect(settled).toBe(true); - } finally { - vi.useRealTimers(); - } + expect(ws.terminateCalls).toBe(1); + expect(settled).toBe(true); }); it("does not clear persisted device auth when explicit shared token is provided", () => { @@ -1347,6 +1334,8 @@ describe("GatewayClient connect auth payload", () => { expect(logDebugMock).toHaveBeenCalledWith( "gateway client hello-ok handler error: Error: hello callback failed", ); + ws.emitClose(1012, "service restart"); + expect(onConnectError).not.toHaveBeenCalled(); } finally { client.stop(); } @@ -1536,6 +1525,23 @@ describe("GatewayClient connect auth payload", () => { client.stop(); }); + it("reports a transport close while the connect request is pending", () => { + const onConnectError = vi.fn(); + const client = new GatewayClient({ + url: "ws://127.0.0.1:18789", + token: "shared-token", + onConnectError, + }); + + const { ws } = startClientAndConnect({ client }); + ws.emitClose(1006, "socket lost"); + + expect(firstMockArg(onConnectError, "connect error")).toMatchObject({ + message: "gateway closed (1006): socket lost", + }); + client.stop(); + }); + it("logs stopped connect handshakes at debug level during teardown", async () => { const onConnectError = vi.fn(); const client = new GatewayClient({ diff --git a/test/vitest/vitest.shared.config.ts b/test/vitest/vitest.shared.config.ts index a816def64570..2130ded548c5 100644 --- a/test/vitest/vitest.shared.config.ts +++ b/test/vitest/vitest.shared.config.ts @@ -193,6 +193,10 @@ export const sharedVitestConfig = { find: "@openclaw/whatsapp/api.js", replacement: path.join(repoRoot, "extensions", "whatsapp", "api.ts"), }, + { + find: "@openclaw/gateway-client/browser", + replacement: path.join(repoRoot, "packages", "gateway-client", "src", "browser.ts"), + }, { find: "@openclaw/gateway-client/readiness", replacement: path.join(repoRoot, "packages", "gateway-client", "src", "readiness.ts"), diff --git a/tsconfig.json b/tsconfig.json index 8d7a4f83aee7..f9a2613f9f8c 100644 --- a/tsconfig.json +++ b/tsconfig.json @@ -66,6 +66,7 @@ ], "@openclaw/model-catalog-core/*": ["./packages/model-catalog-core/src/*"], "@openclaw/gateway-client": ["./packages/gateway-client/src/index.ts"], + "@openclaw/gateway-client/browser": ["./packages/gateway-client/src/browser.ts"], "@openclaw/gateway-client/*": ["./packages/gateway-client/src/*"], "@openclaw/gateway-protocol": ["./packages/gateway-protocol/src/index.ts"], "@openclaw/gateway-protocol/client-info": [ diff --git a/tsdown.config.ts b/tsdown.config.ts index 6f31f93a9d4d..99fcc222bf0b 100644 --- a/tsdown.config.ts +++ b/tsdown.config.ts @@ -383,6 +383,7 @@ function buildGatewayClientDistEntries(): Record { // Keep package entrypoints explicit so package.json exports and root build // config cannot drift when client internals are split again. index: "packages/gateway-client/src/index.ts", + browser: "packages/gateway-client/src/browser.ts", readiness: "packages/gateway-client/src/readiness.ts", timeouts: "packages/gateway-client/src/timeouts.ts", }; @@ -566,11 +567,8 @@ function shouldExternalizeGatewayProtocolDependency(id: string): boolean { } function shouldExternalizeGatewayClientDependency(id: string): boolean { - return ( - id === "ws" || - id.startsWith("ws/") || - id === "@openclaw/gateway-protocol" || - id.startsWith("@openclaw/gateway-protocol/") + return ["ws", "@openclaw/net-policy", "@openclaw/gateway-protocol"].some( + (dependency) => id === dependency || id.startsWith(`${dependency}/`), ); } diff --git a/ui/package.json b/ui/package.json index 7c588c165fc2..f9b55011e975 100644 --- a/ui/package.json +++ b/ui/package.json @@ -20,6 +20,7 @@ "@modelcontextprotocol/ext-apps": "1.7.4", "@modelcontextprotocol/sdk": "1.29.0", "@noble/ed25519": "3.1.0", + "@openclaw/gateway-client": "workspace:*", "@openclaw/libterminal": "0.3.1", "@openclaw/media-core": "workspace:*", "@openclaw/normalization-core": "workspace:*", diff --git a/ui/src/api/gateway.node.test.ts b/ui/src/api/gateway.node.test.ts index a8bf971357ad..99393fa13066 100644 --- a/ui/src/api/gateway.node.test.ts +++ b/ui/src/api/gateway.node.test.ts @@ -1,11 +1,11 @@ -// @vitest-environment node -import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; -import { GATEWAY_CLIENT_CAPS } from "../../../packages/gateway-protocol/src/client-info.js"; -import { ConnectErrorDetailCodes } from "../../../packages/gateway-protocol/src/connect-error-details.js"; +/** @vitest-environment node */ import { + ConnectErrorDetailCodes, + GATEWAY_CLIENT_CAPS, MIN_CLIENT_PROTOCOL_VERSION, PROTOCOL_VERSION, -} from "../../../packages/gateway-protocol/src/version.js"; +} from "@openclaw/gateway-client/browser"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import type { DeviceIdentity } from "../lib/nodes/index.ts"; import { loadDeviceAuthToken as loadScopedDeviceAuthToken, @@ -646,6 +646,9 @@ describe("GatewayBrowserClient", () => { it("reports connect phase timing without credentials or nonce values", async () => { const onConnectTiming = vi.fn(); + vi.stubGlobal("performance", { + now: vi.fn().mockReturnValueOnce(10).mockReturnValueOnce(35).mockReturnValue(40), + }); const client = new GatewayBrowserClient({ url: "ws://127.0.0.1:18789", token: "shared-auth-token", @@ -661,6 +664,7 @@ describe("GatewayBrowserClient", () => { "connect-plan-ready", "request-sent", ]); + expect([sentPayloads[0]?.durationMs, sentPayloads[0]?.phaseDurationMs]).toEqual([25, 25]); for (const payload of sentPayloads) { expect(payload.generation).toBe(1); expect(payload.durationMs).toBeTypeOf("number"); @@ -1566,11 +1570,13 @@ describe("GatewayBrowserClient", () => { it("reports willRetry=false on credential rejections so the UI can fall back to the login gate", async () => { useNodeFakeTimers(); const onClose = vi.fn(); + const onConnectTiming = vi.fn(); const client = new GatewayBrowserClient({ url: "ws://127.0.0.1:18789", password: "wrong-password", onClose, + onConnectTiming, }); const { ws, connectFrame } = await startConnect(client); @@ -1589,6 +1595,12 @@ describe("GatewayBrowserClient", () => { const close = requireFirstMockArg(onClose, "close"); expect(close.willRetry).toBe(false); + expect(connectTimingPayloads(onConnectTiming).at(-1)).toMatchObject({ + phase: "failed", + errorCode: "INVALID_REQUEST", + hasDeviceIdentity: true, + hasPassword: true, + }); await vi.advanceTimersByTimeAsync(30_000); expect(wsInstances).toHaveLength(1); diff --git a/ui/src/api/gateway.ts b/ui/src/api/gateway.ts index 161bb8df40f4..b14f4f4350dc 100644 --- a/ui/src/api/gateway.ts +++ b/ui/src/api/gateway.ts @@ -1,27 +1,32 @@ // Control UI module implements gateway behavior. import { + buildDeviceAuthPayload, GATEWAY_CLIENT_CAPS, GATEWAY_CLIENT_MODES, GATEWAY_CLIENT_NAMES, - type GatewayClientMode, - type GatewayClientName, -} from "../../../packages/gateway-protocol/src/client-info.js"; -import { ConnectErrorDetailCodes, formatConnectErrorMessage, + GatewayProtocolClient, + GatewayProtocolRequestError, + type GatewayClientMode, + type GatewayClientName, + type GatewayProtocolCloseContext, + type GatewayProtocolRequestTiming, + type GatewayProtocolTiming, + type GatewayProtocolSocket, + type GatewayProtocolSocketHandlers, + type ConnectParams, + type ErrorShape, + type EventFrame, + type HelloOk, + shouldPauseGatewayReconnect, readConnectErrorRecoveryAdvice, readConnectErrorDetailCode, - readPairingConnectErrorDetails, -} from "../../../packages/gateway-protocol/src/connect-error-details.js"; -import { isRetryableGatewayStartupUnavailableError, resolveGatewayStartupRetryAfterMs, -} from "../../../packages/gateway-protocol/src/startup-unavailable.js"; -import { MIN_CLIENT_PROTOCOL_VERSION, PROTOCOL_VERSION, -} from "../../../packages/gateway-protocol/src/version.js"; -import { buildDeviceAuthPayload } from "../../../src/gateway/device-auth.js"; +} from "@openclaw/gateway-client/browser"; import { clearDeviceAuthToken, loadDeviceAuthToken, @@ -32,55 +37,24 @@ import { } from "../lib/nodes/index.ts"; import { generateUUID } from "../lib/uuid.ts"; -export type GatewayEventFrame = { - type: "event"; - event: string; - payload?: unknown; - seq?: number; - stateVersion?: { presence: number; health: number }; -}; +export type GatewayEventFrame = EventFrame; -type GatewayResponseFrame = { - type: "res"; - id: string; - ok: boolean; - payload?: unknown; - error?: { - code: string; - message: string; - details?: unknown; - retryable?: boolean; - retryAfterMs?: number; - }; -}; +type GatewayErrorInfo = ErrorShape; -type GatewayErrorInfo = { - code: string; - message: string; - details?: unknown; - retryable?: boolean; - retryAfterMs?: number; -}; - -export class GatewayRequestError extends Error { +export class GatewayRequestError extends GatewayProtocolRequestError { readonly gatewayCode: string; - readonly details?: unknown; - readonly retryable: boolean; - readonly retryAfterMs?: number; + override readonly retryable: boolean; constructor(error: GatewayErrorInfo) { const details = enrichProtocolMismatchDetails(error.message, error.details); - super( - formatConnectErrorMessage({ - message: error.message, - details, - }), - ); + super({ + ...error, + details, + message: formatConnectErrorMessage({ message: error.message, details }), + }); this.name = "GatewayRequestError"; - this.gatewayCode = error.code; - this.details = details; + this.gatewayCode = this.code; this.retryable = error.retryable === true; - this.retryAfterMs = error.retryAfterMs; } } @@ -105,14 +79,6 @@ export function resolveGatewayErrorDetailCode( return readConnectErrorDetailCode(error?.details); } -function shouldContinueReconnectForPairingRequired(details: unknown): boolean { - const pairingDetails = readPairingConnectErrorDetails(details); - return ( - pairingDetails?.pauseReconnect === false || - pairingDetails?.recommendedNextStep === "wait_then_retry" - ); -} - /** * Connect failures that cannot recover while client and server state stay unchanged. * AUTH_TOKEN_MISMATCH stays out: the close handler owns its bounded cached-token retry. @@ -121,40 +87,19 @@ export function isNonRecoverableConnectError(error: { details?: unknown } | unde if (!error) { return false; } - const code = resolveGatewayErrorDetailCode(error); - if ( - code === ConnectErrorDetailCodes.PAIRING_REQUIRED && - shouldContinueReconnectForPairingRequired(error.details) - ) { - return false; - } - return ( - code === ConnectErrorDetailCodes.AUTH_TOKEN_MISSING || - code === ConnectErrorDetailCodes.AUTH_BOOTSTRAP_TOKEN_INVALID || - code === ConnectErrorDetailCodes.AUTH_PASSWORD_MISSING || - code === ConnectErrorDetailCodes.AUTH_PASSWORD_MISMATCH || - code === ConnectErrorDetailCodes.AUTH_RATE_LIMITED || - code === ConnectErrorDetailCodes.AUTH_DEVICE_TOKEN_MISMATCH || - code === ConnectErrorDetailCodes.AUTH_SCOPE_MISMATCH || - code === ConnectErrorDetailCodes.PROTOCOL_MISMATCH || - code === ConnectErrorDetailCodes.PAIRING_REQUIRED || - code === ConnectErrorDetailCodes.CONTROL_UI_DEVICE_IDENTITY_REQUIRED || - code === ConnectErrorDetailCodes.DEVICE_IDENTITY_REQUIRED - ); + return shouldPauseGatewayReconnect({ + details: error.details, + protocolMismatchIsTerminal: true, + }); } function isLoopbackIPv4Host(host: string): boolean { const octets = host.split("."); - if (octets.length !== 4 || octets[0] !== "127") { - return false; - } - return octets.every((octet) => { - if (!/^\d+$/.test(octet)) { - return false; - } - const value = Number(octet); - return value >= 0 && value <= 255; - }); + return ( + octets.length === 4 && + octets[0] === "127" && + octets.every((octet) => /^\d+$/.test(octet) && Number(octet) <= 255) + ); } function isTrustedRetryEndpoint(url: string): boolean { @@ -173,42 +118,12 @@ function isTrustedRetryEndpoint(url: string): boolean { } } -export type GatewayControlUiPluginTab = { - pluginId: string; - id: string; - label: string; - description?: string; - icon?: string; - path?: string; - group?: "control" | "agent"; - order?: number; -}; - -export type GatewayHelloOk = { - type: "hello-ok"; - protocol: number; - server?: { - version?: string; - connId?: string; - }; - features?: { methods?: string[]; events?: string[] }; +export type GatewayControlUiPluginTab = NonNullable[number]; +export type GatewayHelloOk = Omit & { + server?: Partial; + features?: Partial; snapshot?: unknown; - auth: { - deviceToken?: string; - role: string; - scopes: string[]; - issuedAtMs?: number; - }; - controlUiTabs?: GatewayControlUiPluginTab[]; - pluginSurfaceUrls?: Record; - policy?: { tickIntervalMs?: number }; -}; - -type Pending = { - resolve: (value: unknown) => void; - reject: (err: unknown) => void; - method: string; - startedAtMs: number; + policy?: Partial; }; type SelectedConnectAuth = { @@ -219,7 +134,6 @@ type SelectedConnectAuth = { resolvedDeviceToken?: string; storedToken?: string; storedScopes?: string[]; - canFallbackToShared: boolean; }; const CONTROL_UI_OPERATOR_ROLE = "operator"; @@ -239,41 +153,9 @@ export const CONTROL_UI_BOOTSTRAP_OPERATOR_SCOPES = [ "operator.write", ] as const; -type GatewayConnectAuth = { - token?: string; - bootstrapToken?: string; - deviceToken?: string; - password?: string; -}; - -type GatewayConnectDevice = { - id: string; - publicKey: string; - signature: string; - signedAt: number; - nonce: string; -}; - -type GatewayConnectClientInfo = { - id: GatewayClientName; - version: string; - platform: string; - mode: GatewayClientMode; - instanceId?: string; -}; - -type GatewayConnectParams = { - minProtocol: typeof MIN_CLIENT_PROTOCOL_VERSION; - maxProtocol: typeof PROTOCOL_VERSION; - client: GatewayConnectClientInfo; - role: string; - scopes: string[]; - device?: GatewayConnectDevice; - caps: string[]; - auth?: GatewayConnectAuth; - userAgent: string; - locale: string; -}; +type GatewayConnectAuth = NonNullable; +type GatewayConnectDevice = NonNullable; +type GatewayConnectClientInfo = ConnectParams["client"]; type ConnectPlan = { role: string; @@ -315,39 +197,13 @@ export type GatewayBrowserClientOptions = { willRetry: boolean; }) => void; onGap?: (info: { expected: number; received: number }) => void; - onRequestTiming?: (timing: GatewayRequestTiming) => void; + onRequestTiming?: (timing: GatewayProtocolRequestTiming) => void; onConnectTiming?: (timing: GatewayConnectTiming) => void; }; export type GatewayEventListener = (evt: GatewayEventFrame) => void; -type GatewayRequestTiming = { - id: string; - method: string; - ok: boolean; - durationMs: number; - startedAtMs: number; - endedAtMs: number; - errorCode?: string; -}; - -type GatewayConnectTimingPhase = - | "socket-open" - | "challenge" - | "fallback" - | "device-identity-ready" - | "connect-plan-ready" - | "request-sent" - | "hello" - | "failed"; - -type GatewayConnectTiming = { - generation: number; - phase: GatewayConnectTimingPhase; - durationMs: number; - phaseDurationMs: number; - hasChallenge: boolean; - usedFallback: boolean; +type GatewayConnectTiming = Omit, "plan" | "detail"> & { secureContext?: boolean; hasDeviceIdentity?: boolean; hasDevice?: boolean; @@ -358,13 +214,6 @@ type GatewayConnectTiming = { errorCode?: string; }; -type ConnectTimingState = { - startedAtMs: number; - lastAtMs: number; - hasChallenge: boolean; - usedFallback: boolean; -}; - // 4008 = application-defined code (browser rejects 1008 "Policy Violation") const CONNECT_FAILED_CLOSE_CODE = 4008; const STARTUP_RETRY_CLOSE_CODE = 4013; @@ -392,15 +241,15 @@ function getErrorMessage(err: unknown): string { return err instanceof Error && err.message ? err.message : String(err); } +function toGatewayErrorInfo(error: GatewayRequestError): GatewayErrorInfo { + const { gatewayCode: code, message, details, retryable, retryAfterMs } = error; + return { code, message, details, retryable, retryAfterMs }; +} + function getErrorName(err: unknown): string | undefined { - if (err instanceof Error && err.name) { - return err.name; - } - if (err && typeof err === "object" && "name" in err) { - const name = (err as { name?: unknown }).name; - return typeof name === "string" && name.trim() ? name : undefined; - } - return undefined; + const name = + err && typeof err === "object" && "name" in err ? (err as { name?: unknown }).name : undefined; + return typeof name === "string" && name.trim() ? name : undefined; } function isBrowserWebSocketSecurityError(err: unknown): boolean { @@ -418,6 +267,13 @@ function formatBrowserWebSocketConstructorError(err: unknown, url: string): Gate const securityError = isBrowserWebSocketSecurityError(err); const browserMessage = getErrorMessage(err); const isPlaintextWs = url.trim().toLowerCase().startsWith("ws://"); + const details = { + code: securityError + ? BROWSER_WEBSOCKET_SECURITY_ERROR_CODE + : BROWSER_WEBSOCKET_CONSTRUCTOR_ERROR_CODE, + browserErrorName: getErrorName(err), + browserMessage, + }; if (securityError) { return { code: BROWSER_WEBSOCKET_SECURITY_ERROR_CODE, @@ -426,21 +282,13 @@ function formatBrowserWebSocketConstructorError(err: unknown, url: string): Gate (isPlaintextWs ? " Use wss:// when the Control UI is served over HTTPS/Tailscale Serve, or open the loopback dashboard at http://127.0.0.1:18789." : " Check the Gateway WebSocket URL and browser security policy."), - details: { - code: BROWSER_WEBSOCKET_SECURITY_ERROR_CODE, - browserErrorName: getErrorName(err), - browserMessage, - }, + details, }; } return { code: BROWSER_WEBSOCKET_CONSTRUCTOR_ERROR_CODE, message: `Could not create the Gateway WebSocket: ${browserMessage}`, - details: { - code: BROWSER_WEBSOCKET_CONSTRUCTOR_ERROR_CODE, - browserErrorName: getErrorName(err), - browserMessage, - }, + details, }; } @@ -555,218 +403,93 @@ export function shouldRetryWithDeviceToken(params: DeviceTokenRetryDecision): bo } export class GatewayBrowserClient { - private ws: WebSocket | null = null; - private pending = new Map(); - private closed = false; - private lastSeq: number | null = null; - private connectNonce: string | null = null; - private connectSent = false; - private connectTimer: number | null = null; - private connectGeneration = 0; - private backoffMs = 800; - private pendingConnectError: GatewayErrorInfo | undefined; + private readonly client: GatewayProtocolClient; private pendingDeviceTokenRetry = false; private deviceTokenRetryBudgetUsed = false; - private pendingStartupReconnectDelayMs: number | null = null; - private eventListeners = new Set(); - private connectTiming = new Map(); - constructor(private opts: GatewayBrowserClientOptions) {} + constructor(private opts: GatewayBrowserClientOptions) { + this.client = new GatewayProtocolClient({ + createSocket: (handlers) => this.createSocket(handlers), + createRequestId: generateUUID, + createRequestError: (error) => + new GatewayRequestError({ + code: error.code ?? "UNAVAILABLE", + message: error.message ?? "request failed", + details: error.details, + retryable: error.retryable, + retryAfterMs: error.retryAfterMs, + }), + buildConnectPlan: ({ nonce, generation }) => this.buildConnectPlan(nonce, generation), + buildConnectParams: (plan) => this.buildConnectParams(plan), + onConnectHello: (hello, context) => this.handleConnectHello(hello, context.plan), + onHello: (hello) => this.opts.onHello?.(hello), + onConnectFailure: (error, context) => { + this.client.recordTiming("failed", context.generation, context.plan, { + errorCode: error.code, + }); + return this.handleConnectFailure(error, context.plan); + }, + resolveClose: (context) => this.resolveClose(context), + onClose: (context, decision) => { + const error = context.connectFailure?.error; + this.client.recordTiming("failed", context.generation, undefined, { + errorCode: error instanceof GatewayRequestError ? error.code : "SOCKET_CLOSED", + }); + if (decision.notify) { + this.opts.onClose?.({ + code: context.code, + reason: context.reason, + error: error instanceof GatewayRequestError ? toGatewayErrorInfo(error) : undefined, + willRetry: decision.retry, + }); + } + }, + onSocketFactoryError: (error) => this.handleSocketFactoryError(error), + onEvent: (event) => this.opts.onEvent?.(event), + onGap: (info) => this.opts.onGap?.(info), + onTiming: ({ plan, detail, ...timing }) => { + this.opts.onConnectTiming?.({ + ...timing, + ...(plan ? this.connectPlanTimingPayload(plan) : {}), + ...(detail && typeof detail === "object" ? detail : {}), + }); + }, + onRequestTiming: (timing) => this.opts.onRequestTiming?.(timing), + onCallbackError: (label, error) => console.error(`[gateway] ${label} handler error:`, error), + handshake: { mode: "fallback", timeoutMs: 750 }, + reconnect: { initialMs: 800, multiplier: 1.7, maxMs: 15_000 }, + nowMs: () => + typeof performance !== "undefined" && typeof performance.now === "function" + ? performance.now() + : Date.now(), + }); + } start() { - this.closed = false; - this.connect(); + this.client.start(); } stop() { - this.closed = true; - this.clearConnectTimer(); - this.ws?.close(); - this.ws = null; - this.pendingConnectError = undefined; + this.client.stop(); this.pendingDeviceTokenRetry = false; this.deviceTokenRetryBudgetUsed = false; - this.pendingStartupReconnectDelayMs = null; - this.connectTiming.clear(); - this.flushPending(new Error("gateway client stopped")); } get connected() { - return this.ws?.readyState === WebSocket.OPEN; + return this.client.connected; } - private connect() { - if (this.closed) { - return; - } - let ws: WebSocket; - try { - ws = new WebSocket(this.opts.url); - } catch (err) { - const error = formatBrowserWebSocketConstructorError(err, this.opts.url); - this.ws = null; - this.pendingConnectError = undefined; - this.pendingDeviceTokenRetry = false; - this.pendingStartupReconnectDelayMs = null; - this.flushPending(new Error(error.message)); - this.notifyClose({ - code: BROWSER_WEBSOCKET_CLOSE_CODE, - reason: - error.code === BROWSER_WEBSOCKET_SECURITY_ERROR_CODE - ? "security error" - : "websocket error", - error, - // Constructor failures (bad URL, mixed content) never resolve on - // their own; no reconnect is scheduled for them. - willRetry: false, - }); - return; - } - const generation = ++this.connectGeneration; - this.ws = ws; - this.startConnectTiming(generation); - ws.addEventListener("open", () => this.queueConnect(ws, generation)); - ws.addEventListener("message", (ev) => { - if (!this.isActiveSocket(ws, generation)) { - return; - } - this.handleMessage(ws, generation, String(ev.data ?? "")); - }); - ws.addEventListener("close", (ev) => { - if (this.ws !== ws) { - return; - } - const reason = ev.reason ?? ""; - const connectError = this.pendingConnectError; - this.pendingConnectError = undefined; - this.emitConnectTiming(generation, "failed", { - errorCode: connectError?.code ?? "SOCKET_CLOSED", - }); - this.ws = null; - const closeError = connectError - ? new GatewayRequestError(connectError) - : new Error(`gateway closed (${ev.code}): ${reason}`); - if (this.pendingStartupReconnectDelayMs !== null) { - this.flushPending(closeError); - this.scheduleReconnect(); - return; - } - this.flushPending(closeError); - const connectErrorCode = resolveGatewayErrorDetailCode(connectError); - // willRetry drives both the reconnect scheduling below and the app - // layer's "still reconnecting vs gave up" rendering; keep them in sync. - const willRetry = - !this.closed && - (connectErrorCode === ConnectErrorDetailCodes.AUTH_TOKEN_MISMATCH - ? this.pendingDeviceTokenRetry - : !isNonRecoverableConnectError(connectError)); - this.notifyClose({ code: ev.code, reason, error: connectError, willRetry }); - if (willRetry) { - this.scheduleReconnect(); - } - }); - ws.addEventListener("error", () => { - // ignored; close handler will fire - }); - } - - private scheduleReconnect() { - if (this.closed) { - return; - } - const startupDelay = this.pendingStartupReconnectDelayMs; - this.pendingStartupReconnectDelayMs = null; - const delay = startupDelay ?? this.backoffMs; - if (startupDelay === null) { - this.backoffMs = Math.min(this.backoffMs * 1.7, 15_000); - } - this.clearConnectTimer(); - this.connectTimer = window.setTimeout(() => { - this.connectTimer = null; - this.connect(); - }, delay); - } - - private flushPending(err: Error) { - for (const [id, p] of this.pending) { - this.emitRequestTiming(id, p, false, "CLIENT_CLOSED"); - p.reject(err); - } - this.pending.clear(); - } - - private nowMs(): number { - return typeof performance !== "undefined" && typeof performance.now === "function" - ? performance.now() - : Date.now(); - } - - private startConnectTiming(generation: number): void { - const now = this.nowMs(); - this.connectTiming.set(generation, { - startedAtMs: now, - lastAtMs: now, - hasChallenge: false, - usedFallback: false, - }); - } - - private updateConnectTimingState( - generation: number, - updates: Partial>, - ): void { - const state = this.connectTiming.get(generation); - if (!state) { - return; - } - Object.assign(state, updates); - } - - private emitConnectTiming( - generation: number, - phase: GatewayConnectTimingPhase, - payload: Partial = {}, - ): void { - const state = this.connectTiming.get(generation); - if (!state) { - return; - } - const endedAtMs = this.nowMs(); - try { - this.opts.onConnectTiming?.({ - generation, - phase, - durationMs: Math.max(0, endedAtMs - state.startedAtMs), - phaseDurationMs: Math.max(0, endedAtMs - state.lastAtMs), - hasChallenge: state.hasChallenge, - usedFallback: state.usedFallback, - ...payload, - }); - } catch (err) { - console.error("[gateway] connect timing handler error:", err); - } finally { - state.lastAtMs = endedAtMs; - if (phase === "hello" || phase === "failed") { - this.connectTiming.delete(generation); - } - } - } - - private emitRequestTiming(id: string, pending: Pending, ok: boolean, errorCode?: string): void { - const endedAtMs = this.nowMs(); - try { - this.opts.onRequestTiming?.({ - id, - method: pending.method, - ok, - durationMs: Math.max(0, endedAtMs - pending.startedAtMs), - startedAtMs: pending.startedAtMs, - endedAtMs, - errorCode, - }); - } catch (err) { - console.error("[gateway] request timing handler error:", err); - } + private createSocket(handlers: GatewayProtocolSocketHandlers): GatewayProtocolSocket { + const socket = new WebSocket(this.opts.url); + socket.addEventListener("open", handlers.open); + socket.addEventListener("message", (event) => handlers.message(String(event.data ?? ""))); + socket.addEventListener("close", (event) => handlers.close(event.code, event.reason ?? "")); + socket.addEventListener("error", () => handlers.error(new Error("websocket error"))); + return { + isOpen: () => socket.readyState === WebSocket.OPEN, + send: (data) => socket.send(data), + close: (code, reason) => socket.close(code, reason), + }; } private connectPlanTimingPayload(plan: ConnectPlan): Partial { @@ -783,17 +506,7 @@ export class GatewayBrowserClient { }; } - private buildConnectClient(): GatewayConnectClientInfo { - return { - id: this.opts.clientName ?? GATEWAY_CLIENT_NAMES.CONTROL_UI, - version: this.opts.clientVersion ?? "control-ui", - platform: this.opts.platform ?? navigator.platform ?? "web", - mode: this.opts.mode ?? GATEWAY_CLIENT_MODES.WEBCHAT, - instanceId: this.opts.instanceId, - }; - } - - private buildConnectParams(plan: ConnectPlan): GatewayConnectParams { + private buildConnectParams(plan: ConnectPlan): ConnectParams { return { minProtocol: MIN_CLIENT_PROTOCOL_VERSION, maxProtocol: PROTOCOL_VERSION, @@ -817,7 +530,13 @@ export class GatewayBrowserClient { generation: number, ): Promise { const role = CONTROL_UI_OPERATOR_ROLE; - const client = this.buildConnectClient(); + const client: GatewayConnectClientInfo = { + id: this.opts.clientName ?? GATEWAY_CLIENT_NAMES.CONTROL_UI, + version: this.opts.clientVersion ?? "control-ui", + platform: this.opts.platform ?? navigator.platform ?? "web", + mode: this.opts.mode ?? GATEWAY_CLIENT_MODES.WEBCHAT, + instanceId: this.opts.instanceId, + }; const explicitGatewayToken = this.opts.token?.trim() || undefined; const explicitPassword = this.opts.password?.trim() || undefined; @@ -829,12 +548,11 @@ export class GatewayBrowserClient { let selectedAuth: SelectedConnectAuth = { authToken: explicitGatewayToken, authPassword: explicitPassword, - canFallbackToShared: false, }; if (isSecureContext) { deviceIdentity = await loadOrCreateDeviceIdentity(); - this.emitConnectTiming(generation, "device-identity-ready", { + this.client.recordTiming("device-identity-ready", generation, undefined, { secureContext: true, hasDeviceIdentity: true, }); @@ -852,17 +570,7 @@ export class GatewayBrowserClient { authToken: selectedAuth.authBootstrapToken ?? selectedAuth.authToken, connectNonce, }); - this.emitConnectTiming(generation, "connect-plan-ready", { - secureContext: isSecureContext, - hasDeviceIdentity: Boolean(deviceIdentity), - hasDevice: Boolean(device), - hasAuthToken: Boolean(selectedAuth.authToken), - hasBootstrapToken: Boolean(selectedAuth.authBootstrapToken), - hasDeviceToken: Boolean(selectedAuth.authDeviceToken ?? selectedAuth.resolvedDeviceToken), - hasPassword: Boolean(selectedAuth.authPassword), - }); - - return { + const plan: ConnectPlan = { role, scopes, client, @@ -872,38 +580,28 @@ export class GatewayBrowserClient { deviceIdentity, device, }; + if (this.pendingDeviceTokenRetry && plan.selectedAuth.authDeviceToken) { + this.pendingDeviceTokenRetry = false; + } + return plan; } - private handleConnectHello( - hello: GatewayHelloOk, - plan: ConnectPlan, - ws: WebSocket, - generation: number, - ) { - if (!this.isActiveSocket(ws, generation)) { - return; - } + private handleConnectHello(hello: GatewayHelloOk, plan: ConnectPlan) { this.pendingDeviceTokenRetry = false; this.deviceTokenRetryBudgetUsed = false; - this.pendingStartupReconnectDelayMs = null; this.opts.bootstrapToken = undefined; if (hello?.auth?.deviceToken && plan.deviceIdentity) { - this.storeDeviceAuthToken({ + storeDeviceAuthToken({ deviceId: plan.deviceIdentity.deviceId, + gatewayUrl: this.opts.url, role: hello.auth.role ?? plan.role, token: hello.auth.deviceToken, scopes: hello.auth.scopes ?? [], }); } - this.backoffMs = 800; - this.emitConnectTiming(generation, "hello", this.connectPlanTimingPayload(plan)); - this.notifyHello(hello); } - private handleConnectFailure(err: unknown, plan: ConnectPlan, ws: WebSocket, generation: number) { - if (!this.isActiveSocket(ws, generation)) { - return; - } + private handleConnectFailure(err: GatewayProtocolRequestError, plan: ConnectPlan) { const connectErrorCode = err instanceof GatewayRequestError ? resolveGatewayErrorDetailCode(err) : null; const recoveryAdvice = @@ -929,21 +627,6 @@ export class GatewayBrowserClient { this.pendingDeviceTokenRetry = true; this.deviceTokenRetryBudgetUsed = true; } - if (err instanceof GatewayRequestError) { - this.pendingConnectError = { - code: err.gatewayCode, - message: err.message, - details: err.details, - retryable: err.retryable, - retryAfterMs: err.retryAfterMs, - }; - } else { - this.pendingConnectError = undefined; - } - this.emitConnectTiming(generation, "failed", { - ...this.connectPlanTimingPayload(plan), - errorCode: err instanceof GatewayRequestError ? err.gatewayCode : "CLIENT_CONNECT_ERROR", - }); const usedStoredDeviceToken = Boolean(plan.selectedAuth.storedToken) && (plan.selectedAuth.resolvedDeviceToken === plan.selectedAuth.storedToken || @@ -960,155 +643,14 @@ export class GatewayBrowserClient { }); } const startupRetryAfterMs = resolveGatewayStartupRetryAfterMs(err); - if (startupRetryAfterMs !== null) { - this.pendingStartupReconnectDelayMs = startupRetryAfterMs; - } if (isRetryableGatewayStartupUnavailableError(err)) { - ws.close(STARTUP_RETRY_CLOSE_CODE, "gateway starting"); - return; - } - ws.close(CONNECT_FAILED_CLOSE_CODE, "connect failed"); - } - - private isActiveSocket(ws: WebSocket, generation: number): boolean { - return !this.closed && this.ws === ws && this.connectGeneration === generation; - } - - private storeDeviceAuthToken(params: { - deviceId: string; - role: string; - token: string; - scopes?: string[]; - }): void { - storeDeviceAuthToken({ - ...params, - gatewayUrl: this.opts.url, - }); - } - - private async sendConnect(ws: WebSocket, generation: number) { - if (!this.isActiveSocket(ws, generation) || ws.readyState !== WebSocket.OPEN) { - return; - } - if (this.connectSent) { - return; - } - this.connectSent = true; - this.clearConnectTimer(); - - const plan = await this.buildConnectPlan(this.connectNonce, generation); - if (!this.isActiveSocket(ws, generation) || ws.readyState !== WebSocket.OPEN) { - return; - } - if (this.pendingDeviceTokenRetry && plan.selectedAuth.authDeviceToken) { - this.pendingDeviceTokenRetry = false; - } - this.emitConnectTiming(generation, "request-sent", this.connectPlanTimingPayload(plan)); - void this.requestOnSocket(ws, "connect", this.buildConnectParams(plan)) - .then((hello) => this.handleConnectHello(hello, plan, ws, generation)) - .catch((err: unknown) => this.handleConnectFailure(err, plan, ws, generation)); - } - - private handleMessage(ws: WebSocket, generation: number, raw: string) { - let parsed: unknown; - try { - parsed = JSON.parse(raw); - } catch { - return; - } - - const frame = parsed as { type?: unknown }; - if (frame.type === "event") { - const evt = parsed as GatewayEventFrame; - if (evt.event === "connect.challenge") { - const payload = evt.payload as { nonce?: unknown } | undefined; - const nonce = payload && typeof payload.nonce === "string" ? payload.nonce : null; - if (nonce) { - this.connectNonce = nonce; - this.updateConnectTimingState(generation, { hasChallenge: true }); - this.emitConnectTiming(generation, "challenge"); - void this.sendConnect(ws, generation); - } - return; - } - const seq = typeof evt.seq === "number" ? evt.seq : null; - if (seq !== null) { - if (this.lastSeq !== null && seq > this.lastSeq + 1) { - this.notifyGap({ expected: this.lastSeq + 1, received: seq }); - } - this.lastSeq = seq; - } - this.notifyEvent(evt); - for (const listener of this.eventListeners) { - try { - listener(evt); - } catch (err) { - console.error("[gateway] event listener error:", err); - } - } - return; - } - - if (frame.type === "res") { - const res = parsed as GatewayResponseFrame; - const pending = this.pending.get(res.id); - if (!pending) { - return; - } - this.pending.delete(res.id); - if (res.ok) { - this.emitRequestTiming(res.id, pending, true); - pending.resolve(res.payload); - } else { - this.emitRequestTiming(res.id, pending, false, res.error?.code); - pending.reject( - new GatewayRequestError({ - code: res.error?.code ?? "UNAVAILABLE", - message: res.error?.message ?? "request failed", - details: res.error?.details, - retryable: res.error?.retryable, - retryAfterMs: res.error?.retryAfterMs, - }), - ); - } - } - } - - private notifyHello(hello: GatewayHelloOk): void { - try { - this.opts.onHello?.(hello); - } catch (err) { - console.error("[gateway] hello handler error:", err); - } - } - - private notifyClose(info: { - code: number; - reason: string; - error?: GatewayErrorInfo; - willRetry: boolean; - }): void { - try { - this.opts.onClose?.(info); - } catch (err) { - console.error("[gateway] close handler error:", err); - } - } - - private notifyGap(info: { expected: number; received: number }): void { - try { - this.opts.onGap?.(info); - } catch (err) { - console.error("[gateway] gap handler error:", err); - } - } - - private notifyEvent(evt: GatewayEventFrame): void { - try { - this.opts.onEvent?.(evt); - } catch (err) { - console.error("[gateway] event handler error:", err); + return { + closeCode: STARTUP_RETRY_CLOSE_CODE, + closeReason: "gateway starting", + reconnectDelayMs: startupRetryAfterMs ?? undefined, + }; } + return { closeCode: CONNECT_FAILED_CLOSE_CODE, closeReason: "connect failed" }; } private selectConnectAuth(params: { role: string; deviceId: string }): SelectedConnectAuth { @@ -1139,7 +681,6 @@ export class GatewayBrowserClient { authPassword, storedToken: storedToken ?? undefined, storedScopes: storedEntry?.scopes ?? undefined, - canFallbackToShared: false, }; } const authToken = explicitGatewayToken ?? resolvedDeviceToken; @@ -1150,62 +691,49 @@ export class GatewayBrowserClient { resolvedDeviceToken, storedToken: storedToken ?? undefined, storedScopes: storedEntry?.scopes ?? undefined, - canFallbackToShared: Boolean(storedToken && explicitGatewayToken), }; } request(method: string, params?: unknown): Promise { - if (!this.ws || this.ws.readyState !== WebSocket.OPEN) { - return Promise.reject(new Error("gateway not connected")); - } - return this.requestOnSocket(this.ws, method, params); - } - - private requestOnSocket( - ws: WebSocket, - method: string, - params?: unknown, - ): Promise { - if (this.ws !== ws || ws.readyState !== WebSocket.OPEN) { - return Promise.reject(new Error("gateway not connected")); - } - const id = generateUUID(); - const frame = { type: "req", id, method, params }; - const startedAtMs = this.nowMs(); - const p = new Promise((resolve, reject) => { - this.pending.set(id, { resolve: (v) => resolve(v as T), reject, method, startedAtMs }); - }); - ws.send(JSON.stringify(frame)); - return p; + return this.client.request(method, params); } addEventListener(listener: GatewayEventListener): () => void { - this.eventListeners.add(listener); - return () => { - this.eventListeners.delete(listener); - }; + return this.client.addEventListener(listener); } - private queueConnect(ws: WebSocket, generation: number) { - if (!this.isActiveSocket(ws, generation)) { - return; + private resolveClose(context: GatewayProtocolCloseContext) { + const error = context.connectFailure?.error; + const startupDelay = context.connectFailure?.reconnectDelayMs; + if (startupDelay !== undefined) { + return { retry: true, notify: false, reconnectDelayMs: startupDelay, pendingError: error }; } - this.connectNonce = null; - this.connectSent = false; - this.clearConnectTimer(); - this.emitConnectTiming(generation, "socket-open"); - this.connectTimer = window.setTimeout(() => { - this.connectTimer = null; - this.updateConnectTimingState(generation, { usedFallback: true }); - this.emitConnectTiming(generation, "fallback"); - void this.sendConnect(ws, generation); - }, 750); + const connectError = + error instanceof GatewayRequestError ? toGatewayErrorInfo(error) : undefined; + const connectErrorCode = resolveGatewayErrorDetailCode(connectError); + // This decision drives both scheduling and the store's reconnect rendering. + const retry = + connectErrorCode === ConnectErrorDetailCodes.AUTH_TOKEN_MISMATCH + ? this.pendingDeviceTokenRetry + : !isNonRecoverableConnectError(connectError); + return { retry, notify: true, pendingError: error }; } - private clearConnectTimer() { - if (this.connectTimer !== null) { - window.clearTimeout(this.connectTimer); - this.connectTimer = null; + private handleSocketFactoryError(error: Error): void { + const formatted = formatBrowserWebSocketConstructorError(error, this.opts.url); + this.pendingDeviceTokenRetry = false; + try { + this.opts.onClose?.({ + code: BROWSER_WEBSOCKET_CLOSE_CODE, + reason: + formatted.code === BROWSER_WEBSOCKET_SECURITY_ERROR_CODE + ? "security error" + : "websocket error", + error: formatted, + willRetry: false, + }); + } catch (callbackError) { + console.error("[gateway] close handler error:", callbackError); } } } diff --git a/ui/vitest.config.ts b/ui/vitest.config.ts index 58580ae9ef78..575fa182a530 100644 --- a/ui/vitest.config.ts +++ b/ui/vitest.config.ts @@ -14,6 +14,18 @@ import { const here = path.dirname(fileURLToPath(import.meta.url)); const repoRoot = path.resolve(here, ".."); const workspaceSourceAliases = [ + { + find: "@openclaw/gateway-client/browser", + replacement: path.resolve(repoRoot, "packages/gateway-client/src/browser.ts"), + }, + { + find: /^@openclaw\/gateway-protocol\/(.+)$/u, + replacement: path.resolve(repoRoot, "packages/gateway-protocol/src/$1.ts"), + }, + { + find: /^@openclaw\/(gateway-protocol|retry)$/u, + replacement: path.resolve(repoRoot, "packages/$1/src/index.ts"), + }, { find: "../logging/redact.js", replacement: path.resolve(here, "src/lib/browser-redact.ts"),