From f5be31e87da0a5cb041b8a156dc9f14cada69329 Mon Sep 17 00:00:00 2001 From: im10furry Date: Sun, 5 Jul 2026 03:02:55 -0500 Subject: [PATCH] fix(openai): fail degraded streaming results --- .../entrypoints/cli/print/runSingleTurn.ts | 24 +++- .../core/src/ai/llm/openai/queryOpenAI.ts | 47 +++++--- packages/core/src/ai/llm/openai/stream.ts | 113 +++++++++++++++--- packages/core/src/ai/openai/stream.ts | 94 ++++++++++++++- packages/core/src/engine/messages/api.ts | 10 +- .../messages-normalization-reorder.test.ts | 7 +- .../src/test/unit/openai-stream-abort.test.ts | 87 +++++++++++++- .../test/unit/print-mode-api-error.test.ts | 71 +++++++++++ .../src/test/unit/stream-json-session.test.ts | 89 +++++++++++++- .../src/utils/kodeAgentStreamJsonSession.ts | 23 +++- 10 files changed, 513 insertions(+), 52 deletions(-) create mode 100644 packages/core/src/test/unit/print-mode-api-error.test.ts diff --git a/apps/cli/src/entrypoints/cli/print/runSingleTurn.ts b/apps/cli/src/entrypoints/cli/print/runSingleTurn.ts index 9e57be61..95da27f2 100644 --- a/apps/cli/src/entrypoints/cli/print/runSingleTurn.ts +++ b/apps/cli/src/entrypoints/cli/print/runSingleTurn.ts @@ -59,6 +59,10 @@ type MakeSdkResultMessageFn = (args: { uuid?: string }) => unknown +function isApiErrorAssistantMessage(message: Message | null): boolean { + return message?.type === 'assistant' && message.isApiErrorMessage === true +} + export async function runSingleTurnPrint(args: { runTurn: RunTurnFn kodeMessageToSdkMessage: KodeMessageToSdkMessageFn @@ -154,9 +158,16 @@ export async function runSingleTurnPrint(args: { : queryError ? String(queryError) : '' + const hasApiErrorAssistant = isApiErrorAssistantMessage(lastAssistant) let structuredOutput: Record | undefined - if (args.jsonSchema && !queryError && !budgetExceeded && !maxTurnsExceeded) { + if ( + args.jsonSchema && + !queryError && + !hasApiErrorAssistant && + !budgetExceeded && + !maxTurnsExceeded + ) { try { const raw = typeof textFromAssistant === 'string' ? textFromAssistant : '' const fenced = raw.trim() @@ -200,6 +211,11 @@ export async function runSingleTurnPrint(args: { const shouldReturnBudgetExceeded = !shouldReturnMaxTurnsExceeded && (budgetExceeded || queryError instanceof MaxBudgetUsdExceededError) + const shouldReturnDegradedApiError = + !queryError && + !shouldReturnBudgetExceeded && + !shouldReturnMaxTurnsExceeded && + hasApiErrorAssistant const resultNumTurns = (() => { if ( @@ -229,12 +245,14 @@ export async function runSingleTurnPrint(args: { isError: shouldReturnBudgetExceeded || shouldReturnMaxTurnsExceeded ? false - : Boolean(queryError), + : Boolean(queryError) || shouldReturnDegradedApiError, subtype: shouldReturnMaxTurnsExceeded ? 'error_max_turns' : shouldReturnBudgetExceeded ? 'error_max_budget_usd' - : undefined, + : shouldReturnDegradedApiError + ? 'error_during_execution' + : undefined, uuid: randomUUID(), }) diff --git a/packages/core/src/ai/llm/openai/queryOpenAI.ts b/packages/core/src/ai/llm/openai/queryOpenAI.ts index c3ce20ac..1c16363b 100644 --- a/packages/core/src/ai/llm/openai/queryOpenAI.ts +++ b/packages/core/src/ai/llm/openai/queryOpenAI.ts @@ -41,7 +41,7 @@ import { convertOpenAIResponseToAnthropic, } from './conversion' import { buildOpenAIChatCompletionCreateParams, isGPT5Model } from './params' -import { handleMessageStream } from './stream' +import { handleMessageStream, isOpenAIStreamDegradedResponse } from './stream' import { buildAssistantMessageFromUnifiedResponse } from './unifiedResponse' import { getMaxTokensFromProfile, normalizeUsage } from './usage' @@ -58,6 +58,25 @@ function containsCommittedToolResult( return messages.some(message => message.role === 'tool') } +function createAssistantMessageFromOpenAIResponse(args: { + response: OpenAI.ChatCompletion + tools: Tool[] + start: number +}): AssistantMessage { + const message = convertOpenAIResponseToAnthropic(args.response, args.tools) + const assistantMsg: AssistantMessage = { + type: 'assistant', + message, + costUSD: 0, + durationMs: Date.now() - args.start, + uuid: randomUUID() as UUID, + } + if (isOpenAIStreamDegradedResponse(args.response)) { + assistantMsg.isApiErrorMessage = true + } + return assistantMsg +} + export async function queryOpenAI( messages: (UserMessage | AssistantMessage)[], systemPrompt: string[], @@ -296,14 +315,11 @@ export async function queryOpenAI( finalResponse = s } - const message = convertOpenAIResponseToAnthropic(finalResponse, tools) - const assistantMsg: AssistantMessage = { - type: 'assistant', - message, - costUSD: 0, - durationMs: Date.now() - start, - uuid: randomUUID() as UUID, - } + const assistantMsg = createAssistantMessageFromOpenAIResponse({ + response: finalResponse, + tools, + start, + }) return { assistantMessage: assistantMsg, rawResponse: finalResponse, @@ -349,14 +365,11 @@ export async function queryOpenAI( } else { finalResponse = s } - const message = convertOpenAIResponseToAnthropic(finalResponse, tools) - const assistantMsg: AssistantMessage = { - type: 'assistant', - message, - costUSD: 0, - durationMs: Date.now() - start, - uuid: randomUUID() as UUID, - } + const assistantMsg = createAssistantMessageFromOpenAIResponse({ + response: finalResponse, + tools, + start, + }) return { assistantMessage: assistantMsg, rawResponse: finalResponse, diff --git a/packages/core/src/ai/llm/openai/stream.ts b/packages/core/src/ai/llm/openai/stream.ts index a7d48dca..ff459141 100644 --- a/packages/core/src/ai/llm/openai/stream.ts +++ b/packages/core/src/ai/llm/openai/stream.ts @@ -1,5 +1,6 @@ import type { ChatCompletionStream } from 'openai/lib/ChatCompletionStream' import type OpenAI from 'openai' +import { OpenAIStreamError } from '#core/ai/openai/stream' import { debug as debugLogger } from '#core/utils/debugLogger' import { setRequestStatus, @@ -7,6 +8,11 @@ import { updateRequestTokens, } from '#core/utils/requestStatus' +export type OpenAIStreamDegradedCompletion = OpenAI.ChatCompletion & { + __streamDegraded?: true + __streamDegradationReason?: string +} + function messageReducer( previous: OpenAI.ChatCompletionMessage, item: OpenAI.ChatCompletionChunk, @@ -58,6 +64,25 @@ function throwIfAborted(signal?: AbortSignal): void { } } +function hasUsableAssistantOutput( + message: OpenAI.ChatCompletionMessage, +): boolean { + const record = message as unknown as Record + return ( + (typeof message.content === 'string' && message.content.length > 0) || + (Array.isArray(message.tool_calls) && message.tool_calls.length > 0) || + (typeof record.reasoning === 'string' && record.reasoning.length > 0) || + (typeof record.reasoning_content === 'string' && + record.reasoning_content.length > 0) + ) +} + +export function isOpenAIStreamDegradedResponse( + response: OpenAI.ChatCompletion, +): response is OpenAIStreamDegradedCompletion { + return (response as OpenAIStreamDegradedCompletion).__streamDegraded === true +} + export async function handleMessageStream( stream: ChatCompletionStream, signal?: AbortSignal, @@ -68,6 +93,9 @@ export async function handleMessageStream( let errorCount = 0 let hasMarkedStreaming = false let outputTokenCount = 0 + let finishReason: OpenAI.ChatCompletion.Choice['finish_reason'] | null = null + let degradationReason: string | null = null + let lastChunkError: unknown = null debugLogger.api('OPENAI_STREAM_START', { streamStartTime: String(streamStartTime), @@ -140,8 +168,11 @@ export async function handleMessageStream( if (chunk?.usage?.completion_tokens) { updateRequestTokens(chunk.usage.completion_tokens) } + const chunkFinishReason = chunk?.choices?.[0]?.finish_reason + if (chunkFinishReason) finishReason = chunkFinishReason } catch (chunkError) { errorCount++ + lastChunkError = chunkError debugLogger.error('OPENAI_STREAM_CHUNK_ERROR', { chunkNumber: String(chunkCount), errorMessage: @@ -159,6 +190,24 @@ export async function handleMessageStream( throwIfAborted(signal) + if (errorCount > 0 && !hasUsableAssistantOutput(message)) { + throw new OpenAIStreamError( + 'unexpected_error', + `OpenAI stream chunk processing failed before any assistant content: ${ + lastChunkError instanceof Error + ? lastChunkError.message + : String(lastChunkError ?? 'unknown error') + }`, + ) + } + + if (chunkCount === 0 || !hasUsableAssistantOutput(message)) { + throw new OpenAIStreamError( + 'empty_response', + 'OpenAI stream completed without assistant content or tool calls', + ) + } + debugLogger.api('OPENAI_STREAM_COMPLETE', { totalChunks: String(chunkCount), errorCount: String(errorCount), @@ -167,21 +216,48 @@ export async function handleMessageStream( finalMessageId: id || 'undefined', }) } catch (streamError) { - debugLogger.error('OPENAI_STREAM_FATAL_ERROR', { - totalChunks: String(chunkCount), - errorCount: String(errorCount), - errorMessage: - streamError instanceof Error - ? streamError.message - : String(streamError), - errorType: - streamError instanceof Error - ? streamError.constructor.name - : typeof streamError, - }) - throw streamError + if ( + !( + streamError instanceof Error && + streamError.message === 'Request was cancelled' + ) && + hasUsableAssistantOutput(message) + ) { + degradationReason = + streamError instanceof OpenAIStreamError + ? streamError.reason + : streamError instanceof Error + ? streamError.message + : String(streamError) + debugLogger.warn('OPENAI_STREAM_DEGRADED_PARTIAL', { + reason: degradationReason, + chunkCount: String(chunkCount), + }) + } else { + debugLogger.error('OPENAI_STREAM_FATAL_ERROR', { + totalChunks: String(chunkCount), + errorCount: String(errorCount), + errorMessage: + streamError instanceof Error + ? streamError.message + : String(streamError), + errorType: + streamError instanceof Error + ? streamError.constructor.name + : typeof streamError, + }) + throw streamError + } + } + + if (errorCount > 0 && !degradationReason) { + degradationReason = + lastChunkError instanceof Error + ? lastChunkError.message + : 'chunk_processing_error' } - return { + + const completion: OpenAIStreamDegradedCompletion = { id, created, model, @@ -190,10 +266,17 @@ export async function handleMessageStream( { index: 0, message, - finish_reason: 'stop', + finish_reason: finishReason ?? (degradationReason ? 'length' : 'stop'), logprobs: undefined, }, ], usage, } + + if (degradationReason) { + completion.__streamDegraded = true + completion.__streamDegradationReason = degradationReason + } + + return completion } diff --git a/packages/core/src/ai/openai/stream.ts b/packages/core/src/ai/openai/stream.ts index 12f1e2eb..e174a161 100644 --- a/packages/core/src/ai/openai/stream.ts +++ b/packages/core/src/ai/openai/stream.ts @@ -3,6 +3,52 @@ import type { Response } from 'undici' import { debug as debugLogger } from '#core/utils/debugLogger' +export type StreamDegradationReason = + | 'read_error' + | 'json_parse_error' + | 'provider_error' + | 'unexpected_error' + | 'empty_response' + +export class OpenAIStreamError extends Error { + readonly reason: StreamDegradationReason + + constructor(reason: StreamDegradationReason, message: string) { + super(message) + this.name = 'OpenAIStreamError' + this.reason = reason + } +} + +function asRecord(value: unknown): Record | null { + if (!value || typeof value !== 'object' || Array.isArray(value)) return null + return value as Record +} + +function trimForLog(value: string): string { + return value.length <= 500 ? value : `${value.slice(0, 500)}...` +} + +function extractStreamErrorMessage(value: unknown): string | null { + const record = asRecord(value) + if (!record || !('error' in record)) return null + + const error = record.error + if (typeof error === 'string' && error.trim()) return error.trim() + + const errorRecord = asRecord(error) + if (!errorRecord) return 'OpenAI stream returned an error payload' + + const message = errorRecord.message + if (typeof message === 'string' && message.trim()) return message.trim() + + try { + return JSON.stringify(errorRecord) + } catch { + return 'OpenAI stream returned an error payload' + } +} + export function createStreamProcessor( stream: NonNullable, signal?: AbortSignal, @@ -24,7 +70,12 @@ export function createStreamProcessor( debugLogger.warn('OPENAI_STREAM_READ_ERROR', { error: e instanceof Error ? e.message : String(e), }) - break + throw new OpenAIStreamError( + 'read_error', + `OpenAI stream read failed: ${ + e instanceof Error ? e.message : String(e) + }`, + ) } const { done, value } = readResult @@ -47,12 +98,25 @@ export function createStreamProcessor( const data = line.slice(6).trim() if (data) { try { - yield JSON.parse(data) as OpenAI.ChatCompletionChunk + const parsed = JSON.parse(data) + const errorMessage = extractStreamErrorMessage(parsed) + if (errorMessage) { + throw new OpenAIStreamError( + 'provider_error', + `OpenAI stream error: ${errorMessage}`, + ) + } + yield parsed as OpenAI.ChatCompletionChunk } catch (e) { + if (e instanceof OpenAIStreamError) throw e debugLogger.warn('OPENAI_STREAM_JSON_PARSE_ERROR', { - data, + data: trimForLog(data), error: e instanceof Error ? e.message : String(e), }) + throw new OpenAIStreamError( + 'json_parse_error', + `OpenAI stream emitted malformed JSON: ${trimForLog(data)}`, + ) } } } @@ -68,19 +132,39 @@ export function createStreamProcessor( const data = line.slice(6).trim() if (!data) continue try { - yield JSON.parse(data) as OpenAI.ChatCompletionChunk + const parsed = JSON.parse(data) + const errorMessage = extractStreamErrorMessage(parsed) + if (errorMessage) { + throw new OpenAIStreamError( + 'provider_error', + `OpenAI stream error: ${errorMessage}`, + ) + } + yield parsed as OpenAI.ChatCompletionChunk } catch (e) { + if (e instanceof OpenAIStreamError) throw e debugLogger.warn('OPENAI_STREAM_FINAL_JSON_PARSE_ERROR', { - data, + data: trimForLog(data), error: e instanceof Error ? e.message : String(e), }) + throw new OpenAIStreamError( + 'json_parse_error', + `OpenAI stream emitted malformed JSON: ${trimForLog(data)}`, + ) } } } } catch (e) { + if (e instanceof OpenAIStreamError) throw e debugLogger.warn('OPENAI_STREAM_UNEXPECTED_ERROR', { error: e instanceof Error ? e.message : String(e), }) + throw new OpenAIStreamError( + 'unexpected_error', + `OpenAI stream failed unexpectedly: ${ + e instanceof Error ? e.message : String(e) + }`, + ) } finally { try { reader.releaseLock() diff --git a/packages/core/src/engine/messages/api.ts b/packages/core/src/engine/messages/api.ts index 61fbd072..dd4f9cfb 100644 --- a/packages/core/src/engine/messages/api.ts +++ b/packages/core/src/engine/messages/api.ts @@ -11,12 +11,8 @@ import type { AssistantMessage, Message, UserMessage } from '#core/query' export function normalizeMessagesForAPI( messages: Message[], ): (UserMessage | AssistantMessage)[] { - function isSyntheticApiErrorMessage(message: Message): boolean { - return ( - message.type === 'assistant' && - message.isApiErrorMessage === true && - message.message.model === '' - ) + function isApiErrorMessage(message: Message): boolean { + return message.type === 'assistant' && message.isApiErrorMessage === true } function isSyntheticMetaMessage(message: Message): boolean { @@ -73,7 +69,7 @@ export function normalizeMessagesForAPI( const result: (UserMessage | AssistantMessage)[] = [] for (const message of messages) { if (message.type === 'progress') continue - if (isSyntheticApiErrorMessage(message)) continue + if (isApiErrorMessage(message)) continue if (isSyntheticMetaMessage(message)) continue switch (message.type) { diff --git a/packages/core/src/test/unit/messages-normalization-reorder.test.ts b/packages/core/src/test/unit/messages-normalization-reorder.test.ts index 6fbc1fd6..22684c27 100644 --- a/packages/core/src/test/unit/messages-normalization-reorder.test.ts +++ b/packages/core/src/test/unit/messages-normalization-reorder.test.ts @@ -61,10 +61,15 @@ describe('messages normalization + reordering parity', () => { expect(content[2]).toMatchObject({ type: 'text', text: 'meta' }) }) - test('normalizeMessagesForAPI filters synthetic api error assistant messages', () => { + test('normalizeMessagesForAPI filters api error assistant messages', () => { + const degraded = createAssistantMessage('partial response') + degraded.isApiErrorMessage = true + degraded.message.model = 'gpt-4' + const out = normalizeMessagesForAPI([ createUserMessage('hi'), createAssistantAPIErrorMessage('oops'), + degraded, createAssistantMessage('ok'), ]) expect(out.map(m => m.type)).toEqual(['user', 'assistant']) diff --git a/packages/core/src/test/unit/openai-stream-abort.test.ts b/packages/core/src/test/unit/openai-stream-abort.test.ts index e284b2eb..088fbfad 100644 --- a/packages/core/src/test/unit/openai-stream-abort.test.ts +++ b/packages/core/src/test/unit/openai-stream-abort.test.ts @@ -1,5 +1,9 @@ import { describe, expect, test } from 'bun:test' -import { handleMessageStream } from '#core/ai/llm/openai/stream' +import { + handleMessageStream, + isOpenAIStreamDegradedResponse, +} from '#core/ai/llm/openai/stream' +import { createStreamProcessor } from '#core/ai/openai/stream' function chunk(delta: Record) { return { @@ -11,6 +15,18 @@ function chunk(delta: Record) { } } +function sseBody(lines: string[]): ReadableStream { + const encoder = new TextEncoder() + return new ReadableStream({ + start(controller) { + for (const line of lines) { + controller.enqueue(encoder.encode(`${line}\n`)) + } + controller.close() + }, + }) +} + describe('OpenAI stream cancellation', () => { test('rejects when signal is aborted before reading stream chunks', async () => { const controller = new AbortController() @@ -38,3 +54,72 @@ describe('OpenAI stream cancellation', () => { ).rejects.toThrow('Request was cancelled') }) }) + +describe('OpenAI stream degradation', () => { + test('rejects read failures before the first usable assistant token', async () => { + const body = new ReadableStream({ + pull(controller) { + controller.error(new Error('socket closed')) + }, + }) + + await expect( + handleMessageStream(createStreamProcessor(body as any) as any, undefined), + ).rejects.toThrow('socket closed') + }) + + test('rejects malformed-only SSE instead of returning no content', async () => { + const body = sseBody(['data: {bad json}']) + + await expect( + handleMessageStream(createStreamProcessor(body as any) as any, undefined), + ).rejects.toThrow('malformed JSON') + }) + + test('rejects SSE error payloads instead of returning no content', async () => { + const body = sseBody(['data: {"error":{"message":"provider unavailable"}}']) + + await expect( + handleMessageStream(createStreamProcessor(body as any) as any, undefined), + ).rejects.toThrow('provider unavailable') + }) + + test('marks malformed SSE chunks as degraded without blocking partial output', async () => { + const validChunk = JSON.stringify(chunk({ content: 'partial' })) + const body = sseBody([`data: ${validChunk}`, 'data: {bad json}']) + + const result = await handleMessageStream( + createStreamProcessor(body as any) as any, + undefined, + ) + + expect(result.choices[0]?.message.content).toBe('partial') + expect(result.choices[0]?.finish_reason).toBe('length') + expect(isOpenAIStreamDegradedResponse(result)).toBe(true) + }) + + test('marks stream read failures as degraded without blocking partial output', async () => { + const encoder = new TextEncoder() + const validChunk = JSON.stringify(chunk({ content: 'partial' })) + let sent = false + const body = new ReadableStream({ + pull(controller) { + if (!sent) { + sent = true + controller.enqueue(encoder.encode(`data: ${validChunk}\n`)) + return + } + controller.error(new Error('socket closed')) + }, + }) + + const result = await handleMessageStream( + createStreamProcessor(body as any) as any, + undefined, + ) + + expect(result.choices[0]?.message.content).toBe('partial') + expect(result.choices[0]?.finish_reason).toBe('length') + expect(isOpenAIStreamDegradedResponse(result)).toBe(true) + }) +}) diff --git a/packages/core/src/test/unit/print-mode-api-error.test.ts b/packages/core/src/test/unit/print-mode-api-error.test.ts new file mode 100644 index 00000000..6a8b1b9a --- /dev/null +++ b/packages/core/src/test/unit/print-mode-api-error.test.ts @@ -0,0 +1,71 @@ +import { describe, expect, test } from 'bun:test' +import type { Message } from '#core/query' +import { + createAssistantAPIErrorMessage, + createUserMessage, +} from '#core/utils/messages' +import { + kodeMessageToSdkMessage, + makeSdkResultMessage, +} from '#protocol/utils/kodeAgentStreamJson' +import { runSingleTurnPrint } from '#host-cli/entrypoints/cli/print/runSingleTurn' + +describe('print mode API error results', () => { + test('API error assistant messages emit failed results and exit non-zero', async () => { + const written: unknown[] = [] + let exitCode: number | undefined + const originalExit = process.exit + + process.exit = ((code?: string | number | null | undefined) => { + exitCode = typeof code === 'number' ? code : Number(code ?? 0) + throw new Error(`process.exit:${exitCode}`) + }) as typeof process.exit + + try { + await expect( + runSingleTurnPrint({ + runTurn: async function* (): AsyncGenerator { + yield createAssistantAPIErrorMessage( + 'API Error: provider unavailable', + ) + }, + kodeMessageToSdkMessage, + makeSdkResultMessage, + messages: [createUserMessage('hi')], + systemPrompt: [], + context: {}, + canUseTool: (async () => ({ result: true })) as any, + toolUseContext: { + abortController: new AbortController(), + turnCount: 1, + } as any, + sessionId: 'sess_test', + outputFormat: 'stream-json', + writeSdkLine: obj => { + written.push(obj) + }, + sdkMessages: [], + startedAt: Date.now(), + getTotalCostUsd: () => 0, + getTotalApiDurationMs: () => 0, + jsonSchema: null, + verbose: false, + }), + ).rejects.toThrow('process.exit:1') + } finally { + process.exit = originalExit + } + + const result = written.find( + (line): line is Record => + Boolean(line) && + typeof line === 'object' && + (line as Record).type === 'result', + ) + + expect(exitCode).toBe(1) + expect(result?.subtype).toBe('error_during_execution') + expect(result?.is_error).toBe(true) + expect(result?.result).toBe('API Error: provider unavailable') + }) +}) diff --git a/packages/core/src/test/unit/stream-json-session.test.ts b/packages/core/src/test/unit/stream-json-session.test.ts index 17f5c795..ec2b7dbd 100644 --- a/packages/core/src/test/unit/stream-json-session.test.ts +++ b/packages/core/src/test/unit/stream-json-session.test.ts @@ -3,7 +3,11 @@ import { createInterface } from 'node:readline' import { PassThrough } from 'node:stream' import { KodeAgentStructuredStdio } from '#protocol/utils/kodeAgentStructuredStdio' import { runKodeAgentStreamJsonSession } from '#protocol/utils/kodeAgentStreamJsonSession' -import { createAssistantMessage, createUserMessage } from '#core/utils/messages' +import { + createAssistantAPIErrorMessage, + createAssistantMessage, + createUserMessage, +} from '#core/utils/messages' import type { Message } from '#core/query' import type { ToolUseContext } from '#core/tooling/Tool' @@ -225,4 +229,87 @@ describe('stream-json persistent session', () => { rlOut.close() stdout.end() }) + + test('API error assistant messages degrade result subtype without blocking the session', async () => { + const stdin = new PassThrough() + const stdout = new PassThrough() + const rlOut = createInterface({ input: stdout }) + const nextLine = makeLineReader(rlOut) + + const structured = new KodeAgentStructuredStdio(stdin, stdout) + structured.start() + + let queryCalls = 0 + const query = async function* (): AsyncGenerator { + queryCalls += 1 + if (queryCalls === 1) { + yield createAssistantAPIErrorMessage('API Error: provider unavailable') + return + } + yield createAssistantMessage(`turn:${queryCalls}`) + } + + const canUseTool = async () => ({ result: true }) + const toolUseContextBase = { messageId: undefined, readFileTimestamps: {} } + + const sessionPromise = runKodeAgentStreamJsonSession< + Message, + ToolUseContext + >({ + structured, + query, + makeUserMessage: content => { + const text = + typeof content === 'string' ? content : JSON.stringify(content) + return createUserMessage(text) + }, + writeSdkLine: obj => { + stdout.write(JSON.stringify(obj) + '\n') + }, + sessionId: 'sess_test', + systemPrompt: [], + context: {}, + canUseTool, + toolUseContextBase, + replayUserMessages: false, + getTotalCostUsd: () => 0, + }) + + stdin.write( + JSON.stringify({ + type: 'user', + message: { role: 'user', content: 'hi' }, + }) + '\n', + ) + + const assistant = JSON.parse(await nextLine()) + expect(assistant.type).toBe('assistant') + + const result = JSON.parse(await nextLine()) + expect(result.type).toBe('result') + expect(result.subtype).toBe('error_during_execution') + expect(result.is_error).toBe(true) + + stdin.write( + JSON.stringify({ + type: 'user', + message: { role: 'user', content: 'again' }, + }) + '\n', + ) + + const assistant2 = JSON.parse(await nextLine()) + expect(assistant2.type).toBe('assistant') + + const result2 = JSON.parse(await nextLine()) + expect(result2.type).toBe('result') + expect(result2.subtype).toBe('success') + expect(result2.is_error).toBe(false) + + stdin.end() + await sessionPromise + expect(queryCalls).toBe(2) + + rlOut.close() + stdout.end() + }) }) diff --git a/packages/protocol/src/utils/kodeAgentStreamJsonSession.ts b/packages/protocol/src/utils/kodeAgentStreamJsonSession.ts index b4f43470..081597b2 100644 --- a/packages/protocol/src/utils/kodeAgentStreamJsonSession.ts +++ b/packages/protocol/src/utils/kodeAgentStreamJsonSession.ts @@ -47,6 +47,14 @@ function extractAssistantUsage(message: unknown): unknown { return msg?.usage } +function isApiErrorAssistantMessage(message: unknown): boolean { + return ( + isRecord(message) && + message.type === 'assistant' && + message.isApiErrorMessage === true + ) +} + export async function runKodeAgentStreamJsonSession< M extends MessageWithUuid, C extends { abortController: AbortController }, @@ -179,11 +187,13 @@ export async function runKodeAgentStreamJsonSession< totalCostUsd >= args.maxBudgetUsd const maxTurnsExceeded = queryError instanceof MaxTurnsExceededError + const hasApiErrorAssistant = isApiErrorAssistantMessage(lastAssistant) let structuredOutput: Record | undefined if ( args.jsonSchema && !queryError && + !hasApiErrorAssistant && !budgetExceeded && !maxTurnsExceeded ) { @@ -230,7 +240,14 @@ export async function runKodeAgentStreamJsonSession< const isError = !budgetExceeded && !maxTurnsExceeded && - (Boolean(queryError) || turnAbortController.signal.aborted) + (Boolean(queryError) || + turnAbortController.signal.aborted || + hasApiErrorAssistant) + const shouldReturnDegradedApiError = + !queryError && + !budgetExceeded && + !maxTurnsExceeded && + hasApiErrorAssistant args.writeSdkLine( makeSdkResultMessage({ @@ -252,7 +269,9 @@ export async function runKodeAgentStreamJsonSession< ? 'error_max_turns' : budgetExceeded ? 'error_max_budget_usd' - : undefined, + : shouldReturnDegradedApiError + ? 'error_during_execution' + : undefined, uuid: randomUUID(), }), )