fix(core): preserve Responses usage and failure semantics

.fork/features/core.md

@@ -229,3 +229,15 @@ Notes:
 - `RotatingClient.acompletion()` routing logic unchanged
 - `list_models()` and `get_model()` unchanged
 - Ref: b3nw/LLM-API-Key-Proxy#110
+
+## 06.September.2026 — Preserve Responses usage, multimodal input, and failure semantics
+
+Branch: `fix/core-responses-integrity-upstream`; base: `dev`.
+
+- Preserve cached/reasoning token details, including explicit zeros, through Responses and Codex conversions.
+- Keep image tool output out of serialized tool text; preserve call IDs, text, image URLs, and image detail.
+- Emit one terminal failure for upstream errors or truncated streams instead of empty completion.
+- Classify streamed exceptions with null data correctly; stop context-overflow rotation and scope Codex model-eligibility cooldowns to the affected credential/model without exhausting global quota.
+- Files: `src/proxy_app/{main,responses_compat}.py`, `src/rotator_library/client/executor.py`, `src/rotator_library/error_handler.py`, `src/rotator_library/providers/codex_provider.py`, `src/rotator_library/usage/{manager,tracking/engine}.py`, `tests/test_responses_compat.py`, `tests/test_responses_integrity.py`.
+- Verification: changed-file `uv run python3 -m py_compile` and `uv run ruff check --select F401,F811,F821,E9`; full `uv run --with pytest --with pytest-asyncio --with pytest-mock python3 -m pytest -q`. Exact results accompany the PR evidence.
+- Boundaries: no model aliases, deployment configuration, dependency changes, speculative affinity changes, or claim that intermittent upstream cache misses are eliminated.

src/proxy_app/main.py

@@ -1234,16 +1234,14 @@ async def responses_stream_wrapper():
                         events = converter.convert_chunk(chunk_str)
                         if events:
                             yield events
+                    events = converter.finish()
+                    if events:
+                        yield events
                 except Exception as e:
                     logging.error(f"Error during responses stream: {e}")
