fix(agent-core-v2): disable SDK-internal retries that blocked cancellation (#2855)

* fix(agent-core-v2): disable SDK-internal retries that blocked cancellation

The OpenAI and Anthropic SDK clients default to maxRetries=2 with a
backoff sleep that never observes the request AbortSignal, so Ctrl+C
during a 429/5xx/connection-error retry only took effect after the
sleep elapsed, and the hidden attempts were invisible to the engine
(no turn.step.retrying) while double-counting its retry budget.

Build those clients with maxRetries: 0 so retryable failures surface
to the engine's step-retry layer immediately (observable countdown,
abortable sleep, single retry budget). The Google GenAI main request
path only retries when httpOptions.retryOptions is explicitly set, so
there is nothing to disable; instead its error converter now recovers
the server-directed delay from the wire body's google.rpc.RetryInfo
detail, since the SDK's ApiError drops the Retry-After header.

* chore: simplify the retry-cancellation changeset entry

* fix(agent-core-v2): recover GenAI retry delay from prefixed mid-stream error chunks

Mid-stream error chunks throw ApiError with the message wrapped as
"got status: <STATUS>. {json}", so a strict JSON.parse of the whole
message missed the google.rpc.RetryInfo detail. Locate the JSON object
start before parsing; the non-stream path (pure JSON body) is
unaffected.
This commit is contained in:
liruifengv 2026-08-12 20:40:20 +08:00 committed by GitHub
parent ec84a6f9a3
commit 30f56a2d2d
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
7 changed files with 246 additions and 1 deletions

View file

@ -0,0 +1,5 @@
---
"@moonshot-ai/kimi-code": patch
---
Fix Ctrl+C being ignored during automatic retries of failed API requests.

View file

@ -20,6 +20,11 @@
* the trait-composed `convertError` hook consulted, so a vendor riding this
* transport classifies each RAW SDK failure exactly once before the base
* rules run.
*
* The SDK client is built with `maxRetries: 0`: the SDK's internal backoff
* sleep never observes the turn's AbortSignal, so rate-limit / server /
* connection retry is owned by the engine's step-retry layer (observable and
* cancellable), never by the SDK.
*/
import Anthropic, {
@ -1148,6 +1153,7 @@ export class AnthropicChatProvider implements ChatProvider {
authToken: null,
baseURL: this._baseUrl ?? null,
defaultHeaders: this._buildDefaultHeaders(apiKey),
maxRetries: 0,
});
}
}

View file

