Stop upstream 429s from disabling a codex provider

A 429 from the ChatGPT backend is per-quota-bucket, not a provider
fault: x-codex-active-limit names the bucket that refused, and the
account keeps serving from another one meanwhile. In one hour of
production traffic codex_think returned 1888 x 200 interleaved with
1282 x 429. Every 429 was recorded as a provider failure, so three in a
row tripped the three-strikes cooldown and aisbf spent five minutes at a
time rejecting requests itself -- including the majority the upstream
would have answered. One client fired 32 requests during a cooldown and
got 32 x 503 without a single one reaching OpenAI.

- codex: raise RateLimitError on a 429 instead of raise_for_status(),
  without disabling the provider, on both the streaming and
  non-streaming OAuth paths.
- handlers: _should_record_failure() now excludes upstream rate limits
  as well as CoderAI warm-up, and a quota refusal is forwarded to the
  client as 429 rather than a generic 500.

Send ChatGPT-Account-ID again. The header was only set from
tokens.account_id, which is null in every credentials file the login
flow writes; the real value is in the id_token's chatgpt_account_id
claim. Without it the backend picks a workspace itself, so an account
belonging to several workspaces cannot be steered at the configured one.

Make rate_limit actually do something. The spacing timestamp lived on
the handler, but get_provider_handler() builds a fresh handler per
request, so it was always 0 on arrival and no wait was ever applied --
a configured rate_limit was silently inert. Move the timestamps to a
process-wide registry guarded by a per-slot lock, without which N
concurrent requests all read the same stale timestamp and burst
together. Verified: 4 concurrent requests at 0.5s spacing now take
1.50s, previously 0.00s.

Bump version to 0.99.90.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

aisbf/__init__.py

@@ -55,7 +55,7 @@ from .auth.qwen import QwenOAuth2
 from .handlers import RequestHandler, RotationHandler, AutoselectHandler
 from .utils import count_messages_tokens, split_messages_into_chunks, get_max_request_tokens_for_model, get_max_completion_tokens_for_model
 
