Q98: external calls have a time limit and stop with the run

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

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<Guarde
   for (let hop = 0; hop <= MAX_REDIRECTS; hop += 1) {
     const url = parseHttpUrl(current);
     await assertPublicHost(url.hostname);
-    const controller = new AbortController();
-    const timer = setTimeout(() => 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)}`);
     }

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<string, unknown>,
-    options?: { tolerateNotFound?: boolean },
+    options?: { tolerateNotFound?: boolean; timeoutMs?: number },
   ): Promise<Record<string, unknown>> {
+    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<string, unknown> = {}) => {
+        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<string, unknown>);
+            finish(null, JSON.parse(responseData) as Record<string, unknown>);
           } 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();
     });

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<string, unknown>): 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("<p>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("<p>landed</p>", { 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: "<h1>Not Found</h1>" }]).fetcher);
     expect(missing).toBe("ERROR: http://93.184.216.34/gone answered with status 404: Not Found");

tests/unit/composio-tool-pages.test.ts

@@ -8,6 +8,7 @@ const requested: string[] = [];
 const written: string[] = [];
 const pages = new Map<string, Record<string, unknown>>();
 const executions = new Map<string, Record<string, unknown>>();
+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") };