// Proves the Gateway node control plane across real authenticated WebSockets. import { randomUUID } from "node:crypto"; import { existsSync } from "node:fs"; import fs from "node:fs/promises"; import os from "node:os"; import path from "node:path"; import { isRecord } from "@openclaw/normalization-core/record-coerce"; import { GatewayClient } from "openclaw/plugin-sdk/gateway-runtime"; import { afterEach, describe, expect, it, vi } from "vitest"; import { type RawData, WebSocketServer } from "ws"; import { startQaGatewayChild } from "../../../../extensions/qa-lab/api.js"; import { GATEWAY_CLIENT_MODES, GATEWAY_CLIENT_NAMES, } from "../../../../packages/gateway-protocol/src/client-info.js"; import { ConnectErrorDetailCodes } from "../../../../packages/gateway-protocol/src/connect-error-details.js"; import { ErrorCodes, MIN_CLIENT_PROTOCOL_VERSION, MIN_NODE_PROTOCOL_VERSION, PROTOCOL_VERSION, type HelloOk, } from "../../../../packages/gateway-protocol/src/index.js"; import type { OpenClawConfig } from "../../../../src/config/types.openclaw.js"; import { loadOrCreateDeviceIdentity, type DeviceIdentity, } from "../../../../src/infra/device-identity.js"; import { useAutoCleanupTempDirTracker } from "../../../helpers/temp-dir.js"; const TEST_TIMEOUT_MS = 180_000; const REQUEST_TIMEOUT_MS = 20_000; const NODE_DISPLAY_NAME = "QA iPhone"; const NODE_CAPS = ["camera", "location"]; const NODE_COMMANDS = ["camera.list", "location.get"]; const NODE_PERMISSIONS = { accessibility: true, camera: true, location: true, }; const FIXTURE_PLUGIN_ID = "qa-gateway-node-rolling-compat"; const FIXTURE_CAPABILITY = "qa-rolling-surface"; const FIXTURE_COMMAND = "qa.rolling.echo"; const FIXTURE_ROUTE = "/qa-rolling-surface"; type GatewayHandle = Awaited>; type GatewayConnection = Pick; type NodeRead = { nodeId: string; displayName?: string; platform?: string; deviceFamily?: string; caps?: string[]; commands?: string[]; permissions?: Record; approvalState?: string; paired?: boolean; connected?: boolean; lastSeenAtMs?: number; lastSeenReason?: string; }; type NodeInvokeFrame = { id?: string; nodeId?: string; command?: string; paramsJSON?: string | null; }; type InvocationRecord = { id: string; nodeId: string; command: string; params: unknown; }; type ConnectEnvelope = { role: string; mode: string; clientName: string; platform: string; deviceFamily?: string; minProtocol: number; maxProtocol: number; }; type StoppableClient = Pick; type StoppableFixture = { stop: () => Promise; }; describe("Gateway node control plane", () => { const tempDirs = useAutoCleanupTempDirTracker(afterEach); it( "pairs, inventories, invokes, and records presence for one remote device", { timeout: TEST_TIMEOUT_MS }, async () => { const gateway = await startQaGatewayChild({ repoRoot: process.cwd(), command: { executablePath: process.execPath, argsPrefix: ["--import", "tsx", "src/entry.ts"], cwd: process.cwd(), usePackagedPlugins: true, }, transportBaseUrl: "http://127.0.0.1", controlUiEnabled: false, runtimeEnvPatch: { OPENCLAW_DISABLE_BUNDLED_PLUGINS: "1", OPENCLAW_SKIP_CHANNELS: "1", OPENCLAW_SKIP_PROVIDERS: "1", OPENCLAW_TEST_MINIMAL_GATEWAY: "1", }, mutateConfig: (cfg) => { const { plugins: _plugins, ...withoutPlugins } = cfg; return { ...withoutPlugins, gateway: { ...cfg.gateway, nodes: { ...cfg.gateway?.nodes, commands: { allow: NODE_COMMANDS }, }, }, }; }, }); const identity = loadOrCreateDeviceIdentity({ path: path.join(gateway.tempRoot, "control-plane-node.sqlite"), }); const invocations: InvocationRecord[] = []; const handlerErrors: Error[] = []; const invocationResponses: Promise[] = []; let operator: GatewayClient | undefined; let node: GatewayClient | undefined; try { operator = await connectOperator(gateway); node = await connectPairedNode({ gateway, identity, operator, onEvent: (event) => { if (event.event !== "node.invoke.request") { return; } const response = respondToInvocation(node, event.payload, invocations).catch( (error: unknown) => { handlerErrors.push(error instanceof Error ? error : new Error(String(error))); }, ); invocationResponses.push(response); }, }); const listed = await waitForApprovedNode(operator, identity.deviceId, gateway.logs); expect(listed).toMatchObject({ nodeId: identity.deviceId, displayName: NODE_DISPLAY_NAME, platform: "ios", deviceFamily: "iPhone", approvalState: "approved", paired: true, connected: true, permissions: NODE_PERMISSIONS, }); expect(listed.caps?.toSorted()).toEqual(NODE_CAPS); expect(listed.commands?.toSorted()).toEqual(NODE_COMMANDS); const described = await operator.request( "node.describe", { nodeId: identity.deviceId }, { timeoutMs: REQUEST_TIMEOUT_MS }, ); expect(described).toMatchObject({ nodeId: identity.deviceId, displayName: NODE_DISPLAY_NAME, platform: "ios", deviceFamily: "iPhone", approvalState: "approved", paired: true, connected: true, permissions: NODE_PERMISSIONS, }); expect(described.caps?.toSorted()).toEqual(NODE_CAPS); expect(described.commands?.toSorted()).toEqual(NODE_COMMANDS); const cameraParams = { includeUnavailable: false }; const cameraResult = await operator.request<{ ok: boolean; nodeId: string; command: string; payload: unknown; }>( "node.invoke", { nodeId: identity.deviceId, command: "camera.list", params: cameraParams, timeoutMs: REQUEST_TIMEOUT_MS, idempotencyKey: randomUUID(), }, { timeoutMs: REQUEST_TIMEOUT_MS }, ); expect(cameraResult).toMatchObject({ ok: true, nodeId: identity.deviceId, command: "camera.list", payload: { cameras: [{ id: "back-wide", position: "back" }], received: cameraParams, }, }); const locationParams = { accuracy: "balanced" }; const locationResult = await operator.request<{ ok: boolean; nodeId: string; command: string; payload: unknown; }>( "node.invoke", { nodeId: identity.deviceId, command: "location.get", params: locationParams, timeoutMs: REQUEST_TIMEOUT_MS, idempotencyKey: randomUUID(), }, { timeoutMs: REQUEST_TIMEOUT_MS }, ); expect(locationResult).toMatchObject({ ok: true, nodeId: identity.deviceId, command: "location.get", payload: { latitude: 37.3318, longitude: -122.0312, received: locationParams, }, }); expect(invocations).toMatchObject([ { nodeId: identity.deviceId, command: "camera.list", params: cameraParams, }, { nodeId: identity.deviceId, command: "location.get", params: locationParams, }, ]); await Promise.all(invocationResponses); expect(handlerErrors).toEqual([]); const aliveSentAtMs = Date.now(); const aliveResult = await node.request<{ ok: boolean; event: string; handled: boolean; reason?: string; }>( "node.event", { event: "node.presence.alive", payload: { trigger: "manual", sentAtMs: aliveSentAtMs, displayName: NODE_DISPLAY_NAME, platform: "ios", deviceFamily: "iPhone", }, }, { timeoutMs: REQUEST_TIMEOUT_MS }, ); expect(aliveResult).toMatchObject({ ok: true, event: "node.presence.alive", handled: true, reason: "persisted", }); const connectedOperator = operator; await vi.waitFor( async () => { const afterAlive = await readNode(connectedOperator, identity.deviceId); expect(afterAlive, gateway.logs()).toMatchObject({ lastSeenReason: "manual", }); expect(afterAlive?.lastSeenAtMs).toBeGreaterThanOrEqual(aliveSentAtMs); const describedAfterAlive = await connectedOperator.request( "node.describe", { nodeId: identity.deviceId }, { timeoutMs: REQUEST_TIMEOUT_MS }, ); expect(describedAfterAlive).toMatchObject({ lastSeenReason: "manual", }); expect(describedAfterAlive.lastSeenAtMs).toBeGreaterThanOrEqual(aliveSentAtMs); }, { timeout: REQUEST_TIMEOUT_MS, interval: 100 }, ); } finally { await Promise.all(invocationResponses); await Promise.allSettled([ ...(node ? [node.stopAndWait({ timeoutMs: 1_000 })] : []), ...(operator ? [operator.stopAndWait({ timeoutMs: 1_000 })] : []), ]); const tempRoot = gateway.tempRoot; await gateway.stop(); expect(existsSync(tempRoot)).toBe(false); } }, ); it( "captures the default node protocol envelope with a synthetic v3 overlap fixture", { timeout: REQUEST_TIMEOUT_MS * 2 }, async () => { expect(MIN_CLIENT_PROTOCOL_VERSION).toBe(PROTOCOL_VERSION); expect(MIN_NODE_PROTOCOL_VERSION).toBeLessThan(PROTOCOL_VERSION); const fixture = await startProtocolEnvelopeFixture(); let node: GatewayClient | undefined; let fixtureHello: HelloOk | undefined; try { node = await connectClient({ gateway: fixture.gateway, role: "node", clientName: GATEWAY_CLIENT_NAMES.IOS_APP, clientDisplayName: NODE_DISPLAY_NAME, mode: GATEWAY_CLIENT_MODES.NODE, platform: "ios", deviceFamily: "iPhone", scopes: [], deviceIdentity: null, onHelloOk: (value) => { fixtureHello = value; }, }); // hello.protocol reports the fixture Gateway's current server protocol. expect(fixtureHello?.protocol).toBe(MIN_NODE_PROTOCOL_VERSION); expect(fixture.connectFrames).toMatchObject([ { role: "node", mode: GATEWAY_CLIENT_MODES.NODE, minProtocol: MIN_NODE_PROTOCOL_VERSION, maxProtocol: PROTOCOL_VERSION, }, ]); await expect(connectOperator(fixture.gateway)).rejects.toThrow(/protocol mismatch/i); expect(fixture.connectFrames).toMatchObject([ { role: "node", mode: GATEWAY_CLIENT_MODES.NODE, minProtocol: MIN_NODE_PROTOCOL_VERSION, maxProtocol: PROTOCOL_VERSION, }, { role: "operator", mode: GATEWAY_CLIENT_MODES.BACKEND, minProtocol: MIN_CLIENT_PROTOCOL_VERSION, maxProtocol: PROTOCOL_VERSION, }, ]); } finally { await stopProtocolEnvelopeFixture(node, fixture); } }, ); it( "keeps one node-host client converged across Gateway upgrades and rollbacks", { timeout: REQUEST_TIMEOUT_MS * 4 }, async () => { const identity = loadOrCreateDeviceIdentity({ path: path.join(tempDirs.make("openclaw-node-host-negotiation-"), "node.sqlite"), }); const fixture = await startProtocolEnvelopeFixture(); const helloProtocols: number[] = []; let node: GatewayClient | undefined; try { // runNodeHost owns one NODE_HOST GatewayClient. Keep this exact instance // across server transitions so production negotiation owns every retry. node = await connectClient({ gateway: fixture.gateway, role: "node", clientName: GATEWAY_CLIENT_NAMES.NODE_HOST, clientDisplayName: "QA node host", mode: GATEWAY_CLIENT_MODES.NODE, platform: "macos", deviceFamily: "Mac", scopes: [], deviceIdentity: identity, onHelloOk: (hello) => { helloProtocols.push(hello.protocol); }, }); expect(helloProtocols).toEqual([MIN_NODE_PROTOCOL_VERSION]); expect(fixture.connectFrames).toMatchObject([ { clientName: GATEWAY_CLIENT_NAMES.NODE_HOST, platform: "macos", deviceFamily: "Mac", minProtocol: PROTOCOL_VERSION, maxProtocol: PROTOCOL_VERSION, }, { clientName: GATEWAY_CLIENT_NAMES.NODE_HOST, platform: "darwin", minProtocol: MIN_NODE_PROTOCOL_VERSION, maxProtocol: MIN_NODE_PROTOCOL_VERSION, }, ]); expect(fixture.connectFrames[1]).not.toHaveProperty("deviceFamily"); fixture.setProtocol(PROTOCOL_VERSION); fixture.disconnectClients("Gateway upgraded to protocol v4"); await vi.waitFor( () => expect(helloProtocols).toEqual([MIN_NODE_PROTOCOL_VERSION, PROTOCOL_VERSION]), { timeout: REQUEST_TIMEOUT_MS, interval: 100, }, ); expect(fixture.connectFrames.slice(-2)).toMatchObject([ { platform: "darwin", minProtocol: MIN_NODE_PROTOCOL_VERSION, maxProtocol: MIN_NODE_PROTOCOL_VERSION, }, { platform: "macos", deviceFamily: "Mac", minProtocol: PROTOCOL_VERSION, maxProtocol: PROTOCOL_VERSION, }, ]); fixture.setProtocol(MIN_NODE_PROTOCOL_VERSION); fixture.disconnectClients("Gateway rolled back to protocol v3"); await vi.waitFor( () => expect(helloProtocols).toEqual([ MIN_NODE_PROTOCOL_VERSION, PROTOCOL_VERSION, MIN_NODE_PROTOCOL_VERSION, ]), { timeout: REQUEST_TIMEOUT_MS, interval: 100, }, ); expect(fixture.connectFrames.slice(-2)).toMatchObject([ { platform: "macos", deviceFamily: "Mac", minProtocol: PROTOCOL_VERSION, maxProtocol: PROTOCOL_VERSION, }, { platform: "darwin", minProtocol: MIN_NODE_PROTOCOL_VERSION, maxProtocol: MIN_NODE_PROTOCOL_VERSION, }, ]); } finally { await stopProtocolEnvelopeFixture(node, fixture); } }, ); it("stops the protocol fixture when client cleanup rejects", async () => { const clientError = new Error("client cleanup failed"); const stopFixture = vi.fn(async () => {}); await expect( stopProtocolEnvelopeFixture( { stopAndWait: vi.fn(async () => { throw clientError; }), }, { stop: stopFixture }, ), ).rejects.toBe(clientError); expect(stopFixture).toHaveBeenCalledOnce(); }); it( "reconnects the same paired identity across v3-only and v4 envelopes", { timeout: TEST_TIMEOUT_MS }, async () => { expect(MIN_NODE_PROTOCOL_VERSION).toBe(3); expect(PROTOCOL_VERSION).toBe(4); const fixture = await createFixturePlugin(); let gateway: GatewayHandle | undefined; let operator: GatewayClient | undefined; let node: GatewayClient | undefined; let proofError: unknown; const cleanupErrors: unknown[] = []; const invocations: InvocationRecord[] = []; const handlerErrors: Error[] = []; const invocationResponses: Promise[] = []; try { gateway = await startQaGatewayChild({ repoRoot: process.cwd(), command: { executablePath: process.execPath, argsPrefix: ["--import", "tsx", "src/entry.ts"], cwd: process.cwd(), usePackagedPlugins: true, }, transportBaseUrl: "http://127.0.0.1", controlUiEnabled: false, runtimeEnvPatch: { OPENCLAW_DISABLE_BUNDLED_PLUGINS: "1", OPENCLAW_SKIP_CHANNELS: "1", OPENCLAW_SKIP_PROVIDERS: "1", OPENCLAW_TEST_MINIMAL_GATEWAY: "1", }, mutateConfig: (cfg) => { // Bundled plugins are disabled for this focused proof, so their // configured slots cannot remain while the fixture plugin is merged. const { slots: _bundledSlots, ...plugins } = cfg.plugins ?? {}; return withFixturePlugin( { ...cfg, plugins, gateway: { ...cfg.gateway, nodes: { ...cfg.gateway?.nodes, commands: { allow: ["camera.list", FIXTURE_COMMAND] }, }, }, }, fixture.pluginDir, ); }, }); const identity = loadOrCreateDeviceIdentity({ path: path.join(gateway.tempRoot, "rolling-compat-node.sqlite"), }); const declaredCaps = ["camera", FIXTURE_CAPABILITY]; const declaredCommands = ["camera.list", FIXTURE_COMMAND]; const onEvent = (event: { event: string; payload?: unknown }) => { if (event.event !== "node.invoke.request") { return; } const response = respondToInvocation(node, event.payload, invocations).catch( (error: unknown) => { handlerErrors.push(error instanceof Error ? error : new Error(String(error))); }, ); invocationResponses.push(response); }; operator = await connectOperator(gateway); let constrainedHello: HelloOk | undefined; node = await connectPairedNode({ gateway, identity, operator, caps: declaredCaps, commands: declaredCommands, minProtocol: MIN_NODE_PROTOCOL_VERSION, maxProtocol: MIN_NODE_PROTOCOL_VERSION, onEvent, onHelloOk: (hello) => { constrainedHello = hello; }, }); // The client envelope is constrained to v3; hello reports the v4 server protocol. expect(constrainedHello?.protocol).toBe(PROTOCOL_VERSION); expect(constrainedHello?.pluginSurfaceUrls).toBeUndefined(); const legacyListed = await waitForApprovedNode(operator, identity.deviceId, gateway.logs); expect(legacyListed.caps).toEqual(["camera"]); expect(legacyListed.commands).toEqual(["camera.list"]); const cameraParams = { includeUnavailable: false }; const cameraResult = await invokeNodeCommand({ operator, nodeId: identity.deviceId, command: "camera.list", params: cameraParams, }); expect(cameraResult).toMatchObject({ ok: true, nodeId: identity.deviceId, command: "camera.list", payload: { cameras: [{ id: "back-wide", position: "back" }], received: cameraParams, }, }); await Promise.all(invocationResponses); expect(handlerErrors).toEqual([]); await node.stopAndWait({ timeoutMs: 1_000 }); node = undefined; await expectNoPendingPairing(operator, identity.deviceId); let currentHello: HelloOk | undefined; node = await connectClient({ gateway, role: "node", clientName: GATEWAY_CLIENT_NAMES.IOS_APP, clientDisplayName: NODE_DISPLAY_NAME, mode: GATEWAY_CLIENT_MODES.NODE, platform: "ios", deviceFamily: "iPhone", scopes: [], caps: declaredCaps, commands: declaredCommands, permissions: NODE_PERMISSIONS, deviceIdentity: identity, minProtocol: PROTOCOL_VERSION, maxProtocol: PROTOCOL_VERSION, onEvent, onHelloOk: (hello) => { currentHello = hello; }, }); expect(currentHello?.protocol).toBe(PROTOCOL_VERSION); const fixtureSurfaceUrl = currentHello?.pluginSurfaceUrls?.[FIXTURE_CAPABILITY]; if (!fixtureSurfaceUrl) { throw new Error("v4 hello omitted the fixture plugin surface URL"); } expect(new URL(fixtureSurfaceUrl).pathname).toMatch(/^\/__openclaw__\/cap\//); const fixtureSurfaceResponse = await fetch(`${fixtureSurfaceUrl}${FIXTURE_ROUTE}`, { signal: AbortSignal.timeout(REQUEST_TIMEOUT_MS), }); expect(fixtureSurfaceResponse.status).toBe(204); await expectNoPendingPairing(operator, identity.deviceId); const currentListed = await waitForConnectedApprovedNode( operator, identity.deviceId, gateway.logs, ); expect(currentListed.caps?.toSorted()).toEqual(declaredCaps.toSorted()); expect(currentListed.commands?.toSorted()).toEqual(declaredCommands.toSorted()); const pluginParams = { message: "rolling-compatible" }; const pluginResult = await invokeNodeCommand({ operator, nodeId: identity.deviceId, command: FIXTURE_COMMAND, params: pluginParams, }); expect(pluginResult).toMatchObject({ ok: true, nodeId: identity.deviceId, command: FIXTURE_COMMAND, payload: { echoed: pluginParams, }, }); expect(invocations).toMatchObject([ { nodeId: identity.deviceId, command: "camera.list", params: cameraParams, }, { nodeId: identity.deviceId, command: FIXTURE_COMMAND, params: pluginParams, }, ]); await Promise.all(invocationResponses); expect(handlerErrors).toEqual([]); } catch (error) { proofError = error; } finally { await Promise.all(invocationResponses); const clientCleanup = await Promise.allSettled([ ...(node ? [node.stopAndWait({ timeoutMs: 1_000 })] : []), ...(operator ? [operator.stopAndWait({ timeoutMs: 1_000 })] : []), ]); for (const result of clientCleanup) { if (result.status === "rejected") { cleanupErrors.push(result.reason); } } if (gateway) { const tempRoot = gateway.tempRoot; try { await gateway.stop(); expect(existsSync(tempRoot)).toBe(false); } catch (error) { cleanupErrors.push(error); } } try { await fixture.cleanup(); expect(existsSync(fixture.root)).toBe(false); } catch (error) { cleanupErrors.push(error); } } const failures = proofError === undefined ? cleanupErrors : [proofError, ...cleanupErrors]; if (failures.length === 1) { throw failures[0]; } if (failures.length > 1) { throw new AggregateError(failures, "gateway node rolling compatibility proof failed"); } }, ); }); async function stopProtocolEnvelopeFixture( node: StoppableClient | undefined, fixture: StoppableFixture, ): Promise { const results = await Promise.allSettled([ ...(node ? [node.stopAndWait({ timeoutMs: 1_000 })] : []), fixture.stop(), ]); const errors = results.flatMap((result) => (result.status === "rejected" ? [result.reason] : [])); if (errors.length === 1) { throw errors[0]; } if (errors.length > 1) { throw new AggregateError(errors, "protocol envelope fixture cleanup failed"); } } async function connectOperator(gateway: GatewayConnection): Promise { return await connectClient({ gateway, role: "operator", clientName: GATEWAY_CLIENT_NAMES.GATEWAY_CLIENT, clientDisplayName: "Gateway node QA operator", mode: GATEWAY_CLIENT_MODES.BACKEND, scopes: ["operator.admin", "operator.pairing", "operator.read", "operator.write"], deviceIdentity: null, }); } async function connectPairedNode(params: { gateway: GatewayConnection; identity: DeviceIdentity; operator: GatewayClient; caps?: string[]; commands?: string[]; minProtocol?: number; maxProtocol?: number; onEvent: (event: { event: string; payload?: unknown }) => void; onHelloOk?: (hello: HelloOk) => void; }): Promise { const connect = () => connectClient({ gateway: params.gateway, role: "node", clientName: GATEWAY_CLIENT_NAMES.IOS_APP, clientDisplayName: NODE_DISPLAY_NAME, mode: GATEWAY_CLIENT_MODES.NODE, platform: "ios", deviceFamily: "iPhone", scopes: [], caps: params.caps ?? NODE_CAPS, commands: params.commands ?? NODE_COMMANDS, permissions: NODE_PERMISSIONS, deviceIdentity: params.identity, minProtocol: params.minProtocol, maxProtocol: params.maxProtocol, onEvent: params.onEvent, onHelloOk: params.onHelloOk, }); try { return await connect(); } catch (error) { if (!isPairingRequired(error)) { throw error; } await approvePendingNodePairing(params.operator, params.identity.deviceId); return await connect(); } } async function connectClient(params: { gateway: GatewayConnection; role: "operator" | "node"; clientName: | typeof GATEWAY_CLIENT_NAMES.GATEWAY_CLIENT | typeof GATEWAY_CLIENT_NAMES.IOS_APP | typeof GATEWAY_CLIENT_NAMES.NODE_HOST; clientDisplayName: string; mode: typeof GATEWAY_CLIENT_MODES.BACKEND | typeof GATEWAY_CLIENT_MODES.NODE; scopes: string[]; platform?: string; deviceFamily?: string; caps?: string[]; commands?: string[]; permissions?: Record; deviceIdentity: DeviceIdentity | null; minProtocol?: number; maxProtocol?: number; onEvent?: (event: { event: string; payload?: unknown }) => void; onHelloOk?: (hello: HelloOk) => void; }): Promise { return await new Promise((resolve, reject) => { let settled = false; const finish = (error?: Error) => { if (settled) { return; } settled = true; clearTimeout(timeout); if (error) { client.stop(); reject(error); return; } resolve(client); }; const client = new GatewayClient({ url: params.gateway.wsUrl, token: params.gateway.token, env: params.gateway.runtimeEnv, role: params.role, clientName: params.clientName, clientDisplayName: params.clientDisplayName, clientVersion: "1.0.0", platform: params.platform ?? process.platform, deviceFamily: params.deviceFamily, mode: params.mode, scopes: params.scopes, caps: params.caps, commands: params.commands, permissions: params.permissions, deviceIdentity: params.deviceIdentity, minProtocol: params.minProtocol, maxProtocol: params.maxProtocol, requestTimeoutMs: REQUEST_TIMEOUT_MS, onEvent: params.onEvent, onHelloOk: (hello) => { params.onHelloOk?.(hello); finish(); }, onConnectError: (error) => finish(error), onClose: (code, reason) => { // NODE_HOST uses this exact close to switch protocol envelopes. Keep the // same client alive until its negotiated hello reaches the caller. if ( params.clientName === GATEWAY_CLIENT_NAMES.NODE_HOST && code === 1008 && reason === "connect retry" ) { return; } finish(new Error(`Gateway closed (${code}): ${reason}`)); }, }); const timeout = setTimeout( () => finish(new Error(`Gateway client connection timed out:\n${params.gateway.logs()}`)), REQUEST_TIMEOUT_MS, ); timeout.unref(); client.start(); }); } function isPairingRequired(error: unknown): boolean { const details = error && typeof error === "object" ? (error as { details?: { code?: unknown } }).details : undefined; return details?.code === "PAIRING_REQUIRED" || String(error).includes("PAIRING_REQUIRED"); } async function approvePendingNodePairing(operator: GatewayClient, nodeId: string): Promise { let deviceRequestId: string | undefined; await vi.waitFor( async () => { const devices = await operator.request<{ pending?: Array<{ requestId?: string; deviceId?: string; role?: string }>; }>("device.pair.list", {}, { timeoutMs: REQUEST_TIMEOUT_MS }); const pendingDevice = devices.pending?.find( (entry) => entry.deviceId === nodeId || entry.role === "node", ); expect(pendingDevice?.requestId).toBeTruthy(); deviceRequestId = pendingDevice?.requestId; }, { timeout: REQUEST_TIMEOUT_MS, interval: 100 }, ); await operator.request( "device.pair.approve", { requestId: deviceRequestId }, { timeoutMs: REQUEST_TIMEOUT_MS }, ); let nodeRequestId: string | undefined; await vi.waitFor( async () => { const nodes = await operator.request<{ pending?: Array<{ requestId?: string; nodeId?: string }>; }>("node.pair.list", {}, { timeoutMs: REQUEST_TIMEOUT_MS }); const pendingNode = nodes.pending?.find((entry) => entry.nodeId === nodeId); expect(pendingNode?.requestId).toBeTruthy(); nodeRequestId = pendingNode?.requestId; }, { timeout: REQUEST_TIMEOUT_MS, interval: 100 }, ); await operator.request( "node.pair.approve", { requestId: nodeRequestId }, { timeoutMs: REQUEST_TIMEOUT_MS }, ); } async function waitForApprovedNode( operator: GatewayClient, nodeId: string, logs: () => string, ): Promise { let approved: NodeRead | undefined; await vi.waitFor( async () => { await approvePendingNodeSurface(operator, nodeId); approved = await readNode(operator, nodeId); expect(approved, logs()).toMatchObject({ nodeId, approvalState: "approved", connected: true, paired: true, }); }, { timeout: REQUEST_TIMEOUT_MS, interval: 100 }, ); if (!approved) { throw new Error(`approved node never became visible:\n${logs()}`); } return approved; } async function waitForConnectedApprovedNode( operator: GatewayClient, nodeId: string, logs: () => string, ): Promise { let approved: NodeRead | undefined; await vi.waitFor( async () => { approved = await readNode(operator, nodeId); expect(approved, logs()).toMatchObject({ nodeId, approvalState: "approved", connected: true, paired: true, }); }, { timeout: REQUEST_TIMEOUT_MS, interval: 100 }, ); if (!approved) { throw new Error(`approved node never became visible:\n${logs()}`); } return approved; } async function approvePendingNodeSurface(operator: GatewayClient, nodeId: string): Promise { for (const pending of await readPendingNodePairings(operator)) { if (pending.nodeId === nodeId && pending.requestId) { await operator.request( "node.pair.approve", { requestId: pending.requestId }, { timeoutMs: REQUEST_TIMEOUT_MS }, ); } } } async function readPendingNodePairings( operator: GatewayClient, ): Promise> { const nodes = await operator.request<{ pending?: Array<{ requestId?: string; nodeId?: string }>; }>("node.pair.list", {}, { timeoutMs: REQUEST_TIMEOUT_MS }); return nodes.pending ?? []; } async function expectNoPendingPairing(operator: GatewayClient, nodeId: string): Promise { const [devices, nodes] = await Promise.all([ operator.request<{ pending?: Array<{ deviceId?: string }>; }>("device.pair.list", {}, { timeoutMs: REQUEST_TIMEOUT_MS }), readPendingNodePairings(operator), ]); const devicePending = devices.pending?.some((entry) => entry.deviceId === nodeId) ?? false; const nodePending = nodes.some((entry) => entry.nodeId === nodeId); expect(devicePending).toBe(false); expect(nodePending).toBe(false); } async function readNode(operator: GatewayClient, nodeId: string): Promise { const result = await operator.request<{ nodes?: NodeRead[] }>( "node.list", {}, { timeoutMs: REQUEST_TIMEOUT_MS }, ); return result.nodes?.find((entry) => entry.nodeId === nodeId); } async function invokeNodeCommand(params: { operator: GatewayClient; nodeId: string; command: string; params: unknown; }): Promise<{ ok: boolean; nodeId: string; command: string; payload: unknown; }> { return await params.operator.request( "node.invoke", { nodeId: params.nodeId, command: params.command, params: params.params, timeoutMs: REQUEST_TIMEOUT_MS, idempotencyKey: randomUUID(), }, { timeoutMs: REQUEST_TIMEOUT_MS }, ); } async function respondToInvocation( node: GatewayClient | undefined, payload: unknown, invocations: InvocationRecord[], ): Promise { const frame = payload as NodeInvokeFrame; if (!node || !frame.id || !frame.nodeId || !frame.command) { throw new Error(`invalid node.invoke.request: ${JSON.stringify(payload)}`); } const params = frame.paramsJSON ? JSON.parse(frame.paramsJSON) : undefined; invocations.push({ id: frame.id, nodeId: frame.nodeId, command: frame.command, params, }); const response = frame.command === "camera.list" ? { cameras: [{ id: "back-wide", position: "back" }], received: params, } : frame.command === "location.get" ? { latitude: 37.3318, longitude: -122.0312, received: params, } : frame.command === FIXTURE_COMMAND ? { echoed: params, } : undefined; if (!response) { throw new Error(`unexpected node command: ${frame.command}`); } await node.request( "node.invoke.result", { id: frame.id, nodeId: frame.nodeId, ok: true, payloadJSON: JSON.stringify(response), }, { timeoutMs: REQUEST_TIMEOUT_MS }, ); } async function createFixturePlugin(): Promise<{ root: string; pluginDir: string; cleanup: () => Promise; }> { const root = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-gateway-node-rolling-")); const pluginDir = path.join(root, FIXTURE_PLUGIN_ID); try { await fs.mkdir(pluginDir, { recursive: true }); await fs.writeFile( path.join(pluginDir, "openclaw.plugin.json"), `${JSON.stringify( { id: FIXTURE_PLUGIN_ID, activation: { onStartup: true }, configSchema: { type: "object", additionalProperties: false, properties: {} }, }, null, 2, )}\n`, "utf8", ); await fs.writeFile( path.join(pluginDir, "index.js"), `module.exports = { id: ${JSON.stringify(FIXTURE_PLUGIN_ID)}, register(api) { api.registerHttpRoute({ path: ${JSON.stringify(FIXTURE_ROUTE)}, auth: "plugin", nodeCapability: { surface: ${JSON.stringify(FIXTURE_CAPABILITY)} }, handler(_req, res) { res.statusCode = 204; res.end(); return true; }, }); api.registerNodeInvokePolicy({ commands: [${JSON.stringify(FIXTURE_COMMAND)}], defaultPlatforms: ["ios"], handle: async (ctx) => await ctx.invokeNode(), }); }, };\n`, "utf8", ); return { root, pluginDir, cleanup: () => fs.rm(root, { force: true, recursive: true }), }; } catch (error) { try { await fs.rm(root, { force: true, recursive: true }); } catch (cleanupError) { const failure = new AggregateError( [error, cleanupError], "fixture plugin setup and cleanup failed", ); failure.cause = error; throw failure; } throw error; } } function withFixturePlugin(config: OpenClawConfig, pluginDir: string): OpenClawConfig { return { ...config, plugins: { ...config.plugins, enabled: true, allow: [...new Set([...(config.plugins?.allow ?? []), FIXTURE_PLUGIN_ID])], load: { ...config.plugins?.load, paths: [...new Set([...(config.plugins?.load?.paths ?? []), pluginDir])], }, entries: { ...config.plugins?.entries, [FIXTURE_PLUGIN_ID]: { enabled: true }, }, }, }; } function parseConnectEnvelope( data: RawData, ): { id: string; envelope: ConnectEnvelope } | undefined { try { const text = Array.isArray(data) ? Buffer.concat(data.map((chunk) => Buffer.from(chunk))).toString("utf8") : Buffer.isBuffer(data) ? data.toString("utf8") : Buffer.from(data).toString("utf8"); const frame = JSON.parse(text); if (!isRecord(frame) || frame.method !== "connect" || typeof frame.id !== "string") { return undefined; } const params = frame.params; if (!isRecord(params) || !isRecord(params.client)) { return undefined; } if ( typeof params.role !== "string" || typeof params.client.id !== "string" || typeof params.client.mode !== "string" || typeof params.client.platform !== "string" || typeof params.minProtocol !== "number" || typeof params.maxProtocol !== "number" ) { return undefined; } return { id: frame.id, envelope: { role: params.role, mode: params.client.mode, clientName: params.client.id, platform: params.client.platform, ...(typeof params.client.deviceFamily === "string" ? { deviceFamily: params.client.deviceFamily } : {}), minProtocol: params.minProtocol, maxProtocol: params.maxProtocol, }, }; } catch { return undefined; } } async function startProtocolEnvelopeFixture(): Promise<{ gateway: GatewayConnection; connectFrames: ConnectEnvelope[]; setProtocol: (protocol: number) => void; disconnectClients: (reason: string) => void; stop: () => Promise; }> { const connectFrames: ConnectEnvelope[] = []; let protocol: number = MIN_NODE_PROTOCOL_VERSION; let challengeSequence = 0; const server = new WebSocketServer({ host: "127.0.0.1", port: 0 }); server.on("connection", (socket) => { socket.send( JSON.stringify({ type: "event", event: "connect.challenge", seq: 1, payload: { nonce: `qa-protocol-envelope-${++challengeSequence}`, ts: Date.now() }, }), ); socket.on("message", (data) => { const connect = parseConnectEnvelope(data); if (!connect) { return; } connectFrames.push(connect.envelope); const supportsProtocol = connect.envelope.minProtocol <= protocol && connect.envelope.maxProtocol >= protocol; if (!supportsProtocol) { socket.send( JSON.stringify({ type: "res", id: connect.id, ok: false, error: { code: ErrorCodes.INVALID_REQUEST, message: "protocol mismatch", details: { code: ConnectErrorDetailCodes.PROTOCOL_MISMATCH, clientMinProtocol: connect.envelope.minProtocol, clientMaxProtocol: connect.envelope.maxProtocol, expectedProtocol: protocol, }, }, }), ); return; } socket.send( JSON.stringify({ type: "res", id: connect.id, ok: true, payload: { type: "hello-ok", protocol, server: { version: `qa-protocol-${protocol}-fixture`, connId: randomUUID() }, features: { methods: [], events: [] }, snapshot: { presence: [], health: {}, stateVersion: { presence: 1, health: 1 }, uptimeMs: 1, }, auth: { role: connect.envelope.role, scopes: [] }, policy: { maxPayload: 512 * 1024, maxBufferedBytes: 1024 * 1024, tickIntervalMs: 30_000, }, }, }), ); }); }); await new Promise((resolve, reject) => { server.once("listening", resolve); server.once("error", reject); }); const address = server.address(); if (!address || typeof address === "string") { throw new Error("v3 Gateway fixture did not bind a TCP port"); } return { connectFrames, gateway: { wsUrl: `ws://127.0.0.1:${address.port}`, token: "qa-v3-gateway-token", runtimeEnv: {}, logs: () => JSON.stringify(connectFrames), }, setProtocol(nextProtocol) { protocol = nextProtocol; }, disconnectClients(reason) { for (const client of server.clients) { client.close(1012, reason); } }, async stop() { for (const client of server.clients) { client.terminate(); } await new Promise((resolve) => { server.close(() => resolve()); }); }, }; }