mirror of
https://github.com/openclaw/openclaw.git
synced 2026-08-27 12:56:01 -06:00
Revert "fix(protocol): preserve gateway session attribution across node runs"
This reverts commit 735f176b01.
This commit is contained in:
joshavant
@@ -122,16 +122,6 @@ export async function handleClaudeCliNodeInvoke(params: {
|
||||
await params.deps.sendInvalidRequestResult(params.client, params.frame, error);
|
||||
return;
|
||||
}
|
||||
if (Object.hasOwn(params.frame, "sessionKey")) {
|
||||
const sessionKey = params.frame.sessionKey ?? null;
|
||||
const requestWithoutSessionKey = { ...request };
|
||||
delete requestWithoutSessionKey.sessionKey;
|
||||
request = {
|
||||
...requestWithoutSessionKey,
|
||||
...(sessionKey ? { sessionKey } : {}),
|
||||
...(request.systemRunPlan ? { systemRunPlan: { ...request.systemRunPlan, sessionKey } } : {}),
|
||||
};
|
||||
}
|
||||
const approvalCommand = [claudePath, ...request.argv];
|
||||
const preparedApproval = buildSystemRunApprovalPlan({
|
||||
command: approvalCommand,
|
||||
|
||||
@@ -3,7 +3,6 @@ import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { afterEach, describe, expect, it, vi } from "vitest";
|
||||
import type { NodeHostClient } from "./client.js";
|
||||
import type { NodeHostInvokeRuntime } from "./invoke-agent-cli-claude-handler.js";
|
||||
import { decodeClaudeCliNodeRunParams } from "./invoke-agent-cli-claude-params.js";
|
||||
import { runClaudeCliNodeCommand } from "./invoke-agent-cli-claude.js";
|
||||
import { handleInvoke, type NodeInvokeRequestPayload } from "./invoke.js";
|
||||
@@ -227,46 +226,6 @@ describe("Claude CLI node command", () => {
|
||||
});
|
||||
});
|
||||
|
||||
it("clears nested Claude run attribution from an explicit Gateway envelope", async () => {
|
||||
const executable = await executableScript("process.exit(0);");
|
||||
const calls: Array<{ method: string; params: unknown }> = [];
|
||||
const handleSystemRun = vi.fn(
|
||||
async (_options: Parameters<NonNullable<NodeHostInvokeRuntime["handleSystemRun"]>>[0]) =>
|
||||
undefined,
|
||||
);
|
||||
const invokeFrame = frame({
|
||||
argv: ["-p"],
|
||||
sessionKey: "agent:forged:request",
|
||||
systemRunPlan: {
|
||||
argv: [executable, "-p"],
|
||||
cwd: null,
|
||||
commandText: `${executable} -p`,
|
||||
agentId: null,
|
||||
sessionKey: "agent:forged:plan",
|
||||
},
|
||||
idleTimeoutMs: 1_000,
|
||||
timeoutMs: 2_000,
|
||||
});
|
||||
invokeFrame.sessionKey = null;
|
||||
|
||||
await handleInvoke(invokeFrame, client(calls), { current: async () => [] }, undefined, {
|
||||
claudePath: executable,
|
||||
handleSystemRun: handleSystemRun as never,
|
||||
});
|
||||
|
||||
expect(handleSystemRun).toHaveBeenCalledWith(
|
||||
expect.objectContaining({
|
||||
params: expect.objectContaining({
|
||||
systemRunPlan: expect.objectContaining({ sessionKey: null }),
|
||||
}),
|
||||
}),
|
||||
);
|
||||
const runParams = handleSystemRun.mock.calls[0]?.[0]?.params as
|
||||
| { sessionKey?: string }
|
||||
| undefined;
|
||||
expect(runParams).not.toHaveProperty("sessionKey");
|
||||
});
|
||||
|
||||
it("converts forwarded OAuth into a child-only descriptor after approval", async () => {
|
||||
const executable = await executableScript(`
|
||||
const fs = require("node:fs");
|
||||
|
||||
@@ -1,48 +1,5 @@
|
||||
import { describe, expect, it } from "vitest";
|
||||
import { coerceNodeInvokeInputPayload, coerceNodeInvokePayload } from "./invoke-payload.js";
|
||||
|
||||
describe("coerceNodeInvokePayload", () => {
|
||||
it("preserves normalized gateway-owned session attribution", () => {
|
||||
expect(
|
||||
coerceNodeInvokePayload({
|
||||
id: "invoke-1",
|
||||
nodeId: "node-1",
|
||||
command: "system.run",
|
||||
sessionKey: " agent:main:main ",
|
||||
}),
|
||||
).toEqual({
|
||||
id: "invoke-1",
|
||||
nodeId: "node-1",
|
||||
command: "system.run",
|
||||
paramsJSON: null,
|
||||
timeoutMs: null,
|
||||
idempotencyKey: null,
|
||||
sessionKey: "agent:main:main",
|
||||
});
|
||||
});
|
||||
|
||||
it("distinguishes a missing legacy envelope from an explicit clear", () => {
|
||||
expect(
|
||||
coerceNodeInvokePayload({ id: "i", nodeId: "n", command: "system.run" }),
|
||||
).not.toHaveProperty("sessionKey");
|
||||
expect(
|
||||
coerceNodeInvokePayload({
|
||||
id: "i",
|
||||
nodeId: "n",
|
||||
command: "system.run",
|
||||
sessionKey: null,
|
||||
}),
|
||||
).toMatchObject({ sessionKey: null });
|
||||
expect(
|
||||
coerceNodeInvokePayload({
|
||||
id: "i",
|
||||
nodeId: "n",
|
||||
command: "system.run",
|
||||
sessionKey: " ",
|
||||
}),
|
||||
).toMatchObject({ sessionKey: null });
|
||||
});
|
||||
});
|
||||
import { coerceNodeInvokeInputPayload } from "./invoke-payload.js";
|
||||
|
||||
describe("coerceNodeInvokeInputPayload", () => {
|
||||
it("accepts a bounded well-formed input payload", () => {
|
||||
|
||||
@@ -21,7 +21,6 @@ export function coerceNodeInvokePayload(payload: unknown): NodeInvokeRequestPayl
|
||||
: null;
|
||||
const timeoutMs = typeof obj.timeoutMs === "number" ? obj.timeoutMs : null;
|
||||
const idempotencyKey = typeof obj.idempotencyKey === "string" ? obj.idempotencyKey : null;
|
||||
const sessionKey = typeof obj.sessionKey === "string" ? obj.sessionKey.trim() || null : null;
|
||||
return {
|
||||
id,
|
||||
nodeId,
|
||||
@@ -29,7 +28,6 @@ export function coerceNodeInvokePayload(payload: unknown): NodeInvokeRequestPayl
|
||||
paramsJSON,
|
||||
timeoutMs,
|
||||
idempotencyKey,
|
||||
...(Object.hasOwn(obj, "sessionKey") ? { sessionKey } : {}),
|
||||
};
|
||||
}
|
||||
|
||||
|
||||
@@ -912,57 +912,4 @@ describe("node host invoke", () => {
|
||||
allowlistRules: [],
|
||||
});
|
||||
});
|
||||
|
||||
it("clears nested prepare correlation when the Gateway envelope is unattributed", async () => {
|
||||
const request = vi.fn<GatewayClient["request"]>().mockResolvedValue(null);
|
||||
|
||||
await handleInvoke(
|
||||
{
|
||||
id: "invoke-unattributed-prepare",
|
||||
nodeId: "node-1",
|
||||
command: "system.run.prepare",
|
||||
sessionKey: null,
|
||||
paramsJSON: JSON.stringify({
|
||||
command: ["echo", "ok"],
|
||||
sessionKey: "agent:forged:prepare",
|
||||
}),
|
||||
},
|
||||
{ request } as unknown as GatewayClient,
|
||||
{ current: async () => [] },
|
||||
);
|
||||
|
||||
const result = request.mock.calls.find(([method]) => method === "node.invoke.result")?.[1] as {
|
||||
payloadJSON?: string;
|
||||
};
|
||||
const payload = JSON.parse(result.payloadJSON ?? "{}") as {
|
||||
plan?: { sessionKey?: string | null };
|
||||
};
|
||||
expect(payload.plan?.sessionKey).toBeNull();
|
||||
});
|
||||
|
||||
it("preserves nested prepare correlation from a legacy Gateway", async () => {
|
||||
const request = vi.fn<GatewayClient["request"]>().mockResolvedValue(null);
|
||||
|
||||
await handleInvoke(
|
||||
{
|
||||
id: "invoke-legacy-prepare",
|
||||
nodeId: "node-1",
|
||||
command: "system.run.prepare",
|
||||
paramsJSON: JSON.stringify({
|
||||
command: ["echo", "ok"],
|
||||
sessionKey: "agent:legacy:prepare",
|
||||
}),
|
||||
},
|
||||
{ request } as unknown as GatewayClient,
|
||||
{ current: async () => [] },
|
||||
);
|
||||
|
||||
const result = request.mock.calls.find(([method]) => method === "node.invoke.result")?.[1] as {
|
||||
payloadJSON?: string;
|
||||
};
|
||||
const payload = JSON.parse(result.payloadJSON ?? "{}") as {
|
||||
plan?: { sessionKey?: string | null };
|
||||
};
|
||||
expect(payload.plan?.sessionKey).toBe("agent:legacy:prepare");
|
||||
});
|
||||
});
|
||||
|
||||
+8
-35
@@ -133,29 +133,6 @@ function resolveNodeSkillCwdParam<T extends { cwd?: unknown }>(params: T, nodeId
|
||||
return resolved ? { ...params, cwd: resolved } : params;
|
||||
}
|
||||
|
||||
function bindNodeInvokeSessionKey<
|
||||
T extends {
|
||||
sessionKey?: unknown;
|
||||
systemRunPlan?: SystemRunParams["systemRunPlan"];
|
||||
},
|
||||
>(params: T, frame: NodeInvokeRequestPayload): T {
|
||||
if (!Object.hasOwn(frame, "sessionKey")) {
|
||||
return params;
|
||||
}
|
||||
const sessionKey = frame.sessionKey ?? null;
|
||||
// The Gateway envelope owns run correlation. Nested command params are
|
||||
// caller-controlled and must not mint or retain a different session binding.
|
||||
const systemRunPlan =
|
||||
params.systemRunPlan === undefined || params.systemRunPlan === null
|
||||
? params.systemRunPlan
|
||||
: { ...params.systemRunPlan, sessionKey: sessionKey ?? null };
|
||||
return {
|
||||
...params,
|
||||
sessionKey,
|
||||
...(systemRunPlan !== undefined ? { systemRunPlan } : {}),
|
||||
};
|
||||
}
|
||||
|
||||
function buildEnvOverrideRejectionMessage(params: {
|
||||
rejectedOverrideBlockedKeys: string[];
|
||||
rejectedOverrideInvalidKeys: string[];
|
||||
@@ -769,12 +746,11 @@ async function dispatchInvoke(
|
||||
}
|
||||
try {
|
||||
const { pluginCommandIo: io, pluginCommandContext: context } = runtime;
|
||||
const hasSessionKeyEnvelope = Object.hasOwn(frame, "sessionKey");
|
||||
const invokeContext =
|
||||
context && (hasSessionKeyEnvelope || runtime.signal)
|
||||
context && (frame.sessionKey || runtime.signal)
|
||||
? {
|
||||
...context,
|
||||
...(hasSessionKeyEnvelope ? { sessionKey: frame.sessionKey ?? undefined } : {}),
|
||||
...(frame.sessionKey ? { sessionKey: frame.sessionKey } : {}),
|
||||
...(runtime.signal ? { signal: runtime.signal } : {}),
|
||||
}
|
||||
: context;
|
||||
@@ -790,12 +766,9 @@ async function dispatchInvoke(
|
||||
|
||||
if (command === "system.run.prepare") {
|
||||
try {
|
||||
const params = bindNodeInvokeSessionKey(
|
||||
resolveNodeSkillCwdParam(
|
||||
decodeParams<SystemRunPrepareParams>(frame.paramsJSON),
|
||||
frame.nodeId,
|
||||
),
|
||||
frame,
|
||||
const params = resolveNodeSkillCwdParam(
|
||||
decodeParams<SystemRunPrepareParams>(frame.paramsJSON),
|
||||
frame.nodeId,
|
||||
);
|
||||
const prepared = buildSystemRunApprovalPlan(params);
|
||||
if (!prepared.ok) {
|
||||
@@ -852,9 +825,9 @@ async function dispatchInvoke(
|
||||
|
||||
let params: SystemRunParams;
|
||||
try {
|
||||
params = bindNodeInvokeSessionKey(
|
||||
resolveNodeSkillCwdParam(decodeParams<SystemRunParams>(frame.paramsJSON), frame.nodeId),
|
||||
frame,
|
||||
params = resolveNodeSkillCwdParam(
|
||||
decodeParams<SystemRunParams>(frame.paramsJSON),
|
||||
frame.nodeId,
|
||||
);
|
||||
} catch (err) {
|
||||
await sendInvalidRequestResult(client, frame, err);
|
||||
|
||||
@@ -1,490 +0,0 @@
|
||||
import { beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { GatewayClientRequestError, type GatewayClientOptions } from "../gateway/client.js";
|
||||
import type { configureNodeHost } from "./config.js";
|
||||
import type { NodeInvokeRequestPayload } from "./invoke-types.js";
|
||||
import { runNodeHost } from "./runner.js";
|
||||
|
||||
const mocks = vi.hoisted(() => ({
|
||||
capturedGatewayClientOptions: [] as GatewayClientOptions[],
|
||||
capturedGatewayClients: [] as Array<{
|
||||
request: ReturnType<typeof vi.fn<(method: string, params?: unknown) => Promise<unknown>>>;
|
||||
stop: ReturnType<typeof vi.fn>;
|
||||
updateNodeManifest: ReturnType<typeof vi.fn>;
|
||||
}>,
|
||||
activeRuntime: {
|
||||
invoke: vi.fn(async (_payload: NodeInvokeRequestPayload) => {}),
|
||||
handleInput: vi.fn(),
|
||||
cancel: vi.fn(),
|
||||
cancelAll: vi.fn(),
|
||||
close: vi.fn(async () => {}),
|
||||
},
|
||||
configureNodeHost: vi.fn(async (params: Parameters<typeof configureNodeHost>[0]) => ({
|
||||
version: 1 as const,
|
||||
nodeId: params.nodeId?.trim() || "node-test",
|
||||
displayName: params.displayName?.trim() || params.fallbackDisplayName,
|
||||
gateway: params.gateway,
|
||||
})),
|
||||
getRuntimeConfig: vi.fn(() => ({
|
||||
gateway: { handshakeTimeoutMs: 1_000 },
|
||||
})),
|
||||
startGatewayClientWhenEventLoopReady: vi.fn(async () => ({
|
||||
ready: false,
|
||||
aborted: false,
|
||||
elapsedMs: 0,
|
||||
})),
|
||||
}));
|
||||
|
||||
vi.mock("../config/config.js", () => ({
|
||||
getRuntimeConfig: mocks.getRuntimeConfig,
|
||||
}));
|
||||
|
||||
vi.mock("../gateway/client-start-readiness.js", () => ({
|
||||
startGatewayClientWhenEventLoopReady: mocks.startGatewayClientWhenEventLoopReady,
|
||||
}));
|
||||
|
||||
vi.mock("../gateway/client.js", () => ({
|
||||
GatewayClientRequestError: class MockGatewayClientRequestError extends Error {
|
||||
readonly gatewayCode: string;
|
||||
|
||||
constructor(params: { code: string; message: string }) {
|
||||
super(params.message);
|
||||
this.gatewayCode = params.code;
|
||||
}
|
||||
},
|
||||
GatewayClient: function GatewayClient(opts: GatewayClientOptions) {
|
||||
const client = {
|
||||
request: vi.fn<(method: string, params?: unknown) => Promise<unknown>>(async () => ({})),
|
||||
stop: vi.fn(),
|
||||
updateNodeManifest: vi.fn(),
|
||||
};
|
||||
mocks.capturedGatewayClientOptions.push(opts);
|
||||
mocks.capturedGatewayClients.push(client);
|
||||
return client;
|
||||
},
|
||||
}));
|
||||
|
||||
vi.mock("../gateway/credentials-secret-inputs.js", () => ({
|
||||
resolveGatewayCredentialsWithSecretInputs: vi.fn(async () => ({})),
|
||||
}));
|
||||
|
||||
vi.mock("../infra/device-identity.js", () => ({
|
||||
loadOrCreateDeviceIdentity: vi.fn(() => ({
|
||||
id: "device-test",
|
||||
publicKey: "public-key-test",
|
||||
privateKey: "private-key-test",
|
||||
})),
|
||||
}));
|
||||
|
||||
vi.mock("../infra/machine-name.js", () => ({
|
||||
getMachineDisplayName: vi.fn(async () => "test-node"),
|
||||
}));
|
||||
|
||||
vi.mock("../infra/executable-path.js", () => ({
|
||||
resolveExecutableFromPathEnv: vi.fn(() => null),
|
||||
}));
|
||||
|
||||
vi.mock("../infra/path-env.js", () => ({
|
||||
ensureOpenClawCliOnPath: vi.fn(),
|
||||
}));
|
||||
|
||||
vi.mock("./config.js", () => ({
|
||||
configureNodeHost: mocks.configureNodeHost,
|
||||
}));
|
||||
|
||||
vi.mock("./plugin-node-host.js", () => ({
|
||||
ensureNodeHostPluginRegistry: vi.fn(async () => undefined),
|
||||
listRegisteredNodeHostCapsAndCommands: vi.fn(() => ({
|
||||
commands: [],
|
||||
caps: [],
|
||||
nodePluginTools: [],
|
||||
})),
|
||||
watchRegisteredNodeHostCommandAvailability: vi.fn(() => () => undefined),
|
||||
}));
|
||||
|
||||
vi.mock("./mcp.js", () => ({
|
||||
startNodeHostMcpManager: vi.fn(async () => ({
|
||||
configuredServerCount: 0,
|
||||
descriptors: [],
|
||||
callMcpTool: vi.fn(),
|
||||
close: vi.fn(async () => undefined),
|
||||
})),
|
||||
}));
|
||||
|
||||
vi.mock("./skills.js", () => ({
|
||||
scanNodeHostedSkills: vi.fn(() => []),
|
||||
}));
|
||||
|
||||
vi.mock("./startup-state-migrations.js", () => ({
|
||||
runStartupMigrations: vi.fn(async () => undefined),
|
||||
}));
|
||||
|
||||
vi.mock("./runtime.js", async (importOriginal) => {
|
||||
const actual = await importOriginal<typeof import("./runtime.js")>();
|
||||
return {
|
||||
...actual,
|
||||
prepareNodeHostRuntime: async () => ({
|
||||
manifest: { caps: [], commands: [], pathEnv: process.env.PATH ?? "" },
|
||||
initialInventory: { skills: [], pluginTools: [] },
|
||||
start: () => mocks.activeRuntime,
|
||||
}),
|
||||
};
|
||||
});
|
||||
|
||||
function hello(options: GatewayClientOptions | undefined) {
|
||||
options?.onHelloOk?.({
|
||||
protocol: 1,
|
||||
features: { methods: [], events: [] },
|
||||
} as unknown as Parameters<NonNullable<GatewayClientOptions["onHelloOk"]>>[0]);
|
||||
}
|
||||
|
||||
function deferNegotiation(
|
||||
client: (typeof mocks.capturedGatewayClients)[number] | undefined,
|
||||
): () => void {
|
||||
let resolveNegotiation: (() => void) | undefined;
|
||||
client?.request.mockImplementation((method: string) => {
|
||||
if (method === "node.protocolFeatures.update") {
|
||||
return new Promise((resolve) => {
|
||||
resolveNegotiation = () => resolve({});
|
||||
});
|
||||
}
|
||||
return Promise.resolve({});
|
||||
});
|
||||
return () => resolveNegotiation?.();
|
||||
}
|
||||
|
||||
async function waitForProtocolFeaturesNegotiation(
|
||||
client: (typeof mocks.capturedGatewayClients)[number] | undefined,
|
||||
expectedCount = 1,
|
||||
): Promise<void> {
|
||||
await vi.waitFor(() => {
|
||||
expect(
|
||||
client?.request.mock.calls.filter(([method]) => method === "node.protocolFeatures.update"),
|
||||
).toHaveLength(expectedCount);
|
||||
});
|
||||
await new Promise<void>((resolve) => {
|
||||
setImmediate(resolve);
|
||||
});
|
||||
}
|
||||
|
||||
async function startFakeNodeHost() {
|
||||
await expect(runNodeHost({ gatewayHost: "127.0.0.1", gatewayPort: 18789 })).rejects.toThrow(
|
||||
"event loop readiness timeout",
|
||||
);
|
||||
return {
|
||||
options: mocks.capturedGatewayClientOptions[0],
|
||||
client: mocks.capturedGatewayClients[0],
|
||||
};
|
||||
}
|
||||
|
||||
describe("node-host session envelope negotiation", () => {
|
||||
beforeEach(() => {
|
||||
mocks.capturedGatewayClientOptions.length = 0;
|
||||
mocks.capturedGatewayClients.length = 0;
|
||||
vi.clearAllMocks();
|
||||
mocks.getRuntimeConfig.mockReturnValue({
|
||||
gateway: { handshakeTimeoutMs: 1_000 },
|
||||
});
|
||||
});
|
||||
|
||||
it("preserves legacy semantics for invokes received before negotiation completes", async () => {
|
||||
const { options, client } = await startFakeNodeHost();
|
||||
const resolveNegotiation = deferNegotiation(client);
|
||||
|
||||
hello(options);
|
||||
options?.onEvent?.({
|
||||
type: "event",
|
||||
event: "node.invoke.request",
|
||||
payload: {
|
||||
id: "invoke-negotiating",
|
||||
nodeId: "node-1",
|
||||
command: "system.run",
|
||||
paramsJSON: '{"sessionKey":"nested-session"}',
|
||||
},
|
||||
});
|
||||
options?.onEvent?.({
|
||||
type: "event",
|
||||
event: "node.invoke.input",
|
||||
payload: {
|
||||
id: "invoke-negotiating",
|
||||
nodeId: "node-1",
|
||||
seq: 1,
|
||||
payloadJSON: '{"kind":"data"}',
|
||||
},
|
||||
});
|
||||
await vi.waitFor(() => {
|
||||
expect(mocks.activeRuntime.invoke).toHaveBeenCalledOnce();
|
||||
expect(mocks.activeRuntime.handleInput).toHaveBeenCalledWith(
|
||||
"invoke-negotiating",
|
||||
1,
|
||||
'{"kind":"data"}',
|
||||
);
|
||||
});
|
||||
const invokePayload = mocks.activeRuntime.invoke.mock.calls[0]?.[0];
|
||||
expect(invokePayload).toEqual(
|
||||
expect.objectContaining({
|
||||
id: "invoke-negotiating",
|
||||
paramsJSON: '{"sessionKey":"nested-session"}',
|
||||
}),
|
||||
);
|
||||
expect(Object.hasOwn(invokePayload ?? {}, "sessionKey")).toBe(false);
|
||||
expect(mocks.activeRuntime.invoke.mock.invocationCallOrder[0]).toBeLessThan(
|
||||
mocks.activeRuntime.handleInput.mock.invocationCallOrder[0] ?? Number.POSITIVE_INFINITY,
|
||||
);
|
||||
resolveNegotiation();
|
||||
await waitForProtocolFeaturesNegotiation(client);
|
||||
});
|
||||
|
||||
it("does not let negotiation block explicit envelopes or unrelated controls", async () => {
|
||||
const dateNowSpy = vi.spyOn(Date, "now").mockReturnValue(1_000);
|
||||
try {
|
||||
const { options, client } = await startFakeNodeHost();
|
||||
const resolveNegotiation = deferNegotiation(client);
|
||||
|
||||
hello(options);
|
||||
options?.onEvent?.({
|
||||
type: "event",
|
||||
event: "node.invoke.request",
|
||||
payload: {
|
||||
id: "invoke-explicit-envelope",
|
||||
nodeId: "node-1",
|
||||
command: "system.run",
|
||||
timeoutMs: 10,
|
||||
sessionKey: "agent:main:explicit",
|
||||
},
|
||||
});
|
||||
options?.onEvent?.({
|
||||
type: "event",
|
||||
event: "node.invoke.input",
|
||||
payload: {
|
||||
id: "invoke-explicit-envelope",
|
||||
nodeId: "node-1",
|
||||
seq: 1,
|
||||
payloadJSON: '{"kind":"data"}',
|
||||
},
|
||||
});
|
||||
options?.onEvent?.({
|
||||
type: "event",
|
||||
event: "node.invoke.cancel",
|
||||
payload: {
|
||||
invokeId: "invoke-already-running",
|
||||
nodeId: "node-1",
|
||||
},
|
||||
});
|
||||
|
||||
await vi.waitFor(() => {
|
||||
expect(mocks.activeRuntime.invoke).toHaveBeenCalledTimes(1);
|
||||
expect(mocks.activeRuntime.invoke).toHaveBeenLastCalledWith(
|
||||
expect.objectContaining({
|
||||
id: "invoke-explicit-envelope",
|
||||
sessionKey: "agent:main:explicit",
|
||||
timeoutMs: 10,
|
||||
}),
|
||||
);
|
||||
expect(mocks.activeRuntime.handleInput).toHaveBeenCalledWith(
|
||||
"invoke-explicit-envelope",
|
||||
1,
|
||||
'{"kind":"data"}',
|
||||
);
|
||||
expect(mocks.activeRuntime.cancel).toHaveBeenCalledWith("invoke-already-running");
|
||||
});
|
||||
expect(mocks.activeRuntime.invoke.mock.invocationCallOrder[0]).toBeLessThan(
|
||||
mocks.activeRuntime.handleInput.mock.invocationCallOrder[0] ?? Number.POSITIVE_INFINITY,
|
||||
);
|
||||
resolveNegotiation();
|
||||
await waitForProtocolFeaturesNegotiation(client);
|
||||
} finally {
|
||||
dateNowSpy.mockRestore();
|
||||
}
|
||||
});
|
||||
|
||||
it("charges queued dispatch against the invoke deadline", async () => {
|
||||
const dateNowSpy = vi.spyOn(Date, "now");
|
||||
let nowMs = 1_000;
|
||||
dateNowSpy.mockImplementation(() => nowMs);
|
||||
try {
|
||||
const { options, client } = await startFakeNodeHost();
|
||||
const resolveNegotiation = deferNegotiation(client);
|
||||
|
||||
hello(options);
|
||||
options?.onEvent?.({
|
||||
type: "event",
|
||||
event: "node.invoke.request",
|
||||
payload: {
|
||||
id: "invoke-with-deadline",
|
||||
nodeId: "node-1",
|
||||
command: "system.run",
|
||||
timeoutMs: 100,
|
||||
},
|
||||
});
|
||||
|
||||
nowMs += 40;
|
||||
await vi.waitFor(() => {
|
||||
expect(mocks.activeRuntime.invoke).toHaveBeenCalledWith(
|
||||
expect.objectContaining({
|
||||
id: "invoke-with-deadline",
|
||||
timeoutMs: 60,
|
||||
}),
|
||||
);
|
||||
});
|
||||
resolveNegotiation();
|
||||
await waitForProtocolFeaturesNegotiation(client);
|
||||
} finally {
|
||||
dateNowSpy.mockRestore();
|
||||
}
|
||||
});
|
||||
|
||||
it("does not dispatch invokes that expire before queued dispatch", async () => {
|
||||
const dateNowSpy = vi.spyOn(Date, "now");
|
||||
let nowMs = 1_000;
|
||||
dateNowSpy.mockImplementation(() => nowMs);
|
||||
try {
|
||||
const { options, client } = await startFakeNodeHost();
|
||||
const resolveNegotiation = deferNegotiation(client);
|
||||
|
||||
hello(options);
|
||||
options?.onEvent?.({
|
||||
type: "event",
|
||||
event: "node.invoke.request",
|
||||
payload: {
|
||||
id: "invoke-expired",
|
||||
nodeId: "node-1",
|
||||
command: "system.run",
|
||||
timeoutMs: 10,
|
||||
},
|
||||
});
|
||||
options?.onEvent?.({
|
||||
type: "event",
|
||||
event: "node.invoke.input",
|
||||
payload: {
|
||||
id: "invoke-expired",
|
||||
nodeId: "node-1",
|
||||
seq: 0,
|
||||
payloadJSON: '{"kind":"barrier"}',
|
||||
},
|
||||
});
|
||||
|
||||
nowMs += 10;
|
||||
await vi.waitFor(() => {
|
||||
expect(mocks.activeRuntime.handleInput).toHaveBeenCalledWith(
|
||||
"invoke-expired",
|
||||
0,
|
||||
'{"kind":"barrier"}',
|
||||
);
|
||||
});
|
||||
expect(mocks.activeRuntime.invoke).not.toHaveBeenCalled();
|
||||
resolveNegotiation();
|
||||
await waitForProtocolFeaturesNegotiation(client);
|
||||
} finally {
|
||||
dateNowSpy.mockRestore();
|
||||
}
|
||||
});
|
||||
|
||||
it("does not dispatch invokes cancelled before queued dispatch", async () => {
|
||||
const { options, client } = await startFakeNodeHost();
|
||||
const resolveNegotiation = deferNegotiation(client);
|
||||
|
||||
hello(options);
|
||||
options?.onEvent?.({
|
||||
type: "event",
|
||||
event: "node.invoke.request",
|
||||
payload: {
|
||||
id: "invoke-cancelled",
|
||||
nodeId: "node-1",
|
||||
command: "system.run",
|
||||
},
|
||||
});
|
||||
options?.onEvent?.({
|
||||
type: "event",
|
||||
event: "node.invoke.cancel",
|
||||
payload: {
|
||||
invokeId: "invoke-cancelled",
|
||||
nodeId: "node-1",
|
||||
},
|
||||
});
|
||||
|
||||
await vi.waitFor(() => {
|
||||
expect(mocks.activeRuntime.cancel).toHaveBeenCalledWith("invoke-cancelled");
|
||||
});
|
||||
expect(mocks.activeRuntime.invoke).not.toHaveBeenCalled();
|
||||
resolveNegotiation();
|
||||
await waitForProtocolFeaturesNegotiation(client);
|
||||
});
|
||||
|
||||
it("preserves absent envelopes only after an old gateway is confirmed", async () => {
|
||||
const { options, client } = await startFakeNodeHost();
|
||||
client?.request.mockRejectedValueOnce(
|
||||
new GatewayClientRequestError({
|
||||
code: "INVALID_REQUEST",
|
||||
message: "unknown method: node.protocolFeatures.update",
|
||||
}),
|
||||
);
|
||||
|
||||
hello(options);
|
||||
await waitForProtocolFeaturesNegotiation(client);
|
||||
options?.onEvent?.({
|
||||
type: "event",
|
||||
event: "node.invoke.request",
|
||||
payload: {
|
||||
id: "invoke-legacy",
|
||||
nodeId: "node-1",
|
||||
command: "system.run",
|
||||
paramsJSON: '{"sessionKey":"legacy-session"}',
|
||||
},
|
||||
});
|
||||
|
||||
await vi.waitFor(() => expect(mocks.activeRuntime.invoke).toHaveBeenCalledOnce());
|
||||
const payload = mocks.activeRuntime.invoke.mock.calls[0]?.[0];
|
||||
expect(payload && Object.hasOwn(payload, "sessionKey")).toBe(false);
|
||||
});
|
||||
|
||||
it("renegotiates authoritative envelopes after reconnecting from an old gateway", async () => {
|
||||
const { options, client } = await startFakeNodeHost();
|
||||
client?.request.mockRejectedValueOnce(
|
||||
new GatewayClientRequestError({
|
||||
code: "INVALID_REQUEST",
|
||||
message: "unknown method: node.protocolFeatures.update",
|
||||
}),
|
||||
);
|
||||
|
||||
hello(options);
|
||||
options?.onEvent?.({
|
||||
type: "event",
|
||||
event: "node.invoke.request",
|
||||
payload: {
|
||||
id: "invoke-legacy",
|
||||
nodeId: "node-1",
|
||||
command: "system.run",
|
||||
},
|
||||
});
|
||||
await vi.waitFor(() => expect(mocks.activeRuntime.invoke).toHaveBeenCalledOnce());
|
||||
expect(Object.hasOwn(mocks.activeRuntime.invoke.mock.calls[0]?.[0] ?? {}, "sessionKey")).toBe(
|
||||
false,
|
||||
);
|
||||
|
||||
options?.onClose?.(1000, "old gateway closed");
|
||||
const resolveNegotiation = deferNegotiation(client);
|
||||
hello(options);
|
||||
resolveNegotiation();
|
||||
await waitForProtocolFeaturesNegotiation(client, 2);
|
||||
options?.onEvent?.({
|
||||
type: "event",
|
||||
event: "node.invoke.request",
|
||||
payload: {
|
||||
id: "invoke-authoritative",
|
||||
nodeId: "node-1",
|
||||
command: "system.run",
|
||||
},
|
||||
});
|
||||
|
||||
await vi.waitFor(() => expect(mocks.activeRuntime.invoke).toHaveBeenCalledTimes(2));
|
||||
expect(mocks.activeRuntime.invoke.mock.calls[1]?.[0]).toEqual(
|
||||
expect.objectContaining({
|
||||
id: "invoke-authoritative",
|
||||
sessionKey: null,
|
||||
}),
|
||||
);
|
||||
expect(
|
||||
client?.request.mock.calls.filter(([method]) => method === "node.protocolFeatures.update"),
|
||||
).toHaveLength(2);
|
||||
});
|
||||
});
|
||||
@@ -3,7 +3,6 @@ import { beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { ConnectErrorDetailCodes } from "../../packages/gateway-protocol/src/connect-error-details.js";
|
||||
import type { GatewayClientOptions } from "../gateway/client.js";
|
||||
import type { configureNodeHost } from "./config.js";
|
||||
import type { NodeInvokeRequestPayload } from "./invoke-types.js";
|
||||
import { startNodeHostMcpManager, type NodeHostMcpManager } from "./mcp.js";
|
||||
import { runNodeHost } from "./runner.js";
|
||||
|
||||
@@ -11,7 +10,7 @@ const mocks = vi.hoisted(() => ({
|
||||
capturedGatewayClientOptions: [] as GatewayClientOptions[],
|
||||
capturedConfiguredGatewayConfigs: [] as Array<{ contextPath?: string }>,
|
||||
capturedGatewayClients: [] as Array<{
|
||||
request: ReturnType<typeof vi.fn<(method: string, params?: unknown) => Promise<unknown>>>;
|
||||
request: ReturnType<typeof vi.fn>;
|
||||
stop: ReturnType<typeof vi.fn>;
|
||||
updateNodeManifest: ReturnType<typeof vi.fn>;
|
||||
}>,
|
||||
@@ -50,7 +49,7 @@ const mocks = vi.hoisted(() => ({
|
||||
})),
|
||||
resolveGatewayCredentialsWithSecretInputs: vi.fn(async () => ({})),
|
||||
activeRuntime: {
|
||||
invoke: vi.fn(async (_payload: NodeInvokeRequestPayload) => {}),
|
||||
invoke: vi.fn(async () => {}),
|
||||
handleInput: vi.fn(),
|
||||
cancel: vi.fn(),
|
||||
cancelAll: vi.fn(),
|
||||
@@ -67,17 +66,9 @@ vi.mock("../gateway/client-start-readiness.js", () => ({
|
||||
}));
|
||||
|
||||
vi.mock("../gateway/client.js", () => ({
|
||||
GatewayClientRequestError: class MockGatewayClientRequestError extends Error {
|
||||
readonly gatewayCode: string;
|
||||
|
||||
constructor(params: { code: string; message: string }) {
|
||||
super(params.message);
|
||||
this.gatewayCode = params.code;
|
||||
}
|
||||
},
|
||||
GatewayClient: function GatewayClient(opts: GatewayClientOptions) {
|
||||
const client = {
|
||||
request: vi.fn<(method: string, params?: unknown) => Promise<unknown>>(async () => ({})),
|
||||
request: vi.fn(async () => ({})),
|
||||
stop: vi.fn(),
|
||||
updateNodeManifest: vi.fn(),
|
||||
};
|
||||
@@ -254,27 +245,6 @@ describe("runNodeHost", () => {
|
||||
);
|
||||
const options = lastCapturedOptions();
|
||||
|
||||
options?.onEvent?.({
|
||||
type: "event",
|
||||
event: "node.invoke.request",
|
||||
payload: {
|
||||
id: "invoke-1",
|
||||
nodeId: "node-1",
|
||||
command: "system.run",
|
||||
sessionKey: "agent:main:main",
|
||||
},
|
||||
});
|
||||
await vi.waitFor(() =>
|
||||
expect(mocks.activeRuntime.invoke).toHaveBeenCalledWith({
|
||||
id: "invoke-1",
|
||||
nodeId: "node-1",
|
||||
command: "system.run",
|
||||
paramsJSON: null,
|
||||
timeoutMs: null,
|
||||
idempotencyKey: null,
|
||||
sessionKey: "agent:main:main",
|
||||
}),
|
||||
);
|
||||
options?.onEvent?.({
|
||||
type: "event",
|
||||
event: "node.invoke.input",
|
||||
@@ -285,15 +255,10 @@ describe("runNodeHost", () => {
|
||||
event: "node.invoke.cancel",
|
||||
payload: { invokeId: "invoke-1", nodeId: "node-1" },
|
||||
});
|
||||
await vi.waitFor(() => {
|
||||
expect(mocks.activeRuntime.handleInput).toHaveBeenCalledWith(
|
||||
"invoke-1",
|
||||
3,
|
||||
'{"kind":"data"}',
|
||||
);
|
||||
expect(mocks.activeRuntime.cancel).toHaveBeenCalledWith("invoke-1");
|
||||
});
|
||||
options?.onClose?.(1000, "connection closed");
|
||||
|
||||
expect(mocks.activeRuntime.handleInput).toHaveBeenCalledWith("invoke-1", 3, '{"kind":"data"}');
|
||||
expect(mocks.activeRuntime.cancel).toHaveBeenCalledWith("invoke-1");
|
||||
expect(mocks.activeRuntime.cancelAll).toHaveBeenCalledOnce();
|
||||
});
|
||||
|
||||
@@ -532,9 +497,6 @@ describe("runNodeHost", () => {
|
||||
features: { methods: [], events: [] },
|
||||
} as unknown as Parameters<NonNullable<GatewayClientOptions["onHelloOk"]>>[0]);
|
||||
|
||||
expect(client?.request).toHaveBeenCalledWith("node.protocolFeatures.update", {
|
||||
features: ["node-invoke-session-key-envelope-v1"],
|
||||
});
|
||||
expect(client?.request).toHaveBeenCalledWith("node.pluginTools.update", {
|
||||
tools: [
|
||||
{
|
||||
|
||||
+3
-122
@@ -4,7 +4,6 @@ import {
|
||||
GATEWAY_CLIENT_NAMES,
|
||||
} from "../../packages/gateway-protocol/src/client-info.js";
|
||||
import { ConnectErrorDetailCodes } from "../../packages/gateway-protocol/src/connect-error-details.js";
|
||||
import { NODE_INVOKE_SESSION_KEY_ENVELOPE_PROTOCOL_FEATURE } from "../../packages/gateway-protocol/src/schema/nodes.js";
|
||||
import { getRuntimeConfig, type OpenClawConfig } from "../config/config.js";
|
||||
import { startGatewayClientWhenEventLoopReady } from "../gateway/client-start-readiness.js";
|
||||
import {
|
||||
@@ -22,7 +21,6 @@ import {
|
||||
coerceNodeInvokeInputPayload,
|
||||
coerceNodeInvokePayload,
|
||||
} from "./invoke-payload.js";
|
||||
import type { NodeInvokeRequestPayload } from "./invoke-types.js";
|
||||
import { prepareNodeHostRuntime, type NodeHostInventory } from "./runtime.js";
|
||||
import { runStartupMigrations } from "./startup-state-migrations.js";
|
||||
|
||||
@@ -126,35 +124,6 @@ function isUnsupportedNodeSkillsUpdateError(error: unknown): boolean {
|
||||
);
|
||||
}
|
||||
|
||||
function isUnsupportedNodeProtocolFeaturesUpdateError(error: unknown): boolean {
|
||||
return (
|
||||
error instanceof GatewayClientRequestError &&
|
||||
error.gatewayCode === "INVALID_REQUEST" &&
|
||||
error.message.includes("unknown method: node.protocolFeatures.update")
|
||||
);
|
||||
}
|
||||
|
||||
type NodeInvokeSessionEnvelopeMode = "authoritative" | "legacy";
|
||||
|
||||
async function negotiateNodeInvokeSessionEnvelope(
|
||||
client: GatewayClient,
|
||||
): Promise<NodeInvokeSessionEnvelopeMode> {
|
||||
try {
|
||||
await client.request("node.protocolFeatures.update", {
|
||||
features: [NODE_INVOKE_SESSION_KEY_ENVELOPE_PROTOCOL_FEATURE],
|
||||
});
|
||||
return "authoritative";
|
||||
} catch (error) {
|
||||
if (isUnsupportedNodeProtocolFeaturesUpdateError(error)) {
|
||||
return "legacy";
|
||||
}
|
||||
writeStderrLine(`node host protocol feature publish failed: ${String(error)}`);
|
||||
// Only a confirmed unknown-method response enables the legacy nested field.
|
||||
// Other failures keep omitted envelopes fail-closed while the connection lives.
|
||||
return "authoritative";
|
||||
}
|
||||
}
|
||||
|
||||
async function publishNodePluginTools(client: GatewayClient, tools: unknown[]): Promise<void> {
|
||||
try {
|
||||
await client.request("node.pluginTools.update", { tools });
|
||||
@@ -255,38 +224,6 @@ export async function runNodeHost(opts: NodeHostRunOptions): Promise<void> {
|
||||
const url = `${scheme}://${urlHost}:${port}${contextPath}`;
|
||||
let inventory: NodeHostInventory = preparedRuntime.initialInventory;
|
||||
let gatewayHelloReceived = false;
|
||||
let gatewayConnectionGeneration = 0;
|
||||
let nodeInvokeSessionEnvelopeMode =
|
||||
Promise.resolve<NodeInvokeSessionEnvelopeMode>("authoritative");
|
||||
let nodeInvokeSessionEnvelopeNegotiationComplete = true;
|
||||
const nodeInvokeEventDispatchByInvokeId = new Map<string, Promise<void>>();
|
||||
// Cancellation can arrive before the queued request dispatches. Mark it immediately
|
||||
// so the request cannot start before its queued cancel runs.
|
||||
const queuedNodeInvokeCancellations = new Set<string>();
|
||||
const queueNodeInvokeEvent = (
|
||||
invokeId: string,
|
||||
dispatch: (mode: NodeInvokeSessionEnvelopeMode) => void,
|
||||
envelopeMode: Promise<NodeInvokeSessionEnvelopeMode> = Promise.resolve("authoritative"),
|
||||
): void => {
|
||||
const connectionGeneration = gatewayConnectionGeneration;
|
||||
const previous = nodeInvokeEventDispatchByInvokeId.get(invokeId) ?? Promise.resolve();
|
||||
const queued = previous
|
||||
.then(async () => {
|
||||
const mode = await envelopeMode;
|
||||
if (connectionGeneration === gatewayConnectionGeneration) {
|
||||
dispatch(mode);
|
||||
}
|
||||
})
|
||||
.catch((error: unknown) => {
|
||||
writeStderrLine(`node host invoke event dispatch failed: ${String(error)}`);
|
||||
});
|
||||
nodeInvokeEventDispatchByInvokeId.set(invokeId, queued);
|
||||
void queued.then(() => {
|
||||
if (nodeInvokeEventDispatchByInvokeId.get(invokeId) === queued) {
|
||||
nodeInvokeEventDispatchByInvokeId.delete(invokeId);
|
||||
}
|
||||
});
|
||||
};
|
||||
|
||||
const publishInventory = () => {
|
||||
if (!gatewayHelloReceived) {
|
||||
@@ -323,20 +260,14 @@ export async function runNodeHost(opts: NodeHostRunOptions): Promise<void> {
|
||||
if (evt.event === "node.invoke.cancel") {
|
||||
const payload = coerceNodeInvokeCancelPayload(evt.payload);
|
||||
if (payload) {
|
||||
queuedNodeInvokeCancellations.add(payload.invokeId);
|
||||
queueNodeInvokeEvent(payload.invokeId, () => {
|
||||
activeRuntime.cancel(payload.invokeId);
|
||||
queuedNodeInvokeCancellations.delete(payload.invokeId);
|
||||
});
|
||||
activeRuntime.cancel(payload.invokeId);
|
||||
}
|
||||
return;
|
||||
}
|
||||
if (evt.event === "node.invoke.input") {
|
||||
const payload = coerceNodeInvokeInputPayload(evt.payload);
|
||||
if (payload) {
|
||||
queueNodeInvokeEvent(payload.invokeId, () => {
|
||||
activeRuntime.handleInput(payload.invokeId, payload.seq, payload.payloadJSON);
|
||||
});
|
||||
activeRuntime.handleInput(payload.invokeId, payload.seq, payload.payloadJSON);
|
||||
}
|
||||
return;
|
||||
}
|
||||
@@ -347,56 +278,11 @@ export async function runNodeHost(opts: NodeHostRunOptions): Promise<void> {
|
||||
if (!payload) {
|
||||
return;
|
||||
}
|
||||
const receivedAtMs = Date.now();
|
||||
const hasSessionKeyEnvelope = Object.hasOwn(payload, "sessionKey");
|
||||
// Omitted envelopes received before negotiation completes still use the legacy
|
||||
// nested session field. Do not reinterpret them after the response arrives.
|
||||
const envelopeModeAtReceipt = hasSessionKeyEnvelope
|
||||
? Promise.resolve<NodeInvokeSessionEnvelopeMode>("authoritative")
|
||||
: nodeInvokeSessionEnvelopeNegotiationComplete
|
||||
? nodeInvokeSessionEnvelopeMode
|
||||
: Promise.resolve<NodeInvokeSessionEnvelopeMode>("legacy");
|
||||
queueNodeInvokeEvent(
|
||||
payload.id,
|
||||
(mode) => {
|
||||
if (queuedNodeInvokeCancellations.delete(payload.id)) {
|
||||
return;
|
||||
}
|
||||
// Older gateways may send non-empty attribution before negotiation.
|
||||
// Preserve that envelope while still upgrading omitted negotiated requests to a clear.
|
||||
let invokePayload: NodeInvokeRequestPayload =
|
||||
mode === "authoritative" && !hasSessionKeyEnvelope
|
||||
? { ...payload, sessionKey: null }
|
||||
: payload;
|
||||
if (typeof invokePayload.timeoutMs === "number" && invokePayload.timeoutMs > 0) {
|
||||
// The Gateway sends its remaining deadline budget. Charge negotiation
|
||||
// time here so delayed state-changing commands cannot run after expiry.
|
||||
const elapsedMs = Math.max(0, Date.now() - receivedAtMs);
|
||||
const remainingTimeoutMs = Math.max(0, invokePayload.timeoutMs - elapsedMs);
|
||||
if (remainingTimeoutMs === 0) {
|
||||
return;
|
||||
}
|
||||
invokePayload = { ...invokePayload, timeoutMs: remainingTimeoutMs };
|
||||
}
|
||||
void activeRuntime.invoke(invokePayload);
|
||||
},
|
||||
envelopeModeAtReceipt,
|
||||
);
|
||||
void activeRuntime.invoke(payload);
|
||||
},
|
||||
onHelloOk: () => {
|
||||
writeStderrLine(`node host gateway connected: ${url}`);
|
||||
gatewayConnectionGeneration += 1;
|
||||
const connectionGeneration = gatewayConnectionGeneration;
|
||||
nodeInvokeEventDispatchByInvokeId.clear();
|
||||
queuedNodeInvokeCancellations.clear();
|
||||
gatewayHelloReceived = true;
|
||||
nodeInvokeSessionEnvelopeNegotiationComplete = false;
|
||||
nodeInvokeSessionEnvelopeMode = negotiateNodeInvokeSessionEnvelope(client).then((mode) => {
|
||||
if (connectionGeneration === gatewayConnectionGeneration) {
|
||||
nodeInvokeSessionEnvelopeNegotiationComplete = true;
|
||||
}
|
||||
return mode;
|
||||
});
|
||||
publishInventory();
|
||||
},
|
||||
onConnectError: (err) => {
|
||||
@@ -414,12 +300,7 @@ export async function runNodeHost(opts: NodeHostRunOptions): Promise<void> {
|
||||
});
|
||||
},
|
||||
onClose: (code, reason) => {
|
||||
gatewayConnectionGeneration += 1;
|
||||
nodeInvokeEventDispatchByInvokeId.clear();
|
||||
queuedNodeInvokeCancellations.clear();
|
||||
gatewayHelloReceived = false;
|
||||
nodeInvokeSessionEnvelopeMode = Promise.resolve("authoritative");
|
||||
nodeInvokeSessionEnvelopeNegotiationComplete = true;
|
||||
activeRuntime.cancelAll();
|
||||
writeStderrLine(`node host gateway closed (${code}): ${reason}`);
|
||||
},
|
||||
|
||||
Reference in New Issue
Block a user