mirror of
https://github.com/openclaw/openclaw.git
synced 2026-08-24 11:25:50 -06:00
feat(nodes): expose plugin duplex channels (#126961)
* feat(nodes): expose plugin duplex channels * fix(nodes): enforce duplex declarations
This commit is contained in:
Peter Steinberger
committed by
GitHub
GitHub
parent
32fb2fc766
commit
d17bbfc31a
@@ -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<void>`; 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<string>((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.
|
||||
|
||||
<Warning>
|
||||
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`.
|
||||
</Warning>
|
||||
|
||||
</Accordion>
|
||||
|
||||
@@ -10,6 +10,7 @@ function createNodesRuntime(
|
||||
return {
|
||||
list: vi.fn(async () => ({ nodes })),
|
||||
invoke: vi.fn(async () => ({ ok: true })),
|
||||
openDuplex: vi.fn(),
|
||||
};
|
||||
}
|
||||
|
||||
|
||||
@@ -69,6 +69,8 @@ function createContext(opts?: {
|
||||
const invoke = vi.fn(
|
||||
async (params?: {
|
||||
onDispatchReady?: (invokeId: string) => void;
|
||||
onProgress?: (chunk: string) => void;
|
||||
isDispatchAuthorized?: () => boolean;
|
||||
}): Promise<NodeInvokeResult> => {
|
||||
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<PluginApprovalRequestPayload>();
|
||||
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" });
|
||||
|
||||
@@ -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<boolean>;
|
||||
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) {
|
||||
|
||||
@@ -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<string>()),
|
||||
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<string, unknown>;
|
||||
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<GatewayRequestHandlerOptions["client"]> {
|
||||
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,
|
||||
|
||||
@@ -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())) {
|
||||
|
||||
@@ -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 () => []);
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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({
|
||||
|
||||
@@ -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> | 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 ||
|
||||
|
||||
@@ -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<PluginRuntime["nodes"]["openDuplex"]>[0];
|
||||
invokeNode: (
|
||||
params: Parameters<PluginRuntime["nodes"]["invoke"]>[0],
|
||||
stream?: GatewayNodeInvokeStream,
|
||||
signal?: AbortSignal,
|
||||
) => Promise<unknown>;
|
||||
resolveGatewayContext?: GatewayContextResolver;
|
||||
runtimeLifetime?: AbortSignal;
|
||||
}): ReturnType<PluginRuntime["nodes"]["openDuplex"]> {
|
||||
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,
|
||||
|
||||
@@ -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<void>((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<void>((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<void>((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<void>((_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<void>((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<void>((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<void>((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<unknown>;
|
||||
send: (message: Uint8Array) => Promise<void>;
|
||||
}>;
|
||||
};
|
||||
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);
|
||||
|
||||
@@ -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<ReturnType<PluginRuntime["nodes"]["list"]>>["
|
||||
|
||||
export function createGatewayNodesRuntime(
|
||||
resolveGatewayContext?: GatewayContextResolver,
|
||||
runtimeLifetime?: AbortSignal,
|
||||
): PluginRuntime["nodes"] {
|
||||
const invokeNode = async (
|
||||
params: Parameters<PluginRuntime["nodes"]["invoke"]>[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<unknown>(
|
||||
"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<unknown>(
|
||||
"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),
|
||||
},
|
||||
};
|
||||
|
||||
@@ -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, unknown> = {}): 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<void>(() => {});
|
||||
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<void>(() => {}));
|
||||
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<void>((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<void>((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<void>((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<void>((_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<void>(() => {});
|
||||
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<void>((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();
|
||||
});
|
||||
});
|
||||
@@ -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> | 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<void>) | undefined;
|
||||
let sendQueue = Promise.resolve();
|
||||
const drainClosed = createDeferredCore();
|
||||
void drainClosed.promise.catch(() => {});
|
||||
const incomingFragments: Uint8Array[] = [];
|
||||
const pendingMessages: Uint8Array[] = [];
|
||||
const activeDeliveries = new Set<Promise<void>>();
|
||||
|
||||
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<void>): Promise<void> => {
|
||||
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<typeof listener>, 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<string, unknown>) => {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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>): () => 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)]);
|
||||
}
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -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<void>;
|
||||
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;
|
||||
}
|
||||
|
||||
|
||||
@@ -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<void>((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<void>((_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<GatewayClient["request"]>().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;
|
||||
}
|
||||
},
|
||||
);
|
||||
});
|
||||
|
||||
@@ -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();
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -1043,6 +1043,7 @@ export function createPluginRuntimeMock(overrides: DeepPartial<PluginRuntime> =
|
||||
nodes: {
|
||||
list: vi.fn(async () => ({ nodes: [] })),
|
||||
invoke: vi.fn(),
|
||||
openDuplex: vi.fn(),
|
||||
},
|
||||
};
|
||||
|
||||
|
||||
@@ -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();
|
||||
});
|
||||
});
|
||||
|
||||
@@ -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.");
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
@@ -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),
|
||||
};
|
||||
}
|
||||
|
||||
|
||||
@@ -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") {
|
||||
|
||||
@@ -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 () => {
|
||||
|
||||
@@ -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" });
|
||||
});
|
||||
});
|
||||
|
||||
@@ -189,6 +189,7 @@ function createUnavailableNodesRuntime(): PluginRuntime["nodes"] {
|
||||
return {
|
||||
list: unavailable,
|
||||
invoke: unavailable,
|
||||
openDuplex: unavailable,
|
||||
};
|
||||
}
|
||||
|
||||
|
||||
@@ -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<void>;
|
||||
onMessage: (listener: (message: Uint8Array) => void | Promise<void>) => () => void;
|
||||
closed: Promise<unknown>;
|
||||
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<RuntimeNodeListResult>;
|
||||
invoke: (params: RuntimeNodeInvokeParams) => Promise<unknown>;
|
||||
/** Open a connection-scoped binary node command inside the trusted Gateway runtime. */
|
||||
openDuplex: (
|
||||
params: RuntimeNodeInvokeParams & { maxMessageBytes?: number },
|
||||
) => Promise<RuntimeNodeDuplexChannel>;
|
||||
};
|
||||
sandbox: {
|
||||
resolveWorkspaceAuthority: (params: {
|
||||
|
||||
@@ -11,6 +11,11 @@ export type OpenClawPluginNodeHostCommandAvailabilityContext = {
|
||||
export type OpenClawPluginNodeHostCommandIo = {
|
||||
emitChunk(chunk: string): Promise<void>;
|
||||
onInput(callback: (payloadJSON: string) => void): void;
|
||||
/** Complete binary messages; available when the node host dispatches a duplex command. */
|
||||
frames?: {
|
||||
send(message: Uint8Array): Promise<void>;
|
||||
onMessage(listener: (message: Uint8Array) => void | Promise<void>): () => void;
|
||||
};
|
||||
signal: AbortSignal;
|
||||
};
|
||||
|
||||
|
||||
Reference in New Issue
Block a user