-                    error_event = {
-                        "type": "error",
-                        "error": {
-                            "type": "server_error",
-                            "message": str(e),
-                        },
-                    }
-                    yield f"event: error\ndata: {json.dumps(error_event)}\n\n"
+                    events = converter.fail({"type": "server_error", "message": str(e)})
+                    if events:
+                        yield events
 
             return StreamingResponse(
                 responses_stream_wrapper(),

src/proxy_app/responses_compat.py

@@ -109,13 +109,31 @@ def convert_responses_input_to_messages(
         elif item_type == "function_call_output":
             call_id = item.get("call_id", "")
             output = item.get("output", "")
-            if isinstance(output, dict) or isinstance(output, list):
+            # Chat tool messages are text-only. Lift image parts into an adjacent
+            # user message, preserving the tool result ID and never tokenizing
+            # image bytes as JSON text. Codex converts these back to input_image.
+            images = []
+            if isinstance(output, list) and any(
+                isinstance(part, dict) and part.get("type") in ("input_image", "image_url")
+                for part in output
+            ):
+                parts = _convert_input_content(output)
+                images = [part for part in parts if part["type"] == "image_url"]
+                output = "\n".join(part["text"] for part in parts if part["type"] == "text")
+            elif isinstance(output, list) and all(
+                isinstance(part, dict) and part.get("type") in ("input_text", "text")
+                for part in output
+            ):
+                output = _flatten_content(output)
+            elif isinstance(output, (dict, list)):
                 output = json.dumps(output)
             messages.append({
                 "role": "tool",
                 "tool_call_id": call_id,
                 "content": output,
             })
+            if images:
+                messages.append({"role": "user", "content": images})
 
         # Reasoning items are informational — skip them for the Chat Completions pipeline
 
@@ -262,6 +280,24 @@ def build_item_id(prefix: str = "msg") -> str:
     return f"{prefix}_{uuid.uuid4().hex[:24]}"
 
 
+def _convert_usage(usage: Any) -> Dict[str, Any]:
+    """Keep inclusive token totals and map optional cache/reasoning details."""
+    usage = usage or {}
+    result = {
+        "input_tokens": usage.get("prompt_tokens", 0),
+        "output_tokens": usage.get("completion_tokens", 0),
+        "total_tokens": usage.get("total_tokens", 0),
+    }
+    for source, target, field in (
+        ("prompt_tokens_details", "input_tokens_details", "cached_tokens"),
+        ("completion_tokens_details", "output_tokens_details", "reasoning_tokens"),
+    ):
+        details = usage.get(source) or {}
+        if details.get(field) is not None:
+            result[target] = {field: details[field]}
+    return result
+
+
 def convert_chat_response_to_responses(
     cc_response: Any,
     response_id: str,
@@ -311,14 +347,9 @@ def convert_chat_response_to_responses(
             status = "incomplete"
 
     # Build usage in Responses API format
-    usage_raw = resp_dict.get("usage", {})
-    usage = {
-        "input_tokens": usage_raw.get("prompt_tokens", 0),
-        "output_tokens": usage_raw.get("completion_tokens", 0),
-        "total_tokens": usage_raw.get("total_tokens", 0),
-    }
+    usage = _convert_usage(resp_dict.get("usage"))
 
-    return {
+    result = {
         "id": response_id,
         "object": "response",
         "created_at": resp_dict.get("created", int(time.time())),
@@ -328,6 +359,17 @@ def convert_chat_response_to_responses(
         "usage": usage,
         "metadata": request_data.get("metadata", {}),
     }
+    if resp_dict.get("error") is not None or not choices:
+        error = resp_dict.get("error") or {"message": "Upstream response contained no choices"}
+        if not isinstance(error, dict):
+            error = {"message": str(error)}
+        result["status"] = "failed"
+        result["error"] = {
+            "code": error.get("code") or error.get("type") or "server_error",
+            "message": error.get("message") or "Upstream request failed",
+        }
+        result["output"] = []
+    return result
 
 
 class ResponsesStreamConverter:
@@ -357,6 +399,7 @@ def __init__(self, response_id: str, model: str):
         self.finish_reason: Optional[str] = None
         self.output_index: Optional[int] = None
         self.next_output_index = 0
+        self.terminal = False
 
     def _sse(self, event_type: str, data: Dict[str, Any]) -> str:
         data["type"] = event_type
@@ -381,6 +424,9 @@ def convert_chunk(self, chunk_str: str) -> str:
         """Convert a single SSE chunk string from Chat Completions format to Responses API events."""
         events = ""
 
+        if self.terminal:
+            return ""
+
         if not chunk_str.strip() or not chunk_str.startswith("data:"):
             return ""
 
@@ -400,6 +446,9 @@ def convert_chunk(self, chunk_str: str) -> str:
             events += self._sse("response.created", {"response": self._build_response_shell()})
             events += self._sse("response.in_progress", {"response": self._build_response_shell()})
 
+        if chunk.get("error") is not None:
+            return events + self.fail(chunk["error"])
+
         choices = chunk.get("choices", [])
         if not choices:
             if chunk.get("usage"):
@@ -517,8 +566,35 @@ def convert_chunk(self, chunk_str: str) -> str:
 
         return events
 
+    def fail(self, error: Any) -> str:
+        """Emit exactly one terminal failure, including after partial output."""
+        if self.terminal:
+            return ""
+        self.terminal = True
+        if not isinstance(error, dict):
+            error = {"message": str(error)}
+        response = self._build_response_shell("failed")
+        response["error"] = {
+            "code": error.get("code") or error.get("type") or "server_error",
+            "message": error.get("message") or "Upstream request failed",
+        }
+        if self.usage is not None:
+            response["usage"] = _convert_usage(self.usage)
+        return self._sse("response.failed", {"response": response})
+
+    def finish(self) -> str:
+        """Reject an EOF without a terminal marker instead of empty success."""
+        if self.terminal:
+            return ""
+        return self.fail({"code": "incomplete_stream", "message": "Upstream stream ended without [DONE]"})
+
     def _finalize(self) -> str:
-        """Emit final events: output_item.done, content_part.done, response.completed."""
+        """Emit final events once, only after an explicit finish reason."""
+        if self.terminal:
+            return ""
+        if self.finish_reason is None:
+            return self.fail({"code": "incomplete_stream", "message": "Upstream stream ended without a finish reason"})
+        self.terminal = True
         events = ""
         output_items_by_index = []
 
@@ -605,11 +681,7 @@ def _finalize(self) -> str:
         # Build usage
         usage = None
         if self.usage:
-            usage = {
-                "input_tokens": self.usage.get("prompt_tokens", 0),
-                "output_tokens": self.usage.get("completion_tokens", 0),
-                "total_tokens": self.usage.get("total_tokens", 0),
-            }
+            usage = _convert_usage(self.usage)
 
         final_response = self._build_response_shell(status)
         final_response["output"] = output_items
@@ -677,7 +749,13 @@ def _convert_input_content(content: Any) -> Any:
             url = part.get("image_url", "")
             if isinstance(url, dict):
                 url = url.get("url", "")
-            cc_parts.append({"type": "image_url", "image_url": {"url": url}})
+            image = {"url": url}
+            detail = part.get("detail")
+            if isinstance(part.get("image_url"), dict):
+                detail = part["image_url"].get("detail", detail)
+            if detail is not None:
+                image["detail"] = detail
+            cc_parts.append({"type": "image_url", "image_url": image})
     return cc_parts
 
 

src/rotator_library/client/executor.py

@@ -1155,7 +1155,7 @@ async def _execute_streaming(
 
                                 except StreamedAPIError as e:
                                     last_exception = e
-                                    original = getattr(e, "data", e)
+                                    original = getattr(e, "data", None) or e
                                     classified = classify_error(original, provider)
                                     log_failure(
                                         api_key=cred,

src/rotator_library/error_handler.py

@@ -873,6 +873,11 @@ def classify_error(e: Exception, provider: Optional[str] = None) -> ClassifiedEr
     Returns:
         ClassifiedError with error_type, status_code, retry_after, etc.
     """
+    from .core.errors import StreamedAPIError
+
+    if isinstance(e, StreamedAPIError) and getattr(e, "data", None) is not None:
+        return classify_error(e.data, provider)
+
     if isinstance(e, dict):
         payload = e.get("error", e)
         if isinstance(payload, dict):
@@ -921,6 +926,32 @@ def classify_error(e: Exception, provider: Optional[str] = None) -> ClassifiedEr
                 )
 
     error_text = str(e)
+    if isinstance(e, httpx.HTTPStatusError):
+        error_text += " " + e.response.text
+    lowered = error_text.lower()
+    if any(marker in lowered for marker in (
+        "context_length_exceeded", "context_window_exceeded",
+        "exceeds the context window", "maximum context length",
+    )):
+        return ClassifiedError(
+            error_type="context_window_exceeded", original_exception=e, status_code=400
+        )
+    if provider == "codex" and "model" in lowered and (
+        "not supported when using codex with a chatgpt account" in lowered
+        or "not supported for this account" in lowered
+    ):
+        # Eligibility belongs to this credential/model pair, not all models or
+        # all credentials. A bounded model cooldown allows later recovery.
+        return ClassifiedError(
+            error_type="model_not_supported", original_exception=e,
+            status_code=400, retry_after=900,
+        )
+    if isinstance(e, (StreamedAPIError, dict)) and any(marker in lowered for marker in (
+        "server_is_overloaded", "servers are currently overloaded", "service_unavailable_error",
+    )):
+        return ClassifiedError(
+            error_type="server_error", original_exception=e, status_code=503
+        )
     error_type_name = type(e).__name__
     if (
         "MidStreamFallbackError" in error_type_name
@@ -1260,8 +1291,6 @@ def classify_error(e: Exception, provider: Optional[str] = None) -> ClassifiedEr
         )
 
     # StreamedAPIError: errors received inside SSE streams (e.g. Codex response.failed)
-    from .core.errors import StreamedAPIError
-
     if isinstance(e, StreamedAPIError):
         error_msg = str(e).lower()
         if any(

src/rotator_library/providers/codex_provider.py

@@ -725,7 +725,10 @@ def _convert_messages_to_responses_input(
                         elif ptype == "image_url":
                             image_url = part.get("image_url", {})
                             url = image_url.get("url", "") if isinstance(image_url, dict) else image_url
-                            parts.append({"type": "input_image", "image_url": url})
+                            image = {"type": "input_image", "image_url": url}
+                            if isinstance(image_url, dict) and image_url.get("detail") is not None:
+                                image["detail"] = image_url["detail"]
+                            parts.append(image)
                         elif ptype == "input_image":
                             parts.append({"type": "input_image", "image_url": part.get("image_url", "")})
                 if parts:
@@ -1099,11 +1102,12 @@ async def _parse_response_events(
                     total_tokens=u.get("total_tokens", 0),
                 )
                 input_details = u.get("input_tokens_details") or {}
-                cached = input_details.get("cached_tokens", 0) or 0
-                if cached:
-                    usage.prompt_tokens_details = {
-                        "cached_tokens": cached,
-                    }
+                cached = input_details.get("cached_tokens")
+                if cached is not None:
+                    usage.prompt_tokens_details = {"cached_tokens": cached}
+                reasoning = (u.get("output_tokens_details") or {}).get("reasoning_tokens")
+                if reasoning is not None:
+                    usage.completion_tokens_details = {"reasoning_tokens": reasoning}
 
             final_chunk = litellm.ModelResponse(
                 id=response_id,
@@ -1991,11 +1995,12 @@ async def _non_stream_response_inner(
                         )
                         # Map Responses API input_tokens_details to prompt_tokens_details
                         input_details = u.get("input_tokens_details") or {}
-                        cached = input_details.get("cached_tokens", 0) or 0
-                        if cached:
-                            usage.prompt_tokens_details = {
-                                "cached_tokens": cached,
-                            }
+                        cached = input_details.get("cached_tokens")
+                        if cached is not None:
+                            usage.prompt_tokens_details = {"cached_tokens": cached}
+                        reasoning = (u.get("output_tokens_details") or {}).get("reasoning_tokens")
+                        if reasoning is not None:
+                            usage.completion_tokens_details = {"reasoning_tokens": reasoning}
 
                 # Handle errors
                 elif kind == "response.failed":

src/rotator_library/usage/manager.py

@@ -2587,6 +2587,11 @@ async def _record_failure(
                     f"applying {cooldown_duration:.0f}s cooldown (backoff={backoff})"
                 )
 
+            elif error.error_type == "model_not_supported":
+                # Do not block the shared Codex quota group or unrelated models.
+                group_key = normalized_model
+                cooldown_duration = cooldown_duration or 900
+
             # Mark exhausted for quota errors with long cooldown
             elif error.error_type == "quota_exceeded":
                 if (

src/rotator_library/usage/tracking/engine.py

@@ -620,7 +620,7 @@ def _apply_cooldown(
 
         # Check if cooldown qualifies as exhaustion
         cooldown_duration = cooldown_until - now
-        if cooldown_duration >= self._config.exhaustion_cooldown_threshold:
+        if reason != "model_not_supported" and cooldown_duration >= self._config.exhaustion_cooldown_threshold:
             if self._config.fair_cycle.enabled and model_or_group:
                 fair_cycle_key = self._resolve_fair_cycle_key(model_or_group)
                 self._mark_exhausted(state, fair_cycle_key, f"cooldown_{reason}")

tests/test_responses_compat.py

@@ -206,7 +206,8 @@ def test_responses_stream_converter():
     assert "response.output_item.added" in events5
     assert "response.function_call_arguments.delta" in events5
 
-    # Test finalization
+    # Test finalization after the upstream finish marker.
+    converter.convert_chunk('data: {"choices":[{"delta":{},"finish_reason":"tool_calls"}]}')
     events_final = converter.convert_chunk("data: [DONE]")
     assert "response.function_call_arguments.done" in events_final
     assert "response.output_text.done" in events_final
@@ -278,6 +279,7 @@ def test_responses_stream_converter_emits_reasoning_lifecycle():
     assert "response.reasoning_summary_part.added" not in event_types3
 
     # Finalize — should emit summary_text.done, summary_part.done, output_item.done
+    converter.convert_chunk('data: {"choices":[{"delta":{},"finish_reason":"stop"}]}')
     events_final = parse_sse_events(converter.convert_chunk("data: [DONE]"))
     final_types = [e[0] for e in events_final]
     assert "response.reasoning_summary_text.done" in final_types
@@ -307,6 +309,7 @@ def test_responses_stream_converter_allocates_unique_output_indices():
     converter.convert_chunk(f"data: {json.dumps(chunk4)}")
 
     # Finalize
+    converter.convert_chunk('data: {"choices":[{"delta":{},"finish_reason":"tool_calls"}]}')
     events_final = parse_sse_events(converter.convert_chunk("data: [DONE]"))
 
     # Collect all output_index values from output_item.done events