From bf83a4dde8f00e5d330bd00de79cb0232fe6ae09 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Mon, 24 Aug 2026 11:06:52 -0700 Subject: [PATCH] fix(node-host): recover worker hosting after transient initialization failures (#128798) * fix(node-host): retry worker supervisor initialization Clear failed initialization ownership for retry, retain valid container supervisors after transient reconciliation errors, and retry from the runtime until capacity becomes authoritative or shutdown cancels the lifecycle.\n\nRefs #128794 * fix(node-host): preserve permanent container mismatch failures Keep durable engine and daemon-context mismatches on the actionable disabled path while retrying only transient supervisor reconciliation failures.\n\nRefs #128794 * fix(node-host): withdraw hosting on late context mismatch Stop retrying when a later reconciliation attempt proves a permanent container context mismatch, close the supervisor, and withdraw worker hosting from runner inventory.\n\nRefs #128794 --- src/node-host/gateway-platform-identity.ts | 15 + .../node-worker-container-lifecycle.ts | 6 +- .../node-worker-supervisor.recovery.test.ts | 36 +++ src/node-host/node-worker-supervisor.ts | 33 +- src/node-host/runner.ts | 25 +- src/node-host/runtime.test.ts | 153 ---------- src/node-host/runtime.ts | 94 ++++-- src/node-host/runtime.worker-hosting.test.ts | 283 ++++++++++++++++++ 8 files changed, 433 insertions(+), 212 deletions(-) create mode 100644 src/node-host/gateway-platform-identity.ts create mode 100644 src/node-host/runtime.worker-hosting.test.ts diff --git a/src/node-host/gateway-platform-identity.ts b/src/node-host/gateway-platform-identity.ts new file mode 100644 index 000000000000..00e6c006db3d --- /dev/null +++ b/src/node-host/gateway-platform-identity.ts @@ -0,0 +1,15 @@ +export function resolveNodeHostGatewayPlatformIdentity(platform: NodeJS.Platform): { + platform: string; + deviceFamily?: string; +} { + switch (platform) { + case "darwin": + return { platform: "macos", deviceFamily: "Mac" }; + case "win32": + return { platform: "windows", deviceFamily: "Windows" }; + case "linux": + return { platform: "linux", deviceFamily: "Linux" }; + default: + return { platform: "unknown" }; + } +} diff --git a/src/node-host/node-worker-container-lifecycle.ts b/src/node-host/node-worker-container-lifecycle.ts index 0e9858adbacb..e21f62ba18eb 100644 --- a/src/node-host/node-worker-container-lifecycle.ts +++ b/src/node-host/node-worker-container-lifecycle.ts @@ -12,6 +12,8 @@ import { inspectNodeWorkerProcessIdentity } from "./node-worker-process-identity type NodeWorkerContainerOwner = { gatewayNamespace: string; launchId: string }; +export class NodeWorkerContainerContextMismatchError extends Error {} + /** Owns exact container authority and startup cleanup independently of client PIDs. */ export class NodeWorkerContainerLifecycle { constructor( @@ -27,7 +29,7 @@ export class NodeWorkerContainerLifecycle { (receipt.container.engine !== this.engine.id || receipt.container.engineTarget !== this.engine.target) ) { - throw new Error( + throw new NodeWorkerContainerContextMismatchError( `node worker launch ${receipt.launchId} belongs to a different ${receipt.container.engine} engine or daemon; restore its original engine context before enabling worker hosting`, ); } @@ -82,7 +84,7 @@ export class NodeWorkerContainerLifecycle { private requireMatchingEngine(container: NodeWorkerContainerIdentity): NodeWorkerContainerEngine { if (container.engine !== this.engine.id || container.engineTarget !== this.engine.target) { - throw new Error( + throw new NodeWorkerContainerContextMismatchError( `node worker container belongs to a different ${container.engine} engine or daemon context`, ); } diff --git a/src/node-host/node-worker-supervisor.recovery.test.ts b/src/node-host/node-worker-supervisor.recovery.test.ts index b244d4928466..fcb386af3165 100644 --- a/src/node-host/node-worker-supervisor.recovery.test.ts +++ b/src/node-host/node-worker-supervisor.recovery.test.ts @@ -194,6 +194,42 @@ async function waitForIdentityDeath(identity: NodeWorkerProcessIdentity) { } describe("node worker supervisor recovery", () => { + it("coalesces failed initialization and retries reconciliation on the next attempt", async () => { + const { bundleRoot, env } = fixture("node-worker-initialization-retry-"); + const capacitySnapshots: Array<{ total: number; available: number }> = []; + const supervisor = createNodeWorkerSupervisor({ + bundleRoot, + env, + capacity: 2, + onCapacityChanged: (capacity) => capacitySnapshots.push(capacity), + }); + const reconciliation = vi + .spyOn(NodeWorkerLaunchStore.prototype, "listNonterminal") + .mockImplementationOnce(() => { + throw new Error("temporary launch journal failure"); + }); + + try { + const first = supervisor.initialize(); + const concurrent = supervisor.initialize(); + + expect(concurrent).toBe(first); + await expect(first).rejects.toThrow("temporary launch journal failure"); + await expect(supervisor.initialize()).resolves.toBeUndefined(); + expect(reconciliation).toHaveBeenCalledTimes(2); + expect(capacitySnapshots).toEqual([ + { total: 2, available: 0 }, + { total: 2, available: 0 }, + { total: 2, available: 2 }, + ]); + await expect(supervisor.initialize()).resolves.toBeUndefined(); + expect(reconciliation).toHaveBeenCalledTimes(2); + } finally { + reconciliation.mockRestore(); + await supervisor.close().catch(() => undefined); + } + }); + it("atomically adopts pending work only after the previous supervisor is stale", async () => { const { bundleRoot, env, workspaceDir } = fixture("node-worker-stale-pending-"); const supervisor = createNodeWorkerSupervisor({ bundleRoot, env }); diff --git a/src/node-host/node-worker-supervisor.ts b/src/node-host/node-worker-supervisor.ts index 96f35f8d1f71..d355cb0a4325 100644 --- a/src/node-host/node-worker-supervisor.ts +++ b/src/node-host/node-worker-supervisor.ts @@ -103,23 +103,24 @@ class NodeWorkerSupervisor { this.capacity = new NodeWorkerCapacity(this.store, options); } - private requireSupervisorIdentity(): NodeWorkerProcessIdentity { - return (this.supervisorIdentity ??= requireNodeWorkerProcessIdentity(process.pid)); - } - initialize(): Promise { - return (this.initializationPromise ??= this.containerEngine - ? this.initializeContainerHosting() - : this.capacity.initialize(async (receipt) => { - await this.recoverRunning(receipt, false); - })); - } - - private async initializeContainerHosting(): Promise { - await this.containerLifecycle?.initialize(); - await this.capacity.initialize(async (receipt) => { - await this.recoverRunning(receipt, false); + if (this.initializationPromise) { + return this.initializationPromise; + } + const initialization = (async () => { + if (this.containerLifecycle) { + await this.containerLifecycle.initialize(); + } + await this.capacity.initialize(async (receipt) => { + await this.recoverRunning(receipt, false); + }); + })().catch((error: unknown) => { + if (this.initializationPromise === initialization) { + this.initializationPromise = undefined; + } + throw error; }); + return (this.initializationPromise = initialization); } private requireContainerLifecycle(): NodeWorkerContainerLifecycle { @@ -164,7 +165,7 @@ class NodeWorkerSupervisor { return receipt; } } - const supervisor = this.requireSupervisorIdentity(); + const supervisor = (this.supervisorIdentity ??= requireNodeWorkerProcessIdentity(process.pid)); const claimInput = { launchId: input.launchId, planHash, diff --git a/src/node-host/runner.ts b/src/node-host/runner.ts index f8c5d7e1a191..5dae673ab3a1 100644 --- a/src/node-host/runner.ts +++ b/src/node-host/runner.ts @@ -30,6 +30,7 @@ import { resolveNodeHostCloudflareAccess, type NodeHostCloudflareAccessConfig, } from "./gateway-cloudflare-access.js"; +import { resolveNodeHostGatewayPlatformIdentity } from "./gateway-platform-identity.js"; import { coerceNodeInvokeCancelPayload, coerceNodeInvokeInputPayload, @@ -58,22 +59,6 @@ type NodeHostRunOptions = { installedAppsSharing?: boolean; }; -function resolveNodeHostGatewayPlatformIdentity(platform: NodeJS.Platform): { - platform: string; - deviceFamily?: string; -} { - switch (platform) { - case "darwin": - return { platform: "macos", deviceFamily: "Mac" }; - case "win32": - return { platform: "windows", deviceFamily: "Windows" }; - case "linux": - return { platform: "linux", deviceFamily: "Linux" }; - default: - return { platform: "unknown" }; - } -} - function writeStderrLine(message: string): void { process.stderr.write(`${message}\n`); } @@ -286,6 +271,7 @@ export async function runNodeHost(opts: NodeHostRunOptions): Promise { forceWorkerRuns: opts.forceWorkerRuns, installedAppsSharingEnabled: config.installedAppsSharing, }); + let workerHostingEnabled = preparedRuntime.workerHostingEnabled; if (preparedRuntime.workerHostingDisabledReason) { writeStderrLine( `node host worker hosting disabled: ${preparedRuntime.workerHostingDisabledReason}`, @@ -517,7 +503,7 @@ export async function runNodeHost(opts: NodeHostRunOptions): Promise { { protocolFeatures: [NODE_WORKER_SUPERVISOR_PROTOCOL_FEATURE], workerHost: - preparedRuntime.workerHostingEnabled && workerCapacity + workerHostingEnabled && workerCapacity ? { enabled: true, capacity: workerCapacity, @@ -675,6 +661,11 @@ export async function runNodeHost(opts: NodeHostRunOptions): Promise { workerCapacity = capacity; publishRunnerInventory(); }, + onWorkerHostingDisabled: (reason) => { + workerHostingEnabled = false; + writeStderrLine(`node host worker hosting disabled: ${reason}`); + publishRunnerInventory(); + }, onManifestChanged: (manifest) => { // Manifest changes force a reconnect. Retire the current publication queue // now so it cannot drain against the closing connection. diff --git a/src/node-host/runtime.test.ts b/src/node-host/runtime.test.ts index 41282345beab..54db1d1321cc 100644 --- a/src/node-host/runtime.test.ts +++ b/src/node-host/runtime.test.ts @@ -3,7 +3,6 @@ import { NODE_DEVICE_APPS_COMMAND } from "../infra/node-commands.js"; import type { OpenClawPluginNodeHostCommandIo } from "../plugins/types.js"; import { NODE_DESKTOP_STREAM_COMMAND } from "../shared/node-desktop-stream.js"; import type { NodeHostClient } from "./client.js"; -import { createNodeWorkerSupervisor } from "./node-worker-supervisor.js"; import { listRegisteredNodeHostCapsAndCommands } from "./plugin-node-host.js"; import { prepareNodeHostRuntime } from "./runtime.js"; @@ -13,11 +12,6 @@ const mocks = vi.hoisted(() => { closeMcp, closeWorkerSupervisor: vi.fn(async () => undefined), initializeWorkerSupervisor: vi.fn(async () => undefined), - resolveContainerEngine: vi.fn(async (_options?: { env?: NodeJS.ProcessEnv }) => ({ - id: "docker" as const, - command: "docker", - target: "e".repeat(64), - })), handleInvoke: vi.fn(async () => undefined), progressStartHeartbeats: vi.fn(), progressWrite: vi.fn(async (_chunk: string) => undefined), @@ -50,10 +44,6 @@ vi.mock("./node-invoke-progress.js", () => ({ })), })); -vi.mock("./node-worker-container-engine.js", () => ({ - resolveNodeWorkerContainerEngine: mocks.resolveContainerEngine, -})); - vi.mock("./node-worker-supervisor.js", () => ({ createNodeWorkerSupervisor: vi.fn(() => ({ initialize: mocks.initializeWorkerSupervisor, @@ -139,149 +129,6 @@ function holdInvoke(onCommand?: (io: OpenClawPluginNodeHostCommandIo) => void) { }; } -describe("node-host worker manifest", () => { - it("allows environment-managed processes to force worker hosting without durable config", async () => { - const prepared = await prepareNodeHostRuntime({ - config: { nodeHost: { skills: { enabled: false }, workerRuns: { enabled: false } } }, - env: { PATH: "/usr/bin" }, - enableWorkerRuns: true, - forceWorkerRuns: true, - }); - - expect(prepared.workerHostingEnabled).toBe(true); - }); - - it("keeps local consent separate from connection metadata", async () => { - const prepared = await prepareNodeHostRuntime({ - config: { nodeHost: { skills: { enabled: false }, workerRuns: { enabled: true } } }, - env: { PATH: "/usr/bin" }, - enableWorkerRuns: true, - }); - - expect(prepared.workerHostingEnabled).toBe(true); - expect(prepared.manifest).not.toHaveProperty("workerRuns"); - expect(mocks.resolveContainerEngine).not.toHaveBeenCalled(); - }); - - it("disables container-isolated hosting and records why when no engine is usable", async () => { - const reason = - "Container-isolated node workers require Docker or Podman; install and start an engine."; - mocks.resolveContainerEngine.mockRejectedValueOnce(new Error(reason)); - - const prepared = await prepareNodeHostRuntime({ - config: { - nodeHost: { - skills: { enabled: false }, - workerRuns: { enabled: true, isolation: "container" }, - }, - }, - env: { PATH: "/usr/bin" }, - enableWorkerRuns: true, - }); - - expect(prepared.workerHostingEnabled).toBe(false); - expect(prepared.workerHostingDisabledReason).toBe(reason); - const runtime = prepared.start({ - client: { request: vi.fn(async () => ({})) } as unknown as NodeHostClient, - }); - expect(createNodeWorkerSupervisor).not.toHaveBeenCalled(); - await runtime.close(); - }); - - it("disables container-isolated hosting on Windows before probing or advertising an engine", async () => { - const prepared = await prepareNodeHostRuntime({ - config: { - nodeHost: { - skills: { enabled: false }, - workerRuns: { enabled: true, isolation: "container" }, - }, - }, - env: { PATH: "/usr/bin" }, - enableWorkerRuns: true, - platform: "win32", - }); - - expect(prepared.workerHostingEnabled).toBe(false); - expect(prepared.workerHostingDisabledReason).toMatch(/windows.*(?:linux|macos)/iu); - expect(mocks.resolveContainerEngine).not.toHaveBeenCalled(); - expect(createNodeWorkerSupervisor).not.toHaveBeenCalled(); - const runtime = prepared.start({ - client: { request: vi.fn(async () => ({})) } as unknown as NodeHostClient, - }); - expect(createNodeWorkerSupervisor).not.toHaveBeenCalled(); - await runtime.close(); - }); - - it("resolves the container engine once and passes its exact identity to the supervisor", async () => { - mocks.initializeWorkerSupervisor.mockImplementationOnce(async () => { - const options = vi.mocked(createNodeWorkerSupervisor).mock.calls[0]?.[0]; - options?.onCapacityChanged?.({ total: 3, available: 0 }); - options?.onCapacityChanged?.({ total: 3, available: 3 }); - }); - const prepared = await prepareNodeHostRuntime({ - config: { - nodeHost: { - skills: { enabled: false }, - workerRuns: { - enabled: true, - isolation: "container", - containerImage: "registry.example/openclaw-worker:22", - }, - }, - }, - env: { PATH: "/usr/bin" }, - enableWorkerRuns: true, - }); - - expect(prepared.workerHostingEnabled).toBe(true); - expect(mocks.resolveContainerEngine).toHaveBeenCalledOnce(); - expect(mocks.initializeWorkerSupervisor).toHaveBeenCalledOnce(); - const onRunnerCapacityChanged = vi.fn(); - const runtime = prepared.start({ - client: { request: vi.fn(async () => ({})) } as unknown as NodeHostClient, - onRunnerCapacityChanged, - }); - expect(createNodeWorkerSupervisor).toHaveBeenCalledWith( - expect.objectContaining({ - containerEngine: { id: "docker", command: "docker", target: "e".repeat(64) }, - containerImage: "registry.example/openclaw-worker:22", - }), - ); - expect(mocks.resolveContainerEngine).toHaveBeenCalledOnce(); - expect(mocks.initializeWorkerSupervisor).toHaveBeenCalledOnce(); - expect(onRunnerCapacityChanged).toHaveBeenCalledExactlyOnceWith({ total: 3, available: 3 }); - await runtime.close(); - }); - - it("fails closed when container launch reconciliation cannot establish safe ownership", async () => { - mocks.initializeWorkerSupervisor.mockRejectedValueOnce(new Error("orphan sweep failed")); - - const prepared = await prepareNodeHostRuntime({ - config: { - nodeHost: { - skills: { enabled: false }, - workerRuns: { enabled: true, isolation: "container" }, - }, - }, - env: { PATH: "/usr/bin" }, - enableWorkerRuns: true, - }); - - expect(prepared.workerHostingEnabled).toBe(false); - expect(prepared.workerHostingDisabledReason).toContain("orphan sweep failed"); - expect(mocks.closeWorkerSupervisor).toHaveBeenCalledOnce(); - const onRunnerCapacityChanged = vi.fn(); - const runtime = prepared.start({ - client: { request: vi.fn(async () => ({})) } as unknown as NodeHostClient, - onRunnerCapacityChanged, - }); - expect(createNodeWorkerSupervisor).toHaveBeenCalledOnce(); - expect(onRunnerCapacityChanged).not.toHaveBeenCalled(); - await runtime.close(); - expect(mocks.closeWorkerSupervisor).toHaveBeenCalledOnce(); - }); -}); - describe("node-host invocation cancellation", () => { it("cancels ordinary node invocations", async () => { const held = holdInvoke(); diff --git a/src/node-host/runtime.ts b/src/node-host/runtime.ts index f6410b483721..2e323d320ffc 100644 --- a/src/node-host/runtime.ts +++ b/src/node-host/runtime.ts @@ -32,6 +32,7 @@ import { buildNodeEventParams } from "./node-event-params.js"; import { createNodeInvokeProgressWriter } from "./node-invoke-progress.js"; import { NodeWorkerBundleInstaller } from "./node-worker-bundle-installer.js"; import { resolveNodeWorkerContainerEngine } from "./node-worker-container-engine.js"; +import { NodeWorkerContainerContextMismatchError } from "./node-worker-container-lifecycle.js"; import { createNodeWorkerSupervisor } from "./node-worker-supervisor.js"; import { NodeWorkerWorkspaceRuntime } from "./node-worker-workspace.js"; import { @@ -44,6 +45,7 @@ import { import { scanNodeHostedSkills } from "./skills.js"; const DEFAULT_NODE_PATH = "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin"; +const WORKER_INITIALIZATION_RETRY_MS = 5_000; type NodeHostManifest = { caps: string[]; @@ -67,6 +69,7 @@ type PreparedNodeHostRuntime = { onInventoryChanged?: (inventory: NodeHostInventory) => void; onManifestChanged?: (manifest: NodeHostManifest) => void; onRunnerCapacityChanged?: (capacity: NodeWorkerCapacitySnapshot) => void; + onWorkerHostingDisabled?: (reason: string) => void; }): ActiveNodeHostRuntime; }; @@ -296,10 +299,25 @@ export async function prepareNodeHostRuntime(params?: { let preparedContainerWorkspace: NodeWorkerWorkspaceRuntime | undefined; let preparedContainerSupervisor: ReturnType | undefined; let preparedContainerCapacity: NodeWorkerCapacitySnapshot | undefined; + let preparedContainerInitialized = false; let publishContainerCapacity: ((capacity: NodeWorkerCapacitySnapshot) => void) | undefined; let workerHostingDisabledReason: string | undefined; + const disablePreparedContainerHosting = async (error: unknown) => { + let failure = error; + try { + await preparedContainerSupervisor?.close(); + } catch (closeError) { + if (closeError !== error) { + failure = new Error(`${String(error)}; supervisor cleanup failed: ${String(closeError)}`); + } + } + workerRunsEnabled = false; + preparedContainerWorkspace = undefined; + preparedContainerSupervisor = undefined; + preparedContainerCapacity = undefined; + workerHostingDisabledReason = failure instanceof Error ? failure.message : String(failure); + }; if (workerRunsEnabled && config.nodeHost?.workerRuns?.isolation === "container") { - let engineResolved = false; try { if (platform === "win32") { throw new Error( @@ -307,7 +325,6 @@ export async function prepareNodeHostRuntime(params?: { ); } const containerEngine = await resolveNodeWorkerContainerEngine({ env }); - engineResolved = true; preparedContainerWorkspace = new NodeWorkerWorkspaceRuntime({ env }); preparedContainerSupervisor = createNodeWorkerSupervisor({ env, @@ -322,25 +339,19 @@ export async function prepareNodeHostRuntime(params?: { publishContainerCapacity?.(capacity); }, }); - // Container ownership and orphan cleanup must succeed before any worker capacity is advertised. - await preparedContainerSupervisor.initialize(); - } catch (error) { - let failure = error; try { - await preparedContainerSupervisor?.close(); - } catch (closeError) { - if (closeError !== error) { - failure = new Error(`${String(error)}; supervisor cleanup failed: ${String(closeError)}`); + // Container ownership and orphan cleanup must precede positive capacity publication. + await preparedContainerSupervisor.initialize(); + preparedContainerInitialized = true; + } catch (error) { + if (error instanceof NodeWorkerContainerContextMismatchError) { + await disablePreparedContainerHosting(error); + } else { + logDebug(`node-host: worker capacity reconciliation failed: ${String(error)}`); } } - workerRunsEnabled = false; - preparedContainerWorkspace = undefined; - preparedContainerSupervisor = undefined; - preparedContainerCapacity = undefined; - const detail = failure instanceof Error ? failure.message : String(failure); - workerHostingDisabledReason = engineResolved - ? `container worker reconciliation failed: ${detail}; inspect the container engine and worker launch journal before retrying` - : detail; + } catch (error) { + await disablePreparedContainerHosting(error); } } const skills = config.nodeHost?.skills?.enabled === false ? null : scanNodeHostedSkills(); @@ -377,15 +388,24 @@ export async function prepareNodeHostRuntime(params?: { workerHostingEnabled: workerRunsEnabled, ...(workerHostingDisabledReason ? { workerHostingDisabledReason } : {}), initialInventory, - start({ client, onInventoryChanged, onManifestChanged, onRunnerCapacityChanged }) { + start({ + client, + onInventoryChanged, + onManifestChanged, + onRunnerCapacityChanged, + onWorkerHostingDisabled, + }) { const mcpAbort = new AbortController(); + let closing = false; + let closePromise: Promise | undefined; + let initializationRetry: ReturnType | undefined; const workerWorkspace = preparedContainerWorkspace ?? (workerRunsEnabled ? new NodeWorkerWorkspaceRuntime({ env }) : undefined); const workerBundleInstaller = workerRunsEnabled ? new NodeWorkerBundleInstaller({ env }) : undefined; - const workerSupervisor = + let workerSupervisor = preparedContainerSupervisor ?? (workerRunsEnabled ? createNodeWorkerSupervisor({ @@ -400,10 +420,34 @@ export async function prepareNodeHostRuntime(params?: { if (preparedContainerCapacity) { onRunnerCapacityChanged?.(preparedContainerCapacity); } - } else if (workerSupervisor) { - void workerSupervisor.initialize().catch((error: unknown) => { + } + const initializeWorkerSupervisor = () => { + const supervisor = workerSupervisor; + if (!supervisor || closing) { + return; + } + void supervisor.initialize().catch(async (error: unknown) => { logDebug(`node-host: worker capacity reconciliation failed: ${String(error)}`); + if (closing || workerSupervisor !== supervisor) { + return; + } + if (error instanceof NodeWorkerContainerContextMismatchError) { + workerSupervisor = undefined; + onWorkerHostingDisabled?.(error.message); + await supervisor.close().catch((closeError: unknown) => { + logDebug(`node-host: worker supervisor cleanup failed: ${String(closeError)}`); + }); + return; + } + initializationRetry = setTimeout(() => { + initializationRetry = undefined; + initializeWorkerSupervisor(); + }, WORKER_INITIALIZATION_RETRY_MS); + initializationRetry.unref?.(); }); + }; + if (workerSupervisor && !preparedContainerInitialized) { + initializeWorkerSupervisor(); } const skillBins = new SkillBinsCache(client, pathEnv); const activeInvokes = new Map(); @@ -428,8 +472,6 @@ export async function prepareNodeHostRuntime(params?: { } | undefined; let manager: NodeHostMcpManager | undefined; - let closing = false; - let closePromise: Promise | undefined; const publishInventory = () => onInventoryChanged?.( createInventory(skills, currentPluginNodeHost.nodePluginTools, manager?.descriptors), @@ -615,6 +657,10 @@ export async function prepareNodeHostRuntime(params?: { return closePromise; } closing = true; + if (initializationRetry) { + clearTimeout(initializationRetry); + initializationRetry = undefined; + } this.cancelAll(); const preludeErrors: unknown[] = []; try { diff --git a/src/node-host/runtime.worker-hosting.test.ts b/src/node-host/runtime.worker-hosting.test.ts new file mode 100644 index 000000000000..7e3438e13508 --- /dev/null +++ b/src/node-host/runtime.worker-hosting.test.ts @@ -0,0 +1,283 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import type { NodeHostClient } from "./client.js"; +import { NodeWorkerContainerContextMismatchError } from "./node-worker-container-lifecycle.js"; +import { createNodeWorkerSupervisor } from "./node-worker-supervisor.js"; +import { prepareNodeHostRuntime } from "./runtime.js"; + +const mocks = vi.hoisted(() => ({ + closeWorkerSupervisor: vi.fn(async () => undefined), + initializeWorkerSupervisor: vi.fn(async () => undefined), + resolveContainerEngine: vi.fn(async (_options?: { env?: NodeJS.ProcessEnv }) => ({ + id: "docker" as const, + command: "docker", + target: "e".repeat(64), + })), +})); + +vi.mock("../infra/path-env.js", () => ({ ensureOpenClawCliOnPath: vi.fn() })); +vi.mock("./invoke.js", () => ({ handleInvoke: vi.fn(async () => undefined) })); +vi.mock("./mcp.js", () => ({ + startNodeHostMcpManager: vi.fn(async () => ({ + descriptors: [], + close: vi.fn(async () => undefined), + })), +})); +vi.mock("./node-worker-container-engine.js", () => ({ + resolveNodeWorkerContainerEngine: mocks.resolveContainerEngine, +})); +vi.mock("./node-worker-supervisor.js", () => ({ + createNodeWorkerSupervisor: vi.fn(() => ({ + initialize: mocks.initializeWorkerSupervisor, + close: mocks.closeWorkerSupervisor, + })), +})); +vi.mock("./node-worker-workspace.js", () => ({ + NodeWorkerWorkspaceRuntime: class { + readonly exec = vi.fn(); + }, +})); +vi.mock("./plugin-node-host.js", () => ({ + ensureNodeHostPluginRegistry: vi.fn(async () => undefined), + listRegisteredNodeHostCapsAndCommands: vi.fn(() => ({ + caps: [], + commands: [], + nodePluginTools: [], + })), +})); +vi.mock("./skills.js", () => ({ scanNodeHostedSkills: vi.fn(() => []) })); + +const client = { request: vi.fn(async () => ({})) } as unknown as NodeHostClient; + +beforeEach(() => { + vi.clearAllMocks(); + mocks.closeWorkerSupervisor.mockReset().mockResolvedValue(undefined); + mocks.initializeWorkerSupervisor.mockReset().mockResolvedValue(undefined); +}); + +afterEach(() => { + vi.useRealTimers(); +}); + +function prepareWorkerRuntime(isolation?: "container") { + return prepareNodeHostRuntime({ + config: { + nodeHost: { + skills: { enabled: false }, + workerRuns: { enabled: true, ...(isolation ? { isolation } : {}) }, + }, + }, + env: { PATH: "/usr/bin" }, + enableWorkerRuns: true, + }); +} + +describe("node-host worker manifest", () => { + it("allows environment-managed processes to force worker hosting without durable config", async () => { + const prepared = await prepareNodeHostRuntime({ + config: { nodeHost: { skills: { enabled: false }, workerRuns: { enabled: false } } }, + env: { PATH: "/usr/bin" }, + enableWorkerRuns: true, + forceWorkerRuns: true, + }); + + expect(prepared.workerHostingEnabled).toBe(true); + }); + + it("keeps local consent separate from connection metadata", async () => { + const prepared = await prepareWorkerRuntime(); + + expect(prepared.workerHostingEnabled).toBe(true); + expect(prepared.manifest).not.toHaveProperty("workerRuns"); + expect(mocks.resolveContainerEngine).not.toHaveBeenCalled(); + }); + + it("disables container-isolated hosting and records why when no engine is usable", async () => { + const reason = + "Container-isolated node workers require Docker or Podman; install and start an engine."; + mocks.resolveContainerEngine.mockRejectedValueOnce(new Error(reason)); + + const prepared = await prepareWorkerRuntime("container"); + + expect(prepared.workerHostingEnabled).toBe(false); + expect(prepared.workerHostingDisabledReason).toBe(reason); + const runtime = prepared.start({ client }); + expect(createNodeWorkerSupervisor).not.toHaveBeenCalled(); + await runtime.close(); + }); + + it("disables container-isolated hosting on Windows before probing or advertising an engine", async () => { + const prepared = await prepareNodeHostRuntime({ + config: { + nodeHost: { + skills: { enabled: false }, + workerRuns: { enabled: true, isolation: "container" }, + }, + }, + env: { PATH: "/usr/bin" }, + enableWorkerRuns: true, + platform: "win32", + }); + + expect(prepared.workerHostingEnabled).toBe(false); + expect(prepared.workerHostingDisabledReason).toMatch(/windows.*(?:linux|macos)/iu); + expect(mocks.resolveContainerEngine).not.toHaveBeenCalled(); + expect(createNodeWorkerSupervisor).not.toHaveBeenCalled(); + const runtime = prepared.start({ client }); + expect(createNodeWorkerSupervisor).not.toHaveBeenCalled(); + await runtime.close(); + }); + + it("resolves the container engine once and passes its exact identity to the supervisor", async () => { + mocks.initializeWorkerSupervisor.mockImplementationOnce(async () => { + const options = vi.mocked(createNodeWorkerSupervisor).mock.calls[0]?.[0]; + options?.onCapacityChanged?.({ total: 3, available: 0 }); + options?.onCapacityChanged?.({ total: 3, available: 3 }); + }); + const prepared = await prepareNodeHostRuntime({ + config: { + nodeHost: { + skills: { enabled: false }, + workerRuns: { + enabled: true, + isolation: "container", + containerImage: "registry.example/openclaw-worker:22", + }, + }, + }, + env: { PATH: "/usr/bin" }, + enableWorkerRuns: true, + }); + + expect(prepared.workerHostingEnabled).toBe(true); + expect(mocks.resolveContainerEngine).toHaveBeenCalledOnce(); + expect(mocks.initializeWorkerSupervisor).toHaveBeenCalledOnce(); + const onRunnerCapacityChanged = vi.fn(); + const runtime = prepared.start({ client, onRunnerCapacityChanged }); + + expect(createNodeWorkerSupervisor).toHaveBeenCalledWith( + expect.objectContaining({ + containerEngine: { id: "docker", command: "docker", target: "e".repeat(64) }, + containerImage: "registry.example/openclaw-worker:22", + }), + ); + expect(mocks.resolveContainerEngine).toHaveBeenCalledOnce(); + expect(mocks.initializeWorkerSupervisor).toHaveBeenCalledOnce(); + expect(onRunnerCapacityChanged).toHaveBeenCalledExactlyOnceWith({ total: 3, available: 3 }); + await runtime.close(); + }); + + it("retains container hosting after failed reconciliation and recovers its capacity on start", async () => { + mocks.initializeWorkerSupervisor + .mockRejectedValueOnce(new Error("orphan sweep failed")) + .mockImplementationOnce(async () => { + const options = vi.mocked(createNodeWorkerSupervisor).mock.calls[0]?.[0]; + options?.onCapacityChanged?.({ total: 2, available: 0 }); + options?.onCapacityChanged?.({ total: 2, available: 2 }); + }); + + const prepared = await prepareWorkerRuntime("container"); + + expect(prepared.workerHostingEnabled).toBe(true); + expect(prepared.workerHostingDisabledReason).toBeUndefined(); + expect(mocks.closeWorkerSupervisor).not.toHaveBeenCalled(); + expect(mocks.initializeWorkerSupervisor).toHaveBeenCalledOnce(); + const onRunnerCapacityChanged = vi.fn(); + const runtime = prepared.start({ client, onRunnerCapacityChanged }); + + await vi.waitFor(() => + expect(onRunnerCapacityChanged).toHaveBeenLastCalledWith({ total: 2, available: 2 }), + ); + expect(onRunnerCapacityChanged.mock.calls).toEqual([ + [{ total: 2, available: 0 }], + [{ total: 2, available: 2 }], + ]); + expect(mocks.initializeWorkerSupervisor).toHaveBeenCalledTimes(2); + expect(createNodeWorkerSupervisor).toHaveBeenCalledOnce(); + await runtime.close(); + expect(mocks.closeWorkerSupervisor).toHaveBeenCalledOnce(); + }); + + it("keeps a container engine-context mismatch permanently and actionably disabled", async () => { + const mismatch = new NodeWorkerContainerContextMismatchError( + "node worker launch launch-1 belongs to a different docker engine or daemon; restore its original engine context before enabling worker hosting", + ); + mocks.initializeWorkerSupervisor.mockRejectedValueOnce(mismatch); + + const prepared = await prepareWorkerRuntime("container"); + + expect(prepared.workerHostingEnabled).toBe(false); + expect(prepared.workerHostingDisabledReason).toBe(mismatch.message); + expect(mocks.initializeWorkerSupervisor).toHaveBeenCalledOnce(); + expect(mocks.closeWorkerSupervisor).toHaveBeenCalledOnce(); + const onRunnerCapacityChanged = vi.fn(); + const runtime = prepared.start({ client, onRunnerCapacityChanged }); + + expect(createNodeWorkerSupervisor).toHaveBeenCalledOnce(); + expect(onRunnerCapacityChanged).not.toHaveBeenCalled(); + await runtime.close(); + expect(mocks.closeWorkerSupervisor).toHaveBeenCalledOnce(); + }); + + it("disables a retrying container supervisor when a later attempt finds a context mismatch", async () => { + const mismatch = new NodeWorkerContainerContextMismatchError( + "node worker launch launch-1 belongs to a different docker engine or daemon; restore its original engine context before enabling worker hosting", + ); + mocks.initializeWorkerSupervisor + .mockRejectedValueOnce(new Error("launch journal temporarily unavailable")) + .mockRejectedValueOnce(mismatch); + + const prepared = await prepareWorkerRuntime("container"); + + expect(prepared.workerHostingEnabled).toBe(true); + const onWorkerHostingDisabled = vi.fn(); + const runtime = prepared.start({ client, onWorkerHostingDisabled }); + + await vi.waitFor(() => + expect(onWorkerHostingDisabled).toHaveBeenCalledExactlyOnceWith(mismatch.message), + ); + expect(mocks.initializeWorkerSupervisor).toHaveBeenCalledTimes(2); + expect(mocks.closeWorkerSupervisor).toHaveBeenCalledOnce(); + await runtime.close(); + expect(mocks.closeWorkerSupervisor).toHaveBeenCalledOnce(); + }); + + it("retries non-container reconciliation after a bounded delay before publishing capacity", async () => { + vi.useFakeTimers(); + mocks.initializeWorkerSupervisor + .mockRejectedValueOnce(new Error("launch journal temporarily unavailable")) + .mockImplementationOnce(async () => { + const options = vi.mocked(createNodeWorkerSupervisor).mock.calls[0]?.[0]; + options?.onCapacityChanged?.({ total: 2, available: 2 }); + }); + const prepared = await prepareWorkerRuntime(); + const onRunnerCapacityChanged = vi.fn(); + const runtime = prepared.start({ client, onRunnerCapacityChanged }); + + await vi.advanceTimersByTimeAsync(0); + expect(mocks.initializeWorkerSupervisor).toHaveBeenCalledOnce(); + expect(onRunnerCapacityChanged).not.toHaveBeenCalled(); + await vi.advanceTimersByTimeAsync(4_999); + expect(mocks.initializeWorkerSupervisor).toHaveBeenCalledOnce(); + await vi.advanceTimersByTimeAsync(1); + + expect(mocks.initializeWorkerSupervisor).toHaveBeenCalledTimes(2); + expect(onRunnerCapacityChanged).toHaveBeenCalledExactlyOnceWith({ total: 2, available: 2 }); + await runtime.close(); + }); + + it("cancels pending reconciliation retries and closes its supervisor exactly once", async () => { + vi.useFakeTimers(); + mocks.initializeWorkerSupervisor.mockRejectedValueOnce(new Error("launch journal unavailable")); + const prepared = await prepareWorkerRuntime(); + const runtime = prepared.start({ client }); + await vi.advanceTimersByTimeAsync(0); + + const closing = runtime.close(); + expect(runtime.close()).toBe(closing); + await closing; + await vi.advanceTimersByTimeAsync(10_000); + + expect(mocks.initializeWorkerSupervisor).toHaveBeenCalledOnce(); + expect(mocks.closeWorkerSupervisor).toHaveBeenCalledOnce(); + }); +});