fix(proxy): preserve continuation anchors during input normalization

app/core/openai/requests.py

@@ -10,6 +10,7 @@
     BaseModel,
     ConfigDict,
     Field,
+    ModelWrapValidatorHandler,
     PrivateAttr,
     SerializeAsAny,
     SkipValidation,
@@ -640,6 +641,14 @@ class ResponsesTextControls(BaseModel):
     format: ResponsesTextFormat | None = None
 
 
+@dataclass(frozen=True, slots=True)
+class _ResponsesRawInputProvenance:
+    raw_input: str | list[JsonValue]
+    normalized_value: JsonValue
+    original_instructions: str
+    normalized_instructions: str
+
+
 class ResponsesRequest(BaseModel):
     model_config = ConfigDict(extra="allow")
     _codex_lb_client_reasoning_effort: str | None = PrivateAttr(default=None)
@@ -650,6 +659,33 @@ class ResponsesRequest(BaseModel):
     # subscription-overflow dispatch (anchor rule and the source-direction
     # body); the ChatGPT-bound serialization never consults it.
     _codex_lb_client_store: bool | None = PrivateAttr(default=None)
+    _codex_lb_raw_input_provenance: _ResponsesRawInputProvenance | None = PrivateAttr(default=None)
+    _codex_lb_input_shape_wire_version: str | None = PrivateAttr(default="2")
+    _codex_lb_legacy_owner_forwarding_input_shape: bool = PrivateAttr(default=False)
+
+    @model_validator(mode="wrap")
+    @classmethod
+    def _capture_raw_input_string_provenance(
+        cls,
+        data: JsonValue | ResponsesRequest,
+        handler: ModelWrapValidatorHandler[ResponsesRequest],
+    ) -> ResponsesRequest:
+        if isinstance(data, cls):
+            return handler(data)
+        raw_input = data.get("input") if is_json_mapping(data) else None
+        raw_instructions = data.get("instructions") if is_json_mapping(data) else None
+        request = handler(data)
+        request._codex_lb_raw_input_provenance = (
+            _ResponsesRawInputProvenance(
+                raw_input=raw_input,
+                normalized_value=request.input,
+                original_instructions=raw_instructions if isinstance(raw_instructions, str) else "",
+                normalized_instructions=request.instructions,
+            )
+            if isinstance(raw_input, str) or is_json_list(raw_input)
+            else None
+        )
+        return request
 
     @model_validator(mode="before")
     @classmethod
@@ -750,10 +786,10 @@ def model_dump_for_forwarding(self) -> MutableJsonObject:
 
         Like ``model_dump(mode="json", exclude_none=True)`` but without
         synthesizing fields the client never sent. Used by every path that
