fix(proxy): allow verified replay past dead owner route

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

@@ -1965,10 +1965,25 @@ async def _retry_account_model_rejection(
                                     await _release_tracked_stream_lease(current_account_lease)
                                     current_account_lease = None
                                     await _record_or_defer_confirmed_route_backoff(account)
-                                    can_try_other_account = (
-                                        not require_preferred_account
-                                        and account.id != file_preferred_account_id
-                                        and attempt < max_attempts - 1
+                                    verified_owner_replay_moved = False
+                                    if (
+                                        attempt < max_attempts - 1
+                                        and routing_strategy != "single_account"
+                                        and file_preferred_account_id is None
+                                        and turn_state_owner_account_id is None
+                                        and not settlement.downstream_visible
+                                    ):
+                                        verified_owner_replay_moved = _move_verified_fresh_replay_from_owner(
+                                            account_id=account.id,
+                                            outcome="owner_initial_proxy_connect_failure",
+                                        )
+                                    can_try_other_account = bool(
+                                        attempt < max_attempts - 1
+                                        and routing_strategy != "single_account"
+                                        and file_preferred_account_id is None
+                                        and turn_state_owner_account_id is None
+                                        and not settlement.downstream_visible
+                                        and not require_preferred_account
                                     )
                                     if not can_try_other_account:
                                         # Hard account ownership or exhausted
@@ -1980,7 +1995,8 @@ async def _retry_account_model_rejection(
                                     last_pre_dispatch_transport_error = tex
                                     transient_failed_account_id = account.id
                                     excluded_account_ids.add(account.id)
-                                    affinity = replace(affinity, reallocate_sticky=True)
+                                    if not verified_owner_replay_moved:
+                                        affinity = replace(affinity, reallocate_sticky=True)
                                     _facade().logger.info(
                                         "Retrying stream after confirmed pre-dispatch proxy connect failure "
                                         "request_id=%s account_id=%s attempt=%d",

openspec/changes/retry-account-proxy-connect-failures/design.md

@@ -44,6 +44,13 @@ account with the original sanitized failure. Soft prompt-cache and
 process-session affinity may move; the failed account's sticky binding is
 reallocated so the replacement selection does not immediately loop back.
 
+A previous-response continuation is movable when local continuity evidence
+verifies that its input is a complete fresh replay which can be resent without
+the owner anchor. A confirmed pre-dispatch failure may remove that anchor before
+replacement selection. This exception does not override turn-state or uploaded-
+file ownership, forced single-account routing, another required-account contract,
+or downstream-visible output.
+
 ## Account backoff and resource ordering
 
 A confirmed dead account route is stronger evidence than a generic transient

openspec/changes/retry-account-proxy-connect-failures/specs/upstream-proxy-routing/spec.md

@@ -8,7 +8,9 @@ When an account-routed transport reports that it could not connect to the select
 
 This behavior MUST cover raw HTTP/SSE, native Responses WebSocket, and the HTTP responses bridge. Before recording transient account backoff, the service MUST release response-create and stream leases held for the failed account. A request-scoped API-key reservation MUST remain singular across an internal pre-dispatch failover, MUST settle or release at the terminal request outcome before the account-health write, and MUST NOT be reacquired solely for the internal failover. If neither settlement nor fallback release can be confirmed, the service MUST leave the health write unapplied. HTTP-bridge startup cleanup MUST release only an unowned current request lifecycle, and each reservation lifecycle MUST drain only its own health writes after confirmed settlement or release. The confirmed failure MUST place the account at the existing bounded transient error-backoff floor, but MUST NOT pause, deactivate, rate-limit, or quota-penalize it.
 
-The service MUST NOT replay a request when dispatch is unknown or when the request depends on hard previous-response, turn-state, uploaded-file, single-account, or other required account ownership. If no eligible replacement account exists, the service MUST preserve the original sanitized upstream-unavailable failure instead of replacing it with a generated `no_accounts` error.
+When local continuity evidence verifies that a previous-response continuation contains a complete fresh replay which can be resent without its owner anchor, a confirmed pre-dispatch failure on that owner MUST remove the anchor and retry another eligible account. This exception MUST NOT apply when the request also depends on turn-state, uploaded-file, single-account, or other required account ownership, or after any output becomes downstream-visible.
+
+The service MUST NOT replay a request when dispatch is unknown or when the request depends on a previous-response owner without a verified complete fresh replay, turn-state, uploaded-file, single-account, or other required account ownership. If no eligible replacement account exists, the service MUST preserve the original sanitized upstream-unavailable failure instead of replacing it with a generated `no_accounts` error.
 
 #### Scenario: POST uses a healthy endpoint from the same proxy pool
 
@@ -36,6 +38,15 @@ The service MUST NOT replay a request when dispatch is unknown or when the reque
 - **THEN** the service does not send the request to the other account
 - **AND** it returns the sanitized upstream-unavailable failure for the required account
 
+#### Scenario: verified full replay moves off a dead previous-response owner
+
+- **GIVEN** a previous-response continuation whose complete fresh input has been verified locally
+- **AND** the request has no turn-state, uploaded-file, single-account, or other required ownership
+- **WHEN** the previous-response owner's proxy refuses the connection before dispatch
+- **THEN** the service removes the previous-response owner anchor
+- **AND** it excludes the failed owner and completes through another eligible account
+- **AND** no failure event from the failed owner is forwarded downstream
+
 #### Scenario: ambiguous transport failure is not replayed
 
 - **WHEN** a POST transport failure cannot prove that request dispatch was impossible

openspec/changes/retry-account-proxy-connect-failures/tasks.md

@@ -5,6 +5,7 @@
 - [x] Add movable-account failover for raw HTTP/SSE and HTTP bridge startup.
 - [x] Preserve native WebSocket failover while applying the same confirmed-route backoff.
 - [x] Preserve hard continuity/file pins and the original failure when no replacement exists.
+- [x] Move a locally verified complete fresh replay off a dead previous-response owner without overriding other hard ownership.
 - [x] Defer dead-route account-health writes until request-scoped API-key settlement.
 - [x] Track HTTP-bridge settlement ownership per reservation generation and fail closed on unconfirmed release.
 - [x] Add core-client, load-balancer, HTTP/SSE, native WebSocket, and HTTP-bridge regressions.

tests/unit/test_proxy_utils.py

@@ -31618,6 +31618,88 @@ async def fake_stream(payload, headers, access_token, account_id, base_url=None,
     assert streamed_payloads[1].input == full_input
 
 
+@pytest.mark.asyncio
+async def test_stream_verified_fresh_replay_moves_off_dead_owner_after_initial_connect_failure(monkeypatch):
+    settings = _make_proxy_settings()
+    request_logs = _RequestLogsRecorder()
+    service = proxy_service.ProxyService(_repo_factory(request_logs))
+    owner_account = _make_account("acc_stream_replay_dead_owner")
+    replacement_account = _make_account("acc_stream_replay_dead_owner_replacement")
+    session_id = "sid_stream_verified_dead_owner_replay"
+    previous_response_id = "resp_stream_verified_dead_owner"
+    request_logs.response_owner_by_id[(previous_response_id, None, session_id)] = owner_account.id
+    initial_input: list[JsonValue] = [{"role": "user", "content": "first turn"}]
+    full_input: list[JsonValue] = [
+        *initial_input,
+        {"role": "user", "content": "fresh full resend after dead route"},
+    ]
+    service._websocket_continuity_index[(session_id, None)] = proxy_service._WebSocketContinuityState(
+        last_completed_response_id=previous_response_id,
+        last_completed_input_count=len(initial_input),
+        last_completed_input_prefix_fingerprint=proxy_service._fingerprint_input_items(initial_input),
+    )
+    selection_calls: list[dict[str, object]] = []
+    streamed_payloads: list[ResponsesRequest] = []
+    streamed_account_ids: list[str | None] = []
+    record_error_backoff = AsyncMock()
+    handle_stream_error = AsyncMock()
+
+    async def fake_select_account(**kwargs):
+        selection_calls.append(dict(kwargs))
+        if kwargs.get("required_account_id") == owner_account.id:
+            return AccountSelection(account=owner_account, error_message=None)
+        assert kwargs.get("required_account_id") is None
+        assert kwargs.get("exclude_account_ids") == {owner_account.id}
+        assert kwargs.get("reallocate_sticky") is True
+        return AccountSelection(account=replacement_account, error_message=None)
+
+    async def fake_stream(payload, headers, access_token, account_id, base_url=None, raise_for_status=False, **kwargs):
+        del headers, access_token, base_url, raise_for_status, kwargs
+        streamed_payloads.append(payload)
+        streamed_account_ids.append(account_id)
+        if account_id == owner_account.chatgpt_account_id:
+            raise _pre_dispatch_proxy_connect_error("verified owner proxy route unavailable")
+        assert account_id == replacement_account.chatgpt_account_id
+        yield (
+            'data: {"type":"response.completed","response":{"id":"resp_dead_owner_replay_ok",'
+            '"status":"completed","usage":{"input_tokens":1,"output_tokens":1,'
+            '"total_tokens":2}}}\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", AsyncMock(side_effect=fake_select_account))
+    monkeypatch.setattr(service._load_balancer, "record_error_backoff", record_error_backoff)
+    monkeypatch.setattr(service._load_balancer, "record_success", AsyncMock())
+    monkeypatch.setattr(service, "_handle_stream_error", handle_stream_error)
+    monkeypatch.setattr(service, "_ensure_fresh", AsyncMock(side_effect=lambda account, **kwargs: account))
+    monkeypatch.setattr(service, "_settle_stream_api_key_usage", AsyncMock(return_value=True))
+    monkeypatch.setattr(proxy_service, "core_stream_responses", fake_stream)
+
+    payload = ResponsesRequest.model_validate(
+        {
+            "model": "gpt-5.6-sol",
+            "instructions": "test verified full replay after dead route",
+            "input": full_input,
+            "previous_response_id": previous_response_id,
+            "stream": True,
+        }
+    )
+
+    chunks = [chunk async for chunk in service.stream_responses(payload, {"session_id": session_id})]
+
+    assert json.loads(chunks[-1].split("data: ", 1)[1])["response"]["id"] == "resp_dead_owner_replay_ok"
+    assert [streamed.previous_response_id for streamed in streamed_payloads] == [previous_response_id, None]
+    assert streamed_payloads[1].input == full_input
+    assert streamed_account_ids == [
+        owner_account.chatgpt_account_id,
+        replacement_account.chatgpt_account_id,
+    ]
+    assert len(selection_calls) == 2
+    record_error_backoff.assert_awaited_once_with(owner_account)
+    handle_stream_error.assert_not_awaited()
+
+
 @pytest.mark.asyncio
 async def test_stream_verified_fresh_replay_moves_off_owner_after_refresh_connect_failure(monkeypatch):
     settings = _make_proxy_settings()