diff --git a/src/core/proxy.ts b/src/core/proxy.ts index 5757e97d..046628d6 100644 --- a/src/core/proxy.ts +++ b/src/core/proxy.ts @@ -3,7 +3,7 @@ * Adapted by src/node.ts and src/worker.ts; uses only Request/Response/URL/fetch. */ -import { markCacheDead, noteCacheOutcome, responseLeftNoCache } from './session-state.js'; +import { markCacheDead, noteCacheOutcome, noteRateLimitOutcome, responseLeftNoCache } from './session-state.js'; import { transformRequest, type TransformOptions, type TransformInfo } from './transform.js'; import { isClaudeModel, transformOpenAIChatCompletions, transformOpenAIResponses } from './openai.js'; import { isAnthropicMessagesPath, isPxpipeSupportedGptModel, isPxpipeSupportedModel } from './applicability.js'; @@ -2054,6 +2054,10 @@ let teed: Response; usage?.cache_read_input_tokens, usage?.cache_creation_input_tokens, ); + // Track consecutive 429s per session so a session stuck behind a rate + // limit stops re-imaging the same doomed bytes — see + // isRateLimitCircuitOpen. + noteRateLimitOutcome(info?.firstUserSha8, upstreamRes.status); fire( upstreamRes.status, info, diff --git a/src/core/session-state.ts b/src/core/session-state.ts index 0d23ceb1..97020690 100644 --- a/src/core/session-state.ts +++ b/src/core/session-state.ts @@ -96,8 +96,19 @@ interface SessionRecord { * that once had one, and then lost it, has provably free room to re-cut. */ everCacheAlive?: boolean; + /** + * Consecutive 429s for this session, reset on any non-429 response. See + * {@link isRateLimitCircuitOpen}. + */ + consecutive429: number; } +/** + * Consecutive 429s after which a session stops being imaged. See + * {@link isRateLimitCircuitOpen}. + */ +const RATE_LIMIT_CIRCUIT_THRESHOLD = 3; + const sessions = new Map(); function touch(key: string): SessionRecord { @@ -107,7 +118,7 @@ function touch(key: string): SessionRecord { sessions.set(key, existing); return existing; } - const fresh: SessionRecord = { lastSeenMs: 0, freezeStep: 0, cacheDead: false }; + const fresh: SessionRecord = { lastSeenMs: 0, freezeStep: 0, cacheDead: false, consecutive429: 0 }; sessions.set(key, fresh); while (sessions.size > SESSIONS_MAX) { const oldest = sessions.keys().next().value; @@ -249,6 +260,51 @@ export function responseLeftNoCache(status: number, errorBody?: string): boolean return false; } +/** + * Feed a response's status back in for rate-limit tracking. Call once per + * response, alongside {@link noteCacheOutcome}. + * + * A 429 on a request whose bytes are unchanged from the last attempt (retried + * verbatim by the client) means imaging bought nothing: the provider never got + * far enough to populate a fresh prefix cache, the retry re-sends the same + * imaged bytes, and it fails the same way. Three in a row is past "unlucky + * timing" and into "this session cannot get a compressed request through + * right now" — see {@link isRateLimitCircuitOpen}. + * + * Any other status clears the counter: a single retry that gets through means + * whatever was throttling the account has room again. + */ +export function noteRateLimitOutcome(sessionKey: string | undefined, status: number): void { + if (!sessionKey) return; + // touch(), not a get-or-skip: unlike noteCacheOutcome (which only refines a + // record transformRequest is guaranteed to have created earlier in the same + // call), this is the sole writer for consecutive429 and must not silently + // no-op if some other response path reaches here first. + const rec = touch(sessionKey); + rec.consecutive429 = status === 429 ? rec.consecutive429 + 1 : 0; +} + +/** + * Has this session hit {@link RATE_LIMIT_CIRCUIT_THRESHOLD} consecutive 429s? + * Callers use this to skip imaging and send plain text instead, so a session + * stuck behind a rate limit stops re-sending the same doomed imaged bytes and + * gets a chance to complete a request — which is also the only way its own + * prefix cache (see {@link noteCacheOutcome}) gets a chance to come back. + * + * Deliberately has no time-based reset: a session that tripped this is still + * the same session next request, and a fixed cooldown would either reopen too + * early (back to the same failure) or too late (stuck on text past the point + * the rate limit cleared) with no way to know which. Bounded instead by the + * session's own natural lifetime — 512 sessions tracked, oldest evicted first + * (see `SESSIONS_MAX`) — and by the fact that a session back under budget just + * needs one successful request of any kind to clear the counter. + */ +export function isRateLimitCircuitOpen(sessionKey: string | undefined): boolean { + if (!sessionKey) return false; + const rec = sessions.get(sessionKey); + return (rec?.consecutive429 ?? 0) >= RATE_LIMIT_CIRCUIT_THRESHOLD; +} + /** Test seam: drop all session state. */ export function resetSessionState(): void { sessions.clear(); @@ -257,7 +313,7 @@ export function resetSessionState(): void { /** Test/telemetry seam: inspect a session without mutating its clock. */ export function peekSessionState( sessionKey: string, -): { lastSeenMs: number; freezeStep: number; cacheDead: boolean } | undefined { +): { lastSeenMs: number; freezeStep: number; cacheDead: boolean; consecutive429: number } | undefined { const rec = sessions.get(sessionKey); return rec ? { ...rec } : undefined; } diff --git a/src/core/transform.ts b/src/core/transform.ts index 82f5feff..845b94ad 100644 --- a/src/core/transform.ts +++ b/src/core/transform.ts @@ -51,7 +51,7 @@ import { ANTHROPIC_MAX_IMAGES, ANTHROPIC_HISTORY_IMAGE_BUDGET, } from './history.js'; -import { noteHistoryRequest, recordFreezeStep } from './session-state.js'; +import { isRateLimitCircuitOpen, noteHistoryRequest, recordFreezeStep } from './session-state.js'; import type { GptHistoryOptions } from './openai-history.js'; import { CACHE_CREATE_RATE, CACHE_READ_RATE } from './baseline.js'; import { visionTokens, type VisionPricing } from './vision-cost.js'; @@ -2200,6 +2200,18 @@ export async function transformRequest( const firstUserSha = firstUser ? await sha8(firstUser) : undefined; if (firstUserSha) info.firstUserSha8 = firstUserSha; + // Circuit breaker: this session has hit the provider's rate limit on + // several requests in a row. Imaging never gets it further — a 429 means + // the provider never populated a fresh prefix cache, so a retry re-sends + // the same doomed imaged bytes and fails the same way (#234: same + // image_count/image_bytes on every attempt). Fall back to plain text so + // this request has a chance to complete, which is also what lets the + // session's own prefix cache come back. + if (isRateLimitCircuitOpen(firstUserSha)) { + info.reason = 'compress=false (rate_limit_circuit_open)'; + return { body, info }; + } + // Canary: slab tags whose content churns within a session bust the image // cache every turn — report them regardless of the hardcoded lists. if (staticTagContents.size > 0) { diff --git a/tests/rate-limit-circuit-breaker.test.ts b/tests/rate-limit-circuit-breaker.test.ts new file mode 100644 index 00000000..b9f6295f --- /dev/null +++ b/tests/rate-limit-circuit-breaker.test.ts @@ -0,0 +1,116 @@ +/** + * A session stuck behind a provider rate limit must stop re-imaging the same + * doomed bytes. + * + * Observed in production (#234): a session retrying an unchanged imaged + * request got 429 on every attempt, same image_count/image_bytes each time — + * imaging never got it past the rate limit, because a 429 never populates a + * fresh prefix cache for the retry to build on. Three 429s in a row is past + * "unlucky timing"; falling back to plain text gives the session a chance to + * complete a request, which is also the only way its own cache comes back. + * + * Run just this file: pnpm vitest run tests/rate-limit-circuit-breaker.test.ts + */ +import { beforeEach, describe, expect, it } from 'vitest'; +import { transformRequest } from '../src/core/transform.js'; +import { + isRateLimitCircuitOpen, + noteRateLimitOutcome, + peekSessionState, + resetSessionState, +} from '../src/core/session-state.js'; + +const big = (n: number) => 'x'.repeat(n); +const enc = (obj: unknown) => new TextEncoder().encode(JSON.stringify(obj)); + +function bodyFor(userText: string) { + return enc({ + model: 'claude-sonnet-5', + system: [{ type: 'text', text: ['You are a helpful assistant.', big(40_000)].join('\n') }], + messages: [{ role: 'user', content: userText }], + }); +} + +describe('isRateLimitCircuitOpen — unit', () => { + beforeEach(() => resetSessionState()); + + it('stays closed under the threshold', () => { + noteRateLimitOutcome('sess-a', 429); + noteRateLimitOutcome('sess-a', 429); + expect(isRateLimitCircuitOpen('sess-a')).toBe(false); + }); + + it('opens at the third consecutive 429', () => { + noteRateLimitOutcome('sess-a', 429); + noteRateLimitOutcome('sess-a', 429); + noteRateLimitOutcome('sess-a', 429); + expect(isRateLimitCircuitOpen('sess-a')).toBe(true); + }); + + it('resets on any non-429 response', () => { + noteRateLimitOutcome('sess-a', 429); + noteRateLimitOutcome('sess-a', 429); + noteRateLimitOutcome('sess-a', 200); + expect(isRateLimitCircuitOpen('sess-a')).toBe(false); + expect(peekSessionState('sess-a')?.consecutive429).toBe(0); + }); + + it('keeps sessions independent', () => { + noteRateLimitOutcome('sess-a', 429); + noteRateLimitOutcome('sess-a', 429); + noteRateLimitOutcome('sess-a', 429); + expect(isRateLimitCircuitOpen('sess-a')).toBe(true); + expect(isRateLimitCircuitOpen('sess-b')).toBe(false); + }); + + it('is closed for a session never seen', () => { + expect(isRateLimitCircuitOpen('never-seen')).toBe(false); + }); +}); + +describe('transformRequest — falls back to text once the circuit is open', () => { + beforeEach(() => resetSessionState()); + + it('images normally while the circuit is closed', async () => { + const { info } = await transformRequest(bodyFor('turn 1')); + expect(info.imageCount).toBeGreaterThan(0); + expect(info.compressed).toBe(true); + }); + + it('stops imaging the same session after 3 consecutive 429s', async () => { + // First request: establishes the session and its firstUserSha8. + const first = await transformRequest(bodyFor('turn 1')); + expect(first.info.imageCount).toBeGreaterThan(0); + const sessionKey = first.info.firstUserSha8; + expect(sessionKey).toBeDefined(); + + // Three consecutive 429s on that session, exactly what proxy.ts records + // from real upstream responses. + noteRateLimitOutcome(sessionKey, 429); + noteRateLimitOutcome(sessionKey, 429); + noteRateLimitOutcome(sessionKey, 429); + + const after = await transformRequest(bodyFor('turn 1')); + expect(after.info.compressed).toBe(false); + expect(after.info.imageCount ?? 0).toBe(0); + expect(after.info.reason).toContain('rate_limit_circuit_open'); + }); + + it('resumes imaging once a request gets through', async () => { + const first = await transformRequest(bodyFor('turn 1')); + const sessionKey = first.info.firstUserSha8; + + noteRateLimitOutcome(sessionKey, 429); + noteRateLimitOutcome(sessionKey, 429); + noteRateLimitOutcome(sessionKey, 429); + expect((await transformRequest(bodyFor('turn 1'))).info.compressed).toBe(false); + + // A 200 gets through (e.g. the plain-text fallback request above + // succeeded) and clears the counter. + noteRateLimitOutcome(sessionKey, 200); + + const resumed = await transformRequest(bodyFor('turn 1')); + expect(resumed.info.compressed).toBe(true); + expect(resumed.info.imageCount).toBeGreaterThan(0); + }); +});