From 552dfcb6ce7c233926813167e889e538c0b3d0c6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=9B=B7=E7=94=B5=E8=8A=BD=E8=A1=A3?= Date: Thu, 8 Oct 2026 16:12:45 -0400 Subject: [PATCH] Q98: external calls have a time limit and stop with the run Co-Authored-By: Claude Opus 5.5 --- src/main/services/agent-web-tools.ts | 28 +++++++--- src/main/services/composio/composio-client.ts | 51 ++++++++++++------- tests/unit/agent-web-tools.test.ts | 41 ++++++++++++++- tests/unit/composio-tool-pages.test.ts | 21 +++++++- 4 files changed, 113 insertions(+), 28 deletions(-) diff --git a/src/main/services/agent-web-tools.ts b/src/main/services/agent-web-tools.ts index 9330a5aa..8ac38ebe 100644 --- a/src/main/services/agent-web-tools.ts +++ b/src/main/services/agent-web-tools.ts @@ -1,6 +1,7 @@ import { lookup } from "node:dns/promises"; import { isIP } from "node:net"; import { requireString } from "./agent-tool-args"; +import { currentRunSignal, stopReasonOf } from "./run-control"; import type { ToolCall, ToolDefinition } from "./provider-tool-turn"; import { PAGE_TOOL_PROPERTIES, pageText, readPageArgs, type PageRequest } from "./text-paging"; @@ -107,26 +108,37 @@ async function guardedFetch(rawUrl: string, fetcher: WebFetcher): Promise controller.abort(), FETCH_TIMEOUT_MS); + // The limit covers headers and body; a stopped run aborts the fetch too. + const timeout = new AbortController(); + const timer = setTimeout(() => timeout.abort(), FETCH_TIMEOUT_MS); + const runSignal = currentRunSignal(); + const signal = runSignal ? AbortSignal.any([timeout.signal, runSignal]) : timeout.signal; let response: Response; + let text = ""; + let redirect: string | null = null; try { response = await fetcher(url.toString(), { headers: { "user-agent": USER_AGENT, accept: "text/html,application/xhtml+xml,text/plain,*/*" }, redirect: "manual", - signal: controller.signal, + signal, }); + const location = response.headers.get("location"); + if (response.status >= 300 && response.status < 400 && location) { + redirect = new URL(location, url).toString(); + await response.body?.cancel(); + } else { + text = await readBodyCapped(response); + } } catch (error) { - throw new Error(`Fetching ${url.hostname} failed: ${error instanceof Error ? error.message : String(error)}`); + const reason = runSignal?.aborted ? stopReasonOf(runSignal) : timeout.signal.aborted ? `no answer within ${FETCH_TIMEOUT_MS / 1000}s.` : error instanceof Error ? error.message : String(error); + throw new Error(`Fetching ${url.hostname} failed: ${reason}`); } finally { clearTimeout(timer); } - const location = response.headers.get("location"); - if (response.status >= 300 && response.status < 400 && location) { - current = new URL(location, url).toString(); + if (redirect) { + current = redirect; continue; } - const text = await readBodyCapped(response); if (response.status < 200 || response.status >= 300) { throw new Error(`${url.toString()} answered with status ${response.status}: ${statusExcerpt(response.headers.get("content-type") ?? "", text)}`); } diff --git a/src/main/services/composio/composio-client.ts b/src/main/services/composio/composio-client.ts index 6a64f25b..11687b66 100644 --- a/src/main/services/composio/composio-client.ts +++ b/src/main/services/composio/composio-client.ts @@ -30,6 +30,8 @@ export interface ComposioConnectedAccount { export type ComposioToolExecutionResult = { successful: true; data: unknown } | { successful: false; error: string }; const MAX_RESPONSE_BYTES = 2_000_000; +const REQUEST_TIMEOUT_MS = 30_000; +const EXECUTE_TIMEOUT_MS = 120_000; export class ComposioClient { constructor(private readonly apiKey: string) {} @@ -139,7 +141,7 @@ export class ComposioClient { arguments: input.arguments, }; if (input.connectedAccountId) body.connected_account_id = input.connectedAccountId; - const response = await this.request("POST", url, body); + const response = await this.request("POST", url, body, { timeoutMs: EXECUTE_TIMEOUT_MS }); if (typeof response.successful !== "boolean") throw new Error(`Composio returned no execution status for ${input.toolName}.`); if (response.successful) return { successful: true, data: response.data }; return { successful: false, error: typeof response.error === "string" && response.error.trim() ? response.error.trim() : "Composio gave no reason." }; @@ -149,58 +151,71 @@ export class ComposioClient { method: string, url: string, body?: Record, - options?: { tolerateNotFound?: boolean }, + options?: { tolerateNotFound?: boolean; timeoutMs?: number }, ): Promise> { + const timeoutMs = options?.timeoutMs ?? REQUEST_TIMEOUT_MS; return new Promise((resolve, reject) => { - const request = net.request({ method, url }); const runSignal = currentRunSignal(); if (runSignal?.aborted) { reject(new Error(stopReasonOf(runSignal))); return; } - runSignal?.addEventListener("abort", () => { + const request = net.request({ method, url }); + let settled = false; + const finish = (error: Error | null, value: Record = {}) => { + if (settled) return; + settled = true; + clearTimeout(timer); + runSignal?.removeEventListener("abort", onStop); + if (!error) { + resolve(value); + return; + } request.abort(); - reject(new Error(stopReasonOf(runSignal))); - }, { once: true }); + reject(error); + }; + const onStop = () => finish(new Error(stopReasonOf(runSignal!))); + const timer = setTimeout(() => { + console.warn(`[connectors] ${method} ${url} gave no answer within ${timeoutMs / 1000}s`); + finish(new Error(`Composio API ${method} ${url} gave no answer within ${timeoutMs / 1000}s.`)); + }, timeoutMs); + runSignal?.addEventListener("abort", onStop, { once: true }); request.setHeader("x-api-key", this.apiKey); request.setHeader("Accept", "application/json"); if (body) request.setHeader("Content-Type", "application/json"); const chunks: Buffer[] = []; let received = 0; - let aborted = false; request.on("response", (response) => { response.on("data", (chunk) => { - if (aborted) return; + if (settled) return; received += chunk.length; if (received > MAX_RESPONSE_BYTES) { - aborted = true; - request.abort(); - reject(new Error(`Composio API ${method} ${url} returned more than ${MAX_RESPONSE_BYTES} bytes; ask for fewer items or narrower fields.`)); + finish(new Error(`Composio API ${method} ${url} returned more than ${MAX_RESPONSE_BYTES} bytes; ask for fewer items or narrower fields.`)); return; } chunks.push(chunk); }); response.on("end", () => { - if (aborted) return; + if (settled) return; const responseData = Buffer.concat(chunks).toString("utf8"); if (response.statusCode === 404 && options?.tolerateNotFound) { - resolve({}); + finish(null); return; } if (response.statusCode && response.statusCode >= 400) { - reject(new Error(`Composio API ${method} ${url} returned ${response.statusCode}: ${responseData.slice(0, 200)}`)); + finish(new Error(`Composio API ${method} ${url} returned ${response.statusCode}: ${responseData.slice(0, 200)}`)); return; } try { - resolve(JSON.parse(responseData) as Record); + finish(null, JSON.parse(responseData) as Record); } catch { - reject(new Error(`Composio API returned non-JSON response: ${responseData.slice(0, 200)}`)); + finish(new Error(`Composio API returned non-JSON response: ${responseData.slice(0, 200)}`)); } }); - response.on("error", reject); + response.on("error", (error) => finish(error)); }); - request.on("error", reject); + request.on("error", (error) => finish(error)); if (body) request.write(JSON.stringify(body)); request.end(); }); diff --git a/tests/unit/agent-web-tools.test.ts b/tests/unit/agent-web-tools.test.ts index 6333030e..fa6770da 100644 --- a/tests/unit/agent-web-tools.test.ts +++ b/tests/unit/agent-web-tools.test.ts @@ -1,7 +1,8 @@ import { randomUUID } from "node:crypto"; -import { describe, expect, test } from "vitest"; +import { describe, expect, test, vi } from "vitest"; import { executeWebTool, htmlToText, isWebTool, parseSearchResults, WEB_TOOLS, type WebFetcher } from "../../src/main/services/agent-web-tools"; import type { ToolCall } from "../../src/main/services/provider-tool-turn"; +import { stopRun, withRunControl } from "../../src/main/services/run-control"; function call(name: string, args: Record): ToolCall { return { id: `call_${randomUUID()}`, name, arguments: args }; @@ -62,6 +63,44 @@ describe("agent web tools", () => { expect(output).toContain("status 200"); }); + /** A page that sends its first bytes and then never finishes, until the request is aborted. */ + const tricklingFetcher: WebFetcher = async (_url, init) => new Response(new ReadableStream({ + start(stream) { + stream.enqueue(new TextEncoder().encode("

