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(); + }); +});