diff --git a/.changeset/calm-swarms-retry.md b/.changeset/calm-swarms-retry.md new file mode 100644 index 000000000..9518db1d3 --- /dev/null +++ b/.changeset/calm-swarms-retry.md @@ -0,0 +1,5 @@ +--- +"@moonshot-ai/kimi-code": patch +--- + +web: Recover transient subagent rate limits without surfacing them as session errors. diff --git a/apps/kimi-web/src/api/daemon/agentEventProjector.ts b/apps/kimi-web/src/api/daemon/agentEventProjector.ts index befe04d99..4d26110c4 100644 --- a/apps/kimi-web/src/api/daemon/agentEventProjector.ts +++ b/apps/kimi-web/src/api/daemon/agentEventProjector.ts @@ -57,6 +57,7 @@ const MAIN_AGENT_TRANSCRIPT_FRAMES = new Set([ 'tool.result', 'agent.status.updated', 'prompt.completed', + 'error', ]); // --------------------------------------------------------------------------- diff --git a/apps/kimi-web/test/agent-event-projector.test.ts b/apps/kimi-web/test/agent-event-projector.test.ts index 71167a3aa..29be76f60 100644 --- a/apps/kimi-web/test/agent-event-projector.test.ts +++ b/apps/kimi-web/test/agent-event-projector.test.ts @@ -1,3 +1,8 @@ +/** + * Web daemon projector contract for transcript isolation, task progress, and + * client-visible error projection. + */ + import { describe, expect, it } from 'vitest'; import { classifyFrame, createAgentProjector, subagentProgressText } from '../src/api/daemon/agentEventProjector'; @@ -61,6 +66,41 @@ describe('subagent streaming text', () => { }); }); +describe('agent error projection', () => { + it('drops a subagent error instead of surfacing it as a session warning', () => { + const projector = createAgentProjector(); + + expect( + projector.project( + 'error', + { agentId: 'sub-1', code: 'provider.rate_limit', message: 'Rate limited' }, + 's1', + ), + ).toEqual([]); + }); + + it('keeps a main-agent error visible to the session', () => { + const projector = createAgentProjector(); + + expect( + projector.project( + 'error', + { agentId: 'main', code: 'provider.rate_limit', message: 'Rate limited' }, + 's1', + ), + ).toEqual([ + { + type: 'unknown', + raw: { + _agentError: true, + code: 'provider.rate_limit', + message: 'Rate limited', + }, + }, + ]); + }); +}); + describe('cron.fired', () => { it('synthesizes a user message so the cron notice renders live', () => { const projector = createAgentProjector(); diff --git a/packages/agent-core-v2/src/_base/utils/retry.ts b/packages/agent-core-v2/src/_base/utils/retry.ts index 898be2a66..f764b77ac 100644 --- a/packages/agent-core-v2/src/_base/utils/retry.ts +++ b/packages/agent-core-v2/src/_base/utils/retry.ts @@ -1,5 +1,5 @@ /** - * `_base` retry helpers — exponential backoff schedule, abortable retry + * `_base` retry helpers — exponential and server-directed backoff, abortable * sleeps, and error-field extraction shared by retry policies (the loop's * `stepRetry` plugin, full-compaction's self-managed resends). */ @@ -8,9 +8,10 @@ import { abortable } from '#/_base/utils/abort'; export const DEFAULT_MAX_RETRY_ATTEMPTS = 3; -const RETRY_MIN_TIMEOUT_MS = 300; -const RETRY_MAX_TIMEOUT_MS = 5000; +const BASE_DELAY_MS = 500; +const MAX_DELAY_MS = 32_000; const RETRY_FACTOR = 2; +const JITTER_FACTOR = 0.25; export interface RetryErrorFields { readonly errorName: string; @@ -19,13 +20,19 @@ export interface RetryErrorFields { } export function retryBackoffDelays(maxAttempts: number): number[] { - return Array.from({ length: Math.max(maxAttempts - 1, 0) }, (_unused, index) => { - const baseDelay = Math.min( - RETRY_MAX_TIMEOUT_MS, - RETRY_MIN_TIMEOUT_MS * RETRY_FACTOR ** index, - ); - return Math.round(baseDelay * (1 + Math.random())); - }); + const count = Math.max(maxAttempts - 1, 0); + const delays: number[] = []; + for (let i = 0; i < count; i += 1) { + const base = Math.min(BASE_DELAY_MS * Math.pow(RETRY_FACTOR, i), MAX_DELAY_MS); + delays.push(base + Math.random() * JITTER_FACTOR * base); + } + return delays; +} + +export function readRetryAfterMs(error: unknown): number | null { + if (typeof error !== 'object' || error === null) return null; + const value = (error as { retryAfterMs?: unknown }).retryAfterMs; + return typeof value === 'number' && value > 0 ? value : null; } export async function sleepForRetry(delayMs: number, signal?: AbortSignal): Promise { diff --git a/packages/agent-core-v2/src/agent/stepRetry/stepRetryService.ts b/packages/agent-core-v2/src/agent/stepRetry/stepRetryService.ts index 400a628df..5b01b150a 100644 --- a/packages/agent-core-v2/src/agent/stepRetry/stepRetryService.ts +++ b/packages/agent-core-v2/src/agent/stepRetry/stepRetryService.ts @@ -20,6 +20,7 @@ import { InstantiationType } from '#/_base/di/extensions'; import { LifecycleScope, registerScopedService } from '#/_base/di/scope'; import { DEFAULT_MAX_RETRY_ATTEMPTS, + readRetryAfterMs, retryBackoffDelays, retryErrorFields, sleepForRetry, @@ -95,7 +96,9 @@ export class AgentStepRetryService extends Disposable implements IAgentStepRetry return false; } - const delayMs = retryBackoffDelays(maxAttempts)[this.failedAttempts - 1] ?? 0; + const error = unwrapErrorCause(context.error); + const delayMs = + readRetryAfterMs(error) ?? retryBackoffDelays(maxAttempts)[this.failedAttempts - 1] ?? 0; this.eventBus.publish({ type: 'turn.step.retrying', turnId: context.turnId, @@ -105,7 +108,7 @@ export class AgentStepRetryService extends Disposable implements IAgentStepRetry nextAttempt: this.failedAttempts + 1, maxAttempts, delayMs, - ...retryErrorFields(unwrapErrorCause(context.error)), + ...retryErrorFields(error), }); await sleepForRetry(delayMs, context.signal); diff --git a/packages/agent-core-v2/src/app/llmProtocol/errors.ts b/packages/agent-core-v2/src/app/llmProtocol/errors.ts index 371f592e7..9657d6626 100644 --- a/packages/agent-core-v2/src/app/llmProtocol/errors.ts +++ b/packages/agent-core-v2/src/app/llmProtocol/errors.ts @@ -36,12 +36,19 @@ export class APITimeoutError extends ChatProviderError { export class APIStatusError extends ChatProviderError { readonly statusCode: number; readonly requestId: string | null; + readonly retryAfterMs: number | null; - constructor(statusCode: number, message: string, requestId?: string | null) { + constructor( + statusCode: number, + message: string, + requestId?: string | null, + retryAfterMs?: number | null, + ) { super(message); this.name = 'APIStatusError'; this.statusCode = statusCode; this.requestId = requestId ?? null; + this.retryAfterMs = retryAfterMs ?? null; } } @@ -50,8 +57,13 @@ export class APIStatusError extends ChatProviderError { * context window. */ export class APIContextOverflowError extends APIStatusError { - constructor(statusCode: number, message: string, requestId?: string | null) { - super(statusCode, message, requestId); + constructor( + statusCode: number, + message: string, + requestId?: string | null, + retryAfterMs?: number | null, + ) { + super(statusCode, message, requestId, retryAfterMs); this.name = 'APIContextOverflowError'; } } @@ -63,8 +75,13 @@ export class APIContextOverflowError extends APIStatusError { * size rejection is not — it needs media to be dropped or shrunk. */ export class APIRequestTooLargeError extends APIStatusError { - constructor(statusCode: number, message: string, requestId?: string | null) { - super(statusCode, message, requestId); + constructor( + statusCode: number, + message: string, + requestId?: string | null, + retryAfterMs?: number | null, + ) { + super(statusCode, message, requestId, retryAfterMs); this.name = 'APIRequestTooLargeError'; } } @@ -74,8 +91,8 @@ export class APIRequestTooLargeError extends APIStatusError { * request. */ export class APIProviderRateLimitError extends APIStatusError { - constructor(message: string, requestId?: string | null) { - super(429, message, requestId); + constructor(message: string, requestId?: string | null, retryAfterMs?: number | null) { + super(429, message, requestId, retryAfterMs); this.name = 'APIProviderRateLimitError'; } } @@ -87,8 +104,13 @@ export class APIProviderRateLimitError extends APIStatusError { * constraint, the provider is simply saturated — retry with backoff. */ export class APIProviderOverloadedError extends APIStatusError { - constructor(statusCode: number, message: string, requestId?: string | null) { - super(statusCode, message, requestId); + constructor( + statusCode: number, + message: string, + requestId?: string | null, + retryAfterMs?: number | null, + ) { + super(statusCode, message, requestId, retryAfterMs); this.name = 'APIProviderOverloadedError'; } } @@ -124,9 +146,10 @@ export function isRetryableGenerateError(error: unknown): boolean { if (error instanceof APIProviderOverloadedError) { return true; } - return ( - error instanceof APIStatusError && [429, 500, 502, 503, 504, 529].includes(error.statusCode) - ); + if (error instanceof APIStatusError) { + return [408, 409, 429, 500, 502, 503, 504, 529].includes(error.statusCode); + } + return error instanceof ChatProviderError; } const NETWORK_RE = /network|connection|connect|disconnect|terminated/i; @@ -200,22 +223,36 @@ export function normalizeAPIStatusError( statusCode: number, message: string, requestId?: string | null, + retryAfterMs?: number | null, ): APIStatusError { if (statusCode === 429) { - return new APIProviderRateLimitError(message, requestId); + return new APIProviderRateLimitError(message, requestId, retryAfterMs); } // Context overflow first: Vertex returns prompt-too-long as a 413, and a // token overflow must keep routing to compaction even on that status. if (isContextOverflowStatusError(statusCode, message)) { - return new APIContextOverflowError(statusCode, message, requestId); + return new APIContextOverflowError(statusCode, message, requestId, retryAfterMs); } if (isRequestTooLargeStatusError(statusCode, message)) { - return new APIRequestTooLargeError(statusCode, message, requestId); + return new APIRequestTooLargeError(statusCode, message, requestId, retryAfterMs); } if (isProviderOverloadStatusError(statusCode, message)) { - return new APIProviderOverloadedError(statusCode, message, requestId); + return new APIProviderOverloadedError(statusCode, message, requestId, retryAfterMs); } - return new APIStatusError(statusCode, message, requestId); + return new APIStatusError(statusCode, message, requestId, retryAfterMs); +} + +export function parseRetryAfterMs(headers: unknown): number | null { + const raw = + headers !== null && + typeof headers === 'object' && + typeof (headers as { get?: unknown }).get === 'function' + ? (headers as { get(name: string): string | null }).get('retry-after') + : null; + if (raw === null || raw === undefined) return null; + const seconds = Number.parseInt(raw, 10); + if (!Number.isFinite(seconds) || seconds < 0) return null; + return seconds * 1000; } export function isContextOverflowStatusError(statusCode: number, message: string): boolean { diff --git a/packages/agent-core-v2/src/app/llmProtocol/providers/anthropic.ts b/packages/agent-core-v2/src/app/llmProtocol/providers/anthropic.ts index f6fcfc845..ee64d0fa2 100644 --- a/packages/agent-core-v2/src/app/llmProtocol/providers/anthropic.ts +++ b/packages/agent-core-v2/src/app/llmProtocol/providers/anthropic.ts @@ -4,6 +4,7 @@ import { ChatProviderError, classifyBaseApiError, normalizeAPIStatusError, + parseRetryAfterMs, } from '../errors'; import type { ContentPart, Message, StreamedMessagePart, ToolCall } from '../message'; import { isToolDeclarationOnlyMessage } from '../message'; @@ -675,7 +676,12 @@ export function convertAnthropicError(error: unknown): ChatProviderError { // APIError with a status code => status error if (error instanceof AnthropicAPIError && typeof error.status === 'number') { const reqId = error.requestID ?? null; - return normalizeAPIStatusError(error.status, error.message, reqId); + return normalizeAPIStatusError( + error.status, + error.message, + reqId, + parseRetryAfterMs(error.headers), + ); } if (error instanceof AnthropicError) { return new ChatProviderError(`Anthropic error: ${error.message}`); diff --git a/packages/agent-core-v2/src/app/llmProtocol/providers/openai-common.ts b/packages/agent-core-v2/src/app/llmProtocol/providers/openai-common.ts index 25e18e52d..ce42dfe75 100644 --- a/packages/agent-core-v2/src/app/llmProtocol/providers/openai-common.ts +++ b/packages/agent-core-v2/src/app/llmProtocol/providers/openai-common.ts @@ -4,6 +4,7 @@ import { ChatProviderError, classifyBaseApiError, normalizeAPIStatusError, + parseRetryAfterMs, } from '../errors'; import { extractText } from '../message'; import type { ContentPart, Message } from '../message'; @@ -103,7 +104,12 @@ export function convertOpenAIError(error: unknown): ChatProviderError { // APIError with a status code => status error if (error instanceof OpenAIAPIError && typeof error.status === 'number') { const reqId = error.requestID ?? null; - return normalizeAPIStatusError(error.status, error.message, reqId); + return normalizeAPIStatusError( + error.status, + error.message, + reqId, + parseRetryAfterMs(error.headers), + ); } // Base APIError with no status and no body => transport-layer failure. // When the error has a body (e.g. SSE error events from the server), diff --git a/packages/agent-core-v2/src/app/llmProtocol/providers/openai-responses.ts b/packages/agent-core-v2/src/app/llmProtocol/providers/openai-responses.ts index f392dacd8..4ce3f9db9 100644 --- a/packages/agent-core-v2/src/app/llmProtocol/providers/openai-responses.ts +++ b/packages/agent-core-v2/src/app/llmProtocol/providers/openai-responses.ts @@ -235,6 +235,13 @@ function formatResponsesErrorEvent( return `${codeText}: ${message}${paramText}`; } +const EMBEDDED_STATUS_CODE_RE = /\bstatus_code\s*[:=]\s*(\d{3})\b/; + +function readEmbeddedStatusCode(message: string): number | undefined { + const match = EMBEDDED_STATUS_CODE_RE.exec(message); + return match === null ? undefined : Number(match[1]); +} + function errorFromOpenAIResponsesEvent( prefix: string, code: string | null, @@ -246,7 +253,7 @@ function errorFromOpenAIResponsesEvent( if (isContextOverflowErrorCode(code)) { return new APIContextOverflowError(400, fullMessage); } - if (code === 'rate_limit_exceeded') { + if (code === 'rate_limit_exceeded' || readEmbeddedStatusCode(message) === 429) { return new APIProviderRateLimitError(fullMessage); } return new ChatProviderError(fullMessage); diff --git a/packages/agent-core-v2/src/session/swarm/sessionSwarmService.ts b/packages/agent-core-v2/src/session/swarm/sessionSwarmService.ts index d54610baf..e7d3e3053 100644 --- a/packages/agent-core-v2/src/session/swarm/sessionSwarmService.ts +++ b/packages/agent-core-v2/src/session/swarm/sessionSwarmService.ts @@ -189,14 +189,16 @@ export class SessionSwarmService implements ISessionSwarmService { this.realignChildModel(caller, child); const profileName = child.accessor.get(IAgentProfileService).data().profileName ?? RESUMED_PROFILE_FALLBACK; - emitAgentRunSpawned(caller, agentId, { - profileName, - parentToolCallId: options.parentToolCallId, - parentToolCallUuid: options.parentToolCallUuid, - description: options.description, - swarmIndex: options.swarmIndex, - runInBackground: options.runInBackground, - }); + if (!retryTurn) { + emitAgentRunSpawned(caller, agentId, { + profileName, + parentToolCallId: options.parentToolCallId, + parentToolCallUuid: options.parentToolCallUuid, + description: options.description, + swarmIndex: options.swarmIndex, + runInBackground: options.runInBackground, + }); + } const request = retryTurn ? ({ kind: 'retry' } as const) : ({ kind: 'prompt', prompt: options.prompt } as const); diff --git a/packages/agent-core-v2/test/agent/stepRetry/stepRetry.test.ts b/packages/agent-core-v2/test/agent/stepRetry/stepRetry.test.ts index 8d4f1d4d8..2a7706c84 100644 --- a/packages/agent-core-v2/test/agent/stepRetry/stepRetry.test.ts +++ b/packages/agent-core-v2/test/agent/stepRetry/stepRetry.test.ts @@ -1,8 +1,13 @@ import { afterEach, describe, expect, it, vi } from 'vitest'; -import { APIConnectionError, APIStatusError } from '#/app/llmProtocol/errors'; +import { + APIConnectionError, + APIProviderRateLimitError, + APIStatusError, +} from '#/app/llmProtocol/errors'; import { emptyUsage } from '#/app/llmProtocol/usage'; import { IEventBus } from '#/app/event/eventBus'; +import { retryBackoffDelays } from '#/_base/utils/retry'; import { IAgentLoopService } from '#/agent/loop/loop'; import { ContinuationStepRequest } from '#/agent/loop/stepRequest'; @@ -114,6 +119,39 @@ describe('stepRetry plugin', () => { ]); }); + it('honors the provider retry-after delay before retrying', async () => { + let calls = 0; + ctx = createTestAgent( + llmGenerateServices(async () => { + calls += 1; + if (calls === 1) throw new APIProviderRateLimitError('slow down', null, 1); + return { + id: 'retry-after-response', + message: { + role: 'assistant', + content: [{ type: 'text', text: 'recovered' }], + toolCalls: [], + }, + usage: emptyUsage(), + finishReason: 'completed', + rawFinishReason: 'stop', + }; + }), + ); + + ctx.get(IEventBus).publish({ type: 'turn.started', turnId: 1, origin: { kind: 'user' } }); + const loop = ctx.get(IAgentLoopService); + loop.enqueue(new ContinuationStepRequest()); + const result = await loop.run({ turnId: 1 }); + + expect(result.type).toBe('completed'); + expect(rpcEvents('turn.step.retrying')).toEqual([ + expect.objectContaining({ + args: expect.objectContaining({ delayMs: 1 }), + }), + ]); + }); + it('does not retry a non-retryable error', async () => { vi.useFakeTimers(); let calls = 0; @@ -198,3 +236,24 @@ describe('stepRetry plugin', () => { expect(second).toEqual({ type: 'completed', steps: 1, truncated: false }); }); }); + +describe('retryBackoffDelays', () => { + it('starts at 500 milliseconds and doubles with up to 25 percent jitter', () => { + const delays = retryBackoffDelays(3); + + expect(delays[0]).toBeGreaterThanOrEqual(500); + expect(delays[0]).toBeLessThanOrEqual(625); + expect(delays[1]).toBeGreaterThanOrEqual(1_000); + expect(delays[1]).toBeLessThanOrEqual(1_250); + }); + + it('caps high-attempt backoff at 32 seconds plus up to 25 percent jitter', () => { + const delays = retryBackoffDelays(10); + + expect(delays).toHaveLength(9); + expect(delays[6]).toBeGreaterThanOrEqual(32_000); + expect(delays[6]).toBeLessThanOrEqual(40_000); + expect(delays[8]).toBeGreaterThanOrEqual(32_000); + expect(delays[8]).toBeLessThanOrEqual(40_000); + }); +}); diff --git a/packages/agent-core-v2/test/app/llmProtocol/errors.test.ts b/packages/agent-core-v2/test/app/llmProtocol/errors.test.ts index f116707dd..634880700 100644 --- a/packages/agent-core-v2/test/app/llmProtocol/errors.test.ts +++ b/packages/agent-core-v2/test/app/llmProtocol/errors.test.ts @@ -1,3 +1,8 @@ +/** + * `llmProtocol` error contract — provider error classification, normalization, + * and retry metadata shared by generation and swarm recovery. + */ + import { APIConnectionError, APIContextOverflowError, @@ -13,6 +18,7 @@ import { isRetryableGenerateError, isToolExchangeAdjacencyError, normalizeAPIStatusError, + parseRetryAfterMs, } from '#/app/llmProtocol/errors'; import { describe, expect, it } from 'vitest'; @@ -68,6 +74,15 @@ describe('APIStatusError', () => { expect(err.statusCode).toBe(502); expect(err.requestId).toBeNull(); }); + + it('preserves a provider-requested retry delay', () => { + const err = new APIStatusError(429, 'rate limited', 'req-abc', 12_500); + expect(err.retryAfterMs).toBe(12_500); + }); + + it('defaults the provider-requested retry delay to null', () => { + expect(new APIStatusError(429, 'rate limited').retryAfterMs).toBeNull(); + }); }); describe('APIEmptyResponseError', () => { @@ -149,9 +164,12 @@ describe('isRetryableGenerateError', () => { expect(isRetryableGenerateError(new APIEmptyResponseError('empty'))).toBe(true); }); - it.each([429, 500, 502, 503, 504, 529])('treats HTTP %i as retryable', (statusCode) => { - expect(isRetryableGenerateError(new APIStatusError(statusCode, 'retryable'))).toBe(true); - }); + it.each([408, 409, 429, 500, 502, 503, 504, 529])( + 'treats HTTP %i as retryable', + (statusCode) => { + expect(isRetryableGenerateError(new APIStatusError(statusCode, 'retryable'))).toBe(true); + }, + ); it('treats provider overload as retryable', () => { expect(isRetryableGenerateError(new APIProviderOverloadedError(529, 'Overloaded'))).toBe(true); @@ -171,6 +189,10 @@ describe('isRetryableGenerateError', () => { expect(isRetryableGenerateError(new Error('boom'))).toBe(false); expect(isRetryableGenerateError('boom')).toBe(false); }); + + it('retries an unclassified provider error as a transient fallback', () => { + expect(isRetryableGenerateError(new ChatProviderError('upstream failure'))).toBe(true); + }); }); describe('error hierarchy instanceof checks', () => { @@ -216,6 +238,11 @@ describe('normalizeAPIStatusError', () => { expect(error.requestId).toBe('req-rate'); }); + it('propagates the provider-requested retry delay through normalization', () => { + const error = normalizeAPIStatusError(429, 'Too many requests', 'req-rate', 7_000); + expect(error.retryAfterMs).toBe(7_000); + }); + it.each([ [400, 'Context length exceeded'], [400, 'Exceeded max tokens'], @@ -325,6 +352,24 @@ describe('normalizeAPIStatusError', () => { }); }); +describe('parseRetryAfterMs', () => { + it('converts integer retry-after seconds to milliseconds', () => { + expect(parseRetryAfterMs(new Headers({ 'retry-after': '12' }))).toBe(12_000); + }); + + it('ignores an HTTP-date retry-after value', () => { + expect( + parseRetryAfterMs(new Headers({ 'retry-after': 'Wed, 21 Oct 2026 07:28:00 GMT' })), + ).toBeNull(); + }); + + it('ignores missing or malformed header containers', () => { + expect(parseRetryAfterMs(new Headers())).toBeNull(); + expect(parseRetryAfterMs({})).toBeNull(); + expect(parseRetryAfterMs(null)).toBeNull(); + }); +}); + describe('isToolExchangeAdjacencyError', () => { // The exact Anthropic message observed in the field when a tool_use was not // immediately followed by its tool_result. diff --git a/packages/agent-core-v2/test/app/llmProtocol/providers/provider-errors.test.ts b/packages/agent-core-v2/test/app/llmProtocol/providers/provider-errors.test.ts new file mode 100644 index 000000000..f1d4bc7b0 --- /dev/null +++ b/packages/agent-core-v2/test/app/llmProtocol/providers/provider-errors.test.ts @@ -0,0 +1,71 @@ +/** + * `llmProtocol` provider boundary contract — SDK and streamed API failures + * retain rate-limit classification and server-directed retry metadata. + */ + +import { APIError as AnthropicAPIError } from '@anthropic-ai/sdk'; +import { APIError as OpenAIAPIError } from 'openai'; +import { describe, expect, it } from 'vitest'; + +import { APIProviderRateLimitError } from '#/app/llmProtocol/errors'; +import { convertAnthropicError } from '#/app/llmProtocol/providers/anthropic'; +import { convertOpenAIError } from '#/app/llmProtocol/providers/openai-common'; +import { OpenAIResponsesStreamedMessage } from '#/app/llmProtocol/providers/openai-responses'; + +async function* streamEvents(events: readonly Record[]) { + yield* events; +} + +async function consume(stream: AsyncIterable): Promise { + for await (const _part of stream) { + void _part; + } +} + +describe('provider retry-after conversion', () => { + it('preserves OpenAI retry-after seconds on a rate-limit error', () => { + const source = new OpenAIAPIError( + 429, + undefined, + 'Too many requests', + new Headers({ 'retry-after': '12' }), + ); + + const error = convertOpenAIError(source); + + expect(error).toBeInstanceOf(APIProviderRateLimitError); + expect((error as APIProviderRateLimitError).retryAfterMs).toBe(12_000); + }); + + it('preserves Anthropic retry-after seconds on a rate-limit error', () => { + const source = AnthropicAPIError.generate( + 429, + { type: 'error', error: { type: 'rate_limit_error', message: 'rate limited' } }, + 'rate limited', + new Headers({ 'retry-after': '7' }), + ); + + const error = convertAnthropicError(source); + + expect(error).toBeInstanceOf(APIProviderRateLimitError); + expect((error as APIProviderRateLimitError).retryAfterMs).toBe(7_000); + }); +}); + +describe('OpenAI Responses rate-limit conversion', () => { + it('promotes an embedded status_code=429 stream error to a rate-limit error', async () => { + const stream = new OpenAIResponsesStreamedMessage( + streamEvents([ + { + type: 'error', + code: 'upstream_error', + message: 'example.test/responses/response.json status_code=429', + param: null, + }, + ]), + true, + ); + + await expect(consume(stream)).rejects.toBeInstanceOf(APIProviderRateLimitError); + }); +}); diff --git a/packages/agent-core-v2/test/session/swarm/sessionSwarm.test.ts b/packages/agent-core-v2/test/session/swarm/sessionSwarm.test.ts index 6023f7f37..90405d32e 100644 --- a/packages/agent-core-v2/test/session/swarm/sessionSwarm.test.ts +++ b/packages/agent-core-v2/test/session/swarm/sessionSwarm.test.ts @@ -1090,6 +1090,69 @@ describe('SessionSwarmService metadata compatibility', () => { ); }); + it('does not emit spawned again when a rate-limited child retries', async () => { + vi.useFakeTimers(); + try { + agents['agent-retry'] = { + homedir: '/tmp/kimi/s1/agents/agent-retry', + labels: { parentAgentId: 'main' }, + }; + agents['agent-blocker'] = { + homedir: '/tmp/kimi/s1/agents/agent-blocker', + labels: { parentAgentId: 'main' }, + }; + handles.set('agent-retry', agentHandle('agent-retry', lifecycle, eventBus)); + handles.set('agent-blocker', agentHandle('agent-blocker', lifecycle, eventBus)); + const rateLimited = createControlledPromise<{ summary: string }>(); + const blocker = createControlledPromise<{ summary: string }>(); + const published: DomainEvent[] = []; + (eventBus.publish as ReturnType).mockImplementation((event: DomainEvent) => { + published.push(event); + }); + let retryRuns = 0; + runAgent.mockImplementation((agentId, request, options) => { + options?.onReady?.(); + if (agentId === 'agent-retry') { + retryRuns += 1; + return { + agentId, + turn: {} as never, + completion: + retryRuns === 1 + ? rateLimited + : Promise.resolve({ summary: 'recovered summary' }), + }; + } + return { agentId, turn: {} as never, completion: blocker }; + }); + const service = ix.get(ISessionSwarmService); + + const running = service.run({ + callerAgentId: 'main', + tasks: [resumeSessionTask('agent-retry'), resumeSessionTask('agent-blocker')], + }); + await vi.advanceTimersByTimeAsync(0); + rateLimited.reject(new APIProviderRateLimitError('Rate limited')); + await vi.advanceTimersByTimeAsync(0); + blocker.resolve({ summary: 'blocker summary' }); + await vi.advanceTimersByTimeAsync(3_000); + await running; + + expect( + published + .filter((event) => event.type === 'subagent.spawned') + .map((event) => event.subagentId), + ).toEqual(['agent-retry', 'agent-blocker']); + expect( + runAgent.mock.calls + .filter(([agentId]) => agentId === 'agent-retry') + .map(([, request]) => request), + ).toEqual([{ kind: 'prompt', prompt: 'Continue' }, { kind: 'retry' }]); + } finally { + vi.useRealTimers(); + } + }); + it('rejects resume of an already running child before launching or emitting spawned', async () => { agents['agent-existing'] = { homedir: '/tmp/kimi/s1/agents/agent-existing',