diff --git a/app/modules/proxy/_service/http_bridge/streaming.py b/app/modules/proxy/_service/http_bridge/streaming.py index 1fce2706ea..f9c1be3842 100644 --- a/app/modules/proxy/_service/http_bridge/streaming.py +++ b/app/modules/proxy/_service/http_bridge/streaming.py @@ -68,7 +68,6 @@ _effective_http_bridge_idle_ttl_seconds, _http_bridge_durable_lookup_allows_turn_state_takeover, _http_bridge_is_context_overflow_error, - _http_bridge_is_previous_response_owner_unavailable, _http_bridge_models_compatible, _http_bridge_owner_lookup_unavailable_error_envelope, _http_bridge_payload_looks_like_full_resend, @@ -233,6 +232,11 @@ def _proxy_error_code_message(exc: ProxyResponseError) -> tuple[str | None, str return (str(code) if code is not None else None, str(message) if message is not None else None) +def _http_bridge_owner_failure_allows_account_neutral_replay(exc: ProxyResponseError) -> bool: + code, _message = _proxy_error_code_message(exc) + return code in {"previous_response_owner_unavailable", "continuity_owner_conflict"} + + def _http_bridge_account_capacity_wait_seconds(exc: ProxyResponseError) -> float | None: code, message = _proxy_error_code_message(exc) if code == "capacity_exhausted_active_sessions": @@ -1190,7 +1194,7 @@ def owner_unavailable_allows_account_neutral_replay(exc: ProxyResponseError) -> nonlocal durable_full_resend_retains_prior_output if ( - not _http_bridge_is_previous_response_owner_unavailable(exc) + not _http_bridge_owner_failure_allows_account_neutral_replay(exc) or forwarded_request or rewritten_file_account_id is not None or durable_full_resend_anchor_count is None @@ -1293,6 +1297,51 @@ def switch_to_account_neutral_replay() -> None: durable_lookup = None file_required_preferred_account = False + def switch_model_transition_to_account_neutral_fork(exc: ProxyResponseError) -> bool: + nonlocal account_neutral_recovery + nonlocal affinity + nonlocal bridge_session_key + nonlocal force_local_recovery_creation + nonlocal incoming_turn_state_header + nonlocal preferred_account_has_continuity_provenance + nonlocal request_state + nonlocal session_creation_headers + nonlocal session_header_fallback_key + + if ( + durable_model_transition_lookup is None + or not _http_bridge_owner_failure_allows_account_neutral_replay(exc) + or owner_unavailable_allows_account_neutral_replay(exc) + or request_state.previous_response_id is not None + or rewritten_file_account_id is not None + ): + return False + failed_owner_id = request_state.preferred_account_id + _log_http_bridge_event( + "model_transition_owner_conflict_fork", + bridge_session_key, + account_id=failed_owner_id, + model=effective_payload.model, + detail="outcome=retry_without_previous_model_owner", + cache_key_family=bridge_session_key.affinity_kind, + model_class=_extract_model_class(effective_payload.model) if effective_payload.model else None, + owner_check_applied=True, + ) + if failed_owner_id is not None: + fresh_replay_excluded_account_ids.add(failed_owner_id) + session_creation_headers = without_http_bridge_session_affinity_headers(session_creation_headers) + incoming_turn_state_header = None + session_header_fallback_key = None + affinity = _AffinityPolicy() + replay_kind, replay_key = make_http_bridge_account_neutral_replay_key(uuid4().hex) + bridge_session_key = _HTTPBridgeSessionKey(replay_kind, replay_key, bridge_session_key.api_key_id) + account_neutral_recovery = True + force_local_recovery_creation = True + request_state.preferred_account_id = None + request_state.excluded_account_ids.update(fresh_replay_excluded_account_ids) + preferred_account_has_continuity_provenance = False + return True + if required_continuity_owner_missing: owner_unavailable = ProxyResponseError( 502, @@ -1346,6 +1395,8 @@ def switch_to_account_neutral_replay() -> None: exclude_account_ids=fresh_replay_excluded_account_ids or None, ) except ProxyResponseError as exc: + if switch_model_transition_to_account_neutral_fork(exc): + continue if not owner_unavailable_allows_account_neutral_replay(exc): exc_code, _exc_message = _proxy_error_code_message(exc) if not unanchored_fork_spill_attempted and _http_bridge_unanchored_fork_can_spill_on_cap( diff --git a/tests/integration/test_http_responses_bridge.py b/tests/integration/test_http_responses_bridge.py index 989783c5af..a1c81c97cd 100644 --- a/tests/integration/test_http_responses_bridge.py +++ b/tests/integration/test_http_responses_bridge.py @@ -7595,6 +7595,154 @@ async def fake_connect_responses_websocket( assert owner_miss["preferred_account_is_continuity_owner"] is True +@pytest.mark.asyncio +async def test_v1_responses_http_bridge_replays_full_resend_after_owner_conflict(async_client, monkeypatch): + _install_bridge_settings(monkeypatch, enabled=True) + owner_account_id = await _import_account( + async_client, + "acc_http_bridge_conflict_owner", + "http-bridge-conflict-owner@example.com", + ) + alternate_account_id = await _import_account( + async_client, + "acc_http_bridge_conflict_alternate", + "http-bridge-conflict-alternate@example.com", + ) + owner_account = await _get_account(owner_account_id) + alternate_account = await _get_account(alternate_account_id) + owner_chatgpt_account_id = cast(str, owner_account.chatgpt_account_id) + alternate_chatgpt_account_id = cast(str, alternate_account.chatgpt_account_id) + owner_upstream = _ClosingBridgeUpstreamWebSocket("resp_conflict_owner") + alternate_upstream = _FakeBridgeUpstreamWebSocket("resp_conflict_replay") + selection_calls: list[dict[str, object]] = [] + connected_account_ids: list[str] = [] + connect_headers_by_account: dict[str, dict[str, str]] = {} + + async def fake_select_account_with_budget(self, deadline, **kwargs): + del self, deadline + selection_calls.append(dict(kwargs)) + preferred_account_id = cast(str | None, kwargs.get("preferred_account_id")) + excluded_account_ids = cast(set[str], kwargs.get("exclude_account_ids") or set()) + fallback_enabled = bool(kwargs.get("fallback_on_preferred_account_unavailable", True)) + if preferred_account_id == owner_account.id and not fallback_enabled: + assert kwargs.get("preferred_account_is_continuity_owner") is True + return AccountSelection( + account=None, + error_message="Account-owned continuity sources conflict; retry the logical turn", + error_code="continuity_owner_conflict", + ) + if owner_account.id in excluded_account_ids: + return AccountSelection(account=alternate_account, error_message=None, error_code=None) + return AccountSelection(account=owner_account, error_message=None, error_code=None) + + async def fake_ensure_fresh_with_budget(self, target, *, force=False, timeout_seconds): + del self, force, timeout_seconds + return target + + async def fake_connect_responses_websocket( + headers, + access_token, + account_id_header, + *, + base_url=None, + session=None, + ): + del access_token, base_url, session + connected_account_ids.append(account_id_header) + connect_headers_by_account[account_id_header] = dict(headers) + if account_id_header == owner_chatgpt_account_id: + return owner_upstream + assert account_id_header == alternate_chatgpt_account_id + return alternate_upstream + + monkeypatch.setattr(proxy_module.ProxyService, "_select_account_with_budget", fake_select_account_with_budget) + monkeypatch.setattr(proxy_module.ProxyService, "_ensure_fresh_with_budget", fake_ensure_fresh_with_budget) + monkeypatch.setattr(proxy_module, "connect_responses_websocket", fake_connect_responses_websocket) + + historical_input = [ + { + "role": "user", + "content": [{"type": "input_text", "text": "first question"}], + } + ] + first = await asyncio.wait_for( + async_client.post( + "/v1/responses", + json={ + "model": "gpt-5.1", + "instructions": "Return exactly OK.", + "input": historical_input, + "prompt_cache_key": "http-bridge-conflict-replay", + }, + ), + timeout=_TEST_SYNC_TIMEOUT_SECONDS, + ) + assert first.status_code == 200, first.text + + retained_prior_output = { + "type": "message", + "role": "assistant", + "status": "completed", + "content": [{"type": "output_text", "text": "first answer"}], + } + full_resend = [ + *historical_input, + retained_prior_output, + { + "role": "user", + "content": [{"type": "input_text", "text": "second question"}], + }, + ] + second = await asyncio.wait_for( + async_client.post( + "/v1/responses", + json={ + "model": "gpt-5.1", + "instructions": "Return exactly OK.", + "input": full_resend, + "prompt_cache_key": "http-bridge-conflict-replay", + "previous_response_id": first.json()["id"], + }, + headers={ + "session_id": "stale-session", + "x-codex-session-id": "stale-codex-session", + "x-codex-turn-state": "http_turn_stale", + "x-request-trace": "keep-me", + }, + ), + timeout=_TEST_SYNC_TIMEOUT_SECONDS, + ) + + assert second.status_code == 200, second.text + assert second.json()["id"] == "resp_conflict_replay_1" + assert connected_account_ids == [ + owner_chatgpt_account_id, + alternate_chatgpt_account_id, + ] + replay_connect_headers = { + key.lower(): value for key, value in connect_headers_by_account[alternate_chatgpt_account_id].items() + } + assert replay_connect_headers["x-request-trace"] == "keep-me" + assert ( + not { + "session_id", + "x-codex-session-id", + "x-codex-turn-state", + } + & replay_connect_headers.keys() + ) + replay_payload = json.loads(alternate_upstream.sent_text[0]) + assert "previous_response_id" not in replay_payload + assert replay_payload["input"] == full_resend + owner_conflict = next( + call + for call in selection_calls + if call.get("preferred_account_id") == owner_account.id + and call.get("fallback_on_preferred_account_unavailable") is False + ) + assert owner_conflict["preferred_account_is_continuity_owner"] is True + + @pytest.mark.asyncio async def test_backend_responses_http_bridge_real_selector_recovers_full_resend_without_degrading_pool( async_client, monkeypatch diff --git a/tests/unit/test_proxy_http_bridge.py b/tests/unit/test_proxy_http_bridge.py index 00dd232e8e..7ecd6e700e 100644 --- a/tests/unit/test_proxy_http_bridge.py +++ b/tests/unit/test_proxy_http_bridge.py @@ -17997,6 +17997,124 @@ async def fail_first_session_before_output( assert all(call["preferred_account_has_continuity_provenance"] is True for call in creation_calls) +@pytest.mark.asyncio +async def test_stream_via_http_bridge_forks_model_transition_after_owner_conflict( + monkeypatch: pytest.MonkeyPatch, +) -> None: + service = proxy_service.ProxyService(cast(Any, nullcontext())) + payload = proxy_service.ResponsesRequest.model_validate( + { + "model": "gpt-5.6-terra", + "instructions": "hi", + "input": [{"role": "user", "content": "continue on the new model"}], + } + ) + durable_lookup = proxy_service.DurableBridgeLookup( + session_id="durable-model-conflict-parent", + canonical_kind="session_header", + canonical_key="shared-root", + api_key_scope="__anonymous__", + account_id="acc-model-owner", + owner_instance_id=None, + owner_epoch=1, + lease_expires_at=datetime.now(timezone.utc) + timedelta(seconds=60), + state=HttpBridgeSessionState.ACTIVE, + latest_turn_state="http_turn_model_parent", + latest_response_id="resp_model_parent", + model="gpt-5.6-sol", + ) + owner_conflict = ProxyResponseError( + 502, + openai_error( + "continuity_owner_conflict", + "Durable continuity aliases resolve to conflicting upstream owners.", + ), + ) + creation_keys: list[proxy_service._HTTPBridgeSessionKey] = [] + creation_calls: list[dict[str, Any]] = [] + + async def fake_get_or_create( + key: proxy_service._HTTPBridgeSessionKey, + **kwargs: Any, + ) -> proxy_service._HTTPBridgeSession: + creation_keys.append(key) + creation_calls.append(kwargs) + if len(creation_calls) == 1: + raise owner_conflict + session = _make_bridge_session(key=key) + session.account = cast( + Any, + SimpleNamespace(id="acc-model-alternate", status=AccountStatus.ACTIVE), + ) + session.request_model = payload.model + return session + + async def fake_stream_events( + _session: proxy_service._HTTPBridgeSession, + **_kwargs: Any, + ): + yield 'data: {"type":"response.completed"}\n\n' + + monkeypatch.setattr( + proxy_service, + "get_settings_cache", + lambda: cast( + Any, + SimpleNamespace( + get=AsyncMock( + return_value=SimpleNamespace( + sticky_threads_enabled=False, + openai_cache_affinity_max_age_seconds=1800, + http_responses_session_bridge_prompt_cache_idle_ttl_seconds=3600, + http_responses_session_bridge_gateway_safe_mode=False, + ) + ) + ), + ), + ) + monkeypatch.setattr(proxy_service, "get_settings", lambda: _make_app_settings()) + monkeypatch.setattr(service._durable_bridge, "lookup_request_targets", AsyncMock(return_value=durable_lookup)) + monkeypatch.setattr(service, "_resolve_file_account_for_responses", AsyncMock(return_value=None)) + monkeypatch.setattr(service, "_get_or_create_http_bridge_session", fake_get_or_create) + monkeypatch.setattr(service, "_stream_http_bridge_session_events", fake_stream_events) + + chunks = [ + chunk + async for chunk in service._stream_via_http_bridge( + payload, + headers={ + "x-codex-turn-state": "http_turn_model_parent", + "x-codex-session-id": "shared-root", + }, + codex_session_affinity=True, + propagate_http_errors=True, + openai_cache_affinity=True, + api_key=None, + api_key_reservation=None, + suppress_text_done_events=False, + idle_ttl_seconds=120.0, + codex_idle_ttl_seconds=1800.0, + max_sessions=8, + queue_limit=4, + downstream_turn_state="http_turn_model_child", + ) + ] + + assert chunks == ['data: {"type":"response.completed"}\n\n'] + assert len(creation_calls) == 2 + assert creation_keys[0].affinity_kind in {"session_header", "turn_state_header"} + assert is_http_bridge_account_neutral_replay( + kind=creation_keys[1].affinity_kind, + key=creation_keys[1].affinity_key, + ) + assert creation_calls[0]["preferred_account_id"] == "acc-model-owner" + assert creation_calls[0]["preferred_account_has_continuity_provenance"] is True + assert creation_calls[1]["preferred_account_id"] is None + assert creation_calls[1]["preferred_account_has_continuity_provenance"] is False + assert creation_calls[1]["exclude_account_ids"] == {"acc-model-owner"} + assert creation_calls[1]["allow_forward_to_owner"] is False + + @pytest.mark.asyncio async def test_stream_via_http_bridge_preserves_verified_replay_kind_for_durable_model_transition( monkeypatch: pytest.MonkeyPatch,