import { spawn, type ChildProcess } from "node:child_process"; import { randomUUID } from "node:crypto"; import { once } from "node:events"; import path from "node:path"; import { createInterface } from "node:readline"; import { setTimeout as delay } from "node:timers/promises"; import { isRecord } from "@openclaw/normalization-core/record-coerce"; import { expect, vi } from "vitest"; import type { startQaGatewayChild } from "../../../../extensions/qa-lab/api.js"; import type { NodePluginToolDescriptor } from "../../../../packages/gateway-protocol/src/schema/nodes.js"; import type { McpServerConfig } from "../../../../src/config/types.mcp.js"; import { signalProcessTree } from "../../../../src/process/kill-tree.js"; export const TEST_TIMEOUT_MS = 180_000; const WAIT_TIMEOUT_MS = 30_000; const FIXTURE_READY_TYPE = "openclaw-mcp-parity-ready"; const NODE_DISPLAY_NAME = "QA MCP parity node"; export const NODE_MCP_COMMAND = "mcp.tools.call.v1"; export const WAIT_OPTIONS = { timeout: WAIT_TIMEOUT_MS, interval: 100 }; export const MCP_SERVERS = ["sse", "stdio", "streamableHttp"] as const; const MCP_LABELS = { sse: "sse", stdio: "stdio", streamableHttp: "streamable-http" } as const; type McpServerName = keyof typeof MCP_LABELS; export type GatewayHandle = Awaited>; export type CapturedChild = { child: ChildProcess; exited: Promise; logs: () => string; signalTree: (signal: "SIGTERM" | "SIGKILL") => Promise; }; export type HttpFixture = CapturedChild & { pid: number; urls: { streamableHttp: string; sse: string; }; }; type NodeRead = { nodeId: string; displayName?: string; approvalState?: string; paired?: boolean; connected?: boolean; nodePluginTools?: NodePluginToolDescriptor[]; }; export type ProbeResult = { label: string; marker: string; pid: number; }; type EffectiveTool = { id?: string; source?: string; pluginId?: string; }; export type ToolsEffectiveResult = { groups?: Array<{ tools?: EffectiveTool[] }>; }; const CHILD_ENV_KEYS = ["PATH", "PATHEXT", "SystemRoot", "WINDIR", "ComSpec"] as const; function captureChild(child: ChildProcess): CapturedChild { let stdout = ""; let stderr = ""; child.stdout?.on("data", (chunk: Buffer) => { stdout = (stdout + chunk.toString()).slice(-200_000); }); child.stderr?.on("data", (chunk: Buffer) => { stderr = (stderr + chunk.toString()).slice(-200_000); }); const pid = child.pid; let termPromise: Promise | undefined; let killPromise: Promise | undefined; const signalTree = (signal: "SIGTERM" | "SIGKILL") => { const existing = signal === "SIGKILL" ? killPromise : termPromise; if (existing) { return existing; } const signaled = new Promise((resolve) => { if (pid === undefined) { resolve(); return; } signalProcessTree(pid, signal, { detached: process.platform !== "win32", onComplete: resolve, }); }); if (signal === "SIGKILL") { killPromise = signaled; } else { termPromise = signaled; } return signaled; }; const exited = once(child, "exit").then(async () => { // The root PID still identifies this task-owned tree at exit delivery. Reap // descendants before any retained numeric process-group authority can age. await signalTree("SIGKILL"); }); return { child, exited, logs: () => `stdout:\n${stdout}\nstderr:\n${stderr}`, signalTree, }; } export function createChildEnv(params: { home: string; tempDir: string; extra?: Record; }): Record { const env: Record = {}; for (const key of CHILD_ENV_KEYS) { const value = process.env[key]; if (value !== undefined) { env[key] = value; } } return { ...env, HOME: params.home, USERPROFILE: params.home, XDG_CACHE_HOME: path.join(params.home, ".cache"), XDG_CONFIG_HOME: path.join(params.home, ".config"), XDG_DATA_HOME: path.join(params.home, ".local", "share"), TMPDIR: params.tempDir, TMP: params.tempDir, TEMP: params.tempDir, ...params.extra, }; } export async function startHttpFixture(params: { fixturePath: string; labelPrefix: "session" | "node"; env: NodeJS.ProcessEnv; }): Promise { const captured = captureChild( spawn(process.execPath, [params.fixturePath, "http", "--label-prefix", params.labelPrefix], { cwd: process.cwd(), detached: process.platform !== "win32", env: params.env, stdio: ["ignore", "pipe", "pipe"], }), ); let transferred = false; let lines: ReturnType | undefined; try { const pid = captured.child.pid; if (pid === undefined) { throw new Error("HTTP MCP fixture did not start"); } if (!captured.child.stdout) { throw new Error("HTTP MCP fixture stdout was not piped"); } lines = createInterface({ input: captured.child.stdout }); const line = await Promise.race([ new Promise((resolve) => { lines?.once("line", resolve); }), captured.exited.then(() => { throw new Error(`HTTP MCP fixture exited before readiness:\n${captured.logs()}`); }), delay(WAIT_TIMEOUT_MS, undefined, { ref: false }).then(() => { throw new Error(`HTTP MCP fixture readiness timed out:\n${captured.logs()}`); }), ]); const value: unknown = JSON.parse(line); if ( !isRecord(value) || value.type !== FIXTURE_READY_TYPE || !isRecord(value.urls) || typeof value.urls.streamableHttp !== "string" || typeof value.urls.sse !== "string" ) { throw new Error(`HTTP MCP fixture returned invalid readiness: ${line}`); } transferred = true; return { ...captured, pid, urls: { streamableHttp: value.urls.streamableHttp, sse: value.urls.sse }, }; } finally { lines?.close(); if (!transferred) { await stopChild(captured); } } } export function startNodeProcess(gatewayPort: number, nodeEnv: NodeJS.ProcessEnv): CapturedChild { return captureChild( spawn( process.execPath, [ "dist/index.js", "node", "run", "--host", "127.0.0.1", "--port", String(gatewayPort), "--display-name", NODE_DISPLAY_NAME, ], { cwd: process.cwd(), detached: process.platform !== "win32", env: nodeEnv, stdio: ["ignore", "pipe", "pipe"], }, ), ); } export async function stopChild(captured: CapturedChild | undefined): Promise { if (!captured || captured.child.exitCode !== null || captured.child.signalCode !== null) { await captured?.exited.catch(() => {}); return; } await captured.signalTree("SIGTERM"); const graceful = await Promise.race([ captured.exited.then(() => true), delay(10_000, false, { ref: false }), ]); if (!graceful) { await captured.signalTree("SIGKILL"); } await captured.exited; } export function processIsAlive(pid: number): boolean { try { process.kill(pid, 0); return true; } catch (error) { return (error as NodeJS.ErrnoException).code !== "ESRCH"; } } export async function waitForProcessExit(pid: number): Promise { await vi.waitFor(() => expect(processIsAlive(pid)).toBe(false), WAIT_OPTIONS); } export async function approvePairing( gateway: GatewayHandle, kind: "device" | "node", nodeId?: string, ): Promise { let requestId = ""; let approvedId = ""; await vi.waitFor(async () => { const result = (await gateway.call(`${kind}.pair.list`, {})) as { pending?: Array<{ requestId?: string; deviceId?: string; nodeId?: string; role?: string; }>; }; const pending = result.pending?.find((entry) => kind === "device" ? entry.role === "node" : entry.nodeId === nodeId, ); expect(pending?.requestId, gateway.logs()).toBeTruthy(); requestId = pending?.requestId ?? ""; approvedId = kind === "device" ? (pending?.deviceId ?? "") : (pending?.nodeId ?? ""); expect(approvedId, gateway.logs()).toBeTruthy(); }, WAIT_OPTIONS); await gateway.call(`${kind}.pair.approve`, { requestId }); return approvedId; } export async function readNode( gateway: GatewayHandle, nodeId: string, ): Promise { const result = (await gateway.call("node.list", {})) as { nodes?: NodeRead[] }; return result.nodes?.find((entry) => entry.nodeId === nodeId); } export async function waitForNode( gateway: GatewayHandle, nodeId: string, toolCount?: number, ): Promise { let node: NodeRead | undefined; await vi.waitFor(async () => { node = await readNode(gateway, nodeId); expect(node, gateway.logs()).toMatchObject({ nodeId, displayName: NODE_DISPLAY_NAME, approvalState: "approved", paired: true, connected: true, }); if (toolCount !== undefined) { expect(node?.nodePluginTools ?? [], gateway.logs()).toHaveLength(toolCount); } }, WAIT_OPTIONS); if (!node) { throw new Error(`node ${nodeId} did not connect:\n${gateway.logs()}`); } return node; } export function parseNodeMcpTextRecord(value: unknown): Record { const payload = isRecord(value) && isRecord(value.payload) ? value.payload : value; if (!isRecord(payload) || !Array.isArray(payload.content)) { throw new Error(`MCP result omitted content: ${JSON.stringify(value)}`); } const text = payload.content.find( (item) => isRecord(item) && item.type === "text" && typeof item.text === "string", ); if (!isRecord(text) || typeof text.text !== "string") { throw new Error(`MCP result omitted text: ${JSON.stringify(value)}`); } const parsed: unknown = JSON.parse(text.text); if (!isRecord(parsed)) { throw new Error(`MCP text was not an object: ${text.text}`); } return parsed; } export function parseProbeResult(value: unknown): ProbeResult { const parsed = parseNodeMcpTextRecord(value); if ( typeof parsed.label !== "string" || typeof parsed.marker !== "string" || typeof parsed.pid !== "number" ) { throw new Error(`MCP result had invalid probe data: ${JSON.stringify(parsed)}`); } return { label: parsed.label, marker: parsed.marker, pid: parsed.pid }; } export async function invokeNodeMcpPayload(params: { gateway: GatewayHandle; nodeId: string; descriptor: NodePluginToolDescriptor; marker: string; }): Promise { const mcp = params.descriptor.mcp; if (!mcp) { throw new Error(`node descriptor ${params.descriptor.name} omitted MCP routing`); } return await params.gateway.call( "node.invoke", { nodeId: params.nodeId, command: NODE_MCP_COMMAND, params: { server: mcp.server, tool: mcp.tool, arguments: { marker: params.marker } }, timeoutMs: WAIT_TIMEOUT_MS, idempotencyKey: randomUUID(), }, { timeoutMs: WAIT_TIMEOUT_MS }, ); } export async function invokeNodeMcp( params: Parameters[0], ): Promise { return parseProbeResult(await invokeNodeMcpPayload(params)); } export function flattenEffectiveTools(result: ToolsEffectiveResult): EffectiveTool[] { return (result.groups ?? []).flatMap((group) => group.tools ?? []); } export function createMcpServers(params: { placement: "session" | "node"; fixture: HttpFixture; stdioEnv: Record; fixturePath: string; repoRoot: string; }): Record { const common = { connectionTimeoutMs: WAIT_TIMEOUT_MS, requestTimeoutMs: WAIT_TIMEOUT_MS, toolFilter: { include: ["parity_*"], exclude: ["*_hidden"] }, }; return { stdio: { ...common, transport: "stdio", command: process.execPath, args: [params.fixturePath, "stdio", "--label", `${params.placement}-stdio`], cwd: params.repoRoot, env: params.stdioEnv, }, streamableHttp: { ...common, transport: "streamable-http", url: params.fixture.urls.streamableHttp, }, sse: { ...common, transport: "sse", url: params.fixture.urls.sse, }, }; } export function expectedProbeResults( placement: "session" | "node", markerPrefix: string, httpPid: number, servers: readonly McpServerName[] = MCP_SERVERS, ) { return Object.fromEntries( servers.map((server) => [ server, { label: `${placement}-${MCP_LABELS[server]}`, marker: `${markerPrefix}-${server}`, pid: server === "stdio" ? expect.any(Number) : httpPid, }, ]), ); }