fix(proxy): refuse disabled model sources and spare account health from model-scoped rejections

app/modules/model_sources/repository.py

@@ -1,12 +1,27 @@
 from __future__ import annotations
 
-from sqlalchemy import delete, select
+from sqlalchemy import ColumnElement, and_, delete, or_, select
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import selectinload
 
 from app.db.models import ModelSource, ModelSourceModel
 
 
+def _enablement_filter(only_disabled: bool) -> ColumnElement[bool]:
+    """Enabled-state predicate for a per-capability source lookup.
+
+    ``only_disabled`` selects the exact complement of the routable set: rows
+    that match the model and the route shape but that an operator switched off,
+    at the source or at the individual model. Routing needs that complement to
+    tell "no source serves this model" apart from "the source that serves this
+    model is switched off" -- the latter must not fall through to a
+    subscription account, which rejects the model outright.
+    """
+    if only_disabled:
+        return or_(ModelSource.is_enabled.is_(False), ModelSourceModel.is_enabled.is_(False))
+    return and_(ModelSource.is_enabled.is_(True), ModelSourceModel.is_enabled.is_(True))
+
+
 class ModelSourcesRepository:
     def __init__(self, session: AsyncSession) -> None:
         self._session = session
@@ -38,16 +53,16 @@ async def find_chat_source_for_model(
         *,
         allowed_source_ids: set[str] | None = None,
         require_streaming: bool = False,
+        only_disabled: bool = False,
     ) -> ModelSource | None:
         stmt = (
             select(ModelSource)
             .options(selectinload(ModelSource.models))
             .join(ModelSourceModel, ModelSourceModel.source_id == ModelSource.id)
             .where(ModelSource.kind == "openai_compatible")
-            .where(ModelSource.is_enabled.is_(True))
             .where(ModelSource.supports_chat_completions.is_(True))
             .where(ModelSourceModel.model == model)
-            .where(ModelSourceModel.is_enabled.is_(True))
+            .where(_enablement_filter(only_disabled))
             .order_by(ModelSource.name, ModelSource.id)
             .limit(1)
         )
@@ -66,16 +81,16 @@ async def find_responses_source_for_model(
         *,
         allowed_source_ids: set[str] | None = None,
         require_streaming: bool = False,
+        only_disabled: bool = False,
     ) -> ModelSource | None:
         stmt = (
             select(ModelSource)
             .options(selectinload(ModelSource.models))
             .join(ModelSourceModel, ModelSourceModel.source_id == ModelSource.id)
             .where(ModelSource.kind == "openai_compatible")
-            .where(ModelSource.is_enabled.is_(True))
             .where(ModelSource.supports_responses.is_(True))
             .where(ModelSourceModel.model == model)
-            .where(ModelSourceModel.is_enabled.is_(True))
+            .where(_enablement_filter(only_disabled))
             .order_by(ModelSource.name, ModelSource.id)
             .limit(1)
         )

app/modules/model_sources/selection.py

@@ -34,8 +34,19 @@ async def select_responses_model_source(
     *,
     raw_model: str | None = None,
     require_streaming: bool = False,
+    only_disabled: bool = False,
 ) -> tuple[ModelSource, str] | None:
-    """Resolve ``model`` to a Responses-capable model source, if any."""
+    """Resolve ``model`` to a Responses-capable model source, if any.
+
+    ``only_disabled`` inverts the enabled-state filter and leaves every other
+    rule -- candidate order, API key model allowlist, source assignment scope,
+    subscription-registry precedence, route shape, streaming -- untouched, so
+    the answer is exactly "the source this request would have used, had the
+    operator not switched it off". Callers use it after the ordinary lookup
+    misses, to refuse the request instead of handing a source-owned model to a
+    subscription account that cannot serve it. Sharing one function keeps the
+    two lookups from drifting apart the way the transports once did.
+    """
     assigned_source_ids = allowed_source_ids_for_api_key(api_key)
     exact_allowed_models = set(api_key.allowed_models) if api_key and api_key.allowed_models else None
     candidates = [candidate for candidate in (raw_model, model) if candidate]
@@ -55,6 +66,7 @@ async def select_responses_model_source(
                 candidate,
                 allowed_source_ids=assigned_source_ids,
                 require_streaming=require_streaming,
+                only_disabled=only_disabled,
             )
             if source is not None:
                 break

app/modules/proxy/_service/streaming/helpers.py

