diff --git a/.all-contributorsrc b/.all-contributorsrc index 02b873a9de..2624725fd9 100644 --- a/.all-contributorsrc +++ b/.all-contributorsrc @@ -1232,6 +1232,16 @@ "code", "test" ] + }, + { + "login": "kevinsslin", + "name": "Kevin Lin", + "avatar_url": "https://avatars.githubusercontent.com/u/86810837?v=4", + "profile": "https://github.com/kevinsslin", + "contributions": [ + "code", + "test" + ] } ], "contributorsPerLine": 7, diff --git a/README.md b/README.md index e7ba0594a1..1d63923e57 100644 --- a/README.md +++ b/README.md @@ -286,6 +286,7 @@ Thanks goes to these wonderful people ([emoji key](https://allcontributors.org/e DuyBui
DuyBui
💻 ⚠️ + Kevin Lin
Kevin Lin
💻 ⚠️ diff --git a/app/modules/proxy/_service/http_bridge/streaming.py b/app/modules/proxy/_service/http_bridge/streaming.py index c692de0809..1285996fad 100644 --- a/app/modules/proxy/_service/http_bridge/streaming.py +++ b/app/modules/proxy/_service/http_bridge/streaming.py @@ -3642,10 +3642,149 @@ async def retry_precreated_for_idle_recovery( ), ) + def operation_fenced_cooldown_wait_enabled() -> bool: + """Allow a hard turn to wait until its durable fence can arbitrate recovery.""" + return ( + getattr( + _service_get_settings(), + "http_responses_session_bridge_ambiguous_continuation_recovery_mode", + "fail_closed", + ) + in {"server_anchored_replay_once", "server_indefinite_recovery"} + and getattr(_service_get_settings(), "http_responses_session_bridge_operation_ledger_enabled", True) + and request_state.hard_continuity_anchor + and session.durable_session_id is not None + and session.durable_owner_epoch is not None + and request_state.previous_response_id is None + and request_state.response_id is None + and request_state.response_event_count == 0 + ) + def continuity_bound_without_safe_replay() -> bool: """Do not hold a client stream through a cooldown we cannot use.""" return _http_bridge_continuity_bound_without_safe_replay(request_state) and not ( - _http_bridge_server_anchored_replay_enabled(request_state) + _http_bridge_server_anchored_replay_enabled(request_state) or operation_fenced_cooldown_wait_enabled() + ) + + async def wait_through_operation_fenced_startup_cooldown() -> bool: + if session.key.strength != "hard" or not operation_fenced_cooldown_wait_enabled(): + return False + retry_cooldown_seconds = await self._http_bridge_precreated_retry_cooldown_seconds(session) + if retry_cooldown_seconds <= 0: + return False + remaining_budget_seconds = request_deadline - _service_time().monotonic() + if remaining_budget_seconds <= 0: + return False + wait_seconds = min(retry_cooldown_seconds, remaining_budget_seconds) + async with session.pending_lock: + if session.queued_request_count >= queue_limit: + raise ProxyResponseError( + 429, + openai_error( + "bridge_queue_full", + "HTTP responses session bridge queue is full", + error_type="rate_limit_error", + ), + ) + session.queued_request_count += 1 + _log_http_bridge_event( + "wait_operation_fenced_cooldown", + session.key, + account_id=session.account.id, + model=session.request_model, + detail="hard_turn_operation_fence", + cache_key_family=session.key.affinity_kind, + ) + logger.info( + "HTTP bridge waiting through retry-circuit cooldown before durable hard-turn arbitration " + "request_id=%s wait_seconds=%.1f remaining_budget_seconds=%.1f", + request_state.request_id, + wait_seconds, + remaining_budget_seconds, + ) + # No upstream request has been dispatched on this path. After the + # cooldown, normal submission still has to create or claim the + # durable operation fence before response.create can be sent. + try: + current_instance = _service_get_settings().http_responses_session_bridge_instance_id + lease_refresh_interval_seconds = max( + 1.0, + min( + _http_bridge_durable_lease_ttl_seconds() / 3.0, + wait_seconds, + ), + ) + remaining_wait_seconds = wait_seconds + while remaining_wait_seconds > 0: + sleep_seconds = min(remaining_wait_seconds, lease_refresh_interval_seconds) + await asyncio.sleep(sleep_seconds) + remaining_wait_seconds = max(0.0, remaining_wait_seconds - sleep_seconds) + if remaining_wait_seconds <= 0: + break + try: + owner_lookup = await self._durable_bridge.renew_live_session( + session_id=session.durable_session_id, + api_key_id=session.key.api_key_id, + instance_id=current_instance, + owner_epoch=session.durable_owner_epoch, + lease_ttl_seconds=_http_bridge_durable_lease_ttl_seconds(), + latest_turn_state=session.downstream_turn_state, + latest_response_id=None, + ) + except Exception as exc: + session.closed = True + session.upstream_control.reconnect_requested = True + session.upstream_control.retire_after_drain = True + raise ProxyResponseError( + 502, + openai_error( + "bridge_continuity_persistence_failed", + "HTTP responses session ownership could not be renewed; retry the request.", + ), + ) from exc + if ( + owner_lookup is None + or owner_lookup.owner_instance_id != current_instance + or owner_lookup.owner_epoch != session.durable_owner_epoch + ): + session.closed = True + session.upstream_control.reconnect_requested = True + session.upstream_control.retire_after_drain = True + raise ProxyResponseError( + 502, + openai_error( + "bridge_continuity_persistence_failed", + "HTTP responses session ownership changed during cooldown; retry the request.", + ), + ) + finally: + async with session.pending_lock: + session.queued_request_count = max(0, session.queued_request_count - 1) + return True + + async def operation_fenced_request_budget_terminal_event() -> str | None: + if not operation_fenced_cooldown_wait_enabled() or _service_time().monotonic() < request_deadline: + return None + await self._release_websocket_request_state_reservation(request_state) + request_state.api_key_reservation = None + if propagate_http_errors: + raise ProxyResponseError( + 503, + openai_error( + "upstream_request_timeout", + "HTTP responses session bridge recovery exceeded the request budget.", + error_type="server_error", + ), + ) + return format_sse_event( + cast( + Mapping[str, JsonValue], + response_failed_event( + "stream_idle_timeout", + "HTTP responses session bridge recovery exceeded the request budget", + response_id=_websocket_downstream_response_id(request_state), + ), + ) ) async def startup_continuity_cooldown_terminal_event() -> str | None: @@ -3716,6 +3855,12 @@ async def startup_continuity_cooldown_terminal_event() -> str | None: ) while True: + budget_terminal_event = await operation_fenced_request_budget_terminal_event() + if budget_terminal_event is not None: + yield budget_terminal_event + return + if await wait_through_operation_fenced_startup_cooldown(): + continue startup_terminal_event = await startup_continuity_cooldown_terminal_event() if startup_terminal_event is not None: yield startup_terminal_event diff --git a/openspec/changes/hold-operation-fenced-hard-turn-cooldown/.openspec.yaml b/openspec/changes/hold-operation-fenced-hard-turn-cooldown/.openspec.yaml new file mode 100644 index 0000000000..4af864176c --- /dev/null +++ b/openspec/changes/hold-operation-fenced-hard-turn-cooldown/.openspec.yaml @@ -0,0 +1,2 @@ +schema: spec-driven +created: 2026-08-14 diff --git a/openspec/changes/hold-operation-fenced-hard-turn-cooldown/design.md b/openspec/changes/hold-operation-fenced-hard-turn-cooldown/design.md new file mode 100644 index 0000000000..11b8db08ef --- /dev/null +++ b/openspec/changes/hold-operation-fenced-hard-turn-cooldown/design.md @@ -0,0 +1,41 @@ +## Context + +Hard turn-state requests can omit `previous_response_id` while still carrying a +real Codex turn-state continuity anchor. Their replay identity is protected by +the durable operation ledger, but the startup cooldown guard runs before +operation registration. It therefore classifies the request as continuity-bound +without safe replay and returns 503 before the ledger can serialize recovery. + +The HTTP response already includes `Retry-After`, and an already-started SSE +failure includes an SSE `retry:` directive. Production telemetry shows Codex +Desktop retrying in milliseconds anyway, so another client hint does not address +the observed failure mode. + +## Decision + +Treat a turn-state-only hard request as eligible to wait through cooldown only +when all of the following hold: + +- recovery mode is `server_anchored_replay_once` or + `server_indefinite_recovery`; +- the durable operation ledger is enabled; +- the request has a real hard continuity anchor; +- the bridge has both a durable session id and current owner epoch; +- no response id or upstream response event has been observed; and +- request budget remains. + +The wait is clamped to the smaller of cooldown remaining and request budget. +It does not reserve a replay, mutate the operation journal, or send upstream. +When the cooldown expires, normal submission performs the existing operation +fingerprint lookup and atomic recovery claim. One-shot mode keeps its existing +maximum of one recovery dispatch; indefinite mode retains its existing explicit +opt-in semantics. + +## Explicit exclusions + +- No change to the default `fail_closed` mode. +- No transparent replay without a durable session and owner fence. +- No cross-account, file-pinned, image, soft-affinity, or eventful recovery. +- No weakening of operation fingerprint, ownership, or replay-count checks. +- No infinite retry added by this change; bounded one-shot mode is the + recommended deployment setting for this incident class. diff --git a/openspec/changes/hold-operation-fenced-hard-turn-cooldown/proposal.md b/openspec/changes/hold-operation-fenced-hard-turn-cooldown/proposal.md new file mode 100644 index 0000000000..5c6d83bd73 --- /dev/null +++ b/openspec/changes/hold-operation-fenced-hard-turn-cooldown/proposal.md @@ -0,0 +1,27 @@ +## Why + +When two eventless upstream attempts open the HTTP bridge retry circuit, Codex +Desktop immediately retries the same hard turn-state request. The bridge +currently returns a startup 503 before consulting the durable operation ledger. +Codex does not honor the full retry-circuit delay and can exhaust its client +retry budget during the cooldown, pausing the task even though the bridge and +VPS remain healthy. + +## What Changes + +- In an explicitly enabled server recovery mode, hold a turn-state-only hard + continuation through the active retry-circuit cooldown before submission. +- Require a live durable session id and owner epoch, zero response events, and + no response id before waiting. +- Dispatch nothing while waiting. After cooldown, use the existing durable + operation ledger and one-shot/indefinite recovery policy to arbitrate whether + the request may be created, claimed, replayed, or failed closed. +- Preserve the current immediate 503 for the default `fail_closed` mode, + in-memory fallback sessions, soft affinity, and eventful requests. +- Emit a low-cardinality bridge event when the operation-fenced wait begins. + +## Impact + +- HTTP Responses bridge startup behavior during retry-circuit cooldown. +- No database schema, public API, account routing, or default configuration + change. diff --git a/openspec/changes/hold-operation-fenced-hard-turn-cooldown/specs/responses-api-compat/spec.md b/openspec/changes/hold-operation-fenced-hard-turn-cooldown/specs/responses-api-compat/spec.md new file mode 100644 index 0000000000..a487f23e1f --- /dev/null +++ b/openspec/changes/hold-operation-fenced-hard-turn-cooldown/specs/responses-api-compat/spec.md @@ -0,0 +1,67 @@ +## ADDED Requirements + +### Requirement: Operation-fenced hard turns preserve client retry budget during cooldown + +A hard turn-state HTTP bridge request arriving during retry-circuit cooldown MUST remain pending until cooldown expires only if an explicit server recovery mode is enabled, the request has not observed a response id or response event, and the bridge has a live durable session and owner epoch. The proxy MUST NOT dispatch upstream while waiting. After the wait, the request MUST pass through the existing durable operation-ledger admission before any `response.create` is sent. + +#### Scenario: One-shot hard turn waits before durable arbitration + +- **GIVEN** `server_anchored_replay_once` is enabled +- **AND** a turn-state-only hard continuation has a live durable owner +- **AND** its retry circuit is cooling down before submission +- **WHEN** the request reaches bridge startup +- **THEN** the proxy waits for the bounded cooldown instead of returning 503 +- **AND** it sends no upstream request during the wait +- **AND** normal durable operation admission runs after cooldown + +#### Scenario: Missing durable fence remains fail closed + +- **GIVEN** a turn-state-only hard continuation has no durable session or owner + epoch +- **WHEN** its retry circuit is cooling down +- **THEN** the proxy does not wait or dispatch upstream +- **AND** it returns the existing cooldown failure with a retry hint + +#### Scenario: Operation ledger disabled remains fail closed + +- **GIVEN** ambiguous continuation recovery mode is enabled +- **AND** a turn-state-only hard continuation has a live durable session and + owner epoch +- **AND** the durable operation ledger is disabled +- **WHEN** its retry circuit is cooling down before submission +- **THEN** the proxy preserves the existing cooldown failure +- **AND** it does not wait or dispatch upstream + +#### Scenario: Default mode remains fail closed + +- **GIVEN** ambiguous continuation recovery mode is `fail_closed` +- **WHEN** any continuity-bound hard request arrives during cooldown +- **THEN** the proxy preserves the existing immediate cooldown failure +- **AND** it does not create or claim a durable recovery operation + +#### Scenario: Request budget expires while waiting + +- **GIVEN** an operation-fenced hard turn is allowed to wait through cooldown +- **AND** its request budget expires before the cooldown does +- **WHEN** the bounded wait reaches the request deadline +- **THEN** the proxy releases the request reservation and returns a terminal + timeout +- **AND** it does not submit `response.create` after the deadline + +#### Scenario: Cooldown waiter stays within the per-session queue limit + +- **GIVEN** an operation-fenced hard turn is eligible to wait through cooldown +- **AND** the bridge session is already at its configured queue limit +- **WHEN** the request reaches the cooldown wait point before submission +- **THEN** the proxy rejects the request with the existing bridge queue full + error +- **AND** it does not sleep or dispatch upstream + +#### Scenario: Durable ownership is renewed while the cooldown wait is pending + +- **GIVEN** an operation-fenced hard turn is waiting through startup cooldown +- **AND** the cooldown exceeds one durable lease refresh cadence +- **WHEN** the wait continues before submission +- **THEN** the proxy renews and revalidates the durable owner lease before the + wait completes +- **AND** it fails closed if durable ownership changes during the wait diff --git a/openspec/changes/hold-operation-fenced-hard-turn-cooldown/tasks.md b/openspec/changes/hold-operation-fenced-hard-turn-cooldown/tasks.md new file mode 100644 index 0000000000..3f7ccce4ae --- /dev/null +++ b/openspec/changes/hold-operation-fenced-hard-turn-cooldown/tasks.md @@ -0,0 +1,8 @@ +- [x] 1. Reproduce the production turn-state-only startup cooldown as a unit + regression that currently returns 503 before submission. +- [x] 2. Hold only explicitly enabled, zero-event, durable operation-fenced hard + turns through the bounded cooldown. +- [x] 3. Preserve fail-closed behavior when the durable session/owner proof is + absent and keep one-shot recovery bounded by the existing atomic claim. +- [x] 4. Run focused tests, relevant bridge suites, Ruff, type/architecture + checks, whitespace checks, and strict OpenSpec validation. diff --git a/tests/integration/test_proxy_api_extended.py b/tests/integration/test_proxy_api_extended.py index 99431da8fe..a52bfce166 100644 --- a/tests/integration/test_proxy_api_extended.py +++ b/tests/integration/test_proxy_api_extended.py @@ -2965,6 +2965,66 @@ async def stream_responses(self, *args, **kwargs): assert any("response.completed" in chunk for chunk in chunks) +@pytest.mark.asyncio +async def test_codex_route_stream_responses_keeps_client_alive_while_bridge_cooldown_delays_first_event( + monkeypatch, +): + upstream_started = asyncio.Event() + release_upstream = asyncio.Event() + + class _FakeService: + async def rate_limit_headers(self): + return {} + + async def stream_responses(self, *args, **kwargs): + del args, kwargs + upstream_started.set() + _signal_propagated_capacity_startup_ready() + await release_upstream.wait() + yield _sse_event({"type": "response.in_progress", "response": {"id": "resp_cooldown_wait"}}) + yield _sse_event({"type": "response.completed", "response": {"id": "resp_cooldown_wait"}}) + + settings = SimpleNamespace( + http_responses_session_bridge_enabled=False, + sse_keepalive_interval_seconds=0.01, + proxy_account_stream_recovery_reserve=1, + proxy_api_key_fair_share_congestion_threshold_pct=0, + ) + monkeypatch.setattr(proxy_api_module, "get_settings", lambda: settings) + monkeypatch.setattr(proxy_api_module.proxy_service_module, "get_settings", lambda: settings) + + request = Request( + { + "type": "http", + "method": "POST", + "path": "/backend-api/codex/responses", + "headers": [], + } + ) + payload = proxy_api_module.ResponsesRequest.model_validate( + {"model": "gpt-5.1", "instructions": "hi", "input": [], "stream": True} + ) + + response = await proxy_api_module._stream_responses( + request, + payload, + ProxyContext(service=cast(proxy_module.ProxyService, _FakeService())), + api_key=None, + enforce_openai_sdk_contract=False, + ) + + assert isinstance(response, StreamingResponse) + assert upstream_started.is_set() is True + iterator = response.body_iterator.__aiter__() + first_chunk = await asyncio.wait_for(iterator.__anext__(), timeout=0.2) + assert first_chunk == CODEX_KEEPALIVE_FRAME + release_upstream.set() + second_chunk = cast(str, await asyncio.wait_for(iterator.__anext__(), timeout=0.2)) + third_chunk = cast(str, await asyncio.wait_for(iterator.__anext__(), timeout=0.2)) + assert "response.in_progress" in second_chunk + assert "response.completed" in third_chunk + + @pytest.mark.asyncio async def test_proxy_stream_retries_rate_limit_then_success(async_client, monkeypatch): expected_account_id_1 = await _import_account(async_client, "acc_1", "one@example.com") diff --git a/tests/unit/test_proxy_http_bridge.py b/tests/unit/test_proxy_http_bridge.py index 761b631131..9d4ea02384 100644 --- a/tests/unit/test_proxy_http_bridge.py +++ b/tests/unit/test_proxy_http_bridge.py @@ -6046,6 +6046,442 @@ async def test_http_bridge_startup_cooldown_releases_api_key_reservation( assert request_state.api_key_reservation is None +@pytest.mark.asyncio +async def test_http_bridge_one_shot_hard_turn_waits_through_startup_cooldown( + monkeypatch: pytest.MonkeyPatch, +) -> None: + service = proxy_service.ProxyService(cast(Any, nullcontext())) + session = _make_bridge_session(key_value="sid-hard-turn-cooldown-wait") + request_state = proxy_service._WebSocketRequestState( + request_id="req-hard-turn-cooldown-wait", + model="gpt-5.6", + service_tier=None, + reasoning_effort=None, + api_key_reservation=None, + started_at=time.monotonic(), + event_queue=asyncio.Queue(), + transport="http", + session_id="turn-state-hard-anchor", + hard_continuity_anchor=True, + ) + session.durable_session_id = "durable-hard-turn-cooldown-wait" + session.durable_owner_epoch = 7 + cooldown = AsyncMock(side_effect=[0.01, 0.0]) + submit = AsyncMock(side_effect=RuntimeError("submitted after cooldown")) + sleeps: list[float] = [] + + async def sleep(delay: float) -> None: + sleeps.append(delay) + + monkeypatch.setattr( + proxy_service, + "get_settings", + lambda: _make_app_settings( + http_responses_session_bridge_ambiguous_continuation_recovery_mode="server_anchored_replay_once", + ), + ) + monkeypatch.setattr(service, "_http_bridge_precreated_retry_cooldown_seconds", cooldown) + monkeypatch.setattr(service, "_submit_http_bridge_request", submit) + monkeypatch.setattr(http_bridge_streaming_module.asyncio, "sleep", sleep) + + with pytest.raises(RuntimeError, match="submitted after cooldown"): + async for _ in service._stream_http_bridge_session_events( + session, + request_state=request_state, + text_data='{"type":"response.create"}', + queue_limit=8, + propagate_http_errors=True, + downstream_turn_state="turn-state-hard-anchor", + ): + pass + + assert sleeps == [pytest.approx(0.01)] + assert cooldown.await_count == 2 + submit.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_http_bridge_previous_response_anchor_bypasses_hard_turn_cooldown_wait( + monkeypatch: pytest.MonkeyPatch, +) -> None: + service = proxy_service.ProxyService(cast(Any, nullcontext())) + session = _make_bridge_session(key_value="sid-anchored-replay-cooldown-bypass") + session.durable_session_id = "durable-anchored-replay-cooldown-bypass" + session.durable_owner_epoch = 8 + request_state = proxy_service._WebSocketRequestState( + request_id="req-anchored-replay-cooldown-bypass", + model="gpt-5.6", + service_tier=None, + reasoning_effort=None, + api_key_reservation=None, + started_at=time.monotonic(), + event_queue=asyncio.Queue(), + transport="http", + previous_response_id="resp-anchor-before-cooldown", + hard_continuity_anchor=True, + ) + cooldown = AsyncMock(return_value=30.0) + submit = AsyncMock(side_effect=RuntimeError("submitted without cooldown wait")) + sleep = AsyncMock() + + monkeypatch.setattr( + proxy_service, + "get_settings", + lambda: _make_app_settings( + http_responses_session_bridge_ambiguous_continuation_recovery_mode="server_anchored_replay_once", + ), + ) + monkeypatch.setattr(service, "_http_bridge_precreated_retry_cooldown_seconds", cooldown) + monkeypatch.setattr(service, "_submit_http_bridge_request", submit) + monkeypatch.setattr(http_bridge_streaming_module.asyncio, "sleep", sleep) + + with pytest.raises(RuntimeError, match="submitted without cooldown wait"): + async for _ in service._stream_http_bridge_session_events( + session, + request_state=request_state, + text_data='{"type":"response.create","previous_response_id":"resp-anchor-before-cooldown"}', + queue_limit=8, + propagate_http_errors=True, + downstream_turn_state=None, + ): + pass + + cooldown.assert_not_awaited() + sleep.assert_not_awaited() + submit.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_http_bridge_one_shot_hard_turn_without_durable_fence_fails_closed( + monkeypatch: pytest.MonkeyPatch, +) -> None: + service = proxy_service.ProxyService(cast(Any, nullcontext())) + session = _make_bridge_session(key_value="sid-hard-turn-no-durable-fence") + request_state = proxy_service._WebSocketRequestState( + request_id="req-hard-turn-no-durable-fence", + model="gpt-5.6", + service_tier=None, + reasoning_effort=None, + api_key_reservation=None, + started_at=time.monotonic(), + event_queue=asyncio.Queue(), + transport="http", + session_id="turn-state-without-durable-fence", + hard_continuity_anchor=True, + ) + submit = AsyncMock() + sleep = AsyncMock() + monkeypatch.setattr( + proxy_service, + "get_settings", + lambda: _make_app_settings( + http_responses_session_bridge_ambiguous_continuation_recovery_mode="server_anchored_replay_once", + ), + ) + monkeypatch.setattr(service, "_http_bridge_precreated_retry_cooldown_seconds", AsyncMock(return_value=30.0)) + monkeypatch.setattr(service, "_submit_http_bridge_request", submit) + monkeypatch.setattr(http_bridge_streaming_module.asyncio, "sleep", sleep) + + with pytest.raises(ProxyResponseError) as exc_info: + async for _ in service._stream_http_bridge_session_events( + session, + request_state=request_state, + text_data='{"type":"response.create"}', + queue_limit=8, + propagate_http_errors=True, + downstream_turn_state="turn-state-without-durable-fence", + ): + pass + + assert exc_info.value.status_code == 503 + assert exc_info.value.payload["error"]["code"] == "upstream_request_timeout" + submit.assert_not_awaited() + sleep.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_http_bridge_one_shot_hard_turn_requires_operation_ledger_for_cooldown_wait( + monkeypatch: pytest.MonkeyPatch, +) -> None: + service = proxy_service.ProxyService(cast(Any, nullcontext())) + session = _make_bridge_session(key_value="sid-hard-turn-no-ledger") + session.durable_session_id = "durable-hard-turn-no-ledger" + session.durable_owner_epoch = 11 + request_state = proxy_service._WebSocketRequestState( + request_id="req-hard-turn-no-ledger", + model="gpt-5.6", + service_tier=None, + reasoning_effort=None, + api_key_reservation=None, + started_at=time.monotonic(), + event_queue=asyncio.Queue(), + transport="http", + session_id="turn-state-no-ledger", + hard_continuity_anchor=True, + ) + submit = AsyncMock() + sleep = AsyncMock() + monkeypatch.setattr( + proxy_service, + "get_settings", + lambda: _make_app_settings( + http_responses_session_bridge_ambiguous_continuation_recovery_mode="server_anchored_replay_once", + http_responses_session_bridge_operation_ledger_enabled=False, + ), + ) + monkeypatch.setattr(service, "_http_bridge_precreated_retry_cooldown_seconds", AsyncMock(return_value=30.0)) + monkeypatch.setattr(service, "_submit_http_bridge_request", submit) + monkeypatch.setattr(http_bridge_streaming_module.asyncio, "sleep", sleep) + + with pytest.raises(ProxyResponseError) as exc_info: + async for _ in service._stream_http_bridge_session_events( + session, + request_state=request_state, + text_data='{"type":"response.create"}', + queue_limit=8, + propagate_http_errors=True, + downstream_turn_state="turn-state-no-ledger", + ): + pass + + assert exc_info.value.status_code == 503 + assert exc_info.value.payload["error"]["code"] == "upstream_request_timeout" + submit.assert_not_awaited() + sleep.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_http_bridge_one_shot_hard_turn_cooldown_wait_rejects_when_queue_is_full( + monkeypatch: pytest.MonkeyPatch, +) -> None: + service = proxy_service.ProxyService(cast(Any, nullcontext())) + session = _make_bridge_session(key_value="sid-hard-turn-cooldown-queue-full", queued_request_count=8) + session.durable_session_id = "durable-hard-turn-cooldown-queue-full" + session.durable_owner_epoch = 12 + request_state = proxy_service._WebSocketRequestState( + request_id="req-hard-turn-cooldown-queue-full", + model="gpt-5.6", + service_tier=None, + reasoning_effort=None, + api_key_reservation=None, + started_at=time.monotonic(), + event_queue=asyncio.Queue(), + transport="http", + session_id="turn-state-cooldown-queue-full", + hard_continuity_anchor=True, + ) + submit = AsyncMock() + monkeypatch.setattr( + proxy_service, + "get_settings", + lambda: _make_app_settings( + http_responses_session_bridge_ambiguous_continuation_recovery_mode="server_anchored_replay_once", + ), + ) + monkeypatch.setattr(service, "_http_bridge_precreated_retry_cooldown_seconds", AsyncMock(return_value=30.0)) + monkeypatch.setattr(service, "_submit_http_bridge_request", submit) + + with pytest.raises(ProxyResponseError) as exc_info: + async for _ in service._stream_http_bridge_session_events( + session, + request_state=request_state, + text_data='{"type":"response.create"}', + queue_limit=8, + propagate_http_errors=True, + downstream_turn_state="turn-state-cooldown-queue-full", + ): + pass + + assert exc_info.value.status_code == 429 + assert exc_info.value.payload["error"]["code"] == "bridge_queue_full" + submit.assert_not_awaited() + assert session.queued_request_count == 8 + + +@pytest.mark.asyncio +async def test_http_bridge_one_shot_hard_turn_renews_durable_lease_while_waiting( + monkeypatch: pytest.MonkeyPatch, +) -> None: + service = proxy_service.ProxyService(cast(Any, nullcontext())) + session = _make_bridge_session(key_value="sid-hard-turn-renew-wait") + session.durable_session_id = "durable-hard-turn-renew-wait" + session.durable_owner_epoch = 13 + request_state = proxy_service._WebSocketRequestState( + request_id="req-hard-turn-renew-wait", + model="gpt-5.6", + service_tier=None, + reasoning_effort=None, + api_key_reservation=None, + started_at=time.monotonic(), + event_queue=asyncio.Queue(), + transport="http", + session_id="turn-state-renew-wait", + hard_continuity_anchor=True, + ) + renew_live_session = AsyncMock( + return_value=SimpleNamespace( + owner_instance_id="instance-hard-turn-renew", + owner_epoch=13, + ) + ) + service._durable_bridge = cast(Any, SimpleNamespace(renew_live_session=renew_live_session)) + cooldown = AsyncMock(side_effect=[25.0, 0.0]) + submit = AsyncMock(side_effect=RuntimeError("submitted after renewed cooldown")) + slept: list[float] = [] + + async def sleep(delay: float) -> None: + slept.append(delay) + + monkeypatch.setattr( + proxy_service, + "get_settings", + lambda: _make_app_settings( + http_responses_session_bridge_ambiguous_continuation_recovery_mode="server_anchored_replay_once", + http_responses_session_bridge_instance_id="instance-hard-turn-renew", + ), + ) + monkeypatch.setattr(service, "_http_bridge_precreated_retry_cooldown_seconds", cooldown) + monkeypatch.setattr(service, "_submit_http_bridge_request", submit) + monkeypatch.setattr(http_bridge_streaming_module.asyncio, "sleep", sleep) + + with pytest.raises(RuntimeError, match="submitted after renewed cooldown"): + async for _ in service._stream_http_bridge_session_events( + session, + request_state=request_state, + text_data='{"type":"response.create"}', + queue_limit=8, + propagate_http_errors=True, + downstream_turn_state="turn-state-renew-wait", + ): + pass + + assert slept == [pytest.approx(10.0), pytest.approx(10.0), pytest.approx(5.0)] + assert renew_live_session.await_count == 2 + submit.assert_awaited_once() + assert session.queued_request_count == 0 + + +@pytest.mark.asyncio +async def test_http_bridge_one_shot_hard_turn_fails_closed_when_lease_renewal_raises( + monkeypatch: pytest.MonkeyPatch, +) -> None: + service = proxy_service.ProxyService(cast(Any, nullcontext())) + session = _make_bridge_session(key_value="sid-hard-turn-renew-failure") + session.durable_session_id = "durable-hard-turn-renew-failure" + session.durable_owner_epoch = 14 + request_state = proxy_service._WebSocketRequestState( + request_id="req-hard-turn-renew-failure", + model="gpt-5.6", + service_tier=None, + reasoning_effort=None, + api_key_reservation=None, + started_at=time.monotonic(), + event_queue=asyncio.Queue(), + transport="http", + session_id="turn-state-renew-failure", + hard_continuity_anchor=True, + ) + renew_live_session = AsyncMock(side_effect=RuntimeError("durable store unavailable")) + service._durable_bridge = cast(Any, SimpleNamespace(renew_live_session=renew_live_session)) + submit = AsyncMock() + + monkeypatch.setattr( + proxy_service, + "get_settings", + lambda: _make_app_settings( + http_responses_session_bridge_ambiguous_continuation_recovery_mode="server_anchored_replay_once", + http_responses_session_bridge_instance_id="instance-hard-turn-renew-failure", + ), + ) + monkeypatch.setattr( + service, + "_http_bridge_precreated_retry_cooldown_seconds", + AsyncMock(return_value=25.0), + ) + monkeypatch.setattr(service, "_submit_http_bridge_request", submit) + monkeypatch.setattr(http_bridge_streaming_module.asyncio, "sleep", AsyncMock()) + + with pytest.raises(ProxyResponseError) as exc_info: + async for _ in service._stream_http_bridge_session_events( + session, + request_state=request_state, + text_data='{"type":"response.create"}', + queue_limit=8, + propagate_http_errors=True, + downstream_turn_state="turn-state-renew-failure", + ): + pass + + assert exc_info.value.status_code == 502 + assert exc_info.value.payload["error"]["code"] == "bridge_continuity_persistence_failed" + renew_live_session.assert_awaited_once() + submit.assert_not_awaited() + assert session.closed is True + assert session.upstream_control.reconnect_requested is True + assert session.upstream_control.retire_after_drain is True + assert session.queued_request_count == 0 + + +@pytest.mark.asyncio +async def test_http_bridge_one_shot_hard_turn_does_not_submit_after_wait_budget( + monkeypatch: pytest.MonkeyPatch, +) -> None: + service = proxy_service.ProxyService(cast(Any, nullcontext())) + session = _make_bridge_session(key_value="sid-hard-turn-wait-budget") + session.durable_session_id = "durable-hard-turn-wait-budget" + session.durable_owner_epoch = 9 + reservation = cast(Any, object()) + request_state = proxy_service._WebSocketRequestState( + request_id="req-hard-turn-wait-budget", + model="gpt-5.6", + service_tier=None, + reasoning_effort=None, + api_key_reservation=reservation, + started_at=time.monotonic(), + event_queue=asyncio.Queue(), + transport="http", + session_id="turn-state-wait-budget", + hard_continuity_anchor=True, + ) + clock = SimpleNamespace(now=100.0) + submit = AsyncMock() + release = AsyncMock() + + async def sleep(delay: float) -> None: + clock.now += delay + + monkeypatch.setattr( + proxy_service, + "get_settings", + lambda: _make_app_settings( + http_responses_session_bridge_ambiguous_continuation_recovery_mode="server_anchored_replay_once", + ), + ) + monkeypatch.setattr(http_bridge_streaming_module._service_time(), "monotonic", lambda: clock.now) + monkeypatch.setattr(service, "_http_bridge_precreated_retry_cooldown_seconds", AsyncMock(return_value=30.0)) + monkeypatch.setattr(service, "_submit_http_bridge_request", submit) + monkeypatch.setattr(service, "_release_websocket_request_state_reservation", release) + monkeypatch.setattr(http_bridge_streaming_module.asyncio, "sleep", sleep) + + with pytest.raises(ProxyResponseError) as exc_info: + async for _ in service._stream_http_bridge_session_events( + session, + request_state=request_state, + text_data='{"type":"response.create"}', + queue_limit=8, + propagate_http_errors=True, + downstream_turn_state="turn-state-wait-budget", + request_deadline=105.0, + ): + pass + + assert exc_info.value.status_code == 503 + assert exc_info.value.payload["error"]["code"] == "upstream_request_timeout" + submit.assert_not_awaited() + release.assert_awaited_once_with(request_state) + assert request_state.api_key_reservation is None + + @pytest.mark.asyncio async def test_http_bridge_replay_detach_releases_reservation_without_pending_ownership( monkeypatch: pytest.MonkeyPatch,