fix(subagent): bound per-tool execution with timeout + idempotent retry (takeover of #2086)

toolLoop and the legacy Anthropic subagent path ran handler.execute()
unbounded — a wedged pooler or half-open client socket squatted the worker
slot until the JOB-level wall-clock timeout reaped the whole job. Every tool
call is now bounded by GBRAIN_SUBAGENT_TOOL_TIMEOUT_MS (default 60s) and
idempotent tools retry transient/timeout failures up to
GBRAIN_SUBAGENT_TOOL_MAX_ATTEMPTS (default 2, parsed as a count — not through
resolveAiTimeoutMs). Retries emit a typed tool_retry heartbeat (no casts).

The original PR's queue.ts retryJob hunk is dropped: master already resets
started_at/attempts on retry.

Co-authored-by: JiraiyaETH <JiraiyaETH@users.noreply.github.com>
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

src/core/ai/gateway.ts

@@ -80,6 +80,20 @@ const AI_CHAT_TIMEOUT_MS = resolveAiTimeoutMs('GBRAIN_AI_CHAT_TIMEOUT_MS', 300_0
 const AI_EMBED_TIMEOUT_MS = resolveAiTimeoutMs('GBRAIN_AI_EMBED_TIMEOUT_MS', 60_000);
 /** multimodal per request. */
 const AI_MULTIMODAL_TIMEOUT_MS = resolveAiTimeoutMs('GBRAIN_AI_MULTIMODAL_TIMEOUT_MS', 60_000);
+/** Per tool execution inside the tool loop. A hung DB/pooler/client call must
+ * settle as a failed tool result instead of squatting a worker slot until the
+ * job's wall-clock timeout expires. */
+const AI_TOOL_TIMEOUT_MS = resolveAiTimeoutMs('GBRAIN_SUBAGENT_TOOL_TIMEOUT_MS', 60_000);
+/** Max attempts for idempotent transient/timeout tool failures. A COUNT, not a
+ * timeout — parsed by its own resolver, not resolveAiTimeoutMs. */
+const AI_TOOL_MAX_ATTEMPTS = resolveAiCount('GBRAIN_SUBAGENT_TOOL_MAX_ATTEMPTS', 2);
+
+function resolveAiCount(envVar: string, fallback: number): number {
+  const raw = process.env[envVar];
+  if (raw === undefined) return fallback;
+  const n = Number(raw);
+  return Number.isFinite(n) && n >= 1 ? Math.floor(n) : fallback;
+}
 
 /**
  * Compose a caller signal with a default wall-clock timeout. When the caller
@@ -3260,6 +3274,10 @@ export interface ToolLoopOpts {
   abortSignal?: AbortSignal;
   /** Apply Anthropic cache_control to system + last tool. Silently ignored elsewhere. */
   cacheSystem?: boolean;
+  /** Per-tool wall-clock timeout. Defaults to GBRAIN_SUBAGENT_TOOL_TIMEOUT_MS or 60s. */
+  toolTimeoutMs?: number;
+  /** Max attempts for idempotent transient/timeout tool failures. Defaults to GBRAIN_SUBAGENT_TOOL_MAX_ATTEMPTS or 2. */
+  toolMaxAttempts?: number;
 
   /** Crash-replay state. When set, the loop resumes from the recorded position. */
   replayState?: ToolLoopReplayState;
@@ -3334,6 +3352,8 @@ export async function toolLoop(opts: ToolLoopOpts): Promise<ToolLoopResult> {
   const maxTurns = opts.maxTurns ?? 20;
   const maxTokens = opts.maxTokens ?? defaultMaxOutputTokens(opts.model ?? getChatModel());
   const handlers = opts.toolHandlers;
+  const toolTimeoutMs = opts.toolTimeoutMs ?? AI_TOOL_TIMEOUT_MS;
+  const toolMaxAttempts = Math.max(1, Math.floor(opts.toolMaxAttempts ?? AI_TOOL_MAX_ATTEMPTS));
   const totalUsage: ChatResult['usage'] = {
     input_tokens: 0,
     output_tokens: 0,
@@ -3492,10 +3512,25 @@ export async function toolLoop(opts: ToolLoopOpts): Promise<ToolLoopResult> {
         );
       }
 
-      // Step 3: execute (side effect).
+      // Step 3: execute (side effect). Bound every call: DB/pooler/client
+      // hangs must settle as failed tool results instead of squatting a worker
+      // slot until the whole job timeout expires. Only idempotent tools retry.
       opts.onHeartbeat?.('tool_called', { turn_idx: turnIdx, tool_name: call.toolName });
       try {
-        const output = await handler.execute(call.input, opts.abortSignal ?? new AbortController().signal);
+        const output = await executeToolWithTimeoutAndRetry({
+          toolName: call.toolName,
+          baseSignal: opts.abortSignal,
+          timeoutMs: toolTimeoutMs,
+          maxAttempts: toolMaxAttempts,
+          idempotent: handler.idempotent === true,
+          execute: signal => handler.execute(call.input, signal),
+          onRetry: (attempt, error) => opts.onHeartbeat?.('tool_retry', {
+            turn_idx: turnIdx,
+            tool_name: call.toolName,
+            attempt,
+            error: error instanceof Error ? error.message : String(error),
+          }),
+        });
         // Step 4: settle complete.
         await opts.onToolCallComplete?.(gbrainToolUseId, output);
         toolResultBlocks.push({
@@ -3538,6 +3573,110 @@ export async function toolLoop(opts: ToolLoopOpts): Promise<ToolLoopResult> {
   return { finalText, totalTurns: turnIdx, totalUsage, stopReason, messages };
 }
 
+// ---- Per-tool timeout + retry (subagent tool loop) ----
+
+/** Thrown when a single tool execution exceeds its wall-clock timeout. */
+export class ToolCallTimeoutError extends Error {
+  constructor(public toolName: string, public timeoutMs: number) {
+    super(`tool "${toolName}" timed out after ${timeoutMs}ms`);
+    this.name = 'ToolCallTimeoutError';
+  }
+}
+
+export interface ExecuteToolWithTimeoutAndRetryOpts {
+  toolName: string;
+  /** Caller abort (job cancel / worker shutdown). Composed with the per-attempt timeout. */
+  baseSignal?: AbortSignal;
+  /** Per-attempt wall-clock timeout in ms. */
+  timeoutMs: number;
+  /** Total attempts allowed for idempotent tools; non-idempotent tools always run once. */
+  maxAttempts: number;
+  idempotent: boolean;
+  execute: (signal: AbortSignal) => Promise<unknown>;
+  /** Fires before each retry with the 1-based attempt number about to run. */
+  onRetry?: (attempt: number, error: unknown) => void;
+}
+
+/**
+ * Run one tool call bounded by a wall-clock timeout, retrying transient
+ * failures (timeouts + connection-class errors) for IDEMPOTENT tools only.
+ * The timeout wins even when the tool ignores its abort signal — the race
+ * settles and the loop feeds a failed tool result back to the model instead
+ * of wedging the worker until the job-level timeout reaps it.
+ */
+export async function executeToolWithTimeoutAndRetry(
+  opts: ExecuteToolWithTimeoutAndRetryOpts,
+): Promise<unknown> {
+  const maxAttempts = opts.idempotent ? Math.max(1, opts.maxAttempts) : 1;
+  let lastErr: unknown;
+  for (let attempt = 1; attempt <= maxAttempts; attempt++) {
+    try {
+      return await executeSingleToolAttempt(opts);
+    } catch (err) {
+      lastErr = err;
+      if (opts.baseSignal?.aborted) throw err;
+      if (attempt >= maxAttempts || !isRetryableToolError(err)) throw err;
+      opts.onRetry?.(attempt + 1, err);
+      await abortableSleep(Math.min(1000 * attempt, 3000), opts.baseSignal);
+    }
+  }
+  throw lastErr;
+}
+
+async function executeSingleToolAttempt(opts: ExecuteToolWithTimeoutAndRetryOpts): Promise<unknown> {
+  const timeoutController = new AbortController();
+  const timer = setTimeout(() => {
+    timeoutController.abort(new ToolCallTimeoutError(opts.toolName, opts.timeoutMs));
+  }, opts.timeoutMs);
+  const signal = opts.baseSignal
+    ? AbortSignal.any([opts.baseSignal, timeoutController.signal])
+    : timeoutController.signal;
+
+  let abortListener: (() => void) | undefined;
+  const abortPromise = new Promise<never>((_, reject) => {
+    abortListener = () => {
+      const reason = signal.reason;
+      reject(reason instanceof Error ? reason : new Error('tool call aborted'));
+    };
+    if (signal.aborted) abortListener();
+    else signal.addEventListener('abort', abortListener, { once: true });
+  });
+
+  const workPromise = Promise.resolve().then(() => opts.execute(signal));
+  // Promise.race may settle via timeout first; keep the abandoned work's
+  // eventual rejection from becoming an unhandledRejection.
+  workPromise.catch(() => {});
+  try {
+    return await Promise.race([workPromise, abortPromise]);
+  } finally {
+    clearTimeout(timer);
+    if (abortListener) signal.removeEventListener('abort', abortListener);
+  }
+}
+
+function isRetryableToolError(err: unknown): boolean {
+  if (err instanceof ToolCallTimeoutError) return true;
+  const name = err && typeof err === 'object' ? String((err as { name?: unknown }).name ?? '') : '';
+  if (name === 'TimeoutError') return true;
+  const message = err instanceof Error ? err.message : String(err);
+  return /CONNECT_TIMEOUT|connect timeout|Cannot connect|ECONNRESET|ECONNREFUSED|ETIMEDOUT|connection terminated|server closed the connection|pooler/i.test(message);
+}
+
+function abortableSleep(ms: number, signal?: AbortSignal): Promise<void> {
+  if (signal?.aborted) return Promise.reject(signal.reason ?? new Error('aborted'));
+  return new Promise((resolve, reject) => {
+    const onAbort = () => {
+      clearTimeout(timer);
+      reject(signal?.reason ?? new Error('aborted'));
+    };
+    const timer = setTimeout(() => {
+      signal?.removeEventListener('abort', onAbort);
+      resolve();
+    }, ms);
+    signal?.addEventListener('abort', onAbort, { once: true });
+  });
+}
+
 // ---- Reranker (v0.35.0.0+) ----
 
 /** Tagged error class for gateway.rerank() failures. `reason` classifies into the

src/core/minions/handlers/subagent-audit.ts

@@ -37,8 +37,10 @@ export interface SubagentHeartbeatEvent {
   ts: string;
   type: 'heartbeat';
   job_id: number;
-  event: 'llm_call_started' | 'llm_call_completed' | 'tool_called' | 'tool_result' | 'tool_failed';
+  event: 'llm_call_started' | 'llm_call_completed' | 'tool_called' | 'tool_result' | 'tool_failed' | 'tool_retry';
   turn_idx: number;
+  /** 1-based attempt number about to run, for tool_retry. */
+  attempt?: number;
   /** Tool name for tool_* events. Never the input — that may contain secrets. */
   tool_name?: string;
   /** ms elapsed for *_completed / tool_result / tool_failed. */

src/core/minions/handlers/subagent.ts

@@ -49,7 +49,7 @@ import {
 } from './subagent-audit.ts';
 import { resolveModel, isAnthropicProvider, TIER_DEFAULTS } from '../../model-config.ts';
 import { buildSystemPrompt, DEFAULT_SUBAGENT_SYSTEM } from '../system-prompt.ts';
-import { toolLoop as gatewayToolLoop } from '../../ai/gateway.ts';
+import { executeToolWithTimeoutAndRetry, toolLoop as gatewayToolLoop } from '../../ai/gateway.ts';
 import type { ChatToolDef, ChatMessage, ChatBlock, ChatResult, ToolHandler } from '../../ai/gateway.ts';
 import { classifyCapabilities } from '../../ai/capabilities.ts';
 import { randomUUIDv7 } from 'bun';
@@ -419,8 +419,15 @@ export function makeSubagentHandler(deps: SubagentDeps) {
           }
           await persistToolExecPending(engine, ctx.id, last.message_idx, use.id, use.name, use.input);
           try {
-            const output = await toolDef.execute(use.input, {
-              engine, jobId: ctx.id, remote: true, signal: ctx.signal,
+            const output = await executeToolWithTimeoutAndRetry({
+              toolName: use.name,
+              baseSignal: mergeSignals(ctx.signal, ctx.shutdownSignal),
+              timeoutMs: subagentToolTimeoutMs(),
+              maxAttempts: subagentToolMaxAttempts(),
+              idempotent: toolDef.idempotent === true,
+              execute: signal => toolDef.execute(use.input, {
+                engine, jobId: ctx.id, remote: true, signal,
+              }),
             });
             await persistToolExecComplete(engine, ctx.id, use.id, output);
             synthesizedResults.push({
@@ -732,11 +739,26 @@ export function makeSubagentHandler(deps: SubagentDeps) {
 
         const toolStart = Date.now();
         try {
-          const output = await toolDef.execute(use.input, {
-            engine,
-            jobId: ctx.id,
-            remote: true,
-            signal: ctx.signal,
+          const output = await executeToolWithTimeoutAndRetry({
+            toolName,
+            baseSignal: mergeSignals(ctx.signal, ctx.shutdownSignal),
+            timeoutMs: subagentToolTimeoutMs(),
+            maxAttempts: subagentToolMaxAttempts(),
+            idempotent: toolDef.idempotent === true,
+            execute: signal => toolDef.execute(use.input, {
+              engine,
+              jobId: ctx.id,
+              remote: true,
+              signal,
+            }),
+            onRetry: (attempt, error) => logSubagentHeartbeat({
+              job_id: ctx.id,
+              event: 'tool_retry',
+              turn_idx: turnIdx,
+              tool_name: toolName,
+              attempt,
+              error: error instanceof Error ? error.message : String(error),
+            }),
           });
           await persistToolExecComplete(engine, ctx.id, use.id, output);
           logSubagentHeartbeat({
@@ -1493,6 +1515,21 @@ async function persistToolExecFailed(
 
 // ── Internal: helpers ───────────────────────────────────────
 
+/** Per-tool wall-clock timeout for the legacy Anthropic path. Read at call
+ * time (not module load) so operators can tune it per worker restart. Same
+ * env vars as the gateway toolLoop defaults. */
+function subagentToolTimeoutMs(): number {
+  const raw = process.env.GBRAIN_SUBAGENT_TOOL_TIMEOUT_MS;
+  const n = raw === undefined ? 60_000 : Number(raw);
+  return Number.isFinite(n) && n > 0 ? n : 60_000;
+}
+
+function subagentToolMaxAttempts(): number {
+  const raw = process.env.GBRAIN_SUBAGENT_TOOL_MAX_ATTEMPTS;
+  const n = raw === undefined ? 2 : Number(raw);
+  return Number.isFinite(n) && n >= 1 ? Math.floor(n) : 2;
+}
+
 function asStringIfNotObject(value: unknown): string {
   if (typeof value === 'string') return value;
   try {

test/subagent-tool-timeout.test.ts

@@ -0,0 +1,138 @@
+// Per-tool timeout + retry inside the tool loop (takeover of #2086).
+//
+// Pre-fix, `toolLoop` ran `handler.execute()` unbounded: a wedged pooler or
+// half-open client socket squatted the worker slot until the JOB-level
+// wall-clock timeout reaped the whole job. These pin:
+//   - a hung tool settles as ToolCallTimeoutError even when it ignores abort
+//   - idempotent transient failures retry (bounded) and feed the successful
+//     result back into the loop
+//   - non-idempotent tools never retry
+
+import { afterEach, describe, expect, test } from 'bun:test';
+import {
+  __setChatTransportForTests,
+  executeToolWithTimeoutAndRetry,
+  toolLoop,
+  type ChatResult,
+} from '../src/core/ai/gateway.ts';
+
+afterEach(() => {
+  __setChatTransportForTests(null);
+});
+
+const usage = {
+  input_tokens: 1,
+  output_tokens: 1,
+  cache_read_tokens: 0,
+  cache_creation_tokens: 0,
+};
+
+describe('subagent tool execution timeout/retry', () => {
+  test('executeToolWithTimeoutAndRetry bounds a hung tool even if it ignores the abort signal', async () => {
+    const started = Date.now();
+    await expect(executeToolWithTimeoutAndRetry({
+      toolName: 'brain_search',
+      timeoutMs: 20,
+      maxAttempts: 1,
+      idempotent: true,
+      execute: async () => new Promise(() => {}),
+    })).rejects.toThrow('tool "brain_search" timed out after 20ms');
+    expect(Date.now() - started).toBeLessThan(500);
+  });
+
+  test('caller abort (baseSignal) wins over the timeout and is NOT retried', async () => {
+    const ctl = new AbortController();
+    const pending = executeToolWithTimeoutAndRetry({
+      toolName: 'brain_search',
+      baseSignal: ctl.signal,
+      timeoutMs: 60_000,
+      maxAttempts: 3,
+      idempotent: true,
+      execute: async () => new Promise(() => {}),
+    });
+    ctl.abort(new Error('job cancelled'));
+    await expect(pending).rejects.toThrow('job cancelled');
+  });
+
+  test('idempotent transient tool failures retry and feed the successful result back into the loop', async () => {
+    let chatCalls = 0;
+    __setChatTransportForTests(async (): Promise<ChatResult> => {
+      chatCalls += 1;
+      if (chatCalls === 1) {
+        return {
+          text: '',
+          blocks: [{ type: 'tool-call', toolCallId: 'call-1', toolName: 'brain_search', input: { query: 'supabase' } }],
+          stopReason: 'tool_calls',
+          usage,
+          model: 'test:model',
+          providerId: 'test',
+        };
+      }
+      return {
+        text: 'done',
+        blocks: [{ type: 'text', text: 'done' }],
+        stopReason: 'end',
+        usage,
+        model: 'test:model',
+        providerId: 'test',
+      };
+    });
+
+    let attempts = 0;
+    const retries: number[] = [];
+    const result = await toolLoop({
+      model: 'test:model',
+      initialMessages: [{ role: 'user', content: 'search' }],
+      tools: [{ name: 'brain_search', description: 'Search', inputSchema: { type: 'object' } }],
+      toolHandlers: new Map([['brain_search', {
+        idempotent: true,
+        async execute() {
+          attempts += 1;
+          if (attempts === 1) throw new Error('Cannot connect to database: CONNECT_TIMEOUT');
+          return { ok: true };
+        },
+      }]]),
+      toolTimeoutMs: 1000,
+      toolMaxAttempts: 2,
+      onHeartbeat(event, data) {
+        if (event === 'tool_retry') retries.push(data.attempt as number);
+      },
+    });
+
+    expect(result.stopReason).toBe('end');
+    expect(result.finalText).toBe('done');
+    expect(attempts).toBe(2);
+    expect(retries).toEqual([2]);
+    expect(chatCalls).toBe(2);
+  });
+
+  test('non-idempotent transient tool failures do not retry', async () => {
+    let attempts = 0;
+    await expect(executeToolWithTimeoutAndRetry({
+      toolName: 'write_side_effect',
+      timeoutMs: 1000,
+      maxAttempts: 3,
+      idempotent: false,
+      execute: async () => {
+        attempts += 1;
+        throw new Error('Cannot connect to database: CONNECT_TIMEOUT');
+      },
+    })).rejects.toThrow('Cannot connect');
+    expect(attempts).toBe(1);
+  });
+
+  test('non-retryable errors surface immediately even for idempotent tools', async () => {
+    let attempts = 0;
+    await expect(executeToolWithTimeoutAndRetry({
+      toolName: 'brain_search',
+      timeoutMs: 1000,
+      maxAttempts: 3,
+      idempotent: true,
+      execute: async () => {
+        attempts += 1;
+        throw new Error('page not found');
+      },
+    })).rejects.toThrow('page not found');
+    expect(attempts).toBe(1);
+  });
+});