diff --git a/qa/scenarios/runtime/paired-node-worker-lifecycle-wire.yaml b/qa/scenarios/runtime/paired-node-worker-lifecycle-wire.yaml new file mode 100644 index 000000000000..a4bfb55cf8b9 --- /dev/null +++ b/qa/scenarios/runtime/paired-node-worker-lifecycle-wire.yaml @@ -0,0 +1,32 @@ +title: Paired-device worker lifecycle wire + +scenario: + id: paired-node-worker-lifecycle-wire + surface: gateway + category: gateway.nodes-and-remote-capabilities + coverage: + secondary: + - gateway.node-inventory + - gateway.session-apis-session-status + objective: Prove paired-device worker faults stay retryable and isolated from Gateway-local work across the real signed node and worker-child wire. + successCriteria: + - A signed paired node handles private worker invocations through the real bundle installer, supervisor, workspace runtime, and worker child while a Gateway-local session remains usable. + - Bundle retention and status capability v1 expose only installed version or missing status through node.list and environments.list, then reinstall an exactly deleted proof bundle after same-identity reconnect. + - A pre-turn node disconnect produces a visible runner-offline recovery message, preserves the active placement for same-identity retry, and does not terminalize local work. + - Two held real worker children exhaust supervisor capacity 2; a third turn records a bounded capacity failure across RPC, history, and placement, then succeeds after one slot releases. + - Node-role removal reconciles the owned environment and placement before returning, fences the old node connection, and leaves Gateway-local execution available. + docsRefs: + - docs/gateway/protocol.md + - docs/gateway/cloud-workers.md + - docs/concepts/qa-e2e-automation.md + codeRefs: + - src/gateway/worker-environments/node-workspace-retain-coordinator.ts + - src/gateway/worker-environments/node-launch-adapter.ts + - src/gateway/device-worker-revocation.ts + - src/node-host/node-worker-bundle-installer.ts + - src/node-host/node-worker-supervisor.ts + - test/e2e/qa-lab/runtime/paired-node-worker-lifecycle-wire.e2e.test.ts + execution: + kind: vitest + path: test/e2e/qa-lab/runtime/paired-node-worker-lifecycle-wire.e2e.test.ts + summary: Start a real isolated Gateway, pair a signed node host, then prove bundle repair, offline retry, physical capacity recovery, role-removal fencing, and local-session isolation over the real worker wire. diff --git a/src/agents/failover-error.test.ts b/src/agents/failover-error.test.ts index 13b786eac97f..088aaa9d5445 100644 --- a/src/agents/failover-error.test.ts +++ b/src/agents/failover-error.test.ts @@ -829,10 +829,16 @@ describe("failover-error", () => { ).toBe(true); }); - it("returns true for direct and nested runner availability failures", () => { - const unavailable = new Error("The device runner is offline"); - unavailable.name = "WorkerRunnerUnavailableError"; - for (const error of [unavailable, new Error("worker turn failed", { cause: unavailable })]) { + it.each([ + ["availability", "WorkerRunnerUnavailableError", "The device runner is offline"], + ["capacity", "WorkerRunnerCapacityError", "device worker capacity remained full"], + ])("returns true for direct and nested runner %s failures", (_label, name, message) => { + const coordination = new Error(message); + coordination.name = name; + for (const error of [ + coordination, + new Error("worker turn failed", { cause: coordination }), + ]) { expect(isNonProviderRuntimeCoordinationError(error)).toBe(true); expect(resolveModelFallbackError(error)).toEqual({ kind: "coordination", error }); } diff --git a/src/agents/failover-error.ts b/src/agents/failover-error.ts index 51bc6f822767..c1b14d5908b1 100644 --- a/src/agents/failover-error.ts +++ b/src/agents/failover-error.ts @@ -492,8 +492,10 @@ function hasGatewayDrainingFailure(err: unknown): boolean { return errorGraphHasName(err, "GatewayDrainingError"); } -function hasWorkerRunnerUnavailableFailure(err: unknown): boolean { - return errorGraphHasName(err, "WorkerRunnerUnavailableError"); +function hasWorkerRunnerCoordinationFailure(err: unknown): boolean { + return ["WorkerRunnerUnavailableError", "WorkerRunnerCapacityError"].some((name) => + errorGraphHasName(err, name), + ); } function hasDirectProviderFailureIdentity(err: unknown): boolean { @@ -891,7 +893,7 @@ export function resolveModelFallbackError( } // Gateway admission can fail before any provider turn starts. Preserve that // identity through wrappers and aggregates so fallback cannot blame a model. - if (hasGatewayDrainingFailure(err) || hasWorkerRunnerUnavailableFailure(err)) { + if (hasGatewayDrainingFailure(err) || hasWorkerRunnerCoordinationFailure(err)) { return { kind: "coordination", error: err }; } const staleLifecycleFailure = hasStaleAgentRunLifecycleFailure(err); diff --git a/src/agents/model-fallback.test.ts b/src/agents/model-fallback.test.ts index 87355d42151d..8eeef2c79800 100644 --- a/src/agents/model-fallback.test.ts +++ b/src/agents/model-fallback.test.ts @@ -2248,6 +2248,46 @@ describe("runWithModelFallback", () => { expect(onFallbackStep).not.toHaveBeenCalled(); }); + it.each([ + [ + "direct", + () => + Object.assign(new Error("device worker capacity remained full"), { + name: "WorkerRunnerCapacityError", + }), + ], + [ + "wrapped", + () => + new Error("worker turn failed", { + cause: Object.assign(new Error("device worker capacity remained full"), { + name: "WorkerRunnerCapacityError", + }), + }), + ], + ])("aborts fallback on %s device capacity failures", async (_label, makeError) => { + const error = makeError(); + const run = vi.fn().mockRejectedValueOnce(error).mockResolvedValueOnce("too late"); + const onError = vi.fn(); + const onFallbackStep = vi.fn(); + + await expect( + runWithModelFallback({ + cfg: undefined, + provider: "openai", + model: "gpt-5.6-sol", + fallbacksOverride: ["openai/gpt-5.4-mini"], + skipAuthProfileRuntime: true, + run, + onError, + onFallbackStep, + }), + ).rejects.toBe(error); + expect(run).toHaveBeenCalledTimes(1); + expect(onError).not.toHaveBeenCalled(); + expect(onFallbackStep).not.toHaveBeenCalled(); + }); + it("still advances after a genuine provider rate limit", async () => { const rateLimit = Object.assign(new Error("rate limit exceeded"), { status: 429 }); const run = vi.fn().mockRejectedValueOnce(rateLimit).mockResolvedValueOnce("fallback ok"); diff --git a/src/gateway/server-worker-placement-startup.ts b/src/gateway/server-worker-placement-startup.ts index 1c2ba613ed79..a712274845bb 100644 --- a/src/gateway/server-worker-placement-startup.ts +++ b/src/gateway/server-worker-placement-startup.ts @@ -133,6 +133,12 @@ export function createGatewayWorkerPlacementRuntime(params: GatewayWorkerPlaceme const workspaceConflictHandlers = createWorkerWorkspaceConflictTranscriptHandlers( loadWorkerPlacementSessionRuntimeModule, ); + const nodeWorkspaceRetention = createNodeWorkspaceRetainCoordinator({ + gatewayNamespace: params.gatewayNamespace, + placements: params.placements, + environments: params.environments, + warn: params.warn, + }); const resolveWorkspacePath = async ({ sessionId, sessionKey, @@ -333,6 +339,11 @@ export function createGatewayWorkerPlacementRuntime(params: GatewayWorkerPlaceme } return activePlacement; }, + onActivated: (request) => { + if (request.deviceId) { + void nodeWorkspaceRetention.schedule(request.deviceId); + } + }, runReclaimBarrier: async ({ sessionId, sessionKey, agentId, reclaim }) => { const sessionRuntime = await loadWorkerPlacementSessionRuntimeModule(); const { resolveGatewaySessionStoreTargetWithStore } = sessionRuntime; @@ -409,12 +420,6 @@ export function createGatewayWorkerPlacementRuntime(params: GatewayWorkerPlaceme createSessionEvidenceResolver: createWorkerPlacementSessionEvidenceResolver, warn: params.warn, }); - const nodeWorkspaceRetention = createNodeWorkspaceRetainCoordinator({ - gatewayNamespace: params.gatewayNamespace, - placements: params.placements, - environments: params.environments, - warn: params.warn, - }); const admissionProvider = createWorkerSessionTurnPlacementProvider({ environments: params.environments, placements: params.placements, diff --git a/src/gateway/worker-environments/node-launch-adapter.ts b/src/gateway/worker-environments/node-launch-adapter.ts index 545c248bbc0e..92e6598c4b73 100644 --- a/src/gateway/worker-environments/node-launch-adapter.ts +++ b/src/gateway/worker-environments/node-launch-adapter.ts @@ -17,7 +17,7 @@ import type { NodeWorkerSupervisorNodeProof, NodeWorkerSupervisorTransport, } from "../node-registry-private.js"; -import { WorkerRunnerUnavailableError } from "./tunnel-contract.js"; +import { WorkerRunnerCapacityError, WorkerRunnerUnavailableError } from "./tunnel-contract.js"; const DEFAULT_RPC_TIMEOUT_MS = 30_000; const DEFAULT_POLL_INTERVAL_MS = 250; @@ -307,11 +307,12 @@ export function createNodeWorkerLaunchAdapter(options: NodeWorkerLaunchAdapterOp const result = await raceWithSignal(operation, signal); if (!result.ok) { const code = result.error?.code ?? "UNAVAILABLE"; + if (code === NODE_WORKER_CAPACITY_EXHAUSTED_ERROR_CODE) { + throw new WorkerRunnerCapacityError(); + } throw new NodeWorkerLaunchTransportError( code, - code === NODE_WORKER_CAPACITY_EXHAUSTED_ERROR_CODE - ? "device worker capacity remained full" - : `node worker supervisor invocation failed (${code})`, + `node worker supervisor invocation failed (${code})`, ); } return parseInvokeReceipt(result.payloadJSON); @@ -496,10 +497,7 @@ export function createNodeWorkerLaunchAdapter(options: NodeWorkerLaunchAdapterOp } // The node authors this result only after its durable claim stayed absent. // Transport dispatch is therefore not launch ambiguity and needs no cancel. - if ( - error instanceof NodeWorkerLaunchTransportError && - error.code === NODE_WORKER_CAPACITY_EXHAUSTED_ERROR_CODE - ) { + if (error instanceof WorkerRunnerCapacityError) { throw error; } if (!mayHaveLaunched) { diff --git a/src/gateway/worker-environments/placement-dispatch.ts b/src/gateway/worker-environments/placement-dispatch.ts index 9c4d0ca52e2c..8a06a1d71f51 100644 --- a/src/gateway/worker-environments/placement-dispatch.ts +++ b/src/gateway/worker-environments/placement-dispatch.ts @@ -60,6 +60,7 @@ type WorkerPlacementDispatchOptions = { runLocalBarrier: WorkerLocalDispatchBarrier; runActivationBarrier: WorkerActivationBarrier; runReclaimBarrier: WorkerPlacementReclaimBarrier; + onActivated?: (request: WorkerPlacementDispatchRequest) => void; workspaceOperations: WorkerWorkspaceOperationCoordinator; resolveWorkspacePath: (params: { sessionId: string; @@ -253,6 +254,11 @@ export function createWorkerPlacementDispatchService(options: WorkerPlacementDis return activated; }, }); + try { + options.onActivated?.(request); + } catch { + // Maintenance scheduling cannot overturn a durable placement activation. + } return activePlacement; } catch (error) { try { diff --git a/src/gateway/worker-environments/tunnel-contract.ts b/src/gateway/worker-environments/tunnel-contract.ts index 617a3dccbda1..8fbde0c4bc0d 100644 --- a/src/gateway/worker-environments/tunnel-contract.ts +++ b/src/gateway/worker-environments/tunnel-contract.ts @@ -1,3 +1,4 @@ +import { NODE_WORKER_CAPACITY_EXHAUSTED_ERROR_CODE } from "../../infra/node-commands.js"; import type { SpawnResult } from "../../process/exec.js"; import type { WorkerLaunchPlan } from "../../worker/launch-descriptor.js"; import type { NodeWorkerWorkspaceTransferInput } from "../../worker/node-workspace-transfer-protocol.js"; @@ -26,6 +27,15 @@ export class WorkerRunnerUnavailableError extends Error { } } +export class WorkerRunnerCapacityError extends Error { + readonly code = NODE_WORKER_CAPACITY_EXHAUSTED_ERROR_CODE; + + constructor() { + super("device worker capacity remained full"); + this.name = "WorkerRunnerCapacityError"; + } +} + export type WorkerTunnelRequest = { environmentId: string; ownerEpoch: number; diff --git a/src/gateway/worker-environments/worker-turn-launcher-failure-recovery.test.ts b/src/gateway/worker-environments/worker-turn-launcher-failure-recovery.test.ts index 0ddf75b05649..1aa339629529 100644 --- a/src/gateway/worker-environments/worker-turn-launcher-failure-recovery.test.ts +++ b/src/gateway/worker-environments/worker-turn-launcher-failure-recovery.test.ts @@ -8,7 +8,11 @@ import { makeAgentAssistantMessage } from "../../agents/test-helpers/agent-messa import type { SpawnResult } from "../../process/exec.js"; import { WORKER_PROVIDER_REPLAY_LOCAL_RETRY_MESSAGE } from "../../worker/transcript-message.js"; import type { WorkerSessionPlacementStore } from "./placement-store.js"; -import { WorkerRunnerUnavailableError, type WorkerTunnelHandle } from "./tunnel-contract.js"; +import { + WorkerRunnerCapacityError, + WorkerRunnerUnavailableError, + type WorkerTunnelHandle, +} from "./tunnel-contract.js"; import { ENVIRONMENT_ID, MANIFEST_REF, @@ -237,7 +241,20 @@ describe("worker turn launcher failure recovery", () => { expect(environments.destroy).not.toHaveBeenCalled(); }); - it("keeps the placement active when launch fails before transport dispatch", async () => { + it.each([ + { + name: "offline before transport dispatch", + error: new WorkerRunnerUnavailableError(), + dispatched: false, + expectedMessage: "The device runner is offline", + }, + { + name: "capacity rejection after transport dispatch", + error: new WorkerRunnerCapacityError(), + dispatched: true, + expectedMessage: "device worker capacity remained full", + }, + ])("keeps the placement active after $name", async ({ error, dispatched, expectedMessage }) => { seedActivePlacement(); const teardownStates: string[] = []; const observedPlacements: WorkerSessionPlacementStore = { @@ -272,8 +289,11 @@ describe("worker turn launcher failure recovery", () => { resume: vi.fn(async () => {}), })), runWorkspaceCommand: vi.fn(), - launchTurn: vi.fn(async () => { - throw new WorkerRunnerUnavailableError(); + launchTurn: vi.fn(async (request) => { + if (dispatched) { + request.onDispatchReady?.(); + } + throw error; }), syncWorkspace: vi.fn(async () => { throw new Error("unexpected workspace sync"); @@ -309,10 +329,10 @@ describe("worker turn launcher failure recovery", () => { turn("run-failed"), runLocal, ), - ).rejects.toThrow("The device runner is offline"); + ).rejects.toThrow(expectedMessage); expect(runLocal).not.toHaveBeenCalled(); expect(placements.get(SESSION_ID)).toMatchObject({ state: "active", turnClaim: null }); - expect(acknowledgeCredentialDelivery).not.toHaveBeenCalled(); + expect(acknowledgeCredentialDelivery).toHaveBeenCalledTimes(dispatched ? 1 : 0); expect(stopTunnel).not.toHaveBeenCalled(); expect(destroy).not.toHaveBeenCalled(); expect(teardownStates).toEqual([]); diff --git a/src/gateway/worker-environments/worker-turn-launcher.ts b/src/gateway/worker-environments/worker-turn-launcher.ts index 2fb959dea6d2..7600bb2ade19 100644 --- a/src/gateway/worker-environments/worker-turn-launcher.ts +++ b/src/gateway/worker-environments/worker-turn-launcher.ts @@ -22,7 +22,7 @@ import type { WorkerSessionPlacementStore, WorkerSessionTurnClaim, } from "./placement-store.js"; -import { WorkerRunnerUnavailableError } from "./tunnel-contract.js"; +import { WorkerRunnerCapacityError, WorkerRunnerUnavailableError } from "./tunnel-contract.js"; import { resolveWorkerBrowserLaunchPlan } from "./worker-browser-launch-plan.js"; import { claimWorkerTurn, @@ -527,7 +527,10 @@ export function createWorkerSessionTurnPlacementProvider(options: WorkerTurnLaun await options.recoverPendingWorkspaceResult(placement.environmentId); throw error; } - if (error instanceof WorkerRunnerUnavailableError && !handedOff) { + if ( + error instanceof WorkerRunnerCapacityError || + (error instanceof WorkerRunnerUnavailableError && !handedOff) + ) { await releaseClaimIfOwned(options.placements, turnClaim); throw error; } diff --git a/test/e2e/qa-lab/runtime/node-worker-launch-wire.e2e.test.ts b/test/e2e/qa-lab/runtime/node-worker-launch-wire.e2e.test.ts index de06f785b7a5..0e0289ccfa30 100644 --- a/test/e2e/qa-lab/runtime/node-worker-launch-wire.e2e.test.ts +++ b/test/e2e/qa-lab/runtime/node-worker-launch-wire.e2e.test.ts @@ -1,59 +1,40 @@ import { execFile } from "node:child_process"; import fs from "node:fs/promises"; -import { createServer, type Server } from "node:http"; import path from "node:path"; import { performance } from "node:perf_hooks"; import { promisify } from "node:util"; import { GatewayClient } from "openclaw/plugin-sdk/gateway-runtime"; -import { afterEach, describe, expect, it, vi } from "vitest"; -import { startQaGatewayChild } from "../../../../extensions/qa-lab/api.js"; -import { - GATEWAY_CLIENT_MODES, - GATEWAY_CLIENT_NAMES, -} from "../../../../packages/gateway-protocol/src/client-info.js"; -import type { DeviceIdentity } from "../../../../src/infra/device-identity.js"; -import { loadOrCreateDeviceIdentity } from "../../../../src/infra/device-identity.js"; +import { afterEach, describe, expect, it } from "vitest"; import { NODE_WORKER_BUNDLE_INSTALL_COMMAND, NODE_WORKER_SUPERVISOR_LAUNCH_COMMAND, NODE_WORKER_SUPERVISOR_STATUS_COMMAND, NODE_WORKER_WORKSPACE_EXEC_COMMAND, } from "../../../../src/infra/node-commands.js"; -import { - NODE_RUNNER_INVENTORY_UPDATE_METHOD, - NODE_WORKER_SUPERVISOR_PROTOCOL_FEATURE, -} from "../../../../src/infra/node-runner-inventory.js"; -import { handleInvoke, type NodeInvokeRequestPayload } from "../../../../src/node-host/invoke.js"; -import { NodeWorkerBundleInstaller } from "../../../../src/node-host/node-worker-bundle-installer.js"; -import { createNodeWorkerSupervisor } from "../../../../src/node-host/node-worker-supervisor.js"; -import { NodeWorkerWorkspaceRuntime } from "../../../../src/node-host/node-worker-workspace.js"; -import { VERSION } from "../../../../src/version.js"; import { useAutoCleanupTempDirTracker } from "../../../helpers/temp-dir.js"; import { BASELINE_PROMPT, BASELINE_REPLY, - MODEL_REF, PROOF_TIMEOUT_MS, startMidturnProvider, } from "./cloud-worker-midturn-loss-fixture.js"; +import { + closeWireServer, + connectWireClient, + createPairedNodeWorkerHost, + createPublishedWireWorkspace, + startPairedNodeWorkerGateway, + type PairedNodeWorkerHost, + type WireGateway, + wireMessageText, +} from "./paired-node-worker-wire-fixture.js"; const execFileAsync = promisify(execFile); const SESSION_KEY = "agent:qa:node-worker-launch-wire"; -const NODE_DISPLAY_NAME = "QA Gateway-bundle worker node"; const TEST_TIMEOUT_MS = PROOF_TIMEOUT_MS + 60_000; const CONTROL_PROBE_MAX_MS = 4_000; const FINALIZATION_LOAD_CONCURRENCY = 6; -type Gateway = Awaited>; -type GatewayEvent = { event: string; payload?: unknown }; -type NodeRead = { - nodeId: string; - approvalState?: string; - connected?: boolean; - paired?: boolean; - sessionHost?: boolean; -}; - const tempDirs = useAutoCleanupTempDirTracker(afterEach); async function git(cwd: string, ...args: string[]): Promise { @@ -64,261 +45,6 @@ async function git(cwd: string, ...args: string[]): Promise { return stdout.trim(); } -async function closeServer(server: Server): Promise { - server.closeAllConnections(); - await new Promise((resolve, reject) => { - server.close((error) => (error ? reject(error) : resolve())); - }); -} - -async function createPublishedWorkspace(root: string) { - const source = path.join(root, "source"); - const bare = path.join(root, "repo.git"); - await fs.mkdir(source, { recursive: true }); - await execFileAsync("git", ["init", "--bare", bare]); - await git(source, "init", "-b", "main"); - await git(source, "config", "user.name", "OpenClaw QA"); - await git(source, "config", "user.email", "openclaw-qa@example.invalid"); - await fs.mkdir(path.join(source, "nested")); - await fs.writeFile(path.join(source, "launch-wire.txt"), "local-install launch wire\n"); - await fs.writeFile(path.join(source, "nested", "tracked.txt"), "nested tracked input\n"); - await git(source, "add", "."); - await git(source, "commit", "-m", "initialize node worker launch wire workspace"); - await git(source, "remote", "add", "publish", bare); - await git(source, "push", "publish", "main"); - await git(source, "remote", "remove", "publish"); - await git(bare, "update-server-info"); - - const server = createServer((request, response) => { - void (async () => { - const pathname = decodeURIComponent(new URL(request.url ?? "/", "http://localhost").pathname); - if (!pathname.startsWith("/repo.git/")) { - response.writeHead(404).end(); - return; - } - const candidate = path.resolve(bare, pathname.slice("/repo.git/".length)); - if (candidate !== bare && !candidate.startsWith(`${bare}${path.sep}`)) { - response.writeHead(404).end(); - return; - } - try { - const contents = await fs.readFile(candidate); - response.writeHead(200, { - "content-type": pathname.endsWith("/info/refs") - ? "text/plain; charset=utf-8" - : "application/octet-stream", - "content-length": String(contents.byteLength), - }); - response.end(request.method === "HEAD" ? undefined : contents); - } catch { - response.writeHead(404).end(); - } - })(); - }); - await new Promise((resolve, reject) => { - server.once("error", reject); - server.listen(0, "127.0.0.1", resolve); - }); - const address = server.address(); - if (!address || typeof address === "string") { - throw new Error("published workspace server did not bind"); - } - const origin = `http://127.0.0.1:${address.port}/repo.git`; - await git(source, "remote", "add", "origin", origin); - const commit = await git(source, "rev-parse", "HEAD"); - await git(source, "ls-remote", "--exit-code", origin, "refs/heads/main"); - return { commit, source: await fs.realpath(source), server }; -} - -async function connectClient(params: { - gateway: Gateway; - role: "operator" | "node"; - identity: DeviceIdentity | null; - onEvent?: (event: GatewayEvent) => void; - timeoutMs?: number; -}): Promise { - return await new Promise((resolve, reject) => { - let settled = false; - const finish = (error?: Error) => { - if (settled) { - return; - } - settled = true; - clearTimeout(timeout); - if (error) { - client.stop(); - reject(error); - } else { - resolve(client); - } - }; - const timeout = setTimeout( - () => finish(new Error("Gateway client connection timed out")), - params.timeoutMs ?? 30_000, - ); - timeout.unref(); - const node = params.role === "node"; - const client = new GatewayClient({ - url: params.gateway.wsUrl, - token: params.gateway.token, - env: params.gateway.runtimeEnv, - role: params.role, - clientName: node ? GATEWAY_CLIENT_NAMES.NODE_HOST : GATEWAY_CLIENT_NAMES.GATEWAY_CLIENT, - clientDisplayName: node ? NODE_DISPLAY_NAME : "Node worker launch wire operator", - clientVersion: VERSION, - platform: node ? "macos" : process.platform, - deviceFamily: node ? "Mac" : undefined, - mode: node ? GATEWAY_CLIENT_MODES.NODE : GATEWAY_CLIENT_MODES.BACKEND, - scopes: node ? [] : ["operator.admin", "operator.pairing", "operator.read", "operator.write"], - caps: node ? ["system"] : undefined, - commands: node ? [] : undefined, - deviceIdentity: params.identity, - requestTimeoutMs: PROOF_TIMEOUT_MS, - onEvent: params.onEvent, - onHelloOk: () => finish(), - onConnectError: (error) => finish(error), - onClose: (code, reason) => finish(new Error(`Gateway closed (${code}): ${reason}`)), - }); - client.start(); - }); -} - -function isPairingRequired(error: unknown): boolean { - const details = - error && typeof error === "object" - ? (error as { details?: { code?: unknown } }).details - : undefined; - return details?.code === "PAIRING_REQUIRED" || String(error).includes("PAIRING_REQUIRED"); -} - -async function approvePairing(operator: GatewayClient, nodeId: string): Promise { - let deviceRequestId: string | undefined; - await vi.waitFor( - async () => { - const result = await operator.request<{ - pending?: Array<{ requestId?: string; deviceId?: string; role?: string }>; - }>("device.pair.list", {}); - deviceRequestId = result.pending?.find( - (entry) => entry.deviceId === nodeId || entry.role === "node", - )?.requestId; - expect(deviceRequestId).toBeTruthy(); - }, - { timeout: 30_000, interval: 100 }, - ); - await operator.request("device.pair.approve", { requestId: deviceRequestId }); - - await approveNodePairing(operator, nodeId); -} - -async function approveNodePairing(operator: GatewayClient, nodeId: string): Promise { - let nodeRequestId: string | undefined; - await vi.waitFor( - async () => { - const result = await operator.request<{ - pending?: Array<{ requestId?: string; nodeId?: string }>; - }>("node.pair.list", {}); - nodeRequestId = result.pending?.find((entry) => entry.nodeId === nodeId)?.requestId; - expect(nodeRequestId).toBeTruthy(); - }, - { timeout: 30_000, interval: 100 }, - ); - await operator.request("node.pair.approve", { requestId: nodeRequestId }); -} - -async function ensureNodeApproved(operator: GatewayClient, nodeId: string): Promise { - let approvalState: string | undefined; - await vi.waitFor( - async () => { - const result = await operator.request<{ nodes?: NodeRead[] }>("node.list", {}); - approvalState = result.nodes?.find((node) => node.nodeId === nodeId)?.approvalState; - expect(approvalState).toBeTruthy(); - }, - { timeout: 30_000, interval: 100 }, - ); - if (approvalState !== "approved") { - await approveNodePairing(operator, nodeId); - return true; - } - return false; -} - -async function connectPairedNode(params: { - gateway: Gateway; - operator: GatewayClient; - identity: DeviceIdentity; - bundlePrewarm?: 1; - onEvent: (event: GatewayEvent) => void; -}): Promise { - const connect = () => - connectClient({ - gateway: params.gateway, - role: "node", - identity: params.identity, - onEvent: params.onEvent, - }); - let client: GatewayClient; - try { - client = await connect(); - } catch (error) { - if (!isPairingRequired(error)) { - throw error; - } - await approvePairing(params.operator, params.identity.deviceId); - client = await connect(); - } - if (await ensureNodeApproved(params.operator, params.identity.deviceId)) { - await client.stopAndWait({ timeoutMs: 2_000 }); - client = await connect(); - } - await client.request(NODE_RUNNER_INVENTORY_UPDATE_METHOD, { - protocolFeatures: [NODE_WORKER_SUPERVISOR_PROTOCOL_FEATURE], - workerHost: { - enabled: true, - capacity: "available", - ...(params.bundlePrewarm ? { bundlePrewarm: params.bundlePrewarm } : {}), - }, - }); - return client; -} - -async function waitForApprovedNode(operator: GatewayClient, nodeId: string): Promise { - let approved: NodeRead | undefined; - await vi.waitFor( - async () => { - const result = await operator.request<{ nodes?: NodeRead[] }>("node.list", {}); - approved = result.nodes?.find((node) => node.nodeId === nodeId); - expect(approved).toMatchObject({ - nodeId, - approvalState: "approved", - connected: true, - paired: true, - sessionHost: true, - }); - }, - { timeout: 30_000, interval: 100 }, - ); - if (!approved) { - throw new Error("paired worker node did not become available"); - } - return approved; -} - -function messageText(message: unknown): string { - const content = (message as { content?: unknown })?.content; - if (typeof content === "string") { - return content; - } - return Array.isArray(content) - ? content - .flatMap((part) => - part && typeof part === "object" && typeof (part as { text?: unknown }).text === "string" - ? [(part as { text: string }).text] - : [], - ) - .join("") - : ""; -} - describe("node worker launch wire", () => { it( "transfers and reconciles a gateway-push workspace through a device runner", @@ -326,48 +52,12 @@ describe("node worker launch wire", () => { async () => { const root = tempDirs.make("openclaw-node-worker-launch-wire-"); const provider = await startMidturnProvider(); - const published = await createPublishedWorkspace(root); - const nodeEnv = { - ...process.env, - HOME: path.join(root, "node-home"), - NODE_DISABLE_COMPILE_CACHE: undefined, - OPENCLAW_STATE_DIR: path.join(root, "node-state"), - }; - await fs.mkdir(nodeEnv.HOME, { recursive: true }); - const supervisor = createNodeWorkerSupervisor({ - env: nodeEnv, - capacity: FINALIZATION_LOAD_CONCURRENCY, - }); - const bundleInstaller = new NodeWorkerBundleInstaller({ env: nodeEnv }); - const workspace = new NodeWorkerWorkspaceRuntime({ - root: path.join(root, "node-workspaces"), - env: nodeEnv, - }); - const legacyNodeEnv = { - ...process.env, - HOME: path.join(root, "legacy-node-home"), - NODE_DISABLE_COMPILE_CACHE: undefined, - OPENCLAW_STATE_DIR: path.join(root, "legacy-node-state"), - }; - await fs.mkdir(legacyNodeEnv.HOME, { recursive: true }); - const legacySupervisor = createNodeWorkerSupervisor({ env: legacyNodeEnv }); - const legacyBundleInstaller = new NodeWorkerBundleInstaller({ env: legacyNodeEnv }); - const legacyWorkspace = new NodeWorkerWorkspaceRuntime({ - root: path.join(root, "legacy-node-workspaces"), - env: legacyNodeEnv, - }); - let gateway: Gateway | undefined; + const published = await createPublishedWireWorkspace(root); + let gateway: WireGateway | undefined; let operator: GatewayClient | undefined; - let node: GatewayClient | undefined; - let legacyNode: GatewayClient | undefined; - let closing = false; + let workerNode: PairedNodeWorkerHost | undefined; + let legacyWorkerNode: PairedNodeWorkerHost | undefined; let reconnected = false; - const invokeTasks = new Set>(); - const invokeErrors: unknown[] = []; - const legacyInvokeTasks = new Set>(); - const legacyInvokeErrors: unknown[] = []; - const commands: string[] = []; - const legacyCommands: string[] = []; let bundlePrewarm: unknown; let legacyBundlePrewarm: unknown; let launchId: string | undefined; @@ -379,94 +69,46 @@ describe("node worker launch wire", () => { }); try { - gateway = await startQaGatewayChild({ - repoRoot: process.cwd(), - useRepoCli: true, - providerBaseUrl: `${provider.baseUrl}/v1`, - providerMode: "mock-openai", - primaryModel: MODEL_REF, - alternateModel: MODEL_REF, - transportBaseUrl: "http://127.0.0.1", - controlUiEnabled: false, - mutateConfig: (config) => ({ - ...config, - nodeHost: { - ...config.nodeHost, - workerRuns: { enabled: true }, - }, - }), - }); - operator = await connectClient({ gateway, role: "operator", identity: null }); - const identity = loadOrCreateDeviceIdentity({ - path: path.join(root, "node-identity.sqlite"), - }); - const onNodeEvent = (event: GatewayEvent) => { - if (event.event !== "node.invoke.request" || !node) { - return; - } - const receiver = node; - const frame = event.payload as NodeInvokeRequestPayload; - commands.push(frame.command); - if (frame.command === NODE_WORKER_WORKSPACE_EXEC_COMMAND && frame.paramsJSON) { - const workspaceCommand = JSON.parse(frame.paramsJSON) as { - transfer?: { direction?: unknown }; - }; - if ( - observeFinalizationLoad && - workspaceCommand.transfer?.direction === "upload" && - !finalizationStartedAt - ) { - finalizationStartedAt = performance.now(); - resolveFinalizationStarted(); - } - } - if (frame.command === NODE_WORKER_SUPERVISOR_LAUNCH_COMMAND && frame.paramsJSON) { - launchId = (JSON.parse(frame.paramsJSON) as { launchId?: string }).launchId; - } - if (frame.command === NODE_WORKER_BUNDLE_INSTALL_COMMAND && frame.paramsJSON) { - bundlePrewarm = (JSON.parse(frame.paramsJSON) as { bundlePrewarm?: unknown }) - .bundlePrewarm; - } - const task = handleInvoke(frame, receiver, { current: async () => [] }, undefined, { - workerBundleInstaller: bundleInstaller, - workerSupervisor: supervisor, - workerWorkspace: workspace, - gatewayUrl: gateway!.wsUrl, - }) - .then(async () => { - if ( - frame.command === NODE_WORKER_SUPERVISOR_LAUNCH_COMMAND && - !reconnected && - !closing - ) { - reconnected = true; - await receiver.stopAndWait({ timeoutMs: 2_000 }); - if (!closing) { - node = await connectPairedNode({ - gateway: gateway!, - operator: operator!, - identity, - bundlePrewarm: 1, - onEvent: onNodeEvent, - }); - } - } - }) - .catch((error: unknown) => { - invokeErrors.push(error); - }) - .finally(() => invokeTasks.delete(task)); - invokeTasks.add(task); - }; - node = await connectPairedNode({ + gateway = await startPairedNodeWorkerGateway({ providerBaseUrl: provider.baseUrl }); + operator = await connectWireClient({ gateway, role: "operator", identity: null }); + workerNode = await createPairedNodeWorkerHost({ gateway, operator, - identity, - bundlePrewarm: 1, - onEvent: onNodeEvent, + root, + capacity: FINALIZATION_LOAD_CONCURRENCY, + bundlePrewarm: true, + onInvoke: (frame) => { + if (frame.command === NODE_WORKER_WORKSPACE_EXEC_COMMAND && frame.paramsJSON) { + const workspaceCommand = JSON.parse(frame.paramsJSON) as { + transfer?: { direction?: unknown }; + }; + if ( + observeFinalizationLoad && + workspaceCommand.transfer?.direction === "upload" && + !finalizationStartedAt + ) { + finalizationStartedAt = performance.now(); + resolveFinalizationStarted(); + } + } + if (frame.command === NODE_WORKER_SUPERVISOR_LAUNCH_COMMAND && frame.paramsJSON) { + launchId = (JSON.parse(frame.paramsJSON) as { launchId?: string }).launchId; + } + if (frame.command === NODE_WORKER_BUNDLE_INSTALL_COMMAND && frame.paramsJSON) { + bundlePrewarm = (JSON.parse(frame.paramsJSON) as { bundlePrewarm?: unknown }) + .bundlePrewarm; + } + }, + afterInvoke: async (frame, host) => { + if (frame.command !== NODE_WORKER_SUPERVISOR_LAUNCH_COMMAND || reconnected) { + return; + } + reconnected = true; + await host.disconnect(); + await host.connect(); + }, }); - const listed = await waitForApprovedNode(operator, identity.deviceId); - expect(listed.sessionHost).toBe(true); + expect(workerNode.client).toBeTruthy(); await operator.request("sessions.create", { key: SESSION_KEY, @@ -487,7 +129,7 @@ describe("node worker launch wire", () => { ); const dispatched = await gateway.call( "sessions.dispatch", - { key: SESSION_KEY, deviceId: identity.deviceId }, + { key: SESSION_KEY, deviceId: workerNode.identity.deviceId }, { timeoutMs: PROOF_TIMEOUT_MS }, ); const placement = (dispatched as { placement?: Record }).placement; @@ -523,16 +165,18 @@ describe("node worker launch wire", () => { `node worker turn failed: ${JSON.stringify(completed)}\n${gateway.logs().slice(-12_000)}`, ); } - await Promise.all([...invokeTasks]); - expect(invokeErrors).toEqual([]); + await workerNode.waitForInvokes(); + expect(workerNode.invokeErrors).toEqual([]); expect(reconnected).toBe(true); - expect(commands).toContain(NODE_WORKER_BUNDLE_INSTALL_COMMAND); - expect(commands).toContain(NODE_WORKER_WORKSPACE_EXEC_COMMAND); - expect(commands).toContain(NODE_WORKER_SUPERVISOR_LAUNCH_COMMAND); - expect(commands).toContain(NODE_WORKER_SUPERVISOR_STATUS_COMMAND); + expect(workerNode.commands).toContain(NODE_WORKER_BUNDLE_INSTALL_COMMAND); + expect(workerNode.commands).toContain(NODE_WORKER_WORKSPACE_EXEC_COMMAND); + expect(workerNode.commands).toContain(NODE_WORKER_SUPERVISOR_LAUNCH_COMMAND); + expect(workerNode.commands).toContain(NODE_WORKER_SUPERVISOR_STATUS_COMMAND); expect(launchId).toBeTruthy(); expect(bundlePrewarm).toBe(1); - await expect(supervisor.status(launchId!)).resolves.toMatchObject({ state: "completed" }); + await expect(workerNode.supervisor.status(launchId!)).resolves.toMatchObject({ + state: "completed", + }); const history = await operator.request<{ messages?: unknown[] }>("chat.history", { sessionKey: SESSION_KEY, @@ -542,7 +186,7 @@ describe("node worker launch wire", () => { history.messages?.filter( (message) => (message as { role?: unknown }).role === "assistant" && - messageText(message).includes(BASELINE_REPLY), + wireMessageText(message).includes(BASELINE_REPLY), ), ).toHaveLength(1); const described = (await gateway.call("sessions.describe", { key: SESSION_KEY })) as { @@ -563,39 +207,18 @@ describe("node worker launch wire", () => { "device result\n", ); - const legacyIdentity = loadOrCreateDeviceIdentity({ - path: path.join(root, "legacy-node-identity.sqlite"), - }); - const onLegacyNodeEvent = (event: GatewayEvent) => { - if (event.event !== "node.invoke.request" || !legacyNode) { - return; - } - const receiver = legacyNode; - const frame = event.payload as NodeInvokeRequestPayload; - legacyCommands.push(frame.command); - if (frame.command === NODE_WORKER_BUNDLE_INSTALL_COMMAND && frame.paramsJSON) { - legacyBundlePrewarm = (JSON.parse(frame.paramsJSON) as { bundlePrewarm?: unknown }) - .bundlePrewarm; - } - const task = handleInvoke(frame, receiver, { current: async () => [] }, undefined, { - workerBundleInstaller: legacyBundleInstaller, - workerSupervisor: legacySupervisor, - workerWorkspace: legacyWorkspace, - gatewayUrl: gateway!.wsUrl, - }) - .catch((error: unknown) => { - legacyInvokeErrors.push(error); - }) - .finally(() => legacyInvokeTasks.delete(task)); - legacyInvokeTasks.add(task); - }; - legacyNode = await connectPairedNode({ + legacyWorkerNode = await createPairedNodeWorkerHost({ gateway, operator, - identity: legacyIdentity, - onEvent: onLegacyNodeEvent, + root, + label: "legacy-node", + onInvoke: (frame) => { + if (frame.command === NODE_WORKER_BUNDLE_INSTALL_COMMAND && frame.paramsJSON) { + legacyBundlePrewarm = (JSON.parse(frame.paramsJSON) as { bundlePrewarm?: unknown }) + .bundlePrewarm; + } + }, }); - await waitForApprovedNode(operator, legacyIdentity.deviceId); const legacySessionKey = `${SESSION_KEY}-legacy-node`; await operator.request("sessions.create", { key: legacySessionKey, @@ -607,7 +230,7 @@ describe("node worker launch wire", () => { }); await gateway.call( "sessions.dispatch", - { key: legacySessionKey, deviceId: legacyIdentity.deviceId }, + { key: legacySessionKey, deviceId: legacyWorkerNode.identity.deviceId }, { timeoutMs: PROOF_TIMEOUT_MS }, ); const legacyRunId = `node-worker-launch-wire-legacy-${Date.now()}`; @@ -624,9 +247,9 @@ describe("node worker launch wire", () => { { timeoutMs: PROOF_TIMEOUT_MS + 5_000 }, ), ).resolves.toMatchObject({ status: "ok" }); - await Promise.all([...legacyInvokeTasks]); - expect(legacyInvokeErrors).toEqual([]); - expect(legacyCommands).toContain(NODE_WORKER_SUPERVISOR_LAUNCH_COMMAND); + await legacyWorkerNode.waitForInvokes(); + expect(legacyWorkerNode.invokeErrors).toEqual([]); + expect(legacyWorkerNode.commands).toContain(NODE_WORKER_SUPERVISOR_LAUNCH_COMMAND); expect(legacyBundlePrewarm).toBeUndefined(); const loadSessions: string[] = []; @@ -642,7 +265,7 @@ describe("node worker launch wire", () => { }); await gateway.call( "sessions.dispatch", - { key: sessionKey, deviceId: identity.deviceId }, + { key: sessionKey, deviceId: workerNode.identity.deviceId }, { timeoutMs: PROOF_TIMEOUT_MS }, ); loadSessions.push(sessionKey); @@ -675,25 +298,25 @@ describe("node worker launch wire", () => { })(); const loadRunIds = await Promise.all( loadSessions.map(async (sessionKey, index) => { - const runId = `node-worker-finalization-load-${index}-${Date.now()}`; - const started = await operator!.request<{ runId?: string; status?: string }>( + const loadRunId = `node-worker-finalization-load-${index}-${Date.now()}`; + const loadStarted = await operator!.request<{ runId?: string; status?: string }>( "chat.send", { sessionKey, message: BASELINE_PROMPT, deliver: false, - idempotencyKey: runId, + idempotencyKey: loadRunId, }, ); - expect(started).toMatchObject({ runId, status: "started" }); - return runId; + expect(loadStarted).toMatchObject({ runId: loadRunId, status: "started" }); + return loadRunId; }), ); const waits = Promise.all( - loadRunIds.map(async (runId) => { + loadRunIds.map(async (loadRunId) => { const completedLoad = await operator!.request<{ status?: string }>( "agent.wait", - { runId, timeoutMs: PROOF_TIMEOUT_MS }, + { runId: loadRunId, timeoutMs: PROOF_TIMEOUT_MS }, { timeoutMs: PROOF_TIMEOUT_MS + 5_000 }, ); expect(completedLoad.status).toBe("ok"); @@ -701,7 +324,7 @@ describe("node worker launch wire", () => { ); await finalizationStarted; const freshConnectionStartedAt = performance.now(); - const freshClient = await connectClient({ + const freshClient = await connectWireClient({ gateway, role: "operator", identity: null, @@ -723,18 +346,13 @@ describe("node worker launch wire", () => { ); expect(freshConnectionMs).toBeLessThan(CONTROL_PROBE_MAX_MS); } finally { - closing = true; const cleanup = await Promise.allSettled([ - node?.stopAndWait({ timeoutMs: 2_000 }) ?? Promise.resolve(), - legacyNode?.stopAndWait({ timeoutMs: 2_000 }) ?? Promise.resolve(), + workerNode?.stop() ?? Promise.resolve(), + legacyWorkerNode?.stop() ?? Promise.resolve(), operator?.stopAndWait({ timeoutMs: 2_000 }) ?? Promise.resolve(), - Promise.allSettled([...invokeTasks]), - Promise.allSettled([...legacyInvokeTasks]), - supervisor.close(), - legacySupervisor.close(), gateway?.stop() ?? Promise.resolve(), provider.stop(), - closeServer(published.server), + closeWireServer(published.server), ]); const failures = cleanup.flatMap((result) => result.status === "rejected" ? [result.reason] : [], diff --git a/test/e2e/qa-lab/runtime/paired-node-worker-lifecycle-provider.ts b/test/e2e/qa-lab/runtime/paired-node-worker-lifecycle-provider.ts new file mode 100644 index 000000000000..25d6d5bae80b --- /dev/null +++ b/test/e2e/qa-lab/runtime/paired-node-worker-lifecycle-provider.ts @@ -0,0 +1,156 @@ +import { randomUUID } from "node:crypto"; +import { createServer, type ServerResponse } from "node:http"; + +type Deferred = { promise: Promise; resolve: () => void }; + +function createDeferred(): Deferred { + let resolve = () => {}; + const promise = new Promise((settle) => { + resolve = settle; + }); + return { promise, resolve }; +} + +async function readBody(request: AsyncIterable): Promise { + const chunks: Buffer[] = []; + for await (const chunk of request) { + chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(String(chunk))); + } + return Buffer.concat(chunks).toString("utf8"); +} + +function responseText(raw: string): string { + const matches = [...raw.matchAll(/Reply exactly:\s*([A-Z0-9_-]+)/gu)]; + return matches.at(-1)?.[1] ?? "WIRE-OK"; +} + +function writeEvent(response: ServerResponse, event: unknown): void { + response.write(`data: ${JSON.stringify(event)}\n\n`); +} + +function writeReply(response: ServerResponse, text: string): void { + if (!response.headersSent) { + response.writeHead(200, { + "content-type": "text/event-stream", + "cache-control": "no-store", + connection: "keep-alive", + }); + } + const id = `msg_${randomUUID()}`; + const item = { + type: "message", + id, + role: "assistant", + phase: "final_answer", + status: "completed", + content: [{ type: "output_text", text, annotations: [] }], + }; + writeEvent(response, { + type: "response.output_item.added", + output_index: 0, + item: { ...item, status: "in_progress", content: [] }, + }); + writeEvent(response, { + type: "response.output_text.delta", + item_id: id, + output_index: 0, + content_index: 0, + delta: text, + }); + writeEvent(response, { + type: "response.output_text.done", + item_id: id, + output_index: 0, + content_index: 0, + text, + }); + writeEvent(response, { type: "response.output_item.done", output_index: 0, item }); + writeEvent(response, { + type: "response.completed", + response: { + id: `resp_${id}`, + status: "completed", + output: [item], + usage: { input_tokens: 32, output_tokens: 8, total_tokens: 40 }, + }, + }); + response.end("data: [DONE]\n\n"); +} + +export async function startPairedNodeWorkerLifecycleProvider(holdMarkers: readonly string[]) { + const holdSet = new Set(holdMarkers); + const releases = new Map(holdMarkers.map((marker) => [marker, createDeferred()] as const)); + const held = new Set(); + const pendingResponses = new Set(); + const server = createServer((request, response) => { + void (async () => { + if (request.method === "GET" && request.url === "/v1/models") { + response.writeHead(200, { "content-type": "application/json" }); + response.end(JSON.stringify({ data: [{ id: "gpt-5.6-luna", object: "model" }] })); + return; + } + if (request.method !== "POST" || request.url !== "/v1/responses") { + response.writeHead(404).end(); + return; + } + const raw = await readBody(request); + const text = responseText(raw); + if (holdSet.has(text)) { + held.add(text); + pendingResponses.add(response); + response.once("close", () => pendingResponses.delete(response)); + response.writeHead(200, { + "content-type": "text/event-stream", + "cache-control": "no-store", + connection: "keep-alive", + }); + response.flushHeaders(); + await releases.get(text)!.promise; + if (!response.destroyed) { + writeReply(response, text); + } + pendingResponses.delete(response); + return; + } + writeReply(response, text); + })().catch((error: unknown) => { + if (!response.headersSent) { + response.writeHead(500); + } + response.end(error instanceof Error ? error.message : String(error)); + }); + }); + await new Promise((resolve, reject) => { + server.once("error", reject); + server.listen(0, "127.0.0.1", resolve); + }); + const address = server.address(); + if (!address || typeof address === "string") { + throw new Error("paired node lifecycle provider did not bind"); + } + return { + baseUrl: `http://127.0.0.1:${address.port}`, + hasHeld(marker: string): boolean { + return held.has(marker); + }, + release(marker: string): void { + releases.get(marker)?.resolve(); + }, + releaseAll(): void { + for (const release of releases.values()) { + release.resolve(); + } + }, + async stop(): Promise { + for (const release of releases.values()) { + release.resolve(); + } + for (const response of pendingResponses) { + response.destroy(); + } + pendingResponses.clear(); + server.closeAllConnections(); + await new Promise((resolve) => server.close(() => resolve())); + }, + }; +} diff --git a/test/e2e/qa-lab/runtime/paired-node-worker-lifecycle-wire.e2e.test.ts b/test/e2e/qa-lab/runtime/paired-node-worker-lifecycle-wire.e2e.test.ts new file mode 100644 index 000000000000..9494f8e08bfb --- /dev/null +++ b/test/e2e/qa-lab/runtime/paired-node-worker-lifecycle-wire.e2e.test.ts @@ -0,0 +1,351 @@ +import fs from "node:fs/promises"; +import { GatewayClient } from "openclaw/plugin-sdk/gateway-runtime"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { NODE_WORKER_WORKSPACE_RETAIN_COMMAND } from "../../../../src/infra/node-commands.js"; +import { useAutoCleanupTempDirTracker } from "../../../helpers/temp-dir.js"; +import { PROOF_TIMEOUT_MS } from "./cloud-worker-midturn-loss-fixture.js"; +import { startPairedNodeWorkerLifecycleProvider } from "./paired-node-worker-lifecycle-provider.js"; +import { + bundleInstallFrames, + closeWireServer, + connectWireClient, + createPairedNodeWorkerHost, + createPublishedWireWorkspace, + startPairedNodeWorkerGateway, + type PairedNodeWorkerHost, + type PublishedWireWorkspace, + type WireGateway, + type WireNodeRead, + wireMessageText, +} from "./paired-node-worker-wire-fixture.js"; + +const TEST_TIMEOUT_MS = PROOF_TIMEOUT_MS + 180_000; +const SESSION_PREFIX = "agent:qa:paired-node-worker-lifecycle"; +const HOLD_A = "WIRE-HOLD-A"; +const HOLD_B = "WIRE-HOLD-B"; + +type TurnResult = { runId?: string; status?: string; summary?: string }; +type Placement = { environmentId?: string; state?: string; workerBundleHash?: string }; +type EnvironmentRead = { + id: string; + type: string; + workerBundle?: WireNodeRead["workerBundle"]; + worker?: { state?: string }; +}; + +const tempDirs = useAutoCleanupTempDirTracker(afterEach); + +async function createSession(params: { + operator: GatewayClient; + published: PublishedWireWorkspace; + suffix: string; +}): Promise { + const key = `${SESSION_PREFIX}:${params.suffix}`; + await params.operator.request("sessions.create", { + key, + agentId: "qa", + worktree: true, + worktreeName: `paired-node-${params.suffix}`, + worktreeBaseRef: "main", + cwd: params.published.source, + }); + return key; +} + +async function dispatchNodeSession(params: { + gateway: WireGateway; + key: string; + nodeId: string; +}): Promise { + const result = (await params.gateway.call( + "sessions.dispatch", + { key: params.key, deviceId: params.nodeId }, + { timeoutMs: PROOF_TIMEOUT_MS }, + )) as { placement?: Placement }; + expect(result.placement).toMatchObject({ state: "active" }); + return result.placement!; +} + +async function startTurn(params: { + operator: GatewayClient; + key: string; + marker: string; +}): Promise { + const runId = `${params.marker.toLowerCase()}-${Date.now()}`; + const started = await params.operator.request("chat.send", { + sessionKey: params.key, + message: `Reply exactly: ${params.marker}`, + deliver: false, + idempotencyKey: runId, + }); + expect(started).toMatchObject({ runId, status: "started" }); + return runId; +} + +async function waitForTurn(operator: GatewayClient, runId: string): Promise { + return await operator.request( + "agent.wait", + { runId, timeoutMs: PROOF_TIMEOUT_MS }, + { timeoutMs: PROOF_TIMEOUT_MS + 5_000 }, + ); +} + +async function expectSuccessfulTurn(params: { + operator: GatewayClient; + key: string; + marker: string; +}): Promise { + const runId = await startTurn(params); + await expect(waitForTurn(params.operator, runId)).resolves.toMatchObject({ status: "ok" }); + await vi.waitFor( + async () => { + const history = await params.operator.request<{ messages?: unknown[] }>("chat.history", { + sessionKey: params.key, + limit: 100, + }); + expect( + history.messages?.some( + (message) => + (message as { role?: unknown }).role === "assistant" && + wireMessageText(message).includes(params.marker), + ), + ).toBe(true); + }, + { timeout: 30_000, interval: 100 }, + ); +} + +async function describePlacement(gateway: WireGateway, key: string): Promise { + const described = (await gateway.call("sessions.describe", { key })) as { + session?: { placement?: Placement }; + }; + return described.session?.placement ?? {}; +} + +async function readNode( + operator: GatewayClient, + nodeId: string, +): Promise { + const result = await operator.request<{ nodes?: WireNodeRead[] }>("node.list", {}); + return result.nodes?.find((node) => node.nodeId === nodeId); +} + +async function readEnvironments(operator: GatewayClient): Promise { + const result = await operator.request<{ environments?: EnvironmentRead[] }>( + "environments.list", + {}, + ); + return result.environments ?? []; +} + +async function expectPublicBundleStatus(params: { + operator: GatewayClient; + nodeId: string; + status: "installed" | "missing"; +}): Promise { + await vi.waitFor( + async () => { + const node = await readNode(params.operator, params.nodeId); + const environment = (await readEnvironments(params.operator)).find( + (entry) => entry.id === `node:${params.nodeId}`, + ); + expect(node?.workerBundle?.status).toBe(params.status); + expect(environment?.workerBundle?.status).toBe(params.status); + for (const projection of [node?.workerBundle, environment?.workerBundle]) { + expect(Object.keys(projection ?? {}).toSorted()).toEqual( + params.status === "installed" ? ["status", "version"] : ["status"], + ); + expect(projection).not.toHaveProperty("bundleHash"); + expect(projection).not.toHaveProperty("path"); + } + }, + { timeout: 30_000, interval: 100 }, + ); +} + +describe("paired node worker lifecycle wire", () => { + it( + "keeps local control usable across bundle loss, disconnect, capacity, and role removal", + { timeout: TEST_TIMEOUT_MS }, + async () => { + const root = tempDirs.make("openclaw-paired-node-worker-lifecycle-"); + const provider = await startPairedNodeWorkerLifecycleProvider([HOLD_A, HOLD_B]); + const published = await createPublishedWireWorkspace(root); + let gateway: WireGateway | undefined; + let operator: GatewayClient | undefined; + let workerNode: PairedNodeWorkerHost | undefined; + try { + gateway = await startPairedNodeWorkerGateway({ providerBaseUrl: provider.baseUrl }); + operator = await connectWireClient({ gateway, role: "operator", identity: null }); + workerNode = await createPairedNodeWorkerHost({ + gateway, + operator, + root, + capacity: 2, + capacityWaitMs: 750, + bundlePrewarm: true, + bundleRetention: true, + bundleStatus: true, + }); + const nodeId = workerNode.identity.deviceId; + const localKey = await createSession({ operator, published, suffix: "local-control" }); + const retainedKey = await createSession({ operator, published, suffix: "retention" }); + const retainedPlacement = await dispatchNodeSession({ + gateway, + key: retainedKey, + nodeId, + }); + const bundleHash = retainedPlacement.workerBundleHash; + expect(bundleHash).toMatch(/^[a-f0-9]{64}$/u); + + // Local and node-placed sessions share the Gateway without sharing failure state. + await expectSuccessfulTurn({ operator, key: retainedKey, marker: "WIRE-NODE-BASELINE" }); + await expectSuccessfulTurn({ operator, key: localKey, marker: "WIRE-LOCAL-BASELINE" }); + + // Bundle maintenance reports only the public status/version projection and repairs loss. + await expectPublicBundleStatus({ operator, nodeId, status: "installed" }); + const installedBundle = await workerNode.installedBundleDirectory(bundleHash!); + const installCountBeforeLoss = bundleInstallFrames(workerNode).length; + await fs.rm(installedBundle, { recursive: true, force: true }); + await workerNode.disconnect(); + await workerNode.connect(); + await expectPublicBundleStatus({ operator, nodeId, status: "missing" }); + expect( + workerNode.commands.filter((command) => command === NODE_WORKER_WORKSPACE_RETAIN_COMMAND) + .length, + ).toBeGreaterThan(0); + await expectSuccessfulTurn({ operator, key: localKey, marker: "WIRE-LOCAL-AFTER-LOSS" }); + + const repairedKey = await createSession({ operator, published, suffix: "reinstalled" }); + await dispatchNodeSession({ gateway, key: repairedKey, nodeId }); + await vi.waitFor( + () => expect(bundleInstallFrames(workerNode!).length).toBe(installCountBeforeLoss + 1), + { timeout: 30_000, interval: 100 }, + ); + await expectPublicBundleStatus({ operator, nodeId, status: "installed" }); + + // An offline runner fails before handoff, leaves the active placement retryable, and + // does not terminalize the independent local session. + await workerNode.disconnect(); + const offlineRunId = await startTurn({ + operator, + key: repairedKey, + marker: "WIRE-OFFLINE-ATTEMPT", + }); + const offline = await waitForTurn(operator, offlineRunId); + expect(offline.status).not.toBe("ok"); + expect(`${offline.summary ?? ""} ${JSON.stringify(offline)}`).toMatch( + /runner-offline|runner is offline|reconnect/iu, + ); + expect(await describePlacement(gateway, repairedKey)).toMatchObject({ state: "active" }); + await expectSuccessfulTurn({ operator, key: localKey, marker: "WIRE-LOCAL-AFTER-OFFLINE" }); + await workerNode.connect(); + await expectSuccessfulTurn({ operator, key: repairedKey, marker: "WIRE-OFFLINE-RETRY" }); + + // Two nonterminal real worker children consume both physical slots. The third turn + // records a bounded capacity failure at the public RPC/history/placement boundaries. + const holdAKey = await createSession({ operator, published, suffix: "capacity-a" }); + const holdBKey = await createSession({ operator, published, suffix: "capacity-b" }); + const capacityKey = await createSession({ operator, published, suffix: "capacity-c" }); + await dispatchNodeSession({ gateway, key: holdAKey, nodeId }); + await dispatchNodeSession({ gateway, key: holdBKey, nodeId }); + await dispatchNodeSession({ gateway, key: capacityKey, nodeId }); + const holdARunId = await startTurn({ operator, key: holdAKey, marker: HOLD_A }); + const holdBRunId = await startTurn({ operator, key: holdBKey, marker: HOLD_B }); + await vi.waitFor( + () => { + expect(provider.hasHeld(HOLD_A)).toBe(true); + expect(provider.hasHeld(HOLD_B)).toBe(true); + }, + { timeout: 30_000, interval: 100 }, + ); + const capacityRunId = await startTurn({ + operator, + key: capacityKey, + marker: "WIRE-CAPACITY-ATTEMPT", + }); + const capacity = await waitForTurn(operator, capacityRunId); + expect(capacity.status).not.toBe("ok"); + expect(`${capacity.summary ?? ""} ${JSON.stringify(capacity)}`).toMatch(/capacity/iu); + expect(await describePlacement(gateway, capacityKey)).toMatchObject({ state: "active" }); + await vi.waitFor( + async () => { + const history = await operator!.request<{ messages?: unknown[] }>("chat.history", { + sessionKey: capacityKey, + limit: 100, + }); + const messages = history.messages ?? []; + expect( + messages.some( + (message) => + (message as { role?: unknown }).role === "user" && + wireMessageText(message).includes("WIRE-CAPACITY-ATTEMPT"), + ), + ).toBe(true); + expect( + messages.some( + (message) => + (message as { role?: unknown }).role === "assistant" && + wireMessageText(message).includes("WIRE-CAPACITY-ATTEMPT"), + ), + ).toBe(false); + }, + { timeout: 30_000, interval: 100 }, + ); + await expectSuccessfulTurn({ operator, key: localKey, marker: "WIRE-LOCAL-AT-CAPACITY" }); + provider.release(HOLD_A); + await expect(waitForTurn(operator, holdARunId)).resolves.toMatchObject({ status: "ok" }); + await expectSuccessfulTurn({ + operator, + key: capacityKey, + marker: "WIRE-CAPACITY-RETRY", + }); + provider.release(HOLD_B); + await expect(waitForTurn(operator, holdBRunId)).resolves.toMatchObject({ status: "ok" }); + + // Node-role removal waits for environment and placement cleanup before returning, + // invalidates the old connection, and leaves local execution available. + const removalKey = await createSession({ operator, published, suffix: "role-removal" }); + const removalPlacement = await dispatchNodeSession({ gateway, key: removalKey, nodeId }); + expect(workerNode.client).toBeTruthy(); + await expect(operator.request("node.pair.remove", { nodeId })).resolves.toMatchObject({ + nodeId, + }); + const removedEnvironment = (await readEnvironments(operator)).find( + (entry) => entry.id === removalPlacement.environmentId, + ); + expect(["destroyed", "failed", "orphaned"]).toContain(removedEnvironment?.worker?.state); + expect(await describePlacement(gateway, removalKey)).toMatchObject({ state: "failed" }); + await expect(workerNode.publishInventory()).rejects.toBeTruthy(); + await vi.waitFor( + async () => { + const node = await readNode(operator!, nodeId); + expect(node?.connected).not.toBe(true); + }, + { timeout: 30_000, interval: 100 }, + ); + await expectSuccessfulTurn({ operator, key: localKey, marker: "WIRE-LOCAL-AFTER-REMOVAL" }); + await workerNode.waitForInvokes(); + expect(workerNode.invokeErrors).toEqual([]); + } finally { + provider.releaseAll(); + const cleanup = await Promise.allSettled([ + workerNode?.stop() ?? Promise.resolve(), + operator?.stopAndWait({ timeoutMs: 2_000 }) ?? Promise.resolve(), + gateway?.stop() ?? Promise.resolve(), + provider.stop(), + closeWireServer(published.server), + ]); + const failures = cleanup.flatMap((result) => + result.status === "rejected" ? [result.reason] : [], + ); + if (failures.length === 1) { + throw failures[0]; + } + if (failures.length > 1) { + throw new AggregateError(failures, "paired node worker lifecycle cleanup failed"); + } + } + }, + ); +}); diff --git a/test/e2e/qa-lab/runtime/paired-node-worker-wire-fixture.ts b/test/e2e/qa-lab/runtime/paired-node-worker-wire-fixture.ts new file mode 100644 index 000000000000..4e0bcdf7d5ff --- /dev/null +++ b/test/e2e/qa-lab/runtime/paired-node-worker-wire-fixture.ts @@ -0,0 +1,509 @@ +import { execFile } from "node:child_process"; +import fs from "node:fs/promises"; +import { createServer, type Server } from "node:http"; +import path from "node:path"; +import { promisify } from "node:util"; +import { GatewayClient } from "openclaw/plugin-sdk/gateway-runtime"; +import { expect, vi } from "vitest"; +import { startQaGatewayChild } from "../../../../extensions/qa-lab/api.js"; +import { + GATEWAY_CLIENT_MODES, + GATEWAY_CLIENT_NAMES, +} from "../../../../packages/gateway-protocol/src/client-info.js"; +import { WORKER_BUNDLE_PREWARM_VERSION } from "../../../../packages/gateway-protocol/src/schema/worker-admission.js"; +import type { DeviceIdentity } from "../../../../src/infra/device-identity.js"; +import { loadOrCreateDeviceIdentity } from "../../../../src/infra/device-identity.js"; +import { NODE_WORKER_BUNDLE_INSTALL_COMMAND } from "../../../../src/infra/node-commands.js"; +import { + NODE_RUNNER_INVENTORY_UPDATE_METHOD, + NODE_WORKER_BUNDLE_RETENTION_VERSION, + NODE_WORKER_BUNDLE_STATUS_VERSION, + NODE_WORKER_SUPERVISOR_PROTOCOL_FEATURE, +} from "../../../../src/infra/node-runner-inventory.js"; +import { handleInvoke, type NodeInvokeRequestPayload } from "../../../../src/node-host/invoke.js"; +import { NodeWorkerBundleInstaller } from "../../../../src/node-host/node-worker-bundle-installer.js"; +import { createNodeWorkerSupervisor } from "../../../../src/node-host/node-worker-supervisor.js"; +import { NodeWorkerWorkspaceRuntime } from "../../../../src/node-host/node-worker-workspace.js"; +import { VERSION } from "../../../../src/version.js"; +import { MODEL_REF, PROOF_TIMEOUT_MS } from "./cloud-worker-midturn-loss-fixture.js"; + +const execFileAsync = promisify(execFile); +const NODE_DISPLAY_NAME = "QA Gateway-bundle worker node"; + +export type WireGateway = Awaited>; +type WireGatewayEvent = { event: string; payload?: unknown }; +export type WireNodeRead = { + nodeId: string; + approvalState?: string; + connected?: boolean; + paired?: boolean; + sessionHost?: boolean; + workerBundle?: { status: "installed"; version: string } | { status: "missing" }; +}; +export type PublishedWireWorkspace = { + commit: string; + source: string; + server: Server; +}; + +async function git(cwd: string, ...args: string[]): Promise { + const { stdout } = await execFileAsync("git", ["-C", cwd, ...args], { + encoding: "utf8", + timeout: 20_000, + }); + return stdout.trim(); +} + +export async function closeWireServer(server: Server): Promise { + server.closeAllConnections(); + await new Promise((resolve, reject) => { + server.close((error) => (error ? reject(error) : resolve())); + }); +} + +export async function createPublishedWireWorkspace(root: string): Promise { + const source = path.join(root, "source"); + const bare = path.join(root, "repo.git"); + await fs.mkdir(source, { recursive: true }); + await execFileAsync("git", ["init", "--bare", bare]); + await git(source, "init", "-b", "main"); + await git(source, "config", "user.name", "OpenClaw QA"); + await git(source, "config", "user.email", "openclaw-qa@example.invalid"); + await fs.mkdir(path.join(source, "nested")); + await fs.writeFile(path.join(source, "launch-wire.txt"), "local-install launch wire\n"); + await fs.writeFile(path.join(source, "nested", "tracked.txt"), "nested tracked input\n"); + await git(source, "add", "."); + await git(source, "commit", "-m", "initialize node worker launch wire workspace"); + await git(source, "remote", "add", "publish", bare); + await git(source, "push", "publish", "main"); + await git(source, "remote", "remove", "publish"); + await git(bare, "update-server-info"); + + const server = createServer((request, response) => { + void (async () => { + const pathname = decodeURIComponent(new URL(request.url ?? "/", "http://localhost").pathname); + if (!pathname.startsWith("/repo.git/")) { + response.writeHead(404).end(); + return; + } + const candidate = path.resolve(bare, pathname.slice("/repo.git/".length)); + if (candidate !== bare && !candidate.startsWith(`${bare}${path.sep}`)) { + response.writeHead(404).end(); + return; + } + try { + const contents = await fs.readFile(candidate); + response.writeHead(200, { + "content-type": pathname.endsWith("/info/refs") + ? "text/plain; charset=utf-8" + : "application/octet-stream", + "content-length": String(contents.byteLength), + }); + response.end(request.method === "HEAD" ? undefined : contents); + } catch { + response.writeHead(404).end(); + } + })(); + }); + await new Promise((resolve, reject) => { + server.once("error", reject); + server.listen(0, "127.0.0.1", resolve); + }); + const address = server.address(); + if (!address || typeof address === "string") { + throw new Error("published workspace server did not bind"); + } + const origin = `http://127.0.0.1:${address.port}/repo.git`; + await git(source, "remote", "add", "origin", origin); + const commit = await git(source, "rev-parse", "HEAD"); + await git(source, "ls-remote", "--exit-code", origin, "refs/heads/main"); + return { commit, source: await fs.realpath(source), server }; +} + +export async function connectWireClient(params: { + gateway: WireGateway; + role: "operator" | "node"; + identity: DeviceIdentity | null; + onEvent?: (event: WireGatewayEvent) => void; + timeoutMs?: number; +}): Promise { + return await new Promise((resolve, reject) => { + let settled = false; + const finish = (error?: Error) => { + if (settled) { + return; + } + settled = true; + clearTimeout(timeout); + if (error) { + client.stop(); + reject(error); + } else { + resolve(client); + } + }; + const timeout = setTimeout( + () => finish(new Error("Gateway client connection timed out")), + params.timeoutMs ?? 30_000, + ); + timeout.unref(); + const node = params.role === "node"; + const client = new GatewayClient({ + url: params.gateway.wsUrl, + token: params.gateway.token, + env: params.gateway.runtimeEnv, + role: params.role, + clientName: node ? GATEWAY_CLIENT_NAMES.NODE_HOST : GATEWAY_CLIENT_NAMES.GATEWAY_CLIENT, + clientDisplayName: node ? NODE_DISPLAY_NAME : "Paired node worker wire operator", + clientVersion: VERSION, + platform: node ? "macos" : process.platform, + deviceFamily: node ? "Mac" : undefined, + mode: node ? GATEWAY_CLIENT_MODES.NODE : GATEWAY_CLIENT_MODES.BACKEND, + scopes: node ? [] : ["operator.admin", "operator.pairing", "operator.read", "operator.write"], + caps: node ? ["system"] : undefined, + commands: node ? [] : undefined, + deviceIdentity: params.identity, + requestTimeoutMs: PROOF_TIMEOUT_MS, + onEvent: params.onEvent, + onHelloOk: () => finish(), + onConnectError: (error) => finish(error), + onClose: (code, reason) => finish(new Error(`Gateway closed (${code}): ${reason}`)), + }); + client.start(); + }); +} + +function isPairingRequired(error: unknown): boolean { + const details = + error && typeof error === "object" + ? (error as { details?: { code?: unknown } }).details + : undefined; + return details?.code === "PAIRING_REQUIRED" || String(error).includes("PAIRING_REQUIRED"); +} + +async function approveNodePairing(operator: GatewayClient, nodeId: string): Promise { + let nodeRequestId: string | undefined; + await vi.waitFor( + async () => { + const result = await operator.request<{ + pending?: Array<{ requestId?: string; nodeId?: string }>; + }>("node.pair.list", {}); + nodeRequestId = result.pending?.find((entry) => entry.nodeId === nodeId)?.requestId; + expect(nodeRequestId).toBeTruthy(); + }, + { timeout: 30_000, interval: 100 }, + ); + await operator.request("node.pair.approve", { requestId: nodeRequestId }); +} + +async function approvePairing(operator: GatewayClient, nodeId: string): Promise { + let deviceRequestId: string | undefined; + await vi.waitFor( + async () => { + const result = await operator.request<{ + pending?: Array<{ requestId?: string; deviceId?: string; role?: string }>; + }>("device.pair.list", {}); + deviceRequestId = result.pending?.find( + (entry) => entry.deviceId === nodeId || entry.role === "node", + )?.requestId; + expect(deviceRequestId).toBeTruthy(); + }, + { timeout: 30_000, interval: 100 }, + ); + await operator.request("device.pair.approve", { requestId: deviceRequestId }); + await approveNodePairing(operator, nodeId); +} + +async function ensureNodeApproved(operator: GatewayClient, nodeId: string): Promise { + let approvalState: string | undefined; + await vi.waitFor( + async () => { + const result = await operator.request<{ nodes?: WireNodeRead[] }>("node.list", {}); + approvalState = result.nodes?.find((node) => node.nodeId === nodeId)?.approvalState; + expect(approvalState).toBeTruthy(); + }, + { timeout: 30_000, interval: 100 }, + ); + if (approvalState !== "approved") { + await approveNodePairing(operator, nodeId); + return true; + } + return false; +} + +async function waitForApprovedWireNode( + operator: GatewayClient, + nodeId: string, +): Promise { + let approved: WireNodeRead | undefined; + await vi.waitFor( + async () => { + const result = await operator.request<{ nodes?: WireNodeRead[] }>("node.list", {}); + approved = result.nodes?.find((node) => node.nodeId === nodeId); + expect(approved).toMatchObject({ + nodeId, + approvalState: "approved", + connected: true, + paired: true, + sessionHost: true, + }); + }, + { timeout: 30_000, interval: 100 }, + ); + if (!approved) { + throw new Error("paired worker node did not become available"); + } + return approved; +} + +type WireWorkerHostOptions = { + gateway: WireGateway; + operator: GatewayClient; + root: string; + label?: string; + capacity?: number; + capacityWaitMs?: number; + bundlePrewarm?: boolean; + bundleRetention?: boolean; + bundleStatus?: boolean; + onInvoke?: (frame: NodeInvokeRequestPayload) => void; + afterInvoke?: (frame: NodeInvokeRequestPayload, host: PairedNodeWorkerHost) => Promise; +}; + +export type PairedNodeWorkerHost = { + readonly identity: DeviceIdentity; + readonly commands: string[]; + readonly frames: NodeInvokeRequestPayload[]; + readonly invokeErrors: unknown[]; + readonly supervisor: ReturnType; + readonly bundleInstaller: NodeWorkerBundleInstaller; + readonly workspace: NodeWorkerWorkspaceRuntime; + readonly client: GatewayClient | undefined; + connect(): Promise; + disconnect(): Promise; + publishInventory(): Promise; + waitForInvokes(): Promise; + installedBundleDirectory(bundleHash: string): Promise; + stop(): Promise; +}; + +export async function createPairedNodeWorkerHost( + options: WireWorkerHostOptions, +): Promise { + const label = options.label ?? "node"; + const nodeStateDir = path.join(options.root, `${label}-state`); + const nodeHostRoot = path.join(nodeStateDir, "node-host"); + const nodeEnv = { + ...process.env, + HOME: path.join(options.root, `${label}-home`), + NODE_DISABLE_COMPILE_CACHE: undefined, + OPENCLAW_STATE_DIR: nodeStateDir, + }; + await fs.mkdir(nodeEnv.HOME, { recursive: true }); + const workspace = new NodeWorkerWorkspaceRuntime({ root: nodeHostRoot, env: nodeEnv }); + const bundleInstaller = new NodeWorkerBundleInstaller({ root: nodeHostRoot, env: nodeEnv }); + let capacityAvailable = true; + let client: GatewayClient | undefined; + let closing = false; + const invokeTasks = new Set>(); + const invokeErrors: unknown[] = []; + const commands: string[] = []; + const frames: NodeInvokeRequestPayload[] = []; + const identity = loadOrCreateDeviceIdentity({ + path: path.join(options.root, `${label}-identity.sqlite`), + }); + + const inventory = () => ({ + protocolFeatures: [NODE_WORKER_SUPERVISOR_PROTOCOL_FEATURE], + workerHost: { + enabled: true as const, + capacity: capacityAvailable ? ("available" as const) : ("full" as const), + ...(options.bundlePrewarm ? { bundlePrewarm: WORKER_BUNDLE_PREWARM_VERSION } : {}), + ...(options.bundleRetention ? { bundleRetention: NODE_WORKER_BUNDLE_RETENTION_VERSION } : {}), + ...(options.bundleStatus ? { bundleStatus: NODE_WORKER_BUNDLE_STATUS_VERSION } : {}), + }, + }); + + const supervisor = createNodeWorkerSupervisor({ + env: nodeEnv, + workspace, + capacity: options.capacity, + capacityWaitMs: options.capacityWaitMs, + onAvailabilityChanged: (available) => { + capacityAvailable = available; + }, + }); + + let host!: PairedNodeWorkerHost; + const onEvent = (event: WireGatewayEvent) => { + if (closing || event.event !== "node.invoke.request" || !client) { + return; + } + const receiver = client; + const frame = event.payload as NodeInvokeRequestPayload; + commands.push(frame.command); + frames.push(frame); + options.onInvoke?.(frame); + const task = handleInvoke(frame, receiver, { current: async () => [] }, undefined, { + workerBundleInstaller: bundleInstaller, + workerSupervisor: supervisor, + workerWorkspace: workspace, + gatewayUrl: options.gateway.wsUrl, + }) + .then(async () => await options.afterInvoke?.(frame, host)) + .catch((error: unknown) => { + invokeErrors.push(error); + }) + .finally(() => invokeTasks.delete(task)); + invokeTasks.add(task); + }; + + const connect = async () => { + if (closing) { + throw new Error("paired worker node is closing"); + } + const open = () => + connectWireClient({ + gateway: options.gateway, + role: "node", + identity, + onEvent, + }); + let next: GatewayClient; + try { + next = await open(); + } catch (error) { + if (!isPairingRequired(error)) { + throw error; + } + await approvePairing(options.operator, identity.deviceId); + next = await open(); + } + client = next; + if (await ensureNodeApproved(options.operator, identity.deviceId)) { + await client.stopAndWait({ timeoutMs: 2_000 }); + client = await open(); + } + await client.request(NODE_RUNNER_INVENTORY_UPDATE_METHOD, inventory()); + }; + const drainInvokeTasks = async () => { + while (invokeTasks.size > 0) { + await Promise.allSettled([...invokeTasks]); + } + }; + + host = { + identity, + commands, + frames, + invokeErrors, + supervisor, + bundleInstaller, + workspace, + get client() { + return client; + }, + connect, + async disconnect() { + const current = client; + client = undefined; + await current?.stopAndWait({ timeoutMs: 2_000 }); + }, + async publishInventory() { + if (!client) { + throw new Error("paired worker node is disconnected"); + } + await client.request(NODE_RUNNER_INVENTORY_UPDATE_METHOD, inventory()); + }, + async waitForInvokes() { + await drainInvokeTasks(); + }, + async installedBundleDirectory(bundleHash) { + const namespaces = await fs.readdir(nodeHostRoot, { withFileTypes: true }); + const matches: string[] = []; + for (const entry of namespaces) { + if (!entry.isDirectory()) { + continue; + } + const candidate = path.join(nodeHostRoot, entry.name, "bundles", bundleHash); + try { + if ((await fs.stat(candidate)).isDirectory()) { + matches.push(candidate); + } + } catch { + // This namespace does not own the proof bundle. + } + } + if (matches.length !== 1) { + throw new Error(`expected one proof-owned installed bundle, found ${matches.length}`); + } + return matches[0]!; + }, + async stop() { + closing = true; + const current = client; + client = undefined; + const connectionCleanup = await Promise.allSettled([ + current?.stopAndWait({ timeoutMs: 2_000 }) ?? Promise.resolve(), + ]); + await drainInvokeTasks(); + const cleanup = await Promise.allSettled([supervisor.close()]); + const failures = [...connectionCleanup, ...cleanup].flatMap((result) => + result.status === "rejected" ? [result.reason] : [], + ); + if (failures.length === 1) { + throw failures[0]; + } + if (failures.length > 1) { + throw new AggregateError(failures, "paired node worker cleanup failed"); + } + }, + }; + + await supervisor.initialize(); + await connect(); + await waitForApprovedWireNode(options.operator, identity.deviceId); + return host; +} + +export async function startPairedNodeWorkerGateway(params: { + providerBaseUrl: string; +}): Promise { + return await startQaGatewayChild({ + repoRoot: process.cwd(), + useRepoCli: true, + providerBaseUrl: `${params.providerBaseUrl}/v1`, + providerMode: "mock-openai", + primaryModel: MODEL_REF, + alternateModel: MODEL_REF, + transportBaseUrl: "http://127.0.0.1", + controlUiEnabled: false, + mutateConfig: (config) => ({ + ...config, + nodeHost: { + ...config.nodeHost, + workerRuns: { enabled: true }, + }, + }), + }); +} + +export function wireMessageText(message: unknown): string { + const content = (message as { content?: unknown })?.content; + if (typeof content === "string") { + return content; + } + return Array.isArray(content) + ? content + .flatMap((part) => + part && typeof part === "object" && typeof (part as { text?: unknown }).text === "string" + ? [(part as { text: string }).text] + : [], + ) + .join("") + : ""; +} + +export function bundleInstallFrames(host: PairedNodeWorkerHost): NodeInvokeRequestPayload[] { + return host.frames.filter((frame) => frame.command === NODE_WORKER_BUNDLE_INSTALL_COMMAND); +}