fix(session-state): circuit-break imaging after repeated 429s

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,

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<string, SessionRecord>();
 
 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;
 }

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) {

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);
+  });
+});