first")); + init.signal.addEventListener("abort", () => stream.error(init.signal.reason)); + }, + }), { status: 200, headers: { "content-type": "text/html" } }); + + test("a page that never finishes is given up after the time limit", async () => { + vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] }); + try { + const outcome = executeWebTool(call("fetch_url", { url: "http://93.184.216.34/slow" }), tricklingFetcher); + await vi.advanceTimersByTimeAsync(30_000); + expect(await outcome).toBe("ERROR: Fetching 93.184.216.34 failed: no answer within 30s."); + } finally { + vi.useRealTimers(); + } + }); + + test("a redirect whose body never finishes is still followed", async () => { + const requested: string[] = []; + const fetcher: WebFetcher = async (url, init) => { + requested.push(url); + if (requested.length === 2) return new Response("

landed

", { status: 200, headers: { "content-type": "text/html" } }); + const stalled = await tricklingFetcher(url, init); + return new Response(stalled.body, { status: 302, headers: { location: "http://93.184.216.35/final" } }); + }; + expect(await executeWebTool(call("fetch_url", { url: "http://93.184.216.34/start" }), fetcher)).toContain("landed"); + }); + + test("stopping the run stops a fetch in progress", async () => { + const runId = `run_${randomUUID()}`; + const outcome = withRunControl(runId, () => executeWebTool(call("fetch_url", { url: "http://93.184.216.34/slow" }), tricklingFetcher)); + await new Promise((resolve) => setTimeout(resolve, 20)); + stopRun(runId); + expect(await outcome).toBe("ERROR: Fetching 93.184.216.34 failed: Stopped by the operator."); + }); + test("a page or search answered with an error status is a failed call, not data", async () => { const missing = await executeWebTool(call("fetch_url", { url: "http://93.184.216.34/gone" }), fetcherReturning([{ status: 404, body: "

