From 3856b4fa1b353a369891f05891da9c0bca274c0b Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Sat, 1 Aug 2026 14:06:30 -0700 Subject: [PATCH] refactor(cli): unify gateway RPC transport (#117601) --- src/cli/devices-cli.runtime.ts | 37 ++---- src/cli/gateway-cli/call.ts | 52 -------- src/cli/gateway-cli/health-route.test.ts | 9 +- src/cli/gateway-cli/health-route.ts | 21 ++-- .../register.option-collisions.test.ts | 5 +- src/cli/gateway-cli/register.ts | 9 +- src/cli/gateway-rpc.runtime.test.ts | 36 ++++++ src/cli/gateway-rpc.runtime.ts | 37 +++++- src/cli/gateway-rpc.ts | 10 ++ src/cli/help-cold-imports.test.ts | 4 +- src/cli/nodes-cli/rpc.runtime.ts | 117 ------------------ src/cli/nodes-cli/rpc.ts | 75 ++++++++--- src/cli/program.nodes-basic.e2e.test.ts | 4 +- src/commands/sessions-compact.test.ts | 2 +- src/commands/sessions-compact.ts | 10 +- src/gateway/client-bootstrap.test.ts | 18 +-- src/gateway/client-bootstrap.ts | 4 +- src/gateway/connection-auth.ts | 35 ------ ...t.ts => credentials-secret-inputs.test.ts} | 22 ++-- src/gateway/operator-approvals-client.ts | 31 +---- src/infra/exec-approval-channel-runtime.ts | 31 ++--- src/node-host/runner.test.ts | 8 +- src/node-host/runner.ts | 4 +- 23 files changed, 229 insertions(+), 352 deletions(-) delete mode 100644 src/cli/gateway-cli/call.ts delete mode 100644 src/cli/nodes-cli/rpc.runtime.ts delete mode 100644 src/gateway/connection-auth.ts rename src/gateway/{connection-auth.test.ts => credentials-secret-inputs.test.ts} (92%) diff --git a/src/cli/devices-cli.runtime.ts b/src/cli/devices-cli.runtime.ts index 6b281711795a..1f49c944cd51 100644 --- a/src/cli/devices-cli.runtime.ts +++ b/src/cli/devices-cli.runtime.ts @@ -5,10 +5,6 @@ import { normalizeStringifiedOptionalString, } from "@openclaw/normalization-core/string-coerce"; import { uniqueStrings } from "@openclaw/normalization-core/string-normalization"; -import { - GATEWAY_CLIENT_MODES, - GATEWAY_CLIENT_NAMES, -} from "../../packages/gateway-protocol/src/client-info.js"; import { readConnectPairingRequiredMessage, type ConnectPairingRequiredDetails, @@ -16,11 +12,7 @@ import { import { sanitizeForLog } from "../../packages/terminal-core/src/ansi.js"; import { getTerminalTableWidth, renderTable } from "../../packages/terminal-core/src/table.js"; import { theme } from "../../packages/terminal-core/src/theme.js"; -import { - buildGatewayConnectionDetails, - callGateway, - formatGatewayTransportErrorJson, -} from "../gateway/call.js"; +import { buildGatewayConnectionDetails, formatGatewayTransportErrorJson } from "../gateway/call.js"; import { ADMIN_SCOPE, PAIRING_SCOPE, @@ -46,8 +38,7 @@ import { import { parseNodeList } from "../shared/node-list-parse.js"; import type { NodeListNode } from "../shared/node-list-types.js"; import { formatCliCommand } from "./command-format.js"; -import { parseTimeoutMsWithFallback } from "./parse-timeout.js"; -import { withProgress } from "./progress.js"; +import { callGatewayFromCliWithTransport } from "./gateway-rpc.js"; import { quoteCliArg } from "./quote-cli-arg.js"; type DevicesRpcOpts = { @@ -137,25 +128,11 @@ const callGatewayCli = async ( params?: unknown, callOpts?: { scopes?: OperatorScope[] }, ) => - withProgress( - { - label: `Devices ${method}`, - indeterminate: true, - enabled: opts.json !== true, - }, - async () => - await callGateway({ - url: opts.url, - token: opts.token, - password: opts.password, - method, - params, - timeoutMs: parseTimeoutMsWithFallback(opts.timeout, DEFAULT_DEVICES_TIMEOUT_MS), - clientName: GATEWAY_CLIENT_NAMES.CLI, - mode: GATEWAY_CLIENT_MODES.CLI, - scopes: callOpts?.scopes, - }), - ); + callGatewayFromCliWithTransport(method, opts, params, { + label: `Devices ${method}`, + defaultTimeoutMs: DEFAULT_DEVICES_TIMEOUT_MS, + scopes: callOpts?.scopes, + }); function isPendingNodeApprovalState( state: unknown, diff --git a/src/cli/gateway-cli/call.ts b/src/cli/gateway-cli/call.ts deleted file mode 100644 index 5e92fc6d9faa..000000000000 --- a/src/cli/gateway-cli/call.ts +++ /dev/null @@ -1,52 +0,0 @@ -// Progress-wrapped Gateway RPC helper shared by CLI command surfaces. -import { - GATEWAY_CLIENT_MODES, - GATEWAY_CLIENT_NAMES, -} from "../../../packages/gateway-protocol/src/client-info.js"; -import type { OpenClawConfig } from "../../config/types.openclaw.js"; -import { callGateway } from "../../gateway/call.js"; -import { parseTimeoutMsWithFallback } from "../parse-timeout.js"; -import { withProgress } from "../progress.js"; - -export type GatewayRpcOpts = { - config?: OpenClawConfig; - url?: string; - token?: string; - password?: string; - timeout?: string | null; - expectFinal?: boolean; - json?: boolean; - localPortOverride?: number; -}; - -const DEFAULT_GATEWAY_RPC_TIMEOUT_MS = 10_000; - -export const callGatewayCli = async (method: string, opts: GatewayRpcOpts, params?: unknown) => { - const timeoutMs = - opts.timeout === null - ? null - : parseTimeoutMsWithFallback(opts.timeout, DEFAULT_GATEWAY_RPC_TIMEOUT_MS, { - invalidType: "error", - }); - return await withProgress( - { - label: `Gateway ${method}`, - indeterminate: true, - enabled: opts.json !== true, - }, - async () => - await callGateway({ - config: opts.config, - url: opts.url, - token: opts.token, - password: opts.password, - method, - params, - expectFinal: Boolean(opts.expectFinal), - timeoutMs, - localPortOverride: opts.localPortOverride, - clientName: GATEWAY_CLIENT_NAMES.CLI, - mode: GATEWAY_CLIENT_MODES.CLI, - }), - ); -}; diff --git a/src/cli/gateway-cli/health-route.test.ts b/src/cli/gateway-cli/health-route.test.ts index 993bbc01ec04..5a346bf58830 100644 --- a/src/cli/gateway-cli/health-route.test.ts +++ b/src/cli/gateway-cli/health-route.test.ts @@ -37,7 +37,12 @@ describe("runGatewayHealthJsonRoute", () => { }, ); - expect(callGateway).toHaveBeenCalledWith("health", { json: true, timeout: "10000" }); + expect(callGateway).toHaveBeenCalledWith( + "health", + { json: true, timeout: "10000" }, + undefined, + { defaultTimeoutMs: 10_000 }, + ); expect(runtime.writeJson).toHaveBeenCalledWith({ ok: true, durationMs: 6 }, 2); expect(readBestEffortConfig).not.toHaveBeenCalled(); expect(emitReachableGatewayAuthDiagnostic).not.toHaveBeenCalled(); @@ -70,6 +75,8 @@ describe("runGatewayHealthJsonRoute", () => { gateway: { auth: { mode: "token" }, mode: "local", port: 19083 }, }, }), + undefined, + { defaultTimeoutMs: 10_000 }, ); }); diff --git a/src/cli/gateway-cli/health-route.ts b/src/cli/gateway-cli/health-route.ts index 0a463396091c..a94d75b8b121 100644 --- a/src/cli/gateway-cli/health-route.ts +++ b/src/cli/gateway-cli/health-route.ts @@ -1,15 +1,18 @@ // Route-first machine-readable Gateway health command. import type { OpenClawConfig } from "../../config/types.openclaw.js"; import { type RuntimeEnv, writeRuntimeJson } from "../../runtime.js"; -import type { GatewayRpcOpts } from "./call.js"; + +type GatewayHealthRpcOpts = Parameters< + typeof import("../gateway-rpc.js").callGatewayFromCliWithTransport +>[1]; type GatewayHealthJsonRouteArgs = { - rpc: GatewayRpcOpts; + rpc: GatewayHealthRpcOpts; localPortOverride?: number; }; type GatewayHealthRouteDependencies = { - callGateway?: typeof import("./call.js").callGatewayCli; + callGateway?: typeof import("../gateway-rpc.js").callGatewayFromCliWithTransport; readBestEffortConfig?: () => Promise; emitReachableGatewayAuthDiagnostic?: typeof import("../../commands/health.js").emitReachableGatewayAuthDiagnostic; formatGatewayAuthErrorJson?: typeof import("../../gateway/call.js").formatGatewayAuthErrorJson; @@ -20,7 +23,7 @@ type GatewayHealthRouteDependencies = { async function resolveRouteRpcOptions( args: GatewayHealthJsonRouteArgs, deps: GatewayHealthRouteDependencies, -): Promise { +): Promise { if (args.localPortOverride === undefined) { return args.rpc; } @@ -48,11 +51,15 @@ export async function runGatewayHealthJsonRoute( runtime: RuntimeEnv, deps: GatewayHealthRouteDependencies = {}, ): Promise { - let rpc: GatewayRpcOpts | undefined; + let rpc: GatewayHealthRpcOpts | undefined; try { rpc = await resolveRouteRpcOptions(args, deps); - const callGateway = deps.callGateway ?? (await import("./call.js")).callGatewayCli; - writeRuntimeJson(runtime, await callGateway("health", rpc)); + const callGateway = + deps.callGateway ?? (await import("../gateway-rpc.js")).callGatewayFromCliWithTransport; + writeRuntimeJson( + runtime, + await callGateway("health", rpc, undefined, { defaultTimeoutMs: 10_000 }), + ); } catch (error) { if (!rpc) { runtime.error(String(error)); diff --git a/src/cli/gateway-cli/register.option-collisions.test.ts b/src/cli/gateway-cli/register.option-collisions.test.ts index ea5f363162ce..e63e762354ba 100644 --- a/src/cli/gateway-cli/register.option-collisions.test.ts +++ b/src/cli/gateway-cli/register.option-collisions.test.ts @@ -46,8 +46,9 @@ vi.mock("../../commands/gateway-status.js", () => ({ mocks.gatewayStatusCommand(opts, runtime), })); -vi.mock("./call.js", () => ({ - callGatewayCli: (method: string, opts: unknown, params?: unknown) => +vi.mock("../gateway-rpc.js", async () => ({ + ...(await vi.importActual("../gateway-rpc.js")), + callGatewayFromCliWithTransport: (method: string, opts: unknown, params?: unknown) => mocks.callGatewayCli(method, opts, params), })); diff --git a/src/cli/gateway-cli/register.ts b/src/cli/gateway-cli/register.ts index bd28dce614db..ac1963fa942b 100644 --- a/src/cli/gateway-cli/register.ts +++ b/src/cli/gateway-cli/register.ts @@ -21,15 +21,17 @@ import { sleep } from "../../utils/sleep.js"; import { inheritOptionFromParent } from "../command-options.js"; import { addGatewayServiceCommands } from "../daemon-cli/register-service-commands.js"; import { parseGatewayPortOption } from "../gateway-port-option.js"; +import { callGatewayFromCliWithTransport } from "../gateway-rpc.js"; import { formatHelpExamples } from "../help-format.js"; import { parseTimeoutMsWithFallback } from "../parse-timeout.js"; import { setCommandJsonMode } from "../program/json-mode.js"; -import type { GatewayRpcOpts } from "./call.js"; import type { GatewayDiscoverOpts } from "./discover.js"; import { isGatewayMachineOutput } from "./output-mode.js"; import { addGatewayRestartHandoffCommands } from "./register-restart-handoff.js"; import { addGatewayRunCommand } from "./run-command.js"; +type GatewayRpcOpts = Parameters[1]; + const configModuleLoader = createLazyImportLoader( () => import("../../config/read-best-effort-config.runtime.js"), ); @@ -120,8 +122,9 @@ function gatewayCallOpts(cmd: Command, defaultTimeoutMs = DEFAULT_GATEWAY_RPC_TI } async function callGatewayCli(method: string, opts: GatewayRpcOpts, params?: unknown) { - const mod = await import("./call.js"); - return mod.callGatewayCli(method, opts, params); + return await callGatewayFromCliWithTransport(method, opts, params, { + defaultTimeoutMs: DEFAULT_GATEWAY_RPC_TIMEOUT_MS, + }); } function parseGatewayCallParams(value = "{}"): unknown { diff --git a/src/cli/gateway-rpc.runtime.test.ts b/src/cli/gateway-rpc.runtime.test.ts index 3295141fcea9..47f0ab601449 100644 --- a/src/cli/gateway-rpc.runtime.test.ts +++ b/src/cli/gateway-rpc.runtime.test.ts @@ -56,6 +56,42 @@ describe("callGatewayFromCliRuntime", () => { ); }); + it("accepts a caller-specific default timeout", async () => { + await callGatewayFromCliRuntime("health", {}, undefined, { defaultTimeoutMs: 10_000 }); + + expect(callGatewayMock).toHaveBeenCalledWith( + expect.objectContaining({ method: "health", timeoutMs: 10_000 }), + ); + }); + + it("forwards specialized connection and authorization context", async () => { + const config = { gateway: { mode: "local" as const } }; + await callGatewayFromCliRuntime( + "node.list", + { config, localPortOverride: 19_083 }, + {}, + { + timeoutMs: null, + scopes: ["operator.read", "operator.pairing"], + useStoredDeviceAuth: true, + requiredStoredDeviceAuthScopes: ["operator.read", "operator.pairing"], + requireLocalBackendSharedAuth: true, + }, + ); + + expect(callGatewayMock).toHaveBeenCalledWith( + expect.objectContaining({ + config, + localPortOverride: 19_083, + timeoutMs: null, + scopes: ["operator.read", "operator.pairing"], + useStoredDeviceAuth: true, + requiredStoredDeviceAuthScopes: ["operator.read", "operator.pairing"], + requireLocalBackendSharedAuth: true, + }), + ); + }); + it.each([ { name: "token", auth: { token: "test-gateway-token" } }, { name: "password", auth: { password: "test-gateway-password" } }, diff --git a/src/cli/gateway-rpc.runtime.ts b/src/cli/gateway-rpc.runtime.ts index 52d7473d7d31..e058b93bd1e6 100644 --- a/src/cli/gateway-rpc.runtime.ts +++ b/src/cli/gateway-rpc.runtime.ts @@ -3,6 +3,7 @@ import { GATEWAY_CLIENT_MODES, GATEWAY_CLIENT_NAMES, } from "../../packages/gateway-protocol/src/client-info.js"; +import type { OpenClawConfig } from "../config/types.openclaw.js"; import { callGateway } from "../gateway/call.js"; import type { GatewayRpcOpts } from "./gateway-rpc.types.js"; import { parseTimeoutMsWithFallback } from "./parse-timeout.js"; @@ -16,29 +17,51 @@ type CallGatewayFromCliRuntimeExtra = { expectFinal?: boolean; progress?: boolean; scopes?: Parameters[0]["scopes"]; + defaultTimeoutMs?: number; + timeoutMs?: number | null; + label?: string; + useStoredDeviceAuth?: boolean; + requiredStoredDeviceAuthScopes?: Parameters< + typeof callGateway + >[0]["requiredStoredDeviceAuthScopes"]; + requireLocalBackendSharedAuth?: boolean; +}; + +type GatewayCliTransportRpcOpts = Omit & { + config?: OpenClawConfig; + timeout?: string | null; + localPortOverride?: number; }; const DEFAULT_GATEWAY_RPC_TIMEOUT_MS = 30_000; export async function callGatewayFromCliRuntime( method: string, - opts: GatewayRpcOpts, + opts: GatewayCliTransportRpcOpts, params?: unknown, extra?: CallGatewayFromCliRuntimeExtra, ) { // Progress is disabled for JSON output so stdout stays parseable. const showProgress = extra?.progress ?? opts.json !== true; - const timeoutMs = parseTimeoutMsWithFallback(opts.timeout, DEFAULT_GATEWAY_RPC_TIMEOUT_MS, { - invalidType: "error", - }); + const timeoutMs = + extra?.timeoutMs !== undefined + ? extra.timeoutMs + : opts.timeout === null + ? null + : parseTimeoutMsWithFallback( + opts.timeout, + extra?.defaultTimeoutMs ?? DEFAULT_GATEWAY_RPC_TIMEOUT_MS, + { invalidType: "error" }, + ); return await withProgress( { - label: `Gateway ${method}`, + label: extra?.label ?? `Gateway ${method}`, indeterminate: true, enabled: showProgress, }, async () => await callGateway({ + config: opts.config, url: opts.url, token: opts.token, password: opts.password, @@ -47,8 +70,12 @@ export async function callGatewayFromCliRuntime( deviceIdentity: extra?.deviceIdentity, expectFinal: extra?.expectFinal ?? Boolean(opts.expectFinal), scopes: extra?.scopes, + useStoredDeviceAuth: extra?.useStoredDeviceAuth, + requiredStoredDeviceAuthScopes: extra?.requiredStoredDeviceAuthScopes, + requireLocalBackendSharedAuth: extra?.requireLocalBackendSharedAuth, signal: extra?.signal, timeoutMs, + localPortOverride: opts.localPortOverride, clientName: extra?.clientName ?? GATEWAY_CLIENT_NAMES.CLI, mode: extra?.mode ?? GATEWAY_CLIENT_MODES.CLI, }), diff --git a/src/cli/gateway-rpc.ts b/src/cli/gateway-rpc.ts index b86bd1573625..a5753da352ad 100644 --- a/src/cli/gateway-rpc.ts +++ b/src/cli/gateway-rpc.ts @@ -43,6 +43,16 @@ export async function callGatewayFromCli( progress?: boolean; scopes?: OperatorScope[]; }, +) { + return await callGatewayFromCliWithTransport(method, opts, params, extra); +} + +/** Internal CLI facade for callers that need transport or auth policy overrides. */ +export async function callGatewayFromCliWithTransport( + method: string, + opts: Parameters[1], + params?: unknown, + extra?: Parameters[3], ) { const runtime = await loadGatewayRpcRuntime(); return await runtime.callGatewayFromCliRuntime(method, opts, params, extra); diff --git a/src/cli/help-cold-imports.test.ts b/src/cli/help-cold-imports.test.ts index f2845b654884..2c4e391cb5c2 100644 --- a/src/cli/help-cold-imports.test.ts +++ b/src/cli/help-cold-imports.test.ts @@ -20,10 +20,10 @@ vi.mock("./gateway-cli/run.js", () => { }; }); -vi.mock("./gateway-cli/call.js", () => { +vi.mock("./gateway-rpc.runtime.js", () => { loaded.mark("gateway-call-runtime"); return { - callGatewayCli: vi.fn(async () => ({})), + callGatewayFromCliRuntime: vi.fn(async () => ({})), }; }); diff --git a/src/cli/nodes-cli/rpc.runtime.ts b/src/cli/nodes-cli/rpc.runtime.ts deleted file mode 100644 index 1e024549f62a..000000000000 --- a/src/cli/nodes-cli/rpc.runtime.ts +++ /dev/null @@ -1,117 +0,0 @@ -// Runtime gateway RPC helpers for node host and node pairing CLI commands. -import { - GATEWAY_CLIENT_MODES, - GATEWAY_CLIENT_NAMES, -} from "../../../packages/gateway-protocol/src/client-info.js"; -import { callGateway } from "../../gateway/call.js"; -import type { OperatorScope } from "../../gateway/method-scopes.js"; -import { parseTimeoutMsWithFallback } from "../parse-timeout.js"; -import { withProgress } from "../progress.js"; -import type { NodesRpcOpts } from "./types.js"; - -const NODE_PAIR_APPROVAL_GATEWAY_METHODS = new Set(["node.pair.list", "node.pair.approve"]); -const DEFAULT_NODES_RPC_TIMEOUT_MS = 10_000; - -function resolveNodesTransportTimeoutMs( - opts: NodesRpcOpts, - overrideMs?: number, - invokeTimeoutMs?: unknown, -): number | null { - const transportTimeoutMs = - overrideMs ?? parseTimeoutMsWithFallback(opts.timeout, DEFAULT_NODES_RPC_TIMEOUT_MS); - if (invokeTimeoutMs === 0) { - // Zero disables the node deadline; null keeps Gateway startup bounded but the request unbounded. - return null; - } - if ( - typeof invokeTimeoutMs !== "number" || - !Number.isSafeInteger(invokeTimeoutMs) || - invokeTimeoutMs <= 0 - ) { - return transportTimeoutMs; - } - // Gateway transport starts before the node timer; retain one normal RPC timeout for forwarding. - return Math.max(transportTimeoutMs, invokeTimeoutMs + DEFAULT_NODES_RPC_TIMEOUT_MS); -} - -export async function callGatewayCliRuntime( - method: string, - opts: NodesRpcOpts, - params?: unknown, - callOpts?: { - scopes?: OperatorScope[]; - transportTimeoutMs?: number; - useStoredDeviceAuth?: boolean; - requiredStoredDeviceAuthScopes?: OperatorScope[]; - useLocalBackendSharedAuth?: boolean; - }, -) { - const invokeTimeoutMs = - method === "node.invoke" && - params !== null && - typeof params === "object" && - !Array.isArray(params) - ? (params as { timeoutMs?: unknown }).timeoutMs - : undefined; - // Progress is suppressed for JSON callers so stdout remains structured. - return await withProgress( - { - label: `Nodes ${method}`, - indeterminate: true, - enabled: opts.json !== true, - }, - async () => - await callGateway({ - url: opts.url, - token: opts.token, - method, - params, - scopes: callOpts?.scopes, - useStoredDeviceAuth: callOpts?.useStoredDeviceAuth, - requiredStoredDeviceAuthScopes: callOpts?.requiredStoredDeviceAuthScopes, - requireLocalBackendSharedAuth: callOpts?.useLocalBackendSharedAuth, - timeoutMs: resolveNodesTransportTimeoutMs( - opts, - callOpts?.transportTimeoutMs, - invokeTimeoutMs, - ), - clientName: callOpts?.useLocalBackendSharedAuth - ? GATEWAY_CLIENT_NAMES.GATEWAY_CLIENT - : GATEWAY_CLIENT_NAMES.CLI, - mode: callOpts?.useLocalBackendSharedAuth - ? GATEWAY_CLIENT_MODES.BACKEND - : GATEWAY_CLIENT_MODES.CLI, - }), - ); -} - -export async function callNodePairApprovalGatewayCliRuntime( - method: "node.pair.list" | "node.pair.approve", - opts: NodesRpcOpts, - params: unknown, - callOpts: { scopes: OperatorScope[]; transportTimeoutMs?: number }, -) { - if (!NODE_PAIR_APPROVAL_GATEWAY_METHODS.has(method)) { - throw new Error(`unsupported node pair approval gateway method: ${method}`); - } - // Node approval may need the local gateway's backend shared-auth authority - // before the CLI device has been granted the node's required operator scopes. - return await withProgress( - { - label: `Nodes ${method}`, - indeterminate: true, - enabled: opts.json !== true, - }, - async () => - await callGateway({ - url: opts.url, - token: opts.token, - method, - params, - timeoutMs: resolveNodesTransportTimeoutMs(opts, callOpts.transportTimeoutMs), - clientName: GATEWAY_CLIENT_NAMES.GATEWAY_CLIENT, - mode: GATEWAY_CLIENT_MODES.BACKEND, - scopes: callOpts.scopes, - }), - ); -} diff --git a/src/cli/nodes-cli/rpc.ts b/src/cli/nodes-cli/rpc.ts index b8991762f83d..2b59942176cf 100644 --- a/src/cli/nodes-cli/rpc.ts +++ b/src/cli/nodes-cli/rpc.ts @@ -2,6 +2,10 @@ import { randomUUID } from "node:crypto"; import { normalizeLowercaseStringOrEmpty } from "@openclaw/normalization-core/string-coerce"; import type { Command } from "commander"; +import { + GATEWAY_CLIENT_MODES, + GATEWAY_CLIENT_NAMES, +} from "../../../packages/gateway-protocol/src/client-info.js"; import { readConnectErrorDetailCode } from "../../../packages/gateway-protocol/src/connect-error-details.js"; import { readMissingScopeError } from "../../../packages/gateway-protocol/src/gateway-error-details.js"; import type { OperatorScope } from "../../gateway/method-scopes.js"; @@ -10,21 +14,12 @@ import { parseStrictNonNegativeInteger, parseStrictPositiveInteger, } from "../../infra/parse-finite-number.js"; -import { createLazyImportLoader } from "../../shared/lazy-promise.js"; import { resolveNodeFromNodeList } from "../../shared/node-resolve.js"; +import { callGatewayFromCliWithTransport } from "../gateway-rpc.js"; +import { parseTimeoutMsWithFallback } from "../parse-timeout.js"; import { parseNodeList, parsePairingList } from "./format.js"; import type { NodeListNode, NodesRpcOpts } from "./types.js"; -type NodesCliRpcRuntimeModule = typeof import("./rpc.runtime.js"); - -const nodesCliRpcRuntimeLoader = createLazyImportLoader( - () => import("./rpc.runtime.js"), -); - -async function loadNodesCliRpcRuntime(): Promise { - return nodesCliRpcRuntimeLoader.load(); -} - const STORED_DEVICE_AUTH_FALLBACK_DETAIL_CODES = new Set([ "AUTH_REQUIRED", "AUTH_UNAUTHORIZED", @@ -33,6 +28,30 @@ const STORED_DEVICE_AUTH_FALLBACK_DETAIL_CODES = new Set([ "AUTH_SCOPE_MISMATCH", "PAIRING_REQUIRED", ]); +const NODE_PAIR_APPROVAL_GATEWAY_METHODS = new Set(["node.pair.list", "node.pair.approve"]); +const DEFAULT_NODES_RPC_TIMEOUT_MS = 10_000; + +function resolveNodesTransportTimeoutMs( + opts: NodesRpcOpts, + overrideMs?: number, + invokeTimeoutMs?: unknown, +): number | null { + const transportTimeoutMs = + overrideMs ?? parseTimeoutMsWithFallback(opts.timeout, DEFAULT_NODES_RPC_TIMEOUT_MS); + if (invokeTimeoutMs === 0) { + // Zero disables the node deadline; null keeps Gateway startup bounded but the request unbounded. + return null; + } + if ( + typeof invokeTimeoutMs !== "number" || + !Number.isSafeInteger(invokeTimeoutMs) || + invokeTimeoutMs <= 0 + ) { + return transportTimeoutMs; + } + // Gateway transport starts before the node timer; retain one normal RPC timeout for forwarding. + return Math.max(transportTimeoutMs, invokeTimeoutMs + DEFAULT_NODES_RPC_TIMEOUT_MS); +} function isDiagnosticsAuthFallbackError(value: unknown): value is Error { if ( @@ -84,8 +103,26 @@ export const callGatewayCli = async ( useLocalBackendSharedAuth?: boolean; }, ) => { - const runtime = await loadNodesCliRpcRuntime(); - return await runtime.callGatewayCliRuntime(method, opts, params, callOpts); + const invokeTimeoutMs = + method === "node.invoke" && + params !== null && + typeof params === "object" && + !Array.isArray(params) + ? (params as { timeoutMs?: unknown }).timeoutMs + : undefined; + const useLocalBackendSharedAuth = callOpts?.useLocalBackendSharedAuth === true; + return await callGatewayFromCliWithTransport(method, opts, params, { + label: `Nodes ${method}`, + timeoutMs: resolveNodesTransportTimeoutMs(opts, callOpts?.transportTimeoutMs, invokeTimeoutMs), + scopes: callOpts?.scopes, + useStoredDeviceAuth: callOpts?.useStoredDeviceAuth, + requiredStoredDeviceAuthScopes: callOpts?.requiredStoredDeviceAuthScopes, + requireLocalBackendSharedAuth: useLocalBackendSharedAuth, + clientName: useLocalBackendSharedAuth + ? GATEWAY_CLIENT_NAMES.GATEWAY_CLIENT + : GATEWAY_CLIENT_NAMES.CLI, + mode: useLocalBackendSharedAuth ? GATEWAY_CLIENT_MODES.BACKEND : GATEWAY_CLIENT_MODES.CLI, + }); }; /** Read node diagnostics with pairing details when authorized, otherwise keep read-only access. */ @@ -124,8 +161,16 @@ export const callNodePairApprovalGatewayCli = async ( params: unknown, callOpts: { scopes: OperatorScope[]; transportTimeoutMs?: number }, ) => { - const runtime = await loadNodesCliRpcRuntime(); - return await runtime.callNodePairApprovalGatewayCliRuntime(method, opts, params, callOpts); + if (!NODE_PAIR_APPROVAL_GATEWAY_METHODS.has(method)) { + throw new Error(`unsupported node pair approval gateway method: ${method}`); + } + return await callGatewayFromCliWithTransport(method, opts, params, { + label: `Nodes ${method}`, + timeoutMs: resolveNodesTransportTimeoutMs(opts, callOpts.transportTimeoutMs), + scopes: callOpts.scopes, + clientName: GATEWAY_CLIENT_NAMES.GATEWAY_CLIENT, + mode: GATEWAY_CLIENT_MODES.BACKEND, + }); }; /** Build a node.invoke payload with an idempotency key and optional timeout. */ diff --git a/src/cli/program.nodes-basic.e2e.test.ts b/src/cli/program.nodes-basic.e2e.test.ts index 516f7d8b0f82..1a1cb8a74dbe 100644 --- a/src/cli/program.nodes-basic.e2e.test.ts +++ b/src/cli/program.nodes-basic.e2e.test.ts @@ -978,10 +978,10 @@ describe("cli program (nodes basics)", () => { }); it("rejects unsupported node approval backend methods at runtime", async () => { - const { callNodePairApprovalGatewayCliRuntime } = await import("./nodes-cli/rpc.runtime.js"); + const { callNodePairApprovalGatewayCli } = await import("./nodes-cli/rpc.js"); await expect( - callNodePairApprovalGatewayCliRuntime( + callNodePairApprovalGatewayCli( "node.invoke" as never, { json: true }, {}, diff --git a/src/commands/sessions-compact.test.ts b/src/commands/sessions-compact.test.ts index 55a884561144..c4bd5653d0f2 100644 --- a/src/commands/sessions-compact.test.ts +++ b/src/commands/sessions-compact.test.ts @@ -6,7 +6,7 @@ import { sessionsCompactCommand } from "./sessions-compact.js"; const callGatewayCli = vi.hoisted(() => vi.fn()); -vi.mock("../cli/gateway-cli/call.js", () => ({ callGatewayCli })); +vi.mock("../cli/gateway-rpc.js", () => ({ callGatewayFromCliWithTransport: callGatewayCli })); function createRuntime() { return { diff --git a/src/commands/sessions-compact.ts b/src/commands/sessions-compact.ts index 168bd1421c42..6dcd74e6c698 100644 --- a/src/commands/sessions-compact.ts +++ b/src/commands/sessions-compact.ts @@ -7,7 +7,7 @@ * (transport error or an `ok:false` payload) so automation never mistakes a * silent no-op for success. */ -import { callGatewayCli, type GatewayRpcOpts } from "../cli/gateway-cli/call.js"; +import { callGatewayFromCliWithTransport } from "../cli/gateway-rpc.js"; import { formatErrorMessage } from "../infra/errors.js"; import { type RuntimeEnv, writeRuntimeJson } from "../runtime.js"; @@ -45,6 +45,8 @@ type SessionsCompactResult = { }; }; +type SessionsCompactRpcOpts = Parameters[1]; + function describeCompaction(result: SessionsCompactResult, fallbackKey: string): string { const sessionKey = result.key ?? fallbackKey; if (!result.compacted) { @@ -71,7 +73,7 @@ export async function sessionsCompactCommand( opts: SessionsCompactCliOptions, runtime: RuntimeEnv, ): Promise { - const rpcOpts: GatewayRpcOpts = { + const rpcOpts: SessionsCompactRpcOpts = { url: opts.url, token: opts.token, password: opts.password, @@ -88,7 +90,9 @@ export async function sessionsCompactCommand( let result: SessionsCompactResult; try { - result = (await callGatewayCli("sessions.compact", rpcOpts, params)) as SessionsCompactResult; + result = (await callGatewayFromCliWithTransport("sessions.compact", rpcOpts, params, { + defaultTimeoutMs: 10_000, + })) as SessionsCompactResult; } catch (err) { const message = formatErrorMessage(err); if (opts.json) { diff --git a/src/gateway/client-bootstrap.test.ts b/src/gateway/client-bootstrap.test.ts index 1eebef5d5e83..b265a39b08ac 100644 --- a/src/gateway/client-bootstrap.test.ts +++ b/src/gateway/client-bootstrap.test.ts @@ -1,13 +1,13 @@ // Gateway client bootstrap tests keep URL override provenance wired into shared // auth resolution so CLI and env callers authenticate against the intended target. import { beforeEach, describe, expect, it, vi } from "vitest"; -import type { resolveGatewayConnectionAuth } from "./connection-auth.js"; +import type { resolveGatewayCredentialsWithSecretInputs } from "./credentials-secret-inputs.js"; -type AuthResolutionParams = Parameters[0]; +type AuthResolutionParams = Parameters[0]; const mockState = vi.hoisted(() => ({ buildGatewayConnectionDetails: vi.fn(), - resolveGatewayConnectionAuth: vi.fn(), + resolveGatewayCredentialsWithSecretInputs: vi.fn(), })); vi.mock("./connection-details.js", () => ({ @@ -15,9 +15,9 @@ vi.mock("./connection-details.js", () => ({ mockState.buildGatewayConnectionDetails(...args), })); -vi.mock("./connection-auth.js", () => ({ - resolveGatewayConnectionAuth: (...args: unknown[]) => - mockState.resolveGatewayConnectionAuth(...args), +vi.mock("./credentials-secret-inputs.js", () => ({ + resolveGatewayCredentialsWithSecretInputs: (...args: unknown[]) => + mockState.resolveGatewayCredentialsWithSecretInputs(...args), })); const { resolveGatewayClientBootstrap } = await import("./client-bootstrap.js"); @@ -25,7 +25,7 @@ function expectLastAuthResolutionParams(expected: { urlOverride?: string; urlOverrideSource?: "cli" | "env"; }) { - const [params] = mockState.resolveGatewayConnectionAuth.mock.calls.at(-1) ?? []; + const [params] = mockState.resolveGatewayCredentialsWithSecretInputs.mock.calls.at(-1) ?? []; if (params === undefined) { throw new Error("Expected shared auth resolution to be called"); } @@ -38,8 +38,8 @@ function expectLastAuthResolutionParams(expected: { describe("resolveGatewayClientBootstrap", () => { beforeEach(() => { mockState.buildGatewayConnectionDetails.mockReset(); - mockState.resolveGatewayConnectionAuth.mockReset(); - mockState.resolveGatewayConnectionAuth.mockResolvedValue({ + mockState.resolveGatewayCredentialsWithSecretInputs.mockReset(); + mockState.resolveGatewayCredentialsWithSecretInputs.mockResolvedValue({ token: undefined, password: undefined, }); diff --git a/src/gateway/client-bootstrap.ts b/src/gateway/client-bootstrap.ts index 3daf40ea98d0..d27c331b3d84 100644 --- a/src/gateway/client-bootstrap.ts +++ b/src/gateway/client-bootstrap.ts @@ -1,8 +1,8 @@ // Gateway client bootstrap resolver. // Collects URL, auth, and handshake settings before constructing a GatewayClient. import type { OpenClawConfig } from "../config/types.openclaw.js"; -import { resolveGatewayConnectionAuth } from "./connection-auth.js"; import { buildGatewayConnectionDetailsWithResolvers } from "./connection-details.js"; +import { resolveGatewayCredentialsWithSecretInputs } from "./credentials-secret-inputs.js"; import type { ExplicitGatewayAuth } from "./credentials.js"; /** @@ -42,7 +42,7 @@ export async function resolveGatewayClientBootstrap(params: { const urlOverrideSource = resolveGatewayUrlOverrideSource(connection.urlSource); // Only direct CLI/env URL overrides should constrain token/password fallback. Config-derived // remote URLs are canonical config, not a caller override. - const auth = await resolveGatewayConnectionAuth({ + const auth = await resolveGatewayCredentialsWithSecretInputs({ config: params.config, explicitAuth: params.explicitAuth, env: params.env ?? process.env, diff --git a/src/gateway/connection-auth.ts b/src/gateway/connection-auth.ts deleted file mode 100644 index 4ddd12b8800d..000000000000 --- a/src/gateway/connection-auth.ts +++ /dev/null @@ -1,35 +0,0 @@ -// Gateway connection auth facade. -// Resolves config-backed client credentials with or without async SecretRefs. -import type { OpenClawConfig } from "../config/types.openclaw.js"; -import { resolveGatewayCredentialsWithSecretInputs } from "./credentials-secret-inputs.js"; -import type { resolveGatewayCredentialsFromConfig } from "./credentials.js"; - -// Thin public bridge from OpenClawConfig-shaped callers to the lower-level -// credential resolver. Keep this file policy-free; precedence lives in -// credentials-secret-inputs and credentials. -type GatewayCredentialConfigOptions = Parameters[0]; - -/** Connection auth options accepted by gateway clients that already loaded config. */ -type GatewayConnectionAuthOptions = Omit & { - config: OpenClawConfig; -}; - -function toGatewayCredentialOptions( - params: GatewayConnectionAuthOptions, -): GatewayCredentialConfigOptions { - const { config, ...rest } = params; - return { - cfg: config, - ...rest, - }; -} - -/** Resolves gateway connection credentials, including configured SecretRef inputs. */ -export async function resolveGatewayConnectionAuth( - params: GatewayConnectionAuthOptions, -): Promise<{ token?: string; password?: string }> { - return await resolveGatewayCredentialsWithSecretInputs({ - config: params.config, - ...toGatewayCredentialOptions(params), - }); -} diff --git a/src/gateway/connection-auth.test.ts b/src/gateway/credentials-secret-inputs.test.ts similarity index 92% rename from src/gateway/connection-auth.test.ts rename to src/gateway/credentials-secret-inputs.test.ts index 97163817a30e..a0bfce1c65f6 100644 --- a/src/gateway/connection-auth.test.ts +++ b/src/gateway/credentials-secret-inputs.test.ts @@ -1,11 +1,11 @@ -// Gateway connection auth tests document token/password precedence for local, +// Gateway credential resolver tests document token/password precedence for local, // remote, CLI override, env override, and config-secret connection flows. import { describe, expect, it } from "vitest"; import type { OpenClawConfig } from "../config/config.js"; -import { resolveGatewayConnectionAuth } from "./connection-auth.js"; +import { resolveGatewayCredentialsWithSecretInputs } from "./credentials-secret-inputs.js"; type ResolvedAuth = { token?: string; password?: string }; -type GatewayConnectionAuthOptions = Parameters[0]; +type GatewayConnectionAuthOptions = Parameters[0]; type ConnectionAuthCase = { name: string; @@ -41,7 +41,7 @@ const DEFAULT_ENV = { OPENCLAW_GATEWAY_PASSWORD: "env-password", // pragma: allowlist secret } as NodeJS.ProcessEnv; -describe("resolveGatewayConnectionAuth", () => { +describe("resolveGatewayCredentialsWithSecretInputs", () => { const cases: ConnectionAuthCase[] = [ { name: "local mode defaults to config-first token/password", @@ -161,7 +161,7 @@ describe("resolveGatewayConnectionAuth", () => { ]; it.each(cases)("$name", async ({ cfgLocal, env, options, expected }) => { - const asyncResolved = await resolveGatewayConnectionAuth({ + const asyncResolved = await resolveGatewayCredentialsWithSecretInputs({ config: cfgLocal, env, ...options, @@ -187,7 +187,7 @@ describe("resolveGatewayConnectionAuth", () => { LOCAL_SECRET_TOKEN: "resolved-from-secretref", // pragma: allowlist secret } as NodeJS.ProcessEnv; - const resolved = await resolveGatewayConnectionAuth({ + const resolved = await resolveGatewayCredentialsWithSecretInputs({ config, env, }); @@ -199,7 +199,7 @@ describe("resolveGatewayConnectionAuth", () => { it("resolves an env-template local token through the configured auth path", async () => { await expect( - resolveGatewayConnectionAuth({ + resolveGatewayCredentialsWithSecretInputs({ config: cfg({ gateway: { mode: "local", @@ -230,7 +230,7 @@ describe("resolveGatewayConnectionAuth", () => { CONFIG_FIRST_TOKEN: "config-first-token", } as NodeJS.ProcessEnv; - const resolved = await resolveGatewayConnectionAuth({ + const resolved = await resolveGatewayCredentialsWithSecretInputs({ config, env, }); @@ -260,7 +260,7 @@ describe("resolveGatewayConnectionAuth", () => { CONFIG_FIRST_PASSWORD: "config-first-password", // pragma: allowlist secret } as NodeJS.ProcessEnv; - const resolved = await resolveGatewayConnectionAuth({ + const resolved = await resolveGatewayCredentialsWithSecretInputs({ config, env, }); @@ -289,7 +289,7 @@ describe("resolveGatewayConnectionAuth", () => { } as NodeJS.ProcessEnv; await expect( - resolveGatewayConnectionAuth({ + resolveGatewayCredentialsWithSecretInputs({ config, env, }), @@ -316,7 +316,7 @@ describe("resolveGatewayConnectionAuth", () => { } as NodeJS.ProcessEnv; await expect( - resolveGatewayConnectionAuth({ + resolveGatewayCredentialsWithSecretInputs({ config, env, }), diff --git a/src/gateway/operator-approvals-client.ts b/src/gateway/operator-approvals-client.ts index 6f7ea7a5556f..a4dad1c40b32 100644 --- a/src/gateway/operator-approvals-client.ts +++ b/src/gateway/operator-approvals-client.ts @@ -6,6 +6,7 @@ import { GATEWAY_CLIENT_NAMES, } from "../../packages/gateway-protocol/src/client-info.js"; import type { OpenClawConfig } from "../config/types.openclaw.js"; +import { createDeferred } from "../shared/deferred.js"; import { resolveGatewayClientBootstrap } from "./client-bootstrap.js"; import { startGatewayClientWhenEventLoopReady } from "./client-start-readiness.js"; import { GatewayClient, type GatewayClientOptions } from "./client.js"; @@ -81,40 +82,20 @@ export async function withOperatorApprovalsGatewayClient( }, run: (client: GatewayClient) => Promise, ): Promise { - let readySettled = false; - let resolveReady!: () => void; - let rejectReady!: (err: unknown) => void; - const ready = new Promise((resolve, reject) => { - resolveReady = resolve; - rejectReady = reject; - }); - const markReady = () => { - if (readySettled) { - return; - } - readySettled = true; - resolveReady(); - }; - const failReady = (err: unknown) => { - if (readySettled) { - return; - } - readySettled = true; - rejectReady(err); - }; + const ready = createDeferred(); const gatewayClient = await createOperatorApprovalsGatewayClient({ config: params.config, gatewayUrl: params.gatewayUrl, clientDisplayName: params.clientDisplayName, onHelloOk: () => { - markReady(); + ready.resolve(); }, onConnectError: (err) => { - failReady(err); + ready.reject(err); }, onClose: (code, reason) => { - failReady(new Error(`gateway closed (${code}): ${reason}`)); + ready.reject(new Error(`gateway closed (${code}): ${reason}`)); }, }); @@ -129,7 +110,7 @@ export async function withOperatorApprovalsGatewayClient( : "gateway readiness unavailable before approval client start", ); } - await ready; + await ready.promise; return await run(gatewayClient); } finally { await gatewayClient.stopAndWait().catch(() => { diff --git a/src/infra/exec-approval-channel-runtime.ts b/src/infra/exec-approval-channel-runtime.ts index 1eb8c1638061..dc5f7b0b56c0 100644 --- a/src/infra/exec-approval-channel-runtime.ts +++ b/src/infra/exec-approval-channel-runtime.ts @@ -6,6 +6,7 @@ import type { GatewayClient, GatewayReconnectPausedInfo } from "../gateway/clien import { isApprovalMethod } from "../gateway/method-scopes.js"; import { createOperatorApprovalsGatewayClient } from "../gateway/operator-approvals-client.js"; import { createSubsystemLogger } from "../logging/subsystem.js"; +import { createDeferred } from "../shared/deferred.js"; import { getGatewayNativeApprovalRuntime } from "./approval-gateway-runtime-context.js"; import { isGatewayNativeApprovalMethod, @@ -368,22 +369,8 @@ export function createExecApprovalChannelRuntime< return; } - let readySettled = false; - let resolveReady!: () => void; - let rejectReady!: (error: unknown) => void; - const ready = new Promise((resolve, reject) => { - resolveReady = resolve; - rejectReady = reject; - }); + const ready = createDeferred(); let lastConnectError: unknown = null; - const settleReady = (fn: () => void) => { - if (readySettled) { - return; - } - readySettled = true; - // Hello, close, and reconnect-paused callbacks can race during startup. - fn(); - }; const client = await createOperatorApprovalsGatewayClient({ config: adapter.cfg, @@ -392,7 +379,7 @@ export function createExecApprovalChannelRuntime< onEvent: handleGatewayEvent, onHelloOk: () => { log.debug("connected to gateway"); - settleReady(resolveReady); + ready.resolve(); }, onConnectError: (err) => { log.error(`connect error: ${err.message}`); @@ -400,18 +387,14 @@ export function createExecApprovalChannelRuntime< if (readGatewayConnectErrorDetailCode(err)) { return; } - settleReady(() => rejectReady(err)); + ready.reject(err); }, onReconnectPaused: (info) => { - settleReady(() => - rejectReady(new ExecApprovalChannelRuntimeTerminalStartError(info, lastConnectError)), - ); + ready.reject(new ExecApprovalChannelRuntimeTerminalStartError(info, lastConnectError)); }, onClose: (code, reason) => { log.debug(`gateway closed: ${code} ${reason}`); - settleReady(() => - rejectReady(lastConnectError ?? new Error(`gateway closed: ${code} ${reason}`)), - ); + ready.reject(lastConnectError ?? new Error(`gateway closed: ${code} ${reason}`)); }, }); @@ -432,7 +415,7 @@ export function createExecApprovalChannelRuntime< : "gateway readiness unavailable before exec approval runtime start", ); } - await ready; + await ready.promise; if (stopClientIfInactive(client)) { return; } diff --git a/src/node-host/runner.test.ts b/src/node-host/runner.test.ts index b93db0439c3e..d41d3f78e4cf 100644 --- a/src/node-host/runner.test.ts +++ b/src/node-host/runner.test.ts @@ -47,7 +47,7 @@ const mocks = vi.hoisted(() => ({ aborted: false, elapsedMs: 0, })), - resolveGatewayConnectionAuth: vi.fn(async () => ({})), + resolveGatewayCredentialsWithSecretInputs: vi.fn(async () => ({})), activeRuntime: { invoke: vi.fn(async () => {}), handleInput: vi.fn(), @@ -78,8 +78,8 @@ vi.mock("../gateway/client.js", () => ({ }, })); -vi.mock("../gateway/connection-auth.js", () => ({ - resolveGatewayConnectionAuth: mocks.resolveGatewayConnectionAuth, +vi.mock("../gateway/credentials-secret-inputs.js", () => ({ + resolveGatewayCredentialsWithSecretInputs: mocks.resolveGatewayCredentialsWithSecretInputs, })); vi.mock("../infra/device-identity.js", () => ({ @@ -294,7 +294,7 @@ describe("runNodeHost", () => { "event loop readiness timeout", ); - expect(mocks.resolveGatewayConnectionAuth).toHaveBeenCalledWith({ + expect(mocks.resolveGatewayCredentialsWithSecretInputs).toHaveBeenCalledWith({ config: { gateway: { mode: "local", diff --git a/src/node-host/runner.ts b/src/node-host/runner.ts index 5bef2508e0a2..b548c68d44e9 100644 --- a/src/node-host/runner.ts +++ b/src/node-host/runner.ts @@ -11,7 +11,7 @@ import { GatewayClientRequestError, type GatewayReconnectPausedInfo, } from "../gateway/client.js"; -import { resolveGatewayConnectionAuth } from "../gateway/connection-auth.js"; +import { resolveGatewayCredentialsWithSecretInputs } from "../gateway/credentials-secret-inputs.js"; import { loadOrCreateDeviceIdentity } from "../infra/device-identity.js"; import { getMachineDisplayName } from "../infra/machine-name.js"; import { VERSION } from "../version.js"; @@ -153,7 +153,7 @@ async function resolveNodeHostGatewayCredentials(params: { const mode = params.config.gateway?.mode === "remote" ? "remote" : "local"; const configForResolution = mode === "local" ? buildNodeHostLocalAuthConfig(params.config) : params.config; - return await resolveGatewayConnectionAuth({ + return await resolveGatewayCredentialsWithSecretInputs({ config: configForResolution, env: params.env, localPrecedence: "env-first",