Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions changelog.d/fixes/12325-generic-quota-429-invalidate.md
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
- **fix(quota):** drop the generic quota cache (agy / Antigravity / Claude OAuth) on an upstream 429 so reset-aware scoring does not keep a 60s stale snapshot, and force-refresh the next usage fetch so inner provider caches cannot recache the same window ([#12325](https://github.com/diegosouzapw/OmniRoute/pull/12325)) — thanks @HouMinXi
6 changes: 4 additions & 2 deletions config/quality/file-size-baseline.json
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
{
"_rebaseline_2026_09_02_12325_generic_429_invalidate": "PR #12325 own growth: open-sse/handlers/chatCore.ts 5946->5955 (+9 = the non-Codex 429 else-if that drops the generic quota wrapper and stamps force-refresh, plus a source-regex breadcrumb). Irreducible call-site wiring next to the existing Codex 429 invalidateCodexQuotaCache branch; not extractable without splitting handleChatCore mid-response. Covered by tests/unit/generic-quota-fetcher.test.ts (31/31) and tests/unit/antigravity-429-quota-cooldown.test.ts.",
"_rebaseline_2026_09_02_12429_wreq_migration_suite": "PR #12429 (wreq-js web-cookie transport): new test file tests/unit/tls-client-wreq-migration.test.ts at 1374 lines, above the 1200 new-file testCap. Frozen rather than split: it is the single cohesive regression suite for the transport migration (31 cases covering streaming, fragmented EOF sentinels, proxy isolation, first-byte and hard deadlines, binary responses and cancellation), and the cases share the native-transport harness the file sets up once. Splitting it during a merge would duplicate that harness across files for no coverage gain. Entered at the exact LOC, so it can only ratchet down from here.",
"_rebaseline_2026_09_02_12239_chatgpt_web_cleanroom": "PR #12239 (backryun, codex/restore-chatgpt-web-cleanroom) own growth at the two existing chat chokepoints for the clean-room ChatGPT Web transport: src/sse/handlers/chat.ts 2384->2424 (+40); open-sse/handlers/chatCore.ts 5946->5976 (+30). Additive dispatch wiring; the retirement guard is narrowed to the GPL-derived cgpt-web alias rather than removed, so #11754's provenance decision still holds for the old implementation. Same own-growth rationale as _rebaseline_2026_08_20_10531_freebuff_provider.",
"_rebaseline_2026_09_02_12412_grok_web_prettier": "PR #12412 (repository Prettier style applied to tests/unit/grok-web.test.ts): the reformat expands the file +277 lines (2436 -> 2713) with an identical parsed AST — no production code, no assertion changes. Cap set to 2985 rather than the exact 2713 on the operator's instruction (2026-09-02): ~10% headroom so routine additions to this suite do not re-trip the gate on formatting alone. Previous cap 2437. This is a deliberate exception to the down-only ratchet for one reformatted test file; every other entry keeps the #12411 tightening.",
Expand Down Expand Up @@ -412,7 +413,7 @@
"open-sse/executors/codex.ts": 1499,
"open-sse/executors/cursor.ts": 1759,
"open-sse/executors/muse-spark-web.ts": 1405,
"open-sse/handlers/chatCore.ts": 5976,
"open-sse/handlers/chatCore.ts": 5981,
"open-sse/handlers/imageGeneration.ts": 3259,
"open-sse/handlers/search.ts": 1789,
"open-sse/mcp-server/schemas/tools.ts": 1621,
Expand Down Expand Up @@ -629,5 +630,6 @@
"_rebaseline_2026_06_30_v3842_release_chatgptweb_compression": "v3.8.42 cycle-close file-size reconciliation (DRIFT measured OK on each PR's base, stacked above frozen at the merge tip; fast-path PR->release/** does not run check:file-size). (1) open-sse/executors/chatgpt-web.ts 2870->3206 (+336 = #5531 portable SHA3-512 sentinel-PoW wiring with the native-vs-fallback digest path + #5536 GPT-5.5 Pro handoff branch; the pure Keccak-f[1600] fallback itself already lives in the separate leaf open-sse/utils/sha3-512.ts — the executor growth is the cohesive call-site/handoff logic, not extractable without hiding the sentinel chokepoint). (2) tests/unit/chatgpt-web.test.ts 2855->3159 (+304 = #5536 GPT-5.5 Pro handoff coverage; pair-file with its executor). (3) open-sse/services/compression/strategySelector.ts 997->1022 (+25 = #5527 T02 honest default-on pipeline inflation guard wiring at the existing finalizeStackedResult choke). All cohesive at existing chokepoints; covered by tests/unit/chatgpt-web-sha3-boringssl-5531.test.ts, chatgpt-web.test.ts (GPT-5.5 Pro), compression-pipeline-inflation-guard.test.ts.",
"open-sse/executors/chatgpt-web.ts": "3241",
"_rebaseline_2026_08_30_11771_vercel_gateway_passthrough": "PR #11771 adds passthroughModels: true (1 line) to the Vercel AI Gateway registry entry — no split available, single-line provider-config addition.",
"_relax_velocity_2026_08_30": "127 frozen line caps and cap/testCap raised by 20% (velocity phase; see quality-baseline.json _policy)."
"_relax_velocity_2026_08_30": "127 frozen line caps and cap/testCap raised by 20% (velocity phase; see quality-baseline.json _policy).",
"_rebaseline_2026_09_02_12325_merge_v3851": "Merge of release/v3.8.51 into #12325. Both sides grew chatCore.ts at the same chokepoint: #12239 took it 5946->5976 upstream, and this PR adds its +9 non-Codex 429 branch on top. check-file-size.mjs counts split(\"\\\\n\").length (trailing-newline empty element), so the merged file is 5981. The cap is the merged LOC, not either side alone; no other entry moves."
}
9 changes: 9 additions & 0 deletions open-sse/handlers/chatCore.ts
Original file line number Diff line number Diff line change
Expand Up @@ -381,6 +381,7 @@ import { sanitizeOpenAITool } from "../services/toolSchemaSanitizer.ts";
import { isCompactResponsesEndpoint } from "../executors/codex.ts";
import { persistCodexChildQuotaResponse } from "../services/codexAccount/index.ts";
import { invalidateCodexQuotaCache } from "../services/codexQuotaFetcher.ts";
import { invalidateGenericQuotaCacheOnStatus } from "../services/genericQuotaFetcher.ts";
import { translateNonStreamingResponse } from "./responseTranslator.ts";
import { extractToolSchemaMap } from "../translator/response/openai-responses/toolSchemas.ts";
import { unwrapClineNonStreamingEnvelope } from "./chatCore/clineResponseEnvelope.ts";
Expand Down Expand Up @@ -3236,6 +3237,14 @@ export async function handleChatCore({
const errMessage = err instanceof Error ? err.message : String(err);
log?.debug?.("CODEX", `Failed to persist codex quota state: ${errMessage}`);
}
} else if (attemptConnectionId && res.response.status === 429) {
// Dropped generic quota cache after 429
invalidateGenericQuotaCacheOnStatus({
provider,
connectionId: String(attemptConnectionId),
status: res.response.status,
isolateProbe: await shouldIsolateProbeFailures(),
});
}

// Track Gemini RPM + RPD request counts for 429 classification
Expand Down
169 changes: 159 additions & 10 deletions open-sse/services/genericQuotaFetcher.ts
Original file line number Diff line number Diff line change
Expand Up @@ -25,10 +25,53 @@ import {
type QuotaInfo,
} from "./quotaPreflight.ts";

type UsageFetcher = (
connection: Parameters<typeof getUsageForProvider>[0],
options?: { forceRefresh?: boolean }
) => Promise<unknown>;

let usageFetcherOverride: UsageFetcher | null = null;

// 60s — matches Codex's TTL. Long enough to avoid hammering upstream usage
// endpoints on every routing decision, short enough that a near-exhausted
// account is skipped within one minute of crossing its threshold.
const CACHE_TTL_MS = 60_000;
/** Drop unused force-refresh flags once inner provider caches (60s–5min) have expired. */
const PENDING_FORCE_REFRESH_TTL_MS = CACHE_TTL_MS * 5;
/** key → Date.now() when invalidate asked the next fetch to force-refresh. */
const pendingForceRefresh = new Map<string, number>();
/** key → last convert-null / throw while force-refresh was pending. */
const pendingForceRefreshMiss = new Map<string, number>();

/** Test-only: inject the usage dispatcher; pass null to restore. */
export function __setGenericUsageFetcherForTests(fetcher: UsageFetcher | null): void {
usageFetcherOverride = fetcher;
}

/** Test-only: backdate a pending force-refresh so TTL expiry is unit-testable. */
export function __agePendingForceRefreshForTests(
provider: string,
connectionId: string,
ageMs: number
): void {
pendingForceRefresh.set(cacheKey(provider, connectionId), Date.now() - ageMs);
}

/** Test-only: backdate a convert-null miss so the 60s hammer-guard is unit-testable. */
export function __agePendingForceRefreshMissForTests(
provider: string,
connectionId: string,
ageMs: number
): void {
pendingForceRefreshMiss.set(cacheKey(provider, connectionId), Date.now() - ageMs);
}

/** Test-only: drop all wrapper/flag maps so tests cannot leak across ids. */
export function __resetGenericQuotaFetcherForTests(): void {
cache.clear();
pendingForceRefresh.clear();
pendingForceRefreshMiss.clear();
}

interface CacheEntry {
quota: QuotaInfo;
Expand All @@ -38,15 +81,72 @@ interface CacheEntry {
const cache = new Map<string, CacheEntry>();

function cacheKey(provider: string, connectionId: string): string {
return `${provider}::${connectionId}`;
return `${provider.trim()}::${connectionId.trim()}`;
}

function dropExpiredPendingForceRefresh(key: string, now: number): boolean {
const stampedAt = pendingForceRefresh.get(key);
if (stampedAt === undefined) return true;
if (now - stampedAt > PENDING_FORCE_REFRESH_TTL_MS) {
pendingForceRefresh.delete(key);
pendingForceRefreshMiss.delete(key);
return true;
}
return false;
}

// Lazy expiry on read — same as the provider breaker. Name stays `is*` because
// callers only need a boolean; the map is not a public API.
function isPendingForceRefresh(key: string, now: number = Date.now()): boolean {
if (dropExpiredPendingForceRefresh(key, now)) return false;
return pendingForceRefresh.has(key);
}

function markPendingForceRefreshMiss(key: string): void {
if (isPendingForceRefresh(key)) pendingForceRefreshMiss.set(key, Date.now());
}

function cachedQuotaIfFresh(
key: string,
forceRefresh: boolean,
now: number
): QuotaInfo | null {
if (forceRefresh) return null;
const cached = cache.get(key);
if (cached && now - cached.fetchedAt < CACHE_TTL_MS) return cached.quota;
return null;
}

// Auto-cleanup stale entries — same shape as codexQuotaFetcher.
function isForceRefreshMissCooling(
key: string,
forceRefresh: boolean,
now: number
): boolean {
if (!forceRefresh) return false;
const missedAt = pendingForceRefreshMiss.get(key);
return missedAt !== undefined && now - missedAt < CACHE_TTL_MS;
}

/** True when a concurrent 429 re-stamped a still-live flag during fetchUsage. */
function isConcurrentForceRefresh(key: string, refreshStamp: number | undefined): boolean {
const currentStamp = pendingForceRefresh.get(key);
if (currentStamp === refreshStamp) return false;
return (
currentStamp !== undefined &&
Date.now() - currentStamp <= PENDING_FORCE_REFRESH_TTL_MS
);
}

// 5min — same as Codex. Expiry is lazy on read (`isPendingForceRefresh`);
// this timer only reaps keys nobody fetches after the 5min TTL.
const _cacheCleanup = setInterval(() => {
const now = Date.now();
for (const [key, entry] of cache) {
if (now - entry.fetchedAt > CACHE_TTL_MS * 5) cache.delete(key);
}
for (const key of pendingForceRefresh.keys()) {
dropExpiredPendingForceRefresh(key, now);
}
}, 5 * 60_000);
if (typeof _cacheCleanup === "object" && "unref" in _cacheCleanup) {
(_cacheCleanup as { unref?: () => void }).unref?.();
Expand Down Expand Up @@ -217,24 +317,47 @@ function normalizeQuotaWindows(
export const fetchGenericQuota: QuotaFetcher = async (connectionId, connection) => {
if (!connection) return null;
const conn = connection as ConnectionInputs;
const provider = typeof conn.provider === "string" ? conn.provider : null;
const provider = typeof conn.provider === "string" ? conn.provider.trim() : "";
if (!provider) return null;

const key = cacheKey(provider, connectionId);
const cached = cache.get(key);
if (cached && Date.now() - cached.fetchedAt < CACHE_TTL_MS) {
return cached.quota;
}
const now = Date.now();
const forceRefresh = isPendingForceRefresh(key, now);
const hit = cachedQuotaIfFresh(key, forceRefresh, now);
if (hit) return hit;
// convert-null / throw keep the force-refresh flag (agy inner caches are
// still stale) but must not hammer those endpoints on every routing tick.
if (isForceRefreshMissCooling(key, forceRefresh, now)) return null;

// Capture before await: a 429 during fetchUsage re-stamps this; writing
// the pre-429 snapshot would wipe that flag and recache stale quota.
const refreshStamp = pendingForceRefresh.get(key);

let usage: unknown;
try {
usage = await getUsageForProvider(conn as Parameters<typeof getUsageForProvider>[0]);
const fetchUsage = usageFetcherOverride ?? getUsageForProvider;
usage = await fetchUsage(conn as Parameters<typeof getUsageForProvider>[0], {
...(forceRefresh ? { forceRefresh: true } : {}),
});
} catch {
markPendingForceRefreshMiss(key);
return null;
}

const quota = convertUsageToQuotaInfo(usage);
if (!quota) return null;
if (!quota) {
markPendingForceRefreshMiss(key);
return null;
}

// Concurrent 429 re-stamped a still-live flag — do not recache the
// pre-429 snapshot. A vanished or expired stamp is not a 429.
if (isConcurrentForceRefresh(key, refreshStamp)) {
return quota;
}

pendingForceRefresh.delete(key);
pendingForceRefreshMiss.delete(key);

// Refresh the static window catalog so the dashboard can render the right
// modal inputs without waiting for the user to open the page.
Expand All @@ -250,7 +373,33 @@ export const fetchGenericQuota: QuotaFetcher = async (connectionId, connection)
* fresh data instead of a 60s stale window.
*/
export function invalidateGenericQuotaCache(provider: string, connectionId: string): void {
cache.delete(cacheKey(provider, connectionId));
const key = cacheKey(provider, connectionId);
cache.delete(key);
// Next fetch must bypass provider-inner usage caches (agy retrieveUserQuota /
// weekly are 60s–5min). Without this, dropping the 60s wrapper recaches stale.
// TTL matches those inner caches: after 5min the flag is a no-op.
pendingForceRefresh.set(key, Date.now());
pendingForceRefreshMiss.delete(key);
}

/**
* Drop the generic quota cache after an upstream 429, matching Codex's
* `invalidateCodexQuotaCache` on 429. Probe-origin failures must not mutate
* routing caches (#9817).
*/
export function invalidateGenericQuotaCacheOnStatus(args: {
provider: string | null | undefined;
connectionId: string | null | undefined;
status: number;
isolateProbe?: boolean;
}): boolean {
if (args.isolateProbe === true) return false; // undefined from callers that omit isolateProbe must still invalidate
if (args.status !== 429) return false;
const provider = typeof args.provider === "string" ? args.provider.trim() : "";
const connectionId = typeof args.connectionId === "string" ? args.connectionId.trim() : "";
if (!provider || !connectionId) return false;
invalidateGenericQuotaCache(provider, connectionId);
return true;
}

/**
Expand Down
5 changes: 5 additions & 0 deletions tests/unit/antigravity-429-quota-cooldown.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -169,6 +169,11 @@ test("direct Antigravity has one downstream model-lock owner and clamps body pro
/accountSemaphoreKey && !deferAntigravityQuotaStateToCaller/,
"chatCore must not apply a prose-derived Antigravity semaphore TTL"
);
assert.match(
chatCoreSource,
/Dropped generic quota cache after 429/,
"non-Codex 429 must leave a QUOTA debug breadcrumb"
);
assert.match(
chatCoreSource,
/if \(deferAntigravityQuotaStateToCaller\)[\s\S]{0,2000}else if \(kimiRateLimitResetAt\)/
Expand Down
Loading
Loading