diff --git a/app/modules/proxy/_service/http_bridge/request_submit.py b/app/modules/proxy/_service/http_bridge/request_submit.py
index 29823ca125..364179fbf4 100644
--- a/app/modules/proxy/_service/http_bridge/request_submit.py
+++ b/app/modules/proxy/_service/http_bridge/request_submit.py
@@ -122,6 +122,7 @@
     _upstream_response_create_max_bytes,
     _websocket_auth_failure_permanent_code,
     _websocket_auth_failure_requires_reauth,
+    _websocket_request_text_is_account_neutral_fresh_replay,
 )
 from app.modules.proxy._service.observability import (
     _hash_identifier as _hash_identifier,
@@ -3094,6 +3095,7 @@ def request_is_retryable(request_state: _WebSocketRequestState) -> bool:
             )
             if request_state.replay_count >= 1 and not additional_clean_close_retry:
                 return False
+            account_bound_replay = False
             if request_state.previous_response_id is not None:
                 require_preferred_reconnect = False
                 if account_neutral_recovery:
@@ -3125,11 +3127,26 @@ def request_is_retryable(request_state: _WebSocketRequestState) -> bool:
                 # Account-scoped uploaded files cannot be replayed on a
                 # different owner. Keep the preferred account mandatory for
                 # both silent recovery and clean-close recovery.
-                require_preferred_reconnect = account_neutral_recovery or request_state.file_required_preferred_account
+                candidate_text = (
+                    request_state.fresh_upstream_request_text
+                    if request_state.fresh_upstream_request_is_retry_safe and request_state.fresh_upstream_request_text
+                    else request_state.request_text
+                )
+                # The send boundary decorates durable operations with
+                # codex_lb_operation_id after selection. Keep that operation
+                # identity on its owner unless a dedicated rebind path has
+                # already replaced the operation ID.
+                candidate_portable = request_state.operation_id is None and (
+                    _websocket_request_text_is_account_neutral_fresh_replay(candidate_text)
+                )
                 request_text = _prepare_websocket_request_state_for_visible_output_replay(request_state)
-                if request_text is None:
+                if request_text is None or request_text != candidate_text:
                     return False
-                if account_neutral_recovery:
+                account_bound_replay = not candidate_portable
+                require_preferred_reconnect = (
+                    account_neutral_recovery or account_bound_replay or request_state.file_required_preferred_account
+                )
+                if account_neutral_recovery or account_bound_replay:
                     request_state.preferred_account_id = session.account.id
                 elif not request_state.file_required_preferred_account:
                     if hard_owner_bound and not model_fallback_replay and not fresh_hard_request_account_switch_allowed:
@@ -3210,7 +3227,7 @@ def request_is_retryable(request_state: _WebSocketRequestState) -> bool:
                 await self._reconnect_http_bridge_session(
                     session,
                     request_state=request_state,
-                    require_same_account=account_neutral_recovery,
+                    require_same_account=account_neutral_recovery or account_bound_replay,
                     require_preferred_account=True,
                     **reconnect_reader_kwargs,
                 )
@@ -3319,7 +3336,22 @@ async def _retry_http_bridge_precreated_auth_request(
         error_message: str | None,
     ) -> Literal["not_replayable", "retried", "failed"]:
         permanent_failure_code = _websocket_auth_failure_permanent_code(error_message)
-        request_text = _prepare_websocket_request_state_for_auth_replay(request_state)
+        bound_to_current_account = request_state.replay_required_account_id == session.account.id
+        if bound_to_current_account and (
+            _websocket_auth_failure_requires_reauth(error_message)
+            or request_state.auth_replay_counts_by_account.get(session.account.id, 0) > 0
+        ):
+            failure_code = permanent_failure_code or _WEBSOCKET_AUTH_INVALIDATED_FAILURE_CODE
+            await self._load_balancer.mark_permanent_failure(session.account, failure_code)
+            setattr(request_state, "account_health_error_handled", True)
+            request_state.force_refresh_account_id = None
+            request_state.preferred_account_id = None
+            request_state.excluded_account_ids.add(session.account.id)
+            return "not_replayable"
+        request_text = _prepare_websocket_request_state_for_auth_replay(
+            request_state,
+            current_account_id=session.account.id,
+        )
         if request_text is None:
             await self._load_balancer.mark_permanent_failure(session.account, permanent_failure_code)
             setattr(request_state, "account_health_error_handled", True)