@ -11,6 +11,11 @@
* module's abort plumbing (abortPromise racing,
* per-chunk checks, the catch guard that rethrows DOMException aborts before
* error conversion) is self-contained by design.
*
* Error conversion recovers the server-directed retry delay from the wire
* body: the SDK's `ApiError` drops response headers, so the
* `google.rpc.RetryInfo` detail inside the stringified error body is the
* only carrier of that wait time.
*/
import { ApiError as GoogleApiError, GoogleGenAI as GenAIClient } from '@google/genai';
@ -626,7 +631,12 @@ const TIMEOUT_RE = /timed?\s*out|timeout|deadline/i;
export function convertGoogleGenAIError(error: unknown): ChatProviderError {
if (error instanceof GoogleApiError) {
return normalizeAPIStatusError(error.status, error.message);
return normalizeAPIStatusError(
error.status,
error.message,
undefined,
parseRetryInfoDelayMs(error.message),
);
}
if (error instanceof Error) {
const msg = error.message;
@ -645,6 +655,32 @@ export function convertGoogleGenAIError(error: unknown): ChatProviderError {
return new ChatProviderError(`GoogleGenAI error: ${String(error)}`);
}
function parseRetryInfoDelayMs(message: string): number | null {
const jsonStart = message.indexOf('{');
if (jsonStart < 0) return null;
try {
const body: unknown = JSON.parse(message.slice(jsonStart));
if (typeof body !== 'object' || body === null) return null;
const details = (body as { error?: { details?: unknown } }).error?.details;
if (!Array.isArray(details)) return null;
for (const detail of details) {
if (typeof detail !== 'object' || detail === null) continue;
const type = (detail as { '@type'?: unknown })['@type'];
if (typeof type !== 'string' || !type.endsWith('google.rpc.RetryInfo')) continue;
const retryDelay = (detail as { retryDelay?: unknown }).retryDelay;
if (typeof retryDelay !== 'string') continue;
const match = /^(\d+(?:\.\d+)?)s$/.exec(retryDelay.trim());
if (match?.[1] === undefined) continue;
const seconds = Number.parseFloat(match[1]);
if (!Number.isFinite(seconds) || seconds < 0) continue;
return Math.round(seconds * 1000);
}
return null;
} catch {
return null;
}
}
export class GoogleGenAIChatProvider implements ChatProvider {
readonly name: string = 'google_genai';

View file

@ -22,6 +22,11 @@
* tool-result `extract_text` fallback and tool-declaration-only skip are
* handed over to the trait wholesale: every history message is
* base-converted, post-processed by the hook, and dropped on `null`.
*
* The SDK client is built with `maxRetries: 0`: the SDK's internal backoff
* sleep never observes the turn's AbortSignal, so rate-limit / server /
* connection retry is owned by the engine's step-retry layer (observable and
* cancellable), never by the SDK.
*/
import OpenAI from 'openai';
@ -751,6 +756,7 @@ export class OpenAILegacyChatProvider implements ChatProvider {
const clientOpts: Record<string, unknown> = {
apiKey,
baseURL: this._baseUrl,
maxRetries: 0,
};
const defaultHeaders = mergeRequestHeaders(this._defaultHeaders, auth?.headers);
if (defaultHeaders !== undefined) {

View file

@ -11,6 +11,11 @@
* classification (already-converted errors crossing an outer catch pass
* through without re-consulting). The developer-role model detection lives
* here.
*
* The SDK client is built with `maxRetries: 0`: the SDK's internal backoff
* sleep never observes the turn's AbortSignal, so rate-limit / server /
* connection retry is owned by the engine's step-retry layer (observable and
* cancellable), never by the SDK.
*/
import OpenAI from 'openai';
@ -1203,6 +1208,7 @@ export class OpenAIResponsesChatProvider implements ChatProvider {
const clientOpts: Record<string, unknown> = {
apiKey,
baseURL: this._baseUrl,
maxRetries: 0,
};
const defaultHeaders = mergeRequestHeaders(this._defaultHeaders, auth?.headers);
if (defaultHeaders !== undefined) {

View file

@ -33,11 +33,19 @@
* profile, and the OpenAI `reasoning_effort` auto-enable with its
* load-bearing kill switch (a `withThinking` hook disables it).
*
* Plus one construction invariant: every wire base builds its SDK client
* with `maxRetries: 0` retry is owned by the engine's step-retry layer,
* never by the SDK (whose backoff sleep ignores the turn's AbortSignal). The
* closing section proves it over a real HTTP 429: the first response reaches
* the caller after exactly one request, carrying the server-directed delay.
*
* Note: base/definition registries are module-level state shared across this
* file, so the contribs and test-vendor definitions are imported/registered
* exactly once here.
*/
import { createServer } from 'node:http';
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
import { APIError as AnthropicAPIError } from '@anthropic-ai/sdk';
@ -47,6 +55,7 @@ import {
APIConnectionError,
APIProviderQuotaExhaustedError,
APIProviderRateLimitError,
APIStatusError,
isRetryableGenerateError,
} from '#/kosong/contract/errors';
import type { Message } from '#/kosong/contract/message';
@ -344,6 +353,17 @@ describe('createChatProvider', () => {
});
});
describe('SDK-internal retry disabled (engine-owned step retry)', () => {
it.each([
{ protocol: 'openai', modelName: 'gpt-4o' },
{ protocol: 'openai_responses', modelName: 'gpt-5' },
{ protocol: 'anthropic', modelName: 'claude-opus-4-6' },
] as const)('builds the $protocol SDK client with maxRetries 0', ({ protocol, modelName }) => {
const provider = registry.createChatProvider({ protocol, modelName, apiKey: 'sk-probe' });
expect((sdkClient(provider) as { maxRetries?: number }).maxRetries).toBe(0);
});
});
describe('google-genai vertex mode (providerOptions)', () => {
it('forwards vertexai + project + location from providerOptions to the base', () => {
const provider = registry.createChatProvider({
@ -1234,3 +1254,88 @@ describe('OpenAI reasoning_effort path (issue #1616)', () => {
expect(explicit['reasoning_effort']).toBe('low');
});
});
describe('429 wire behavior over real HTTP (no hidden SDK retry)', () => {
async function with429Server(
body: Record<string, unknown>,
run: (port: number, requestCount: () => number) => Promise<void>,
): Promise<void> {
let count = 0;
const server = createServer((_req, res) => {
count += 1;
res.writeHead(429, { 'content-type': 'application/json', 'retry-after': '5' });
res.end(JSON.stringify(body));
});
await new Promise<void>((resolve) => {
server.listen(0, '127.0.0.1', resolve);
});
try {
const address = server.address();
if (address === null || typeof address === 'string') {
throw new Error('server has no address');
}
await run(address.port, () => count);
} finally {
await new Promise<void>((resolve) => {
server.close(() => {
resolve();
});
});
}
}
it.each([
{
protocol: 'openai',
modelName: 'gpt-4o',
baseUrlPath: '/v1',
body: { error: { message: 'slow down', type: 'rate_limit_error' } },
},
{
protocol: 'openai_responses',
modelName: 'gpt-5',
baseUrlPath: '/v1',
body: { error: { message: 'slow down', type: 'rate_limit_error' } },
},
{
protocol: 'anthropic',
modelName: 'claude-opus-4-6',
baseUrlPath: '',
body: { type: 'error', error: { type: 'rate_limit_error', message: 'slow down' } },
},
{
protocol: 'google-genai',
modelName: 'gemini-2.5-flash',
baseUrlPath: '',
body: {
error: {
code: 429,
message: 'Resource exhausted',
status: 'RESOURCE_EXHAUSTED',
details: [{ '@type': 'type.googleapis.com/google.rpc.RetryInfo', retryDelay: '5s' }],
},
},
},
] as const)(
'the first 429 reaches the caller after exactly one request with the 5s server delay ($protocol)',
async ({ protocol, modelName, baseUrlPath, body }) => {
await with429Server(body, async (port, requestCount) => {
const provider = registry.createChatProvider({
protocol,
modelName,
apiKey: 'sk-probe',
baseUrl: `http://127.0.0.1:${String(port)}${baseUrlPath}`,
});
const rejected: unknown = await provider.generate('sys', [], PROBE_HISTORY).then(
() => {
throw new Error('expected generate to reject');
},
(error: unknown) => error,
);
expect(rejected).toBeInstanceOf(APIProviderRateLimitError);
expect((rejected as APIStatusError).retryAfterMs).toBe(5000);
expect(requestCount()).toBe(1);
});
},
);
});

View file

@ -17,9 +17,15 @@
* last-declarer-wins semantics and consulted after the abort guard with
* the raw failure at every catch seam, including the Responses in-stream
* error-event path.
*
* The Google GenAI converter coverage checks the recovery of the
* server-directed retry delay from the wire body's `google.rpc.RetryInfo`
* detail the SDK's `ApiError` drops response headers, so the body detail
* is the only carrier of that wait time.
*/
import { APIError as AnthropicAPIError } from '@anthropic-ai/sdk';
import { ApiError as GoogleApiError } from '@google/genai';
import { APIError as OpenAIAPIError } from 'openai';
import { describe, expect, it } from 'vitest';
@ -34,6 +40,7 @@ import {
import type { ProtocolAdapterConfig } from '#/kosong/protocol/protocol';
import { traitConvertError, type TraitContext } from '#/kosong/protocol/protocolTrait';
import { convertAnthropicError } from '#/kosong/provider/bases/anthropic/anthropic';
import { convertGoogleGenAIError } from '#/kosong/provider/bases/google-genai/google-genai';
import { convertOpenAIError } from '#/kosong/provider/bases/openai/openai-common';
import { OpenAIResponsesStreamedMessage } from '#/kosong/provider/bases/openai/openai-responses';
import { composeOpenAIChatHooks } from '#/kosong/provider/bases/openai/openaiHooks';
@ -301,3 +308,77 @@ describe('OpenAI Responses quota-exhausted conversion', () => {
expect(seen[0]).toMatchObject({ type: 'error', code: 'vendor_quota_gone' });
});
});
describe('convertGoogleGenAIError RetryInfo recovery', () => {
function googleApiError(status: number, body: unknown): GoogleApiError {
return new GoogleApiError({ message: JSON.stringify(body), status });
}
it('recovers the server-directed retry delay from the RetryInfo detail', () => {
const error = convertGoogleGenAIError(
googleApiError(429, {
error: {
code: 429,
message: 'Resource exhausted',
status: 'RESOURCE_EXHAUSTED',
details: [
{ '@type': 'type.googleapis.com/google.rpc.QuotaFailure', violations: [] },
{ '@type': 'type.googleapis.com/google.rpc.RetryInfo', retryDelay: '5s' },
],
},
}),
);
expect(error).toBeInstanceOf(APIProviderRateLimitError);
expect((error as APIStatusError).retryAfterMs).toBe(5000);
expect(isRetryableGenerateError(error)).toBe(true);
});
it('parses fractional proto Duration delays', () => {
const error = convertGoogleGenAIError(
googleApiError(429, {
error: {
details: [
{ '@type': 'type.googleapis.com/google.rpc.RetryInfo', retryDelay: '5.5s' },
],
},
}),
);
expect((error as APIStatusError).retryAfterMs).toBe(5500);
});
it('keeps a 429 without RetryInfo a retryable rate limit with no server delay', () => {
const error = convertGoogleGenAIError(
googleApiError(429, { error: { code: 429, message: 'Too many requests' } }),
);
expect(error).toBeInstanceOf(APIProviderRateLimitError);
expect((error as APIStatusError).retryAfterMs).toBeNull();
expect(isRetryableGenerateError(error)).toBe(true);
});
it('tolerates a non-JSON ApiError message', () => {
const error = convertGoogleGenAIError(
new GoogleApiError({ message: 'Too many requests', status: 429 }),
);
expect(error).toBeInstanceOf(APIProviderRateLimitError);
expect((error as APIStatusError).retryAfterMs).toBeNull();
});
it('recovers the delay from a mid-stream error chunk carrying the "got status" prefix', () => {
const chunk = {
error: {
code: 429,
message: 'Resource exhausted',
status: 'RESOURCE_EXHAUSTED',
details: [{ '@type': 'type.googleapis.com/google.rpc.RetryInfo', retryDelay: '5s' }],
},
};
const error = convertGoogleGenAIError(
new GoogleApiError({
message: `got status: RESOURCE_EXHAUSTED. ${JSON.stringify(chunk)}`,
status: 429,
}),
);
expect(error).toBeInstanceOf(APIProviderRateLimitError);
expect((error as APIStatusError).retryAfterMs).toBe(5000);
});
});