Skip to content

Commit e7538b4

Browse files
vincentkocsteipete
vincentkoc
authored and
steipete
committed
perf(qa): drop per-rpc gateway cli forks
1 parent 02bd9e8 commit e7538b4

4 files changed

Lines changed: 254 additions & 52 deletions

File tree

‎extensions/qa-lab/src/gateway-child.ts‎

Lines changed: 9 additions & 52 deletions
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,7 @@ import net from "node:net";
66
import os from "node:os";
77
import path from "node:path";
88
import { setTimeout as sleep } from "node:timers/promises";
9-
import { formatErrorMessage } from "openclaw/plugin-sdk/error-runtime";
9+
import { startQaGatewayRpcClient } from "./gateway-rpc-client.js";
1010
import { seedQaAgentWorkspace } from "./qa-agent-workspace.js";
1111
import { buildQaGatewayConfig } from "./qa-gateway-config.js";
1212

@@ -157,34 +157,6 @@ async function waitForGatewayReady(params: {
157157
throw new Error(`gateway failed to become healthy:\n${params.logs()}`);
158158
}
159159

160-
async function runCliJson(params: { cwd: string; env: NodeJS.ProcessEnv; args: string[] }) {
161-
const stdout: Buffer[] = [];
162-
const stderr: Buffer[] = [];
163-
await new Promise<void>((resolve, reject) => {
164-
const child = spawn(process.execPath, params.args, {
165-
cwd: params.cwd,
166-
env: params.env,
167-
stdio: ["ignore", "pipe", "pipe"],
168-
});
169-
child.stdout.on("data", (chunk) => stdout.push(Buffer.from(chunk)));
170-
child.stderr.on("data", (chunk) => stderr.push(Buffer.from(chunk)));
171-
child.once("error", reject);
172-
child.once("exit", (code) => {
173-
if (code === 0) {
174-
resolve();
175-
return;
176-
}
177-
reject(
178-
new Error(
179-
`gateway cli failed (${code ?? "unknown"}): ${Buffer.concat(stderr).toString("utf8")}`,
180-
),
181-
);
182-
});
183-
});
184-
const text = Buffer.concat(stdout).toString("utf8").trim();
185-
return text ? (JSON.parse(text) as unknown) : {};
186-
}
187-
188160
export function resolveQaControlUiRoot(params: { repoRoot: string; controlUiEnabled?: boolean }) {
189161
if (params.controlUiEnabled === false) {
190162
return undefined;
@@ -288,12 +260,18 @@ export async function startQaGatewayChild(params: {
288260
`${Buffer.concat(stdout).toString("utf8")}\n${Buffer.concat(stderr).toString("utf8")}`.trim();
289261
const keepTemp = process.env.OPENCLAW_QA_KEEP_TEMP === "1";
290262

263+
let rpcClient;
291264
try {
292265
await waitForGatewayReady({
293266
baseUrl,
294267
logs,
295268
child,
296269
});
270+
rpcClient = await startQaGatewayRpcClient({
271+
wsUrl,
272+
token: gatewayToken,
273+
logs,
274+
});
297275
} catch (error) {
298276
child.kill("SIGTERM");
299277
throw error;
@@ -314,31 +292,10 @@ export async function startQaGatewayChild(params: {
314292
rpcParams?: unknown,
315293
opts?: { expectFinal?: boolean; timeoutMs?: number },
316294
) {
317-
return await runCliJson({
318-
cwd: runtimeCwd,
319-
env,
320-
args: [
321-
distEntryPath,
322-
"gateway",
323-
"call",
324-
method,
325-
"--url",
326-
wsUrl,
327-
"--token",
328-
gatewayToken,
329-
"--json",
330-
"--timeout",
331-
String(opts?.timeoutMs ?? 20_000),
332-
...(opts?.expectFinal ? ["--expect-final"] : []),
333-
"--params",
334-
JSON.stringify(rpcParams ?? {}),
335-
],
336-
}).catch((error) => {
337-
const details = formatErrorMessage(error);
338-
throw new Error(`${details}\nGateway logs:\n${logs()}`);
339-
});
295+
return await rpcClient.request(method, rpcParams, opts);
340296
},
341297
async stop(opts?: { keepTemp?: boolean }) {
298+
await rpcClient.stop().catch(() => {});
342299
if (!child.killed) {
343300
child.kill("SIGTERM");
344301
await Promise.race([

‎extensions/qa-lab/src/gateway-rpc-client.test.ts‎

Lines changed: 143 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,143 @@
1+
import { beforeEach, describe, expect, it, vi } from "vitest";
2+
3+
const gatewayClientMock = vi.hoisted(() => {
4+
const request = vi.fn(async (_method?: string, _params?: unknown, _opts?: unknown) => ({
5+
ok: true,
6+
}));
7+
const stopAndWait = vi.fn(async () => {});
8+
const stop = vi.fn();
9+
const constructorCalls: Array<Record<string, unknown>> = [];
10+
let startMode: "hello" | "connect-error" = "hello";
11+
12+
class MockGatewayClient {
13+
private readonly options: Record<string, unknown>;
14+
15+
constructor(options: Record<string, unknown>) {
16+
this.options = options;
17+
constructorCalls.push(options);
18+
}
19+
20+
start() {
21+
queueMicrotask(() => {
22+
if (startMode === "connect-error") {
23+
const onConnectError = this.options.onConnectError;
24+
if (typeof onConnectError === "function") {
25+
onConnectError(new Error("connect boom"));
26+
}
27+
return;
28+
}
29+
const onHelloOk = this.options.onHelloOk;
30+
if (typeof onHelloOk === "function") {
31+
onHelloOk({});
32+
}
33+
});
34+
}
35+
36+
async request(method: string, params?: unknown, opts?: unknown) {
37+
return await request(method, params, opts);
38+
}
39+
40+
async stopAndWait() {
41+
await stopAndWait();
42+
}
43+
44+
stop() {
45+
stop();
46+
}
47+
}
48+
49+
return {
50+
MockGatewayClient,
51+
request,
52+
stopAndWait,
53+
stop,
54+
constructorCalls,
55+
reset() {
56+
request.mockReset().mockResolvedValue({ ok: true });
57+
stopAndWait.mockReset().mockResolvedValue(undefined);
58+
stop.mockReset();
59+
constructorCalls.splice(0, constructorCalls.length);
60+
startMode = "hello";
61+
},
62+
setStartMode(mode: "hello" | "connect-error") {
63+
startMode = mode;
64+
},
65+
};
66+
});
67+
68+
vi.mock("./runtime-api.js", () => ({
69+
GatewayClient: gatewayClientMock.MockGatewayClient,
70+
}));
71+
72+
import { startQaGatewayRpcClient } from "./gateway-rpc-client.js";
73+
74+
describe("startQaGatewayRpcClient", () => {
75+
beforeEach(() => {
76+
gatewayClientMock.reset();
77+
});
78+
79+
it("starts a gateway client without device identity and forwards requests", async () => {
80+
const client = await startQaGatewayRpcClient({
81+
wsUrl: "ws://127.0.0.1:18789",
82+
token: "qa-token",
83+
logs: () => "qa logs",
84+
});
85+
86+
expect(gatewayClientMock.constructorCalls[0]).toEqual(
87+
expect.objectContaining({
88+
url: "ws://127.0.0.1:18789",
89+
token: "qa-token",
90+
deviceIdentity: null,
91+
scopes: [
92+
"operator.admin",
93+
"operator.read",
94+
"operator.write",
95+
"operator.approvals",
96+
"operator.pairing",
97+
"operator.talk.secrets",
98+
],
99+
}),
100+
);
101+
102+
await expect(
103+
client.request("agent.run", { prompt: "hi" }, { expectFinal: true, timeoutMs: 45_000 }),
104+
).resolves.toEqual({ ok: true });
105+
106+
expect(gatewayClientMock.request).toHaveBeenCalledWith(
107+
"agent.run",
108+
{ prompt: "hi" },
109+
{
110+
expectFinal: true,
111+
timeoutMs: 45_000,
112+
},
113+
);
114+
115+
await client.stop();
116+
expect(gatewayClientMock.stopAndWait).toHaveBeenCalledTimes(1);
117+
});
118+
119+
it("wraps request failures with gateway logs", async () => {
120+
gatewayClientMock.request.mockRejectedValueOnce(new Error("gateway not connected"));
121+
const client = await startQaGatewayRpcClient({
122+
wsUrl: "ws://127.0.0.1:18789",
123+
token: "qa-token",
124+
logs: () => "qa logs",
125+
});
126+
127+
await expect(client.request("health")).rejects.toThrow(
128+
"gateway not connected\nGateway logs:\nqa logs",
129+
);
130+
});
131+
132+
it("wraps connect failures with gateway logs", async () => {
133+
gatewayClientMock.setStartMode("connect-error");
134+
135+
await expect(
136+
startQaGatewayRpcClient({
137+
wsUrl: "ws://127.0.0.1:18789",
138+
token: "qa-token",
139+
logs: () => "qa logs",
140+
}),
141+
).rejects.toThrow("connect boom\nGateway logs:\nqa logs");
142+
});
143+
});

‎extensions/qa-lab/src/gateway-rpc-client.ts‎

Lines changed: 101 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,101 @@
1+
import { formatErrorMessage } from "openclaw/plugin-sdk/error-runtime";
2+
import { GatewayClient } from "./runtime-api.js";
3+
4+
type QaGatewayRpcRequestOptions = {
5+
expectFinal?: boolean;
6+
timeoutMs?: number;
7+
};
8+
9+
const QA_GATEWAY_RPC_SCOPES = [
10+
"operator.admin",
11+
"operator.read",
12+
"operator.write",
13+
"operator.approvals",
14+
"operator.pairing",
15+
"operator.talk.secrets",
16+
] as const;
17+
18+
export type QaGatewayRpcClient = {
19+
request(method: string, rpcParams?: unknown, opts?: QaGatewayRpcRequestOptions): Promise<unknown>;
20+
stop(): Promise<void>;
21+
};
22+
23+
function formatQaGatewayRpcError(error: unknown, logs: () => string) {
24+
const details = formatErrorMessage(error);
25+
return new Error(`${details}\nGateway logs:\n${logs()}`);
26+
}
27+
28+
export async function startQaGatewayRpcClient(params: {
29+
wsUrl: string;
30+
token: string;
31+
logs: () => string;
32+
}): Promise<QaGatewayRpcClient> {
33+
let readySettled = false;
34+
let stopping = false;
35+
let resolveReady!: () => void;
36+
let rejectReady!: (err: unknown) => void;
37+
38+
const ready = new Promise<void>((resolve, reject) => {
39+
resolveReady = resolve;
40+
rejectReady = reject;
41+
});
42+
43+
const settleReady = (error?: unknown) => {
44+
if (readySettled) {
45+
return;
46+
}
47+
readySettled = true;
48+
if (error) {
49+
rejectReady(error);
50+
return;
51+
}
52+
resolveReady();
53+
};
54+
55+
const wrapError = (error: unknown) => formatQaGatewayRpcError(error, params.logs);
56+
57+
const client = new GatewayClient({
58+
url: params.wsUrl,
59+
token: params.token,
60+
deviceIdentity: null,
61+
// Mirror the old gateway CLI caller scopes so the faster path stays behavior-identical.
62+
scopes: [...QA_GATEWAY_RPC_SCOPES],
63+
onHelloOk: () => {
64+
settleReady();
65+
},
66+
onConnectError: (error) => {
67+
settleReady(wrapError(error));
68+
},
69+
onClose: (code, reason) => {
70+
if (stopping) {
71+
return;
72+
}
73+
const reasonText = reason.trim() || "no close reason";
74+
settleReady(wrapError(new Error(`gateway closed (${code}): ${reasonText}`)));
75+
},
76+
});
77+
78+
client.start();
79+
await ready;
80+
81+
return {
82+
async request(method, rpcParams, opts) {
83+
try {
84+
return await client.request(method, rpcParams, {
85+
expectFinal: opts?.expectFinal,
86+
timeoutMs: opts?.timeoutMs ?? 20_000,
87+
});
88+
} catch (error) {
89+
throw wrapError(error);
90+
}
91+
},
92+
async stop() {
93+
stopping = true;
94+
try {
95+
await client.stopAndWait();
96+
} catch {
97+
client.stop();
98+
}
99+
},
100+
};
101+
}

‎extensions/qa-lab/src/runtime-api.ts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
export type { Command } from "commander";
22
export type { OpenClawConfig, PluginRuntime } from "openclaw/plugin-sdk/core";
33
export { definePluginEntry } from "openclaw/plugin-sdk/core";
4+
export { GatewayClient } from "openclaw/plugin-sdk/gateway-runtime";
45
export {
56
buildQaTarget,
67
createQaBusThread,

0 commit comments

Comments
 (0)