@@ -3368,10 +3400,14 @@ async def _retry_http_bridge_precreated_auth_request(
             await self._reconnect_http_bridge_session(
                 session,
                 request_state=request_state,
-                require_same_account=is_http_bridge_account_neutral_replay(
-                    kind=session.key.affinity_kind,
-                    key=session.key.affinity_key,
+                require_same_account=(
+                    bound_to_current_account
+                    or is_http_bridge_account_neutral_replay(
+                        kind=session.key.affinity_kind,
+                        key=session.key.affinity_key,
+                    )
                 ),
+                require_preferred_account=bound_to_current_account,
             )
             request_text = self._http_bridge_text_with_account_installation_id(session, request_state, request_text)
             await _send_http_bridge_request_text_with_archive_id(session, request_state, request_text)
@@ -3413,13 +3449,13 @@ async def _retry_http_bridge_security_work_request(
             key=session.key.affinity_key,
         ):
             return False
-        retry_text = request_state.request_text
-        if not retry_text:
-            return False
         if request_state.file_required_preferred_account:
             return False
         if not _websocket_request_can_replay_before_visible_output(request_state):
             return False
+        retry_text = _prepare_websocket_request_state_for_account_switch(request_state)
+        if retry_text is None:
+            return False
 
         owner_account_id = session.account.id
         previous_replay_count = request_state.replay_count
@@ -3436,11 +3472,6 @@ async def _retry_http_bridge_security_work_request(
             session.turn_state_alias_registration_generations
         )
         previous_session_headers = session.headers
-        if request_state.previous_response_id is not None:
-            retry_text = _prepare_websocket_request_state_for_account_switch(request_state)
-        if retry_text is None:
-            return False
-
         request_state.preferred_account_id = None
         request_state.excluded_account_ids.add(owner_account_id)
         request_state.affinity_policy = replace(
diff --git a/app/modules/proxy/_service/http_bridge/service_stubs.py b/app/modules/proxy/_service/http_bridge/service_stubs.py
index 3db2e16c15..86ed81c9a3 100644
--- a/app/modules/proxy/_service/http_bridge/service_stubs.py
+++ b/app/modules/proxy/_service/http_bridge/service_stubs.py
@@ -452,6 +452,10 @@ def _prepare_websocket_request_state_for_account_switch(*args: Any, **kwargs: An
     return _service_global("_prepare_websocket_request_state_for_account_switch")(*args, **kwargs)
 
 
+def _websocket_request_text_is_account_neutral_fresh_replay(*args: Any, **kwargs: Any) -> Any:
+    return _service_global("_websocket_request_text_is_account_neutral_fresh_replay")(*args, **kwargs)
+
+
 def _matching_websocket_request_states_for_previous_response_error(*args: Any, **kwargs: Any) -> Any:
     return _service_global("_matching_websocket_request_states_for_previous_response_error")(*args, **kwargs)
 
diff --git a/app/modules/proxy/_service/streaming/retry.py b/app/modules/proxy/_service/streaming/retry.py
index 0b28853a44..c0d2c43abe 100644
--- a/app/modules/proxy/_service/streaming/retry.py
+++ b/app/modules/proxy/_service/streaming/retry.py
@@ -79,6 +79,7 @@
     is_upstream_model_capacity_error,
 )
 from app.modules.proxy.load_balancer import AccountLease, AccountSelection
+from app.modules.proxy.replay_safety import responses_payload_is_account_neutral_fresh_replay
 from app.modules.proxy.selection_errors import USAGE_LIMIT_REACHED, selection_failure_response
 
 _REQUEST_TRANSPORT_HTTP = "http"
@@ -176,7 +177,10 @@ def _verified_cross_transport_fresh_replay(
         stored_fingerprint=continuity_state.last_completed_input_prefix_fingerprint,
     ):
         return None
-    return payload.model_copy(update={"previous_response_id": None})
+    fresh_payload = payload.model_copy(update={"previous_response_id": None})
+    if not responses_payload_is_account_neutral_fresh_replay(fresh_payload.to_replay_safety_payload()):
+        return None
+    return fresh_payload
 
 
 def _effective_http_downstream_transport_policy(
@@ -396,6 +400,7 @@ async def _stream_with_retry(
         deferred_capacity_account: Account | None = None
         deferred_capacity_lease: AccountLease | None = None
         preferred_account_id: str | None = None
+        payload_replay_required_account_id: str | None = None
         file_preferred_account_id: str | None = rewritten_file_account_id
         require_preferred_account = False
         last_retryable_stream_error: _RetryableStreamError | None = None
@@ -577,18 +582,39 @@ async def _settle_process_network_budget_exhaustion(
             )
             settled = await _settle_stream_usage_before_pending_penalty(settlement)
 
+        def _authorize_payload_dispatch(account: Account) -> bool:
+            required_account_id = payload_replay_required_account_id
+            if required_account_id is not None and required_account_id != account.id:
+                raise ProxyResponseError(
+                    502,
+                    openai_error(
+                        "previous_response_owner_unavailable",
+                        "Request payload owner account is unavailable; retry later.",
+                        error_type="server_error",
+                    ),
+                )
+            return required_account_id is None and not responses_payload_is_account_neutral_fresh_replay(
+                payload.to_replay_safety_payload()
+            )
+
         def _move_verified_fresh_replay_from_owner(*, account_id: str, outcome: str) -> bool:
             # Only a proxy-injected owner anchor with locally verified full
             # input may move; the failed owner stays excluded so sticky
             # selection cannot immediately loop back to it.
-            nonlocal affinity, payload, preferred_account_id, require_preferred_account, verified_fresh_replay_payload
+            nonlocal affinity, payload, payload_replay_required_account_id
+            nonlocal preferred_account_id, require_preferred_account, verified_fresh_replay_payload
             if not (
                 require_preferred_account
                 and preferred_account_id == account_id
                 and verified_fresh_replay_payload is not None
             ):
                 return False
+            if not responses_payload_is_account_neutral_fresh_replay(
+                verified_fresh_replay_payload.to_replay_safety_payload()
+            ):
+                return False
             payload = verified_fresh_replay_payload
+            payload_replay_required_account_id = None
             verified_fresh_replay_payload = None
             excluded_account_ids.add(account_id)
             preferred_account_id = None
@@ -1037,6 +1063,13 @@ async def _retry_account_model_rejection(
                     yield format_sse_event(_facade()._proxy_request_timeout_event(request_id))
                     return
                 while True:
+                    effective_preferred_account_id = resolve_required_account_id(
+                        ("continuation", preferred_account_id),
+                        ("dispatched payload", payload_replay_required_account_id),
+                    )
+                    effective_require_preferred_account = (
+                        require_preferred_account or payload_replay_required_account_id is not None
+                    )
                     try:
                         selection = await proxy._select_account_with_budget_compatible(
                             deadline,
@@ -1050,7 +1083,7 @@ async def _retry_account_model_rejection(
                             model=payload.model,
                             service_tier=payload.service_tier,
                             exclude_account_ids=excluded_account_ids,
-                            preferred_account_id=preferred_account_id,
+                            preferred_account_id=effective_preferred_account_id,
                             require_security_work_authorized=require_security_work_authorized,
                             lease_kind="stream",
                             estimated_lease_tokens=estimated_lease_tokens,
@@ -1058,7 +1091,7 @@ async def _retry_account_model_rejection(
                             # verified-fresh replay branch below removes its
                             # anchor before it permits cross-account movement.
                             fallback_on_preferred_account_unavailable=not (
-                                require_preferred_account or file_required_preferred_account
+                                effective_require_preferred_account or file_required_preferred_account
                             ),
                         )
                     except ProxyResponseError as exc:
@@ -1846,6 +1879,7 @@ async def _retry_account_model_rejection(
                         )
                         try:
                             settlement = _StreamSettlement()
+                            register_payload_owner = _authorize_payload_dispatch(account)
                             inner_stream = proxy._stream_once(
                                 account,
                                 payload,
@@ -1887,8 +1921,21 @@ async def _retry_account_model_rejection(
                                 enforce_openai_sdk_contract=enforce_openai_sdk_contract,
                             )
                             try:
-                                async for line in inner_stream:
-                                    yield line
+                                try:
+                                    async for line in inner_stream:
+                                        if register_payload_owner:
+                                            payload_replay_required_account_id = account.id
+                                            register_payload_owner = False
+                                        yield line
+                                    if register_payload_owner:
+                                        payload_replay_required_account_id = account.id
+                                except BaseException as exc:
+                                    if register_payload_owner and not (
+                                        isinstance(exc, ProxyResponseError)
+                                        and is_confirmed_pre_dispatch_transport_error(exc)
+                                    ):
+                                        payload_replay_required_account_id = account.id
+                                    raise
                             finally:
                                 close_task = asyncio.create_task(
                                     inner_stream.aclose(),
diff --git a/app/modules/proxy/_service/support.py b/app/modules/proxy/_service/support.py
index fd11e00f20..8a79957e72 100644
--- a/app/modules/proxy/_service/support.py
+++ b/app/modules/proxy/_service/support.py
@@ -1066,6 +1066,9 @@ class _WebSocketRequestState:
     fresh_upstream_request_responses_lite_model: str | None = None
     request_stage: str = "first_turn"
     preferred_account_id: str | None = None
+    # Once an account-bound body has been dispatched, retries remain pinned to
+    # that owner even when stale-anchor recovery removes previous_response_id.
+    replay_required_account_id: str | None = None
     require_security_work_authorized: bool = False
     durable_capability_lineage_required: bool = False
     file_required_preferred_account: bool = False
diff --git a/app/modules/proxy/_service/websocket/helpers.py b/app/modules/proxy/_service/websocket/helpers.py
index 349895c2f3..83858cc645 100644
--- a/app/modules/proxy/_service/websocket/helpers.py
+++ b/app/modules/proxy/_service/websocket/helpers.py
@@ -339,6 +339,7 @@
 from app.modules.proxy.http_bridge_forwarding import (
     OwnerForwardRelayFailure as OwnerForwardRelayFailure,
 )
+from app.modules.proxy.replay_safety import responses_payload_is_account_neutral_fresh_replay
 
 
 def _facade() -> Any:
@@ -459,37 +460,75 @@ def _websocket_owner_switch_has_other_pending_requests(
     return any(pending is not request_state for pending in pending_requests)
 
 
-def _prepare_websocket_request_state_for_account_switch(
+def _websocket_request_text_is_account_neutral_fresh_replay(request_text: str | None) -> bool:
+    if not isinstance(request_text, str):
+        return False
+    try:
+        payload = json.loads(request_text)
+    except json.JSONDecodeError:
+        return False
+    if not isinstance(payload, dict):
+        return False
+    event_type = payload.get("type")
+    if event_type is not None and event_type != "response.create":
+        return False
+    payload.pop("type", None)
+    return responses_payload_is_account_neutral_fresh_replay(cast(dict[str, JsonValue], payload))
+
+
+def _bind_websocket_request_dispatch_owner(
+    request_state: "_WebSocketRequestState",
+    *,
+    account_id: str,
+    exact_request_text: str,
+) -> bool:
+    required_account_id = request_state.replay_required_account_id
+    if _websocket_request_text_is_account_neutral_fresh_replay(exact_request_text):
+        return required_account_id is None or required_account_id == account_id
+    if required_account_id is not None and required_account_id != account_id:
+        return False
+    request_state.preferred_account_id = account_id
+    request_state.replay_required_account_id = account_id
+    return True
+
+
+def _install_verified_fresh_replay(
     request_state: "_WebSocketRequestState",
+    *,
+    require_proxy_injected_previous_response_id: bool = True,
+    require_account_neutral: bool = True,
 ) -> str | None:
-    """Return an unsent request body only when moving accounts is proven safe."""
-    if request_state.previous_response_id is None:
-        return request_state.request_text
-    if not (
-        request_state.proxy_injected_previous_response_id
-        and request_state.fresh_upstream_request_is_retry_safe
-        and request_state.fresh_upstream_request_text
-    ):
+    if not (request_state.fresh_upstream_request_is_retry_safe and request_state.fresh_upstream_request_text):
         return None
-    try:
-        fresh_payload = json.loads(request_state.fresh_upstream_request_text)
-    except (TypeError, json.JSONDecodeError):
+    if require_proxy_injected_previous_response_id and not request_state.proxy_injected_previous_response_id:
         return None
-    fresh_input = fresh_payload.get("input")
-    if extract_input_file_ids(fresh_input):
-        # A retained full body can be replay-safe for text continuity while
-        # still naming an account-scoped uploaded file.  Keep its injected
-        # anchor instead of moving that file reference to another account.
+    fresh_request_text = request_state.fresh_upstream_request_text
+    account_neutral = _websocket_request_text_is_account_neutral_fresh_replay(fresh_request_text)
+    if require_account_neutral and not account_neutral:
         return None
-
-    request_state.request_text = request_state.fresh_upstream_request_text
+    replay_required_account_id = request_state.replay_required_account_id or request_state.preferred_account_id
+    if not account_neutral and replay_required_account_id is None:
+        return None
+    request_state.request_text = fresh_request_text
     request_state.previous_response_id = None
     request_state.preferred_account_id = None
+    request_state.replay_required_account_id = None if account_neutral else replay_required_account_id
     request_state.proxy_injected_previous_response_id = False
     request_state.fresh_upstream_request_is_retry_safe = False
     request_state.responses_lite_model = request_state.fresh_upstream_request_responses_lite_model
     _refresh_websocket_request_input_fingerprint_from_text(request_state)
-    return request_state.request_text
+    return fresh_request_text
+
+
+def _prepare_websocket_request_state_for_account_switch(
+    request_state: "_WebSocketRequestState",
+) -> str | None:
+    """Return an unsent request body only when moving accounts is proven safe."""
+    if request_state.previous_response_id is None:
+        if not _websocket_request_text_is_account_neutral_fresh_replay(request_state.request_text):
+            return None
+        return request_state.request_text
+    return _install_verified_fresh_replay(request_state)
 
 
 def _websocket_continuity_anchor_for_payload(
@@ -900,35 +939,43 @@ def _websocket_auth_request_can_switch_account(request_state: _WebSocketRequestS
     if request_state.file_required_preferred_account:
         return False
     if request_state.previous_response_id is None:
-        return True
+        return request_state.request_text is None or _websocket_request_text_is_account_neutral_fresh_replay(
+            request_state.request_text
+        )
     if not (
         request_state.proxy_injected_previous_response_id
         and request_state.fresh_upstream_request_is_retry_safe
         and request_state.fresh_upstream_request_text
     ):
         return False
-    return not _websocket_fresh_request_blocks_account_switch(request_state)
+    return _websocket_request_text_is_account_neutral_fresh_replay(
+        request_state.fresh_upstream_request_text
+    ) and not _websocket_fresh_request_blocks_account_switch(request_state)
 
 
 def _prepare_websocket_request_state_for_auth_replay(
     request_state: _WebSocketRequestState,
+    *,
+    current_account_id: str | None = None,
 ) -> str | None:
     if request_state.last_downstream_sequence_number is not None:
         return None
-    if not _websocket_auth_request_can_switch_account(request_state):
+    can_switch_account = _websocket_auth_request_can_switch_account(request_state)
+    can_retry_bound_owner = (
+        request_state.auth_replay_count == 0
+        and current_account_id is not None
+        and request_state.replay_required_account_id == current_account_id
+        and isinstance(request_state.request_text, str)
+    )
+    if not can_switch_account and not can_retry_bound_owner:
         return None
-    if (
+    if can_switch_account and (
         request_state.proxy_injected_previous_response_id
         and request_state.fresh_upstream_request_is_retry_safe
         and request_state.fresh_upstream_request_text
     ):
-        request_state.request_text = request_state.fresh_upstream_request_text
-        request_state.previous_response_id = None
-        request_state.preferred_account_id = None
-        request_state.proxy_injected_previous_response_id = False
-        request_state.fresh_upstream_request_is_retry_safe = False
-        request_state.responses_lite_model = request_state.fresh_upstream_request_responses_lite_model
-        _refresh_websocket_request_input_fingerprint_from_text(request_state)
+        if _install_verified_fresh_replay(request_state) is None:
+            return None
     request_text = request_state.request_text
     if not isinstance(request_text, str):
         return None
diff --git a/app/modules/proxy/_service/websocket/mixin.py b/app/modules/proxy/_service/websocket/mixin.py
index b1e948dab7..1c268f7ea9 100644
--- a/app/modules/proxy/_service/websocket/mixin.py
+++ b/app/modules/proxy/_service/websocket/mixin.py
@@ -383,8 +383,10 @@
 from app.modules.proxy._service.websocket.helpers import (
     _app_error_to_websocket_event,
     _assign_websocket_response_id,
+    _bind_websocket_request_dispatch_owner,
     _find_websocket_request_state_by_response_id,
     _forget_websocket_stale_previous_response,
+    _install_verified_fresh_replay,
     _is_websocket_response_create,
     _is_websocket_stale_previous_response,
     _match_websocket_request_state_for_anonymous_event,
@@ -2575,6 +2577,19 @@ def take_reader_replay_request_state() -> _WebSocketRequestState | None:
                     if text_data is not None:
                         archive_request_id = None if request_state is None else request_state.archive_request_id
                         if request_state is not None and payload is not None and _is_websocket_response_create(payload):
+                            if account is None or not _bind_websocket_request_dispatch_owner(
+                                request_state,
+                                account_id=account.id,
+                                exact_request_text=text_data,
+                            ):
+                                raise ProxyResponseError(
+                                    502,
+                                    openai_error(
+                                        "previous_response_owner_unavailable",
+                                        "Request payload owner account is unavailable; retry later.",
+                                        error_type="server_error",
+                                    ),
+                                )
                             request_state.response_create_sent_at = time.monotonic()
                         with _websocket_archive_request_context(archive_request_id):
                             await upstream.send_text(text_data)
@@ -3441,13 +3456,18 @@ async def _record_or_defer_confirmed_route_backoff(account: Account) -> None:
         for attempt in range(max_attempts):
             is_retry = attempt > 0
             forced_refresh_account_id = request_state.force_refresh_account_id
-            preferred_account_id = forced_refresh_account_id or request_state.preferred_account_id
+            preferred_account_id = (
+                request_state.replay_required_account_id
+                or forced_refresh_account_id
+                or request_state.preferred_account_id
+            )
             turn_state_owner_required = (
                 request_state.affinity_policy.codex_session_source == "turn_state"
                 and request_state.preferred_account_id is not None
             )
             require_preferred_account = (
                 (request_state.previous_response_id is not None and request_state.preferred_account_id is not None)
+                or request_state.replay_required_account_id is not None
                 or request_state.file_required_preferred_account
                 or turn_state_owner_required
             )
@@ -3761,6 +3781,14 @@ async def _heartbeat(remaining_seconds: float) -> None:
                 break
 
         account = selection.account
+        if (
+            account is not None
+            and request_state.replay_required_account_id is None
+            and request_state.request_text is not None
+            and not _facade()._websocket_request_text_is_account_neutral_fresh_replay(request_state.request_text)
+        ):
+            request_state.preferred_account_id = account.id
+            request_state.replay_required_account_id = account.id
         if (
             account is not None
             and require_preferred_account
@@ -5544,18 +5572,21 @@ async def _process_upstream_websocket_text(
                     # transparently retried.
                     retry_error_code = None
                 else:
-                    upstream_control.reconnect_requested = True
-                    request_state.request_text = request_state.fresh_upstream_request_text
-                    request_state.previous_response_id = None
-                    request_state.proxy_injected_previous_response_id = False
-                    request_state.fresh_upstream_request_is_retry_safe = False
-                    request_state.responses_lite_model = request_state.fresh_upstream_request_responses_lite_model
-                    request_state.replay_count += 1
-                    request_state.awaiting_response_created = True
-                    request_state.response_id = None
-                    _clear_websocket_request_error_overrides(request_state)
-                    upstream_control.suppress_downstream_event = True
-                    upstream_control.replay_request_state = request_state
+                    replay_text = _install_verified_fresh_replay(
+                        request_state,
+                        require_proxy_injected_previous_response_id=False,
+                        require_account_neutral=False,
+                    )
+                    if replay_text is None:
+                        retry_error_code = None
+                    else:
+                        upstream_control.reconnect_requested = True
+                        request_state.replay_count += 1
+                        request_state.awaiting_response_created = True
+                        request_state.response_id = None
+                        _clear_websocket_request_error_overrides(request_state)
+                        upstream_control.suppress_downstream_event = True
+                        upstream_control.replay_request_state = request_state
             else:
                 upstream_control.reconnect_requested = True
                 request_state.replay_count += 1
@@ -5678,10 +5709,31 @@ async def _handle_precreated_websocket_auth_failure(
     ) -> bool:
         proxy = cast(_WebSocketServiceProtocol, self)
         _ = proxy
-        if _prepare_websocket_request_state_for_auth_replay(request_state) is None:
+        bound_to_current_account = request_state.replay_required_account_id == account.id
+        requires_reauth = _websocket_auth_failure_requires_reauth(error_message)
+        if bound_to_current_account and (
+            requires_reauth or request_state.auth_replay_counts_by_account.get(account.id, 0) > 0
+        ):
+            failure_code = (
+                _facade()._WEBSOCKET_SESSION_EXPIRED_FAILURE_CODE
+                if requires_reauth
+                else _facade()._WEBSOCKET_AUTH_INVALIDATED_FAILURE_CODE
+            )
+            await proxy._load_balancer.mark_permanent_failure(account, failure_code)
+            request_state.force_refresh_account_id = None
+            request_state.preferred_account_id = None
+            request_state.excluded_account_ids.add(account.id)
+            return False
+        if (
+            _prepare_websocket_request_state_for_auth_replay(
+                request_state,
+                current_account_id=account.id,
+            )
+            is None
+        ):
             return False
 
-        if _websocket_auth_failure_requires_reauth(error_message):
+        if requires_reauth:
             failure_code = _facade()._WEBSOCKET_SESSION_EXPIRED_FAILURE_CODE
         elif request_state.auth_replay_counts_by_account.get(account.id, 0) == 0:
             request_state.auth_replay_counts_by_account[account.id] = 1
diff --git a/app/modules/proxy/service.py b/app/modules/proxy/service.py
index 898b41d88e..cf6fcbafd0 100644
--- a/app/modules/proxy/service.py
+++ b/app/modules/proxy/service.py
@@ -704,6 +704,7 @@
     _websocket_precreated_auth_error_code,  # noqa: F401
     _websocket_precreated_retry_error_code,  # noqa: F401
     _websocket_receive_timeout_for_pending_requests,  # noqa: F401
+    _websocket_request_text_is_account_neutral_fresh_replay,  # noqa: F401
     _websocket_response_id,  # noqa: F401
     _websocket_top_level_error_payload,  # noqa: F401
     _wrapped_websocket_error_event,  # noqa: F401
diff --git a/openspec/changes/archive/2026-08-19-bind-previous-response-replays-to-dispatch-owner/.openspec.yaml b/openspec/changes/archive/2026-08-19-bind-previous-response-replays-to-dispatch-owner/.openspec.yaml
new file mode 100644
index 0000000000..41c30bab88
--- /dev/null
+++ b/openspec/changes/archive/2026-08-19-bind-previous-response-replays-to-dispatch-owner/.openspec.yaml
@@ -0,0 +1,2 @@
+schema: spec-driven
+created: 2026-08-19
diff --git a/openspec/changes/archive/2026-08-19-bind-previous-response-replays-to-dispatch-owner/context.md b/openspec/changes/archive/2026-08-19-bind-previous-response-replays-to-dispatch-owner/context.md
new file mode 100644
index 0000000000..08de111df6
--- /dev/null
+++ b/openspec/changes/archive/2026-08-19-bind-previous-response-replays-to-dispatch-owner/context.md
@@ -0,0 +1,54 @@
+# Previous-response replay owner fencing
+
+## Purpose
+
+This change distinguishes continuation-anchor recovery from payload
+portability. A retry may safely remove a stale anchor yet still be forbidden
+from changing accounts because retained request items remain account-scoped.
+
+## Example
+
+Account A first receives:
+
+```json
+{
+  "previous_response_id": "resp_owner",
+  "input": [
+    {
+      "type": "reasoning",
+      "id": "rs_owner",
+      "encrypted_content": "owner-bound-ciphertext"
+    }
+  ]
+}
+```
+
+If a pre-visible failure triggers stale-anchor recovery, the proxy may remove
+`previous_response_id` only as part of a verified replay. Because the retained
+encrypted reasoning is not account-neutral, the replacement remains bound to
+account A. Account B must never receive it.
+
+An ordinary fresh request containing only portable user input can pass the
+canonical predicate and may use normal account selection.
+
+A selected account is not recorded as owner when transport evidence proves the
+request failed before dispatch. The body may then make its first real dispatch
+on another eligible account. Ambiguous failures remain pinned.
+
+HTTP bridge operation IDs are proxy-owned but still identify an in-flight
+operation. A bridge retry carrying an existing operation ID remains on its
+current account unless the operation is explicitly rebound before selection.
+
+## Operational Notes
+
+- Owner-unavailable failures are internal retry decisions; they do not add a
+  setting or require operator action.
+- Existing file ownership remains an independent strict pin.
+- Verified fresh-body installation clears the old dispatch owner atomically
+  with replacing the request body.
+- A bound request may perform one forced authentication refresh on the same
+  owner; it does not become eligible for cross-account auth failover.
+- API-key reservation settlement still completes before deferred account-health
+  writes.
+- The change covers HTTP streaming, HTTP bridge, and direct WebSocket paths so
+  operators do not observe transport-dependent account ownership.
diff --git a/openspec/changes/archive/2026-08-19-bind-previous-response-replays-to-dispatch-owner/design.md b/openspec/changes/archive/2026-08-19-bind-previous-response-replays-to-dispatch-owner/design.md
new file mode 100644
index 0000000000..2b0ab9f16c
--- /dev/null
+++ b/openspec/changes/archive/2026-08-19-bind-previous-response-replays-to-dispatch-owner/design.md
@@ -0,0 +1,110 @@
+## Context
+
+Responses requests can carry both a server-side continuation anchor and
+client-retained material. Removing a stale `previous_response_id` does not make
+the remaining body portable: encrypted reasoning, account-scoped probe items,
+and other retained state can still belong to the account that first received
+the request.
+
+The selector already supports strict required-account routing for known
+previous-response and file owners. The missing state is payload dispatch
+provenance: after a pre-visible retry excludes an account, later selection can
+no longer tell that the retained body was already dispatched there.
+
+## Goals / Non-Goals
+
+**Goals:**
+
+- Bind nonportable payloads to their first dispatch account.
+- Enforce the binding consistently in HTTP streaming, HTTP bridge, and direct
+  WebSocket retry paths.
+- Allow cross-account replay only after exact-wire verification proves the
+  resulting request is an account-neutral fresh replay.
+- Preserve existing settlement, health-write, and file-owner invariants.
+
+**Non-Goals:**
+
+- Changing stale previous-response error classification from PR #1818.
+- Preserving or reshaping unrelated bare/raw upstream error fields.
+- Changing public API envelopes, retry counts, quota accounting, or settings.
+- Making encrypted reasoning or account-scoped probe items portable.
+
+## Decisions
+
+### Use the canonical portability predicate
+
+Every candidate body is evaluated with
+`responses_payload_is_account_neutral_fresh_replay`. Ad hoc checks for files or
+`previous_response_id` are insufficient because account scope can live in
+retained input items.
+
+Alternative: extend each transport's file checks. Rejected because it
+duplicates an incomplete allowlist and already failed to catch encrypted
+reasoning.
+
+### Bind on first nonportable dispatch
+
+A request-local dispatch-owner ID is authorized before the first nonportable
+payload is sent and persisted after the first upstream event or normal stream
+completion. Ambiguous/post-dispatch failures also preserve that owner, while a
+positively confirmed pre-dispatch transport failure does not create one. Every
+later selection treats a persisted owner like any other strict continuity
+requirement.
+
+Alternative: infer ownership from the current preferred account. Rejected
+because retry branches intentionally clear or replace preference state.
+
+### Clear ownership only after verified neutral replay
+
+Verified stale-anchor recovery may replace the wire body with a reconstructed
+fresh request. The dispatch binding is cleared only when that exact replacement
+passes the canonical account-neutral predicate. Body replacement and
+owner-fence clearing occur in one transition so a retry cannot observe mixed
+state. A verified nonneutral replacement may be installed for a same-owner
+retry, but that transition preserves the existing dispatch-owner fence.
+
+Alternative: clear ownership whenever the anchor is removed. Rejected because
+the reproduced defect retained owner-bound ciphertext after anchor removal.
+
+### Treat proxy-owned operation metadata as account-bound
+
+HTTP bridge sends may add `codex_lb_operation_id` after request preparation.
+Until a dedicated rebind path replaces that operation identity, selection
+treats the request as nonportable and requires the current account.
+
+Alternative: remove the proxy-owned field before portability checks. Rejected
+because normalization would authorize a different account while preserving the
+same operation identity on the final wire request.
+
+### Fail closed across transport-specific recovery
+
+Trusted Access migration/degradation, owner exclusion, bridge reconnect, and
+WebSocket account switching may not bypass payload ownership. If the owner
+cannot satisfy the retry, the proxy returns the stable owner-unavailable error
+without dispatching the retained body elsewhere.
+
+A generic authentication failure is split into two decisions: one forced token
+refresh may replay a bound body on the same owner, while owner exclusion or
+cross-account migration still requires an atomically installed neutral body.
+Permanent authentication failure remains terminal for a bound body.
+
+## Risks / Trade-offs
+
+- **Fewer automatic retries for account-bound bodies** → This is intentional;
+  confidentiality and continuation correctness outrank cross-account fallback.
+- **False nonportability** → The canonical predicate is an explicit allowlist,
+  so unknown retained item types fail closed.
+- **Transport drift** → Shared helpers plus focused HTTP, bridge, and WebSocket
+  regressions keep the invariant aligned.
+- **Settlement regression** → The change does not move reservation settlement
+  or deferred health writes; existing settlement tests remain mandatory.
+
+## Migration Plan
+
+No data or configuration migration is required. Deploy the proxy code normally.
+Rollback is a code rollback; no persisted format changes.
+
+## Open Questions
+
+None. Current-main runtime probes reproduce the cross-account dispatch and the
+existing selector already provides the strict owner-routing primitive.
diff --git a/openspec/changes/archive/2026-08-19-bind-previous-response-replays-to-dispatch-owner/proposal.md b/openspec/changes/archive/2026-08-19-bind-previous-response-replays-to-dispatch-owner/proposal.md
new file mode 100644
index 0000000000..8ad755213e
--- /dev/null
+++ b/openspec/changes/archive/2026-08-19-bind-previous-response-replays-to-dispatch-owner/proposal.md
@@ -0,0 +1,46 @@
+## Why
+
+A pre-visible Responses retry can remove a `previous_response_id` anchor while
+retaining account-scoped request material, exclude the original account, and
+dispatch the retained payload to another account. Encrypted reasoning was
+reproduced crossing accounts through the HTTP stream path; the same missing
+dispatch provenance affects HTTP bridge and direct WebSocket retries.
+
+PR #1818 fixed parameterless stale-response classification but intentionally
+did not add payload-owner fencing. The remaining defect violates account
+ownership even when session/file continuity and API-key settlement work as
+designed.
+
+## What Changes
+
+- Classify exact-wire replay candidates with the canonical
+  account-neutral-fresh-replay predicate.
+- Bind every nonportable Responses payload to its first dispatch account.
+- Merge payload ownership with previous-response and file ownership during
+  every HTTP stream, HTTP bridge, and direct WebSocket selection.
+- Fail closed rather than excluding the owner or moving retained
+  account-scoped material during Trusted Access migration/degradation.
+- Clear payload ownership only after verified anchor removal produces a
+  canonical account-neutral fresh replay.
+- Keep proxy-owned operation metadata on its current account unless a
+  dedicated operation-rebind path replaces that identity before selection.
+- Preserve existing file pinning, API-key settlement ordering, error
+  classification, and raw error-envelope behavior.
+
+## Capabilities
+
+### Modified Capabilities
+
+- `responses-api-compat`: require account-bound retry payloads to remain on
+  their dispatch owner across all Responses transports.
+
+## Impact
+
+- **Affected code:** Responses replay safety, HTTP streaming retries, HTTP
+  bridge reconnects, and direct WebSocket account switching.
+- **Affected tests:** proxy streaming utilities and WebSocket Responses
+  integration tests.
+- **API/schema changes:** none.
+- **Configuration changes:** none.
+- **Security impact:** prevents account-scoped request material from crossing
+  account boundaries during internal retries.
diff --git a/openspec/changes/archive/2026-08-19-bind-previous-response-replays-to-dispatch-owner/specs/responses-api-compat/spec.md b/openspec/changes/archive/2026-08-19-bind-previous-response-replays-to-dispatch-owner/specs/responses-api-compat/spec.md
new file mode 100644
index 0000000000..e09e31f394
--- /dev/null
+++ b/openspec/changes/archive/2026-08-19-bind-previous-response-replays-to-dispatch-owner/specs/responses-api-compat/spec.md
@@ -0,0 +1,92 @@
+## ADDED Requirements
+
+### Requirement: Account-bound retries remain on their dispatch owner
+
+The proxy MUST bind a Responses request body that is not a canonical
+account-neutral fresh replay to the account that first receives that exact
+body. Every later selection for that request MUST treat the dispatch owner as a
+strict required account across HTTP streaming, HTTP bridge, and direct
+WebSocket transports.
+
+The proxy MUST NOT exclude the dispatch owner and send the retained body to a
+different account during stale-anchor recovery, retryable account failure,
+Trusted Access migration or degradation, bridge reconnect, or WebSocket account
+switching. If the required owner is unavailable, the proxy MUST fail closed
+without dispatching the retained body to another account.
+
+The proxy MAY perform one forced authentication refresh and replay a retained
+account-bound body on the same dispatch owner. It MUST NOT use that refresh to
+exclude the owner or migrate the body to another account, and a permanent
+authentication failure MUST remain terminal for the bound body.
+
+The proxy MAY clear the dispatch-owner binding only after verified recovery
+replaces the exact wire body and the replacement passes the canonical
+account-neutral-fresh-replay predicate. Removing `previous_response_id` alone
+MUST NOT make retained account-scoped input portable.
+
+Proxy-owned operation metadata that will be added at the send boundary MUST
+remain bound to the current account unless an explicit operation-rebind path
+replaces that identity before account selection. Installing a verified fresh
+body and clearing its dispatch-owner binding MUST occur as one state
+transition.
+
+#### Scenario: Encrypted reasoning remains on its first dispatch account
+
+- **GIVEN** account A first receives a Responses request containing encrypted
+  reasoning or another account-scoped retained item
+- **WHEN** a pre-visible retry excludes account A or requests a differently
+  authorized account
+- **THEN** the proxy does not dispatch the retained body to account B
+- **AND** the retry fails closed when account A is unavailable
+
+#### Scenario: Verified account-neutral fresh replay may change accounts
+
+- **GIVEN** verified recovery removes a stale continuation anchor
+- **AND** the exact replacement body contains only canonical account-neutral
+  fresh input
+- **WHEN** normal retry selection chooses account B
+- **THEN** the proxy may dispatch the replacement body to account B
+
+#### Scenario: Confirmed pre-dispatch failure does not create an owner
+
+- **GIVEN** account A is selected for a nonportable Responses body
+- **WHEN** transport evidence confirms the request failed before any upstream
+  bytes were dispatched
+- **THEN** the proxy does not record account A as the dispatch owner
+- **AND** normal retry selection may dispatch the body first on account B
+
+#### Scenario: HTTP bridge preserves payload ownership
+
+- **GIVEN** an HTTP bridge request has already dispatched a nonportable body to
+  account A
+- **WHEN** pre-created recovery or reconnect selection excludes account A
+- **THEN** the bridge does not submit that body on account B
+
+#### Scenario: Direct WebSocket preserves payload ownership
+
+- **GIVEN** a direct WebSocket request has already dispatched a nonportable body
+  to account A
+- **WHEN** retry handling prepares an account switch
+- **THEN** the proxy rejects the switch unless the exact replacement body is a
+  canonical account-neutral fresh replay
+
+#### Scenario: Bound authentication refresh stays on the owner
+
+- **GIVEN** a nonportable body is bound to account A
+- **WHEN** account A reports a refreshable authentication failure before
+  visible output
+- **THEN** the proxy may refresh and replay once on account A
+- **AND** it does not dispatch the retained body to account B
+
+#### Scenario: HTTP bridge operation identity remains on its owner
+
+- **GIVEN** an HTTP bridge retry retains a proxy-owned operation identity
+- **AND** no explicit operation rebind has replaced that identity
+- **WHEN** retry selection evaluates another account
+- **THEN** the bridge requires the current operation owner
+
+#### Scenario: Existing settlement ordering is unchanged
+
+- **GIVEN** an API-key reservation requires settlement during the failed retry
+- **WHEN** account health is updated
+- **THEN** required settlement still completes before deferred health writes
diff --git a/openspec/changes/archive/2026-08-19-bind-previous-response-replays-to-dispatch-owner/tasks.md b/openspec/changes/archive/2026-08-19-bind-previous-response-replays-to-dispatch-owner/tasks.md
new file mode 100644
index 0000000000..53d6b3264a
--- /dev/null
+++ b/openspec/changes/archive/2026-08-19-bind-previous-response-replays-to-dispatch-owner/tasks.md
@@ -0,0 +1,29 @@
+## 1. Regression coverage
+
+- [x] 1.1 Add a deterministic HTTP streaming regression proving account-bound
+      encrypted reasoning never dispatches to a Trusted Access replacement.
+- [x] 1.2 Add direct WebSocket regressions for unanchored and verified-fresh
+      account-bound request bodies.
+- [x] 1.3 Add HTTP bridge coverage for owner exclusion and account-neutral
+      replacement controls.
+- [x] 1.4 Add a confirmed pre-dispatch regression proving owner registration
+      waits for actual upstream dispatch.
+
+## 2. Owner-fencing implementation
+
+- [x] 2.1 Route every replay candidate through the canonical account-neutral
+      fresh-replay predicate.
+- [x] 2.2 Bind nonportable HTTP stream payloads to their first dispatch owner
+      and require that owner during later selections.
+- [x] 2.3 Enforce the same binding in HTTP bridge and direct WebSocket account
+      switching without changing settlement ordering.
+
+## 3. Verification and publication
+
+- [x] 3.1 Capture genuine focused RED, implement the minimal owner fence, and
+      run focused HTTP/bridge/WebSocket tests GREEN.
+- [x] 3.2 Run diagnostics, Ruff, typecheck, architecture gates, full affected
+      tests, and strict affected OpenSpec validation.
+- [x] 3.3 Execute an isolated real-surface account-switch scenario proving no
+      cross-account dispatch and an account-neutral control.
+- [x] 3.4 Complete independent review and sync the verified change for archive.
diff --git a/openspec/specs/responses-api-compat/context.md b/openspec/specs/responses-api-compat/context.md
index e2a75c6ca4..b9e4547395 100644
--- a/openspec/specs/responses-api-compat/context.md
+++ b/openspec/specs/responses-api-compat/context.md
@@ -195,6 +195,36 @@ retry once with full local history. If the original request already contained a
 self-contained full resend, codex-lb instead reconnects and replays that body
 without the rejected anchor.
 
+## Previous-response replay owner fencing
+
+Removing a stale continuation anchor does not make every retained body
+portable. Encrypted reasoning, account-scoped items, file references, and
+durable bridge operation identities remain owned by the account that first
+received them. The proxy records that dispatch owner and requires it on later
+HTTP streaming, HTTP bridge, and direct WebSocket selections.
+
+For example, if account A first receives encrypted reasoning and then returns a
+pre-visible Trusted Access or authentication failure, account B must never
+receive the retained ciphertext. One forced token refresh may replay the body
+on account A; permanent failure or owner unavailability fails closed.
+
+Verified recovery installs a replacement body and updates owner state
+atomically. A canonical account-neutral replacement clears the owner and may
+use normal failover. A verified nonneutral replacement, including a
+Responses-Lite full resend, may replay only on the same owner and preserves the
+fence.
+
+HTTP bridge tracing archive IDs do not pin neutral requests. A real durable
+`operation_id` does pin the request until an explicit operation-rebind path
+replaces that identity. Existing file pins and API-key settlement-before-health
+ordering remain independent invariants.
+
+Streaming selection authorizes owner compatibility before opening upstream, but
+persists a new owner only after dispatch is observed. A transport failure that
+is positively classified as pre-dispatch therefore leaves the body unowned and
+eligible for its first real dispatch on another account. Ambiguous failures
+remain owner-bound.
+
 ## Known Client Integrations (Reference)
 
 Third-party agents that consume the `/v1` Responses surface documented by this
diff --git a/openspec/specs/responses-api-compat/spec.md b/openspec/specs/responses-api-compat/spec.md
index b201e659e9..3bf0552710 100644
--- a/openspec/specs/responses-api-compat/spec.md
+++ b/openspec/specs/responses-api-compat/spec.md
@@ -5342,3 +5342,94 @@ Responses-compatible routes MUST accept the canonical `ultrafast` service tier a
 
 - **WHEN** upstream completes a request with `response.service_tier: "ultrafast"`
 - **THEN** the actual and billable request-log tiers are `ultrafast`
+
+### Requirement: Account-bound retries remain on their dispatch owner
+
+The proxy MUST bind a Responses request body that is not a canonical
+account-neutral fresh replay to the account that first receives that exact
+body. Every later selection for that request MUST treat the dispatch owner as a
+strict required account across HTTP streaming, HTTP bridge, and direct
+WebSocket transports.
+
+The proxy MUST NOT exclude the dispatch owner and send the retained body to a
+different account during stale-anchor recovery, retryable account failure,
+Trusted Access migration or degradation, bridge reconnect, or WebSocket account
+switching. If the required owner is unavailable, the proxy MUST fail closed
+without dispatching the retained body to another account.
+
+The proxy MAY perform one forced authentication refresh and replay a retained
+account-bound body on the same dispatch owner. It MUST NOT use that refresh to
+exclude the owner or migrate the body to another account, and a permanent
+authentication failure MUST remain terminal for the bound body.
+
+The proxy MAY clear the dispatch-owner binding only after verified recovery
+replaces the exact wire body and the replacement passes the canonical
+account-neutral-fresh-replay predicate. Removing `previous_response_id` alone
+MUST NOT make retained account-scoped input portable.
+
+Proxy-owned operation metadata that will be added at the send boundary MUST
+remain bound to the current account unless an explicit operation-rebind path
+replaces that identity before account selection. Installing a verified fresh
+body and clearing its dispatch-owner binding MUST occur as one state
+transition.
+
+#### Scenario: Encrypted reasoning remains on its first dispatch account
+
+- **GIVEN** account A first receives a Responses request containing encrypted
+  reasoning or another account-scoped retained item
+- **WHEN** a pre-visible retry excludes account A or requests a differently
+  authorized account
+- **THEN** the proxy does not dispatch the retained body to account B
+- **AND** the retry fails closed when account A is unavailable
+
+#### Scenario: Verified account-neutral fresh replay may change accounts
+
+- **GIVEN** verified recovery removes a stale continuation anchor
+- **AND** the exact replacement body contains only canonical account-neutral
+  fresh input
+- **WHEN** normal retry selection chooses account B
+- **THEN** the proxy may dispatch the replacement body to account B
+
+#### Scenario: Confirmed pre-dispatch failure does not create an owner
+
+- **GIVEN** account A is selected for a nonportable Responses body
+- **WHEN** transport evidence confirms the request failed before any upstream
+  bytes were dispatched
+- **THEN** the proxy does not record account A as the dispatch owner
+- **AND** normal retry selection may dispatch the body first on account B
+
+#### Scenario: HTTP bridge preserves payload ownership
+
+- **GIVEN** an HTTP bridge request has already dispatched a nonportable body to
+  account A
+- **WHEN** pre-created recovery or reconnect selection excludes account A
+- **THEN** the bridge does not submit that body on account B
+
+#### Scenario: Direct WebSocket preserves payload ownership
+
+- **GIVEN** a direct WebSocket request has already dispatched a nonportable body
+  to account A
+- **WHEN** retry handling prepares an account switch
+- **THEN** the proxy rejects the switch unless the exact replacement body is a
+  canonical account-neutral fresh replay
+
+#### Scenario: Bound authentication refresh stays on the owner
+
+- **GIVEN** a nonportable body is bound to account A
+- **WHEN** account A reports a refreshable authentication failure before
+  visible output
+- **THEN** the proxy may refresh and replay once on account A
+- **AND** it does not dispatch the retained body to account B
+
+#### Scenario: HTTP bridge operation identity remains on its owner
+
+- **GIVEN** an HTTP bridge retry retains a proxy-owned operation identity
+- **AND** no explicit operation rebind has replaced that identity
+- **WHEN** retry selection evaluates another account
+- **THEN** the bridge requires the current operation owner
+
+#### Scenario: Existing settlement ordering is unchanged
+
+- **GIVEN** an API-key reservation requires settlement during the failed retry
+- **WHEN** account health is updated
+- **THEN** required settlement still completes before deferred health writes
diff --git a/tests/integration/test_proxy_websocket_responses.py b/tests/integration/test_proxy_websocket_responses.py
index 70c1e2e265..16c5ee2e1e 100644
--- a/tests/integration/test_proxy_websocket_responses.py
+++ b/tests/integration/test_proxy_websocket_responses.py
@@ -4409,6 +4409,8 @@ def test_v1_responses_websocket_reuses_upstream_for_sequential_requests(app_inst
         ],
     )
     connect_calls: list[dict[str, object]] = []
+    dispatch_owner_snapshots: list[tuple[str | None, str | None]] = []
+    original_bind_dispatch_owner = websocket_mixin_module._bind_websocket_request_dispatch_owner
 
     class _FakeSettingsCache:
         async def get(self):
@@ -4449,7 +4451,18 @@ async def fake_connect_proxy_websocket(
                 "model": model,
             }
         )
-        return SimpleNamespace(id=f"acct_ws_proxy_{len(connect_calls)}"), first_upstream
+        return SimpleNamespace(id="acct_ws_proxy_owner"), first_upstream
+
+    def capture_dispatch_owner(*args, **kwargs):
+        bound = original_bind_dispatch_owner(*args, **kwargs)
+        request_state = args[0] if args else kwargs["request_state"]
+        dispatch_owner_snapshots.append(
+            (
+                request_state.preferred_account_id,
+                request_state.replay_required_account_id,
+            )
+        )
+        return bound
 
     async def fake_write_request_log(self, **kwargs):
         del self, kwargs
@@ -4459,6 +4472,11 @@ async def fake_write_request_log(self, **kwargs):
     monkeypatch.setattr(proxy_module, "get_settings_cache", lambda: _FakeSettingsCache())
     monkeypatch.setattr(proxy_module.ProxyService, "_connect_proxy_websocket", fake_connect_proxy_websocket)
     monkeypatch.setattr(proxy_module.ProxyService, "_write_request_log", fake_write_request_log)
+    monkeypatch.setattr(
+        websocket_mixin_module,
+        "_bind_websocket_request_dispatch_owner",
+        capture_dispatch_owner,
+    )
 
     first_request = {
         "type": "response.create",
@@ -4471,6 +4489,7 @@ async def fake_write_request_log(self, **kwargs):
         "type": "response.create",
         "model": "gpt-5.5",
         "input": "second",
+        "account_bound_probe": "owner-bound",
         "promptCacheKey": "thread_b",
         "stream": True,
     }
@@ -4489,6 +4508,10 @@ async def fake_write_request_log(self, **kwargs):
     assert connect_calls[0]["sticky_key"] == "thread_a"
     assert connect_calls[0]["sticky_kind"] == proxy_module.StickySessionKind.PROMPT_CACHE
     assert connect_calls[0]["model"] == "gpt-5.4"
+    assert dispatch_owner_snapshots == [
+        (None, None),
+        ("acct_ws_proxy_owner", "acct_ws_proxy_owner"),
+    ]
     _assert_upstream_payloads(
         first_upstream.sent_text,
         [
@@ -4505,6 +4528,7 @@ async def fake_write_request_log(self, **kwargs):
                 "model": "gpt-5.5",
                 "instructions": "",
                 "input": [{"role": "user", "content": [{"type": "input_text", "text": "second"}]}],
+                "account_bound_probe": "owner-bound",
                 "store": False,
                 "include": [],
                 "prompt_cache_key": "thread_b",
diff --git a/tests/unit/test_proxy_utils.py b/tests/unit/test_proxy_utils.py
index b8d03ed5a2..665eff1219 100644
--- a/tests/unit/test_proxy_utils.py
+++ b/tests/unit/test_proxy_utils.py
@@ -479,6 +479,150 @@ def test_websocket_account_switch_keeps_anchor_when_fresh_replay_references_file
     assert request_state.preferred_account_id == "acc_file_owner"
 
 
+def test_websocket_account_switch_blocks_unanchored_account_bound_request():
+    request_state = proxy_service._WebSocketRequestState(
+        request_id="req_ws_account_bound",
+        model="gpt-5.6-sol",
+        service_tier="priority",
+        reasoning_effort="high",
+        api_key_reservation=None,
+        started_at=time.monotonic(),
+        request_text=(
+            '{"type":"response.create","model":"gpt-5.6-sol","input":['
+            '{"type":"reasoning","id":"rs_owner","encrypted_content":"owner-bound"}]}'
+        ),
+        preferred_account_id="acc_owner",
+    )
+
+    assert websocket_mixin._prepare_websocket_request_state_for_account_switch(request_state) is None
+    assert request_state.preferred_account_id == "acc_owner"
+
+
+def test_websocket_account_switch_blocks_account_bound_fresh_replay():
+    fresh_text = (
+        '{"type":"response.create","model":"gpt-5.6-sol","input":['
+        '{"role":"user","content":"hello"},'
+        '{"type":"reasoning","id":"rs_owner","encrypted_content":"owner-bound"},'
+        '{"type":"function_call","name":"lookup","call_id":"call_1","arguments":"{}"},'
+        '{"type":"function_call_output","call_id":"call_1","output":"ok"}]}'
+    )
+    request_state = proxy_service._WebSocketRequestState(
+        request_id="req_ws_account_bound_fresh",
+        model="gpt-5.6-sol",
+        service_tier="priority",
+        reasoning_effort="high",
+        api_key_reservation=None,
+        started_at=time.monotonic(),
+        request_text='{"type":"response.create","previous_response_id":"resp_proxy"}',
+        previous_response_id="resp_proxy",
+        preferred_account_id="acc_owner",
+        proxy_injected_previous_response_id=True,
+        fresh_upstream_request_is_retry_safe=True,
+        fresh_upstream_request_text=fresh_text,
+    )
+
+    assert websocket_mixin._prepare_websocket_request_state_for_account_switch(request_state) is None
+    assert request_state.previous_response_id == "resp_proxy"
+    assert request_state.preferred_account_id == "acc_owner"
+
+
+def test_websocket_dispatch_owner_rejects_account_bound_socket_reuse():
+    request_text = (
+        '{"type":"response.create","model":"gpt-5.6-sol","input":['
+        '{"type":"reasoning","id":"rs_owner","encrypted_content":"owner-bound"}]}'
+    )
+    request_state = proxy_service._WebSocketRequestState(
+        request_id="req_ws_dispatch_owner",
+        model="gpt-5.6-sol",
+        service_tier=None,
+        reasoning_effort=None,
+        api_key_reservation=None,
+        started_at=time.monotonic(),
+        request_text=request_text,
+    )
+
+    assert websocket_mixin._bind_websocket_request_dispatch_owner(
+        request_state,
+        account_id="acc_owner",
+        exact_request_text=request_text,
+    )
+    assert not websocket_mixin._bind_websocket_request_dispatch_owner(
+        request_state,
+        account_id="acc_other",
+        exact_request_text=request_text,
+    )
+    assert request_state.preferred_account_id == "acc_owner"
+    assert request_state.replay_required_account_id == "acc_owner"
+
+
+def test_websocket_verified_fresh_replay_clears_dispatch_owner_atomically():
+    fresh_text = '{"type":"response.create","model":"gpt-5.6-sol","input":"portable user input"}'
+    request_state = proxy_service._WebSocketRequestState(
+        request_id="req_ws_verified_fresh_owner",
+        model="gpt-5.6-sol",
+        service_tier=None,
+        reasoning_effort=None,
+        api_key_reservation=None,
+        started_at=time.monotonic(),
+        request_text='{"type":"response.create","previous_response_id":"resp_proxy"}',
+        previous_response_id="resp_proxy",
+        preferred_account_id="acc_owner",
+        replay_required_account_id="acc_owner",
+        proxy_injected_previous_response_id=True,
+        fresh_upstream_request_is_retry_safe=True,
+        fresh_upstream_request_text=fresh_text,
+    )
+
+    assert websocket_mixin._install_verified_fresh_replay(request_state) == fresh_text
+    assert request_state.request_text == fresh_text
+    assert request_state.previous_response_id is None
+    assert request_state.preferred_account_id is None
+    assert request_state.replay_required_account_id is None
+
+
+def test_websocket_bound_auth_replay_allows_one_same_owner_refresh():
+    request_text = (
+        '{"type":"response.create","model":"gpt-5.6-sol","input":['
+        '{"type":"reasoning","id":"rs_owner","encrypted_content":"owner-bound"}]}'
+    )
+    request_state = proxy_service._WebSocketRequestState(
+        request_id="req_ws_bound_auth_refresh",
+        model="gpt-5.6-sol",
+        service_tier=None,
+        reasoning_effort=None,
+        api_key_reservation=None,
+        started_at=time.monotonic(),
+        request_text=request_text,
+        preferred_account_id="acc_owner",
+        replay_required_account_id="acc_owner",
+    )
+
+    assert (
+        websocket_mixin._prepare_websocket_request_state_for_auth_replay(
+            request_state,
+            current_account_id="acc_other",
+        )
+        is None
+    )
+    assert request_state.auth_replay_count == 0
+    assert (
+        websocket_mixin._prepare_websocket_request_state_for_auth_replay(
+            request_state,
+            current_account_id="acc_owner",
+        )
+        == request_text
+    )
+    assert request_state.auth_replay_count == 1
+    assert request_state.replay_required_account_id == "acc_owner"
+    assert (
+        websocket_mixin._prepare_websocket_request_state_for_auth_replay(
+            request_state,
+            current_account_id="acc_owner",
+        )
+        is None
+    )
+
+
 def test_websocket_owner_switch_detects_other_pending_request() -> None:
     current = proxy_service._WebSocketRequestState(
         request_id="req_owner_switch",
@@ -17248,6 +17392,134 @@ async def fake_stream(payload, headers, access_token, account_id, base_url=None,
     assert authorized_lease in released_leases
 
 
+@pytest.mark.asyncio
+async def test_stream_responses_account_bound_pre_dispatch_failure_retries_other_account(monkeypatch):
+    settings = _make_proxy_settings()
+    service = proxy_service.ProxyService(_repo_factory(_RequestLogsRecorder()))
+    first_account = _make_account("acc_pre_dispatch_bound_first")
+    second_account = _make_account("acc_pre_dispatch_bound_second")
+    select_account = AsyncMock(
+        side_effect=[
+            AccountSelection(account=first_account, error_message=None),
+            AccountSelection(account=second_account, error_message=None),
+        ]
+    )
+    attempted_account_ids: list[str] = []
+
+    async def fake_stream(payload, headers, access_token, account_id, base_url=None, raise_for_status=False):
+        del payload, headers, access_token, base_url, raise_for_status
+        attempted_account_ids.append(account_id)
+        if account_id == first_account.chatgpt_account_id:
+            raise _pre_dispatch_proxy_connect_error("first bound account proxy route unavailable")
+        yield 'data: {"type":"response.completed","response":{"id":"resp_bound_fallback"}}\n\n'
+
+    monkeypatch.setattr(proxy_service, "get_settings_cache", lambda: _SettingsCache(settings))
+    monkeypatch.setattr(proxy_service, "get_settings", lambda: settings)
+    monkeypatch.setattr(service._load_balancer, "select_account", select_account)
+    monkeypatch.setattr(service._load_balancer, "record_error", AsyncMock())
+    monkeypatch.setattr(service._load_balancer, "record_success", AsyncMock())
+    monkeypatch.setattr(service, "_ensure_fresh", AsyncMock(side_effect=lambda account, **kwargs: account))
+    monkeypatch.setattr(proxy_service, "core_stream_responses", fake_stream)
+
+    payload = ResponsesRequest.model_validate(
+        {
+            "model": "gpt-5.1",
+            "instructions": "",
+            "input": [
+                {
+                    "type": "reasoning",
+                    "id": "rs_pre_dispatch_owner",
+                    "encrypted_content": "owner-bound",
+                }
+            ],
+            "stream": True,
+        }
+    )
+
+    chunks = [chunk async for chunk in service.stream_responses(payload, {"session_id": "sid-pre-dispatch"})]
+
+    assert select_account.await_count == 2
+    assert attempted_account_ids == [
+        first_account.chatgpt_account_id,
+        second_account.chatgpt_account_id,
+    ]
+    assert any("resp_bound_fallback" in chunk for chunk in chunks)
+
+
+@pytest.mark.asyncio
+async def test_stream_responses_refuses_account_bound_security_work_retry(monkeypatch):
+    settings = _make_proxy_settings()
+    service = proxy_service.ProxyService(_repo_factory(_RequestLogsRecorder()))
+    regular_account = _make_account("acc_regular_security_account_bound")
+    authorized_account = _make_account("acc_authorized_security_account_bound")
+    authorized_account.security_work_authorized = True
+    select_account = AsyncMock(
+        side_effect=[
+            AccountSelection(account=regular_account, error_message=None),
+            AccountSelection(account=authorized_account, error_message=None),
+        ]
+    )
+    cyber_message = (
+        "This chat was flagged for possible cybersecurity risk. "
+        "If this seems wrong, try rephrasing your request. "
+        "To get authorized for security work, join the Trusted Access for Cyber program. "
+        "https://chatgpt.com/cyber"
+    )
+    dispatched_account_ids: list[str] = []
+
+    async def fake_stream(payload, headers, access_token, account_id, base_url=None, raise_for_status=False):
+        del payload, headers, access_token, base_url, raise_for_status
+        dispatched_account_ids.append(account_id)
+        if account_id == regular_account.chatgpt_account_id:
+            yield (
+                "data: "
+                + json.dumps(
+                    {
+                        "type": "response.failed",
+                        "response": {
+                            "id": "resp_cyber_account_bound",
+                            "error": {
+                                "code": "invalid_request_error",
+                                "type": "invalid_request_error",
+                                "message": cyber_message,
+                            },
+                        },
+                    }
+                )
+                + "\n\n"
+            )
+            return
+        yield 'data: {"type":"response.completed","response":{"id":"resp_cross_account"}}\n\n'
+
+    monkeypatch.setattr(proxy_service, "get_settings_cache", lambda: _SettingsCache(settings))
+    monkeypatch.setattr(proxy_service, "get_settings", lambda: settings)
+    monkeypatch.setattr(service._load_balancer, "select_account", select_account)
+    monkeypatch.setattr(service._load_balancer, "record_error", AsyncMock())
+    monkeypatch.setattr(service._load_balancer, "record_success", AsyncMock())
+    monkeypatch.setattr(service, "_ensure_fresh", AsyncMock(side_effect=lambda account, **kwargs: account))
+    monkeypatch.setattr(proxy_service, "core_stream_responses", fake_stream)
+
+    payload = ResponsesRequest.model_validate(
+        {
+            "model": "gpt-5.1",
+            "instructions": "check api",
+            "input": [
+                {
+                    "type": "reasoning",
+                    "id": "rs_owner",
+                    "encrypted_content": "owner-bound",
+                }
+            ],
+            "stream": True,
+        }
+    )
+
+    chunks = [chunk async for chunk in service.stream_responses(payload, {"session_id": "sid-stream"})]
+
+    assert dispatched_account_ids == [regular_account.chatgpt_account_id]
+    assert all("resp_cross_account" not in chunk for chunk in chunks)
+
+
 @pytest.mark.asyncio
 async def test_stream_responses_treats_missing_security_work_pool_as_optional(monkeypatch):
     settings = _make_proxy_settings()
@@ -17737,6 +18009,63 @@ async def fake_reconnect_http_bridge_session(
     assert request_state.event_queue.empty()
 
 
+@pytest.mark.asyncio
+async def test_http_bridge_refuses_account_bound_security_work_retry(monkeypatch):
+    service = proxy_service.ProxyService(_repo_factory(_RequestLogsRecorder()))
+    regular_account = _make_account("acc_bridge_security_account_bound")
+    request_text = json.dumps(
+        {
+            "type": "response.create",
+            "model": "gpt-5.1",
+            "input": [
+                {
+                    "type": "reasoning",
+                    "id": "rs_owner",
+                    "encrypted_content": "owner-bound",
+                }
+            ],
+        },
+        separators=(",", ":"),
+    )
+    request_state = proxy_service._WebSocketRequestState(
+        request_id="bridge_req_security_account_bound",
+        model="gpt-5.1",
+        service_tier=None,
+        reasoning_effort=None,
+        api_key_reservation=None,
+        started_at=1.0,
+        awaiting_response_created=True,
+        event_queue=asyncio.Queue(),
+        transport="http",
+        request_text=request_text,
+        preferred_account_id=regular_account.id,
+    )
+    session = proxy_service._HTTPBridgeSession(
+        key=proxy_service._HTTPBridgeSessionKey("turn_state_header", "turn-security-owner", None),
+        headers={},
+        affinity=proxy_service._AffinityPolicy(),
+        request_model="gpt-5.1",
+        account=regular_account,
+        upstream=AsyncMock(),
+        upstream_control=proxy_service._WebSocketUpstreamControl(),
+        pending_requests=deque([request_state]),
+        pending_lock=anyio.Lock(),
+        response_create_gate=asyncio.Semaphore(1),
+        queued_request_count=1,
+        last_used_at=1.0,
+        idle_ttl_seconds=300.0,
+    )
+    reconnect = AsyncMock()
+    monkeypatch.setattr(service, "_reconnect_http_bridge_session", reconnect)
+
+    assert await service._retry_http_bridge_security_work_request(session, request_state) is False
+
+    reconnect.assert_not_awaited()
+    assert request_state.preferred_account_id == regular_account.id
+    assert request_state.excluded_account_ids == set()
+    assert request_state.request_text == request_text
+
+
 @pytest.mark.parametrize(
     ("item_type", "expected_deferred"),
     [
@@ -18181,6 +18510,83 @@ async def test_http_bridge_recovery_auth_reconnect_failure_preserves_original_au
     assert await request_state.event_queue.get() is None
 
 
+@pytest.mark.asyncio
+async def test_retry_http_bridge_bound_auth_refresh_never_sends_on_other_account(monkeypatch):
+    service = proxy_service.ProxyService(_repo_factory(_RequestLogsRecorder()))
+    owner = _make_account("acc_bridge_bound_auth_owner")
+    other = _make_account("acc_bridge_bound_auth_other")
+    request_text = (
+        '{"type":"response.create","model":"gpt-5.6-sol","input":['
+        '{"type":"reasoning","id":"rs_owner","encrypted_content":"owner-bound"}]}'
+    )
+    request_state = proxy_service._WebSocketRequestState(
+        request_id="bridge_bound_auth_refresh",
+        model="gpt-5.6-sol",
+        service_tier=None,
+        reasoning_effort=None,
+        api_key_reservation=None,
+        started_at=0.0,
+        awaiting_response_created=True,
+        transport="http",
+        request_text=request_text,
+        preferred_account_id=owner.id,
+        replay_required_account_id=owner.id,
+    )
+    owner_upstream = AsyncMock()
+    other_upstream = AsyncMock()
+    session = proxy_service._HTTPBridgeSession(
+        key=proxy_service._HTTPBridgeSessionKey("prompt_cache", "bridge-bound-auth", None),
+        headers={},
+        affinity=proxy_service._AffinityPolicy(),
+        request_model="gpt-5.6-sol",
+        account=owner,
+        upstream=owner_upstream,
+        upstream_control=proxy_service._WebSocketUpstreamControl(),
+        pending_requests=deque([request_state]),
+        pending_lock=anyio.Lock(),
+        response_create_gate=asyncio.Semaphore(1),
+        queued_request_count=1,
+        last_used_at=0.0,
+        idle_ttl_seconds=30.0,
+    )
+
+    async def adversarial_reconnect(
+        reconnect_session,
+        *,
+        request_state,
+        require_same_account=False,
+        require_preferred_account=False,
+        **kwargs,
+    ):
+        del request_state, kwargs
+        if not (require_same_account and require_preferred_account):
+            reconnect_session.account = other
+            reconnect_session.upstream = other_upstream
+            return
+        raise proxy_module.ProxyResponseError(
+            502,
+            openai_error(
+                "previous_response_owner_unavailable",
+                "Request payload owner account is unavailable; retry later.",
+                error_type="server_error",
+            ),
+        )
+
+    monkeypatch.setattr(service, "_reconnect_http_bridge_session", adversarial_reconnect)
+
+    result = await service._retry_http_bridge_precreated_auth_request(
+        session,
+        request_state,
+        error_message="Authentication failed",
+    )
+
+    assert result == "failed"
+    assert session.account is owner
+    owner_upstream.send_text.assert_not_awaited()
+    other_upstream.send_text.assert_not_awaited()
+    assert request_state.replay_required_account_id == owner.id
+
+
 @pytest.mark.asyncio
 async def test_http_bridge_keeps_previous_response_pinned_security_work_error(monkeypatch):
     request_logs = _RequestLogsRecorder()
@@ -20093,6 +20499,68 @@ async def select_account(deadline: float, **kwargs: object) -> AccountSelection:
     assert request_logs.calls[0]["account_id"] == account_owner.id
 
 
+@pytest.mark.asyncio
+async def test_connect_proxy_websocket_account_bound_replay_stays_on_owner(monkeypatch):
+    request_logs = _RequestLogsRecorder()
+    service = proxy_service.ProxyService(_repo_factory(request_logs))
+    account_owner = _make_account("acc_ws_replay_owner")
+    account_other = _make_account("acc_ws_replay_other")
+    select_account = AsyncMock(
+        side_effect=[
+            AccountSelection(account=account_owner, error_message=None),
+            AccountSelection(account=account_other, error_message=None),
+        ]
+    )
+    handshake_error = proxy_module.ProxyResponseError(
+        429,
+        openai_error("usage_limit_reached", "usage limit reached"),
+    )
+    monkeypatch.setattr(service, "_select_account_with_budget", select_account)
+    monkeypatch.setattr(service._load_balancer, "mark_rate_limit", AsyncMock())
+    monkeypatch.setattr(service, "_ensure_fresh", AsyncMock(return_value=account_owner))
+    open_upstream = AsyncMock(side_effect=[handshake_error])
+    monkeypatch.setattr(service, "_open_upstream_websocket", open_upstream)
+    monkeypatch.setattr(service, "_release_websocket_reservation", AsyncMock())
+
+    request_state = proxy_service._WebSocketRequestState(
+        request_id="ws_req_account_bound_replay",
+        model="gpt-5.1",
+        service_tier=None,
+        reasoning_effort=None,
+        api_key_reservation=None,
+        started_at=0.0,
+        request_text=(
+            '{"type":"response.create","model":"gpt-5.1","input":['
+            '{"type":"reasoning","id":"rs_owner","encrypted_content":"owner-bound"}]}'
+        ),
+    )
+    websocket_send = AsyncMock()
+
+    selected_account, selected_upstream = await service._connect_proxy_websocket(
+        {},
+        sticky_key=None,
+        sticky_kind=None,
+        prefer_earlier_reset=False,
+        prefer_earlier_reset_window="secondary",
+        routing_strategy="usage_weighted",
+        model="gpt-5.1",
+        request_state=request_state,
+        api_key=None,
+        client_send_lock=anyio.Lock(),
+        websocket=cast(WebSocket, SimpleNamespace(send_text=websocket_send)),
+    )
+
+    assert selected_account is None
+    assert selected_upstream is None
+    assert select_account.await_count == 2
+    assert select_account.await_args_list[1].kwargs["preferred_account_id"] == account_owner.id
+    open_upstream.assert_awaited_once()
+    websocket_send_args = websocket_send.await_args
+    assert websocket_send_args is not None
+    sent_payload = json.loads(websocket_send_args.args[0])
+    assert sent_payload["error"]["code"] == "previous_response_owner_unavailable"
+
+
 @pytest.mark.asyncio
 async def test_connect_proxy_websocket_surfaces_local_connect_overload_without_penalizing_account(monkeypatch):
     settings = _make_proxy_settings()
@@ -36273,6 +36741,51 @@ def test_cross_transport_fresh_replay_requires_matching_ws_continuity_prefix():
     assert fresh.input == full_input
 
 
+def test_cross_transport_fresh_replay_rejects_account_bound_payload():
+    service = proxy_service.ProxyService(_repo_factory(_RequestLogsRecorder()))
+    first_input: list[JsonValue] = [
+        {"role": "user", "content": [{"type": "input_text", "text": "call echo"}]},
+    ]
+    full_input: list[JsonValue] = [
+        *first_input,
+        {
+            "type": "reasoning",
+            "id": "rs_owner",
+            "encrypted_content": "owner-bound",
+        },
+        {
+            "type": "function_call",
+            "name": "echo",
+            "call_id": "call_1",
+            "arguments": '{"value":"ok"}',
+        },
+        {"type": "function_call_output", "call_id": "call_1", "output": "ok"},
+    ]
+    service._websocket_continuity_index[("turn_generated_by_ws", None)] = proxy_service._WebSocketContinuityState(
+        last_completed_response_id="resp_ws_owner",
+        last_completed_input_count=len(first_input),
+        last_completed_input_prefix_fingerprint=proxy_service._fingerprint_input_items(first_input),
+    )
+    payload = ResponsesRequest.model_validate(
+        {
+            "model": "gpt-5.6-sol",
+            "instructions": "test",
+            "previous_response_id": "resp_ws_owner",
+            "input": full_input,
+        }
+    )
+
+    assert (
+        streaming_retry_module._verified_cross_transport_fresh_replay(
+            cast(Any, service),
+            payload=payload,
+            headers={"x-codex-session-id": "sid-cross-transport"},
+            api_key=None,
+        )
+        is None
+    )
+
+
 def test_cross_transport_fresh_replay_rejects_unverified_client_full_resend():
     service = proxy_service.ProxyService(_repo_factory(_RequestLogsRecorder()))
     payload = ResponsesRequest.model_validate(
@@ -44793,6 +45306,171 @@ async def test_retry_http_bridge_precreated_request_migrates_only_safe_initial_t
     upstream.send_text.assert_awaited_once_with(request_state.request_text)
 
 
+@pytest.mark.asyncio
+async def test_retry_http_bridge_precreated_request_keeps_account_bound_body_on_owner(monkeypatch):
+    service = proxy_service.ProxyService(_repo_factory(_RequestLogsRecorder()))
+    account = _make_account("acc_bridge_account_bound")
+    request_state = proxy_service._WebSocketRequestState(
+        request_id="req_bridge_account_bound",
+        model="gpt-5.6-sol",
+        service_tier=None,
+        reasoning_effort=None,
+        api_key_reservation=None,
+        started_at=0.0,
+        awaiting_response_created=True,
+        transport="http",
+        request_text=(
+            '{"type":"response.create","model":"gpt-5.6-sol","input":['
+            '{"type":"reasoning","id":"rs_owner","encrypted_content":"owner-bound"}]}'
+        ),
+        preferred_account_id=account.id,
+    )
+    upstream = AsyncMock()
+    session = proxy_service._HTTPBridgeSession(
+        key=proxy_service._HTTPBridgeSessionKey("prompt_cache", "bridge-account-bound", None),
+        headers={"x-codex-turn-state": "turn_state_owner"},
+        affinity=proxy_service._AffinityPolicy(),
+        request_model="gpt-5.6-sol",
+        account=account,
+        upstream=upstream,
+        upstream_control=proxy_service._WebSocketUpstreamControl(),
+        pending_requests=deque([request_state]),
+        pending_lock=anyio.Lock(),
+        response_create_gate=asyncio.Semaphore(1),
+        queued_request_count=1,
+        last_used_at=0.0,
+        idle_ttl_seconds=30.0,
+        upstream_turn_state="turn_state_owner",
+        downstream_turn_state="turn_state_owner",
+    )
+    reconnect = AsyncMock(return_value=None)
+    monkeypatch.setattr(service, "_reconnect_http_bridge_session", reconnect)
+
+    assert await service._retry_http_bridge_precreated_request(session) is True
+
+    reconnect.assert_awaited_once_with(
+        session,
+        request_state=request_state,
+        require_same_account=True,
+        require_preferred_account=True,
+    )
+    assert request_state.preferred_account_id == account.id
+    assert request_state.excluded_account_ids == set()
+    assert session.upstream_turn_state == "turn_state_owner"
+    upstream.send_text.assert_awaited_once_with(request_state.request_text)
+
+
+@pytest.mark.asyncio
+async def test_retry_http_bridge_precreated_request_keeps_operation_id_on_owner(monkeypatch):
+    service = proxy_service.ProxyService(_repo_factory(_RequestLogsRecorder()))
+    account = _make_account("acc_bridge_operation_owner")
+    request_text = '{"type":"response.create","model":"gpt-5.6-sol","input":"portable user input"}'
+    request_state = proxy_service._WebSocketRequestState(
+        request_id="req_bridge_operation_owner",
+        model="gpt-5.6-sol",
+        service_tier=None,
+        reasoning_effort=None,
+        api_key_reservation=None,
+        started_at=0.0,
+        awaiting_response_created=True,
+        transport="http",
+        request_text=request_text,
+        preferred_account_id=account.id,
+        archive_request_id="archive_bridge_operation_owner",
+        operation_id="op_bridge_owner",
+    )
+    upstream = AsyncMock()
+    session = proxy_service._HTTPBridgeSession(
+        key=proxy_service._HTTPBridgeSessionKey("prompt_cache", "bridge-operation-owner", None),
+        headers={},
+        affinity=proxy_service._AffinityPolicy(),
+        request_model="gpt-5.6-sol",
+        account=account,
+        upstream=upstream,
+        upstream_control=proxy_service._WebSocketUpstreamControl(),
+        pending_requests=deque([request_state]),
+        pending_lock=anyio.Lock(),
+        response_create_gate=asyncio.Semaphore(1),
+        queued_request_count=1,
+        last_used_at=0.0,
+        idle_ttl_seconds=30.0,
+    )
+    reconnect = AsyncMock(return_value=None)
+    monkeypatch.setattr(service, "_reconnect_http_bridge_session", reconnect)
+
+    assert await service._retry_http_bridge_precreated_request(session) is True
+
+    reconnect.assert_awaited_once_with(
+        session,
+        request_state=request_state,
+        require_same_account=True,
+        require_preferred_account=True,
+    )
+    assert request_state.excluded_account_ids == set()
+    assert request_state.operation_id == "op_bridge_owner"
+    upstream.send_text.assert_awaited_once()
+    send_args = upstream.send_text.await_args
+    assert send_args is not None
+    assert json.loads(send_args.args[0]) == {
+        "type": "response.create",
+        "model": "gpt-5.6-sol",
+        "input": "portable user input",
+        "client_metadata": {"codex_lb_operation_id": "op_bridge_owner"},
+    }
+
+
+@pytest.mark.asyncio
+async def test_retry_http_bridge_precreated_request_allows_prepared_neutral_archive(monkeypatch):
+    service = proxy_service.ProxyService(_repo_factory(_RequestLogsRecorder()))
+    account = _make_account("acc_bridge_prepared_neutral")
+    payload = ResponsesRequest.model_validate(
+        {
+            "model": "gpt-5.6-sol",
+            "instructions": "",
+            "input": "portable user input",
+            "stream": True,
+        }
+    )
+    request_state, request_text = service._prepare_response_bridge_request_state(
+        payload,
+        api_key=None,
+        api_key_reservation=None,
+        include_type_field=True,
+        attach_event_queue=False,
+        transport=proxy_service._REQUEST_TRANSPORT_HTTP,
+        client_metadata=None,
+    )
+    request_state.preferred_account_id = account.id
+    upstream = AsyncMock()
+    session = proxy_service._HTTPBridgeSession(
+        key=proxy_service._HTTPBridgeSessionKey("prompt_cache", "bridge-prepared-neutral", None),
+        headers={},
+        affinity=proxy_service._AffinityPolicy(),
+        request_model="gpt-5.6-sol",
+        account=account,
+        upstream=upstream,
+        upstream_control=proxy_service._WebSocketUpstreamControl(),
+        pending_requests=deque([request_state]),
+        pending_lock=anyio.Lock(),
+        response_create_gate=asyncio.Semaphore(1),
+        queued_request_count=1,
+        last_used_at=0.0,
+        idle_ttl_seconds=30.0,
+    )
+    reconnect = AsyncMock(return_value=None)
+    monkeypatch.setattr(service, "_reconnect_http_bridge_session", reconnect)
+
+    assert request_state.archive_request_id is not None
+    assert request_state.operation_id is None
+    assert await service._retry_http_bridge_precreated_request(session) is True
+
+    reconnect.assert_awaited_once_with(
+        session,
+        request_state=request_state,
+    )
+    upstream.send_text.assert_awaited_once_with(request_text)
+
+
 @pytest.mark.asyncio
 async def test_retry_http_bridge_precreated_request_keeps_hard_session_owner_bound(monkeypatch):
     service = proxy_service.ProxyService(_repo_factory(_RequestLogsRecorder()))