fix(proxy-responses): tolerate structured content type values

app/core/openai/requests.py

@@ -45,6 +45,7 @@
 _COMPACT_TOOL_CALL_OUTPUT_ITEM_TYPES = frozenset(
     {"function_call_output", "custom_tool_call_output", "apply_patch_call_output"}
 )
+_EXPLICIT_PROMPT_CACHE_CONTENT_TYPES = frozenset({"input_text", "input_image", "input_file"})
 _GOAL_CONTINUATION_CONTEXT_PREFIX = '<codex_internal_context source="goal">'
 _PLAN_MODE_CONTEXT_PREFIX = "<collaboration_mode># Plan Mode"
 
@@ -800,6 +801,7 @@ def to_payload(self) -> JsonObject:
 def _strip_unsupported_fields(payload: MutableJsonObject) -> MutableJsonObject:
     _normalize_openai_compatible_aliases(payload)
     _normalize_service_tier_aliases(payload)
+    _strip_subscription_prompt_cache_controls(payload)
     _sanitize_interleaved_reasoning_input(payload)
     _strip_poisoned_local_compact_fallback_items(payload)
     # ``tools`` is deliberately NOT canonicalized here: the wire payload must
@@ -812,6 +814,55 @@ def _strip_unsupported_fields(payload: MutableJsonObject) -> MutableJsonObject:
     return payload
 
 
+def responses_request_has_explicit_prompt_cache_controls(payload: ResponsesRequest) -> bool:
+    """Whether a request asks for public-API explicit prompt caching."""
+
+    extra = payload.model_extra
+    if isinstance(extra, dict) and "prompt_cache_options" in extra:
+        return True
+    return _contains_explicit_prompt_cache_breakpoint(payload.input)
+
+
+def _contains_explicit_prompt_cache_breakpoint(value: JsonValue) -> bool:
+    if isinstance(value, list):
+        return any(_contains_explicit_prompt_cache_breakpoint(item) for item in value)
+    if not isinstance(value, dict):
+        return False
+    value_type = value.get("type")
+    if (
+        isinstance(value_type, str)
+        and value_type in _EXPLICIT_PROMPT_CACHE_CONTENT_TYPES
+        and "prompt_cache_breakpoint" in value
+    ):
+        return True
+    return any(_contains_explicit_prompt_cache_breakpoint(child) for child in value.values())
+
+
+def _strip_subscription_prompt_cache_controls(payload: MutableJsonObject) -> None:
+    """Remove controls rejected by the Codex subscription upstream.
+
+    OpenAI-compatible model sources use ``model_dump_for_forwarding`` and do
+    not pass through this subscription-only serializer.
+    """
+
+    payload.pop("prompt_cache_options", None)
+    _strip_subscription_prompt_cache_breakpoints(payload.get("input"))
+
+
+def _strip_subscription_prompt_cache_breakpoints(value: JsonValue | None) -> None:
+    if isinstance(value, list):
+        for item in value:
+            _strip_subscription_prompt_cache_breakpoints(item)
+        return
+    if not isinstance(value, dict):
+        return
+    value_type = value.get("type")
+    if isinstance(value_type, str) and value_type in _EXPLICIT_PROMPT_CACHE_CONTENT_TYPES:
+        value.pop("prompt_cache_breakpoint", None)
+    for child in value.values():
+        _strip_subscription_prompt_cache_breakpoints(child)
+
+
 def _strip_poisoned_local_compact_fallback_items(payload: MutableJsonObject) -> None:
     input_value = payload.get("input")
     if not is_json_list(input_value):

app/modules/proxy/api.py

@@ -126,6 +126,7 @@
     ResponsesRequest,
     extract_input_file_ids,
     normalize_tool_type,
+    responses_request_has_explicit_prompt_cache_controls,
 )
 from app.core.openai.v1_requests import V1ResponsesCompactRequest, V1ResponsesRequest
 from app.core.request_locality import (
@@ -302,6 +303,14 @@
 )
 _PUBLIC_RESPONSES_PRE_CREATED_BUFFER_LIMIT = 64
 _SOURCE_LIMITED_STREAM_BUFFER_BYTES = 16 * 1024 * 1024
