fix(node): cancel remote commands and stabilize local inference (#114935)

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<void>((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();

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<string | undefined>,
+  signal?: AbortSignal,
+): Promise<string | undefined> {
+  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;

extensions/ollama/src/node-inference.test.ts

@@ -27,6 +27,7 @@ async function withOllamaServer<T>(
     chatRequests: Record<string, unknown>[],
     showRequests: string[],
   ) => Promise<T>,
+  options?: { models: Array<Record<string, unknown>> },
 ): Promise<T> {
   const chatRequests: Record<string, unknown>[] = [];
   const showRequests: string[] = [];
@@ -35,7 +36,7 @@ async function withOllamaServer<T>(
     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<T>(
         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(

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 = {

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);

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<string, Promise<OllamaModelShowInfo>>();
 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<OllamaModelWithContext[]> {
+  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",

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,

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<string, string> | undefined,
     timeoutMs: number | undefined,
+    signal?: AbortSignal,
   ) => Promise<RunResult>;
   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<void> {
+  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);

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))"],

src/node-host/invoke.test-support.ts

@@ -10,6 +10,7 @@ type NodeHostInvokeTestApi = {
     cwd: string | undefined,
     env: Record<string, string> | undefined,
     timeoutMs: number | undefined,
+    signal?: AbortSignal,
   ): Promise<RunResult>;
 };
 
@@ -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);
   },
 };

src/node-host/invoke.ts

@@ -335,6 +335,7 @@ async function runCommand(
   cwd: string | undefined,
   env: Record<string, string> | undefined,
   timeoutMs: number | undefined,
+  signal?: AbortSignal,
 ): Promise<RunResult> {
   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,

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<void>((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();

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<string, ActiveNodeInvoke>();
       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<string>(
-                    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<string>(
+                  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}`);
           }
         },

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<unknown>((resolve, reject) => {
       const timer = setTimeout(() => {
         this.pending.delete(id);

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<Record<string, unknown>> = [];
+    const client = new NodeHostWorkerBridgeClient((message) => {
+      messages.push(message as Record<string, unknown>);
+    });
+
+    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", () => {