-        forwards this request as a JSON body — the multi-instance owner
-        forward (``HTTPBridgeOwnerClient``) and model-source Responses
-        egress — so that field omission survives the hop and the receiving
-        side does not re-mark ``tools`` as explicitly set.
+        forwards this request to a model source so that field omission
+        survives the hop and the receiver does not re-mark ``tools`` as
+        explicitly set. The multi-instance owner wire has a dedicated dump
+        because it must also preserve pre-validation input shape.
         """
         payload: MutableJsonObject = self.model_dump(mode="json", exclude_none=True)
         if "tools" not in self.model_fields_set:
@@ -770,6 +806,23 @@ def model_dump_for_forwarding(self) -> MutableJsonObject:
             payload.pop("tools", None)
         return payload
 
+    def model_dump_for_http_bridge_owner_forwarding(self) -> MutableJsonObject:
+        """Dump the exact request body posted to an internal bridge owner.
+
+        Request validation normalizes a client string into one input item.
+        Preserve the original string on this internal hop so the owner can
+        revalidate it without changing the full-resend boundary. Keep model
+        source forwarding normalized; only the internal owner needs to repeat
+        local request-shape classification.
+        """
+        payload = self.model_dump_for_forwarding()
+        provenance = self._codex_lb_raw_input_provenance
+        if provenance is not None and self.input is provenance.normalized_value:
+            payload["input"] = provenance.raw_input
+            if self.instructions == provenance.normalized_instructions:
+                payload["instructions"] = provenance.original_instructions
+        return payload
+
     def to_payload(self) -> JsonObject:
         payload = _strip_unsupported_fields(self.model_dump_for_forwarding())
         _normalize_compaction_trigger_singleton(payload)

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

@@ -71,6 +71,7 @@
 )
 from app.core.resilience.overload import local_overload_error
 from app.core.types import JsonValue
+from app.core.utils.json_guards import is_json_list, is_json_mapping
 from app.core.utils.request_id import get_request_id
 from app.core.utils.shared_future import _await_task_deferring_cancellation, wait_on_shared_future
 from app.core.utils.sse import format_sse_event, parse_sse_data_json
@@ -131,6 +132,7 @@
 )
 from app.modules.proxy._service.support import (
     _HARD_HTTP_BRIDGE_AFFINITY_KINDS,  # noqa: F401
+    _PENDING_TOOL_CALL_OUTPUT_ITEM_TYPES,
     _REQUEST_TRANSPORT_HTTP,
     _WEBSOCKET_FULL_REPLAY_WAIT_POLL_SECONDS,  # noqa: F401
     _http_bridge_session_supports_service_tier,
@@ -2078,20 +2080,73 @@ def _http_bridge_session_retiring_with_visible_requests(session: "_HTTPBridgeSes
 
 
 def _http_bridge_payload_looks_like_full_resend(payload: ResponsesRequest) -> bool:
-    input_value = payload.input
+    raw_input_provenance = payload._codex_lb_raw_input_provenance
+    if raw_input_provenance is not None and payload.input is raw_input_provenance.normalized_value:
+        input_value = raw_input_provenance.raw_input
+    else:
+        input_value = payload.input
+
     if isinstance(input_value, str):
         return len(input_value) >= 4096
     if isinstance(input_value, Sequence) and not isinstance(input_value, (str, bytes, bytearray)):
         if len(input_value) > 1:
-            return True
+            if payload._codex_lb_legacy_owner_forwarding_input_shape:
+                return True
+            # Multiple outputs can be one delta-only continuation for
+            # parallel calls completed by the anchored response. Without the
+            # corresponding call items this payload is not self-contained and
+            # must keep its durable anchor.
+            return not all(
+                _http_bridge_input_item_type(item) in _PENDING_TOOL_CALL_OUTPUT_ITEM_TYPES for item in input_value
+            )
         if len(input_value) == 1:
+            if (
+                not payload._codex_lb_legacy_owner_forwarding_input_shape
+                and _http_bridge_input_item_type(input_value[0]) in _PENDING_TOOL_CALL_OUTPUT_ITEM_TYPES
+            ):
+                return False
+            if payload._codex_lb_legacy_owner_forwarding_input_shape:
+                normalized_text_length = _http_bridge_legacy_normalized_input_text_length(input_value)
+                if normalized_text_length is not None:
+                    return normalized_text_length >= 4096
+                # Pre-classifier owners measured a one-item array's item,
+                # rather than the array envelope. Keep that predicate for
+                # genuinely array-shaped legacy forwards so a large item is
+                # not silently downgraded to a delta.
+                try:
+                    return len(json.dumps(input_value[0], ensure_ascii=True, separators=(",", ":"))) >= 4096
+                except (OverflowError, TypeError, ValueError):
+                    return False
             try:
-                return len(json.dumps(input_value[0], ensure_ascii=True, separators=(",", ":"))) >= 4096
-            except TypeError:
+                return len(json.dumps(input_value, ensure_ascii=True, separators=(",", ":"))) >= 4096
+            except (OverflowError, TypeError, ValueError):
                 return False
     return False
 
 
+def _http_bridge_legacy_normalized_input_text_length(input_value: Sequence[JsonValue]) -> int | None:
+    """Recognize the canonical one-item array produced from a raw string.
+
+    A legacy owner receives the normalized array without the original string
+    provenance. Its envelope is implementation detail, so classify this
+    exact shape by the contained text length and use the legacy item-size
+    rule only for genuinely array-shaped inputs.
+    """
+    if len(input_value) != 1 or not is_json_mapping(input_value[0]):
+        return None
+    item = input_value[0]
+    if set(item) != {"role", "content"} or item.get("role") != "user":
+        return None
+    content = item.get("content")
+    if not is_json_list(content) or len(content) != 1 or not is_json_mapping(content[0]):
+        return None
+    part = content[0]
+    if set(part) != {"type", "text"} or part.get("type") != "input_text":
+        return None
+    text = part.get("text")
+    return len(text) if isinstance(text, str) else None
+
+
 def _preferred_http_bridge_reconnect_turn_state(session: "_HTTPBridgeSession") -> str | None:
     if (
         session.codex_session
@@ -2547,6 +2602,15 @@ def _durable_bridge_lookup_active_owner(lookup: DurableBridgeLookup | None) -> s
     return lookup.owner_instance_id
 
 
+def _owner_process_epoch(
+    lookup: DurableBridgeLookup | None,
+    owner_instance: str,
+) -> str | None:
+    if lookup is None or lookup.owner_instance_id != owner_instance:
+        return None
+    return lookup.owner_process_epoch
+
+
 def _durable_bridge_lookup_allows_local_reuse(
     lookup: DurableBridgeLookup | None,
     *,
@@ -3238,13 +3302,15 @@ def _http_bridge_turn_state_anchor_for_owner_failure(
     headers: Mapping[str, str],
     previous_response_id: str | None,
 ) -> str | None:
-    """Return the turn-state anchor when an owner forward failed unreachable."""
+    """Return turn state for an unreachable owner or typed pre-dispatch shape failure."""
 
     if previous_response_id is not None:
         return None
     turn_state = _sticky_key_from_turn_state_header(headers)
     if turn_state is None:
         return None
+    if exc.failure_phase == "owner_forward" and exc.failure_detail == "owner_input_shape_upgrade_required":
+        return turn_state
     payload = exc.payload
     if not isinstance(payload, dict):
         return None
@@ -3455,6 +3521,8 @@ def _http_bridge_reconnect_connect_failure(
 
 
 def _http_bridge_should_attempt_local_previous_response_recovery(exc: ProxyResponseError) -> bool:
+    if exc.failure_phase == "owner_forward" and exc.failure_detail == "owner_input_shape_upgrade_required":
+        return True
     payload = exc.payload
     if not isinstance(payload, dict):
         return False
@@ -3599,6 +3667,8 @@ def _http_bridge_should_attempt_local_bootstrap_rebind(
         return False
     if _sticky_key_from_turn_state_header(headers) is not None:
         return False
+    if exc.failure_phase == "owner_forward" and exc.failure_detail == "owner_input_shape_upgrade_required":
+        return True
     payload = exc.payload
     if not isinstance(payload, dict):
         return False

app/modules/proxy/_service/http_bridge/mixin.py

@@ -47,10 +47,7 @@
 )
 from app.core.utils.locks import fast_lock
 from app.core.utils.request_id import ensure_request_scope_id
-from app.db.models import (
-    DashboardSettings,
-    StickySessionKind,
-)
+from app.db.models import DashboardSettings, StickySessionKind
 from app.modules.api_keys.service import (
     ApiKeyData,
     ApiKeyRequestUsageBudget,
@@ -116,6 +113,7 @@
     _log_http_bridge_event,
     _log_http_bridge_startup_wait_timeout,
     _mark_http_bridge_reader_handoff_reconnect_failed,
+    _owner_process_epoch,
     _persist_http_bridge_replacement_account,
     _persistent_http_bridge_affinity,
     _plan_http_bridge_lru_capacity_closes,
@@ -1056,6 +1054,7 @@ def bind_account_neutral_recovery_owner(session: _HTTPBridgeSession) -> None:
                                             owner_instance=owner_instance,
                                             owner_endpoint=owner_endpoint,
                                             key=key,
+                                            owner_process_epoch=_owner_process_epoch(durable_lookup, owner_instance),
                                         )
                             else:
                                 if _http_bridge_has_durable_recovery_anchor(

app/modules/proxy/_service/http_bridge/owner_forwarding.py

@@ -2,6 +2,7 @@
 
 import asyncio
 import logging
+from dataclasses import replace
 from enum import StrEnum
 from typing import Any, AsyncIterator, Mapping, TypeVar
 
@@ -155,7 +156,9 @@
 from app.modules.proxy.http_bridge_forwarding import (
     HTTPBridgeForwardContext,
     OwnerForwardRelayFailure,
+    _http_bridge_owner_forward_requires_shape_upgrade,
 )
+from app.modules.proxy.ring_membership import HTTP_BRIDGE_INPUT_SHAPE_CLASSIFIER_CAPABILITY
 
 logger = logging.getLogger("app.modules.proxy.service")
 T = TypeVar("T")
@@ -476,13 +479,38 @@ def owner_response_rejected() -> None:
             if api_key_reservation is not None:
                 _signal_propagated_responses_owner_forward_rejected()
 
+        owner_supports_input_shape_classifier = False
+        if (
+            _http_bridge_owner_forward_requires_shape_upgrade(payload)
+            and self._ring_membership is not None
+            and owner_forward.owner_process_epoch is not None
+        ):
+            try:
+                owner_supports_input_shape_classifier = await self._ring_membership.owner_supports_capability(
+                    owner_forward.owner_instance,
+                    owner_process_epoch=owner_forward.owner_process_epoch,
+                    capability=HTTP_BRIDGE_INPUT_SHAPE_CLASSIFIER_CAPABILITY,
+                )
+            except Exception:
+                logger.debug(
+                    "Failed to prove HTTP bridge owner input-shape capability",
+                    exc_info=True,
+                )
+
+        if owner_supports_input_shape_classifier:
+            forward_context = replace(
+                forward_context,
+                expected_owner_process_epoch=owner_forward.owner_process_epoch,
+            )
+
         try:
             async for event_block in self._http_bridge_owner_client.stream_responses(
                 owner_endpoint=owner_forward.owner_endpoint,
                 payload=payload,
                 headers=forward_headers,
                 context=forward_context,
                 request_started_at=request_started_at,
+                owner_supports_input_shape_classifier=owner_supports_input_shape_classifier,
                 on_request_dispatched=owner_request_dispatched,
                 on_response_rejected=owner_response_rejected,
                 on_response_wait=_signal_propagated_capacity_startup_wait,