fix(searxng): stop web search requests when runs are aborted (#104216)

* fix(searxng): cancel pending searches with parent runs

* fix(searxng): preserve parent cancellation

* refactor(web-search): isolate provider execution

* fix(web-search): use stable provider type

---------

Co-authored-by: Peter Steinberger <steipete@gmail.com>

extensions/searxng/src/searxng-client.test.ts

@@ -4,14 +4,19 @@ import type { LookupFn } from "openclaw/plugin-sdk/ssrf-runtime";
 import { beforeEach, describe, expect, it, vi } from "vitest";
 
 const endpointMockState = vi.hoisted(() => ({
-  calls: [] as Array<{ url: string; timeoutSeconds: number; init: RequestInit }>,
+  calls: [] as Array<{
+    url: string;
+    timeoutSeconds: number;
+    init: RequestInit;
+    signal?: AbortSignal;
+  }>,
   responses: [] as Response[],
 }));
 
 vi.mock("openclaw/plugin-sdk/provider-web-search", async (importOriginal) => {
   const actual = await importOriginal<typeof import("openclaw/plugin-sdk/provider-web-search")>();
   const runEndpoint = async (
-    params: { url: string; timeoutSeconds: number; init: RequestInit },
+    params: { url: string; timeoutSeconds: number; init: RequestInit; signal?: AbortSignal },
     run: (response: Response) => Promise<unknown>,
   ) => {
     endpointMockState.calls.push(params);
@@ -153,6 +158,23 @@ describe("searxng client", () => {
     });
   });
 
+  it("forwards the abort signal to the guarded endpoint", async () => {
+    endpointMockState.responses.push(
+      new Response(JSON.stringify({ results: [] }), { status: 200 }),
+    );
+    const controller = new AbortController();
+
+    await runSearxngSearch({
+      baseUrl: "http://127.0.0.1:8888",
+      query: "openclaw",
+      categories: "general",
+      signal: controller.signal,
+    });
+
+    expect(endpointMockState.calls).toHaveLength(1);
+    expect(endpointMockState.calls[0]?.signal).toBe(controller.signal);
+  });
+
   it("rejects partial response bodies without blaming the size limit", async () => {
     const chunk = new TextEncoder().encode("partial");
     let sentChunk = false;

extensions/searxng/src/searxng-client.transport.test.ts

@@ -0,0 +1,99 @@
+import { createServer, type Server } from "node:http";
+import { afterEach, describe, expect, it } from "vitest";
+import { runSearxngSearch, testing } from "./searxng-client.js";
+
+const servers = new Set<Server>();
+
+async function listen(server: Server): Promise<string> {
+  servers.add(server);
+  await new Promise<void>((resolve, reject) => {
+    server.once("error", reject);
+    server.listen(0, "127.0.0.1", () => {
+      server.off("error", reject);
+      resolve();
+    });
+  });
+  const address = server.address();
+  if (!address || typeof address === "string") {
+    throw new Error("Expected a TCP listener address");
+  }
+  return `http://127.0.0.1:${address.port}`;
+}
+
+async function closeServer(server: Server): Promise<void> {
+  server.closeAllConnections();
+  await new Promise<void>((resolve) => {
+    server.close(() => resolve());
+  });
+}
+
+afterEach(async () => {
+  testing.SEARXNG_SEARCH_CACHE.clear();
+  await Promise.all([...servers].map(closeServer));
+  servers.clear();
+});
+
+describe("searxng real transport", () => {
+  it("reads JSON results from a loopback endpoint", async () => {
+    const server = createServer((_request, response) => {
+      response.writeHead(200, { "Content-Type": "application/json" });
+      response.end(
+        JSON.stringify({
+          results: [
+            {
+              title: "OpenClaw",
+              url: "https://docs.openclaw.ai/",
+              content: "OpenClaw documentation",
+            },
+          ],
+        }),
+      );
+    });
+    const baseUrl = await listen(server);
+
+    await expect(
+      runSearxngSearch({
+        baseUrl,
+        query: "openclaw",
+        categories: "general",
+      }),
+    ).resolves.toMatchObject({
+      provider: "searxng",
+      count: 1,
+      results: [{ url: "https://docs.openclaw.ai/" }],
+    });
+  });
+
+  it("aborts a stalled response body and closes the request", async () => {
+    let resolveRequestStarted: (() => void) | undefined;
+    const requestStarted = new Promise<void>((resolve) => {
+      resolveRequestStarted = resolve;
+    });
+    let resolveClientClosed: (() => void) | undefined;
+    const clientClosed = new Promise<void>((resolve) => {
+      resolveClientClosed = resolve;
+    });
+    const server = createServer((request, response) => {
+      request.socket.once("close", () => resolveClientClosed?.());
+      response.writeHead(200, { "Content-Type": "application/json" });
+      response.write('{"results":[');
+      response.flushHeaders();
+      resolveRequestStarted?.();
+    });
+    const baseUrl = await listen(server);
+    const controller = new AbortController();
+    const pending = runSearxngSearch({
+      baseUrl,
+      query: "stalled response",
+      categories: "general",
+      timeoutSeconds: 30,
+      signal: controller.signal,
+    });
+
+    await requestStarted;
+    controller.abort();
+
+    await expect(pending).rejects.toMatchObject({ name: "AbortError" });
+    await expect(clientClosed).resolves.toBeUndefined();
+  });
+});

extensions/searxng/src/searxng-client.ts

@@ -189,6 +189,7 @@ async function fetchSearxngResults(params: {
   timeoutSeconds: number;
   count: number;
   endpointMode: SearxngEndpointMode;
+  signal?: AbortSignal;
 }): Promise<SearxngResult[]> {
   const url = buildSearxngSearchUrl({
     baseUrl: params.baseUrl,
@@ -205,6 +206,7 @@ async function fetchSearxngResults(params: {
     {
       url,
       timeoutSeconds: params.timeoutSeconds,
+      signal: params.signal,
       init: {
         method: "GET",
         headers: {
@@ -238,7 +240,9 @@ export async function runSearxngSearch(params: {
   baseUrl?: string;
   timeoutSeconds?: number;
   cacheTtlMinutes?: number;
+  signal?: AbortSignal;
 }): Promise<Record<string, unknown>> {
+  params.signal?.throwIfAborted();
   const count = resolveSearchCount(params.count, DEFAULT_SEARCH_COUNT);
   const categories = params.categories ?? resolveSearxngCategories(params.config);
   const language = params.language ?? resolveSearxngLanguage(params.config);
@@ -252,6 +256,7 @@ export async function runSearxngSearch(params: {
     );
   }
   const endpointMode = await validateSearxngBaseUrl(baseUrl);
+  params.signal?.throwIfAborted();
 
   const cacheKey = normalizeCacheKey(
     JSON.stringify({
@@ -277,7 +282,9 @@ export async function runSearxngSearch(params: {
     timeoutSeconds,
     count,
     endpointMode,
+    signal: params.signal,
   });
+  params.signal?.throwIfAborted();
   if (results.length === 0 && shouldRetryEmptyCategorySearchWithGeneral(categories)) {
     results = await fetchSearxngResults({
       baseUrl,
@@ -287,7 +294,9 @@ export async function runSearxngSearch(params: {
       timeoutSeconds,
       count,
       endpointMode,
+      signal: params.signal,
     });
+    params.signal?.throwIfAborted();
   }
 
   const payload = {

extensions/searxng/src/searxng-search-provider.test.ts

@@ -93,6 +93,28 @@ describe("searxng web search provider", () => {
     });
   });
 
+  it("forwards the execution abort signal to the SearXNG client", async () => {
+    const provider = createSearxngWebSearchProvider();
+    const tool = provider.createTool({
+      config: { test: true },
+    } as never);
+    if (!tool) {
+      throw new Error("Expected tool definition");
+    }
+    const controller = new AbortController();
+
+    await tool.execute({ query: "openclaw docs" }, { signal: controller.signal });
+
+    expect(runSearxngSearch).toHaveBeenCalledWith({
+      config: { test: true },
+      query: "openclaw docs",
+      count: undefined,
+      categories: undefined,
+      language: undefined,
+      signal: controller.signal,
+    });
+  });
+
   it("rejects fractional and out-of-range counts before searching", async () => {
     const provider = createSearxngWebSearchProvider();
     const tool = provider.createTool({

extensions/searxng/src/searxng-search-provider.ts

@@ -59,7 +59,7 @@ export function createSearxngWebSearchProvider(): WebSearchProviderPlugin {
       description:
         "Search the web using a self-hosted SearXNG instance. Returns titles, URLs, and snippets.",
       parameters: SearxngSearchSchema,
-      execute: async (args) => {
+      execute: async (args, context) => {
         const { runSearxngSearch } = await loadSearxngClientModule();
         return await runSearxngSearch({
           config: ctx.config,
@@ -70,6 +70,7 @@ export function createSearxngWebSearchProvider(): WebSearchProviderPlugin {
           }),
           categories: readStringParam(args, "categories"),
           language: readStringParam(args, "language"),
+          signal: context?.signal,
         });
       },
     }),

src/web-search/runtime-execution.ts

@@ -0,0 +1,75 @@
+// Web search provider execution owns cancellation precedence and automatic fallback.
+import type { OpenClawConfig } from "../config/types.openclaw.js";
+import type { PluginWebSearchProviderEntry } from "../plugins/web-provider-types.js";
+import type { RuntimeWebSearchMetadata } from "../secrets/runtime-web-tools.types.js";
+import type { RunWebSearchResult } from "./runtime-types.js";
+
+type ExecuteWebSearchCandidatesParams = {
+  candidates: readonly PluginWebSearchProviderEntry[];
+  config?: OpenClawConfig;
+  searchConfig?: Record<string, unknown>;
+  runtimeMetadata?: RuntimeWebSearchMetadata;
+  agentDir?: string;
+  args: Record<string, unknown>;
+  signal?: AbortSignal;
+  allowFallback: boolean;
+};
+
+function isStructuredAvailabilityError(result: unknown): result is { error: string } {
+  if (!result || typeof result !== "object" || !("error" in result)) {
+    return false;
+  }
+  const error = (result as { error?: unknown }).error;
+  return typeof error === "string" && /^missing_[a-z0-9_]*api_key$/i.test(error);
+}
+
+export async function executeWebSearchCandidates(
+  params: ExecuteWebSearchCandidatesParams,
+): Promise<RunWebSearchResult> {
+  let lastError: unknown;
+  let sawUnavailableProvider = false;
+
+  for (const candidate of params.candidates) {
+    params.signal?.throwIfAborted();
+    try {
+      const definition = candidate.createTool({
+        config: params.config,
+        agentDir: params.agentDir,
+        searchConfig: params.searchConfig,
+        runtimeMetadata: params.runtimeMetadata,
+      });
+      if (!definition) {
+        if (!params.allowFallback) {
+          throw new Error(`web_search provider "${candidate.id}" is not available.`);
+        }
+        sawUnavailableProvider = true;
+        continue;
+      }
+      const executed = await definition.execute(params.args, { signal: params.signal });
+      // Cancellation wins races with provider completion or cleanup failures. Otherwise an
+      // ignored signal could return stale work or trigger another provider fallback.
+      params.signal?.throwIfAborted();
+      if (params.allowFallback && isStructuredAvailabilityError(executed)) {
+        // Some providers report missing credentials as structured tool output.
+        // Treat that like unavailable only during auto-detected fallback.
+        lastError = new Error(`web_search provider "${candidate.id}" returned ${executed.error}`);
+        continue;
+      }
+      return {
+        provider: candidate.id,
+        result: executed,
+      };
+    } catch (error) {
+      params.signal?.throwIfAborted();
+      lastError = error;
+      if (!params.allowFallback) {
+        throw error;
+      }
+    }
+  }
+
+  if (sawUnavailableProvider && lastError === undefined) {
+    throw new Error("web_search is enabled but no provider is currently available.");
+  }
+  throw lastError instanceof Error ? lastError : new Error(String(lastError));
+}

src/web-search/runtime.test.ts

@@ -267,6 +267,84 @@ describe("web search runtime", () => {
     );
   });
 
+  it("does not fall back to another provider after parent cancellation", async () => {
+    const controller = new AbortController();
+    const abortReason = new Error("parent run cancelled");
+    abortReason.name = "AbortError";
+    const firstExecute = vi.fn(
+      async (_args: Record<string, unknown>, context?: { signal?: AbortSignal }) => {
+        expect(context?.signal).toBe(controller.signal);
+        controller.abort(abortReason);
+        throw new Error("provider cleanup failed after cancellation");
+      },
+    );
+    const fallbackExecute = vi.fn(async () => ({ ok: true }));
+    resolveRuntimeWebSearchProvidersMock.mockReturnValue([
+      createCustomSearchProvider({
+        credentialPath: "",
+        createTool: () => ({
+          description: "first",
+          parameters: {},
+          execute: firstExecute,
+        }),
+      }),
+      createCustomSearchProvider({
+        pluginId: "fallback-search",
+        id: "fallback",
+        credentialPath: "",
+        autoDetectOrder: 2,
+        getConfiguredCredentialValue: () => "configured",
+        createTool: () => ({
+          description: "fallback",
+          parameters: {},
+          execute: fallbackExecute,
+        }),
+      }),
+    ]);
+
+    await expect(
+      runWebSearch({
+        config: createCustomSearchConfig("custom-config-key"),
+        args: { query: "abort fallback" },
+        signal: controller.signal,
+      }),
+    ).rejects.toBe(abortReason);
+    expect(firstExecute).toHaveBeenCalledOnce();
+    expect(fallbackExecute).not.toHaveBeenCalled();
+  });
+
+  it("rejects a provider result that completes after parent cancellation", async () => {
+    const controller = new AbortController();
+    const abortReason = new Error("parent run cancelled");
+    abortReason.name = "AbortError";
+    const execute = vi.fn(async () => {
+      controller.abort(abortReason);
+      return { ok: true };
+    });
+    resolveRuntimeWebSearchProvidersMock.mockReturnValue([
+      createCustomSearchProvider({
+        credentialPath: "tools.web.search.custom.apiKey",
+        requiresCredential: false,
+        createTool: () => ({
+          description: "custom",
+          parameters: {},
+          execute,
+        }),
+      }),
+    ]);
+
+    await expect(
+      runWebSearch({
+        config: {
+          tools: { web: { search: { provider: "custom" } } },
+        },
+        args: { query: "late abort" },
+        signal: controller.signal,
+      }),
+    ).rejects.toBe(abortReason);
+    expect(execute).toHaveBeenCalledOnce();
+  });
+
   it("auto-detects a provider from canonical plugin-owned credentials", async () => {
     const provider = createCustomSearchProvider();
     resolveRuntimeWebSearchProvidersMock.mockReturnValue([provider]);

src/web-search/runtime.ts

@@ -28,6 +28,7 @@ import {
 import { sortWebSearchProvidersForAutoDetect } from "../plugins/web-search-providers.shared.js";
 import { getActiveRuntimeWebToolsMetadata } from "../secrets/runtime-web-tools-state.js";
 import type { RuntimeWebSearchMetadata } from "../secrets/runtime-web-tools.types.js";
+import { executeWebSearchCandidates } from "./runtime-execution.js";
 import type {
   ResolveWebSearchDefinitionParams,
   RunWebSearchParams,
@@ -437,14 +438,6 @@ function hasExplicitWebSearchSelection(params: {
   return false;
 }
 
-function isStructuredAvailabilityError(result: unknown): result is { error: string } {
-  if (!result || typeof result !== "object" || !("error" in result)) {
-    return false;
-  }
-  const error = (result as { error?: unknown }).error;
-  return typeof error === "string" && /^missing_[a-z0-9_]*api_key$/i.test(error);
-}
-
 /** Executes web_search with fallback when selection was not explicit. */
 export async function runWebSearch(params: RunWebSearchParams): Promise<RunWebSearchResult> {
   const config = resolveWebSearchRuntimeConfig({
@@ -468,45 +461,14 @@ export async function runWebSearch(params: RunWebSearchParams): Promise<RunWebSe
     providerId: params.providerId,
     providers: candidates,
   });
-  let lastError: unknown;
-  let sawUnavailableProvider = false;
-
-  for (const candidate of candidates) {
-    try {
-      const definition = candidate.createTool({
-        config,
-        agentDir: params.agentDir,
-        searchConfig: search as Record<string, unknown> | undefined,
-        runtimeMetadata: runtimeWebSearch,
-      });
-      if (!definition) {
-        if (!allowFallback) {
-          throw new Error(`web_search provider "${candidate.id}" is not available.`);
-        }
-        sawUnavailableProvider = true;
-        continue;
-      }
-      const executed = await definition.execute(params.args, { signal: params.signal });
-      if (allowFallback && isStructuredAvailabilityError(executed)) {
-        // Some providers report missing credentials as structured tool output.
-        // Treat that like unavailable only during auto-detected fallback.
-        lastError = new Error(`web_search provider "${candidate.id}" returned ${executed.error}`);
-        continue;
-      }
-      return {
-        provider: candidate.id,
-        result: executed,
-      };
-    } catch (error) {
-      lastError = error;
-      if (!allowFallback) {
-        throw error;
-      }
-    }
-  }
-
-  if (sawUnavailableProvider && lastError === undefined) {
-    throw new Error("web_search is enabled but no provider is currently available.");
-  }
-  throw lastError instanceof Error ? lastError : new Error(String(lastError));
+  return await executeWebSearchCandidates({
+    candidates,
+    config,
+    searchConfig: search as Record<string, unknown> | undefined,
+    runtimeMetadata: runtimeWebSearch,
+    agentDir: params.agentDir,
+    args: params.args,
+    signal: params.signal,
+    allowFallback,
+  });
 }