-__version__ = "0.99.89"
+__version__ = "0.99.90"
 __all__ = [
     # Config
     "config",

aisbf/handlers.py

@@ -80,6 +80,28 @@ def _is_coderai_warmup_error(error) -> bool:
     return 'No active CoderAI broker session' in str(error or '')
 
 
+def _is_upstream_rate_limit(error) -> bool:
+    """True when the upstream refused on quota rather than the provider breaking.
+
+    A 429 says one quota bucket is exhausted right now; it says nothing about
+    whether the next request will succeed, and providers routinely keep serving
+    from another bucket while one is capped. Counting these toward the
+    three-strikes cooldown makes aisbf reject traffic itself that the upstream
+    would have answered, so they are excluded from failure recording.
+    """
+    if isinstance(error, RateLimitError):
+        return True
+    status = getattr(getattr(error, 'response', None), 'status_code', None)
+    if status == 429:
+        return True
+    return '429 Too Many Requests' in str(error or '') or 'usage_limit_reached' in str(error or '')
+
+
+def _should_record_failure(error) -> bool:
+    """Whether an exception should count against the provider's failure budget."""
+    return not _is_coderai_warmup_error(error) and not _is_upstream_rate_limit(error)
+
+
 def _raise_if_provider_unavailable(handler, provider_id: str) -> None:
     """Reject the request with 503 when the provider is disabled, saying why.
 
@@ -995,8 +1017,7 @@ class RequestHandler:
             logger.info(f"=== RequestHandler.handle_chat_completion END ===")
             return response
         except Exception as e:
-            # CoderAI warm-up (no broker session) is transient — don't disable it.
-            if not _is_coderai_warmup_error(e):
+            if _should_record_failure(e):
                 handler.record_failure()
 
             # Record failed request analytics
@@ -1048,7 +1069,11 @@ class RequestHandler:
                     )
             except Exception as analytics_error:
                 logger.warning(f"Analytics recording for failed request failed: {analytics_error}")
-            
+
+            # Forward an upstream quota refusal as a 429 so the caller can back off
+            # or fail over on a real signal instead of a generic server error.
+            if _is_upstream_rate_limit(e):
+                raise HTTPException(status_code=429, detail=str(e))
             raise HTTPException(status_code=500, detail=str(e))
 
     async def handle_streaming_chat_completion(self, request: Request, provider_id: str, request_data: Dict):
@@ -1734,8 +1759,7 @@ class RequestHandler:
                     logger.warning(f"Analytics recording for streaming request failed: {analytics_error}")
                 
             except Exception as e:
-                # CoderAI warm-up (no broker session) is transient — don't disable it.
-                if not _is_coderai_warmup_error(e):
+                if _should_record_failure(e):
                     handler.record_failure()
 
                 # Record analytics for failed streaming request
@@ -2241,7 +2265,7 @@ class RequestHandler:
                 logger.warning(f"Market settlement failed for audio transcription {provider_id}: {market_error}")
             return result
         except Exception as e:
-            if not _is_coderai_warmup_error(e):
+            if _should_record_failure(e):
                 handler.record_failure()
             raise HTTPException(status_code=500, detail=str(e))
     
@@ -2275,7 +2299,7 @@ class RequestHandler:
                 logger.warning(f"Market settlement failed for text to speech {provider_id}: {market_error}")
             return result
         except Exception as e:
-            if not _is_coderai_warmup_error(e):
+            if _should_record_failure(e):
                 handler.record_failure()
             raise HTTPException(status_code=500, detail=str(e))
     
@@ -2319,7 +2343,7 @@ class RequestHandler:
                 logger.warning(f"Market settlement failed for image generation {provider_id}: {market_error}")
             return result
         except Exception as e:
-            if not _is_coderai_warmup_error(e):
+            if _should_record_failure(e):
                 handler.record_failure()
             raise HTTPException(status_code=500, detail=str(e))
     
@@ -2353,7 +2377,7 @@ class RequestHandler:
                 logger.warning(f"Market settlement failed for embeddings {provider_id}: {market_error}")
             return result
         except Exception as e:
-            if not _is_coderai_warmup_error(e):
+            if _should_record_failure(e):
                 handler.record_failure()
             raise HTTPException(status_code=500, detail=str(e))
     
@@ -5157,7 +5181,7 @@ class RotationHandler:
                 #    bad request, unsupported model, auth) mean the provider is up and
                 #    simply rejected this request; disabling it would pull a healthy
                 #    provider out of rotation for the whole cooldown.
-                if not _is_coderai_warmup_error(e) and not self._is_non_retryable_error(e):
+                if _should_record_failure(e) and not self._is_non_retryable_error(e):
                     handler.record_failure()
                 latency_ms = (time.time() - request_start_time) * 1000 if request_start_time else 0
                 self._record_dashboard_proxy_event(

aisbf/providers/base.py

@@ -34,6 +34,12 @@ from ..batching import get_request_batcher
 # Check if debug mode is enabled
 AISBF_DEBUG = os.environ.get('AISBF_DEBUG', '').lower() in ('true', '1', 'yes')
 
+# Request spacing state, keyed by "provider:user" (and ":model" for per-model
+# limits). Process-wide because a handler instance lives for exactly one request,
+# so per-instance timestamps can never space anything apart. See _throttle().
+_rate_limit_last_request: Dict[str, float] = {}
+_rate_limit_locks: Dict[str, asyncio.Lock] = {}
+
 
 # Newer OpenAI models (reasoning models and the GPT-5/codex families) rejected the
 # legacy `max_tokens` parameter and only accept `max_completion_tokens`. Older
@@ -1669,60 +1675,57 @@ class BaseProviderHandler:
         logger.error(f"Disabled until: {self.error_tracking['disabled_until']}")
         logger.error(f"Provider will be automatically re-enabled after cooldown")
 
-    async def apply_rate_limit(self, rate_limit: Optional[float] = None):
-        """Apply rate limiting by waiting if necessary, using adaptive rate limiting."""
+    async def _throttle(self, slot: str, rate_limit: float, label: str):
+        """Space requests sharing `slot` at least `rate_limit` seconds apart.
+
+        The timestamp lives in a process-wide registry rather than on the handler
+        because get_provider_handler() builds a fresh handler for every request:
+        an instance attribute is always 0 on arrival, so the spacing never applied
+        and a configured rate_limit did nothing at all. The lock is what makes the
+        limit hold under concurrency — without it, N simultaneous requests all
+        read the same stale timestamp, decide no wait is due, and burst together.
+        """
         import logging
         logger = logging.getLogger(__name__)
-        
-        # Use adaptive rate limiter if enabled
-        if self.adaptive_limiter.enabled:
-            adaptive_limit = self.adaptive_limiter.get_rate_limit()
-            
-            if rate_limit is None:
-                rate_limit = adaptive_limit
-            else:
-                # Use the higher of the two (more conservative)
-                rate_limit = max(rate_limit, adaptive_limit)
-        elif rate_limit is None:
-            rate_limit = self.rate_limit
-
-        if rate_limit and rate_limit > 0:
-            current_time = time.time()
-            time_since_last_request = current_time - self.last_request_time
-            required_wait = rate_limit - time_since_last_request
 
+        lock = _rate_limit_locks.setdefault(slot, asyncio.Lock())
+        async with lock:
+            last_time = _rate_limit_last_request.get(slot, 0)
+            required_wait = rate_limit - (time.time() - last_time)
             if required_wait > 0:
-                logger.info(f"[RateLimit] Provider {self.provider_id}: waiting {required_wait:.2f}s (adaptive: {self.adaptive_limiter.enabled})")
+                logger.info(
+                    f"[RateLimit] {label}: waiting {required_wait:.2f}s "
+                    f"(limit: {rate_limit:.3f}s between requests, adaptive: {self.adaptive_limiter.enabled})"
+                )
                 await asyncio.sleep(required_wait)
+            _rate_limit_last_request[slot] = time.time()
 
-            self.last_request_time = time.time()
-
-    async def apply_model_rate_limit(self, model: str, rate_limit: Optional[float] = None):
-        """Apply rate limiting for a specific model, using adaptive rate limiting."""
-        import logging
-        logger = logging.getLogger(__name__)
-        
-        # Use adaptive rate limiter if enabled
+    def _effective_rate_limit(self, rate_limit: Optional[float]) -> Optional[float]:
+        """Combine the caller's limit with the adaptive one, taking the wider."""
         if self.adaptive_limiter.enabled:
             adaptive_limit = self.adaptive_limiter.get_rate_limit()
-            
             if rate_limit is None:
-                rate_limit = adaptive_limit
-            else:
-                rate_limit = max(rate_limit, adaptive_limit)
-        elif rate_limit is None:
-            rate_limit = self.rate_limit
+                return adaptive_limit
+            # Use the higher of the two (more conservative)
+            return max(rate_limit, adaptive_limit)
+        if rate_limit is None:
+            return self.rate_limit
+        return rate_limit
 
+    async def apply_rate_limit(self, rate_limit: Optional[float] = None):
+        """Apply rate limiting by waiting if necessary, using adaptive rate limiting."""
+        rate_limit = self._effective_rate_limit(rate_limit)
         if rate_limit and rate_limit > 0:
-            current_time = time.time()
-            last_time = self.model_last_request_time.get(model, 0)
-            time_since_last_request = current_time - last_time
-            required_wait = rate_limit - time_since_last_request
-
-            if required_wait > 0:
-                logger.info(f"[RateLimit] Model {model}: waiting {required_wait:.2f}s (adaptive: {self.adaptive_limiter.enabled})")
-                await asyncio.sleep(required_wait)
+            slot = f"{self.provider_id}:{self.user_id}"
+            await self._throttle(slot, rate_limit, f"Provider {self.provider_id}")
+            self.last_request_time = time.time()
 
+    async def apply_model_rate_limit(self, model: str, rate_limit: Optional[float] = None):
+        """Apply rate limiting for a specific model, using adaptive rate limiting."""
+        rate_limit = self._effective_rate_limit(rate_limit)
+        if rate_limit and rate_limit > 0:
+            slot = f"{self.provider_id}:{self.user_id}:{model}"
+            await self._throttle(slot, rate_limit, f"Model {model}")
             self.model_last_request_time[model] = time.time()
 
     def record_failure(self):

aisbf/providers/codex.py

@@ -21,6 +21,7 @@ along with this program.  If not, see <https://www.gnu.org/licenses/>.
 
 Why did the programmer quit his job? Because he didn't get arrays!
 """
+import base64
 import json
 import logging
 import time
@@ -35,6 +36,7 @@ from ..config import config
 from ..utils import count_messages_tokens
 from .base import (
     BaseProviderHandler,
+    RateLimitError,
     AISBF_DEBUG,
     adapt_request_for_unsupported_param,
     apply_max_tokens_param,
@@ -220,19 +222,91 @@ class CodexProviderHandler(BaseProviderHandler):
             save_callback=lambda creds: self._save_oauth2_to_db(creds)
         )
     
+    def _raise_if_usage_limited(self, status_code: int, body: bytes, headers=None) -> None:
+        """Turn a Codex usage-limit refusal into RateLimitError without disabling us.
+
+        A 429 from the ChatGPT backend is per-quota-bucket, not a provider fault:
+        x-codex-active-limit says which bucket refused, and the account routinely
+        still serves requests from another one. In one hour of production traffic
+        this provider returned 1888 × 200 interleaved with 1282 × 429. Recording
+        those 429s as failures tripped the three-strikes cooldown and made aisbf
+        reject everything itself for five minutes at a time — including the
+        majority that would have succeeded.
+
+        So: surface the upstream refusal to the caller, and leave the provider
+        enabled so the next request is actually attempted.
+        """
+        if status_code != 429:
+            return
+        detail = ''
+        try:
+            payload = json.loads(body or b'{}')
+            error = payload.get('error') or {}
+            detail = error.get('type') or error.get('message') or ''
+        except Exception:
+            detail = (body or b'').decode('utf-8', 'replace')[:200]
+
+        bucket = (headers or {}).get('x-codex-active-limit') if headers else None
+        logger.warning(
+            f"[{self.provider_id}] Upstream usage limit ({detail or '429'}"
+            f"{f', bucket={bucket}' if bucket else ''}) — passing 429 through, provider stays enabled"
+        )
+        # record_429 lets the adaptive limiter widen the request spacing on its own.
+        try:
+            self.adaptive_limiter.record_429(0)
+        except Exception:
+            pass
+        raise RateLimitError(
+            f"Provider {self.provider_id} upstream usage limit reached: {detail or '429 Too Many Requests'}",
+            status_code=429,
+        )
+
+    @staticmethod
+    def _account_id_from_credentials(credentials: Optional[Dict]) -> Optional[str]:
+        """Resolve the ChatGPT account id the request should be billed to.
+
+        The login flow only sometimes writes tokens.account_id — in practice it is
+        often null — but the id_token always carries chatgpt_account_id in its
+        OpenAI auth claim. Without it the ChatGPT-Account-ID header is omitted and
+        the backend picks a workspace itself, so an account belonging to several
+        workspaces cannot be steered at the one the operator configured. Fall back
+        to the claim rather than sending nothing.
+        """
+        tokens = (credentials or {}).get('tokens') or {}
+        account_id = tokens.get('account_id')
+        if account_id:
+            return account_id
+        id_token = tokens.get('id_token')
+        if not id_token:
+            return None
+        try:
+            payload = id_token.split('.')[1]
+            payload += '=' * (-len(payload) % 4)
+            claims = json.loads(base64.urlsafe_b64decode(payload))
+            return claims.get('https://api.openai.com/auth', {}).get('chatgpt_account_id')
+        except Exception as e:
+            logger.debug(f"CodexProviderHandler: could not read account id from id_token: {e}")
+            return None
+
     async def _get_valid_api_key(self) -> str:
         """Get a valid API key, refreshing OAuth2 if needed."""
         # If we have an API key, use it (prefer passed api_key, then stored config)
         if self.api_key:
             return self.api_key
-        
+
         # Try OAuth2 token
         token = await self.oauth2.get_valid_token_with_refresh()
         if token:
             # Extract account ID from credentials if available
-            if self.oauth2.credentials and self.oauth2.credentials.get('tokens'):
-                self._account_id = self.oauth2.credentials['tokens'].get('account_id')
-            
+            self._account_id = self._account_id_from_credentials(self.oauth2.credentials)
+            if self._account_id:
+                logger.info(f"[{self.provider_id}] Billing requests to ChatGPT account {self._account_id}")
+            else:
+                logger.warning(
+                    f"[{self.provider_id}] No ChatGPT account id available — the backend will "
+                    f"choose the workspace, which may not be the configured one"
+                )
+
             # Switch to ChatGPT backend if OAuth2 is now authenticated
             if not self._use_api_key_mode and self.base_url != "https://chatgpt.com/backend-api":
                 self.base_url = "https://chatgpt.com/backend-api"
@@ -660,6 +734,7 @@ class CodexProviderHandler(BaseProviderHandler):
                         f"CodexProviderHandler: Streaming error {response.status_code}: "
                         f"{error_body.decode('utf-8')}"
                     )
+                    self._raise_if_usage_limited(response.status_code, error_body, response.headers)
                     response.raise_for_status()
 
                 async for event in self._parse_sse_stream(response):
@@ -818,6 +893,7 @@ class CodexProviderHandler(BaseProviderHandler):
                     error_body = await response.aread()
                     logger.error(f"CodexProviderHandler: Error response status: {response.status_code}")
                     logger.error(f"CodexProviderHandler: Error response body: {error_body.decode('utf-8')}")
+                    self._raise_if_usage_limited(response.status_code, error_body, response.headers)
 
                 response.raise_for_status()
 
@@ -893,6 +969,10 @@ class CodexProviderHandler(BaseProviderHandler):
             
             return response
                 
+        except RateLimitError:
+            # An upstream quota refusal is not a provider fault — leave the failure
+            # counter alone so the next request is still attempted.
+            raise
         except Exception as e:
             logger.error(f"CodexProviderHandler: Error: {str(e)}", exc_info=True)
             self.record_failure()

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "aisbf"
-version = "0.99.89"
+version = "0.99.90"
 description = "AISBF - AI Service Broker Framework || AI Should Be Free - A modular proxy server for managing multiple AI provider integrations"
 readme = "README.md"
 license = "GPL-3.0-or-later"

setup.py

@@ -106,7 +106,7 @@ class InstallCommand(_install):
 
 setup(
     name="aisbf",
-    version="0.99.89",
+    version="0.99.90",
     author="AISBF Contributors",
     author_email="stefy@nexlab.net",
     description="AISBF - AI Service Broker Framework || AI Should Be Free - A modular proxy server for managing multiple AI provider integrations",