+_PROMPT_CACHE_MODE_HEADER = "X-Codex-LB-Prompt-Cache-Mode"
+_SUBSCRIPTION_IMPLICIT_PROMPT_CACHE_MODE = "subscription-implicit"
+
+
+def _mark_subscription_prompt_cache_fallback(response: Response, payload: ResponsesRequest) -> Response:
+    if response.status_code < 400 and responses_request_has_explicit_prompt_cache_controls(payload):
+        response.headers[_PROMPT_CACHE_MODE_HEADER] = _SUBSCRIPTION_IMPLICIT_PROMPT_CACHE_MODE
+    return response
 
 
 class _V1ResetCreditFreshCredentials:
@@ -1054,7 +1063,7 @@ async def responses(
         service_tier_was_enforced=service_tier_was_enforced,
     )
 
-    return await _stream_responses(
+    response = await _stream_responses(
         request,
         responses_payload,
         context,
@@ -1069,6 +1078,7 @@ async def responses(
         enforce_openai_sdk_contract=openai_sdk_request,
         native_codex_heartbeat=native_codex_heartbeat,
     )
+    return _mark_subscription_prompt_cache_fallback(response, responses_payload)
 
 
 @router.get("/opportunistic/admission")
@@ -1181,7 +1191,7 @@ async def v1_responses(
         service_tier_was_enforced=service_tier_was_enforced,
     )
     if responses_payload.stream:
-        return await _stream_responses(
+        response = await _stream_responses(
             request,
             responses_payload,
             context,
@@ -1191,16 +1201,18 @@ async def v1_responses(
             prefer_http_bridge=True,
             prohibit_fast_mode=prohibit_fast_mode,
         )
-    return await _collect_responses(
-        request,
-        responses_payload,
-        context,
-        api_key,
-        codex_session_affinity=False,
-        openai_cache_affinity=True,
-        prefer_http_bridge=True,
-        prohibit_fast_mode=prohibit_fast_mode,
-    )
+    else:
+        response = await _collect_responses(
+            request,
+            responses_payload,
+            context,
+            api_key,
+            codex_session_affinity=False,
+            openai_cache_affinity=True,
+            prefer_http_bridge=True,
+            prohibit_fast_mode=prohibit_fast_mode,
+        )
+    return _mark_subscription_prompt_cache_fallback(response, responses_payload)
 
 
 @internal_router.post(

openspec/changes/adapt-subscription-prompt-cache-controls/proposal.md

@@ -0,0 +1,39 @@
+# Adapt subscription prompt-cache controls
+
+## Why
+
+The public OpenAI Responses API supports GPT-5.6 explicit prompt caching through
+`prompt_cache_options` and per-content `prompt_cache_breakpoint` markers. The
+Codex subscription upstream currently rejects both controls before response
+creation. A client using the documented public shape therefore receives a 400
+through codex-lb even though the same request can still use subscription-side
+implicit caching and `prompt_cache_key` affinity.
+
+Model-source requests are different: an OpenAI-compatible API-key source may
+support the public controls and must receive them unchanged. The adaptation
+therefore belongs at the subscription egress boundary, not in shared request
+validation.
+
+## What Changes
+
+- Subscription Responses egress omits `prompt_cache_options` and explicit
+  breakpoint markers while preserving prompt content, order, and
+  `prompt_cache_key`.
+- Successful HTTP responses that used this fallback expose
+  `X-Codex-LB-Prompt-Cache-Mode: subscription-implicit` so clients do not
+  mistake the fallback for exact explicit-prefix caching.
+- OpenAI-compatible model-source egress preserves the explicit controls.
+
+## Capabilities
+
+### Modified Capabilities
+
+- `responses-api-compat`: documented prompt-cache controls are adapted only for
+  the subscription upstream and their semantic downgrade is observable.
+
+## Impact
+
+- Code: Responses request serialization and HTTP route response metadata.
+- Tests: subscription and model-source regressions at `/v1/responses`.
+- API/schema: one informational response header; no database or configuration
+  change.

openspec/changes/adapt-subscription-prompt-cache-controls/specs/responses-api-compat/spec.md

@@ -0,0 +1,37 @@
+# responses-api-compat Delta Specification
+
+## ADDED Requirements
+
+### Requirement: Subscription Responses adapts unsupported explicit prompt-cache controls
+
+The proxy MUST omit public explicit prompt-cache controls from an HTTP Responses
+request routed to the Codex subscription upstream. This applies to
+`prompt_cache_options` and `prompt_cache_breakpoint` on a supported prompt
+content block. It MUST preserve the prompt content and ordering and MUST
+continue forwarding a client-supplied `prompt_cache_key` unchanged.
+
+A successful HTTP response for such a request MUST include
+`X-Codex-LB-Prompt-Cache-Mode: subscription-implicit`, because subscription
+implicit caching and account affinity do not provide the exact explicit-prefix
+semantics requested by the client. The proxy MUST NOT include that downgrade
+header when the request is routed to an OpenAI-compatible model source, and the
+model-source wire payload MUST preserve the explicit controls unchanged.
+
+#### Scenario: Subscription request falls back to implicit caching
+
+- **GIVEN** a `/v1/responses` request contains a `prompt_cache_key`,
+  `prompt_cache_options`, and an explicit breakpoint on an `input_text` block
+- **WHEN** the request is routed to a subscription account
+- **THEN** the upstream subscription payload omits `prompt_cache_options` and
+  the breakpoint
+- **AND** preserves the input text, input order, and `prompt_cache_key`
+- **AND** a successful response reports
+  `X-Codex-LB-Prompt-Cache-Mode: subscription-implicit`
+
+#### Scenario: Model source preserves public explicit-cache semantics
+
+- **GIVEN** the same `/v1/responses` request is routed to an OpenAI-compatible
+  model source
+- **THEN** the model-source payload retains `prompt_cache_options`, every
+  explicit breakpoint, and `prompt_cache_key`
+- **AND** the response does not report a subscription implicit fallback

openspec/changes/adapt-subscription-prompt-cache-controls/tasks.md

@@ -0,0 +1,21 @@
+# Tasks: adapt-subscription-prompt-cache-controls
+
+## 1. Implementation
+
+- [x] 1.1 Strip public explicit prompt-cache controls only from subscription
+      Responses egress while preserving `prompt_cache_key` and prompt content
+- [x] 1.2 Report successful subscription fallback through
+      `X-Codex-LB-Prompt-Cache-Mode: subscription-implicit`
+- [x] 1.3 Preserve explicit controls on OpenAI-compatible model-source egress
+
+## 2. Regression coverage
+
+- [x] 2.1 Exercise the exact `/v1/responses` subscription request shape and
+      assert upstream serialization plus response header
+- [x] 2.2 Exercise the model-source route as a negative control and assert the
+      controls remain intact
+
+## 3. Verification
+
+- [x] 3.1 Run focused unit/integration tests and strict OpenSpec validation
+- [ ] 3.2 Re-run the bounded live request against the overlay-preserving stack

tests/integration/test_openai_compat_features.py

@@ -5,7 +5,9 @@
 from typing import cast
 
 import pytest
+from fastapi.responses import JSONResponse
 
+import app.modules.proxy.api as proxy_api_module
 import app.modules.proxy.service as proxy_module
 from app.core.openai.requests import ResponsesRequest
 
@@ -246,6 +248,91 @@ async def fake_stream(payload, headers, access_token, account_id, base_url=None,
     assert "prompt_cache_retention" not in seen["payload"]
 
 
+@pytest.mark.asyncio
+async def test_v1_responses_downgrades_explicit_prompt_cache_for_subscription(async_client, monkeypatch):
+    await _import_account(async_client, "acc_prompt_cache_explicit", "prompt-cache-explicit@example.com")
+
+    seen = {}
+
+    async def fake_stream(payload, headers, access_token, account_id, base_url=None, raise_for_status=False):
+        seen["payload"] = payload.to_payload()
+        yield _completed_event("resp_prompt_cache_explicit")
+
+    monkeypatch.setattr(proxy_module, "core_stream_responses", fake_stream)
+
+    payload = {
+        "model": "gpt-5.6-sol",
+        "input": [
+            {
+                "role": "user",
+                "content": [
+                    {
+                        "type": "input_text",
+                        "text": "stable prefix",
+                        "prompt_cache_breakpoint": {"mode": "explicit"},
+                    },
+                    {"type": "input_text", "text": "changing suffix"},
+                ],
+            }
+        ],
+        "prompt_cache_key": "explicit-thread",
+        "prompt_cache_options": {"mode": "explicit"},
+    }
+    resp = await async_client.post("/v1/responses", json=payload)
+
+    assert resp.status_code == 200
+    assert resp.headers["x-codex-lb-prompt-cache-mode"] == "subscription-implicit"
+    forwarded = seen["payload"]
+    assert forwarded["prompt_cache_key"] == "explicit-thread"
+    assert "prompt_cache_options" not in forwarded
+    assert "prompt_cache_breakpoint" not in forwarded["input"][0]["content"][0]
+    assert forwarded["input"][0]["content"][0]["text"] == "stable prefix"
+
+
+@pytest.mark.asyncio
+async def test_v1_responses_preserves_explicit_prompt_cache_for_model_source(async_client, monkeypatch):
+    await _import_account(async_client, "acc_prompt_cache_source", "prompt-cache-source@example.com")
+
+    seen = {}
+    source = object()
+
+    async def fake_select(model, api_key, *, raw_model=None, require_streaming=False):
+        return source, model
+
+    async def fake_source_response(request, payload, *, source, api_key, rate_limit_headers):
+        seen["payload"] = payload.model_dump_for_forwarding()
+        return JSONResponse({"id": "resp_prompt_cache_source", "status": "completed", "output": []})
+
+    monkeypatch.setattr(proxy_api_module, "_select_responses_model_source", fake_select)
+    monkeypatch.setattr(proxy_api_module, "_source_responses_response", fake_source_response)
+
+    payload = {
+        "model": "gpt-5.6-sol",
+        "input": [
+            {
+                "role": "user",
+                "content": [
+                    {
+                        "type": "input_text",
+                        "text": "stable prefix",
+                        "prompt_cache_breakpoint": {"mode": "explicit"},
+                    }
+                ],
+            }
+        ],
+        "prompt_cache_key": "source-thread",
+        "prompt_cache_options": {"mode": "explicit"},
+    }
+    resp = await async_client.post("/v1/responses", json=payload)
+
+    assert resp.status_code == 200
+    assert "x-codex-lb-prompt-cache-mode" not in resp.headers
+    forwarded = seen["payload"]
+    assert forwarded["prompt_cache_options"] == {"mode": "explicit"}
+    assert forwarded["input"][0]["content"][0]["prompt_cache_breakpoint"] == {"mode": "explicit"}
+    assert forwarded["prompt_cache_key"] == "source-thread"
+
+
 @pytest.mark.asyncio
 async def test_v1_responses_normalizes_prompt_cache_aliases(async_client, monkeypatch):
     await _import_account(async_client, "acc_prompt_cache_alias", "prompt-cache-alias@example.com")

tests/unit/test_proxy_api_key_usage.py

@@ -93,3 +93,30 @@ def test_estimate_api_key_request_usage_uses_conservative_input_for_file_referen
     budget = estimate_api_key_request_usage(payload)
 
     assert budget.input_tokens is None
+
+
+def test_estimate_api_key_request_usage_allows_structured_content_type_values() -> None:
+    payload = ResponsesRequest.model_validate(
+        {
+            "model": "gpt-5.5",
+            "instructions": "continue",
+            "input": [
+                {
+                    "role": "assistant",
+                    "content": [
+                        {
+                            "type": {
+                                "namespace": "multi_agent_v1",
+                                "name": "tool_search_output",
+                            },
+                            "text": "deferred tool metadata",
+                        }
+                    ],
+                }
+            ],
+        }
+    )
+
+    budget = estimate_api_key_request_usage(payload)
+
+    assert budget.input_tokens is not None