diff --git a/docs/plugins/sdk-runtime.md b/docs/plugins/sdk-runtime.md index 66e01cd15794..e06a243b06ec 100644 --- a/docs/plugins/sdk-runtime.md +++ b/docs/plugins/sdk-runtime.md @@ -507,6 +507,95 @@ snapshots; OpenClaw owns all persistence and lifecycle coordination. in-flight requests and release local resources. Existing calls that omit the signal retain their previous behavior. + Gateway-loaded plugins can open a connection-scoped binary channel to a + registered node-host command with `nodes.openDuplex(...)`: + + ```typescript + const controller = new AbortController(); + const channel = await api.runtime.nodes.openDuplex({ + nodeId: "paired-node", + command: "my-plugin.image-bridge", + params: { format: "png" }, + timeoutMs: 30000, + maxMessageBytes: 4 * 1024 * 1024, + signal: controller.signal, + }); + + const unsubscribe = channel.onMessage((message: Uint8Array) => { + console.log("Received one complete binary message:", message.byteLength); + }); + + try { + await channel.send(Uint8Array.of(1, 2, 3)); + const result = await channel.closed; + } finally { + unsubscribe(); + channel.close(); + } + ``` + + `openDuplex` accepts the same node, command, parameters, timeout, + idempotency key, session key, caller signal, and requested scopes as + `nodes.invoke`, plus an optional `maxMessageBytes`. The limit defaults to + 100 MiB and can be reduced, but never increased beyond 100 MiB. OpenClaw + splits each binary message into ordered 8 KiB payload fragments that fit the + existing 16 KiB transport-frame limit; callers always send and receive + complete `Uint8Array` messages. Concurrent sends preserve message + boundaries. + + Register the channel's single message listener immediately after + `openDuplex` resolves. Before a listener is registered, OpenClaw buffers at + most eight complete messages and 1 MiB total; exceeding either limit closes + the invocation. The unsubscribe callback removes that listener. Listeners + may return `Promise`; a thrown error or rejected promise, caller + abort, `close()`, node disconnect, pairing change, plugin reload or + retirement, or Gateway shutdown closes the channel and cancels outstanding + node work. Successful node command completion and `channel.closed` wait + for asynchronous message listeners already in progress. `close()` is + idempotent, and retained channel methods reject after closure. + `channel.closed` resolves with the successful command result or rejects + with the node, authorization, transport, or cancellation error. Channels + cannot reconnect or survive a node disconnection. + + The node plugin declares `duplex: true` and registers a message listener + through the optional framed command I/O capability: + + ```typescript + api.registerNodeHostCommand({ + command: "my-plugin.image-bridge", + duplex: true, + async handle(_paramsJSON, io) { + if (!io?.frames) { + throw new Error("Framed node command I/O is unavailable."); + } + + const frames = io.frames; + return await new Promise((resolve, reject) => { + frames.onMessage((message) => { + void frames.send(message).then(() => resolve('{"ok":true}'), reject); + }); + io.signal.addEventListener( + "abort", + () => reject(new Error("Node command was canceled.")), + { once: true }, + ); + }); + }, + }); + ``` + + Register `frames.onMessage(...)` before sending: the node announces framed + readiness only after the listener exists, and `openDuplex` resolves only + after both command dispatch and framed readiness. This prevents input from + arriving before the plugin can consume it. The existing raw `emitChunk` + and `onInput` helpers remain available to terminal-style commands. + + `openDuplex` is available only to a current, trusted in-process Gateway + plugin runtime. Plugin CLI runtimes reject it with an actionable error; + there is no remote polling or local fallback. Every invocation uses the + same pairing, declared-command allowlist, plugin policy, approval, + authorization, and connection-ownership checks as `nodes.invoke`. + `nodes.list(...)` includes each connected node's advertised `nodePluginTools` descriptors when that node exposes plugin or MCP-backed tools to the agent. Those descriptors are live connection state: the Gateway @@ -518,7 +607,7 @@ snapshots; OpenClaw owns all persistence and lifecycle coordination. Plugins that expose node-hosted agent tools can set `agentTool.defaultPlatforms` for non-dangerous commands that should be allowlisted by default. Omit it when operators must opt in with `gateway.nodes.commands.allow`. Dangerous node-host commands should register a node-invoke policy with `api.registerNodeInvokePolicy(...)`; the policy runs in the Gateway after command allowlist checks and before the command is forwarded to the node, so direct `node.invoke` calls, node-hosted plugin tools, and higher-level plugin tools share the same enforcement path. - The optional `scopes` field requests Gateway operator scopes for the invocation. OpenClaw honors it only for bundled plugins and trusted official plugin installations; requests from other plugins do not elevate the call. Use it only when a trusted plugin must invoke a node command with a stricter Gateway scope, such as `operator.admin`. + The optional `scopes` field requests Gateway operator scopes for the invocation. OpenClaw honors it only for bundled plugins and trusted official plugin installations; requests from other plugins do not elevate the call. When `openDuplex` runs inside an authenticated Gateway request, its effective scopes never exceed that authenticated caller's actual scopes, even if a trusted plugin requests stronger scopes. Without an authenticated incoming client, existing trusted-plugin scope behavior applies. Use requested scopes only when a trusted plugin must invoke a node command with a stricter Gateway scope, such as `operator.admin`. diff --git a/extensions/canvas/src/widget-presenter.test.ts b/extensions/canvas/src/widget-presenter.test.ts index 2d3ed338133f..78b52f6eb16f 100644 --- a/extensions/canvas/src/widget-presenter.test.ts +++ b/extensions/canvas/src/widget-presenter.test.ts @@ -10,6 +10,7 @@ function createNodesRuntime( return { list: vi.fn(async () => ({ nodes })), invoke: vi.fn(async () => ({ ok: true })), + openDuplex: vi.fn(), }; } diff --git a/src/gateway/node-invoke-plugin-policy.test.ts b/src/gateway/node-invoke-plugin-policy.test.ts index b7db8cca9714..da7ee4b05a04 100644 --- a/src/gateway/node-invoke-plugin-policy.test.ts +++ b/src/gateway/node-invoke-plugin-policy.test.ts @@ -69,6 +69,8 @@ function createContext(opts?: { const invoke = vi.fn( async (params?: { onDispatchReady?: (invokeId: string) => void; + onProgress?: (chunk: string) => void; + isDispatchAuthorized?: () => boolean; }): Promise => { params?.onDispatchReady?.("invoke-1"); return { @@ -84,7 +86,11 @@ function createContext(opts?: { getRuntimeConfig: opts?.getRuntimeConfig ?? (() => ({ gateway: { nodes: { commands: { allow: [DEMO_COMMAND] } } } })), - nodeRegistry: { get: () => nodeSession, invoke }, + nodeRegistry: { + get: () => nodeSession, + getForPairingGeneration: () => nodeSession, + invoke, + }, broadcast: vi.fn(), broadcastToConnIds: vi.fn(), pluginApprovalManager: opts?.pluginApprovalManager, @@ -283,6 +289,77 @@ describe("applyPluginNodeInvokePolicy", () => { }); }); + it("streams approved dangerous commands through the existing scoped policy transport", async () => { + const manager = new ExecApprovalManager(); + const nodeSession = createNodeSession(); + nodeSession.pairingGeneration = "paired-generation-1"; + const reviewer = createOperatorClient(); + reviewer.connId = "conn-owner-approval"; + setDangerousDemoCommandRegistry([ + createDemoPolicy(async (policyContext) => { + expect(policyContext.client?.scopes).toEqual(["operator.approvals"]); + const approval = await policyContext.approvals?.request({ + title: "Open fixture duplex", + description: "Approve the declared node command", + }); + if (approval?.decision !== "allow-once") { + return { ok: false, code: "APPROVAL_DENIED", message: "node command was not approved" }; + } + return await policyContext.invokeNode(); + }), + ]); + const { context, invoke } = createContext({ + nodeSession, + pluginApprovalManager: manager, + getApprovalClientConnIds: createApprovalClientLookup([reviewer]), + }); + let runtimeCurrent = true; + const stream = { + onProgress: vi.fn(), + onDispatchReady: vi.fn(), + idleTimeoutMs: 5_000, + isRuntimeCurrent: () => runtimeCurrent, + }; + invoke.mockImplementationOnce(async (params) => { + params?.onDispatchReady?.("approved-duplex-invoke"); + params?.onProgress?.("approved-duplex-progress"); + return { ok: true, payload: { approved: true }, payloadJSON: null, error: null }; + }); + const resultPromise = applyPluginNodeInvokePolicy({ + context, + client: { + ...createOperatorClient(), + internal: { + syntheticClient: true, + pluginRuntimeOwnerId: DEMO_PLUGIN_ID, + nodeInvokeStream: stream, + }, + }, + nodeSession, + command: DEMO_COMMAND, + params: DEMO_PARAMS, + nodeInvokeStream: stream, + }); + + const approval = await expectSinglePendingApproval(manager); + expect(invoke).not.toHaveBeenCalled(); + expect(manager.resolve(approval.id, "allow-once")).toBe(true); + + await expect(resultPromise).resolves.toMatchObject({ ok: true }); + expect(stream.onDispatchReady).toHaveBeenCalledWith("approved-duplex-invoke"); + expect(stream.onProgress).toHaveBeenCalledWith("approved-duplex-progress"); + expect(invoke).toHaveBeenCalledWith( + expect.objectContaining({ + expectedConnId: "conn-1", + expectedPairingGeneration: "paired-generation-1", + idleTimeoutMs: 5_000, + }), + ); + + runtimeCurrent = false; + expect(invoke.mock.calls[0]?.[0]?.isDispatchAuthorized?.()).toBe(false); + }); + it("classifies exact arguments before the policy handler and transport", async () => { const policy = createDemoPolicy((ctx: OpenClawPluginNodeInvokePolicyContext) => { expect(ctx.risk).toEqual({ level: "high", family: "fixture_mutation" }); diff --git a/src/gateway/node-invoke-plugin-policy.ts b/src/gateway/node-invoke-plugin-policy.ts index ac2c82bd5681..125bcae1abc8 100644 --- a/src/gateway/node-invoke-plugin-policy.ts +++ b/src/gateway/node-invoke-plugin-policy.ts @@ -24,6 +24,7 @@ import { buildRequestedApprovalEvent, handlePendingApprovalRequest, } from "./server-methods/approval-shared.js"; +import type { GatewayNodeInvokeStream } from "./server-methods/shared-types.js"; import type { GatewayClient, GatewayRequestContext, RespondFn } from "./server-methods/types.js"; // Plugin node.invoke policies are the last gateway-side guard before a @@ -241,6 +242,7 @@ export async function applyPluginNodeInvokePolicy(params: { signal?: AbortSignal; resolveRemainingTimeoutMs?: () => number | undefined; onNodeCommandDispatched?: () => void; + nodeInvokeStream?: GatewayNodeInvokeStream; idempotencyKey?: string; isInvocationCurrent?: () => boolean | Promise; isApprovalAuthorityActive?: () => boolean; @@ -394,15 +396,21 @@ export async function applyPluginNodeInvokePolicy(params: { timeoutMs, ...(params.signal ? { signal: params.signal } : {}), idempotencyKey: override.idempotencyKey ?? params.idempotencyKey, + ...(params.nodeInvokeStream && { + onProgress: params.nodeInvokeStream.onProgress, + idleTimeoutMs: params.nodeInvokeStream.idleTimeoutMs, + }), isDispatchAuthorized: () => + (params.nodeInvokeStream?.isRuntimeCurrent() ?? true) && (!callerIdentity || params.context.validateAgentRuntimeApprovalAuthority?.(callerIdentity) === true) && params.isApprovalAuthorityActive?.() !== false, - onDispatchReady: () => { + onDispatchReady: (invokeId) => { // Only the registry knows that the transport send succeeded. Preserve // pre-send failures as retry-safe while making later failures ambiguous. nodeCommandDispatched = true; params.onNodeCommandDispatched?.(); + params.nodeInvokeStream?.onDispatchReady(invokeId); }, }); if (!res.ok) { diff --git a/src/gateway/server-methods/nodes.invoke-abort.test.ts b/src/gateway/server-methods/nodes.invoke-abort.test.ts index 7961466c3420..10dcbd00074b 100644 --- a/src/gateway/server-methods/nodes.invoke-abort.test.ts +++ b/src/gateway/server-methods/nodes.invoke-abort.test.ts @@ -4,7 +4,7 @@ import { NODE_WORKER_PRIVATE_COMMANDS } from "../../infra/node-commands.js"; import { isNodeWakeLifecycleCurrent } from "../node-wake-state.js"; import { resetNodeWakeStateForTest } from "../node-wake-state.test-support.js"; import { nodeInvokeHandlers } from "./nodes.invoke.js"; -import type { GatewayRequestHandlerOptions } from "./shared-types.js"; +import type { GatewayNodeInvokeStream, GatewayRequestHandlerOptions } from "./shared-types.js"; const mocks = vi.hoisted(() => ({ captureNodePairingGeneration: vi.fn(async (nodeId: string) => ({ @@ -12,13 +12,19 @@ const mocks = vi.hoisted(() => ({ key: `generation:${nodeId}:1`, })), isNodePairingGenerationCurrent: vi.fn(async () => true), - isNodeCommandAllowed: vi.fn(() => ({ ok: true as const })), + isNodeCommandAllowed: vi.fn((): { ok: true } | { ok: false; reason: string } => ({ ok: true })), resolveNodeCommandAllowlist: vi.fn(() => new Set()), applyPluginNodeInvokePolicy: vi.fn(async () => undefined), - sanitizeNodeInvokeParamsForForwarding: vi.fn(({ rawParams }: { rawParams: unknown }) => ({ - ok: true as const, - params: rawParams, - })), + sanitizeNodeInvokeParamsForForwarding: vi.fn( + ({ + rawParams, + }: { + rawParams: unknown; + }): { ok: true; params: unknown } | { ok: false; message: string } => ({ + ok: true, + params: rawParams, + }), + ), })); vi.mock("../../infra/device-pairing-node-state.js", () => ({ @@ -64,6 +70,7 @@ function startNodeInvoke(options: { command?: string; config?: Record; commands?: string[]; + client?: GatewayRequestHandlerOptions["client"]; }) { const respond = vi.fn(); const handler = nodeInvokeHandlers["node.invoke"]; @@ -79,7 +86,7 @@ function startNodeInvoke(options: { timeoutMs: 10_000, idempotencyKey: "paired-inference-idempotency-key", }, - client: null, + client: options.client ?? null, isWebchatConnect: () => false, respond, context: { @@ -99,16 +106,178 @@ function startNodeInvoke(options: { return { invocation, respond }; } +function createNodeInvokeStreamClient( + stream: GatewayNodeInvokeStream, + options?: { synthetic?: boolean; owner?: boolean }, +): NonNullable { + return { + connect: { + minProtocol: 3, + maxProtocol: 3, + client: { id: "gateway-client", version: "internal", platform: "node", mode: "backend" }, + role: "operator", + scopes: ["operator.write"], + }, + internal: { + ...(options?.synthetic === false ? {} : { syntheticClient: true }), + ...(options?.owner === false ? {} : { pluginRuntimeOwnerId: "duplex-fixture" }), + nodeInvokeStream: stream, + }, + }; +} + describe("node.invoke caller cancellation", () => { + it("carries trusted plugin duplex hooks through the canonical paired dispatch", async () => { + let runtimeCurrent = true; + const stream = { + onProgress: vi.fn(), + onDispatchReady: vi.fn(), + idleTimeoutMs: 5_000, + isRuntimeCurrent: () => runtimeCurrent, + }; + const invoke = vi.fn( + async (params: { + onProgress?: (chunk: string) => void; + onDispatchReady?: (invokeId: string) => void; + isDispatchAuthorized?: () => boolean; + }) => { + params.onDispatchReady?.("paired-stream-invoke"); + params.onProgress?.("paired-stream-progress"); + return { ok: true, payload: { delivered: true } }; + }, + ); + + const { invocation, respond } = startNodeInvoke({ + invoke, + client: createNodeInvokeStreamClient(stream), + }); + await invocation; + + expect(stream.onDispatchReady).toHaveBeenCalledWith("paired-stream-invoke"); + expect(stream.onProgress).toHaveBeenCalledWith("paired-stream-progress"); + expect(invoke).toHaveBeenCalledWith( + expect.objectContaining({ + expectedConnId: "paired-node-connection", + expectedPairingGeneration: "generation:paired-node:1", + idleTimeoutMs: 5_000, + }), + ); + expect(respond).toHaveBeenCalledWith( + true, + expect.objectContaining({ nodeId: "paired-node", command: "ollama.chat" }), + undefined, + ); + + runtimeCurrent = false; + expect(invoke.mock.calls[0]?.[0].isDispatchAuthorized?.()).toBe(false); + }); + + it.each([ + { name: "network client", synthetic: false }, + { name: "ownerless synthetic client", owner: false }, + ])("ignores duplex hooks on an untrusted $name", async (clientOptions) => { + const stream = { + onProgress: vi.fn(), + onDispatchReady: vi.fn(), + isRuntimeCurrent: () => false, + }; + const invoke = vi.fn( + async (params: { + onProgress?: (chunk: string) => void; + onDispatchReady?: (invokeId: string) => void; + isDispatchAuthorized?: () => boolean; + }) => { + params.onDispatchReady?.("untrusted-stream-invoke"); + params.onProgress?.("untrusted-stream-progress"); + return { ok: true, payload: {} }; + }, + ); + + const { invocation } = startNodeInvoke({ + invoke, + client: createNodeInvokeStreamClient(stream, clientOptions), + }); + await invocation; + + expect(stream.onDispatchReady).not.toHaveBeenCalled(); + expect(stream.onProgress).not.toHaveBeenCalled(); + expect(invoke.mock.calls[0]?.[0].isDispatchAuthorized?.()).toBe(true); + }); + + it("rejects undeclared commands before trusted duplex hooks receive dispatch", async () => { + mocks.isNodeCommandAllowed.mockReturnValueOnce({ + ok: false, + reason: "command not declared by node", + }); + const stream = { + onProgress: vi.fn(), + onDispatchReady: vi.fn(), + isRuntimeCurrent: () => true, + }; + const invoke = vi.fn(); + + const { invocation, respond } = startNodeInvoke({ + invoke, + command: "plugin.undeclared", + commands: ["ollama.chat"], + client: createNodeInvokeStreamClient(stream), + }); + await invocation; + + expect(invoke).not.toHaveBeenCalled(); + expect(stream.onDispatchReady).not.toHaveBeenCalled(); + expect(respond).toHaveBeenCalledWith( + false, + undefined, + expect.objectContaining({ message: expect.stringContaining("does not support") }), + ); + }); + + it("does not bypass system.run approval sanitization for trusted duplex hooks", async () => { + mocks.sanitizeNodeInvokeParamsForForwarding.mockReturnValueOnce({ + ok: false, + message: "system.run approval could not be verified", + }); + const stream = { + onProgress: vi.fn(), + onDispatchReady: vi.fn(), + isRuntimeCurrent: () => true, + }; + const invoke = vi.fn(); + + const { invocation, respond } = startNodeInvoke({ + invoke, + command: "system.run", + commands: ["system.run"], + client: createNodeInvokeStreamClient(stream), + }); + await invocation; + + expect(mocks.sanitizeNodeInvokeParamsForForwarding).toHaveBeenCalledOnce(); + expect(invoke).not.toHaveBeenCalled(); + expect(stream.onDispatchReady).not.toHaveBeenCalled(); + expect(respond).toHaveBeenCalledWith( + false, + undefined, + expect.objectContaining({ message: "system.run approval could not be verified" }), + ); + }); + it.each(NODE_WORKER_PRIVATE_COMMANDS)( "rejects private control %s before public policy and dispatch", async (command) => { const invoke = vi.fn(); + const stream = { + onProgress: vi.fn(), + onDispatchReady: vi.fn(), + isRuntimeCurrent: () => true, + }; const { invocation, respond } = startNodeInvoke({ invoke, command, commands: [command], config: { gateway: { nodes: { commands: { allow: [command] } } } }, + client: createNodeInvokeStreamClient(stream), }); await invocation; @@ -116,6 +285,7 @@ describe("node.invoke caller cancellation", () => { expect(invoke).not.toHaveBeenCalled(); expect(mocks.resolveNodeCommandAllowlist).not.toHaveBeenCalled(); expect(mocks.applyPluginNodeInvokePolicy).not.toHaveBeenCalled(); + expect(stream.onDispatchReady).not.toHaveBeenCalled(); expect(respond).toHaveBeenCalledWith( false, undefined, diff --git a/src/gateway/server-methods/nodes.invoke.ts b/src/gateway/server-methods/nodes.invoke.ts index d93eb8cff18d..a5d9e999df73 100644 --- a/src/gateway/server-methods/nodes.invoke.ts +++ b/src/gateway/server-methods/nodes.invoke.ts @@ -81,6 +81,10 @@ export const nodeInvokeHandlers: GatewayRequestHandlers = { const nodeId = normalizeOptionalString(p.nodeId) ?? ""; const command = normalizeOptionalString(p.command) ?? ""; const sessionKey = normalizeOptionalString(p.sessionKey); + const nodeInvokeStream = + client?.internal?.syntheticClient === true && client.internal.pluginRuntimeOwnerId + ? client.internal.nodeInvokeStream + : undefined; if (!nodeId || !command) { respond( false, @@ -454,6 +458,7 @@ export const nodeInvokeHandlers: GatewayRequestHandlers = { isInvocationCurrent: () => isNodePairingWorkCurrent({ nodeId, generation, lifecycle: wakeLifecycle }), isApprovalAuthorityActive: isForwardedApprovalAuthorityActive, + ...(nodeInvokeStream ? { nodeInvokeStream } : {}), }), invokeDeadlineAtMs, ); @@ -578,14 +583,20 @@ export const nodeInvokeHandlers: GatewayRequestHandlers = { signal: invocationLifecycle, idempotencyKey: p.idempotencyKey, ...(sessionKey ? { sessionKey } : {}), + ...(nodeInvokeStream && { + onProgress: nodeInvokeStream.onProgress, + idleTimeoutMs: nodeInvokeStream.idleTimeoutMs, + }), isDispatchAuthorized: () => + (nodeInvokeStream?.isRuntimeCurrent() ?? true) && resolveNodeInvokeRuntimeAuthorityError({ context, client, approvalAuthority: forwardedParams.approvalAuthority, }) === undefined, - onDispatchReady: () => { + onDispatchReady: (invokeId) => { nodeCommandDispatched = true; + nodeInvokeStream?.onDispatchReady(invokeId); }, }); if (!(await continuePairingWork())) { diff --git a/src/gateway/server-methods/session-catalog.test.ts b/src/gateway/server-methods/session-catalog.test.ts index 2c4205298928..3c03221afc5e 100644 --- a/src/gateway/server-methods/session-catalog.test.ts +++ b/src/gateway/server-methods/session-catalog.test.ts @@ -476,7 +476,11 @@ describe("session catalog Gateway methods", () => { bindPluginRegistryRuntime( hoisted.activeRegistry as PluginRegistry, createPluginRuntime({ - nodes: { list: dispatchNodeList, invoke: vi.fn(async () => undefined) }, + nodes: { + list: dispatchNodeList, + invoke: vi.fn(async () => undefined), + openDuplex: vi.fn(), + }, }), ); const catalogUsingNodes = (id: string) => @@ -503,7 +507,11 @@ describe("session catalog Gateway methods", () => { bindPluginRegistryRuntime( hoisted.activeRegistry as PluginRegistry, createPluginRuntime({ - nodes: { list: dispatchNodeList, invoke: vi.fn(async () => undefined) }, + nodes: { + list: dispatchNodeList, + invoke: vi.fn(async () => undefined), + openDuplex: vi.fn(), + }, }), ); const selectedList = vi.fn(async () => []); diff --git a/src/gateway/server-methods/shared-types.ts b/src/gateway/server-methods/shared-types.ts index fe52d3314397..9842845110ee 100644 --- a/src/gateway/server-methods/shared-types.ts +++ b/src/gateway/server-methods/shared-types.ts @@ -89,6 +89,14 @@ export type TrustedAgentToolCaller = Readonly<{ sessionKey: string; }>; +/** Closure-bound streaming hooks attached only to trusted plugin-owned synthetic clients. */ +export type GatewayNodeInvokeStream = { + onProgress: (chunk: string) => void; + onDispatchReady: (invokeId: string) => void; + idleTimeoutMs?: number; + isRuntimeCurrent: () => boolean; +}; + /** Per-connection client metadata captured after the gateway handshake. */ export type GatewayClient = { connect: ConnectParams; @@ -128,6 +136,8 @@ export type GatewayClient = { cronRunContinuation?: boolean; agentRuntimeIdentity?: AgentRuntimeIdentity; pluginRuntimeOwnerId?: string; + /** Plugin-owned in-process invoke hooks; never accepted from Gateway wire params. */ + nodeInvokeStream?: GatewayNodeInvokeStream; agentRunTracking?: GatewayAgentRunTaskOwner; /** Host-captured requester lineage for opt-in plugin subagent completion delivery. */ pluginSubagentRequester?: PluginSubagentRequesterContext; diff --git a/src/gateway/server-plugin-in-process-dispatch.authorization.test.ts b/src/gateway/server-plugin-in-process-dispatch.authorization.test.ts index 3f070660e847..effc2c622870 100644 --- a/src/gateway/server-plugin-in-process-dispatch.authorization.test.ts +++ b/src/gateway/server-plugin-in-process-dispatch.authorization.test.ts @@ -5,10 +5,17 @@ import { GATEWAY_CLIENT_MODES, } from "../../packages/gateway-protocol/src/client-info.js"; import { PROTOCOL_VERSION } from "../../packages/gateway-protocol/src/version.js"; +import { createOperationalRunInstanceRef } from "../agents/admitted-run-context.js"; import { upsertSessionEntryCore } from "../config/sessions/session-accessor.js"; import { withPluginRuntimeGatewayRequestScope } from "../plugins/runtime/gateway-request-scope.js"; import { withOpenClawTestState } from "../test-utils/openclaw-test-state.js"; -import type { GatewayRequestContext, GatewayRequestOptions } from "./server-methods/types.js"; +import { createGatewayMethodRegistry } from "./methods/registry.js"; +import { resolveNodeInvokeRuntimeAuthorityError } from "./server-methods/nodes.invoke-authority.js"; +import type { + GatewayRequestContext, + GatewayRequestHandlerOptions, + GatewayRequestOptions, +} from "./server-methods/types.js"; import { dispatchGatewayMethodInProcess } from "./server-plugin-in-process-dispatch.js"; const startTurn = vi.hoisted(() => vi.fn()); @@ -104,6 +111,129 @@ describe("typed in-process agent authorization", () => { waitForTurn.mockReset(); }); + it.each([ + { + name: "non-synthetic client", + method: "node.invoke", + options: { pluginRuntimeOwnerId: "duplex-fixture" }, + }, + { + name: "ownerless synthetic client", + method: "node.invoke", + options: { forceSyntheticClient: true }, + }, + { + name: "different gateway method", + method: "node.list", + options: { forceSyntheticClient: true, pluginRuntimeOwnerId: "duplex-fixture" }, + }, + ])("rejects node duplex hooks on an $name", async ({ method, options }) => { + const onDispatchReady = vi.fn(); + + await expect( + dispatchGatewayMethodInProcess( + method, + {}, + { + ...options, + nodeInvokeStream: { + onProgress: vi.fn(), + onDispatchReady, + isRuntimeCurrent: () => true, + }, + resolveGatewayContext: createContext, + }, + ), + ).rejects.toThrow("owner-bound trusted synthetic client"); + + expect(onDispatchReady).not.toHaveBeenCalled(); + }); + + it("retains the authenticated caller and its closure-bound authority for node duplex", async () => { + const client = createOperatorClient({ + profileId: "duplex-owner", + scopes: ["operator.write", "operator.approvals"], + }); + const operationalRunInstance = createOperationalRunInstanceRef("duplex-owned-run"); + const agentRuntimeIdentity = { + kind: "agentRuntime" as const, + agentId: "main", + sessionKey: "agent:main:duplex-owner", + operationalRunInstance, + delegatedAuthority: { + kind: "local" as const, + operationalRunInstance, + lifecycleGeneration: "duplex-generation", + claimId: "duplex-claim", + }, + }; + client.isDeviceTokenAuth = true; + client.internal = { + agentRuntimeIdentity, + approvalRuntime: true, + senderAttribution: { id: "duplex-sender" }, + }; + let authorityCurrent = true; + const dispatched: { client: GatewayRequestOptions["client"] } = { client: null }; + const context = createContext(); + context.validateAgentRuntimeApprovalAuthority = (identity) => + authorityCurrent && identity === agentRuntimeIdentity; + const methodRegistry = createGatewayMethodRegistry([ + { + name: "node.invoke", + scope: "operator.write", + owner: { kind: "core", area: "nodes" }, + handler: ({ client: resolvedClient, respond }: GatewayRequestHandlerOptions) => { + dispatched.client = resolvedClient; + respond(true, { ok: true }); + }, + }, + ]); + context.getGatewayMethodRegistry = () => methodRegistry; + + await withPluginRuntimeGatewayRequestScope( + { client, context, isWebchatConnect: () => false }, + async () => + await dispatchGatewayMethodInProcess( + "node.invoke", + {}, + { + forceSyntheticClient: true, + pluginRuntimeOwnerId: "duplex-fixture", + syntheticScopes: ["operator.write", "operator.approvals"], + nodeInvokeStream: { + onProgress: vi.fn(), + onDispatchReady: vi.fn(), + isRuntimeCurrent: () => true, + }, + }, + ), + ); + + expect(dispatched.client).toMatchObject({ + connId: "conn-duplex-owner", + authenticatedUserId: "duplex-owner@example.com", + authenticatedUserProfile: { profileId: "duplex-owner" }, + isDeviceTokenAuth: true, + connect: { scopes: ["operator.write", "operator.approvals"] }, + internal: { + syntheticClient: true, + pluginRuntimeOwnerId: "duplex-fixture", + approvalRuntime: true, + senderAttribution: { id: "duplex-sender" }, + }, + }); + expect(dispatched.client?.internal?.agentRuntimeIdentity).toBe(agentRuntimeIdentity); + expect( + resolveNodeInvokeRuntimeAuthorityError({ context, client: dispatched.client }), + ).toBeUndefined(); + + authorityCurrent = false; + expect(resolveNodeInvokeRuntimeAuthorityError({ context, client: dispatched.client })).toBe( + "agent runtime approval authority closed before node dispatch", + ); + }); + it("rejects a scoped agent turn without operator.write", async () => { await expect( dispatchScopedAgent({ diff --git a/src/gateway/server-plugin-in-process-dispatch.ts b/src/gateway/server-plugin-in-process-dispatch.ts index a9ad283a6b82..cf1b91815811 100644 --- a/src/gateway/server-plugin-in-process-dispatch.ts +++ b/src/gateway/server-plugin-in-process-dispatch.ts @@ -15,6 +15,7 @@ import type { TrustedSessionCreation } from "./server-methods/session-creation-p import type { GatewayAgentRunTaskOwner, GatewayContextResolver, + GatewayNodeInvokeStream, GatewayRequestContext, GatewayRequestOptions, TrustedAgentToolCaller, @@ -42,6 +43,7 @@ type DispatchGatewayMethodInProcessOptions = { forceSyntheticClient?: boolean; internalDeliveryMediaUrls?: string[]; internalDeliverySuppressText?: boolean; + nodeInvokeStream?: GatewayNodeInvokeStream; onAccepted?: (payload: unknown) => void; onSignalAbort?: () => Promise | void; pluginRuntimeOwnerId?: string; @@ -87,6 +89,12 @@ function resolveInProcessGatewayDispatch( typeof options?.pluginRuntimeOwnerId === "string" && options.pluginRuntimeOwnerId.trim() ? options.pluginRuntimeOwnerId.trim() : undefined; + if ( + options?.nodeInvokeStream && + (method !== "node.invoke" || !pluginRuntimeOwnerId || options.forceSyntheticClient !== true) + ) { + throw new Error("Node invoke streaming requires an owner-bound trusted synthetic client."); + } const delegatedToolPolicyHandoffId = options?.delegatedToolPolicyHandoff ? registerSubagentCompletionToolHandoff(options.delegatedToolPolicyHandoff) : undefined; @@ -111,13 +119,30 @@ function resolveInProcessGatewayDispatch( ...(options?.sessionCreation ? { sessionCreation: options.sessionCreation } : {}), scopes: options?.syntheticScopes, }); - const agentRuntimeIdentity = readInProcessAgentRuntimeIdentity(options); - const syntheticClient = agentRuntimeIdentity - ? { - ...baseSyntheticClient, - internal: { ...baseSyntheticClient.internal, agentRuntimeIdentity }, - } - : baseSyntheticClient; + const scopedStreamClient = options?.nodeInvokeStream ? scope?.client : undefined; + const agentRuntimeIdentity = + scopedStreamClient?.internal?.agentRuntimeIdentity ?? + readInProcessAgentRuntimeIdentity(options); + const syntheticClient = + agentRuntimeIdentity || options?.nodeInvokeStream + ? { + ...(scopedStreamClient ?? baseSyntheticClient), + ...(scopedStreamClient + ? { + connect: { + ...scopedStreamClient.connect, + scopes: baseSyntheticClient.connect.scopes, + }, + } + : {}), + internal: { + ...scopedStreamClient?.internal, + ...baseSyntheticClient.internal, + ...(agentRuntimeIdentity ? { agentRuntimeIdentity } : {}), + ...(options?.nodeInvokeStream ? { nodeInvokeStream: options.nodeInvokeStream } : {}), + }, + } + : baseSyntheticClient; const scopedClient = mergePluginRuntimeClientInternal( scope?.client, pluginRuntimeOwnerId || diff --git a/src/gateway/server-plugins-node-runtime.ts b/src/gateway/server-plugins-node-runtime.ts index 019849d28ba8..a4ae427d30a5 100644 --- a/src/gateway/server-plugins-node-runtime.ts +++ b/src/gateway/server-plugins-node-runtime.ts @@ -1,6 +1,12 @@ +import { NODE_DUPLEX_INVOKE_IDLE_TIMEOUT_MS } from "../infra/node-commands.js"; +import { createNodeDuplexEndpoint } from "../infra/node-duplex-framing.js"; import { getPluginRuntimeGatewayRequestScope } from "../plugins/runtime/gateway-request-scope.js"; +import type { PluginRuntime } from "../plugins/runtime/types.js"; +import { createDeferredCore } from "../shared/deferred.js"; import { isNodeCommandAllowed, resolveNodeCommandAllowlist } from "./node-command-policy.js"; +import type { GatewayNodeInvokeStream } from "./server-methods/shared-types.js"; import type { GatewayContextResolver, GatewayRequestContext } from "./server-methods/types.js"; +import { getInProcessGatewayRequestContext } from "./server-plugin-in-process-dispatch.js"; export function hasInProcessGatewayContext( resolveGatewayContext?: GatewayContextResolver, @@ -9,6 +15,127 @@ export function hasInProcessGatewayContext( return Boolean(resolveGatewayContext?.() ?? scope?.resolveGatewayContext?.() ?? scope?.context); } +/** Opens one lifecycle-fenced binary channel through the canonical node invocation owner. */ +export async function openGatewayNodeDuplex(options: { + params: Parameters[0]; + invokeNode: ( + params: Parameters[0], + stream?: GatewayNodeInvokeStream, + signal?: AbortSignal, + ) => Promise; + resolveGatewayContext?: GatewayContextResolver; + runtimeLifetime?: AbortSignal; +}): ReturnType { + const { params, resolveGatewayContext, runtimeLifetime, invokeNode } = options; + const scope = getPluginRuntimeGatewayRequestScope(); + if (!scope?.pluginId?.trim()) { + throw new Error("Plugin node duplex commands require an active owning plugin identity."); + } + const registrations = scope.pluginRegistry?.nodeHostCommands.filter( + (entry) => entry.command.command === params.command, + ); + if ( + registrations?.length !== 1 || + registrations[0]?.pluginId !== scope.pluginId || + registrations[0]?.command.duplex !== true + ) { + throw new Error( + `Node command "${params.command}" must be registered exactly once by plugin "${scope.pluginId}" and declare duplex: true.`, + ); + } + const callerIdentity = scope.client?.internal?.agentRuntimeIdentity; + const context = getInProcessGatewayRequestContext(resolveGatewayContext); + if (!context?.nodeRegistry) { + throw new Error("Plugin node duplex commands require an active Gateway node registry."); + } + const controller = new AbortController(); + const signals = [controller.signal, runtimeLifetime, params.signal].filter( + (candidate): candidate is AbortSignal => candidate !== undefined, + ); + const signal = AbortSignal.any(signals); + const abortError = () => + signal.reason instanceof Error ? signal.reason : new Error("Node duplex invocation cancelled."); + if (signal.aborted) { + throw abortError(); + } + let invokeId: string | undefined; + let framedReady = false; + const ready = createDeferredCore(); + const isRuntimeCurrent = () => + !signal.aborted && + (!resolveGatewayContext || resolveGatewayContext() === context) && + (!callerIdentity || context.validateAgentRuntimeApprovalAuthority?.(callerIdentity) === true); + const assertRuntimeCurrent = () => { + if (!isRuntimeCurrent()) { + const error = signal.aborted + ? abortError() + : new Error("Plugin Gateway runtime authority is no longer current."); + controller.abort(error); + throw error; + } + }; + const endpoint = createNodeDuplexEndpoint({ + requireReady: true, + maxMessageBytes: params.maxMessageBytes, + sendFrame(frame) { + assertRuntimeCurrent(); + if (!invokeId || !framedReady) { + throw new Error("Node duplex command is not ready for binary messages."); + } + context.nodeRegistry.sendInvokeInput(invokeId, JSON.parse(frame)); + }, + onReady() { + if (!invokeId) { + throw new Error("Node duplex command announced readiness before its dispatch."); + } + framedReady = true; + ready.resolve(); + }, + onError: (error) => controller.abort(error), + }); + const onAbort = () => endpoint.close(); + signal.addEventListener("abort", onAbort, { once: true }); + const closed = invokeNode( + params, + { + onProgress: (chunk) => { + assertRuntimeCurrent(); + endpoint.receive(chunk); + }, + onDispatchReady: (id) => { + assertRuntimeCurrent(); + invokeId = id; + }, + isRuntimeCurrent, + idleTimeoutMs: NODE_DUPLEX_INVOKE_IDLE_TIMEOUT_MS, + }, + signal, + ) + .then(async (result) => { + if (!invokeId || !framedReady) { + throw new Error("Node command completed without opening a ready duplex invocation."); + } + await endpoint.drain(); + return result; + }) + .finally(() => { + signal.removeEventListener("abort", onAbort); + endpoint.close(); + controller.abort(new Error("Node duplex command has closed.")); + }); + void closed.catch(ready.reject); + await ready.promise; + return { + send: (message) => endpoint.send(message), + onMessage: (listener) => { + assertRuntimeCurrent(); + return endpoint.onMessage(listener); + }, + closed, + close: () => controller.abort(new Error("Node duplex channel closed by its caller.")), + }; +} + export function projectGatewayRuntimeNodes( nodes: unknown[], context: GatewayRequestContext | undefined, diff --git a/src/gateway/server-plugins.test.ts b/src/gateway/server-plugins.test.ts index e864c5b245cd..c56deb43088e 100644 --- a/src/gateway/server-plugins.test.ts +++ b/src/gateway/server-plugins.test.ts @@ -20,6 +20,7 @@ import type { PluginRuntimeGatewayRequestScope } from "../plugins/runtime/gatewa import type { PluginRuntime } from "../plugins/runtime/types.js"; import { withEnv } from "../test-utils/env.js"; import type { GatewayRequestContext, GatewayRequestOptions } from "./server-methods/types.js"; +import { createSyntheticPluginRuntimeClient } from "./server-plugin-runtime-client.js"; const loadOpenClawPlugins = vi.hoisted(() => vi.fn()); const loadPluginLookUpTable = vi.hoisted(() => @@ -169,6 +170,17 @@ function addLoadedPlugin( return registry; } +function createDuplexPluginRegistry(command = "image.bridge"): PluginRegistry { + const registry = addLoadedPlugin(createRegistry([]), { id: "duplex-plugin" }); + registry.nodeHostCommands.push({ + pluginId: "duplex-plugin", + pluginName: "Duplex plugin", + command: { command, duplex: true, handle: async () => "{}" }, + source: "test", + }); + return registry; +} + function createLookUpTableForTest(params: { installRecords?: PluginLookUpTable["index"]["installRecords"]; manifestRegistry?: PluginLookUpTable["manifestRegistry"]; @@ -1487,6 +1499,431 @@ describe("loadGatewayPlugins", () => { expect(getLastDispatchedClientInternal().pluginRuntimeOwnerId).toBe("third-party"); }); + test("rejects an owned non-duplex node command before invoking its handler", async () => { + const handle = vi.fn(async () => '{"ok":true}'); + const registry = addLoadedPlugin(createRegistry([]), { id: "duplex-plugin" }); + registry.nodeHostCommands.push({ + pluginId: "duplex-plugin", + pluginName: "Duplex plugin", + command: { command: "image.bridge", handle }, + source: "test", + }); + loadOpenClawPlugins.mockReturnValue(registry); + loadGatewayStartupPluginsForTest(); + serverPluginsModule.setFallbackGatewayContext({ + nodeRegistry: { sendInvokeInput: vi.fn() }, + } as unknown as GatewayRequestContext); + handleGatewayRequest.mockImplementationOnce(async (opts: HandleGatewayRequestOptions) => { + await handle(); + opts.respond(true, { ok: true }); + }); + const runtime = createRuntimeFromLastGatewayLoad(); + + const error = await gatewayRequestScopeModule.withPluginRuntimeRegistryScope(registry, () => + gatewayRequestScopeModule + .withPluginRuntimePluginScope({ pluginId: "duplex-plugin", pluginOrigin: "bundled" }, () => + runtime.nodes.openDuplex({ nodeId: "node-1", command: "image.bridge" }), + ) + .catch((reason: unknown) => reason), + ); + + expect(handle).not.toHaveBeenCalled(); + expect(handleGatewayRequest).not.toHaveBeenCalled(); + expect(error).toBeInstanceOf(Error); + expect((error as Error).message).toMatch(/declare.*duplex: true/i); + }); + + test.each([ + { label: "unknown command", owners: [] }, + { label: "another plugin's duplex command", owners: ["another-plugin"] }, + { label: "ambiguous plugin ownership", owners: ["duplex-plugin", "another-plugin"] }, + { label: "duplicate caller-owned declarations", owners: ["duplex-plugin", "duplex-plugin"] }, + { label: "missing scoped registry", owners: ["duplex-plugin"], scopedRegistry: false }, + ])("rejects a $label before node dispatch", async ({ owners, scopedRegistry }) => { + const registry = addLoadedPlugin(createRegistry([]), { id: "duplex-plugin" }); + registry.nodeHostCommands.push( + ...owners.map((pluginId) => ({ + pluginId, + pluginName: pluginId, + command: { command: "image.bridge", duplex: true, handle: vi.fn(async () => "{}") }, + source: "test", + })), + ); + loadOpenClawPlugins.mockReturnValue(registry); + loadGatewayStartupPluginsForTest(); + serverPluginsModule.setFallbackGatewayContext({ + nodeRegistry: { sendInvokeInput: vi.fn() }, + } as unknown as GatewayRequestContext); + const runtime = createRuntimeFromLastGatewayLoad(); + const openDuplex = () => + gatewayRequestScopeModule.withPluginRuntimePluginScope( + { pluginId: "duplex-plugin", pluginOrigin: "bundled" }, + () => runtime.nodes.openDuplex({ nodeId: "node-1", command: "image.bridge" }), + ); + + await expect( + scopedRegistry === false + ? openDuplex() + : gatewayRequestScopeModule.withPluginRuntimeRegistryScope(registry, openDuplex), + ).rejects.toThrow(/registered exactly once.*duplex: true/i); + expect(handleGatewayRequest).not.toHaveBeenCalled(); + }); + + test.each(["operator.read", "no scopes"])( + "does not elevate a scoped %s caller when forcing a synthetic duplex client", + async (scopeLabel) => { + const scopes = scopeLabel === "no scopes" ? [] : ["operator.read"]; + const registry = createDuplexPluginRegistry(); + loadOpenClawPlugins.mockReturnValue(registry); + loadGatewayStartupPluginsForTest(); + const context = { + nodeRegistry: { sendInvokeInput: vi.fn() }, + } as unknown as GatewayRequestContext; + serverPluginsModule.setFallbackGatewayContext(context); + handleGatewayRequest.mockImplementationOnce(async (opts: HandleGatewayRequestOptions) => { + opts.respond(false, undefined, { + code: "INVALID_REQUEST", + message: "missing operator.write scope", + }); + }); + const requestScope = { + context, + client: { connect: { scopes } } as GatewayRequestOptions["client"], + isWebchatConnect: () => false, + pluginRegistry: registry, + } satisfies PluginRuntimeGatewayRequestScope; + const runtime = createRuntimeFromLastGatewayLoad(); + + await expect( + gatewayRequestScopeModule.withPluginRuntimeGatewayRequestScope(requestScope, () => + gatewayRequestScopeModule.withPluginRuntimePluginScope( + { pluginId: "duplex-plugin", pluginOrigin: "bundled" }, + () => runtime.nodes.openDuplex({ nodeId: "node-1", command: "image.bridge" }), + ), + ), + ).rejects.toThrow("missing operator.write scope"); + expect(getLastDispatchedClientScopes()).toEqual(scopes); + expect(getLastDispatchedClientInternal().pluginRuntimeOwnerId).toBe("duplex-plugin"); + expect(getLastDispatchedParams()).not.toHaveProperty("nodeInvokeStream"); + }, + ); + + test.each([ + { callerScope: "operator.read", requestedScope: "operator.write" }, + { callerScope: "no scopes", requestedScope: "operator.write" }, + { callerScope: "operator.write", requestedScope: "operator.admin" }, + { callerScope: "operator.write", requestedScope: "operator.approvals" }, + ] as const)( + "rejects explicit $requestedScope duplex escalation from an authenticated $callerScope caller", + async ({ callerScope, requestedScope }) => { + const scopes = callerScope === "no scopes" ? [] : [callerScope]; + const callerAbort = new AbortController(); + const registry = createDuplexPluginRegistry(); + loadOpenClawPlugins.mockReturnValue(registry); + loadGatewayStartupPluginsForTest(); + const context = { + nodeRegistry: { sendInvokeInput: vi.fn() }, + } as unknown as GatewayRequestContext; + serverPluginsModule.setFallbackGatewayContext(context); + const requestScope = { + context, + client: { connect: { scopes } } as GatewayRequestOptions["client"], + isWebchatConnect: () => false, + pluginRegistry: registry, + } satisfies PluginRuntimeGatewayRequestScope; + const runtime = createRuntimeFromLastGatewayLoad(); + + await expect( + gatewayRequestScopeModule.withPluginRuntimeGatewayRequestScope(requestScope, () => + gatewayRequestScopeModule.withPluginRuntimePluginScope( + { pluginId: "duplex-plugin", pluginOrigin: "bundled" }, + () => + runtime.nodes.openDuplex({ + nodeId: "node-1", + command: "image.bridge", + scopes: [requestedScope], + signal: callerAbort.signal, + }), + ), + ), + ).rejects.toThrow("exceed the authenticated Gateway caller's authority"); + expect(handleGatewayRequest).not.toHaveBeenCalled(); + callerAbort.abort(new Error("denied invocation caller retired")); + await new Promise((resolve) => { + setImmediate(resolve); + }); + }, + ); + + test("waits for framed readiness and carries binary messages through canonical invoke transport", async () => { + const registry = createDuplexPluginRegistry(); + loadOpenClawPlugins.mockReturnValue(registry); + loadGatewayStartupPluginsForTest(); + const sendInvokeInput = vi.fn(); + const context = { + nodeRegistry: { sendInvokeInput }, + } as unknown as GatewayRequestContext; + serverPluginsModule.setFallbackGatewayContext(context); + let invokeOptions: HandleGatewayRequestOptions | undefined; + let finishInvoke: (() => void) | undefined; + handleGatewayRequest.mockImplementationOnce(async (opts: HandleGatewayRequestOptions) => { + invokeOptions = opts; + opts.client?.internal?.nodeInvokeStream?.onDispatchReady("duplex-ready-invoke"); + await new Promise((resolve) => { + finishInvoke = () => { + opts.respond(true, { ok: true, payload: { complete: true } }); + resolve(); + }; + }); + }); + const runtime = createRuntimeFromLastGatewayLoad(); + const requestScope = { + context, + client: { + connect: { scopes: ["operator.read", "operator.write"] }, + } as GatewayRequestOptions["client"], + isWebchatConnect: () => false, + pluginRegistry: registry, + } satisfies PluginRuntimeGatewayRequestScope; + let settled = false; + const opening = gatewayRequestScopeModule.withPluginRuntimeGatewayRequestScope( + requestScope, + () => + gatewayRequestScopeModule.withPluginRuntimePluginScope( + { pluginId: "duplex-plugin", pluginOrigin: "bundled" }, + () => + runtime.nodes.openDuplex({ + nodeId: "node-1", + command: "image.bridge", + scopes: ["operator.write"], + }), + ), + ); + void opening.then(() => { + settled = true; + }); + await vi.waitFor(() => expect(invokeOptions).toBeDefined()); + expect(settled).toBe(false); + expect(getLastDispatchedClientScopes()).toEqual(["operator.write"]); + + const stream = invokeOptions?.client?.internal?.nodeInvokeStream; + stream?.onProgress(JSON.stringify({ v: 1, kind: "ready" })); + const channel = await opening; + const onMessage = vi.fn(); + channel.onMessage(onMessage); + stream?.onProgress( + JSON.stringify({ v: 1, kind: "data", message: 0, index: 0, last: true, data: "BAU=" }), + ); + await channel.send(Uint8Array.of(1, 2, 3)); + + expect(onMessage).toHaveBeenCalledWith(Uint8Array.of(4, 5)); + expect(sendInvokeInput).toHaveBeenCalledWith( + "duplex-ready-invoke", + expect.objectContaining({ kind: "data", message: 0, index: 0, data: "AQID" }), + ); + expect(stream?.idleTimeoutMs).toBe(30_000); + finishInvoke?.(); + await expect(channel.closed).resolves.toEqual({ ok: true, payload: { complete: true } }); + await expect(channel.send(Uint8Array.of(1))).rejects.toThrow(/closed/i); + }); + + test.each(["listener rejection", "caller cancellation"])( + "waits for terminal asynchronous message delivery and handles %s", + async (terminalAction) => { + const registry = createDuplexPluginRegistry(); + loadOpenClawPlugins.mockReturnValue(registry); + loadGatewayStartupPluginsForTest(); + serverPluginsModule.setFallbackGatewayContext({ + nodeRegistry: { sendInvokeInput: vi.fn() }, + } as unknown as GatewayRequestContext); + let invokeOptions: HandleGatewayRequestOptions | undefined; + let finishInvoke: (() => void) | undefined; + handleGatewayRequest.mockImplementationOnce(async (opts: HandleGatewayRequestOptions) => { + invokeOptions = opts; + opts.client?.internal?.nodeInvokeStream?.onDispatchReady("duplex-terminal-delivery"); + await new Promise((resolve) => { + finishInvoke = () => { + opts.respond(true, { ok: true }); + resolve(); + }; + }); + }); + const runtime = createRuntimeFromLastGatewayLoad(); + const opening = gatewayRequestScopeModule.withPluginRuntimeRegistryScope(registry, () => + gatewayRequestScopeModule.withPluginRuntimePluginScope( + { pluginId: "duplex-plugin", pluginOrigin: "bundled" }, + () => runtime.nodes.openDuplex({ nodeId: "node-1", command: "image.bridge" }), + ), + ); + await vi.waitFor(() => expect(invokeOptions).toBeDefined()); + const stream = invokeOptions?.client?.internal?.nodeInvokeStream; + stream?.onProgress(JSON.stringify({ v: 1, kind: "ready" })); + const channel = await opening; + let rejectDelivery: ((error: Error) => void) | undefined; + channel.onMessage( + async () => + await new Promise((_resolve, reject) => { + rejectDelivery = reject; + }), + ); + + stream?.onProgress( + JSON.stringify({ v: 1, kind: "data", message: 0, index: 0, last: true, data: "AQ==" }), + ); + finishInvoke?.(); + let closedSettled = false; + void channel.closed.then( + () => { + closedSettled = true; + }, + () => { + closedSettled = true; + }, + ); + await new Promise((resolve) => { + setImmediate(resolve); + }); + expect(closedSettled).toBe(false); + + if (terminalAction === "listener rejection") { + rejectDelivery?.(new Error("terminal message listener rejected")); + await expect(channel.closed).rejects.toThrow("terminal message listener rejected"); + } else { + channel.close(); + await expect(channel.closed).rejects.toThrow(/closed|cancel/i); + } + }, + ); + + test("cancels a retained duplex invocation when its delegated caller authority closes", async () => { + const registry = createDuplexPluginRegistry(); + loadOpenClawPlugins.mockReturnValue(registry); + loadGatewayStartupPluginsForTest(); + const sendInvokeInput = vi.fn(); + const validateAgentRuntimeApprovalAuthority = vi.fn(() => true); + const context = { + nodeRegistry: { sendInvokeInput }, + validateAgentRuntimeApprovalAuthority, + } as unknown as GatewayRequestContext; + serverPluginsModule.setFallbackGatewayContext(context); + let invokeSignal: AbortSignal | undefined; + handleGatewayRequest.mockImplementationOnce(async (opts: HandleGatewayRequestOptions) => { + invokeSignal = opts.signal; + opts.client?.internal?.nodeInvokeStream?.onDispatchReady("delegated-duplex"); + opts.client?.internal?.nodeInvokeStream?.onProgress(JSON.stringify({ v: 1, kind: "ready" })); + await new Promise((resolve) => { + opts.signal?.addEventListener("abort", () => resolve(), { once: true }); + }); + }); + const operationalRunInstance = { + instanceId: "delegated-instance", + runId: "delegated-run", + }; + const client = createSyntheticPluginRuntimeClient({ scopes: ["operator.write"] }); + client.internal = { + ...client.internal, + agentRuntimeIdentity: { + kind: "agentRuntime", + agentId: "main", + sessionKey: "agent:main:delegated", + operationalRunInstance, + delegatedAuthority: { + kind: "local", + lifecycleGeneration: "delegated-generation", + claimId: "delegated-claim", + operationalRunInstance, + }, + }, + }; + const requestScope = { + context, + client, + isWebchatConnect: () => false, + pluginRegistry: registry, + } satisfies PluginRuntimeGatewayRequestScope; + const runtime = createRuntimeFromLastGatewayLoad(); + const channel = await gatewayRequestScopeModule.withPluginRuntimeGatewayRequestScope( + requestScope, + () => + gatewayRequestScopeModule.withPluginRuntimePluginScope( + { pluginId: "duplex-plugin", pluginOrigin: "bundled" }, + () => runtime.nodes.openDuplex({ nodeId: "node-1", command: "image.bridge" }), + ), + ); + + validateAgentRuntimeApprovalAuthority.mockReturnValue(false); + + await expect(channel.send(Uint8Array.of(1))).rejects.toThrow(/authority.*no longer current/i); + expect(invokeSignal?.aborted).toBe(true); + expect(sendInvokeInput).not.toHaveBeenCalled(); + await expect(channel.closed).rejects.toThrow(/authority.*no longer current/i); + expect(() => channel.onMessage(vi.fn())).toThrow(/authority.*no longer current/i); + }); + + test("cancels an open node duplex invocation before retiring its plugin runtime", async () => { + const registry = createDuplexPluginRegistry("plugin.duplex.v1"); + loadOpenClawPlugins.mockReturnValue(registry); + const context = { + nodeRegistry: { sendInvokeInput: vi.fn() }, + } as unknown as GatewayRequestContext; + serverPluginsModule.setFallbackGatewayContext(context); + const loaded = serverPluginsModule.loadGatewayPlugins({ + cfg: {}, + workspaceDir: "/tmp", + log: createTestLog(), + coreGatewayHandlers: {}, + baseMethods: [], + pluginIds: ["duplex-plugin"], + resolveGatewayContext: () => resolveTestGatewayContext(), + }); + runtimeRegistryModule.setActivePluginRegistry(loaded.pluginRegistry); + + let invokeSignal: AbortSignal | undefined; + handleGatewayRequest.mockImplementationOnce(async (opts: HandleGatewayRequestOptions) => { + invokeSignal = opts.signal; + const stream = opts.client?.internal as + | { + nodeInvokeStream?: { + onDispatchReady: (invokeId: string) => void; + onProgress: (chunk: string) => void; + }; + } + | undefined; + stream?.nodeInvokeStream?.onDispatchReady("duplex-retire-invoke"); + stream?.nodeInvokeStream?.onProgress(JSON.stringify({ v: 1, kind: "ready" })); + await new Promise((resolve) => { + opts.signal?.addEventListener( + "abort", + () => { + opts.respond(false, undefined, { code: "ABORTED", message: "node invoke cancelled" }); + resolve(); + }, + { once: true }, + ); + }); + }); + + const runtime = createRuntimeFromLastGatewayLoad(); + const nodes = runtime.nodes as PluginRuntime["nodes"] & { + openDuplex: (params: { nodeId: string; command: string }) => Promise<{ + closed: Promise; + send: (message: Uint8Array) => Promise; + }>; + }; + const channel = await gatewayRequestScopeModule.withPluginRuntimeRegistryScope(registry, () => + gatewayRequestScopeModule.withPluginRuntimePluginScope( + { pluginId: "duplex-plugin", pluginOrigin: "bundled" }, + () => nodes.openDuplex({ nodeId: "node-1", command: "plugin.duplex.v1" }), + ), + ); + + loaded.retireGatewayRuntimeBindings(); + + expect(invokeSignal?.aborted).toBe(true); + await expect(channel.closed).rejects.toThrow(/retired|cancel/i); + await expect(channel.send(Uint8Array.of(1))).rejects.toThrow(/retired|closed/i); + }); + test("forwards provider and model overrides when the request scope is authorized", async () => { const serverPlugins = serverPluginsModule; const runtime = await createSubagentRuntime(serverPlugins); diff --git a/src/gateway/server-plugins.ts b/src/gateway/server-plugins.ts index 523483f1db85..5438c6b1455d 100644 --- a/src/gateway/server-plugins.ts +++ b/src/gateway/server-plugins.ts @@ -31,8 +31,9 @@ import type { RuntimeGatewayRequestOptions, } from "../plugins/runtime/types.js"; import type { PluginLogger, PluginOrigin } from "../plugins/types.js"; -import { ADMIN_SCOPE } from "./method-scopes.js"; +import { ADMIN_SCOPE, authorizeOperatorScopesForRequiredScope } from "./method-scopes.js"; import { normalizeOperatorScopeList, type OperatorScope } from "./operator-scopes.js"; +import type { GatewayNodeInvokeStream } from "./server-methods/shared-types.js"; import type { GatewayContextResolver, GatewayRequestHandler, @@ -51,6 +52,7 @@ import { } from "./server-plugin-subagent-runtime.js"; import { hasInProcessGatewayContext, + openGatewayNodeDuplex, projectGatewayRuntimeNodes, } from "./server-plugins-node-runtime.js"; @@ -391,7 +393,57 @@ type GatewayRuntimeNodes = Awaited>[" export function createGatewayNodesRuntime( resolveGatewayContext?: GatewayContextResolver, + runtimeLifetime?: AbortSignal, ): PluginRuntime["nodes"] { + const invokeNode = async ( + params: Parameters[0], + stream?: GatewayNodeInvokeStream, + signal = params.signal, + ) => { + const scope = getPluginRuntimeGatewayRequestScope(); + const pluginId = scope?.pluginId?.trim() || undefined; + const requestedScopes = resolveRuntimeNodeInvokeSyntheticScopes({ + pluginId, + pluginOrigin: scope?.pluginOrigin, + pluginTrustedOfficialInstall: scope?.pluginTrustedOfficialInstall, + requestedScopes: normalizeOperatorScopeList(params.scopes), + }); + const callerScopes = + stream && scope?.client + ? (normalizeOperatorScopeList(scope.client.connect.scopes) ?? []) + : undefined; + if ( + callerScopes && + requestedScopes?.some( + (requestedScope) => + !authorizeOperatorScopesForRequiredScope(requestedScope, callerScopes).allowed, + ) + ) { + throw new Error("Requested node scopes exceed the authenticated Gateway caller's authority."); + } + // Forced synthetic stream clients must retain their authenticated caller's exact scopes. + const syntheticScopes = requestedScopes ?? callerScopes; + return dispatchGatewayMethodInProcess( + "node.invoke", + { + nodeId: params.nodeId, + command: params.command, + ...(params.params !== undefined && { params: params.params }), + timeoutMs: params.timeoutMs, + idempotencyKey: params.idempotencyKey || randomUUID(), + ...(params.sessionKey ? { sessionKey: params.sessionKey } : {}), + }, + { + ...(pluginId ? { pluginRuntimeOwnerId: pluginId } : {}), + ...(syntheticScopes ? { syntheticScopes } : {}), + ...(stream || syntheticScopes ? { forceSyntheticClient: true } : {}), + ...(stream ? { nodeInvokeStream: stream } : {}), + ...(signal ? { signal } : {}), + resolveGatewayContext, + }, + ); + }; + return { async list(params) { const context = getInProcessGatewayRequestContext(resolveGatewayContext); @@ -415,36 +467,9 @@ export function createGatewayNodesRuntime( nodes: projectGatewayRuntimeNodes(filteredNodes, context) as GatewayRuntimeNodes, }; }, - async invoke(params) { - const scope = getPluginRuntimeGatewayRequestScope(); - const pluginId = - typeof scope?.pluginId === "string" && scope.pluginId.trim() - ? scope.pluginId.trim() - : undefined; - const syntheticScopes = resolveRuntimeNodeInvokeSyntheticScopes({ - pluginId, - pluginOrigin: scope?.pluginOrigin, - pluginTrustedOfficialInstall: scope?.pluginTrustedOfficialInstall, - requestedScopes: normalizeOperatorScopeList(params.scopes), - }); - return await dispatchGatewayMethodInProcess( - "node.invoke", - { - nodeId: params.nodeId, - command: params.command, - ...(params.params !== undefined && { params: params.params }), - timeoutMs: params.timeoutMs, - idempotencyKey: params.idempotencyKey || randomUUID(), - ...(params.sessionKey ? { sessionKey: params.sessionKey } : {}), - }, - { - ...(pluginId ? { pluginRuntimeOwnerId: pluginId } : {}), - ...(syntheticScopes ? { forceSyntheticClient: true, syntheticScopes } : {}), - ...(params.signal ? { signal: params.signal } : {}), - resolveGatewayContext, - }, - ); - }, + invoke: invokeNode, + openDuplex: (params) => + openGatewayNodeDuplex({ params, invokeNode, resolveGatewayContext, runtimeLifetime }), }; } @@ -457,9 +482,11 @@ function createGatewayPluginRuntimeBindings( retire: () => void; } { let active = true; + const lifetime = new AbortController(); const resolveBoundGatewayContext = () => (active ? resolveGatewayContext() : undefined); return { retire: () => { + lifetime.abort(new Error("Plugin Gateway runtime retired; duplex invocation cancelled.")); active = false; }, runtime: { @@ -483,7 +510,7 @@ function createGatewayPluginRuntimeBindings( request: (method, params, options) => dispatchTrustedPluginGatewayMethod(method, params, options, resolveBoundGatewayContext), }, - nodes: createGatewayNodesRuntime(resolveBoundGatewayContext), + nodes: createGatewayNodesRuntime(resolveBoundGatewayContext, lifetime.signal), subagent: createGatewaySubagentRuntime(resolveBoundGatewayContext, overridePolicies), }, }; diff --git a/src/infra/node-duplex-framing.test.ts b/src/infra/node-duplex-framing.test.ts new file mode 100644 index 000000000000..03b5d219e24e --- /dev/null +++ b/src/infra/node-duplex-framing.test.ts @@ -0,0 +1,570 @@ +import { describe, expect, it, vi } from "vitest"; +import { createNodeDuplexEndpoint } from "./node-duplex-framing.js"; + +const FRAGMENT_BYTES = 8 * 1024; +const MAX_MESSAGE_BYTES = 100 * 1024 * 1024; + +function dataFrame(overrides: Record = {}): string { + return JSON.stringify({ + v: 1, + kind: "data", + message: 0, + index: 0, + last: true, + data: Buffer.from("message").toString("base64"), + ...overrides, + }); +} + +describe("node duplex message framing", () => { + it("transfers binary messages larger than transport frames in both directions", async () => { + const outboundFrames: string[] = []; + const inboundFrames: string[] = []; + const leftMessages: Uint8Array[] = []; + const rightMessages: Uint8Array[] = []; + const left = createNodeDuplexEndpoint({ + sendFrame(frame) { + outboundFrames.push(frame); + right.receive(frame); + }, + }); + const right = createNodeDuplexEndpoint({ + sendFrame(frame) { + inboundFrames.push(frame); + left.receive(frame); + }, + }); + left.onMessage((message) => { + leftMessages.push(message); + }); + right.onMessage((message) => { + rightMessages.push(message); + }); + + const outbound = Uint8Array.from({ length: 40_000 }, (_, index) => index % 251); + const inbound = Uint8Array.from({ length: 25_000 }, (_, index) => 255 - (index % 251)); + await left.send(outbound); + await right.send(inbound); + + expect(rightMessages).toEqual([outbound]); + expect(leftMessages).toEqual([inbound]); + expect(outboundFrames.length).toBeGreaterThan(2); + expect(inboundFrames.length).toBeGreaterThan(2); + expect([...outboundFrames, ...inboundFrames]).toSatisfy((frames: string[]) => + frames.every((frame) => Buffer.byteLength(frame, "utf8") < 16 * 1024), + ); + }); + + it("preserves complete message boundaries across concurrent asynchronous sends", async () => { + const received: Uint8Array[] = []; + const first = Uint8Array.from({ length: 20_000 }, () => 1); + const second = Uint8Array.from({ length: 18_000 }, () => 2); + const receiver = createNodeDuplexEndpoint({ sendFrame: () => {} }); + receiver.onMessage((message) => { + received.push(message); + }); + const sender = createNodeDuplexEndpoint({ + async sendFrame(frame) { + await Promise.resolve(); + receiver.receive(frame); + }, + }); + + await Promise.all([sender.send(first), sender.send(second)]); + + expect(received).toEqual([first, second]); + }); + + it("serializes framed readiness ahead of a concurrent message", async () => { + const events: string[] = []; + const receiver = createNodeDuplexEndpoint({ + sendFrame: () => {}, + onReady: () => events.push("ready"), + }); + receiver.onMessage(() => { + events.push("message"); + }); + const sender = createNodeDuplexEndpoint({ + async sendFrame(frame) { + await Promise.resolve(); + receiver.receive(frame); + }, + }); + + await Promise.all([sender.sendReady(), sender.send(Uint8Array.of(1, 2))]); + + expect(events).toEqual(["ready", "message"]); + }); + + it("rejects data before required readiness while node-host input needs no reciprocal ready", () => { + const gatewayError = vi.fn(); + const gateway = createNodeDuplexEndpoint({ + sendFrame: () => {}, + onError: gatewayError, + requireReady: true, + }); + + expect(() => gateway.receive(dataFrame())).toThrow(/before framed readiness/i); + expect(gatewayError).toHaveBeenCalledOnce(); + + const hostMessage = vi.fn(); + const host = createNodeDuplexEndpoint({ sendFrame: () => {} }); + host.onMessage(hostMessage); + host.receive(dataFrame()); + expect(hostMessage).toHaveBeenCalledOnce(); + }); + + it("preserves empty binary messages as distinct complete messages", async () => { + const received: Uint8Array[] = []; + const receiver = createNodeDuplexEndpoint({ sendFrame: () => {} }); + receiver.onMessage((message) => { + received.push(message); + }); + const sender = createNodeDuplexEndpoint({ sendFrame: (frame) => receiver.receive(frame) }); + + await sender.send(new Uint8Array()); + await sender.send(Uint8Array.of(7)); + + expect(received).toEqual([new Uint8Array(), Uint8Array.of(7)]); + }); + + it.each([ + ["malformed JSON", "{"], + ["wrong version", dataFrame({ v: 2 })], + ["unknown kind", dataFrame({ kind: "unknown" })], + ["extra field", dataFrame({ extra: true })], + ["noncanonical base64", dataFrame({ data: "bWVzc2FnZQ" })], + ["invalid base64", dataFrame({ data: "%%%%" })], + ["negative message id", dataFrame({ message: -1 })], + ["unsafe message id", dataFrame({ message: Number.MAX_SAFE_INTEGER + 1 })], + ["message gap", dataFrame({ message: 1 })], + ["fragment gap", dataFrame({ index: 1 })], + ["negative fragment index", dataFrame({ index: -1 })], + ["unsafe fragment index", dataFrame({ index: Number.MAX_SAFE_INTEGER + 1 })], + ["undersized nonterminal fragment", dataFrame({ last: false })], + ["mixed ready fields", JSON.stringify({ v: 1, kind: "ready", data: "" })], + [ + "oversized fragment", + dataFrame({ data: Buffer.alloc(FRAGMENT_BYTES + 1).toString("base64") }), + ], + ["oversized wire frame", `{"data":"${"a".repeat(16 * 1024)}"}`], + ])("fails closed on %s", (_reason, frame) => { + const onError = vi.fn(); + const received = vi.fn(); + const endpoint = createNodeDuplexEndpoint({ sendFrame: () => {}, onError }); + endpoint.onMessage(received); + + expect(() => endpoint.receive(frame)).toThrow(); + expect(onError).toHaveBeenCalledOnce(); + expect(received).not.toHaveBeenCalled(); + expect(() => endpoint.receive(dataFrame())).toThrow(/closed/i); + expect(onError).toHaveBeenCalledOnce(); + }); + + it.each([ + [ + "duplicate fragment", + [ + dataFrame({ + last: false, + data: Buffer.alloc(FRAGMENT_BYTES).toString("base64"), + }), + dataFrame({ index: 0 }), + ], + ], + [ + "fragment gap", + [ + dataFrame({ + last: false, + data: Buffer.alloc(FRAGMENT_BYTES).toString("base64"), + }), + dataFrame({ index: 2 }), + ], + ], + [ + "interleaved message", + [ + dataFrame({ + last: false, + data: Buffer.alloc(FRAGMENT_BYTES).toString("base64"), + }), + dataFrame({ message: 1, index: 1 }), + ], + ], + ["duplicate completed message", [dataFrame(), dataFrame()]], + [ + "duplicate readiness", + [JSON.stringify({ v: 1, kind: "ready" }), JSON.stringify({ v: 1, kind: "ready" })], + ], + ["late readiness", [dataFrame(), JSON.stringify({ v: 1, kind: "ready" })]], + ])("rejects %s without delivering subsequent messages", (_reason, frames) => { + const onError = vi.fn(); + const endpoint = createNodeDuplexEndpoint({ sendFrame: () => {}, onError }); + endpoint.onMessage(() => {}); + endpoint.receive(frames[0]!); + + expect(() => endpoint.receive(frames[1]!)).toThrow(); + expect(onError).toHaveBeenCalledOnce(); + }); + + it("bounds pending message count and bytes before a listener subscribes", async () => { + const countError = vi.fn(); + const countBounded = createNodeDuplexEndpoint({ sendFrame: () => {}, onError: countError }); + for (let message = 0; message < 8; message += 1) { + countBounded.receive(dataFrame({ message })); + } + expect(() => countBounded.receive(dataFrame({ message: 8 }))).toThrow(/pending/i); + expect(countError).toHaveBeenCalledOnce(); + + const bytesError = vi.fn(); + const bytesBounded = createNodeDuplexEndpoint({ sendFrame: () => {}, onError: bytesError }); + const sender = createNodeDuplexEndpoint({ + sendFrame: (frame) => bytesBounded.receive(frame), + }); + await sender.send(new Uint8Array(600_000)); + await expect(sender.send(new Uint8Array(600_000))).rejects.toThrow(/pending/i); + expect(bytesError).toHaveBeenCalledOnce(); + }); + + it("bounds incomplete fragments against bytes already buffered before listener registration", () => { + const onError = vi.fn(); + const endpoint = createNodeDuplexEndpoint({ sendFrame: () => {}, onError }); + endpoint.receive(dataFrame({ data: "eA==" })); + const fragment = Buffer.alloc(FRAGMENT_BYTES).toString("base64"); + for (let index = 0; index < 127; index += 1) { + endpoint.receive(dataFrame({ message: 1, index, last: false, data: fragment })); + } + + expect(() => + endpoint.receive(dataFrame({ message: 1, index: 127, last: false, data: fragment })), + ).toThrow(/pending/i); + expect(onError).toHaveBeenCalledOnce(); + expect(() => endpoint.receive(dataFrame({ message: 1, index: 128 }))).toThrow(/closed/i); + }); + + it("accepts logical messages above the pending-byte limit after listener registration", async () => { + const received = vi.fn(); + const receiver = createNodeDuplexEndpoint({ sendFrame: () => {} }); + receiver.onMessage(received); + const sender = createNodeDuplexEndpoint({ sendFrame: (frame) => receiver.receive(frame) }); + const message = new Uint8Array(1024 * 1024 + 1); + + await sender.send(message); + + expect(received).toHaveBeenCalledExactlyOnceWith(message); + }); + + it("delivers buffered whole messages in order when the listener subscribes", () => { + const endpoint = createNodeDuplexEndpoint({ sendFrame: () => {} }); + endpoint.receive(dataFrame()); + endpoint.receive(dataFrame({ message: 1, data: Buffer.from("second").toString("base64") })); + const received: string[] = []; + + endpoint.onMessage((message) => { + received.push(Buffer.from(message).toString()); + }); + + expect(received).toEqual(["message", "second"]); + }); + + it("rejects oversized outbound and inbound logical messages", async () => { + const sendFrame = vi.fn(); + const outboundError = vi.fn(); + const sender = createNodeDuplexEndpoint({ + sendFrame, + onError: outboundError, + maxMessageBytes: 5, + }); + + await expect(sender.send(Uint8Array.of(1, 2, 3, 4, 5, 6))).rejects.toThrow(/maximum/i); + expect(sendFrame).not.toHaveBeenCalled(); + expect(outboundError).toHaveBeenCalledOnce(); + + const inboundError = vi.fn(); + const receiver = createNodeDuplexEndpoint({ + sendFrame: () => {}, + onError: inboundError, + maxMessageBytes: 5, + }); + expect(() => receiver.receive(dataFrame())).toThrow(/maximum/i); + expect(inboundError).toHaveBeenCalledOnce(); + }); + + it.each([0, -1, 1.5, Number.NaN, MAX_MESSAGE_BYTES + 1])( + "rejects an unsafe logical message limit of %s bytes", + (maxMessageBytes) => { + expect(() => createNodeDuplexEndpoint({ sendFrame: () => {}, maxMessageBytes })).toThrow( + /maximum/i, + ); + }, + ); + + it("rejects accumulated message overflow and excessive fragment counts", () => { + const fragment = Buffer.alloc(FRAGMENT_BYTES).toString("base64"); + const overflow = createNodeDuplexEndpoint({ + sendFrame: () => {}, + maxMessageBytes: FRAGMENT_BYTES + 1, + }); + overflow.receive(dataFrame({ last: false, data: fragment })); + expect(() => + overflow.receive(dataFrame({ index: 1, data: Buffer.from("xx").toString("base64") })), + ).toThrow(/maximum/i); + + const excessive = createNodeDuplexEndpoint({ + sendFrame: () => {}, + maxMessageBytes: FRAGMENT_BYTES, + }); + excessive.receive(dataFrame({ last: false, data: fragment })); + expect(() => excessive.receive(dataFrame({ index: 1, data: "" }))).toThrow(/fragment/i); + }); + + it("ignores empty heartbeat frames without disturbing message ordering", () => { + const received = vi.fn(); + const endpoint = createNodeDuplexEndpoint({ sendFrame: () => {} }); + endpoint.onMessage(received); + + endpoint.receive(""); + endpoint.receive(dataFrame()); + endpoint.receive(""); + endpoint.receive(dataFrame({ message: 1 })); + + expect(received).toHaveBeenCalledTimes(2); + }); + + it("closes when a message listener throws and never invokes it afterward", () => { + const failure = new Error("listener exploded"); + const onError = vi.fn(); + const listener = vi.fn(() => { + throw failure; + }); + const endpoint = createNodeDuplexEndpoint({ sendFrame: () => {}, onError }); + endpoint.onMessage(listener); + + expect(() => endpoint.receive(dataFrame())).toThrow(failure); + expect(onError).toHaveBeenCalledExactlyOnceWith(failure); + expect(() => endpoint.receive(dataFrame({ message: 1 }))).toThrow(/closed/i); + expect(listener).toHaveBeenCalledOnce(); + }); + + it.each(["immediate", "buffered"] as const)( + "closes after an asynchronous %s message listener rejects", + async (delivery) => { + const failure = new Error("asynchronous listener exploded"); + const onError = vi.fn(); + const listener = vi.fn(() => { + const rejection = Promise.reject(failure); + void rejection.catch(() => {}); + return rejection; + }); + const endpoint = createNodeDuplexEndpoint({ sendFrame: () => {}, onError }); + if (delivery === "buffered") { + endpoint.receive(dataFrame()); + } + endpoint.onMessage(listener); + if (delivery === "immediate") { + endpoint.receive(dataFrame()); + } + + await vi.waitFor(() => expect(onError).toHaveBeenCalledExactlyOnceWith(failure)); + expect(() => endpoint.receive(dataFrame({ message: 1 }))).toThrow(/closed/i); + expect(listener).toHaveBeenCalledOnce(); + }, + ); + + it("bounds outstanding asynchronous listener deliveries before invoking another callback", () => { + const onError = vi.fn(); + const neverSettles = new Promise(() => {}); + const listener = vi.fn(() => neverSettles); + const endpoint = createNodeDuplexEndpoint({ sendFrame: () => {}, onError }); + endpoint.onMessage(listener); + for (let message = 0; message < 8; message += 1) { + endpoint.receive(dataFrame({ message })); + } + + expect(() => endpoint.receive(dataFrame({ message: 8 }))).toThrow(/pending|in.flight/i); + expect(listener).toHaveBeenCalledTimes(8); + expect(onError).toHaveBeenCalledOnce(); + }); + + it("bounds combined bytes held by outstanding asynchronous listener deliveries", () => { + const onError = vi.fn(); + const listener = vi.fn(() => new Promise(() => {})); + const endpoint = createNodeDuplexEndpoint({ + sendFrame: () => {}, + onError, + maxMessageBytes: 16, + }); + endpoint.onMessage(listener); + endpoint.receive(dataFrame({ data: Buffer.alloc(10).toString("base64") })); + + expect(() => + endpoint.receive(dataFrame({ message: 1, data: Buffer.alloc(7).toString("base64") })), + ).toThrow(/pending|in.flight/i); + expect(listener).toHaveBeenCalledOnce(); + expect(onError).toHaveBeenCalledOnce(); + }); + + it.each(["immediate", "buffered"] as const)( + "drains an asynchronous %s listener before allowing invocation completion", + async (delivery) => { + let finishListener: (() => void) | undefined; + const listenerFinished = new Promise((resolve) => { + finishListener = resolve; + }); + const endpoint = createNodeDuplexEndpoint({ sendFrame: () => {} }); + if (delivery === "buffered") { + endpoint.receive(dataFrame()); + } + endpoint.onMessage(() => listenerFinished); + if (delivery === "immediate") { + endpoint.receive(dataFrame()); + } + let drained = false; + const drain = endpoint.drain().then(() => { + drained = true; + }); + await Promise.resolve(); + expect(drained).toBe(false); + + finishListener?.(); + await drain; + + expect(drained).toBe(true); + }, + ); + + it("continues draining listener work that arrives while an earlier delivery is pending", async () => { + const finishListeners: Array<() => void> = []; + const endpoint = createNodeDuplexEndpoint({ sendFrame: () => {} }); + endpoint.onMessage( + () => + new Promise((resolve) => { + finishListeners.push(resolve); + }), + ); + endpoint.receive(dataFrame()); + let drained = false; + const drain = endpoint.drain().then(() => { + drained = true; + }); + endpoint.receive(dataFrame({ message: 1 })); + + finishListeners[0]?.(); + await new Promise((resolve) => { + setImmediate(resolve); + }); + expect(drained).toBe(false); + + finishListeners[1]?.(); + await drain; + expect(drained).toBe(true); + }); + + it("preserves the original asynchronous listener failure while draining", async () => { + let rejectListener: ((error: Error) => void) | undefined; + const listenerFinished = new Promise((_resolve, reject) => { + rejectListener = reject; + }); + const failure = new Error("asynchronous drain listener exploded"); + const onError = vi.fn(); + const endpoint = createNodeDuplexEndpoint({ sendFrame: () => {}, onError }); + endpoint.onMessage(() => listenerFinished); + endpoint.receive(dataFrame()); + const drain = endpoint.drain(); + + rejectListener?.(failure); + + await expect(drain).rejects.toBe(failure); + expect(onError).toHaveBeenCalledExactlyOnceWith(failure); + }); + + it("rejects drain immediately when closing with a listener that never settles", async () => { + const listenerFinished = new Promise(() => {}); + const endpoint = createNodeDuplexEndpoint({ sendFrame: () => {} }); + endpoint.onMessage(() => listenerFinished); + endpoint.receive(dataFrame()); + const drain = endpoint.drain(); + + endpoint.close(); + + await expect(drain).rejects.toThrow(/closed/i); + }); + + it("closes and reports asynchronous frame transport failure exactly once", async () => { + const failure = new Error("node transport disconnected"); + const onError = vi.fn(); + const endpoint = createNodeDuplexEndpoint({ + async sendFrame() { + throw failure; + }, + onError, + }); + + await expect(endpoint.send(Uint8Array.of(1))).rejects.toThrow(failure); + await expect(endpoint.send(Uint8Array.of(2))).rejects.toThrow(/closed/i); + expect(onError).toHaveBeenCalledExactlyOnceWith(failure); + }); + + it.each(["message", "ready"] as const)( + "rejects %s when the endpoint closes during its final transport await", + async (operation) => { + let releaseTransport: (() => void) | undefined; + const transportReleased = new Promise((resolve) => { + releaseTransport = resolve; + }); + const endpoint = createNodeDuplexEndpoint({ + async sendFrame() { + await transportReleased; + }, + }); + const pending = + operation === "message" ? endpoint.send(Uint8Array.of(1)) : endpoint.sendReady(); + await Promise.resolve(); + + endpoint.close(); + releaseTransport?.(); + + await expect(pending).rejects.toThrow(/closed/i); + }, + ); + + it("closes when the framed-ready callback rejects unexpected readiness", () => { + const failure = new Error("node readiness preceded dispatch"); + const onError = vi.fn(); + const endpoint = createNodeDuplexEndpoint({ + sendFrame: () => {}, + onReady() { + throw failure; + }, + onError, + }); + + expect(() => endpoint.receive(JSON.stringify({ v: 1, kind: "ready" }))).toThrow(failure); + expect(onError).toHaveBeenCalledExactlyOnceWith(failure); + }); + + it("rejects a second active listener and subscriptions after closure", () => { + const endpoint = createNodeDuplexEndpoint({ sendFrame: () => {} }); + const unsubscribe = endpoint.onMessage(() => {}); + + expect(() => endpoint.onMessage(() => {})).toThrow(/listener/i); + unsubscribe(); + endpoint.onMessage(() => {}); + endpoint.close(); + expect(() => endpoint.onMessage(() => {})).toThrow(/closed/i); + }); + + it("rejects retained send and incoming data after an idempotent close", async () => { + const listener = vi.fn(); + const endpoint = createNodeDuplexEndpoint({ sendFrame: () => {} }); + endpoint.onMessage(listener); + + endpoint.close(); + endpoint.close(); + + await expect(endpoint.send(Uint8Array.of(1))).rejects.toThrow(/closed/i); + expect(() => endpoint.receive(dataFrame())).toThrow(/closed/i); + expect(listener).not.toHaveBeenCalled(); + }); +}); diff --git a/src/infra/node-duplex-framing.ts b/src/infra/node-duplex-framing.ts new file mode 100644 index 000000000000..5b5aa3380c5a --- /dev/null +++ b/src/infra/node-duplex-framing.ts @@ -0,0 +1,241 @@ +import { isRecord } from "@openclaw/normalization-core/record-coerce"; +import { createDeferredCore } from "../shared/deferred.js"; + +const NODE_DUPLEX_FRAGMENT_BYTES = 8 * 1024; +const NODE_DUPLEX_MAX_MESSAGE_BYTES = 100 * 1024 * 1024; + +const MAX_PENDING_MESSAGES = 8; +const MAX_PENDING_BYTES = 1024 * 1024; + +/** Owns ordered, bounded binary messages carried by existing node-invoke string frames. */ +export function createNodeDuplexEndpoint(options: { + sendFrame: (frame: string) => Promise | void; + onReady?: () => void; + onError?: (error: Error) => void; + requireReady?: boolean; + maxMessageBytes?: number; +}) { + const maxMessageBytes = options.maxMessageBytes ?? NODE_DUPLEX_MAX_MESSAGE_BYTES; + const invalidMessageLimit = !Number.isSafeInteger(maxMessageBytes) || maxMessageBytes < 1; + if (invalidMessageLimit || maxMessageBytes > NODE_DUPLEX_MAX_MESSAGE_BYTES) { + throw new Error("node duplex maximum message bytes must be between 1 and 100 MiB"); + } + let closed = false; + const ready = { sent: false, received: false }; + let nextOutgoingMessage = 0; + const incoming = { message: 0, fragment: 0 }; + let activeDeliveryBytes = 0; + let listener: ((message: Uint8Array) => void | Promise) | undefined; + let sendQueue = Promise.resolve(); + const drainClosed = createDeferredCore(); + void drainClosed.promise.catch(() => {}); + const incomingFragments: Uint8Array[] = []; + const pendingMessages: Uint8Array[] = []; + const activeDeliveries = new Set>(); + + const assertOpen = () => { + if (closed) { + throw new Error("node duplex channel is closed"); + } + }; + + const close = (reason = new Error("node duplex channel is closed")) => { + closed = true; + drainClosed.reject(reason); + listener = undefined; + incomingFragments.length = 0; + pendingMessages.length = 0; + activeDeliveries.clear(); + }; + + const fail = (cause: unknown): Error => { + const error = cause instanceof Error ? cause : new Error(String(cause)); + if (!closed) { + close(error); + options.onError?.(error); + } + return error; + }; + + const enqueue = (task: () => Promise): Promise => { + const operation = sendQueue.then(task); + // Keep rejected sends observed without allowing concurrent messages to interleave. + sendQueue = operation.catch(() => {}); + return operation.catch((error: unknown) => { + throw fail(error); + }); + }; + + const observeListener = (callback: NonNullable, message: Uint8Array) => { + const bytesExceeded = activeDeliveryBytes + message.byteLength > maxMessageBytes; + if (activeDeliveries.size >= MAX_PENDING_MESSAGES || bytesExceeded) { + throw new Error("node duplex pending listener delivery exceeded its bounded capacity"); + } + const result = callback(message); + if (!result) { + return; + } + activeDeliveryBytes += message.byteLength; + const delivery = Promise.resolve(result).finally(() => { + activeDeliveries.delete(delivery); + activeDeliveryBytes -= message.byteLength; + }); + activeDeliveries.add(delivery); + void delivery.catch(fail); + }; + + const acceptData = (frame: Record) => { + if ( + Object.keys(frame).length !== 6 || + !Number.isSafeInteger(frame.message) || + !Number.isSafeInteger(frame.index) || + typeof frame.last !== "boolean" || + typeof frame.data !== "string" + ) { + throw new Error("node duplex data frame has an invalid closed shape"); + } + if (frame.message !== incoming.message || frame.index !== incoming.fragment) { + throw new Error("node duplex message or fragment arrived out of order"); + } + const fragment = Buffer.from(frame.data, "base64"); + if ( + fragment.toString("base64") !== frame.data || + fragment.byteLength > NODE_DUPLEX_FRAGMENT_BYTES || + (!frame.last && fragment.byteLength !== NODE_DUPLEX_FRAGMENT_BYTES) || + (frame.last && fragment.byteLength === 0 && incoming.fragment > 0) + ) { + throw new Error("node duplex fragment has invalid canonical base64 or bounded size"); + } + const incomingBytes = incoming.fragment * NODE_DUPLEX_FRAGMENT_BYTES; + if (incomingBytes + fragment.byteLength > maxMessageBytes) { + throw new Error("node duplex logical message exceeds its maximum size"); + } + const pendingBytes = pendingMessages.reduce((total, message) => total + message.byteLength, 0); + if (!listener && pendingBytes + incomingBytes + fragment.byteLength > MAX_PENDING_BYTES) { + throw new Error("node duplex pending message buffer exceeded its bounded capacity"); + } + incomingFragments.push(fragment); + incoming.fragment += 1; + if (!frame.last) { + return; + } + const assembled = Buffer.concat(incomingFragments, incomingBytes + fragment.byteLength); + const message = new Uint8Array(assembled.buffer, assembled.byteOffset, assembled.byteLength); + incomingFragments.length = 0; + incoming.fragment = 0; + incoming.message += 1; + if (listener) { + observeListener(listener, message); + return; + } + if (pendingMessages.length >= MAX_PENDING_MESSAGES) { + throw new Error("node duplex pending message buffer exceeded its bounded capacity"); + } + pendingMessages.push(message); + }; + + return { + send(message: Uint8Array): Promise { + return enqueue(async () => { + if (!(message instanceof Uint8Array) || message.byteLength > maxMessageBytes) { + throw new Error("node duplex logical message exceeds its maximum size"); + } + if (!Number.isSafeInteger(nextOutgoingMessage)) { + throw new Error("node duplex message sequence exceeded its maximum"); + } + const messageId = nextOutgoingMessage++; + const fragments = Math.max(1, Math.ceil(message.byteLength / NODE_DUPLEX_FRAGMENT_BYTES)); + for (let index = 0; index < fragments; index += 1) { + assertOpen(); + const start = index * NODE_DUPLEX_FRAGMENT_BYTES; + const fragment = message.subarray(start, start + NODE_DUPLEX_FRAGMENT_BYTES); + await options.sendFrame( + JSON.stringify({ + v: 1, + kind: "data", + message: messageId, + index, + last: index === fragments - 1, + data: Buffer.from(fragment).toString("base64"), + }), + ); + assertOpen(); + } + }); + }, + + sendReady(): Promise { + return enqueue(async () => { + assertOpen(); + if (ready.sent || nextOutgoingMessage > 0) { + throw new Error("node duplex framed readiness is duplicate or out of order"); + } + ready.sent = true; + await options.sendFrame(JSON.stringify({ v: 1, kind: "ready" })); + assertOpen(); + }); + }, + + receive(frame: string): void { + if (!frame) { + return; + } + assertOpen(); + try { + if (Buffer.byteLength(frame, "utf8") > 16 * 1024) { + throw new Error("node duplex wire frame exceeds the 16 KiB transport limit"); + } + const parsed: unknown = JSON.parse(frame); + if (!isRecord(parsed) || parsed.v !== 1) { + throw new Error("node duplex frame has an unsupported version or shape"); + } + if (parsed.kind === "ready") { + const receivedData = incoming.message > 0 || incoming.fragment > 0; + if (Object.keys(parsed).length !== 2 || ready.received || receivedData) { + throw new Error("node duplex framed readiness is malformed, duplicate, or late"); + } + ready.received = true; + options.onReady?.(); + return; + } + if (parsed.kind !== "data" || (options.requireReady && !ready.received)) { + throw new Error( + "node duplex frame has unsupported kind or arrived before framed readiness", + ); + } + acceptData(parsed); + } catch (error) { + throw fail(error); + } + }, + + onMessage(callback: (message: Uint8Array) => void | Promise): () => void { + assertOpen(); + if (listener) { + throw new Error("node duplex channel already has an active message listener"); + } + listener = callback; + try { + while (pendingMessages.length > 0) { + const message = pendingMessages.shift()!; + observeListener(callback, message); + } + } catch (error) { + throw fail(error); + } + return () => { + if (listener === callback) { + listener = undefined; + } + }; + }, + + close, + drain: async () => { + assertOpen(); + while (activeDeliveries.size > 0) { + await Promise.race([drainClosed.promise, Promise.all(activeDeliveries)]); + } + }, + }; +} diff --git a/src/node-host/invoke.ts b/src/node-host/invoke.ts index b31ced342d86..1430b5f9e803 100644 --- a/src/node-host/invoke.ts +++ b/src/node-host/invoke.ts @@ -82,6 +82,8 @@ const OUTPUT_EVENT_TAIL = 20_000; const DEFAULT_NODE_PATH = "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin"; type NodeHostPrivateInvokeRuntime = NodeHostInvokeRuntime & { + canReportAbortedFailure?: (error: unknown) => boolean; + flushPluginCommandIo?: () => Promise; workerBundleInstaller?: NodeWorkerBundleInstallerControl; workerSupervisor?: NodeWorkerSupervisorControl; workerWorkspace?: NodeWorkerWorkspaceRuntime; @@ -581,7 +583,7 @@ export async function handleInvoke( ) { const invocationClient = createNodeHostInvocationClient(client, runtime.signal); try { - await dispatchInvoke(frame, invocationClient, skillBins, mcpManager, runtime); + await dispatchInvoke(frame, invocationClient, client, skillBins, mcpManager, runtime); } catch (err) { // Gateway events launch this handler without awaiting it. Consume unexpected // failures here so one bad request cannot terminate the node-host process. @@ -603,6 +605,7 @@ export async function handleInvoke( async function dispatchInvoke( frame: NodeInvokeRequestPayload, client: NodeHostClient, + abortedFailureClient: NodeHostClient, skillBins: SkillBinsProvider, mcpManager?: NodeHostMcpManager, runtime: NodeHostPrivateInvokeRuntime = {}, @@ -827,11 +830,14 @@ async function dispatchInvoke( : context; const pluginResult = await invokePlugin(command, frame.paramsJSON, io, invokeContext); if (pluginResult !== null) { + await runtime.flushPluginCommandIo?.(); await sendRawPayloadResult(client, frame, pluginResult); return; } } catch (err) { - await sendInvalidRequestResult(client, frame, err); + // Only the exact current owner's exact framed failure may bypass its aborted-client fence. + const failureClient = runtime.canReportAbortedFailure?.(err) ? abortedFailureClient : client; + await sendInvalidRequestResult(failureClient, frame, err); return; } diff --git a/src/node-host/plugin-node-host.abort.test.ts b/src/node-host/plugin-node-host.abort.test.ts index 2f75dbd75b0e..57483f7c3aa1 100644 --- a/src/node-host/plugin-node-host.abort.test.ts +++ b/src/node-host/plugin-node-host.abort.test.ts @@ -1,9 +1,13 @@ /** Verifies non-duplex plugin commands inherit the node invocation lifetime. */ import { afterEach, describe, expect, it, vi } from "vitest"; import type { GatewayClient } from "../gateway/client.js"; +import { createNodeDuplexEndpoint } from "../infra/node-duplex-framing.js"; import { createEmptyPluginRegistry } from "../plugins/registry-empty.js"; import { resetPluginRuntimeStateForTest, setActivePluginRegistry } from "../plugins/runtime.js"; -import type { OpenClawPluginNodeHostCommandContext } from "../plugins/types.node-host.js"; +import type { + OpenClawPluginNodeHostCommandContext, + OpenClawPluginNodeHostCommandIo, +} from "../plugins/types.node-host.js"; import { handleInvoke } from "./invoke.js"; afterEach(() => { @@ -107,4 +111,119 @@ describe("non-duplex node-host plugin cancellation", () => { expect.objectContaining({ ok: true, payloadJSON: '{"ok":true}' }), ); }); + + it.each(["success", "failure", "cancellation", "supersession", "different-error"] as const)( + "settles pending asynchronous plugin listener delivery before result (%s)", + async (outcome) => { + let resolveListener!: () => void; + let rejectListener!: (error: Error) => void; + const listenerCompleted = new Promise((resolve, reject) => { + resolveListener = resolve; + rejectListener = reject; + }); + const controller = new AbortController(); + let currentInvocation = true; + let framedFailure: Error | undefined; + const framedIo = createNodeDuplexEndpoint({ + sendFrame: async () => undefined, + onError: (error) => { + framedFailure = error; + controller.abort(error); + }, + }); + controller.signal.addEventListener("abort", () => framedIo.close(), { once: true }); + const io: OpenClawPluginNodeHostCommandIo = { + signal: controller.signal, + emitChunk: vi.fn(async (_chunk: string) => undefined), + onInput: vi.fn(), + frames: framedIo, + }; + const handle = vi.fn( + async (_paramsJSON?: string | null, commandIo?: OpenClawPluginNodeHostCommandIo) => { + commandIo?.frames?.onMessage(async () => await listenerCompleted); + framedIo.receive( + JSON.stringify({ v: 1, kind: "data", message: 0, index: 0, last: true, data: "Bw==" }), + ); + if (outcome === "different-error") { + await new Promise((_resolve, reject) => { + controller.signal.addEventListener( + "abort", + () => reject(new Error("identical framed failure message")), + { once: true }, + ); + }); + } + return '{"ok":true}'; + }, + ); + const registry = createEmptyPluginRegistry(); + registry.nodeHostCommands = [ + { + pluginId: "frames-fixture", + pluginName: "Frames fixture", + command: { command: "fixture.duplex", duplex: true, handle }, + source: "test", + }, + ]; + setActivePluginRegistry(registry); + const request = vi.fn().mockResolvedValue(null); + + const invocation = handleInvoke( + { id: "pending-listener", nodeId: "paired-node", command: "fixture.duplex" }, + { request } as unknown as GatewayClient, + { current: async () => [] }, + undefined, + { + signal: controller.signal, + pluginCommandIo: io, + flushPluginCommandIo: framedIo.drain, + canReportAbortedFailure: (error) => + currentInvocation && error === framedFailure && error === controller.signal.reason, + }, + ); + + try { + await vi.waitFor(() => expect(handle).toHaveBeenCalledOnce()); + expect(request).not.toHaveBeenCalled(); + if (outcome === "failure") { + rejectListener(new Error("asynchronous plugin listener rejected")); + } else if (outcome === "cancellation") { + controller.abort(new Error("plugin command canceled")); + } else if (outcome === "supersession") { + currentInvocation = false; + rejectListener(new Error("superseded plugin listener rejected")); + } else if (outcome === "different-error") { + rejectListener(new Error("identical framed failure message")); + } else { + resolveListener(); + } + await invocation; + + if (outcome === "success") { + expect(request).toHaveBeenCalledWith( + "node.invoke.result", + expect.objectContaining({ ok: true, payloadJSON: '{"ok":true}' }), + ); + } else if (outcome === "failure") { + expect(request).toHaveBeenCalledWith( + "node.invoke.result", + expect.objectContaining({ + ok: false, + error: { + code: "INVALID_REQUEST", + message: "Error: asynchronous plugin listener rejected", + }, + }), + ); + } else { + expect(controller.signal.aborted).toBe(true); + expect(request).not.toHaveBeenCalled(); + } + } finally { + resolveListener(); + framedIo.close(); + await invocation; + } + }, + ); }); diff --git a/src/node-host/runtime.test.ts b/src/node-host/runtime.test.ts index a58e9e421ed7..5eef9946832c 100644 --- a/src/node-host/runtime.test.ts +++ b/src/node-host/runtime.test.ts @@ -14,7 +14,7 @@ const mocks = vi.hoisted(() => { initializeWorkerSupervisor: vi.fn(async () => undefined), handleInvoke: vi.fn(async () => undefined), progressStartHeartbeats: vi.fn(), - progressWrite: vi.fn(async () => undefined), + progressWrite: vi.fn(async (_chunk: string) => undefined), startMcp: vi.fn(async (_servers: unknown, _deps?: { signal?: AbortSignal }) => ({ descriptors: [], callMcpTool: vi.fn(), @@ -99,7 +99,7 @@ async function startRuntime() { }); } -function holdInvoke() { +function holdInvoke(onCommand?: (io: OpenClawPluginNodeHostCommandIo) => void) { let io: OpenClawPluginNodeHostCommandIo | undefined; let signal: AbortSignal | undefined; let release: (() => void) | undefined; @@ -113,6 +113,9 @@ function holdInvoke() { }; io = runtime.pluginCommandIo; signal = runtime.signal; + if (io) { + onCommand?.(io); + } await held; }); return { @@ -325,6 +328,276 @@ describe("node-host invoke input dispatch", () => { vi.clearAllMocks(); }); + it("provides framed binary message IO to duplex plugin commands", async () => { + const held = holdInvoke(); + const runtime = await startRuntime(); + const invoking = runtime.invoke(frame); + + try { + await vi.waitFor(() => expect(held.io).toBeDefined()); + expect(held.io).toMatchObject({ + frames: { + send: expect.any(Function), + onMessage: expect.any(Function), + }, + }); + } finally { + held.release(); + await invoking; + await runtime.close(); + } + }); + + it("announces framed readiness only after the plugin registers its message listener", async () => { + const held = holdInvoke(); + const runtime = await startRuntime(); + const invoking = runtime.invoke(frame); + + try { + await vi.waitFor(() => expect(held.io).toBeDefined()); + expect(mocks.progressWrite).not.toHaveBeenCalled(); + + const unsubscribe = held.io?.frames?.onMessage(vi.fn()); + + await vi.waitFor(() => + expect(mocks.progressWrite).toHaveBeenCalledWith(JSON.stringify({ v: 1, kind: "ready" })), + ); + expect(unsubscribe).toEqual(expect.any(Function)); + unsubscribe?.(); + } finally { + held.release(); + await invoking; + await runtime.close(); + } + }); + + it("round-trips binary messages through an external-style duplex plugin command", async () => { + const received = vi.fn(); + const pluginCommand = { + command: "test.duplex", + duplex: true, + handle: (_paramsJSON: string | null, io: OpenClawPluginNodeHostCommandIo) => { + io.frames?.onMessage((message) => { + received(message); + void io.frames?.send(message); + }); + }, + }; + const held = holdInvoke((io) => pluginCommand.handle(frame.paramsJSON, io)); + const runtime = await startRuntime(); + const invoking = runtime.invoke(frame); + + try { + await vi.waitFor(() => expect(mocks.progressWrite).toHaveBeenCalledOnce()); + runtime.handleInput( + frame.id, + 0, + JSON.stringify({ + v: 1, + kind: "data", + message: 0, + index: 0, + last: true, + data: "AP8B", + }), + ); + + await vi.waitFor(() => expect(mocks.progressWrite).toHaveBeenCalledTimes(2)); + expect(received).toHaveBeenCalledWith(Uint8Array.from([0, 255, 1])); + expect(JSON.parse(mocks.progressWrite.mock.calls[1]?.[0] ?? "null")).toMatchObject({ + v: 1, + kind: "data", + message: 0, + index: 0, + last: true, + data: "AP8B", + }); + } finally { + held.release(); + await invoking; + await runtime.close(); + } + }); + + it("preserves binary message boundaries and fragments output below the transport limit", async () => { + const held = holdInvoke(); + const runtime = await startRuntime(); + const invoking = runtime.invoke(frame); + + try { + await vi.waitFor(() => expect(held.io?.frames).toBeDefined()); + const received = vi.fn(); + held.io?.frames?.onMessage(received); + await vi.waitFor(() => expect(mocks.progressWrite).toHaveBeenCalledOnce()); + mocks.progressWrite.mockClear(); + + const incoming = Uint8Array.from({ length: 20_000 }, (_, index) => index % 256); + const incomingFragments = [ + incoming.slice(0, 8_192), + incoming.slice(8_192, 16_384), + incoming.slice(16_384), + ]; + for (const [index, fragment] of incomingFragments.entries()) { + runtime.handleInput( + frame.id, + index, + JSON.stringify({ + v: 1, + kind: "data", + message: 0, + index, + last: index === incomingFragments.length - 1, + data: Buffer.from(fragment).toString("base64"), + }), + ); + } + runtime.handleInput( + frame.id, + incomingFragments.length, + JSON.stringify({ + v: 1, + kind: "data", + message: 1, + index: 0, + last: true, + data: Buffer.from([0, 255]).toString("base64"), + }), + ); + expect(received.mock.calls).toEqual([[incoming], [Uint8Array.from([0, 255])]]); + + const outgoing = Uint8Array.from({ length: 20_000 }, (_, index) => (index * 7) % 256); + await Promise.all([ + held.io?.frames?.send(outgoing), + held.io?.frames?.send(Uint8Array.from([4, 5, 6])), + ]); + + const fragments = mocks.progressWrite.mock.calls.map(([value]) => { + expect(Buffer.byteLength(value, "utf8")).toBeLessThan(16 * 1024); + return JSON.parse(value) as { + v: number; + kind: string; + message: number; + index: number; + last: boolean; + data: string; + }; + }); + expect(fragments.map(({ message }) => message)).toEqual([0, 0, 0, 1]); + expect(fragments.map(({ index }) => index)).toEqual([0, 1, 2, 0]); + expect( + Buffer.concat( + fragments + .filter(({ message }) => message === 0) + .map(({ data }) => Buffer.from(data, "base64")), + ), + ).toEqual(Buffer.from(outgoing)); + expect(Buffer.from(fragments[3]?.data ?? "", "base64")).toEqual(Buffer.from([4, 5, 6])); + } finally { + held.release(); + await invoking; + await runtime.close(); + } + }); + + it.each(["cancel", "result"] as const)( + "closes framed plugin IO after invocation %s", + async (terminalState) => { + const held = holdInvoke(); + const runtime = await startRuntime(); + const invoking = runtime.invoke(frame); + + try { + await vi.waitFor(() => expect(held.io?.frames).toBeDefined()); + const received = vi.fn(); + held.io?.frames?.onMessage(received); + await vi.waitFor(() => expect(mocks.progressWrite).toHaveBeenCalledOnce()); + + if (terminalState === "cancel") { + runtime.cancel(frame.id); + } else { + held.release(); + await invoking; + } + runtime.handleInput( + frame.id, + 0, + JSON.stringify({ + v: 1, + kind: "data", + message: 0, + index: 0, + last: true, + data: "eA==", + }), + ); + + expect(received).not.toHaveBeenCalled(); + await expect(held.io?.frames?.send(Uint8Array.from([1]))).rejects.toThrow(/closed/i); + } finally { + held.release(); + await invoking; + await runtime.close(); + } + }, + ); + + it("aborts a framed plugin command on malformed input without throwing through the transport", async () => { + const held = holdInvoke(); + const runtime = await startRuntime(); + const invoking = runtime.invoke(frame); + + try { + await vi.waitFor(() => expect(held.io?.frames).toBeDefined()); + held.io?.frames?.onMessage(vi.fn()); + await vi.waitFor(() => expect(mocks.progressWrite).toHaveBeenCalledOnce()); + + expect(() => runtime.handleInput(frame.id, 0, "not-json")).not.toThrow(); + expect(held.io?.signal.aborted).toBe(true); + await expect(held.io?.frames?.send(Uint8Array.from([1]))).rejects.toThrow(/closed/i); + } finally { + held.release(); + await invoking; + await runtime.close(); + } + }); + + it("aborts the invocation when its framed plugin message listener fails", async () => { + const held = holdInvoke(); + const runtime = await startRuntime(); + const invoking = runtime.invoke(frame); + + try { + await vi.waitFor(() => expect(held.io?.frames).toBeDefined()); + held.io?.frames?.onMessage(() => { + throw new Error("plugin message rejected"); + }); + await vi.waitFor(() => expect(mocks.progressWrite).toHaveBeenCalledOnce()); + + expect(() => + runtime.handleInput( + frame.id, + 0, + JSON.stringify({ + v: 1, + kind: "data", + message: 0, + index: 0, + last: true, + data: "eA==", + }), + ), + ).not.toThrow(); + expect(held.io?.signal.aborted).toBe(true); + expect(held.io?.signal.reason).toEqual( + expect.objectContaining({ message: "plugin message rejected" }), + ); + } finally { + held.release(); + await invoking; + await runtime.close(); + } + }); + it("buffers frames before the command registers input and flushes them in order", async () => { const held = holdInvoke(); const runtime = await startRuntime(); diff --git a/src/node-host/runtime.ts b/src/node-host/runtime.ts index cac4477f20a5..f2dc073c0b9f 100644 --- a/src/node-host/runtime.ts +++ b/src/node-host/runtime.ts @@ -15,6 +15,7 @@ import { NODE_SYSTEM_RUN_COMMANDS, NODE_TERMINAL_UPLOAD_COMMAND, } from "../infra/node-commands.js"; +import { createNodeDuplexEndpoint } from "../infra/node-duplex-framing.js"; import type { NodeWorkerCapacitySnapshot } from "../infra/node-runner-inventory.js"; import { ensureOpenClawCliOnPath } from "../infra/path-env.js"; import { ensureTerminalUploadCleanup } from "../infra/terminal-file-upload.js"; @@ -90,6 +91,7 @@ type NodeInvokeInputTarget = { type ActiveNodeInvoke = { controller: AbortController; + framedFailure?: Error; input?: NodeInvokeInputTarget; }; @@ -438,8 +440,22 @@ export async function prepareNodeHostRuntime(params?: { if (duplexCommand) { progress?.startHeartbeats(); } - const pluginCommandIo: OpenClawPluginNodeHostCommandIo | undefined = + const framedIo = input && progress + ? createNodeDuplexEndpoint({ + sendFrame: async (payloadJSON) => await progress.write(payloadJSON), + onError: (error) => { + active.framedFailure = error; + controller.abort(error); + }, + }) + : undefined; + if (framedIo) { + controller.signal.addEventListener("abort", () => framedIo.close(), { once: true }); + } + let framedInputRegistered = false; + const pluginCommandIo: OpenClawPluginNodeHostCommandIo | undefined = + input && progress && framedIo ? { signal: controller.signal, emitChunk: async (chunk) => await progress.write(chunk), @@ -448,13 +464,37 @@ export async function prepareNodeHostRuntime(params?: { registerNodeInvokeInputHandler(input, callback); } }, + frames: { + send: async (message) => await framedIo.send(message), + onMessage: (callback) => { + const unsubscribe = framedIo.onMessage(callback); + if (!framedInputRegistered) { + framedInputRegistered = true; + registerNodeInvokeInputHandler(input, (payloadJSON) => { + try { + framedIo.receive(payloadJSON); + } catch (error) { + controller.abort(error); + } + }); + void framedIo.sendReady().catch(controller.abort.bind(controller)); + } + return unsubscribe; + }, + }, } : undefined; try { await handleInvoke(frame, client, skillBins, manager, { ...(claudePath ? { claudePath } : {}), signal: controller.signal, - ...(pluginCommandIo ? { pluginCommandIo } : {}), + pluginCommandIo, + flushPluginCommandIo: framedIo?.drain, + canReportAbortedFailure: (error) => + controller.signal.aborted && + error === active.framedFailure && + error === controller.signal.reason && + activeInvokes.get(frame.id) === active, ...(gatewayConnection?.url ? { gatewayUrl: gatewayConnection.url } : {}), ...(gatewayConnection?.tlsFingerprint ? { gatewayTlsFingerprint: gatewayConnection.tlsFingerprint } @@ -472,6 +512,7 @@ export async function prepareNodeHostRuntime(params?: { ...(workerWorkspace ? { workerWorkspace } : {}), }); } finally { + framedIo?.close(); progress?.stop(); await progress?.flush(); if (activeInvokes.get(frame.id) === active) { diff --git a/src/plugin-sdk/test-helpers/plugin-runtime-mock.ts b/src/plugin-sdk/test-helpers/plugin-runtime-mock.ts index 86f9221d9563..32593b2c74bc 100644 --- a/src/plugin-sdk/test-helpers/plugin-runtime-mock.ts +++ b/src/plugin-sdk/test-helpers/plugin-runtime-mock.ts @@ -1043,6 +1043,7 @@ export function createPluginRuntimeMock(overrides: DeepPartial = nodes: { list: vi.fn(async () => ({ nodes: [] })), invoke: vi.fn(), + openDuplex: vi.fn(), }, }; diff --git a/src/plugins/cli-gateway-nodes-runtime.test.ts b/src/plugins/cli-gateway-nodes-runtime.test.ts index 2cee05ecd34f..6ed7bbeec018 100644 --- a/src/plugins/cli-gateway-nodes-runtime.test.ts +++ b/src/plugins/cli-gateway-nodes-runtime.test.ts @@ -100,4 +100,13 @@ describe("createPluginCliGatewayNodesRuntime", () => { expect(callGatewayMock.mock.calls[0]?.[0]).not.toHaveProperty("signal"); expect(callGatewayMock.mock.calls[0]?.[0].params).not.toHaveProperty("signal"); }); + + it("rejects duplex commands without opening a polling Gateway fallback", async () => { + const nodes = createPluginCliGatewayNodesRuntime(); + + await expect(nodes.openDuplex({ nodeId: "node-1", command: "image.bridge" })).rejects.toThrow( + "unavailable in the CLI", + ); + expect(callGatewayMock).not.toHaveBeenCalled(); + }); }); diff --git a/src/plugins/cli-gateway-nodes-runtime.ts b/src/plugins/cli-gateway-nodes-runtime.ts index d527426307cd..f3d183d04097 100644 --- a/src/plugins/cli-gateway-nodes-runtime.ts +++ b/src/plugins/cli-gateway-nodes-runtime.ts @@ -81,5 +81,8 @@ export function createPluginCliGatewayNodesRuntime(): PluginRuntime["nodes"] { ...(params.signal ? { signal: params.signal } : {}), }); }, + async openDuplex() { + throw new Error("Node duplex is unavailable in the CLI; run this plugin inside the Gateway."); + }, }; } diff --git a/src/plugins/loader-runtime-load.ts b/src/plugins/loader-runtime-load.ts index c53aac7b7a77..d7ebad95bb91 100644 --- a/src/plugins/loader-runtime-load.ts +++ b/src/plugins/loader-runtime-load.ts @@ -59,6 +59,7 @@ function createDeferredGatewayNodesRuntime(runtime: PluginRuntime): PluginRuntim return { list: (...args) => runtime.nodes.list(...args), invoke: (...args) => runtime.nodes.invoke(...args), + openDuplex: (...args) => runtime.nodes.openDuplex(...args), }; } diff --git a/src/plugins/registry-runtime.ts b/src/plugins/registry-runtime.ts index b484a4f75e6d..df11efa10f2d 100644 --- a/src/plugins/registry-runtime.ts +++ b/src/plugins/registry-runtime.ts @@ -41,6 +41,7 @@ import { getGatewayContextResolver, withPluginRuntimePluginIdScope, withPluginRuntimePluginScope, + withPluginRuntimeRegistryScope, } from "./runtime/gateway-request-scope.js"; import type { PluginRuntime } from "./runtime/types.js"; @@ -793,6 +794,10 @@ export function createPluginRuntimeResolver(state: PluginRegistryState) { return { list: (params) => runWithPluginScope(() => nodes.list(params)), invoke: (params) => runWithPluginScope(() => nodes.invoke(params)), + openDuplex: (params) => + withPluginRuntimeRegistryScope(registry, () => + runWithPluginScope(() => nodes.openDuplex(params)), + ), } satisfies PluginRuntime["nodes"]; } if (prop === "agent") { diff --git a/src/plugins/registry.runtime-config.test.ts b/src/plugins/registry.runtime-config.test.ts index 97f390f18adf..bc3d4849b34e 100644 --- a/src/plugins/registry.runtime-config.test.ts +++ b/src/plugins/registry.runtime-config.test.ts @@ -324,6 +324,7 @@ describe("plugin registry runtime config scope", () => { it("runs node helpers with the owning plugin scope", async () => { let listScope = getPluginRuntimeGatewayRequestScope(); let invokeScope = getPluginRuntimeGatewayRequestScope(); + let duplexScope = getPluginRuntimeGatewayRequestScope(); const runtime = createPluginRuntime(); runtime.nodes = { list: vi.fn(async () => { @@ -334,6 +335,15 @@ describe("plugin registry runtime config scope", () => { invokeScope = getPluginRuntimeGatewayRequestScope(); return { ok: true }; }), + openDuplex: vi.fn(async () => { + duplexScope = getPluginRuntimeGatewayRequestScope(); + return { + send: vi.fn(async () => {}), + onMessage: vi.fn(() => () => {}), + closed: Promise.resolve({ ok: true }), + close: vi.fn(), + }; + }), }; const pluginRegistry = createTestRegistry(runtime); const record = createPluginRecord({ @@ -352,6 +362,7 @@ describe("plugin registry runtime config scope", () => { command: "browser.proxy", scopes: ["operator.admin"], }); + await api.runtime.nodes.openDuplex({ nodeId: "node-1", command: "image.bridge" }); expect(listScope).toMatchObject({ pluginId: "google-meet", @@ -361,6 +372,11 @@ describe("plugin registry runtime config scope", () => { pluginId: "google-meet", pluginSource: "/plugins/google-meet/index.js", }); + expect(duplexScope).toMatchObject({ + pluginId: "google-meet", + pluginSource: "/plugins/google-meet/index.js", + }); + expect(duplexScope?.pluginRegistry).toBe(pluginRegistry.registry); }); it("runs gateway requests with the owning plugin scope", async () => { diff --git a/src/plugins/runtime/index.test.ts b/src/plugins/runtime/index.test.ts index 33980d806280..abfa4308e239 100644 --- a/src/plugins/runtime/index.test.ts +++ b/src/plugins/runtime/index.test.ts @@ -468,6 +468,14 @@ describe("plugin runtime command execution", () => { expectGatewaySubagentRunFailure(runtime, { sessionKey: "s-1", message: "hello" }); }); + it("exposes a node duplex capability even when Gateway access is unavailable", () => { + const nodes = createPluginRuntime().nodes; + expect(nodes).toHaveProperty("openDuplex", expect.any(Function)); + expect(() => nodes.openDuplex({ nodeId: "node-1", command: "image.bridge" })).toThrow( + "only available inside the Gateway", + ); + }); + it("uses an explicit subagent runtime", async () => { const run = vi.fn().mockResolvedValue({ runId: "run-1" }); const runtime = createPluginRuntime({ @@ -486,6 +494,7 @@ describe("plugin runtime command execution", () => { const nodes = { list: vi.fn().mockResolvedValue({ nodes: [] }), invoke: vi.fn().mockResolvedValue({ ok: true }), + openDuplex: vi.fn().mockResolvedValue({ closed: Promise.resolve({ ok: true }) }), }; const runtime = createPluginRuntime({ nodes }); @@ -495,5 +504,9 @@ describe("plugin runtime command execution", () => { ).resolves.toEqual({ ok: true }); expect(nodes.list).toHaveBeenCalledWith({ connected: true }); expect(nodes.invoke).toHaveBeenCalledWith({ nodeId: "node-1", command: "browser.proxy" }); + await expect( + runtime.nodes.openDuplex({ nodeId: "node-1", command: "image.bridge" }), + ).resolves.toMatchObject({ closed: expect.any(Promise) }); + expect(nodes.openDuplex).toHaveBeenCalledWith({ nodeId: "node-1", command: "image.bridge" }); }); }); diff --git a/src/plugins/runtime/index.ts b/src/plugins/runtime/index.ts index 04d7aab4c351..991241079ea0 100644 --- a/src/plugins/runtime/index.ts +++ b/src/plugins/runtime/index.ts @@ -189,6 +189,7 @@ function createUnavailableNodesRuntime(): PluginRuntime["nodes"] { return { list: unavailable, invoke: unavailable, + openDuplex: unavailable, }; } diff --git a/src/plugins/runtime/types.ts b/src/plugins/runtime/types.ts index 189147b83000..49d449c4a650 100644 --- a/src/plugins/runtime/types.ts +++ b/src/plugins/runtime/types.ts @@ -105,6 +105,14 @@ type RuntimeNodeInvokeParams = { scopes?: OperatorScope[]; }; +/** A lifecycle-bound, complete-message binary channel for one node invocation. */ +type RuntimeNodeDuplexChannel = { + send: (message: Uint8Array) => Promise; + onMessage: (listener: (message: Uint8Array) => void | Promise) => () => void; + closed: Promise; + close: () => void; +}; + export type RuntimeGatewayRequestOptions = { timeoutMs?: number; /** Requested Gateway scopes. Honored only for bundled or trusted official plugins. */ @@ -134,6 +142,10 @@ export type PluginRuntime = PluginRuntimeCore & { nodes: { list: (params?: RuntimeNodeListParams) => Promise; invoke: (params: RuntimeNodeInvokeParams) => Promise; + /** Open a connection-scoped binary node command inside the trusted Gateway runtime. */ + openDuplex: ( + params: RuntimeNodeInvokeParams & { maxMessageBytes?: number }, + ) => Promise; }; sandbox: { resolveWorkspaceAuthority: (params: { diff --git a/src/plugins/types.node-host.ts b/src/plugins/types.node-host.ts index db596265b668..0a69c4dd300b 100644 --- a/src/plugins/types.node-host.ts +++ b/src/plugins/types.node-host.ts @@ -11,6 +11,11 @@ export type OpenClawPluginNodeHostCommandAvailabilityContext = { export type OpenClawPluginNodeHostCommandIo = { emitChunk(chunk: string): Promise; onInput(callback: (payloadJSON: string) => void): void; + /** Complete binary messages; available when the node host dispatches a duplex command. */ + frames?: { + send(message: Uint8Array): Promise; + onMessage(listener: (message: Uint8Array) => void | Promise): () => void; + }; signal: AbortSignal; };