diff --git a/extensions/lmstudio/src/stream.test.ts b/extensions/lmstudio/src/stream.test.ts index 3776383deb5d..7f552cb80711 100644 --- a/extensions/lmstudio/src/stream.test.ts +++ b/extensions/lmstudio/src/stream.test.ts @@ -490,6 +490,51 @@ describe("lmstudio stream wrapper", () => { expect(ensureLmstudioModelLoadedMock).toHaveBeenCalledTimes(1); }); + it("cancels one shared preload waiter without cancelling another inference", async () => { + let resolvePreload: (() => void) | undefined; + ensureLmstudioModelLoadedMock.mockImplementationOnce( + () => + new Promise((resolve) => { + resolvePreload = resolve; + }), + ); + const baseStream = buildDoneStreamFn(); + const wrapped = createWrappedLmstudioStream(baseStream); + const controller = new AbortController(); + const first = collectEvents( + runWrappedLmstudioStream(wrapped, { contextWindow: 32_768 }, { signal: controller.signal }), + ); + let firstOutcome: string | undefined; + void first.then( + () => { + firstOutcome = "completed"; + }, + (error: unknown) => { + firstOutcome = error instanceof Error ? error.name : "unknown"; + }, + ); + const second = collectEvents(runWrappedLmstudioStream(wrapped, { contextWindow: 32_768 })); + + try { + await vi.waitFor(() => expect(resolvePreload).toBeDefined()); + controller.abort(new DOMException("inference cancelled", "AbortError")); + + await vi.waitFor(() => expect(firstOutcome).toBe("AbortError"), { + timeout: 250, + }); + expect(baseStream).not.toHaveBeenCalled(); + expect(ensureLmstudioModelLoadedMock).toHaveBeenCalledTimes(1); + + resolvePreload?.(); + + expectSingleDoneEvent(await second); + expect(baseStream).toHaveBeenCalledTimes(1); + } finally { + resolvePreload?.(); + await Promise.allSettled([first, second]); + } + }); + it("skips preload on the second attempt while the failure backoff is active", async () => { ensureLmstudioModelLoadedMock.mockRejectedValue(new Error("out of memory")); const baseStream = buildDoneStreamFn(); diff --git a/extensions/lmstudio/src/stream.ts b/extensions/lmstudio/src/stream.ts index 6d08347edee1..75639a529727 100644 --- a/extensions/lmstudio/src/stream.ts +++ b/extensions/lmstudio/src/stream.ts @@ -179,6 +179,37 @@ function createPreloadKey(params: { return `${params.baseUrl}::${params.modelKey}::${params.requestedContextLength ?? "default"}`; } +function toLmstudioPreloadError(reason: unknown, message: string): Error { + return reason instanceof Error ? reason : new Error(message, { cause: reason }); +} + +function waitForLmstudioPreload( + preload: Promise, + signal?: AbortSignal, +): Promise { + if (!signal) { + return preload; + } + if (signal.aborted) { + return Promise.reject(toLmstudioPreloadError(signal.reason, "LM Studio preload aborted")); + } + return new Promise((resolve, reject) => { + const onAbort = () => + reject(toLmstudioPreloadError(signal.reason, "LM Studio preload aborted")); + signal.addEventListener("abort", onAbort, { once: true }); + void preload.then( + (modelKey) => { + signal.removeEventListener("abort", onAbort); + resolve(modelKey); + }, + (error: unknown) => { + signal.removeEventListener("abort", onAbort); + reject(toLmstudioPreloadError(error, "LM Studio model preload failed")); + }, + ); + }); +} + async function ensureLmstudioModelLoadedBestEffort(params: { baseUrl: string; modelKey: string; @@ -291,8 +322,11 @@ export function wrapLmstudioInferencePreload(ctx: ProviderWrapStreamFnContext): let resolvedModelKey: string | undefined; if (preloadPromise) { try { - resolvedModelKey = await preloadPromise; + resolvedModelKey = await waitForLmstudioPreload(preloadPromise, options?.signal); } catch (error) { + // A caller owns its wait, not the shared model load needed by other + // in-flight requests; cancellation must never become preload backoff. + options?.signal?.throwIfAborted(); const annotated = error as { cause?: unknown; consecutiveFailures?: number; diff --git a/extensions/ollama/src/node-inference.test.ts b/extensions/ollama/src/node-inference.test.ts index e68f9723574a..7e0e5dd35a27 100644 --- a/extensions/ollama/src/node-inference.test.ts +++ b/extensions/ollama/src/node-inference.test.ts @@ -27,6 +27,7 @@ async function withOllamaServer( chatRequests: Record[], showRequests: string[], ) => Promise, + options?: { models: Array> }, ): Promise { const chatRequests: Record[] = []; const showRequests: string[] = []; @@ -35,7 +36,7 @@ async function withOllamaServer( if (request.url === "/api/tags") { response.end( JSON.stringify({ - models: [ + models: options?.models ?? [ { name: "remote:cloud", size: 1, @@ -84,7 +85,7 @@ async function withOllamaServer( response.end(JSON.stringify({ error: "show failed" })); return; } - const embedding = body.name === "embedding:latest"; + const embedding = body.name?.startsWith("embedding") === true; response.end( JSON.stringify({ capabilities: embedding ? ["embedding"] : ["completion", "tools"], @@ -170,6 +171,29 @@ describe("Ollama node host inference", () => { }); }); + it("discovers a local chat model after 200 embedding-only node models", async () => { + const models = [ + ...Array.from({ length: 200 }, (_, index) => ({ name: `embedding-${index}:latest` })), + { name: "chat:small", size: 500 }, + ]; + + await withOllamaServer( + async (baseUrl, _chatRequests, showRequests) => { + const result = JSON.parse(await commandByName(baseUrl, OLLAMA_MODELS_COMMAND).handle()) as { + provider: string; + models: Array<{ name: string }>; + }; + + expect(result).toEqual({ + provider: "ollama", + models: [expect.objectContaining({ name: "chat:small" })], + }); + expect(showRequests).toHaveLength(201); + }, + { models }, + ); + }); + it("runs bounded chat and returns compact usage", async () => { await withOllamaServer(async (baseUrl, chatRequests, showRequests) => { const result = JSON.parse( diff --git a/extensions/ollama/src/node-inference.ts b/extensions/ollama/src/node-inference.ts index 953fc0444951..7fa0b07fa833 100644 --- a/extensions/ollama/src/node-inference.ts +++ b/extensions/ollama/src/node-inference.ts @@ -22,6 +22,7 @@ import { Type } from "typebox"; import { OLLAMA_DEFAULT_BASE_URL } from "./defaults.js"; import { buildOllamaBaseUrlSsrFPolicy, + enrichOllamaCompletionModels, enrichOllamaModelsWithContext, fetchOllamaModels, isOllamaCloudModel, @@ -41,7 +42,6 @@ const MAX_INFERENCE_TIMEOUT_MS = 10 * 60_000; const MAX_TOKENS = 8192; const MAX_PROMPT_CHARS = 128_000; const MAX_SYSTEM_PROMPT_CHARS = 32_000; -const MAX_DISCOVERED_MODELS = 200; const MAX_ERROR_BODY_BYTES = 500; type NodeModel = { @@ -185,17 +185,16 @@ async function discoverOllamaNodeModels( if (!discovered.reachable) { throw new Error(`Ollama is not running at ${apiBase}`); } - const localModels = discovered.models - .filter((model) => !model.remote_host?.trim() && !isOllamaCloudModel(model.name)) - .slice(0, MAX_DISCOVERED_MODELS); + const localModels = discovered.models.filter( + (model) => !model.remote_host?.trim() && !isOllamaCloudModel(model.name), + ); const [models, loadedNames] = await Promise.all([ - enrichOllamaModelsWithContext(apiBase, localModels), + // Paired nodes must positively confirm completion; unlike provider catalogs, + // failed or legacy show probes must never expose unrunnable remote commands. + enrichOllamaCompletionModels(apiBase, localModels, { requireCompletionCapability: true }), fetchLoadedModelNames(apiBase), ]); const rows = models - // Nodes advertise only models Ollama positively identifies as chat-capable. - // Failed /api/show probes must not turn embedding models into runnable choices. - .filter((model) => model.capabilities?.includes("completion") === true) .map((model): NodeModel => { const details = model.details; const row: NodeModel = { diff --git a/extensions/ollama/src/provider-models.test.ts b/extensions/ollama/src/provider-models.test.ts index 367b8318f4df..72fa17a98b49 100644 --- a/extensions/ollama/src/provider-models.test.ts +++ b/extensions/ollama/src/provider-models.test.ts @@ -148,6 +148,36 @@ describe("ollama provider models", () => { expect(fetchMock).toHaveBeenCalledTimes(2); }); + it("discovers a chat model after 200 embedding-only catalog entries", async () => { + const embeddingModels = Array.from({ length: 200 }, (_, index) => ({ + name: `embedding-${index}:latest`, + })); + const fetchMock = vi.fn(async (input: string | URL | Request, init?: RequestInit) => { + const url = requestUrl(input); + if (url.endsWith("/api/tags")) { + return jsonResponse({ + models: [...embeddingModels, { name: "qwen-chat:latest" }], + }); + } + if (url.endsWith("/api/show")) { + const body = JSON.parse(requestBodyText(init?.body)) as { name?: string }; + const completion = body.name === "qwen-chat:latest"; + return jsonResponse({ + capabilities: completion ? ["completion", "tools"] : ["embedding"], + model_info: completion ? { "qwen.context_length": 32_768 } : {}, + }); + } + throw new Error(`Unexpected fetch: ${url}`); + }); + vi.stubGlobal("fetch", fetchMock); + + const provider = await buildOllamaProvider("http://127.0.0.1:11434"); + + expect(provider.models?.map((model) => model.id)).toEqual(["qwen-chat:latest"]); + expect(provider.models?.[0]?.contextWindow).toBe(32_768); + expect(fetchMock).toHaveBeenCalledTimes(202); + }); + it("scopes cached show metadata by credential", async () => { const fetchMock = vi.fn(async (input: string | URL | Request, init?: RequestInit) => { const url = requestUrl(input); diff --git a/extensions/ollama/src/provider-models.ts b/extensions/ollama/src/provider-models.ts index b873b1bcb472..aab781fb172f 100644 --- a/extensions/ollama/src/provider-models.ts +++ b/extensions/ollama/src/provider-models.ts @@ -40,6 +40,7 @@ export type OllamaModelWithContext = OllamaTagModel & { const OLLAMA_SHOW_CONCURRENCY = 8; const OLLAMA_CONTEXT_ENRICH_LIMIT = 200; +const MAX_OLLAMA_DISCOVERY_PROBES = OLLAMA_CONTEXT_ENRICH_LIMIT * 4; const MAX_OLLAMA_SHOW_CACHE_ENTRIES = 256; const ollamaModelShowInfoCache = new Map>(); const OLLAMA_ALWAYS_BLOCKED_HOSTNAMES = new Set(["metadata.google.internal"]); @@ -259,6 +260,37 @@ export async function enrichOllamaModelsWithContext( return enriched; } +export async function enrichOllamaCompletionModels( + apiBase: string, + models: OllamaTagModel[], + opts?: { apiKey?: string; requireCompletionCapability?: boolean }, +): Promise { + const completionModels: OllamaModelWithContext[] = []; + const probeLimit = Math.min(models.length, MAX_OLLAMA_DISCOVERY_PROBES); + for ( + let index = 0; + index < probeLimit && completionModels.length < OLLAMA_CONTEXT_ENRICH_LIMIT; + index += OLLAMA_SHOW_CONCURRENCY + ) { + const batch = await enrichOllamaModelsWithContext( + apiBase, + models.slice(index, Math.min(index + OLLAMA_SHOW_CONCURRENCY, probeLimit)), + opts?.apiKey ? { apiKey: opts.apiKey } : undefined, + ); + for (const model of batch) { + const canComplete = model.capabilities?.includes("completion"); + if (!canComplete && (opts?.requireCompletionCapability || model.capabilities)) { + continue; + } + completionModels.push(model); + if (completionModels.length === OLLAMA_CONTEXT_ENRICH_LIMIT) { + break; + } + } + } + return completionModels; +} + export function isOllamaCloudModel(modelName: string | undefined): boolean { return isCloudModelRef(modelName); } @@ -404,11 +436,7 @@ export async function buildOllamaProvider( if (!reachable && !opts?.quiet) { console.warn(`Ollama could not be reached at ${apiBase}.`); } - const discovered = await enrichOllamaModelsWithContext( - apiBase, - models.slice(0, OLLAMA_CONTEXT_ENRICH_LIMIT), - auth, - ); + const discovered = await enrichOllamaCompletionModels(apiBase, models, auth); return { baseUrl: apiBase, api: "ollama", diff --git a/src/node-host/invoke-system-run.test.ts b/src/node-host/invoke-system-run.test.ts index 88ba2c78eb4e..a96dfa01d828 100644 --- a/src/node-host/invoke-system-run.test.ts +++ b/src/node-host/invoke-system-run.test.ts @@ -521,6 +521,7 @@ describe("handleSystemRunInvoke mac app exec host routing", () => { autoReviewer?: ExecAutoReviewer; commitExecAuthorization?: HandleSystemRunInvokeOptions["commitExecAuthorization"]; prepareDelayedApprovalPlan?: boolean; + signal?: AbortSignal; }): Promise<{ runCommand: MockedRunCommand; runViaMacAppExecHost: MockedRunViaMacAppExecHost; @@ -618,6 +619,7 @@ describe("handleSystemRunInvoke mac app exec host routing", () => { skillBins: { current: params.skillBinsCurrent ?? (async () => []), }, + signal: params.signal, execHostEnforced: false, execHostFallbackAllowed: true, resolveExecSecurity: params.resolveExecSecurity ?? (() => params.security ?? "full"), @@ -645,6 +647,61 @@ describe("handleSystemRunInvoke mac app exec host routing", () => { }; } + it("forwards cancellation to locally spawned node commands", async () => { + const controller = new AbortController(); + const result = await runSystemInvoke({ + preferMacAppExecHost: false, + signal: controller.signal, + }); + + expect(result.runCommand.mock.calls[0]?.[4]).toBe(controller.signal); + }); + + it("does not spawn an already-cancelled node command", async () => { + const controller = new AbortController(); + controller.abort(); + + const result = await runSystemInvoke({ + preferMacAppExecHost: false, + signal: controller.signal, + }); + + expect(result.runCommand).not.toHaveBeenCalled(); + expect(result.runViaMacAppExecHost).not.toHaveBeenCalled(); + }); + + it("does not publish a cancelled local command completion", async () => { + const controller = new AbortController(); + const result = await runSystemInvoke({ + preferMacAppExecHost: false, + signal: controller.signal, + runCommand: async () => { + controller.abort(); + return createLocalRunResult("cancelled"); + }, + }); + + expect(result.runCommand).toHaveBeenCalledOnce(); + expect(result.sendInvokeResult).not.toHaveBeenCalled(); + expect(result.sendExecFinishedEvent).not.toHaveBeenCalled(); + }); + + it("does not publish a cancelled Mac exec-host completion", async () => { + const controller = new AbortController(); + const result = await runSystemInvoke({ + preferMacAppExecHost: true, + signal: controller.signal, + runViaMacAppExecHost: async () => { + controller.abort(); + return { ok: true, payload: createLocalRunResult("cancelled") }; + }, + }); + + expect(result.runViaMacAppExecHost).toHaveBeenCalledOnce(); + expect(result.sendInvokeResult).not.toHaveBeenCalled(); + expect(result.sendExecFinishedEvent).not.toHaveBeenCalled(); + }); + it("routes local, mac host, and canonical shell-wrapper requests", async () => { const localInvoke = await runSystemInvoke({ preferMacAppExecHost: false, diff --git a/src/node-host/invoke-system-run.ts b/src/node-host/invoke-system-run.ts index fcbb9e8daf41..0745dabebc6d 100644 --- a/src/node-host/invoke-system-run.ts +++ b/src/node-host/invoke-system-run.ts @@ -263,6 +263,7 @@ type HandleSystemRunInvokeOptions = { client: NodeHostClient; params: SystemRunParams; skillBins: SkillBinsProvider; + signal?: AbortSignal; execHostEnforced: boolean; execHostFallbackAllowed: boolean; resolveExecSecurity: (value?: string) => ExecSecurity; @@ -274,6 +275,7 @@ type HandleSystemRunInvokeOptions = { cwd: string | undefined, env: Record | undefined, timeoutMs: number | undefined, + signal?: AbortSignal, ) => Promise; runViaMacAppExecHost: (params: { approvals: ExecApprovalsResolved; @@ -979,6 +981,9 @@ async function executeSystemRunPhase( approvals: phase.approvals, request: execRequest, }); + if (opts.signal?.aborted) { + return; + } if (!response) { if (opts.execHostEnforced || !opts.execHostFallbackAllowed) { await sendSystemRunDenied(opts, phase.execution, { @@ -1067,7 +1072,15 @@ async function executeSystemRunPhase( return; } - const result = await opts.runCommand(execArgv, phase.cwd, phase.env, phase.timeoutMs); + if (opts.signal?.aborted) { + return; + } + const result = await (opts.signal + ? opts.runCommand(execArgv, phase.cwd, phase.env, phase.timeoutMs, opts.signal) + : opts.runCommand(execArgv, phase.cwd, phase.env, phase.timeoutMs)); + if (opts.signal?.aborted) { + return; + } applyOutputTruncation(result); await sendSystemRunCompleted( opts, @@ -1086,12 +1099,15 @@ async function executeSystemRunPhase( /** Executes a validated system.run request, emitting lifecycle events and approvals. */ export async function handleSystemRunInvoke(opts: HandleSystemRunInvokeOptions): Promise { + if (opts.signal?.aborted) { + return; + } const parsed = await parseSystemRunPhase(opts); - if (!parsed) { + if (!parsed || opts.signal?.aborted) { return; } const policyPhase = await evaluateSystemRunPolicyPhase(opts, parsed); - if (!policyPhase) { + if (!policyPhase || opts.signal?.aborted) { return; } await executeSystemRunPhase(opts, policyPhase); diff --git a/src/node-host/invoke.run-command.test.ts b/src/node-host/invoke.run-command.test.ts index 15b792d81360..d0bd20b86a22 100644 --- a/src/node-host/invoke.run-command.test.ts +++ b/src/node-host/invoke.run-command.test.ts @@ -76,6 +76,26 @@ describe("runCommand", () => { expect(Date.now() - startedAt).toBeLessThan(2_000); }); + it.runIf(process.platform !== "win32")("force-kills cancelled command trees", async () => { + const controller = new AbortController(); + const startedAt = Date.now(); + const cancelling = setTimeout(() => controller.abort(), 25); + try { + const result = await testing.runCommand( + [process.execPath, "-e", "process.on('SIGTERM', () => {}); setInterval(() => {}, 1000)"], + undefined, + undefined, + undefined, + controller.signal, + ); + + expect(result).toMatchObject({ timedOut: false, success: false, error: null }); + expect(Date.now() - startedAt).toBeLessThan(2_000); + } finally { + clearTimeout(cancelling); + } + }); + it("keeps the combined output prefix bounded", async () => { const result = await testing.runCommand( [process.execPath, "-e", "process.stdout.write('x'.repeat(200_001))"], diff --git a/src/node-host/invoke.test-support.ts b/src/node-host/invoke.test-support.ts index f34d0c018b00..0bc98f0c0a56 100644 --- a/src/node-host/invoke.test-support.ts +++ b/src/node-host/invoke.test-support.ts @@ -10,6 +10,7 @@ type NodeHostInvokeTestApi = { cwd: string | undefined, env: Record | undefined, timeoutMs: number | undefined, + signal?: AbortSignal, ): Promise; }; @@ -29,7 +30,7 @@ export const testing: NodeHostInvokeTestApi = { clarifyNodeExecCwdSpawnError(error, cwd) { return getTestApi().clarifyNodeExecCwdSpawnError(error, cwd); }, - runCommand(argv, cwd, env, timeoutMs) { - return getTestApi().runCommand(argv, cwd, env, timeoutMs); + runCommand(argv, cwd, env, timeoutMs, signal) { + return getTestApi().runCommand(argv, cwd, env, timeoutMs, signal); }, }; diff --git a/src/node-host/invoke.ts b/src/node-host/invoke.ts index ed8f14877093..61bb85220be7 100644 --- a/src/node-host/invoke.ts +++ b/src/node-host/invoke.ts @@ -335,6 +335,7 @@ async function runCommand( cwd: string | undefined, env: Record | undefined, timeoutMs: number | undefined, + signal?: AbortSignal, ): Promise { try { const result = await runCommandWithTimeout(argv, { @@ -345,6 +346,7 @@ async function runCommand( maxOutputBytes: OUTPUT_CAP, outputCapture: "head", input: Buffer.alloc(0), + signal, timeoutMs: timeoutMs && timeoutMs > 0 ? timeoutMs : undefined, }); const timedOut = result.termination === "timeout"; @@ -804,6 +806,7 @@ async function dispatchInvoke( client, params, skillBins, + signal: runtime.signal, execHostEnforced, execHostFallbackAllowed, resolveExecSecurity, diff --git a/src/node-host/runtime.test.ts b/src/node-host/runtime.test.ts index 23aaa8111398..301ab2501d68 100644 --- a/src/node-host/runtime.test.ts +++ b/src/node-host/runtime.test.ts @@ -72,22 +72,114 @@ async function startRuntime() { function holdInvoke() { let io: OpenClawPluginNodeHostCommandIo | undefined; + let signal: AbortSignal | undefined; let release: (() => void) | undefined; const held = new Promise((resolve) => { release = resolve; }); mocks.handleInvoke.mockImplementationOnce(async (...args: unknown[]) => { - io = (args[4] as { pluginCommandIo?: OpenClawPluginNodeHostCommandIo }).pluginCommandIo; + const runtime = args[4] as { + pluginCommandIo?: OpenClawPluginNodeHostCommandIo; + signal?: AbortSignal; + }; + io = runtime.pluginCommandIo; + signal = runtime.signal; await held; }); return { get io() { return io; }, + get signal() { + return signal; + }, release: () => release?.(), }; } +describe("node-host invocation cancellation", () => { + beforeEach(() => { + vi.clearAllMocks(); + }); + + it("cancels ordinary node invocations", async () => { + const held = holdInvoke(); + const runtime = await startRuntime(); + const invoking = runtime.invoke({ ...frame, command: "system.run" }); + await vi.waitFor(() => expect(held.signal).toBeDefined()); + + runtime.cancel(frame.id); + + expect(held.signal?.aborted).toBe(true); + expect(held.io).toBeUndefined(); + held.release(); + await invoking; + await runtime.close(); + }); + + it("cancels a superseded invocation without orphaning its replacement", async () => { + const first = holdInvoke(); + const second = holdInvoke(); + const runtime = await startRuntime(); + const firstInvoke = runtime.invoke({ ...frame, command: "system.run" }); + await vi.waitFor(() => expect(first.signal).toBeDefined()); + + const secondInvoke = runtime.invoke({ ...frame, command: "system.run" }); + await vi.waitFor(() => expect(second.signal).toBeDefined()); + + expect(first.signal?.aborted).toBe(true); + expect(second.signal?.aborted).toBe(false); + + first.release(); + await firstInvoke; + expect(second.signal?.aborted).toBe(false); + runtime.cancel(frame.id); + + expect(second.signal?.aborted).toBe(true); + second.release(); + await secondInvoke; + await runtime.close(); + }); + + it("cancels every ordinary invocation when the gateway disconnects", async () => { + const first = holdInvoke(); + const second = holdInvoke(); + const runtime = await startRuntime(); + const firstInvoke = runtime.invoke({ ...frame, command: "system.run" }); + const secondInvoke = runtime.invoke({ + ...frame, + id: "invoke-2", + command: "system.run", + }); + await vi.waitFor(() => { + expect(first.signal).toBeDefined(); + expect(second.signal).toBeDefined(); + }); + + runtime.cancelAll(); + + expect(first.signal?.aborted).toBe(true); + expect(second.signal?.aborted).toBe(true); + first.release(); + second.release(); + await Promise.all([firstInvoke, secondInvoke]); + await runtime.close(); + }); + + it("cancels ordinary invocations when the node runtime closes", async () => { + const held = holdInvoke(); + const runtime = await startRuntime(); + const invoking = runtime.invoke({ ...frame, command: "system.run" }); + await vi.waitFor(() => expect(held.signal).toBeDefined()); + + await runtime.close(); + + expect(held.signal?.aborted).toBe(true); + held.release(); + await invoking; + }); +}); + describe("node-host invoke input dispatch", () => { beforeEach(() => { vi.clearAllMocks(); diff --git a/src/node-host/runtime.ts b/src/node-host/runtime.ts index 815f14f25471..0ee9718f2753 100644 --- a/src/node-host/runtime.ts +++ b/src/node-host/runtime.ts @@ -72,6 +72,11 @@ type NodeInvokeInputTarget = { inputFailed: boolean; }; +type ActiveNodeInvoke = { + controller: AbortController; + input?: NodeInvokeInputTarget; +}; + const MAX_PENDING_INVOKE_INPUT_BYTES = 64 * 1024; function dispatchNodeInvokeInput( @@ -295,10 +300,7 @@ export async function prepareNodeHostRuntime(params?: { start({ client, onInventoryChanged, onManifestChanged }) { const mcpAbort = new AbortController(); const skillBins = new SkillBinsCache(client, pathEnv); - const activeInvokes = new Map< - string, - NodeInvokeInputTarget & { controller: AbortController } - >(); + const activeInvokes = new Map(); const pluginCommandContext: OpenClawPluginNodeHostCommandContext = { sendNodeEvent: async (event, payload) => await client.request("node.event", buildNodeEventParams(event, payload)), @@ -346,49 +348,48 @@ export async function prepareNodeHostRuntime(params?: { return { async invoke(frame) { const duplexCommand = duplexEnabled && isRegisteredNodeHostCommandDuplex(frame.command); - const controller = - (claudePath && frame.command === NODE_AGENT_CLI_CLAUDE_RUN_COMMAND) || duplexCommand - ? new AbortController() - : undefined; - const active: (NodeInvokeInputTarget & { controller: AbortController }) | undefined = - controller - ? { - controller, - nextInputSeq: 0, - pendingInput: new BoundedBuffer( - MAX_PENDING_INVOKE_INPUT_BYTES, - { - mode: "fail-closed", - onOverflow: () => - controller.abort( - new Error("terminal input exceeded the 64 KiB pre-spawn buffer"), - ), - }, - (payload) => Buffer.byteLength(payload, "utf8"), - ), - inputFailed: false, - } - : undefined; - if (active) { - activeInvokes.set(frame.id, active); - } + const controller = new AbortController(); + // Every command must remain cancellable after dispatch; only duplex + // commands own ordered input and its pre-spawn buffer. + const input: NodeInvokeInputTarget | undefined = duplexCommand + ? { + nextInputSeq: 0, + pendingInput: new BoundedBuffer( + MAX_PENDING_INVOKE_INPUT_BYTES, + { + mode: "fail-closed", + onOverflow: () => + controller.abort( + new Error("terminal input exceeded the 64 KiB pre-spawn buffer"), + ), + }, + (payload) => Buffer.byteLength(payload, "utf8"), + ), + inputFailed: false, + } + : undefined; + const active: ActiveNodeInvoke = { controller, ...(input ? { input } : {}) }; + // Redelivered IDs must not orphan the original command's process or + // let its cleanup unregister the replacement invocation. + activeInvokes.get(frame.id)?.controller.abort(); + activeInvokes.set(frame.id, active); const progress = duplexCommand ? createNodeInvokeProgressWriter({ client, frame, idleTimeoutMs: NODE_DUPLEX_INVOKE_IDLE_TIMEOUT_MS, - onError: () => controller?.abort(), + onError: () => controller.abort(), }) : undefined; progress?.startHeartbeats(); const pluginCommandIo: OpenClawPluginNodeHostCommandIo | undefined = - controller && active && progress + input && progress ? { signal: controller.signal, emitChunk: async (chunk) => await progress.write(chunk), onInput: (callback) => { if (activeInvokes.get(frame.id) === active) { - registerNodeInvokeInputHandler(active, callback); + registerNodeInvokeInputHandler(input, callback); } }, } @@ -396,7 +397,7 @@ export async function prepareNodeHostRuntime(params?: { try { await handleInvoke(frame, client, skillBins, manager, { ...(claudePath ? { claudePath } : {}), - ...(controller ? { signal: controller.signal } : {}), + signal: controller.signal, ...(pluginCommandIo ? { pluginCommandIo } : {}), installedAppsSharingEnabled, installedAppsPlatform: platform, @@ -405,14 +406,14 @@ export async function prepareNodeHostRuntime(params?: { } finally { progress?.stop(); await progress?.flush(); - if (active && activeInvokes.get(frame.id) === active) { + if (activeInvokes.get(frame.id) === active) { activeInvokes.delete(frame.id); } } }, handleInput(invokeId, seq, payloadJSON) { - const active = activeInvokes.get(invokeId); - if (!dispatchNodeInvokeInput(active, seq, payloadJSON)) { + const input = activeInvokes.get(invokeId)?.input; + if (!dispatchNodeInvokeInput(input, seq, payloadJSON)) { logDebug(`node-host: dropped inactive or duplicate input for invoke ${invokeId}`); } }, diff --git a/src/node-host/worker-support.ts b/src/node-host/worker-support.ts index 4cc65889b52e..368f1ae750c6 100644 --- a/src/node-host/worker-support.ts +++ b/src/node-host/worker-support.ts @@ -1,3 +1,4 @@ +import { resolveTimerTimeoutMs } from "@openclaw/normalization-core/number-coercion"; import { asNullableRecord as asRecord } from "@openclaw/normalization-core/record-coerce"; import type { GatewayClientRequestOptions } from "../gateway/client.js"; import type { NodeHostClient } from "./client.js"; @@ -89,7 +90,7 @@ export class NodeHostWorkerBridgeClient implements NodeHostClient { } const id = `gateway-${this.nextRequestId++}`; - const timeoutMs = Math.max(1, opts?.timeoutMs ?? 15_000); + const timeoutMs = resolveTimerTimeoutMs(opts?.timeoutMs, 15_000); const response = new Promise((resolve, reject) => { const timer = setTimeout(() => { this.pending.delete(id); diff --git a/src/node-host/worker.test.ts b/src/node-host/worker.test.ts index 320d427b1eb2..c413139fe9c7 100644 --- a/src/node-host/worker.test.ts +++ b/src/node-host/worker.test.ts @@ -1,4 +1,5 @@ -import { describe, expect, it } from "vitest"; +import { MAX_TIMER_TIMEOUT_MS } from "@openclaw/normalization-core/number-coercion"; +import { afterEach, describe, expect, it, vi } from "vitest"; import { NodeHostWorkerBridgeClient, parseNodeHostWorkerInput, @@ -30,6 +31,10 @@ describe("parseNodeHostWorkerInput", () => { }); describe("NodeHostWorkerBridgeClient", () => { + afterEach(() => { + vi.useRealTimers(); + }); + it("forwards invoke results and events without creating gateway request waits", async () => { const messages: unknown[] = []; const client = new NodeHostWorkerBridgeClient((message) => messages.push(message)); @@ -119,6 +124,44 @@ describe("NodeHostWorkerBridgeClient", () => { await expect(response).rejects.toThrow("node-host worker stopped"); }); + + it.each([ + { requested: Number.MAX_SAFE_INTEGER, expected: MAX_TIMER_TIMEOUT_MS }, + { requested: Number.POSITIVE_INFINITY, expected: 15_000 }, + { requested: Number.NaN, expected: 15_000 }, + { requested: 0, expected: 1 }, + { requested: -5, expected: 1 }, + { requested: 7.9, expected: 7 }, + ])("normalizes a gateway request timeout of $requested", async ({ requested, expected }) => { + vi.useFakeTimers(); + const messages: Array> = []; + const client = new NodeHostWorkerBridgeClient((message) => { + messages.push(message as Record); + }); + + const response = client.request("skills.bins", {}, { timeoutMs: requested }); + + expect(messages).toEqual([ + { + type: "gateway-request", + id: "gateway-1", + method: "skills.bins", + params: {}, + timeoutMs: expected, + }, + ]); + expect(vi.getTimerCount()).toBe(1); + expect( + client.handleResponse({ + type: "gateway-response", + id: "gateway-1", + ok: true, + result: { bins: [] }, + }), + ).toBe(true); + await expect(response).resolves.toEqual({ bins: [] }); + expect(vi.getTimerCount()).toBe(0); + }); }); describe("stopNodeHostWorkerFromSignal", () => {