diff --git a/.changeset/normalize-responses-rate-limit.md b/.changeset/normalize-responses-rate-limit.md new file mode 100644 index 0000000000..1ff4b7f167 --- /dev/null +++ b/.changeset/normalize-responses-rate-limit.md @@ -0,0 +1,6 @@ +--- +"@moonshot-ai/kosong": patch +"@moonshot-ai/kimi-code": patch +--- + +Normalize malformed Responses stream rate limit errors as provider rate limit failures. diff --git a/packages/kosong/src/providers/openai-responses.ts b/packages/kosong/src/providers/openai-responses.ts index 59c0c852da..00e31e3cf9 100644 --- a/packages/kosong/src/providers/openai-responses.ts +++ b/packages/kosong/src/providers/openai-responses.ts @@ -1,5 +1,10 @@ import type { ModelCapability } from '#/capability'; -import { APIContextOverflowError, ChatProviderError, isContextOverflowErrorCode } from '#/errors'; +import { + APIContextOverflowError, + APIStatusError, + ChatProviderError, + isContextOverflowErrorCode, +} from '#/errors'; import type { ContentPart, Message, StreamedMessagePart, ToolCall } from '#/message'; import { extractText } from '#/message'; import type { @@ -111,6 +116,10 @@ function readStringField(object: RawObject, key: string): string | undefined { return typeof value === 'string' ? value : undefined; } +function hasOwn(object: RawObject, key: string): boolean { + return Object.prototype.hasOwnProperty.call(object, key); +} + function readNullableStringField(object: RawObject, key: string): string | null | undefined { const value = object[key]; if (value === null) return null; @@ -237,9 +246,65 @@ function errorFromOpenAIResponsesEvent( if (isContextOverflowErrorCode(code)) { return new APIContextOverflowError(400, fullMessage); } + if (code === 'rate_limit_exceeded') { + return new APIStatusError(429, fullMessage); + } return new ChatProviderError(fullMessage); } +function parseNestedGatewayStreamError(message: string): + | { + code: string | null; + message: string; + param: string | null; + } + | undefined { + const marker = 'received error while streaming:'; + const markerIndex = message.indexOf(marker); + if (markerIndex === -1) return undefined; + + const jsonText = message.slice(markerIndex + marker.length).trim(); + if (jsonText.length === 0) return undefined; + + let parsed: unknown; + try { + parsed = JSON.parse(jsonText); + } catch { + return undefined; + } + + const error = asRawObject(parsed); + if (error === null) return undefined; + + const nestedMessage = readStringField(error, 'message'); + if (nestedMessage === undefined) return undefined; + + return { + code: readNullableStringField(error, 'code') ?? null, + message: nestedMessage, + param: readNullableStringField(error, 'param') ?? null, + }; +} + +function malformedStreamErrorEvent(message: string): ChatProviderError { + const nested = parseNestedGatewayStreamError(message); + if (nested !== undefined) { + return errorFromOpenAIResponsesEvent( + 'OpenAI Responses malformed stream error', + nested.code, + nested.message, + nested.param, + ); + } + + return errorFromOpenAIResponsesEvent( + 'OpenAI Responses malformed stream error', + null, + message, + null, + ); +} + function readResponsesFailedResponseError(response: RawObject): | { code: string | null; @@ -699,7 +764,16 @@ export class OpenAIResponsesStreamedMessage implements StreamedMessage { try { for await (const chunk of response) { - const type = requireStringField(chunk, 'type', 'stream event'); + const type = readStringField(chunk, 'type'); + if (type === undefined) { + if (!hasOwn(chunk, 'type')) { + const message = readStringField(chunk, 'message'); + if (message !== undefined) { + throw malformedStreamErrorEvent(message); + } + } + failResponsesDecode('stream event.type', 'must be a string.'); + } switch (type) { case 'response.output_text.delta': diff --git a/packages/kosong/test/openai-responses.test.ts b/packages/kosong/test/openai-responses.test.ts index 572527cba4..3cfbd41ca8 100644 --- a/packages/kosong/test/openai-responses.test.ts +++ b/packages/kosong/test/openai-responses.test.ts @@ -1,4 +1,4 @@ -import { APIContextOverflowError, ChatProviderError } from '#/errors'; +import { APIContextOverflowError, APIStatusError, ChatProviderError } from '#/errors'; import { generate } from '#/generate'; import type { ContentPart, Message, StreamedMessagePart, ToolCall } from '#/message'; import { @@ -1682,8 +1682,52 @@ describe('OpenAIResponsesChatProvider', () => { ]; const stream = new OpenAIResponsesStreamedMessage(makeAsyncIterable(events), true); + let caughtError: unknown; + try { + await collectStreamParts(stream); + } catch (error) { + caughtError = error; + } + + expect(caughtError).toBeInstanceOf(APIStatusError); + expect((caughtError as APIStatusError).statusCode).toBe(429); + expect((caughtError as Error).message).toMatch(/rate_limit_exceeded.*too many/); + }); + + it('normalizes malformed gateway error frames with nested rate-limit JSON', async () => { + const events = [ + { + message: + 'received error while streaming: {"type":"tokens","code":"rate_limit_exceeded","message":"Rate limit reached for gpt-5.5. Please try again in 325ms.","param":null}', + }, + ]; + const stream = new OpenAIResponsesStreamedMessage(makeAsyncIterable(events), true); + + let caughtError: unknown; + try { + await collectStreamParts(stream); + } catch (error) { + caughtError = error; + } + + expect(caughtError).toBeInstanceOf(APIStatusError); + expect((caughtError as APIStatusError).statusCode).toBe(429); + expect((caughtError as Error).message).toContain('Rate limit reached for gpt-5.5'); + expect((caughtError as Error).message).not.toContain('stream event.type must be a string'); + }); + + it('rejects malformed stream events with a non-string type even when message is present', async () => { + const events = [ + { + type: 42, + message: + 'received error while streaming: {"type":"tokens","code":"rate_limit_exceeded","message":"too many requests","param":null}', + }, + ]; + const stream = new OpenAIResponsesStreamedMessage(makeAsyncIterable(events), true); + await expect(collectStreamParts(stream)).rejects.toThrow( - /rate_limit_exceeded.*too many/, + 'OpenAI Responses decode error: stream event.type must be a string.', ); });