@@ -387,6 +387,7 @@
 from app.modules.proxy.helpers import (
     _normalize_error_code,
     classify_upstream_failure,
+    is_model_scoped_upstream_rejection,
     is_upstream_model_capacity_error,
 )
 from app.modules.proxy.http_bridge_forwarding import (
@@ -867,6 +868,36 @@ def _is_account_neutral_request_rejection(
     return bool(_facade()._is_missing_tool_output_message(message))
 
 
+def _is_model_scoped_rejection(
+    *,
+    http_status: int | None,
+    message: str | None,
+) -> bool:
+    """Return whether upstream rejected the requested model, not the account.
+
+    The ChatGPT model-entitlement rejection is scoped to the model named in the
+    message. It proves nothing about the account's ability to serve the models
+    it *is* entitled to, so it must never move that account's ``error_count``:
+    otherwise one client looping on a model no account can serve drives every
+    serving account into ``ERROR_BACKOFF_THRESHOLD`` backoff and starves all
+    unrelated traffic on those accounts -- the exact failure mode when a model
+    reaches subscription selection because its OpenAI-compatible model source
+    is disabled or unreachable.
+
+    Failover is deliberately untouched: the classified failure is still
+    returned, so the caller keeps trying other accounts, whose entitlements may
+    differ.
+
+    The normalized code is not part of the match. Upstream delivers this
+    rejection with neither ``code`` nor ``type`` on the streaming path, which
+    normalizes to ``upstream_error``; on other paths it arrives as
+    ``invalid_request_error``. Only the exact message shape decides membership.
+    """
+    if http_status is not None and http_status != 400:
+        return False
+    return is_model_scoped_upstream_rejection(message)
+
+
 async def _handle_stream_error(
     proxy: Any,
     account: Account,
@@ -896,6 +927,17 @@ async def _handle_stream_error(
             code,
         )
         return classified
+    if _is_model_scoped_rejection(
+        http_status=http_status,
+        message=error.get("message"),
+    ):
+        _facade().logger.info(
+            "Skipped account error penalty for model-scoped upstream rejection account_id=%s request_id=%s code=%s",
+            "<redacted>" if privacy_policy.redacts_sensitive_details else account.id,
+            get_request_id(),
+            code,
+        )
+        return classified
     if classified["failure_class"] == "rate_limit":
         await proxy._load_balancer.mark_rate_limit(account, error)
     elif classified["failure_class"] == "quota":

app/modules/proxy/api.py

@@ -1132,6 +1132,17 @@ async def responses(
         if source_selection is not None:
             source, selected_model = source_selection
             responses_payload.model = selected_model
+        else:
+            disabled_denial = await _disabled_model_source_denial(
+                request,
+                responses_payload.model,
+                api_key,
+                route="responses",
+                raw_model=raw_source_model,
+                require_streaming=True,
+            )
+            if disabled_denial is not None:
+                return disabled_denial
     if source is not None:
         # Opportunistic admission gates subscription *account* capacity;
         # source-routed requests use no account, so a closed/empty pool must
@@ -1272,9 +1283,10 @@ async def v1_responses(
     # File-referencing Responses requests pin to the subscription account that
     # registered the upload; that account-scoped invariant applies to /v1
     # streams too, so such requests must not be source-routed.
+    file_pinned = bool(extract_input_file_ids(responses_payload.input))
     source_selection = (
         None
-        if extract_input_file_ids(responses_payload.input)
+        if file_pinned
         else await _select_responses_model_source(
             responses_payload.model,
             api_key,
@@ -1285,6 +1297,17 @@ async def v1_responses(
     source = source_selection[0] if source_selection is not None else None
     if source_selection is not None:
         responses_payload.model = source_selection[1]
+    elif not file_pinned:
+        disabled_denial = await _disabled_model_source_denial(
+            request,
+            responses_payload.model,
+            api_key,
+            route="responses",
+            raw_model=raw_source_model,
+            require_streaming=responses_payload.stream is True,
+        )
+        if disabled_denial is not None:
+            return disabled_denial
     if source is not None:
         # Opportunistic admission gates subscription *account* capacity;
         # source-routed requests use no account, so a closed/empty pool must
@@ -4213,18 +4236,32 @@ async def v1_chat_completions(
     if prohibit_fast_mode and _is_fast_mode_model_alias(effective_model):
         effective_model = responses_payload.model
     validate_model_access(api_key, responses_payload.model)
+    source_route_attempted = not responses_shaped_payload and payload.messages is not None
     source_selection = (
         await _select_chat_model_source(
             responses_payload.model,
             api_key,
             raw_model=effective_model,
             require_streaming=payload.stream is True,
         )
-        if not responses_shaped_payload and payload.messages is not None
+        if source_route_attempted
         else None
     )
     source = source_selection[0] if source_selection is not None else None
     request_model = source_selection[1] if source_selection is not None else responses_payload.model
+    if source is None and source_route_attempted:
+        # Before any reservation is taken, so a refusal strands nothing.
+        disabled_denial = await _disabled_model_source_denial(
+            request,
+            responses_payload.model,
+            api_key,
+            route="chat",
+            raw_model=effective_model,
+            require_streaming=payload.stream is True,
+            headers=rate_limit_headers,
+        )
+        if disabled_denial is not None:
+            return disabled_denial
     if source is None:
         apply_enforced_service_tier_model_fallback(
             responses_payload,
@@ -4360,7 +4397,15 @@ async def _select_chat_model_source(
     *,
     raw_model: str | None = None,
     require_streaming: bool = False,
+    only_disabled: bool = False,
 ) -> tuple[ModelSource, str] | None:
+    """Resolve ``model`` to a Chat Completions-capable model source, if any.
+
+    ``only_disabled`` mirrors :func:`select_responses_model_source`: every rule
+    stays the same except the enabled-state filter, which is inverted, so the
+    result names the source the request would have used had it not been
+    switched off.
+    """
     assigned_source_ids = _allowed_source_ids_for_api_key(api_key)
     exact_allowed_models = set(api_key.allowed_models) if api_key and api_key.allowed_models else None
     candidates = [candidate for candidate in (raw_model, model) if candidate]
@@ -4380,6 +4425,7 @@ async def _select_chat_model_source(
                 candidate,
                 allowed_source_ids=assigned_source_ids,
                 require_streaming=require_streaming,
+                only_disabled=only_disabled,
             )
             if source is not None:
                 break
@@ -4398,6 +4444,7 @@ async def _select_responses_model_source(
     *,
     raw_model: str | None = None,
     require_streaming: bool = False,
+    only_disabled: bool = False,
 ) -> tuple[ModelSource, str] | None:
     # Shared with the WebSocket path so both transports agree on which models
     # belong to a model source.
@@ -4406,7 +4453,64 @@ async def _select_responses_model_source(
         api_key,
         raw_model=raw_model,
         require_streaming=require_streaming,
+        only_disabled=only_disabled,
+    )
+
+
+async def _disabled_model_source_denial(
+    request: Request,
+    model: str,
+    api_key: ApiKeyData | None,
+    *,
+    route: Literal["chat", "responses"],
+    raw_model: str | None = None,
+    require_streaming: bool = False,
+    headers: Mapping[str, str] | None = None,
+) -> JSONResponse | None:
+    """Refuse a request whose model source exists but is switched off.
+
+    Called only after the ordinary lookup missed. A miss has two very different
+    causes that the routing code cannot otherwise tell apart: the model belongs
+    to nobody, or it belongs to a model source (or a source model) an operator
+    disabled. Only the first may fall through to subscription selection --
+    handing a source-owned slug to a ChatGPT account produces
+    ``The '<model>' model is not supported when using Codex with a ChatGPT
+    account.``, which burns the account's health signal and tells the operator
+    nothing about the source they switched off.
+
+    Returns ``None`` when no disabled source claims the model, leaving every
+    other request on its existing path.
+    """
+    selection = (
+        await _select_responses_model_source(
+            model,
+            api_key,
+            raw_model=raw_model,
+            require_streaming=require_streaming,
+            only_disabled=True,
+        )
+        if route == "responses"
+        else await _select_chat_model_source(
+            model,
+            api_key,
+            raw_model=raw_model,
+            require_streaming=require_streaming,
+            only_disabled=True,
+        )
+    )
+    if selection is None:
+        return None
+    source, matched_model = selection
+    # The source name is operator-facing configuration, not a client-visible
+    # identifier, so the envelope names the model and the condition only.
+    reason = "is disabled" if not source.is_enabled else "has that model disabled"
+    error = openai_error(
+        "model_source_disabled",
+        f"The model '{matched_model}' is served by an OpenAI-compatible model source that {reason}. "
+        "Enable the source and its model in codex-lb, or request a different model.",
+        error_type="upstream_error",
     )
+    return _logged_error_json_response(request, 503, error, headers=headers)
 
 
 async def _select_embeddings_model_source(model: str, api_key: ApiKeyData | None) -> ModelSource | None:

app/modules/proxy/helpers.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import re
 from typing import Iterable
 
 from pydantic import ValidationError
@@ -40,6 +41,29 @@
     {"server_error", "upstream_error", "stream_incomplete", "overloaded_error", "server_is_overloaded"}
 )
 _MODEL_CAPACITY_MESSAGE_MARKERS = ("selected model is at capacity",)
+_MODEL_UNSUPPORTED_MESSAGE_RE = re.compile(
+    r"^The '.+' model is not supported when using Codex with a ChatGPT account\.$"
+)
+
+
+def is_model_scoped_upstream_rejection(message: str | None) -> bool:
+    """Match the ChatGPT model-entitlement rejection for *any* requested model.
+
+    The rejection names the model, not the account: it reproduces on every
+    request for that model and says nothing about whether the serving account
+    can still stream the models it is entitled to. Callers use it to keep the
+    rejection out of account health while leaving failover alone -- a different
+    account may hold a different entitlement.
+
+    Unlike ``_is_account_model_unsupported_error`` this does not require the
+    caller to know the requested model or the normalized error code. Upstream
+    delivers this rejection over the Codex WebSocket with neither ``code`` nor
+    ``type`` populated, which normalizes to the ``upstream_error`` fallback, so
+    a code-gated match misses it on the live stream path.
+    """
+    if message is None:
+        return False
+    return _MODEL_UNSUPPORTED_MESSAGE_RE.fullmatch(" ".join(message.split())) is not None
 
 
 def _is_account_model_unsupported_error(