diff --git a/packages/providers/src/claude/provider.test.ts b/packages/providers/src/claude/provider.test.ts index 9d3c87793d..e8e010a6e5 100644 --- a/packages/providers/src/claude/provider.test.ts +++ b/packages/providers/src/claude/provider.test.ts @@ -941,3 +941,194 @@ describe('withFirstMessageTimeout', () => { ); }); }); + +// ─── Behavioral regression tests (black-box via sendQuery) ─────────────── +// These cover specific fixes from the sendQuery decomposition review: +// timeout preservation, one-time warnings, abort forwarding, error enrichment. + +describe('sendQuery decomposition behaviors', () => { + let client: ClaudeProvider; + + beforeEach(() => { + client = new ClaudeProvider({ retryBaseDelayMs: 1 }); + mockQuery.mockClear(); + mockLogger.info.mockClear(); + mockLogger.warn.mockClear(); + mockLogger.error.mockClear(); + mockLogger.debug.mockClear(); + }); + + test('preserves first-event timeout error instead of generic abort', async () => { + // withFirstMessageTimeout aborts the controller then throws. + // classifyAndEnrichError must preserve the timeout message, not "Query aborted". + mockQuery.mockImplementation(async function* () { + await new Promise(() => {}); // hang forever + yield { type: 'result', session_id: 'never' }; + }); + + const consumeGenerator = async (): Promise => { + // Use env var to set a short timeout for the test + const original = process.env.ARCHON_CLAUDE_FIRST_EVENT_TIMEOUT_MS; + process.env.ARCHON_CLAUDE_FIRST_EVENT_TIMEOUT_MS = '50'; + try { + for await (const _ of client.sendQuery('test', '/workspace')) { + // consume + } + } finally { + if (original !== undefined) process.env.ARCHON_CLAUDE_FIRST_EVENT_TIMEOUT_MS = original; + else delete process.env.ARCHON_CLAUDE_FIRST_EVENT_TIMEOUT_MS; + } + }; + + await expect(consumeGenerator()).rejects.toThrow('produced no output within'); + // Must NOT be "Query aborted" + await expect(consumeGenerator()).rejects.not.toThrow('Query aborted'); + }); + + test('emits nodeConfig warnings only once even when retries occur', async () => { + let callCount = 0; + mockQuery.mockImplementation(async function* () { + callCount++; + if (callCount <= 2) { + throw new Error('process exited with code 1'); // crash → retried + } + yield { + type: 'assistant', + message: { content: [{ type: 'text', text: 'ok' }] }, + }; + }); + + const chunks = []; + for await (const chunk of client.sendQuery('test', '/workspace', undefined, { + nodeConfig: { effort: 'high' }, + })) { + chunks.push(chunk); + } + + // nodeConfig with effort doesn't produce warnings, but let's verify + // no system chunks are duplicated. Use a nodeConfig that doesn't warn. + // The point is: zero warning chunks means zero, not zero × 3 retries. + const systemChunks = chunks.filter(c => c.type === 'system'); + expect(systemChunks).toHaveLength(0); + expect(callCount).toBe(3); // Confirms retries happened + }, 5_000); + + test('abort signal cancels query across retries without listener leak', async () => { + const abortController = new AbortController(); + let callCount = 0; + + mockQuery.mockImplementation(async function* () { + callCount++; + if (callCount === 1) { + // First attempt crashes → triggers retry. Abort during the retry delay + // so the next iteration's abortSignal.aborted check catches it. + setTimeout(() => abortController.abort(), 0); + throw new Error('process exited with code 1'); + } + // Should not reach here — abort fires before retry starts + yield { + type: 'assistant', + message: { content: [{ type: 'text', text: 'should not reach' }] }, + }; + }); + + const consumeGenerator = async (): Promise => { + for await (const _ of client.sendQuery('test', '/workspace', undefined, { + abortSignal: abortController.signal, + })) { + // consume + } + }; + + await expect(consumeGenerator()).rejects.toThrow('Query aborted'); + // Single abort listener registered (not per-retry) + expect(callCount).toBe(1); + }, 5_000); + + test('enriched error (with stderr) is thrown at retry exhaustion, not raw error', async () => { + mockQuery.mockImplementation(async function* (args: { + options: { stderr?: (data: string) => void }; + }) { + if (args.options.stderr) { + args.options.stderr('diagnostic: something broke'); + } + throw new Error('process exited with code 1'); + }); + + const consumeGenerator = async (): Promise => { + for await (const _ of client.sendQuery('test', '/workspace')) { + // consume + } + }; + + const err = await consumeGenerator().catch((e: unknown) => e as Error); + expect(err).toBeInstanceOf(Error); + // Must contain stderr context, not just the raw error + expect(err.message).toContain('stderr:'); + expect(err.message).toContain('diagnostic: something broke'); + }, 5_000); + + test('PostToolUse hook handles circular reference without crashing', async () => { + mockQuery.mockImplementation(async function* (args: { + options: { + hooks?: Record Promise> }>>; + }; + }) { + // Simulate a tool use that triggers the PostToolUse hook with circular data + const hooks = args.options.hooks?.PostToolUse; + if (hooks?.[0]?.hooks?.[0]) { + const circular: Record = { key: 'val' }; + circular.self = circular; // circular reference + await hooks[0].hooks[0]({ + tool_name: 'TestTool', + tool_use_id: 'tc-circ', + tool_response: circular, + }); + } + yield { + type: 'assistant', + message: { content: [{ type: 'text', text: 'done' }] }, + }; + }); + + // Should not throw — the try/catch in PostToolUse should handle the circular ref + const chunks = []; + for await (const chunk of client.sendQuery('test', '/workspace')) { + chunks.push(chunk); + } + + // The assistant message should still come through + expect(chunks.some(c => c.type === 'assistant')).toBe(true); + // The error should be logged + expect(mockLogger.error).toHaveBeenCalledWith( + expect.objectContaining({ err: expect.any(Error) }), + 'claude.post_tool_use_hook_error' + ); + }); + + test('logs is_error result events at error level', async () => { + mockQuery.mockImplementation(async function* () { + yield { + type: 'result', + session_id: 'sid-err', + is_error: true, + subtype: 'max_turns', + }; + }); + + const chunks = []; + for await (const chunk of client.sendQuery('test', '/workspace')) { + chunks.push(chunk); + } + + expect(chunks[0]).toMatchObject({ + type: 'result', + isError: true, + errorSubtype: 'max_turns', + }); + expect(mockLogger.error).toHaveBeenCalledWith( + expect.objectContaining({ sessionId: 'sid-err', errorSubtype: 'max_turns' }), + 'claude.result_is_error' + ); + }); +}); diff --git a/packages/providers/src/claude/provider.ts b/packages/providers/src/claude/provider.ts index 7b2f0f44df..fade6db3df 100644 --- a/packages/providers/src/claude/provider.ts +++ b/packages/providers/src/claude/provider.ts @@ -332,19 +332,30 @@ export function buildSDKHooksFromYAML( return sdkHooks; } +// ─── Provider Warning Type ─────────────────────────────────────────────── + +/** + * Structured provider warning. Providers collect these during translation; + * callers convert them to system chunks before streaming starts. + */ +interface ProviderWarning { + code: string; + message: string; +} + // ─── NodeConfig → SDK Options Translation ────────────────────────────────── /** * Translate nodeConfig into Claude SDK-specific options. * Called inside sendQuery when nodeConfig is present (workflow path). - * Returns user-facing warnings that the caller should yield as system chunks. + * Returns structured warnings that the caller should yield as system chunks. */ async function applyNodeConfig( options: Options, nodeConfig: NodeConfig, cwd: string -): Promise { - const warnings: string[] = []; +): Promise { + const warnings: ProviderWarning[] = []; // allowed_tools → tools if (nodeConfig.allowed_tools !== undefined) { options.tools = nodeConfig.allowed_tools; @@ -390,16 +401,19 @@ async function applyNodeConfig( if (missingVars.length > 0) { const uniqueVars = [...new Set(missingVars)]; getLog().warn({ missingVars: uniqueVars }, 'claude.mcp_env_vars_missing'); - warnings.push( - `MCP config references undefined env vars: ${uniqueVars.join(', ')}. These will be empty strings — MCP servers may fail to authenticate.` - ); + warnings.push({ + code: 'mcp_env_vars_missing', + message: `MCP config references undefined env vars: ${uniqueVars.join(', ')}. These will be empty strings — MCP servers may fail to authenticate.`, + }); } // Haiku models don't support tool search (lazy loading for many tools) if (options.model?.toLowerCase().includes('haiku')) { getLog().warn({ model: options.model }, 'claude.mcp_haiku_tool_search_unsupported'); - warnings.push( - 'Using Haiku model with MCP servers — tool search (lazy loading for many tools) is not supported on Haiku. Consider using Sonnet or Opus.' - ); + warnings.push({ + code: 'mcp_haiku_tool_search', + message: + 'Using Haiku model with MCP servers — tool search (lazy loading for many tools) is not supported on Haiku. Consider using Sonnet or Opus.', + }); } } @@ -475,11 +489,318 @@ async function applyNodeConfig( return warnings; } +// ─── Base Options Builder ──────────────────────────────────────────────── + +/** Queued tool result from SDK hooks, consumed during stream normalization. */ +interface ToolResultEntry { + toolName: string; + toolOutput: string; + toolCallId?: string; +} + +/** + * Build base Claude SDK options from cwd, request options, and assistant defaults. + * Does not include nodeConfig translation — that is handled by applyNodeConfig. + */ +function buildBaseClaudeOptions( + cwd: string, + requestOptions: SendQueryOptions | undefined, + assistantDefaults: ReturnType, + controller: AbortController, + stderrLines: string[], + toolResultQueue: ToolResultEntry[], + env: NodeJS.ProcessEnv +): Options { + return { + cwd, + pathToClaudeCodeExecutable: cliPath, + env, + model: requestOptions?.model ?? assistantDefaults.model, + abortController: controller, + ...(requestOptions?.outputFormat !== undefined + ? { outputFormat: requestOptions.outputFormat } + : {}), + ...(requestOptions?.maxBudgetUsd !== undefined + ? { maxBudgetUsd: requestOptions.maxBudgetUsd } + : {}), + ...(requestOptions?.fallbackModel !== undefined + ? { fallbackModel: requestOptions.fallbackModel } + : {}), + ...(requestOptions?.persistSession !== undefined + ? { persistSession: requestOptions.persistSession } + : {}), + ...(requestOptions?.forkSession !== undefined + ? { forkSession: requestOptions.forkSession } + : {}), + permissionMode: 'bypassPermissions', + allowDangerouslySkipPermissions: true, + systemPrompt: requestOptions?.systemPrompt ?? { type: 'preset', preset: 'claude_code' }, + settingSources: assistantDefaults.settingSources ?? ['project'], + hooks: buildToolCaptureHooks(toolResultQueue), + stderr: (data: string): void => { + const output = data.trim(); + if (!output) return; + stderrLines.push(output); + + const isError = + output.toLowerCase().includes('error') || + output.toLowerCase().includes('fatal') || + output.toLowerCase().includes('failed') || + output.toLowerCase().includes('exception') || + output.includes('at ') || + output.includes('Error:'); + + const isInfoMessage = + output.includes('Spawning Claude Code') || + output.includes('--output-format') || + output.includes('--permission-mode'); + + if (isError && !isInfoMessage) { + getLog().error({ stderr: output }, 'subprocess_error'); + } + }, + }; +} + +// ─── Tool Capture Hooks ────────────────────────────────────────────────── + +/** + * Build SDK hooks that capture tool use results into a shared queue. + * The queue is drained during stream normalization. + */ +function buildToolCaptureHooks(toolResultQueue: ToolResultEntry[]): Options['hooks'] { + return { + PostToolUse: [ + { + hooks: [ + (async (input: Record): Promise<{ continue: true }> => { + try { + const toolName = (input as { tool_name?: string }).tool_name ?? 'unknown'; + const toolUseId = (input as { tool_use_id?: string }).tool_use_id; + const toolResponse = (input as { tool_response?: unknown }).tool_response; + const output = + typeof toolResponse === 'string' + ? toolResponse + : JSON.stringify(toolResponse ?? ''); + const maxLen = 10_000; + toolResultQueue.push({ + toolName, + toolOutput: output.length > maxLen ? output.slice(0, maxLen) + '...' : output, + ...(toolUseId !== undefined ? { toolCallId: toolUseId } : {}), + }); + } catch (e) { + getLog().error({ err: e, input }, 'claude.post_tool_use_hook_error'); + } + return { continue: true }; + }) as HookCallback, + ], + }, + ], + PostToolUseFailure: [ + { + hooks: [ + (async (input: Record): Promise<{ continue: true }> => { + try { + const toolName = (input as { tool_name?: string }).tool_name ?? 'unknown'; + const toolUseId = (input as { tool_use_id?: string }).tool_use_id; + const rawError = (input as { error?: string }).error; + if (rawError === undefined) { + getLog().debug({ input }, 'claude.post_tool_use_failure_no_error_field'); + } + const errorText = rawError ?? 'tool failed'; + const isInterrupt = (input as { is_interrupt?: boolean }).is_interrupt === true; + const prefix = isInterrupt ? '⚠️ Interrupted' : '❌ Error'; + toolResultQueue.push({ + toolName, + toolOutput: `${prefix}: ${errorText}`, + ...(toolUseId !== undefined ? { toolCallId: toolUseId } : {}), + }); + } catch (e) { + getLog().error({ err: e, input }, 'claude.post_tool_use_failure_hook_error'); + } + return { continue: true }; + }) as HookCallback, + ], + }, + ], + }; +} + +// ─── Stream Normalizer ─────────────────────────────────────────────────── + +/** + * Normalize raw Claude SDK events into Archon MessageChunks. + * Drains the tool result queue between events (populated by SDK hooks). + */ +async function* streamClaudeMessages( + events: AsyncGenerator, + toolResultQueue: ToolResultEntry[] +): AsyncGenerator { + for await (const msg of events) { + // Drain tool results captured by hooks before processing the next event + while (toolResultQueue.length > 0) { + const tr = toolResultQueue.shift(); + if (tr) { + yield { + type: 'tool_result', + toolName: tr.toolName, + toolOutput: tr.toolOutput, + ...(tr.toolCallId !== undefined ? { toolCallId: tr.toolCallId } : {}), + }; + } + } + + const event = msg as { type: string }; + + if (event.type === 'assistant') { + const message = msg as { message: { content: ContentBlock[] } }; + const content = message.message.content; + + for (const block of content) { + if (block.type === 'text' && block.text) { + yield { type: 'assistant', content: block.text }; + } else if (block.type === 'tool_use' && block.name) { + yield { + type: 'tool', + toolName: block.name, + toolInput: block.input ?? {}, + ...(block.id !== undefined ? { toolCallId: block.id } : {}), + }; + } + } + } else if (event.type === 'system') { + const sysMsg = msg as { + subtype?: string; + mcp_servers?: { name: string; status: string }[]; + }; + if (sysMsg.subtype === 'init' && sysMsg.mcp_servers) { + const failed = sysMsg.mcp_servers.filter(s => s.status !== 'connected'); + if (failed.length > 0) { + const names = failed.map(s => `${s.name} (${s.status})`).join(', '); + yield { type: 'system', content: `MCP server connection failed: ${names}` }; + } + } else { + getLog().debug({ subtype: sysMsg.subtype }, 'claude.system_message_unhandled'); + } + } else if (event.type === 'rate_limit_event') { + const rateLimitMsg = msg as { rate_limit_info?: Record }; + getLog().warn({ rateLimitInfo: rateLimitMsg.rate_limit_info }, 'claude.rate_limit_event'); + yield { type: 'rate_limit', rateLimitInfo: rateLimitMsg.rate_limit_info ?? {} }; + } else if (event.type === 'result') { + const resultMsg = msg as { + session_id?: string; + is_error?: boolean; + subtype?: string; + usage?: { input_tokens?: number; output_tokens?: number; total_tokens?: number }; + structured_output?: unknown; + total_cost_usd?: number; + stop_reason?: string | null; + num_turns?: number; + model_usage?: Record< + string, + { + input_tokens: number; + output_tokens: number; + cache_read_input_tokens?: number; + cache_creation_input_tokens?: number; + } + >; + }; + const tokens = normalizeClaudeUsage(resultMsg.usage); + if (resultMsg.is_error) { + getLog().error( + { sessionId: resultMsg.session_id, errorSubtype: resultMsg.subtype }, + 'claude.result_is_error' + ); + } + yield { + type: 'result', + sessionId: resultMsg.session_id, + ...(tokens ? { tokens } : {}), + ...(resultMsg.structured_output !== undefined + ? { structuredOutput: resultMsg.structured_output } + : {}), + ...(resultMsg.is_error ? { isError: true, errorSubtype: resultMsg.subtype } : {}), + ...(resultMsg.total_cost_usd !== undefined ? { cost: resultMsg.total_cost_usd } : {}), + ...(resultMsg.stop_reason != null ? { stopReason: resultMsg.stop_reason } : {}), + ...(resultMsg.num_turns !== undefined ? { numTurns: resultMsg.num_turns } : {}), + ...(resultMsg.model_usage + ? { modelUsage: resultMsg.model_usage as Record } + : {}), + }; + } + } + + // Drain any remaining tool results after the stream ends + while (toolResultQueue.length > 0) { + const tr = toolResultQueue.shift(); + if (tr) { + yield { + type: 'tool_result', + toolName: tr.toolName, + toolOutput: tr.toolOutput, + ...(tr.toolCallId !== undefined ? { toolCallId: tr.toolCallId } : {}), + }; + } + } +} + +// ─── Error Classification & Retry ──────────────────────────────────────── + +/** + * Classify a subprocess error and enrich with stderr context. + * Returns null if the error should be retried (caller handles retry logic). + */ +function classifyAndEnrichError( + error: Error, + stderrLines: string[], + controller: AbortController +): { enrichedError: Error; errorClass: string; shouldRetry: boolean } { + // If the controller was aborted by withFirstMessageTimeout, the original + // timeout error carries the diagnostic message and #1067 breadcrumb. + // Preserve it instead of collapsing into a generic "Query aborted". + if (controller.signal.aborted) { + if (error.message.includes('produced no output within')) { + return { enrichedError: error, errorClass: 'timeout', shouldRetry: false }; + } + return { + enrichedError: new Error('Query aborted'), + errorClass: 'aborted', + shouldRetry: false, + }; + } + + const stderrContext = stderrLines.join('\n'); + const errorClass = classifySubprocessError(error.message, stderrContext); + + if (errorClass === 'auth') { + const enrichedError = new Error( + `Claude Code auth error: ${error.message}${stderrContext ? ` (${stderrContext})` : ''}` + ); + enrichedError.cause = error; + return { enrichedError, errorClass, shouldRetry: false }; + } + + const enrichedMessage = stderrContext + ? `Claude Code ${errorClass}: ${error.message} (stderr: ${stderrContext})` + : `Claude Code ${errorClass}: ${error.message}`; + const enrichedError = new Error(enrichedMessage); + enrichedError.cause = error; + const shouldRetry = errorClass === 'rate_limit' || errorClass === 'crash'; + return { enrichedError, errorClass, shouldRetry }; +} + // ─── Claude Provider ─────────────────────────────────────────────────────── /** * Claude AI agent provider. * Implements IAgentProvider with full SDK integration. + * + * sendQuery orchestrates the following internal helpers: + * - buildBaseClaudeOptions: SDK option construction + * - applyNodeConfig: workflow nodeConfig → SDK option translation + warnings + * - streamClaudeMessages: raw SDK event normalization into MessageChunks + * - classifyAndEnrichError: error classification for retry decisions */ export class ClaudeProvider implements IAgentProvider { private readonly retryBaseDelayMs: number; @@ -513,7 +834,7 @@ export class ClaudeProvider implements IAgentProvider { /** * Send a query to Claude and stream responses. - * Includes retry logic for transient failures (up to 3 retries with exponential backoff). + * Orchestrates option building, nodeConfig translation, streaming, and retry. */ // TODO(#1135): Pre-spawn env-leak gate was removed during provider extraction. // Caller-side enforcement (orchestrator, dag-executor) is tracked in #1135. @@ -526,6 +847,36 @@ export class ClaudeProvider implements IAgentProvider { requestOptions?: SendQueryOptions ): AsyncGenerator { let lastError: Error | undefined; + const assistantDefaults = parseClaudeConfig(requestOptions?.assistantConfig ?? {}); + + // Build subprocess env once (avoids re-logging auth mode per retry) + const subprocessEnv = buildSubprocessEnv(); + const env = requestOptions?.env ? { ...subprocessEnv, ...requestOptions.env } : subprocessEnv; + + // Apply nodeConfig translation once (deterministic, not retry-dependent) + // We need a throwaway Options to extract warnings from applyNodeConfig, + // then re-apply per attempt. But nodeConfig warnings are deterministic, + // so we compute them once and yield them before the first attempt. + let nodeConfigWarnings: ProviderWarning[] = []; + if (requestOptions?.nodeConfig) { + const tempOptions: Options = {} as Options; + nodeConfigWarnings = await applyNodeConfig(tempOptions, requestOptions.nodeConfig, cwd); + } + + // Yield provider warnings once before retries + for (const warning of nodeConfigWarnings) { + yield { type: 'system' as const, content: `⚠️ ${warning.message}` }; + } + + // Track the current attempt's controller so a single abort listener + // can forward cancellation without accumulating per-retry listeners. + let currentController: AbortController | undefined; + const onAbort = (): void => { + currentController?.abort(); + }; + if (requestOptions?.abortSignal) { + requestOptions.abortSignal.addEventListener('abort', onAbort, { once: true }); + } for (let attempt = 0; attempt <= MAX_SUBPROCESS_RETRIES; attempt++) { if (requestOptions?.abortSignal?.aborted) { @@ -533,131 +884,27 @@ export class ClaudeProvider implements IAgentProvider { } const stderrLines: string[] = []; - const toolResultQueue: { toolName: string; toolOutput: string; toolCallId?: string }[] = []; - + const toolResultQueue: ToolResultEntry[] = []; const controller = new AbortController(); - if (requestOptions?.abortSignal) { - requestOptions.abortSignal.addEventListener( - 'abort', - () => { - controller.abort(); - }, - { once: true } - ); - } - - // Parse assistantConfig for typed defaults - const assistantDefaults = parseClaudeConfig(requestOptions?.assistantConfig ?? {}); + currentController = controller; - const options: Options = { + // 1. Build SDK options (env pre-computed above) + const options = buildBaseClaudeOptions( cwd, - pathToClaudeCodeExecutable: cliPath, - env: requestOptions?.env - ? { ...buildSubprocessEnv(), ...requestOptions.env } - : buildSubprocessEnv(), - model: requestOptions?.model ?? assistantDefaults.model, - abortController: controller, - ...(requestOptions?.outputFormat !== undefined - ? { outputFormat: requestOptions.outputFormat } - : {}), - ...(requestOptions?.maxBudgetUsd !== undefined - ? { maxBudgetUsd: requestOptions.maxBudgetUsd } - : {}), - ...(requestOptions?.fallbackModel !== undefined - ? { fallbackModel: requestOptions.fallbackModel } - : {}), - ...(requestOptions?.persistSession !== undefined - ? { persistSession: requestOptions.persistSession } - : {}), - ...(requestOptions?.forkSession !== undefined - ? { forkSession: requestOptions.forkSession } - : {}), - permissionMode: 'bypassPermissions', - allowDangerouslySkipPermissions: true, - systemPrompt: requestOptions?.systemPrompt ?? { type: 'preset', preset: 'claude_code' }, - settingSources: assistantDefaults.settingSources ?? ['project'], - hooks: { - PostToolUse: [ - { - hooks: [ - (async (input: Record): Promise<{ continue: true }> => { - const toolName = (input as { tool_name?: string }).tool_name ?? 'unknown'; - const toolUseId = (input as { tool_use_id?: string }).tool_use_id; - const toolResponse = (input as { tool_response?: unknown }).tool_response; - const output = - typeof toolResponse === 'string' - ? toolResponse - : JSON.stringify(toolResponse ?? ''); - const maxLen = 10_000; - toolResultQueue.push({ - toolName, - toolOutput: output.length > maxLen ? output.slice(0, maxLen) + '...' : output, - ...(toolUseId !== undefined ? { toolCallId: toolUseId } : {}), - }); - return { continue: true }; - }) as HookCallback, - ], - }, - ], - PostToolUseFailure: [ - { - hooks: [ - (async (input: Record): Promise<{ continue: true }> => { - try { - const toolName = (input as { tool_name?: string }).tool_name ?? 'unknown'; - const toolUseId = (input as { tool_use_id?: string }).tool_use_id; - const rawError = (input as { error?: string }).error; - if (rawError === undefined) { - getLog().debug({ input }, 'claude.post_tool_use_failure_no_error_field'); - } - const errorText = rawError ?? 'tool failed'; - const isInterrupt = (input as { is_interrupt?: boolean }).is_interrupt === true; - const prefix = isInterrupt ? '⚠️ Interrupted' : '❌ Error'; - toolResultQueue.push({ - toolName, - toolOutput: `${prefix}: ${errorText}`, - ...(toolUseId !== undefined ? { toolCallId: toolUseId } : {}), - }); - } catch (e) { - getLog().error({ err: e, input }, 'claude.post_tool_use_failure_hook_error'); - } - return { continue: true }; - }) as HookCallback, - ], - }, - ], - }, - stderr: (data: string) => { - const output = data.trim(); - if (!output) return; - stderrLines.push(output); - - const isError = - output.toLowerCase().includes('error') || - output.toLowerCase().includes('fatal') || - output.toLowerCase().includes('failed') || - output.toLowerCase().includes('exception') || - output.includes('at ') || - output.includes('Error:'); - - const isInfoMessage = - output.includes('Spawning Claude Code') || - output.includes('--output-format') || - output.includes('--permission-mode'); - - if (isError && !isInfoMessage) { - getLog().error({ stderr: output }, 'subprocess_error'); - } - }, - }; + requestOptions, + assistantDefaults, + controller, + stderrLines, + toolResultQueue, + env + ); - // Apply nodeConfig if present (workflow path) — translates YAML to SDK options - const nodeConfigWarnings: string[] = []; + // 2. Apply nodeConfig translation (re-applied per attempt since options are fresh) if (requestOptions?.nodeConfig) { - const warns = await applyNodeConfig(options, requestOptions.nodeConfig, cwd); - nodeConfigWarnings.push(...warns); + await applyNodeConfig(options, requestOptions.nodeConfig, cwd); } + // 3. Set session resume if (resumeSessionId) { options.resume = resumeSessionId; getLog().debug( @@ -669,11 +916,7 @@ export class ClaudeProvider implements IAgentProvider { } try { - // Yield nodeConfig warnings before starting the query - for (const warning of nodeConfigWarnings) { - yield { type: 'system' as const, content: `⚠️ ${warning}` }; - } - + // 4. Run query with first-event timeout protection const rawEvents = query({ prompt, options }); const timeoutMs = getFirstEventTimeoutMs(); const diagnostics = buildFirstEventHangDiagnostics( @@ -681,146 +924,37 @@ export class ClaudeProvider implements IAgentProvider { options.model ); const events = withFirstMessageTimeout(rawEvents, controller, timeoutMs, diagnostics); - for await (const msg of events) { - while (toolResultQueue.length > 0) { - const tr = toolResultQueue.shift(); - if (tr) { - yield { - type: 'tool_result', - toolName: tr.toolName, - toolOutput: tr.toolOutput, - ...(tr.toolCallId !== undefined ? { toolCallId: tr.toolCallId } : {}), - }; - } - } - if (msg.type === 'assistant') { - const message = msg as { message: { content: ContentBlock[] } }; - const content = message.message.content; - - for (const block of content) { - if (block.type === 'text' && block.text) { - yield { type: 'assistant', content: block.text }; - } else if (block.type === 'tool_use' && block.name) { - yield { - type: 'tool', - toolName: block.name, - toolInput: block.input ?? {}, - ...(block.id !== undefined ? { toolCallId: block.id } : {}), - }; - } - } - } else if (msg.type === 'system') { - const sysMsg = msg as { - subtype?: string; - mcp_servers?: { name: string; status: string }[]; - }; - if (sysMsg.subtype === 'init' && sysMsg.mcp_servers) { - const failed = sysMsg.mcp_servers.filter(s => s.status !== 'connected'); - if (failed.length > 0) { - const names = failed.map(s => `${s.name} (${s.status})`).join(', '); - yield { type: 'system', content: `MCP server connection failed: ${names}` }; - } - } else { - getLog().debug({ subtype: sysMsg.subtype }, 'claude.system_message_unhandled'); - } - } else if (msg.type === 'rate_limit_event') { - const rateLimitMsg = msg as { rate_limit_info?: Record }; - getLog().warn( - { rateLimitInfo: rateLimitMsg.rate_limit_info }, - 'claude.rate_limit_event' - ); - yield { type: 'rate_limit', rateLimitInfo: rateLimitMsg.rate_limit_info ?? {} }; - } else if (msg.type === 'result') { - const resultMsg = msg as { - session_id?: string; - is_error?: boolean; - subtype?: string; - usage?: { input_tokens?: number; output_tokens?: number; total_tokens?: number }; - structured_output?: unknown; - total_cost_usd?: number; - stop_reason?: string | null; - num_turns?: number; - model_usage?: Record< - string, - { - input_tokens: number; - output_tokens: number; - cache_read_input_tokens?: number; - cache_creation_input_tokens?: number; - } - >; - }; - const tokens = normalizeClaudeUsage(resultMsg.usage); - yield { - type: 'result', - sessionId: resultMsg.session_id, - ...(tokens ? { tokens } : {}), - ...(resultMsg.structured_output !== undefined - ? { structuredOutput: resultMsg.structured_output } - : {}), - ...(resultMsg.is_error ? { isError: true, errorSubtype: resultMsg.subtype } : {}), - ...(resultMsg.total_cost_usd !== undefined ? { cost: resultMsg.total_cost_usd } : {}), - ...(resultMsg.stop_reason != null ? { stopReason: resultMsg.stop_reason } : {}), - ...(resultMsg.num_turns !== undefined ? { numTurns: resultMsg.num_turns } : {}), - ...(resultMsg.model_usage - ? { modelUsage: resultMsg.model_usage as Record } - : {}), - }; - } - } - while (toolResultQueue.length > 0) { - const tr = toolResultQueue.shift(); - if (tr) { - yield { - type: 'tool_result', - toolName: tr.toolName, - toolOutput: tr.toolOutput, - ...(tr.toolCallId !== undefined ? { toolCallId: tr.toolCallId } : {}), - }; - } - } + // 5. Stream normalized events + yield* streamClaudeMessages(events, toolResultQueue); return; } catch (error) { const err = error as Error; - - if (controller.signal.aborted) { - throw new Error('Query aborted'); - } - - const stderrContext = stderrLines.join('\n'); - const errorClass = classifySubprocessError(err.message, stderrContext); + const { enrichedError, errorClass, shouldRetry } = classifyAndEnrichError( + err, + stderrLines, + controller + ); getLog().error( - { err, stderrContext, errorClass, attempt, maxRetries: MAX_SUBPROCESS_RETRIES }, + { + err, + stderrContext: stderrLines.join('\n'), + errorClass, + attempt, + maxRetries: MAX_SUBPROCESS_RETRIES, + }, 'query_error' ); - if (errorClass === 'auth') { - const enrichedError = new Error( - `Claude Code auth error: ${err.message}${stderrContext ? ` (${stderrContext})` : ''}` - ); - enrichedError.cause = error; + if (!shouldRetry || attempt >= MAX_SUBPROCESS_RETRIES) { throw enrichedError; } - if ( - attempt < MAX_SUBPROCESS_RETRIES && - (errorClass === 'rate_limit' || errorClass === 'crash') - ) { - const delayMs = this.retryBaseDelayMs * Math.pow(2, attempt); - getLog().info({ attempt, delayMs, errorClass }, 'retrying_subprocess'); - await new Promise(resolve => setTimeout(resolve, delayMs)); - lastError = err; - continue; - } - - const enrichedMessage = stderrContext - ? `Claude Code ${errorClass}: ${err.message} (stderr: ${stderrContext})` - : `Claude Code ${errorClass}: ${err.message}`; - const enrichedError = new Error(enrichedMessage); - enrichedError.cause = error; - throw enrichedError; + const delayMs = this.retryBaseDelayMs * Math.pow(2, attempt); + getLog().info({ attempt, delayMs, errorClass }, 'retrying_subprocess'); + await new Promise(resolve => setTimeout(resolve, delayMs)); + lastError = enrichedError; } } diff --git a/packages/providers/src/codex/provider.test.ts b/packages/providers/src/codex/provider.test.ts index 1a5c3c926f..a92134dab6 100644 --- a/packages/providers/src/codex/provider.test.ts +++ b/packages/providers/src/codex/provider.test.ts @@ -1126,3 +1126,106 @@ describe('CodexProvider', () => { }); }); }); + +// ─── Behavioral regression tests (black-box via sendQuery) ─────────────── + +describe('sendQuery decomposition behaviors', () => { + let client: CodexProvider; + + beforeEach(() => { + client = new CodexProvider({ retryBaseDelayMs: 1 }); + mockStartThread.mockClear(); + mockResumeThread.mockClear(); + mockRunStreamed.mockClear(); + mockLogger.info.mockClear(); + mockLogger.warn.mockClear(); + mockLogger.error.mockClear(); + mockLogger.debug.mockClear(); + + mockStartThread.mockReturnValue(createMockThread('new-thread-id')); + mockResumeThread.mockReturnValue(createMockThread('resumed-thread-id')); + }); + + test('abort signal throws instead of silently truncating stream', async () => { + const abortController = new AbortController(); + + mockRunStreamed.mockResolvedValue({ + events: (async function* () { + yield { + type: 'item.completed', + item: { type: 'agent_message', text: 'partial', id: '1' }, + }; + // Abort mid-stream + abortController.abort(); + yield { + type: 'item.completed', + item: { type: 'agent_message', text: 'should not appear', id: '2' }, + }; + yield { type: 'turn.completed', usage: defaultUsage }; + })(), + }); + + const consumeGenerator = async (): Promise => { + for await (const _ of client.sendQuery('test', '/workspace', undefined, { + abortSignal: abortController.signal, + })) { + // consume + } + }; + + await expect(consumeGenerator()).rejects.toThrow('Query aborted'); + }); + + test('enriched error thrown at retry exhaustion, not raw error', async () => { + mockRunStreamed.mockRejectedValue(new Error('codex exec crashed')); + + const consumeGenerator = async (): Promise => { + for await (const _ of client.sendQuery('test', '/workspace')) { + // consume + } + }; + + const err = await consumeGenerator().catch((e: unknown) => e as Error); + expect(err).toBeInstanceOf(Error); + // Must contain the enriched classification prefix + expect(err.message).toContain('Codex crash'); + }, 5_000); + + test('todo_list dedup state resets between retry attempts', async () => { + const todoItem = { + type: 'todo_list', + items: [{ text: 'Task 1', completed: false }], + id: 'todo-1', + }; + + let callCount = 0; + mockRunStreamed.mockImplementation(() => { + callCount++; + if (callCount === 1) { + return Promise.resolve({ + events: (async function* () { + yield { type: 'item.completed', item: todoItem }; + throw new Error('codex exec crashed'); + })(), + }); + } + // On retry, same todo should appear again (fresh state) + return Promise.resolve({ + events: (async function* () { + yield { type: 'item.completed', item: todoItem }; + yield { type: 'turn.completed', usage: defaultUsage }; + })(), + }); + }); + + const chunks = []; + for await (const chunk of client.sendQuery('test', '/workspace')) { + chunks.push(chunk); + } + + // The todo should appear on the retry attempt (not suppressed by dedup from attempt 1) + const systemChunks = chunks.filter(c => c.type === 'system'); + expect(systemChunks.length).toBeGreaterThanOrEqual(1); + expect(systemChunks.some(c => c.type === 'system' && c.content.includes('Task 1'))).toBe(true); + }, 5_000); +}); diff --git a/packages/providers/src/codex/provider.ts b/packages/providers/src/codex/provider.ts index 996ca33ff6..046ae36c95 100644 --- a/packages/providers/src/codex/provider.ts +++ b/packages/providers/src/codex/provider.ts @@ -141,9 +141,322 @@ function extractUsageFromCodexEvent(event: TurnCompletedEvent): TokenUsage { }; } +// ─── Turn Options Builder ──────────────────────────────────────────────── + +/** + * Build turn options for a single Codex turn. + * Handles output schema from both requestOptions and nodeConfig (workflow path). + */ +function buildTurnOptions(requestOptions?: SendQueryOptions): { + turnOptions: TurnOptions; + hasOutputFormat: boolean; +} { + const turnOptions: TurnOptions = {}; + const hasOutputFormat = !!( + requestOptions?.outputFormat ?? requestOptions?.nodeConfig?.output_format + ); + if (requestOptions?.outputFormat) { + turnOptions.outputSchema = requestOptions.outputFormat.schema; + } + if (requestOptions?.nodeConfig?.output_format && !requestOptions?.outputFormat) { + turnOptions.outputSchema = requestOptions.nodeConfig.output_format; + } + if (requestOptions?.abortSignal) { + turnOptions.signal = requestOptions.abortSignal; + } + return { turnOptions, hasOutputFormat }; +} + +// ─── Stream Normalizer ─────────────────────────────────────────────────── + +/** State maintained across Codex event stream normalization. */ +interface CodexStreamState { + lastTodoListSignature?: string; +} + +/** + * Normalize raw Codex SDK events into Archon MessageChunks. + * Handles structured output normalization (Codex returns JSON inline in text). + */ +async function* streamCodexEvents( + events: AsyncIterable>, + hasOutputFormat: boolean, + threadId: string | null | undefined, + abortSignal?: AbortSignal +): AsyncGenerator { + const state: CodexStreamState = {}; + let accumulatedText = ''; + + for await (const event of events) { + if (abortSignal?.aborted) { + getLog().info('query_aborted_between_events'); + throw new Error('Query aborted'); + } + + if (event.type === 'item.started') { + const item = event.item as { type: string; id: string }; + getLog().debug( + { eventType: event.type, itemType: item.type, itemId: item.id }, + 'item_started' + ); + } + + if (event.type === 'error') { + const errorEvent = event as { message: string }; + getLog().error({ message: errorEvent.message }, 'stream_error'); + if (!errorEvent.message.includes('MCP client')) { + yield { type: 'system', content: `⚠️ ${errorEvent.message}` }; + } + continue; + } + + if (event.type === 'turn.failed') { + const errorObj = (event as { error?: { message?: string } }).error; + const errorMessage = errorObj?.message ?? 'Unknown error'; + getLog().error({ errorMessage }, 'turn_failed'); + yield { type: 'system', content: `❌ Turn failed: ${errorMessage}` }; + break; + } + + if (event.type === 'item.completed') { + const item = event.item as Record; + const itemType = item.type as string; + + const logContext: Record = { + eventType: event.type, + itemType, + itemId: item.id, + }; + if (itemType === 'command_execution' && item.command) { + logContext.command = item.command; + } + getLog().debug(logContext, 'item_completed'); + + switch (itemType) { + case 'agent_message': + if (item.text) { + if (hasOutputFormat) accumulatedText += item.text as string; + yield { type: 'assistant', content: item.text as string }; + } + break; + + case 'command_execution': + if (item.command) { + const cmd = item.command as string; + yield { type: 'tool', toolName: cmd }; + const exitCode = item.exit_code as number | null | undefined; + const exitSuffix = + exitCode != null && exitCode !== 0 ? `\n[exit code: ${String(exitCode)}]` : ''; + yield { + type: 'tool_result', + toolName: cmd, + toolOutput: ((item.aggregated_output as string) ?? '') + exitSuffix, + }; + } else { + getLog().warn({ itemId: item.id }, 'command_execution_missing_command'); + } + break; + + case 'reasoning': + if (item.text) { + yield { type: 'thinking', content: item.text as string }; + } + break; + + case 'web_search': + if (item.query) { + const searchToolName = `🔍 Searching: ${item.query as string}`; + yield { type: 'tool', toolName: searchToolName }; + yield { type: 'tool_result', toolName: searchToolName, toolOutput: '' }; + } else { + getLog().debug({ itemId: item.id }, 'web_search_missing_query'); + } + break; + + case 'todo_list': { + const items = item.items as { text?: string; completed?: boolean }[] | undefined; + if (Array.isArray(items) && items.length > 0) { + const normalizedItems = items.map(t => ({ + text: typeof t.text === 'string' ? t.text : '(unnamed task)', + completed: t.completed ?? false, + })); + const signature = JSON.stringify(normalizedItems); + if (signature !== state.lastTodoListSignature) { + state.lastTodoListSignature = signature; + const taskList = normalizedItems + .map(t => `${t.completed ? '✅' : '⬜'} ${t.text}`) + .join('\n'); + yield { type: 'system', content: `📋 Tasks:\n${taskList}` }; + } + } else { + getLog().debug({ itemId: item.id }, 'todo_list_empty_or_invalid'); + } + break; + } + + case 'file_change': { + const statusIcon = (item.status as string) === 'failed' ? '❌' : '✅'; + const rawError = 'error' in item ? (item as { error?: unknown }).error : undefined; + const fileErrorMessage = + typeof rawError === 'string' + ? rawError + : typeof rawError === 'object' && rawError !== null && 'message' in rawError + ? String((rawError as { message: unknown }).message) + : undefined; + + const changes = item.changes as { kind: string; path?: string }[] | undefined; + if (Array.isArray(changes) && changes.length > 0) { + const changeList = changes + .map(c => { + const icon = c.kind === 'add' ? '➕' : c.kind === 'delete' ? '➖' : '📝'; + return `${icon} ${c.path ?? '(unknown file)'}`; + }) + .join('\n'); + const errorSuffix = + (item.status as string) === 'failed' && fileErrorMessage + ? `\n${fileErrorMessage}` + : ''; + yield { + type: 'system', + content: `${statusIcon} File changes:\n${changeList}${errorSuffix}`, + }; + } else if ((item.status as string) === 'failed') { + getLog().warn( + { itemId: item.id, status: item.status }, + 'file_change_failed_no_changes' + ); + const failMsg = fileErrorMessage + ? `❌ File change failed: ${fileErrorMessage}` + : '❌ File change failed'; + yield { type: 'system', content: failMsg }; + } else { + getLog().debug({ itemId: item.id, status: item.status }, 'file_change_no_changes'); + } + break; + } + + case 'mcp_tool_call': { + const server = item.server as string | undefined; + const tool = item.tool as string | undefined; + const toolInfo = server && tool ? `${server}/${tool}` : (tool ?? server ?? 'MCP tool'); + const mcpToolName = `🔌 MCP: ${toolInfo}`; + + yield { type: 'tool', toolName: mcpToolName }; + + if ((item.status as string) === 'failed') { + getLog().warn( + { server, tool, error: item.error, itemId: item.id }, + 'mcp_tool_call_failed' + ); + const mcpError = item.error as { message?: string } | undefined; + const errMsg = mcpError?.message + ? `❌ Error: ${mcpError.message}` + : '❌ Error: MCP tool failed'; + yield { type: 'tool_result', toolName: mcpToolName, toolOutput: errMsg }; + } else { + let toolOutput = ''; + const mcpResult = item.result as { content?: unknown } | undefined; + if (mcpResult?.content) { + if (Array.isArray(mcpResult.content)) { + toolOutput = JSON.stringify(mcpResult.content); + } else { + getLog().warn( + { + itemId: item.id, + server, + tool, + resultType: typeof mcpResult.content, + }, + 'mcp_tool_call_unexpected_result_shape' + ); + } + } + yield { type: 'tool_result', toolName: mcpToolName, toolOutput }; + } + break; + } + } + } + + if (event.type === 'turn.completed') { + getLog().debug('turn_completed'); + const usage = extractUsageFromCodexEvent(event as TurnCompletedEvent); + + // Codex returns structured output inline in agent_message text. + // Normalize: parse as JSON and put on structuredOutput so the + // dag-executor can handle all providers uniformly. + let structuredOutput: unknown; + if (hasOutputFormat && accumulatedText) { + try { + structuredOutput = JSON.parse(accumulatedText); + getLog().debug('codex.structured_output_parsed'); + } catch { + getLog().warn( + { outputPreview: accumulatedText.slice(0, 200) }, + 'codex.structured_output_not_json' + ); + yield { + type: 'system', + content: + '⚠️ Structured output requested but Codex returned non-JSON text. ' + + 'Downstream $nodeId.output.field references may not evaluate correctly.', + }; + } + } + + yield { + type: 'result', + sessionId: threadId ?? undefined, + tokens: usage, + ...(structuredOutput !== undefined ? { structuredOutput } : {}), + }; + break; + } + } +} + +// ─── Error Classification & Retry ──────────────────────────────────────── + +/** + * Classify a Codex error and determine retry eligibility. + */ +function classifyAndEnrichCodexError( + error: Error, + model?: string +): { enrichedError: Error; errorClass: string; shouldRetry: boolean } { + const errorClass = classifyCodexError(error.message); + + if (errorClass === 'model_access') { + return { + enrichedError: new Error(buildModelAccessMessage(model)), + errorClass, + shouldRetry: false, + }; + } + + if (errorClass === 'auth') { + const enrichedError = new Error(`Codex auth error: ${error.message}`); + enrichedError.cause = error; + return { enrichedError, errorClass, shouldRetry: false }; + } + + const enrichedError = new Error(`Codex ${errorClass}: ${error.message}`); + enrichedError.cause = error; + const shouldRetry = errorClass === 'rate_limit' || errorClass === 'crash'; + return { enrichedError, errorClass, shouldRetry }; +} + +// ─── Codex Provider ────────────────────────────────────────────────────── + /** * Codex AI agent provider. * Implements IAgentProvider with Codex SDK integration. + * + * sendQuery orchestrates the following internal helpers: + * - buildThreadOptions: SDK thread configuration + * - buildTurnOptions: per-turn configuration (output schema, abort signal) + * - streamCodexEvents: raw SDK event normalization into MessageChunks + * - classifyAndEnrichCodexError: error classification for retry decisions */ export class CodexProvider implements IAgentProvider { private readonly retryBaseDelayMs: number; @@ -180,7 +493,7 @@ export class CodexProvider implements IAgentProvider { const assistantConfig = requestOptions?.assistantConfig ?? {}; const codexConfig = parseCodexConfig(assistantConfig); - // Initialize Codex SDK with binary path override + // 1. Initialize SDK and build thread options const codex = await getCodex(codexConfig.codexBinaryPath); const threadOptions = buildThreadOptions(cwd, requestOptions?.model, assistantConfig); @@ -188,6 +501,7 @@ export class CodexProvider implements IAgentProvider { throw new Error('Query aborted'); } + // 2. Create or resume thread let sessionResumeFailed = false; let thread; if (resumeSessionId) { @@ -227,7 +541,8 @@ export class CodexProvider implements IAgentProvider { }; } - let lastTodoListSignature: string | undefined; + // 3. Build turn options + const { turnOptions, hasOutputFormat } = buildTurnOptions(requestOptions); let lastError: Error | undefined; for (let attempt = 0; attempt <= MAX_SUBPROCESS_RETRIES; attempt++) { @@ -249,254 +564,16 @@ export class CodexProvider implements IAgentProvider { } try { - const turnOptions: TurnOptions = {}; - const hasOutputFormat = !!( - requestOptions?.outputFormat ?? requestOptions?.nodeConfig?.output_format - ); - if (requestOptions?.outputFormat) { - turnOptions.outputSchema = requestOptions.outputFormat.schema; - } - // Also check nodeConfig.output_format (workflow path) - if (requestOptions?.nodeConfig?.output_format && !requestOptions?.outputFormat) { - turnOptions.outputSchema = requestOptions.nodeConfig.output_format; - } - // Track accumulated text for structured output normalization - let accumulatedText = ''; - if (requestOptions?.abortSignal) { - turnOptions.signal = requestOptions.abortSignal; - } - + // 4. Run streamed turn const result = await thread.runStreamed(prompt, turnOptions); - for await (const event of result.events) { - if (requestOptions?.abortSignal?.aborted) { - getLog().info('query_aborted_between_events'); - break; - } - - if (event.type === 'item.started') { - const item = event.item; - getLog().debug( - { eventType: event.type, itemType: item.type, itemId: item.id }, - 'item_started' - ); - } - - if (event.type === 'error') { - getLog().error({ message: event.message }, 'stream_error'); - if (!event.message.includes('MCP client')) { - yield { type: 'system', content: `⚠️ ${event.message}` }; - } - continue; - } - - if (event.type === 'turn.failed') { - const errorObj = event.error as { message?: string } | undefined; - const errorMessage = errorObj?.message ?? 'Unknown error'; - getLog().error({ errorMessage }, 'turn_failed'); - yield { - type: 'system', - content: `❌ Turn failed: ${errorMessage}`, - }; - break; - } - - if (event.type === 'item.completed') { - const item = event.item; - - const logContext: Record = { - eventType: event.type, - itemType: item.type, - itemId: item.id, - }; - if (item.type === 'command_execution' && item.command) { - logContext.command = item.command; - } - getLog().debug(logContext, 'item_completed'); - - switch (item.type) { - case 'agent_message': - if (item.text) { - if (hasOutputFormat) accumulatedText += item.text; - yield { type: 'assistant', content: item.text }; - } - break; - - case 'command_execution': - if (item.command) { - yield { type: 'tool', toolName: item.command }; - const exitSuffix = - item.exit_code != null && item.exit_code !== 0 - ? `\n[exit code: ${item.exit_code}]` - : ''; - yield { - type: 'tool_result', - toolName: item.command, - toolOutput: (item.aggregated_output ?? '') + exitSuffix, - }; - } else { - getLog().warn({ itemId: item.id }, 'command_execution_missing_command'); - } - break; - - case 'reasoning': - if (item.text) { - yield { type: 'thinking', content: item.text }; - } - break; - - case 'web_search': - if (item.query) { - const searchToolName = `🔍 Searching: ${item.query}`; - yield { type: 'tool', toolName: searchToolName }; - yield { type: 'tool_result', toolName: searchToolName, toolOutput: '' }; - } else { - getLog().debug({ itemId: item.id }, 'web_search_missing_query'); - } - break; - - case 'todo_list': - if (Array.isArray(item.items) && item.items.length > 0) { - const normalizedItems = item.items.map(t => ({ - text: typeof t.text === 'string' ? t.text : '(unnamed task)', - completed: t.completed ?? false, - })); - const signature = JSON.stringify(normalizedItems); - if (signature !== lastTodoListSignature) { - lastTodoListSignature = signature; - const taskList = normalizedItems - .map(t => `${t.completed ? '✅' : '⬜'} ${t.text}`) - .join('\n'); - yield { type: 'system', content: `📋 Tasks:\n${taskList}` }; - } - } else { - getLog().debug({ itemId: item.id }, 'todo_list_empty_or_invalid'); - } - break; - - case 'file_change': { - const statusIcon = item.status === 'failed' ? '❌' : '✅'; - const rawError = 'error' in item ? (item as { error?: unknown }).error : undefined; - const fileErrorMessage = - typeof rawError === 'string' - ? rawError - : typeof rawError === 'object' && rawError !== null && 'message' in rawError - ? String((rawError as { message: unknown }).message) - : undefined; - - if (Array.isArray(item.changes) && item.changes.length > 0) { - const changeList = item.changes - .map(c => { - const icon = c.kind === 'add' ? '➕' : c.kind === 'delete' ? '➖' : '📝'; - return `${icon} ${c.path ?? '(unknown file)'}`; - }) - .join('\n'); - const errorSuffix = - item.status === 'failed' && fileErrorMessage ? `\n${fileErrorMessage}` : ''; - yield { - type: 'system', - content: `${statusIcon} File changes:\n${changeList}${errorSuffix}`, - }; - } else if (item.status === 'failed') { - getLog().warn( - { itemId: item.id, status: item.status }, - 'file_change_failed_no_changes' - ); - const failMsg = fileErrorMessage - ? `❌ File change failed: ${fileErrorMessage}` - : '❌ File change failed'; - yield { type: 'system', content: failMsg }; - } else { - getLog().debug( - { itemId: item.id, status: item.status }, - 'file_change_no_changes' - ); - } - break; - } - - case 'mcp_tool_call': { - const toolInfo = - item.server && item.tool - ? `${item.server}/${item.tool}` - : (item.tool ?? item.server ?? 'MCP tool'); - const mcpToolName = `🔌 MCP: ${toolInfo}`; - - yield { type: 'tool', toolName: mcpToolName }; - - if (item.status === 'failed') { - getLog().warn( - { - server: item.server, - tool: item.tool, - error: item.error, - itemId: item.id, - }, - 'mcp_tool_call_failed' - ); - const errMsg = item.error?.message - ? `❌ Error: ${item.error.message}` - : '❌ Error: MCP tool failed'; - yield { type: 'tool_result', toolName: mcpToolName, toolOutput: errMsg }; - } else { - let toolOutput = ''; - if (item.result?.content) { - if (Array.isArray(item.result.content)) { - toolOutput = JSON.stringify(item.result.content); - } else { - getLog().warn( - { - itemId: item.id, - server: item.server, - tool: item.tool, - resultType: typeof item.result.content, - }, - 'mcp_tool_call_unexpected_result_shape' - ); - } - } - yield { type: 'tool_result', toolName: mcpToolName, toolOutput }; - } - break; - } - } - } - - if (event.type === 'turn.completed') { - getLog().debug('turn_completed'); - const usage = extractUsageFromCodexEvent(event); - - // Codex returns structured output inline in agent_message text. - // Normalize: parse as JSON and put on structuredOutput so the - // dag-executor can handle all providers uniformly. - let structuredOutput: unknown; - if (hasOutputFormat && accumulatedText) { - try { - structuredOutput = JSON.parse(accumulatedText); - getLog().debug('codex.structured_output_parsed'); - } catch { - getLog().warn( - { outputPreview: accumulatedText.slice(0, 200) }, - 'codex.structured_output_not_json' - ); - yield { - type: 'system', - content: - '⚠️ Structured output requested but Codex returned non-JSON text. ' + - 'Downstream $nodeId.output.field references may not evaluate correctly.', - }; - } - } - - yield { - type: 'result', - sessionId: thread.id ?? undefined, - tokens: usage, - ...(structuredOutput !== undefined ? { structuredOutput } : {}), - }; - break; - } - } + // 5. Stream normalized events (fresh state per attempt to avoid dedup leaks) + yield* streamCodexEvents( + result.events as AsyncIterable>, + hasOutputFormat, + thread.id, + requestOptions?.abortSignal + ); return; } catch (error) { const err = error as Error; @@ -505,36 +582,24 @@ export class CodexProvider implements IAgentProvider { throw new Error('Query aborted'); } - const errorClass = classifyCodexError(err.message); + const { enrichedError, errorClass, shouldRetry } = classifyAndEnrichCodexError( + err, + requestOptions?.model + ); + getLog().error( { err, errorClass, attempt, maxRetries: MAX_SUBPROCESS_RETRIES }, 'query_error' ); - if (errorClass === 'model_access') { - throw new Error(buildModelAccessMessage(requestOptions?.model)); - } - - if (errorClass === 'auth') { - const enrichedError = new Error(`Codex auth error: ${err.message}`); - enrichedError.cause = error; + if (!shouldRetry || attempt >= MAX_SUBPROCESS_RETRIES) { throw enrichedError; } - if ( - attempt < MAX_SUBPROCESS_RETRIES && - (errorClass === 'rate_limit' || errorClass === 'crash') - ) { - const delayMs = this.retryBaseDelayMs * Math.pow(2, attempt); - getLog().info({ attempt, delayMs, errorClass }, 'retrying_query'); - await new Promise(resolve => setTimeout(resolve, delayMs)); - lastError = err; - continue; - } - - const enrichedError = new Error(`Codex ${errorClass}: ${err.message}`); - enrichedError.cause = error; - throw enrichedError; + const delayMs = this.retryBaseDelayMs * Math.pow(2, attempt); + getLog().info({ attempt, delayMs, errorClass }, 'retrying_query'); + await new Promise(resolve => setTimeout(resolve, delayMs)); + lastError = enrichedError; } }