From e9a740efaa8dd89210b2f5bd8186dfdfe8539234 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Albert=20Castellana=20LLu=C3=ADs?= Date: Tue, 18 Aug 2026 17:28:04 +0200 Subject: [PATCH] fix(codex): hold the scarcity demote until the quota actually resets MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A 429 tells the router the subscription is exhausted; a reset header tells it until when. Only the first was used, so the demote decayed after quota_429_window_s (120s by default) and the router re-probed an exhaustion it already knew about — every couple of minutes, indefinitely, paying a failed call at the head of the cascade each time. Worse, the reset header could never arrive. The observer's filter matched ratelimit|usage|quota|percent, and the vendor's own naming is x-codex-primary-* (see the used-percent header in test_codex_scarcity), so a field like x-codex-primary-reset-after-seconds contains none of those words and was dropped. Since polling the endpoint would burn the very quota it measures, observation of real traffic is the only safe signal — a header dropped there is lost for good. - codex_backend: extract the filter as quota_headers/1 and widen it to *reset* and retry-after. Deliberately still narrow: response headers carry cookies and tokens, and this map is stored and rendered on the dashboard. - sources/codex: parse a reset time off a 429 (absolute epoch or seconds-from-observation, whichever the vendor sends) and hold the demote at full while it is still ahead. Falls back to the existing 429 ramp when no such header is present, so behaviour without one is unchanged. A reset header on a healthy 200 reports when the window rolls over, not exhaustion, and is deliberately not read as a demote signal. --- codex_backend.py | 21 +++++++++++-- sources/codex.py | 59 +++++++++++++++++++++++++++++++----- tests/test_codex.py | 22 ++++++++++++++ tests/test_codex_scarcity.py | 55 +++++++++++++++++++++++++++++++++ 4 files changed, 147 insertions(+), 10 deletions(-) diff --git a/codex_backend.py b/codex_backend.py index 595896e..37acb4e 100644 --- a/codex_backend.py +++ b/codex_backend.py @@ -12,6 +12,7 @@ import asyncio import json +import re from contextlib import AsyncExitStack from typing import Any, Iterable @@ -25,6 +26,21 @@ CODEX_BASE_URL = "https://chatgpt.com/backend-api/codex" +# The quota signal the router can never re-fetch: polling the codex endpoint +# would burn the very quota it measures, so observation of real traffic is the +# only safe source. A header dropped here is therefore lost for good — which is +# why the match covers WHEN the limit lifts (`*-reset-*`, `retry-after`) and not +# only how much is used. Deliberately still narrow: response headers can carry +# cookies and tokens, and this map is stored and rendered on the dashboard. +_QUOTA_HEADER_RE = re.compile(r"ratelimit|usage|quota|percent|reset|^retry-after$", re.I) + + +def quota_headers(headers) -> dict: + """The quota-relevant subset of a response's headers, lowercased.""" + return {k.lower(): v for k, v in dict(headers or {}).items() + if _QUOTA_HEADER_RE.search(k)} + + def _err(kind: str, status: int, latency_ms: int, message: str) -> dict: return {"ok": False, "error_kind": kind, "http_status": status, "latency_ms": latency_ms, "error_message": message} @@ -311,9 +327,8 @@ def _notify(status: int, headers=None) -> None: if observe is None: return try: - hdrs = {k.lower(): v for k, v in dict(headers or {}).items() - if _re.search(r"ratelimit|usage|quota|percent", k, _re.I)} - observe({"status": status, "headers": hdrs, "ts": int(time.time())}) + observe({"status": status, "headers": quota_headers(headers), + "ts": int(time.time())}) except Exception: pass diff --git a/sources/codex.py b/sources/codex.py index fe11a47..3fb7484 100644 --- a/sources/codex.py +++ b/sources/codex.py @@ -9,9 +9,11 @@ policy would route ALL of a family's traffic to it until it 429s, then oscillate. So as the subscription gets strained the host imputes a RISING ranking price so paid routes take over before the 429 wall — and it decays back as pressure eases. -Two signals feed the ramp: the `*used-percent*` quota header when codex exposes -one, AND recently observed 429s (the only signal when it doesn't). Billing stays -$0 (executed cost) — this is ranking-only. +Three signals feed the ramp: the `*used-percent*` quota header when codex +exposes one, a `*reset*` / `retry-after` header on a 429 (which pins the demote +until the quota actually rolls over, instead of letting it decay and re-probe an +exhaustion already known), AND recently observed 429s (the only signal when +neither header is present). Billing stays $0 (executed cost) — ranking-only. """ from __future__ import annotations @@ -29,6 +31,31 @@ # the window so it recovers. +def _reset_at(headers: dict, observed_ts: int | None) -> int | None: + """Epoch seconds at which the codex quota is said to reset, or None. + + The value is accepted in either shape without knowing the vendor's choice: + a large number is an absolute epoch, a small one is seconds-from-observation + (so an old event's short reset is correctly already past). Unparseable or + non-positive values yield None and the caller falls back to the 429 ramp. + """ + for name, raw in (headers or {}).items(): + n = str(name).lower() + if "reset" not in n and n != "retry-after": + continue + try: + v = float(str(raw).strip()) + except (TypeError, ValueError): + continue + if v <= 0: + continue + if v > 10**9: # absolute epoch + return int(v) + base = observed_ts if observed_ts is not None else int(time.time()) + return int(base + v) # seconds from when it was observed + return None + + class CodexSource: name = "codex" # A local tick (NOT an endpoint probe): re-imputes the scarcity price so it @@ -61,17 +88,27 @@ def ingest(self, provider_id: str, signal: dict) -> None: # ---- scarcity ------------------------------------------------------ def _demote_frac(self) -> float: - """0 (codex free, wins) → 1 (fully demoted) from the quota header and/or - recently observed 429s, whichever is higher.""" + """0 (codex free, wins) → 1 (fully demoted) from the quota header, a + known reset time, and/or recently observed 429s, whichever is higher.""" bal = (self._balances_sync().get(self.provider_ids[0]) or {}) used = bal.get("value") - recent_429 = (bal.get("detail") or {}).get("recent_429_count") or 0 + detail = bal.get("detail") or {} + recent_429 = detail.get("recent_429_count") or 0 start = settings.get("codex.quota_demote_start") shed = settings.get("codex.quota_429_shed") header_frac = (max(0.0, (float(used) - start) / (1.0 - start)) if used is not None and start < 1.0 else 0.0) rl_frac = (min(1.0, recent_429 / shed) if shed > 0 else 0.0) - return max(0.0, min(1.0, max(header_frac, rl_frac))) + + # A 429 says the subscription is out; a reset header says until WHEN. + # Without the second the 429 ramp decays after quota_429_window_s and + # the router re-probes an exhaustion it already knows about, paying a + # failed call at the head of the cascade every window. While the reset + # is still ahead, stay fully demoted so paid routes carry the traffic. + reset_at = detail.get("reset_at") + reset_frac = 1.0 if reset_at is not None and time.time() < reset_at else 0.0 + + return max(0.0, min(1.0, max(header_frac, rl_frac, reset_frac))) def _push_scarcity_prices(self) -> None: if self._host is None or not self._families: @@ -93,6 +130,7 @@ def _balances_sync(self) -> dict[str, Balance]: observed: dict[str, str] = {} last_429 = None recent_429 = 0 + reset_at = None now_ts = int(time.time()) for e in events: for k, v in (e.get("headers") or {}).items(): @@ -107,10 +145,17 @@ def _balances_sync(self) -> dict[str, Balance]: ts = e.get("ts") if ts is None or (now_ts - ts) <= settings.get("codex.quota_429_window_s"): recent_429 += 1 + # Only a 429 carries "you are out until X"; the same header on a + # healthy 200 just reports when the window rolls over and must + # not be read as exhaustion. + at = _reset_at(e.get("headers") or {}, ts) + if at is not None and (reset_at is None or at > reset_at): + reset_at = at return {self.provider_ids[0]: { "kind": "quota_window", "value": used_fraction, "detail": {"recent_429_count": recent_429, "last_429_at": last_429, + "reset_at": reset_at, "observed_headers": observed, "events": len(events)}, "fetched_at": int(time.time()), }} diff --git a/tests/test_codex.py b/tests/test_codex.py index 09d3c93..4930c97 100644 --- a/tests/test_codex.py +++ b/tests/test_codex.py @@ -386,3 +386,25 @@ def test_content_to_text_coerces_part_arrays(): assert cb._content_to_text([{"type": "text", "text": "a"}, {"type": "text", "text": "b"}]) == "ab" assert cb._content_to_text(7) == "7" + + +def test_quota_headers_keeps_reset_signals_not_just_usage(): + """The observer is the ONLY quota signal (polling would burn the quota it + measures), so a header it drops is a signal nobody can ever recover. The + vendor's own naming is `x-codex-primary-*`, and a reset field in that family + contains none of ratelimit/usage/quota/percent.""" + import codex_backend as cb + + kept = cb.quota_headers({ + "x-codex-primary-used-percent": "98", + "x-codex-primary-reset-after-seconds": "3600", + "retry-after": "120", + "content-type": "application/json", + "set-cookie": "session=abc", + }) + + assert "x-codex-primary-used-percent" in kept # unchanged + assert "x-codex-primary-reset-after-seconds" in kept + assert "retry-after" in kept + assert "content-type" not in kept # still narrow + assert "set-cookie" not in kept # never widen onto secrets diff --git a/tests/test_codex_scarcity.py b/tests/test_codex_scarcity.py index 48030da..52c5873 100644 --- a/tests/test_codex_scarcity.py +++ b/tests/test_codex_scarcity.py @@ -81,3 +81,58 @@ async def test_pricing_returns_scarcity_for_each_family(): fams = {p["model_family"]: p for p in prices} assert set(fams) == {"gpt-5.5", "gpt-5.3-codex-spark"} assert fams["gpt-5.5"]["price_in_usd_per_mtok"] == 5.0 + + +# --- reset-aware hold ------------------------------------------------------- +# A 429 says "you are out"; the reset header says "until when". Without the +# second, the ramp decays after quota_429_window_s (120s) and the router +# re-probes a subscription it already knows is exhausted, every couple of +# minutes, paying the full failed-call latency at the head of the cascade. + + +def test_reset_header_holds_full_demote_until_it_passes(): + s, h = _src() + now = int(time.time()) + # ONE 429 (the 429 ramp alone would give 1/3), plus a reset an hour out. + s.ingest("openai_codex", { + "status": 429, + "headers": {"x-codex-primary-reset-after-seconds": "3600"}, + "ts": now, + }) + assert _price_in(h) == 5.0 # held fully demoted until the quota resets + + +def test_reset_already_elapsed_falls_back_to_the_429_ramp(): + s, h = _src() + now = int(time.time()) + s.ingest("openai_codex", { + "status": 429, + "headers": {"x-codex-primary-reset-after-seconds": "1"}, + "ts": now - 600, # observed 10 min ago, so the 1s reset is long past + }) + # the 429 itself has also aged out of the 120s window → recovered + assert _price_in(h) == 0.0 + + +def test_epoch_form_reset_header_is_understood(): + s, h = _src() + now = int(time.time()) + s.ingest("openai_codex", { + "status": 429, + "headers": {"x-codex-primary-reset-at": str(now + 3600)}, + "ts": now, + }) + assert _price_in(h) == 5.0 + + +def test_reset_header_without_a_429_does_not_demote(): + s, h = _src() + now = int(time.time()) + # a healthy 200 that merely reports when the window rolls over must not + # be read as exhaustion + s.ingest("openai_codex", { + "status": 200, + "headers": {"x-codex-primary-reset-after-seconds": "3600"}, + "ts": now, + }) + assert _price_in(h) == 0.0