dilim/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-08-13 17:07:40 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(search): propagate cancellation through provider-owned requests (#118895)

Browse Source
This commit is contained in:
Peter Steinberger
2026-08-03 12:46:17 -07:00
committed by GitHub
parent f1a87bf364
commit 084833b324
22 changed files with 470 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
extensions/brave/src/brave-web-search-provider.runtime.ts
View File

@@ -206,6 +206,7 @@ async function runBraveJsonRequest<T>(
apiKey: string;
timeoutSeconds: number;
diagnostics?: BraveHttpDiagnostics;
signal?: AbortSignal;
configureUrl: (url: URL) => void;
},
errorLabel: string,
@@ -228,6 +229,7 @@ async function runBraveJsonRequest<T>(
{
url: url.toString(),
timeoutSeconds: params.timeoutSeconds,
signal: params.signal,
init: {
method: "GET",
headers: {
@@ -256,6 +258,7 @@ async function runBraveLlmContextSearch(params: {
apiKey: string;
timeoutSeconds: number;
diagnostics?: BraveHttpDiagnostics;
signal?: AbortSignal;
country?: string;
search_lang?: string;
freshness?: string;
@@ -279,6 +282,7 @@ async function runBraveLlmContextSearch(params: {
apiKey: params.apiKey,
timeoutSeconds: params.timeoutSeconds,
diagnostics: params.diagnostics,
signal: params.signal,
configureUrl: (url) => {
setBraveSearchUrlParams(url, params);
},
@@ -296,6 +300,7 @@ async function runBraveWebSearch(params: {
apiKey: string;
timeoutSeconds: number;
diagnostics?: BraveHttpDiagnostics;
signal?: AbortSignal;
country?: string;
search_lang?: string;
ui_lang?: string;
@@ -312,6 +317,7 @@ async function runBraveWebSearch(params: {
apiKey: params.apiKey,
timeoutSeconds: params.timeoutSeconds,
diagnostics: params.diagnostics,
signal: params.signal,
configureUrl: (url) => {
setBraveSearchUrlParams(url, {
...params,
@@ -346,6 +352,7 @@ export async function executeBraveSearch(
searchConfig?: SearchConfigRecord,
options?: {
diagnosticsEnabled?: boolean;
signal?: AbortSignal;
},
): Promise<Record<string, unknown>> {
const apiKey = resolveBraveApiKey(searchConfig);
@@ -485,12 +492,14 @@ export async function executeBraveSearch(
apiKey,
timeoutSeconds,
diagnostics,
signal: options?.signal,
country: country ?? undefined,
search_lang: normalizedLanguage.search_lang,
freshness,
dateAfter,
dateBefore,
});
options?.signal?.throwIfAborted();
const payload = {
query,
provider: "brave",
@@ -530,6 +539,7 @@ export async function executeBraveSearch(
apiKey,
timeoutSeconds,
diagnostics,
signal: options?.signal,
country: country ?? undefined,
search_lang: normalizedLanguage.search_lang,
ui_lang: normalizedLanguage.ui_lang,
@@ -537,6 +547,7 @@ export async function executeBraveSearch(
dateAfter,
dateBefore,
});
options?.signal?.throwIfAborted();
const payload = {
query,
provider: "brave",

46
extensions/brave/src/brave-web-search-provider.test.ts
View File

@@ -182,6 +182,52 @@ describe("brave web search provider", () => {
});
});
it.each(["web", "llm-context"] as const)(
"does not start an already canceled %s search",
async (mode) => {
const fetchMock = vi.fn(async () => emptyWebSearchResponse());
global.fetch = fetchMock as typeof global.fetch;
const tool = createBraveTool({ webSearch: { apiKey: "brave-test-key", mode } });
const controller = new AbortController();
controller.abort(new Error("Brave caller canceled"));
await expect(
tool.execute({ query: `brave pre-canceled ${mode}` }, { signal: controller.signal }),
).rejects.toThrow("Brave caller canceled");
expect(fetchMock).not.toHaveBeenCalled();
},
);
it.each(["web", "llm-context"] as const)(
"aborts an in-flight %s request with the caller's reason",
async (mode) => {
const fetchMock = vi.fn(
async (_url: string, init?: RequestInit) =>
await new Promise<Response>((_resolve, reject) => {
const signal = init?.signal;
if (!signal) {
reject(new Error("Brave request lost caller cancellation"));
return;
}
signal.addEventListener("abort", () => reject(signal.reason as Error), { once: true });
}),
);
global.fetch = fetchMock as typeof global.fetch;
const tool = createBraveTool({ webSearch: { apiKey: "brave-test-key", mode } });
const controller = new AbortController();
const result = tool.execute(
{ query: `brave in-flight cancellation ${mode}` },
{ signal: controller.signal },
);
await vi.waitFor(() => expect(fetchMock).toHaveBeenCalledOnce());
controller.abort(new Error("Brave request canceled in flight"));
await expect(result).rejects.toThrow("Brave request canceled in flight");
expect(fetchMock.mock.calls[0]?.[1]?.signal?.aborted).toBe(true);
},
);
it("normalizes brave language parameters and swaps reversed ui/search inputs", () => {
expect(
testing.normalizeBraveLanguageParams({

8
extensions/brave/src/brave-web-search-provider.ts
View File

@@ -82,9 +82,13 @@ function createBraveToolDefinition(
? "Search the web using Brave Search LLM Context API. Returns pre-extracted page content (text chunks, tables, code blocks) optimized for LLM grounding."
: "Search the web using Brave Search API. Supports region-specific and localized search via country and language parameters. Returns titles, URLs, and snippets for fast research.",
parameters: BraveSearchSchema,
execute: async (args) => {
execute: async (args, context) => {
context?.signal?.throwIfAborted();
const { executeBraveSearch } = await loadBraveWebSearchRuntime();
return await executeBraveSearch(args, searchConfig, { diagnosticsEnabled });
return await executeBraveSearch(args, searchConfig, {
diagnosticsEnabled,
signal: context?.signal,
});
},
};
}

3
extensions/duckduckgo/src/ddg-client.ts
View File

@@ -126,6 +126,7 @@ export async function runDuckDuckGoSearch(params: {
safeSearch?: DdgSafeSearch;
timeoutSeconds?: number;
cacheTtlMinutes?: number;
signal?: AbortSignal;
}): Promise<Record<string, unknown>> {
const count = resolveSearchCount(params.count, DEFAULT_SEARCH_COUNT);
const region = params.region ?? resolveDdgRegion(params.config);
@@ -163,6 +164,7 @@ export async function runDuckDuckGoSearch(params: {
{
url: url.toString(),
timeoutSeconds,
signal: params.signal,
init: {
method: "GET",
headers: {
@@ -187,6 +189,7 @@ export async function runDuckDuckGoSearch(params: {
},
);
params.signal?.throwIfAborted();
const payload = {
query: params.query,
provider: "duckduckgo",

63
extensions/duckduckgo/src/ddg-search-provider.test.ts
View File

@@ -15,6 +15,7 @@ vi.mock("./ddg-client.js", () => ({
describe("duckduckgo web search provider", () => {
let createDuckDuckGoWebSearchProvider: typeof import("./ddg-search-provider.js").createDuckDuckGoWebSearchProvider;
let ddgClientTesting: typeof import("./ddg-client.js").testing;
let runActualDuckDuckGoSearch: typeof import("./ddg-client.js").runDuckDuckGoSearch;
afterAll(() => {
vi.doUnmock("./ddg-client.js");
@@ -23,7 +24,7 @@ describe("duckduckgo web search provider", () => {
beforeAll(async () => {
({ createDuckDuckGoWebSearchProvider } = await import("./ddg-search-provider.js"));
({ testing: ddgClientTesting } =
({ testing: ddgClientTesting, runDuckDuckGoSearch: runActualDuckDuckGoSearch } =
await vi.importActual<typeof import("./ddg-client.js")>("./ddg-client.js"));
await import("../index.js");
});
@@ -105,6 +106,66 @@ describe("duckduckgo web search provider", () => {
expect(runDuckDuckGoSearch).not.toHaveBeenCalled();
});
it("forwards caller cancellation without starting an already canceled search", async () => {
const tool = createDuckDuckGoWebSearchProvider().createTool({ config: {} });
if (!tool) {
throw new Error("Expected tool definition");
}
const active = new AbortController();
await tool.execute({ query: "duckduckgo cancellation forwarding" }, { signal: active.signal });
expect(runDuckDuckGoSearch).toHaveBeenCalledWith(
expect.objectContaining({ signal: active.signal }),
);
runDuckDuckGoSearch.mockClear();
const canceled = new AbortController();
canceled.abort(new Error("DuckDuckGo caller canceled"));
await expect(
tool.execute({ query: "duckduckgo pre-canceled" }, { signal: canceled.signal }),
).rejects.toThrow("DuckDuckGo caller canceled");
expect(runDuckDuckGoSearch).not.toHaveBeenCalled();
});
it("aborts an in-flight DuckDuckGo request without caching its result", async () => {
const fetchMock = vi.spyOn(globalThis, "fetch").mockImplementation(
async (_url, init) =>
await new Promise<Response>((_resolve, reject) => {
if (!init?.signal) {
reject(new Error("DuckDuckGo request lost caller cancellation"));
return;
}
init.signal.addEventListener("abort", () => reject(init.signal?.reason as Error), {
once: true,
});
}),
);
const controller = new AbortController();
const result = runActualDuckDuckGoSearch({
query: "duckduckgo in-flight cancellation",
signal: controller.signal,
});
try {
await vi.waitFor(() => expect(fetchMock).toHaveBeenCalledOnce());
controller.abort(new Error("DuckDuckGo request canceled in flight"));
await expect(result).rejects.toThrow("DuckDuckGo request canceled in flight");
expect(fetchMock.mock.calls[0]?.[1]?.signal?.aborted).toBe(true);
fetchMock.mockResolvedValueOnce(
new Response('<a class="result__a" href="https://example.com">Example</a>', {
headers: { "content-type": "text/html" },
}),
);
await runActualDuckDuckGoSearch({ query: "duckduckgo in-flight cancellation" });
expect(fetchMock).toHaveBeenCalledTimes(2);
} finally {
fetchMock.mockRestore();
}
});
it("bounds successful DuckDuckGo HTML bodies without using response.text()", async () => {
const streamed = createStreamingResponse({
chunkCount: 32,

4
extensions/duckduckgo/src/ddg-search-provider.ts
View File

@@ -35,7 +35,8 @@ export function createDuckDuckGoWebSearchProvider(): WebSearchProviderPlugin {
description:
"Search the web using DuckDuckGo. Returns titles, URLs, and snippets with no API key required.",
parameters: DuckDuckGoSearchSchema,
execute: async (args) => {
execute: async (args, context) => {
context?.signal?.throwIfAborted();
const { runDuckDuckGoSearch } = await loadDuckDuckGoClientModule();
return await runDuckDuckGoSearch({
config: ctx.config,
@@ -50,6 +51,7 @@ export function createDuckDuckGoWebSearchProvider(): WebSearchProviderPlugin {
| "moderate"
| "off"
| undefined,
...(context?.signal ? { signal: context.signal } : {}),
});
},
}),

5
extensions/exa/src/exa-web-search-provider.runtime.ts
View File

@@ -392,6 +392,7 @@ async function runExaSearch(params: {
type: ExaSearchType;
contents?: ExaContentsArgs;
timeoutSeconds: number;
signal?: AbortSignal;
}): Promise<ExaSearchResult[]> {
const body: Record<string, unknown> = {
query: params.query,
@@ -413,6 +414,7 @@ async function runExaSearch(params: {
{
url: params.endpoint,
timeoutSeconds: params.timeoutSeconds,
signal: params.signal,
init: {
method: "POST",
headers: {
@@ -471,6 +473,7 @@ function buildExaCacheKey(params: {
export async function executeExaWebSearchProviderTool(
ctx: { config?: Record<string, unknown>; searchConfig?: SearchConfigRecord },
args: Record<string, unknown>,
signal?: AbortSignal,
): Promise<Record<string, unknown>> {
const searchConfig = mergeScopedSearchConfig(
ctx.searchConfig,
@@ -570,8 +573,10 @@ export async function executeExaWebSearchProviderTool(
type,
contents,
timeoutSeconds: resolveSearchTimeoutSeconds(searchConfig),
signal,
});
signal?.throwIfAborted();
const payload = {
query,
provider: "exa",

67
extensions/exa/src/exa-web-search-provider.test.ts
View File

@@ -54,6 +54,73 @@ function streamingJsonResponse(params: { chunkCount: number; chunkSize: number }
}
describe("exa web search provider", () => {
it("does not send or cache an already canceled search", async () => {
const fetchMock = vi.spyOn(globalThis, "fetch").mockResolvedValue(
new Response(JSON.stringify({ results: [] }), {
status: 200,
headers: { "content-type": "application/json" },
}),
);
const tool = createExaWebSearchProvider().createTool({
config: {
plugins: { entries: { exa: { config: { webSearch: { apiKey: "exa-test-key" } } } } },
},
searchConfig: {},
});
if (!tool) {
throw new Error("Expected tool definition");
}
const controller = new AbortController();
controller.abort(new Error("Exa caller canceled"));
try {
await expect(
tool.execute({ query: "exa pre-canceled" }, { signal: controller.signal }),
).rejects.toThrow("Exa caller canceled");
expect(fetchMock).not.toHaveBeenCalled();
} finally {
fetchMock.mockRestore();
}
});
it("aborts the guarded Exa request without losing the caller's reason", async () => {
const fetchMock = vi.spyOn(globalThis, "fetch").mockImplementation(
async (_url, init) =>
await new Promise<Response>((_resolve, reject) => {
if (!init?.signal) {
reject(new Error("Exa request lost caller cancellation"));
return;
}
init.signal.addEventListener("abort", () => reject(init.signal?.reason as Error), {
once: true,
});
}),
);
const tool = createExaWebSearchProvider().createTool({
config: {
plugins: { entries: { exa: { config: { webSearch: { apiKey: "exa-test-key" } } } } },
},
searchConfig: {},
});
if (!tool) {
throw new Error("Expected tool definition");
}
const controller = new AbortController();
const result = tool.execute(
{ query: "exa in-flight cancellation" },
{ signal: controller.signal },
);
try {
await vi.waitFor(() => expect(fetchMock).toHaveBeenCalledOnce());
controller.abort(new Error("Exa request canceled in flight"));
await expect(result).rejects.toThrow("Exa request canceled in flight");
expect(fetchMock.mock.calls[0]?.[1]?.signal?.aborted).toBe(true);
} finally {
fetchMock.mockRestore();
}
});
it("exposes the expected metadata and selection wiring", () => {
const provider = createExaWebSearchProvider();
if (!provider.applySelectionConfig) {

5
extensions/exa/src/exa-web-search-provider.ts
View File

@@ -67,9 +67,10 @@ export function createExaWebSearchProvider(): WebSearchProviderPlugin {
description:
"Search the web using Exa AI. Supports neural or keyword search, publication date filters, and optional highlights or text extraction.",
parameters: ExaSearchSchema,
execute: async (args) => {
execute: async (args, context) => {
context?.signal?.throwIfAborted();
const { executeExaWebSearchProviderTool } = await loadExaWebSearchRuntime();
return await executeExaWebSearchProviderTool(ctx, args);
return await executeExaWebSearchProviderTool(ctx, args, context?.signal);
},
}),
};

5
extensions/minimax/src/minimax-web-search-provider.runtime.ts
View File

@@ -127,6 +127,7 @@ async function runMiniMaxSearch(params: {
apiKey: string;
endpoint: string;
timeoutSeconds: number;
signal?: AbortSignal;
}): Promise<{
results: Array<Record<string, unknown>>;
relatedSearches?: string[];
@@ -135,6 +136,7 @@ async function runMiniMaxSearch(params: {
{
url: params.endpoint,
timeoutSeconds: params.timeoutSeconds,
signal: params.signal,
init: {
method: "POST",
headers: {
@@ -202,6 +204,7 @@ function missingMiniMaxKeyPayload() {
export async function executeMiniMaxWebSearchProviderTool(
ctx: { config?: Record<string, unknown>; searchConfig?: SearchConfigRecord },
args: Record<string, unknown>,
signal?: AbortSignal,
): Promise<Record<string, unknown>> {
const searchConfig = mergeScopedSearchConfig(
ctx.searchConfig,
@@ -244,8 +247,10 @@ export async function executeMiniMaxWebSearchProviderTool(
apiKey,
endpoint,
timeoutSeconds,
signal,
});
signal?.throwIfAborted();
const payload: Record<string, unknown> = {
query,
provider: "minimax",

74
extensions/minimax/src/minimax-web-search-provider.test.ts
View File

@@ -1,6 +1,7 @@
// Minimax tests cover minimax web search provider plugin behavior.
import { afterEach, beforeEach, describe, expect, it } from "vitest";
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
import { minimaxWebSearchTesting } from "../test-api.js";
import { createMiniMaxWebSearchProvider } from "./minimax-web-search-provider.js";
const {
MINIMAX_SEARCH_ENDPOINT_GLOBAL,
@@ -42,6 +43,77 @@ describe("minimax web search provider", () => {
restoreEnvValue("MINIMAX_API_KEY", originalApiKey);
});
it("does not send an already canceled MiniMax search", async () => {
const fetchMock = vi.spyOn(globalThis, "fetch").mockResolvedValue(
new Response(JSON.stringify({ organic: [], base_resp: { status_code: 0 } }), {
status: 200,
headers: { "content-type": "application/json" },
}),
);
const tool = createMiniMaxWebSearchProvider().createTool({
config: {
plugins: {
entries: { minimax: { config: { webSearch: { apiKey: "minimax-test-key" } } } },
},
},
searchConfig: {},
});
if (!tool) {
throw new Error("Expected tool definition");
}
const controller = new AbortController();
controller.abort(new Error("MiniMax caller canceled"));
try {
await expect(
tool.execute({ query: "minimax pre-canceled" }, { signal: controller.signal }),
).rejects.toThrow("MiniMax caller canceled");
expect(fetchMock).not.toHaveBeenCalled();
} finally {
fetchMock.mockRestore();
}
});
it("aborts the guarded MiniMax request with the caller's reason", async () => {
const fetchMock = vi.spyOn(globalThis, "fetch").mockImplementation(
async (_url, init) =>
await new Promise<Response>((_resolve, reject) => {
if (!init?.signal) {
reject(new Error("MiniMax request lost caller cancellation"));
return;
}
init.signal.addEventListener("abort", () => reject(init.signal?.reason as Error), {
once: true,
});
}),
);
const tool = createMiniMaxWebSearchProvider().createTool({
config: {
plugins: {
entries: { minimax: { config: { webSearch: { apiKey: "minimax-test-key" } } } },
},
},
searchConfig: {},
});
if (!tool) {
throw new Error("Expected tool definition");
}
const controller = new AbortController();
const result = tool.execute(
{ query: "minimax in-flight cancellation" },
{ signal: controller.signal },
);
try {
await vi.waitFor(() => expect(fetchMock).toHaveBeenCalledOnce());
controller.abort(new Error("MiniMax request canceled in flight"));
await expect(result).rejects.toThrow("MiniMax request canceled in flight");
expect(fetchMock.mock.calls[0]?.[1]?.signal?.aborted).toBe(true);
} finally {
fetchMock.mockRestore();
}
});
describe("resolveMiniMaxRegion", () => {
it("returns global by default", () => {
expect(resolveMiniMaxRegion()).toBe("global");

5
extensions/minimax/src/minimax-web-search-provider.ts
View File

@@ -52,9 +52,10 @@ export function createMiniMaxWebSearchProvider(): WebSearchProviderPlugin {
description:
"Search the web using MiniMax Search API. Returns titles, URLs, snippets, and related search suggestions.",
parameters: MiniMaxSearchSchema,
execute: async (args) => {
execute: async (args, context) => {
context?.signal?.throwIfAborted();
const { executeMiniMaxWebSearchProviderTool } = await loadMiniMaxWebSearchRuntime();
return await executeMiniMaxWebSearchProviderTool(ctx, args);
return await executeMiniMaxWebSearchProviderTool(ctx, args, context?.signal);
},
}),
};

30
extensions/ollama/src/web-search-provider.test.ts
View File

@@ -116,6 +116,7 @@ async function runOllamaWebSearch(
type GuardedRequest = {
url: string;
signal?: AbortSignal;
init: {
method: string;
headers: Record<string, string>;
@@ -227,6 +228,35 @@ describe("ollama web search provider", () => {
expect(fetchRequest().url).toBe("http://ollama.local:11434/api/experimental/web_search");
});
it("passes caller cancellation to the guard without replacing its owned timeout", async () => {
mockSuccessfulSearchResponse();
const tool = createOllamaWebSearchProvider().createTool({ config: createOllamaConfig() });
if (!tool) {
throw new Error("Expected tool definition");
}
const controller = new AbortController();
await tool.execute({ query: "ollama active cancellation" }, { signal: controller.signal });
expect(fetchRequest().signal).toBe(controller.signal);
expect(fetchRequest().init.signal).toBeUndefined();
});
it("does not start fallback attempts after caller cancellation", async () => {
mockSuccessfulSearchResponse();
const tool = createOllamaWebSearchProvider().createTool({ config: createOllamaConfig() });
if (!tool) {
throw new Error("Expected tool definition");
}
const controller = new AbortController();
controller.abort(new Error("Ollama caller canceled"));
await expect(
tool.execute({ query: "ollama pre-canceled" }, { signal: controller.signal }),
).rejects.toThrow("Ollama caller canceled");
expect(fetchWithSsrFGuardMock).not.toHaveBeenCalled();
});
it.each<[string, () => OpenClawConfig, string]>([
[
"prefers the plugin web search base URL over the model provider host",

13
extensions/ollama/src/web-search-provider.ts
View File

@@ -189,6 +189,7 @@ async function runOllamaWebSearch(params: {
config?: OpenClawConfig;
query: string;
count?: number;
signal?: AbortSignal;
}): Promise<Record<string, unknown>> {
const query = params.query.trim();
if (!query) {
@@ -209,6 +210,7 @@ async function runOllamaWebSearch(params: {
let payload: OllamaWebSearchResponse | undefined;
let lastError: Error | undefined;
for (const attempt of attempts) {
params.signal?.throwIfAborted();
const headers: Record<string, string> = { "Content-Type": "application/json" };
if (attempt.apiKey) {
headers.Authorization = `Bearer ${attempt.apiKey}`;
@@ -222,6 +224,7 @@ async function runOllamaWebSearch(params: {
},
// Guard-owned timeoutMs also bounds DNS/proxy preflight; init.signal does not.
timeoutMs: DEFAULT_OLLAMA_WEB_SEARCH_TIMEOUT_MS,
...(params.signal ? { signal: params.signal } : {}),
policy: buildOllamaBaseUrlSsrFPolicy(attempt.baseUrl),
auditContext: "ollama-web-search.search",
});
@@ -246,6 +249,7 @@ async function runOllamaWebSearch(params: {
throw new Error(message);
}
payload = await readOllamaWebSearchResponse(response);
params.signal?.throwIfAborted();
break;
} catch (error) {
if (error instanceof Error) {
@@ -358,15 +362,18 @@ export function createOllamaWebSearchProvider(): WebSearchProviderPlugin {
description:
"Search the web using Ollama's web search API. Returns titles, URLs, and snippets from the configured Ollama host.",
parameters: OLLAMA_WEB_SEARCH_SCHEMA,
execute: async (args) =>
await runOllamaWebSearch({
execute: async (args, context) => {
context?.signal?.throwIfAborted();
return await runOllamaWebSearch({
config: ctx.config,
query: readStringParam(args, "query", { required: true }),
count: readPositiveIntegerParam(args, "count", {
max: 10,
message: "count must be an integer from 1 to 10.",
}),
}),
signal: context?.signal,
});
},
}),
};
}

5
extensions/parallel/src/parallel-web-search-provider.runtime.ts
View File

@@ -123,6 +123,7 @@ async function runParallelSearch(params: {
sessionId?: string;
clientModel?: string;
timeoutSeconds: number;
signal?: AbortSignal;
}): Promise<ParallelSearchResponse> {
const body: Record<string, unknown> = {
search_queries: [...params.searchQueries],
@@ -142,6 +143,7 @@ async function runParallelSearch(params: {
{
url: params.endpoint,
timeoutSeconds: params.timeoutSeconds,
signal: params.signal,
init: {
method: "POST",
headers: {
@@ -170,6 +172,7 @@ async function runParallelSearch(params: {
export async function executeParallelWebSearchProviderTool(
ctx: { config?: Record<string, unknown>; searchConfig?: SearchConfigRecord },
args: Record<string, unknown>,
signal?: AbortSignal,
): Promise<Record<string, unknown>> {
const searchConfig = mergeScopedSearchConfig(
ctx.searchConfig,
@@ -234,7 +237,9 @@ export async function executeParallelWebSearchProviderTool(
sessionId,
clientModel,
timeoutSeconds: resolveSearchTimeoutSeconds(searchConfig),
signal,
});
signal?.throwIfAborted();
const results = mapParallelResults(response);
const payload: Record<string, unknown> = {

44
extensions/parallel/src/parallel-web-search-provider.test.ts
View File

@@ -1,7 +1,12 @@
import { expectDefined } from "@openclaw/normalization-core";
import { beforeEach, describe, expect, it, vi } from "vitest";
import { createStreamingResponse } from "../../test-support/streaming-error-response.js";
type EndpointCall = { url: string; timeoutSeconds: number; init: RequestInit };
type EndpointCall = {
url: string;
timeoutSeconds: number;
init: RequestInit;
signal?: AbortSignal;
};
type JsonRecord = Record<string, unknown>;
type ToolParameters = {
properties: Record<
@@ -307,6 +312,30 @@ describe("parallel web search provider", () => {
expect(result).not.toHaveProperty("objective");
expect(result).toMatchObject({ provider: "parallel" });
});
it("forwards paid-search cancellation to the guarded endpoint", async () => {
enqueueJson();
const controller = new AbortController();
await paidTool().execute(
{ search_queries: ["parallel active cancellation"] },
{ signal: controller.signal },
);
expect(endpointCall(0).signal).toBe(controller.signal);
});
it("does not bill an already canceled paid search", async () => {
enqueueJson();
const controller = new AbortController();
controller.abort(new Error("Parallel caller canceled"));
await expect(
paidTool().execute(
{ search_queries: ["parallel pre-canceled"] },
{ signal: controller.signal },
),
).rejects.toThrow("Parallel caller canceled");
expect(endpointMockState.calls).toHaveLength(0);
});
it("returns an error payload when search_queries is missing or empty", async () => {
const tool = paidTool();
expect(await tool.execute({ objective: "Find OpenClaw on GitHub" })).toMatchObject({
@@ -669,6 +698,19 @@ describe("runParallelMcpSearch", () => {
});
});
describe("parallel-free web search provider", () => {
it("keeps caller cancellation attached to every free MCP handshake step", async () => {
pushMcpHandshake({ search_id: "free-cancellation", results: [] });
const controller = new AbortController();
await freeTool().execute(
{ search_queries: ["parallel free cancellation control"] },
{ signal: controller.signal },
);
expect(endpointMockState.calls).toHaveLength(3);
expect(endpointMockState.calls.every((call) => call.signal === controller.signal)).toBe(true);
});
it("exposes keyless metadata without claiming auto-detect fallback", () => {
const provider = createParallelFreeWebSearchProvider();
expect(provider.id).toBe("parallel-free");

5
extensions/parallel/src/parallel-web-search-provider.ts
View File

@@ -62,9 +62,10 @@ export function createParallelWebSearchProvider(): WebSearchProviderPlugin {
description:
"Search the web using Parallel. Returns ranked, LLM-optimized dense excerpts from web sources. Pass an `objective` describing the underlying question along with 2-3 short keyword `search_queries` (Parallel's recommended pairing). For multi-step research, thread the prior result's `sessionId` back in as `session_id` to keep Parallel's context grouped.",
parameters: ParallelSearchSchema,
execute: async (args) => {
execute: async (args, context) => {
context?.signal?.throwIfAborted();
const { executeParallelWebSearchProviderTool } = await loadParallelWebSearchRuntime();
return await executeParallelWebSearchProviderTool(ctx, args);
return await executeParallelWebSearchProviderTool(ctx, args, context?.signal);
},
}),
};

8
extensions/perplexity/src/perplexity-web-search-provider.runtime.ts
View File

@@ -201,6 +201,7 @@ async function runPerplexitySearchApi(params: {
apiKey: string;
count: number;
timeoutSeconds: number;
signal?: AbortSignal;
country?: string;
searchDomainFilter?: string[];
searchRecencyFilter?: string;
@@ -243,6 +244,7 @@ async function runPerplexitySearchApi(params: {
{
url: PERPLEXITY_SEARCH_ENDPOINT,
timeoutSeconds: params.timeoutSeconds,
signal: params.signal,
init: {
method: "POST",
headers: buildPerplexityRequestHeaders(params.apiKey, true),
@@ -274,6 +276,7 @@ async function runPerplexitySearch(params: {
baseUrl: string;
model: string;
timeoutSeconds: number;
signal?: AbortSignal;
freshness?: string;
}): Promise<{ content: string; citations: string[] }> {
const endpoint = `${params.baseUrl.trim().replace(/\/$/, "")}/chat/completions`;
@@ -289,6 +292,7 @@ async function runPerplexitySearch(params: {
{
url: endpoint,
timeoutSeconds: params.timeoutSeconds,
signal: params.signal,
init: {
method: "POST",
headers: buildPerplexityRequestHeaders(params.apiKey),
@@ -311,6 +315,7 @@ async function runPerplexitySearch(params: {
export async function executePerplexitySearch(
args: Record<string, unknown>,
searchConfig?: SearchConfigRecord,
signal?: AbortSignal,
): Promise<Record<string, unknown>> {
const perplexityConfig = resolvePerplexityConfig(searchConfig);
const runtime = resolvePerplexityTransport(perplexityConfig);
@@ -499,6 +504,7 @@ export async function executePerplexitySearch(
baseUrl: runtime.baseUrl,
model: runtime.model,
timeoutSeconds,
signal,
freshness,
});
return {
@@ -523,6 +529,7 @@ export async function executePerplexitySearch(
apiKey: runtime.apiKey,
count: resolveSearchCount(count, DEFAULT_SEARCH_COUNT),
timeoutSeconds,
signal,
country: country ?? undefined,
searchDomainFilter: domainFilter,
searchRecencyFilter: freshness,
@@ -541,6 +548,7 @@ export async function executePerplexitySearch(
(payload as { tookMs: number }).tookMs = Date.now() - start;
}
signal?.throwIfAborted();
writeCachedSearchPayload(cacheKey, payload, resolveSearchCacheTtlMs(searchConfig));
return payload;
}

66
extensions/perplexity/src/perplexity-web-search-provider.test.ts
View File

@@ -43,6 +43,72 @@ describe("perplexity web search provider", () => {
);
});
it.each([
{ name: "native Search API", webSearch: { apiKey: "pplx-test" } },
{
name: "chat completions",
webSearch: { apiKey: "pplx-test", baseUrl: "https://api.perplexity.ai" },
},
])("does not start an already canceled $name request", async ({ webSearch }) => {
withTrustedWebSearchEndpointMock.mockReset();
withTrustedWebSearchEndpointMock.mockResolvedValue({ results: [] });
const tool = createPerplexityWebSearchProvider().createTool({
config: { plugins: { entries: { perplexity: { config: { webSearch } } } } },
searchConfig: {},
});
if (!tool) {
throw new Error("Expected tool definition");
}
const controller = new AbortController();
controller.abort(new Error("Perplexity caller canceled"));
await expect(
tool.execute({ query: "perplexity pre-canceled" }, { signal: controller.signal }),
).rejects.toThrow("Perplexity caller canceled");
expect(withTrustedWebSearchEndpointMock).not.toHaveBeenCalled();
});
it.each([
{ name: "native Search API", webSearch: { apiKey: "pplx-test" } },
{
name: "chat completions",
webSearch: { apiKey: "pplx-test", baseUrl: "https://api.perplexity.ai" },
},
])("cancels an in-flight $name request", async ({ name, webSearch }) => {
withTrustedWebSearchEndpointMock.mockReset();
withTrustedWebSearchEndpointMock.mockImplementation(
async (params: { signal?: AbortSignal }) =>
await new Promise<never>((_resolve, reject) => {
if (!params.signal) {
reject(new Error("Perplexity request lost caller cancellation"));
return;
}
params.signal.addEventListener("abort", () => reject(params.signal?.reason as Error), {
once: true,
});
}),
);
const tool = createPerplexityWebSearchProvider().createTool({
config: { plugins: { entries: { perplexity: { config: { webSearch } } } } },
searchConfig: {},
});
if (!tool) {
throw new Error("Expected tool definition");
}
const controller = new AbortController();
const result = tool.execute(
{ query: `perplexity in-flight cancellation ${name}` },
{ signal: controller.signal },
);
await vi.waitFor(() => expect(withTrustedWebSearchEndpointMock).toHaveBeenCalledOnce());
controller.abort(new Error("Perplexity request canceled in flight"));
await expect(result).rejects.toThrow("Perplexity request canceled in flight");
expect(withTrustedWebSearchEndpointMock.mock.calls[0]?.[0]?.signal).toBe(controller.signal);
withTrustedWebSearchEndpointMock.mockReset();
});
it("infers provider routing from api key prefixes", () => {
expect(testing.inferPerplexityBaseUrlFromApiKey("pplx-abc")).toBe("direct");
expect(testing.inferPerplexityBaseUrlFromApiKey("sk-or-v1-abc")).toBe("openrouter");

5
extensions/perplexity/src/perplexity-web-search-provider.ts
View File

@@ -97,9 +97,10 @@ function createPerplexityToolDefinition(
? "Search the web using Perplexity Sonar via Perplexity/OpenRouter chat completions. Returns AI-synthesized answers with citations from web-grounded search."
: "Search the web using Perplexity. Runtime routing decides between native Search API and Sonar chat-completions compatibility. Structured filters are available on the native Search API path.",
parameters: createPerplexityParameters(schemaTransport),
execute: async (args) => {
execute: async (args, context) => {
context?.signal?.throwIfAborted();
const { executePerplexitySearch } = await loadPerplexityWebSearchRuntime();
return await executePerplexitySearch(args, searchConfig);
return await executePerplexitySearch(args, searchConfig, context?.signal);
},
};
}

13
extensions/qa-lab/src/qa-web-search-provider.test.ts
View File

@@ -46,6 +46,19 @@ describe("qa-lab web search provider", () => {
expect(JSON.stringify(result)).toContain("Deterministic QA Lab web_search result");
});
it("preserves caller cancellation instead of returning a fixture result", async () => {
const tool = createQaLabWebSearchProvider().createTool({});
if (!tool) {
throw new Error("expected QA Lab web search tool");
}
const controller = new AbortController();
controller.abort(new Error("QA Lab caller canceled"));
await expect(
tool.execute({ query: "qa pre-canceled" }, { signal: controller.signal }),
).rejects.toThrow("QA Lab caller canceled");
});
it("keeps malformed failure-path calls as tool failures", async () => {
const provider = createQaLabWebSearchProvider();
const tool = provider.createTool({});

3
extensions/qa-lab/src/qa-web-search-provider.ts
View File

@@ -52,7 +52,8 @@ export function createQaLabWebSearchProvider(): WebSearchProviderPlugin {
description:
"Search a deterministic QA Lab fixture corpus. This provider is for QA runtime parity only and never calls the public web.",
parameters: QaLabWebSearchSchema,
execute: async (args) => {
execute: async (args, context) => {
context?.signal?.throwIfAborted();
const query = readStringParam(args, "query", { required: true });
if (query === QA_LAB_WEB_SEARCH_DENIED_INPUT_QUERY) {
throw new Error("QA Lab web_search denied input sentinel");