From 2de65136c590307c16cb746ac54a5b0ec6c7ec22 Mon Sep 17 00:00:00 2001 From: Garry Tan Date: Tue, 21 Jul 2026 14:50:43 -0700 Subject: [PATCH] fix(subagent): bound per-tool execution with timeout + idempotent retry (takeover of #2086) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 Co-Authored-By: Claude Fable 5 --- src/core/ai/gateway.ts | 143 +++++++++++++++++++- src/core/minions/handlers/subagent-audit.ts | 4 +- src/core/minions/handlers/subagent.ts | 53 ++++++-- test/subagent-tool-timeout.test.ts | 138 +++++++++++++++++++ 4 files changed, 327 insertions(+), 11 deletions(-) create mode 100644 test/subagent-tool-timeout.test.ts diff --git a/src/core/ai/gateway.ts b/src/core/ai/gateway.ts index 8a2674484..429c50112 100644 --- a/src/core/ai/gateway.ts +++ b/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 { 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 { ); } - // 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 { 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; + /** 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 { + 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 { + 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((_, 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 { + 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 diff --git a/src/core/minions/handlers/subagent-audit.ts b/src/core/minions/handlers/subagent-audit.ts index f10a3feb0..bf1f1dc64 100644 --- a/src/core/minions/handlers/subagent-audit.ts +++ b/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. */ diff --git a/src/core/minions/handlers/subagent.ts b/src/core/minions/handlers/subagent.ts index 3852469ed..cf4ccfbfe 100644 --- a/src/core/minions/handlers/subagent.ts +++ b/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 { diff --git a/test/subagent-tool-timeout.test.ts b/test/subagent-tool-timeout.test.ts new file mode 100644 index 000000000..f004cd23c --- /dev/null +++ b/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 => { + 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); + }); +});