Not Found

" }]).fetcher); expect(missing).toBe("ERROR: http://93.184.216.34/gone answered with status 404: Not Found"); diff --git a/tests/unit/composio-tool-pages.test.ts b/tests/unit/composio-tool-pages.test.ts index a0f2e493..3ebe1389 100644 --- a/tests/unit/composio-tool-pages.test.ts +++ b/tests/unit/composio-tool-pages.test.ts @@ -8,6 +8,7 @@ const requested: string[] = []; const written: string[] = []; const pages = new Map>(); const executions = new Map>(); +const aborted: string[] = []; vi.mock("electron", () => ({ net: { @@ -15,9 +16,10 @@ vi.mock("electron", () => ({ const request = new EventEmitter() as EventEmitter & { setHeader: () => void; write: (body: string) => void; end: () => void; abort: () => void }; request.setHeader = () => undefined; request.write = (body: string) => { written.push(`${url} ${body}`); }; - request.abort = () => undefined; + request.abort = () => { aborted.push(url); }; request.end = () => { requested.push(url); + if (url.includes("/tools/execute/GITHUB_HANG")) return; const query = new URL(url).searchParams; const page = url.includes("/tools/execute/") ? executions.get(url.split("/tools/execute/")[1]) ?? { data: { number: 42 }, error: null, successful: true } : pages.get(`${query.get("toolkit_slug")}:${query.get("cursor") ?? ""}`); const response = new EventEmitter() as EventEmitter & { statusCode: number }; @@ -54,6 +56,23 @@ describe("a toolkit's tools are all listed", () => { }); }); +describe("a Composio call that never answers", () => { + test("is given up after its time limit and its request is aborted", async () => { + vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] }); + try { + const pending = new ComposioClient("ak_hang").executeTool({ toolName: "GITHUB_HANG", arguments: {}, userId: "u" }); + const outcome = pending.then(() => "answered", (error: Error) => error.message); + await vi.advanceTimersByTimeAsync(119_000); + expect(aborted.some((url) => url.includes("GITHUB_HANG"))).toBe(false); + await vi.advanceTimersByTimeAsync(1_000); + expect(await outcome).toBe("Composio API POST https://backend.composio.dev/api/v3/tools/execute/GITHUB_HANG gave no answer within 120s."); + expect(aborted.some((url) => url.includes("GITHUB_HANG"))).toBe(true); + } finally { + vi.useRealTimers(); + } + }); +}); + describe("agents reach every action through search and run", () => { const vault = { encryptSecret: (secret: string) => Buffer.from(secret, "utf8"), decryptSecret: (ciphertext: Buffer) => ciphertext.toString("utf8") };