diff --git a/app/modules/api_keys/repository.py b/app/modules/api_keys/repository.py index 8fb9b77a8..0c7df5b90 100644 --- a/app/modules/api_keys/repository.py +++ b/app/modules/api_keys/repository.py @@ -380,6 +380,12 @@ async def delete(self, key_id: str) -> bool: async def commit(self) -> None: await self._session.commit() + async def update_last_used(self, key_id: str, *, commit: bool = True) -> None: + """Compatibility touch for maintenance and durability checks.""" + await self._session.execute(update(ApiKey).where(ApiKey.id == key_id).values(last_used_at=utcnow())) + if commit: + await self._session.commit() + async def rollback(self) -> None: await self._session.rollback() diff --git a/app/modules/proxy/_service/http_bridge/mixin.py b/app/modules/proxy/_service/http_bridge/mixin.py index b5d7b3389..8dea063e0 100644 --- a/app/modules/proxy/_service/http_bridge/mixin.py +++ b/app/modules/proxy/_service/http_bridge/mixin.py @@ -2367,6 +2367,7 @@ async def abort_selected_handoff() -> None: await self._unregister_http_bridge_turn_states(session) await self._unregister_http_bridge_previous_response_ids(session) session.last_completed_response_id = None + session.last_completed_response_account_id = None session.last_completed_input_count = 0 session.last_completed_input_prefix_fingerprint = None session.last_pending_tool_calls.clear() diff --git a/app/modules/proxy/_service/http_bridge/streaming.py b/app/modules/proxy/_service/http_bridge/streaming.py index da677d8c3..50b42cdeb 100644 --- a/app/modules/proxy/_service/http_bridge/streaming.py +++ b/app/modules/proxy/_service/http_bridge/streaming.py @@ -2173,29 +2173,58 @@ def switch_to_account_neutral_replay() -> None: and isinstance(recovery_payload.input, list) and len(recovery_payload.input) > durable_full_resend_anchor_count ): - recovery_input = cast(list[JsonValue], recovery_payload.input) - recovery_anchor_input_count = len(recovery_input) - recovery_anchor_input_fingerprint = _fingerprint_input_items(recovery_input) - recovery_payload = recovery_payload.model_copy( - update={ - "previous_response_id": durable_lookup.latest_response_id, - "input": recovery_input[durable_full_resend_anchor_count:], - } - ) - if durable_lookup.latest_response_id != session.last_completed_response_id: - session.last_pending_tool_calls = {} - session.last_completed_response_id = durable_lookup.latest_response_id - session.last_completed_input_count = durable_full_resend_anchor_count - session.last_completed_input_prefix_fingerprint = durable_full_resend_anchor_fingerprint - _log_http_bridge_event( - "owner_forward_recovery_anchor_injected", - bridge_session_key, - account_id=durable_lookup.account_id, - model=recovery_payload.model, - detail=f"response_id={durable_lookup.latest_response_id}", - cache_key_family=bridge_session_key.affinity_kind, - model_class=_extract_model_class(recovery_payload.model) if recovery_payload.model else None, - ) + # The recovery rebind above is allowed to bind this session to + # an account other than the durable owner. A previous_response_id + # is account-scoped upstream, so replaying the durable anchor on + # a different account sends an anchor upstream cannot resolve + # with the history trimmed away: no response.created arrives and + # the per-bridge response-create gate wedges. Resend the full + # history on the serving account instead. + if durable_lookup.account_id != session.account.id: + _log_http_bridge_event( + "cross_account_anchor_declined", + bridge_session_key, + account_id=session.account.id, + model=recovery_payload.model, + detail=( + "site=owner_forward_recovery, " + f"response_id={durable_lookup.latest_response_id}, " + f"anchor_account_id={durable_lookup.account_id}, " + "outcome=full_history_resend" + ), + cache_key_family=bridge_session_key.affinity_kind, + model_class=_extract_model_class(recovery_payload.model) + if recovery_payload.model + else None, + owner_check_applied=True, + ) + else: + recovery_input = cast(list[JsonValue], recovery_payload.input) + recovery_anchor_input_count = len(recovery_input) + recovery_anchor_input_fingerprint = _fingerprint_input_items(recovery_input) + recovery_payload = recovery_payload.model_copy( + update={ + "previous_response_id": durable_lookup.latest_response_id, + "input": recovery_input[durable_full_resend_anchor_count:], + } + ) + if durable_lookup.latest_response_id != session.last_completed_response_id: + session.last_pending_tool_calls = {} + session.last_completed_response_id = durable_lookup.latest_response_id + session.last_completed_response_account_id = durable_lookup.account_id + session.last_completed_input_count = durable_full_resend_anchor_count + session.last_completed_input_prefix_fingerprint = durable_full_resend_anchor_fingerprint + _log_http_bridge_event( + "owner_forward_recovery_anchor_injected", + bridge_session_key, + account_id=durable_lookup.account_id, + model=recovery_payload.model, + detail=f"response_id={durable_lookup.latest_response_id}", + cache_key_family=bridge_session_key.affinity_kind, + model_class=_extract_model_class(recovery_payload.model) + if recovery_payload.model + else None, + ) recovery_injected_input = _http_bridge_interrupted_tool_outputs_input( session, payload=recovery_payload, @@ -2305,6 +2334,12 @@ def switch_to_account_neutral_replay() -> None: # must not trigger interrupted-output injection. session.last_pending_tool_calls = {} session.last_completed_response_id = durable_lookup.latest_response_id + # The durable anchor is owned by the durable session's account, which + # may differ from this session's account after a failover. Record the + # owner so the session-anchor injection below can refuse to replay a + # cross-account previous_response_id (upstream cannot resolve it and + # would stall with no response.created — a wedged response-create gate). + session.last_completed_response_account_id = durable_lookup.account_id session.last_completed_input_count = durable_full_resend_anchor_count session.last_completed_input_prefix_fingerprint = durable_full_resend_anchor_fingerprint # --- Session-level previous_response_id injection --- @@ -2333,17 +2368,45 @@ def switch_to_account_neutral_replay() -> None: stored_count=stored_count_preview, stored_fingerprint=stored_fingerprint_preview, ) + # A previous_response_id is account-scoped upstream: only the account that + # created the response can resume it. If this session's serving account is + # not the anchor's owner (e.g. the session failed over after the durable + # owner became unavailable), injecting the anchor sends an unresolvable + # previous_response_id upstream with the history trimmed away — upstream + # then never emits response.created and the response-create gate wedges + # ("idle timeout waiting for SSE"). Fall through to a full-history resend. + session_anchor_account_owned = ( + session.last_completed_response_account_id is not None + and session.last_completed_response_account_id == session.account.id + ) recovery_session_can_anchor = is_http_bridge_account_neutral_replay( kind=session.key.affinity_kind, key=session.key.affinity_key, ) and (not _http_bridge_payload_looks_like_full_resend(effective_payload) or session_anchor_trimmable) - if ( + session_anchor_candidate = ( session.codex_session and not proxy_injected_previous_response_id and effective_payload.previous_response_id is None and session.last_completed_response_id is not None and (session_anchor_trimmable or recovery_session_can_anchor) - ): + ) + if session_anchor_candidate and not session_anchor_account_owned: + _log_http_bridge_event( + "cross_account_anchor_declined", + session.key, + account_id=session.account.id, + model=effective_payload.model, + detail=( + "site=session_anchor, " + f"response_id={session.last_completed_response_id}, " + f"anchor_account_id={session.last_completed_response_account_id}, " + "outcome=full_history_resend" + ), + cache_key_family=session.key.affinity_kind, + model_class=_extract_model_class(effective_payload.model) if effective_payload.model else None, + owner_check_applied=True, + ) + if session_anchor_candidate and session_anchor_account_owned: fresh_upstream_request_text = text_data session_level_payload_looks_like_full_resend = _http_bridge_payload_looks_like_full_resend( effective_payload diff --git a/app/modules/proxy/_service/http_bridge/upstream_events.py b/app/modules/proxy/_service/http_bridge/upstream_events.py index e9fe0c660..30f237585 100644 --- a/app/modules/proxy/_service/http_bridge/upstream_events.py +++ b/app/modules/proxy/_service/http_bridge/upstream_events.py @@ -2016,6 +2016,10 @@ async def _process_parsed_http_bridge_upstream_event( # anchor for continuity lookups. if response_id is not None: session.last_completed_response_id = response_id + # This response was completed on the session's current account, so + # that account owns the anchor. Record it so the anchor is only + # replayed on the same account (never after a cross-account failover). + session.last_completed_response_account_id = session.account.id # Remember which tool-call items the completed response left # pending so an anchored follow-up that omits their outputs # (interrupted turn) can receive synthetic interrupted diff --git a/app/modules/proxy/_service/support.py b/app/modules/proxy/_service/support.py index 7c26e6f81..ca7233e00 100644 --- a/app/modules/proxy/_service/support.py +++ b/app/modules/proxy/_service/support.py @@ -1022,6 +1022,12 @@ class _HTTPBridgeSession: previous_response_alias_registration_generations: dict[str, int] = field(default_factory=dict) last_completed_input_count: int = 0 last_completed_response_id: str | None = None + # Account that owns ``last_completed_response_id``. A previous_response_id + # anchor is account-scoped upstream, so it may only be replayed on the same + # account; when the session fails over to a different account this diverges + # from ``account.id`` and the anchor must NOT be injected. Kept in sync with + # ``last_completed_response_id`` at every setter. + last_completed_response_account_id: str | None = None last_completed_input_prefix_fingerprint: str | None = None last_pending_tool_calls: dict[str, str] = field(default_factory=dict) durable_session_id: str | None = None diff --git a/openspec/changes/fix-cross-account-previous-response-anchor/proposal.md b/openspec/changes/fix-cross-account-previous-response-anchor/proposal.md new file mode 100644 index 000000000..29acea927 --- /dev/null +++ b/openspec/changes/fix-cross-account-previous-response-anchor/proposal.md @@ -0,0 +1,65 @@ +# Fix cross-account compact previous_response_id anchor wedge + +## Why + +The HTTP-bridge "compact anchor" continuity optimization injects +`previous_response_id = session.last_completed_response_id` and trims the +already-stored history prefix so a follow-up turn only carries the new items. + +A `previous_response_id` is **account-scoped** on the upstream Responses API: +only the account that created the response can resume it. The injection had no +account-ownership check. When a Codex session fails over to a different account +(e.g. the durable owner account became unavailable, or a durable session record +is restored onto a new session bound to another account), the anchor points at a +response the serving account never created. Upstream then accepts the WebSocket +`response.create` but **never emits `response.created`**, and because the history +was trimmed away there is no fallback. The per-bridge `response_create_gate` +(a `Semaphore(1)`) stays held: the holder's own client eventually reports +`stream disconnected before completion: idle timeout waiting for SSE`, and later +requests on the same session time out as `codex-lb is temporarily overloaded +during http_bridge_response_create_gate`. + +Observed live on 2026-07-10: sol sessions that fanned across three accounts +repeatedly wedged on the same anchor (`resp_0bc0310d…`) even though the accounts +had quota. Freeing the gate (the stuck-gate retire backstop) does not stop the +recurrence because the session re-injects the same cross-account anchor. + +## What Changes + +- Track the account that owns `last_completed_response_id` on the bridge session + (`last_completed_response_account_id`), set in lockstep at every setter: the + real upstream `response.completed` path records the session's current account; + the durable-restore path records the durable owner account. +- The session-level compact-anchor injection MUST only fire when the anchor's + owning account equals the session's serving account. Otherwise the request + falls through to a full-history resend (correct output, slightly more tokens), + never a cross-account `previous_response_id`. +- The owner-forward recovery anchor injection gets the same guard. That rebind + runs with `allow_previous_response_recovery_rebind` / + `allow_bootstrap_owner_rebind`, which are explicitly allowed to bind the + session to an account other than the durable owner, so it is the second site + where a proxy-injected anchor can cross an account boundary. +- Both declines emit a `cross_account_anchor_declined` bridge event naming the + injection site and the anchor's owning account, so the wedge family stays + observable from bridge event logs. + +The remaining proxy-side injection site, the pre-binding durable "fresh reattach" +anchor, is already covered: setting `previous_response_id` there makes the durable +owner a *required* account (`require_preferred_account`), so session creation and +session reuse both refuse to serve that request on another account, and an +unavailable owner degrades along the account-neutral full-resend path instead. + +## Impact + +- Affected specs: `sticky-session-operations` +- Affected code: `_service/support.py` (`_HTTPBridgeSession`), + `_service/http_bridge/streaming.py` (both injection guards + durable-restore + owner), `_service/http_bridge/upstream_events.py` (completion owner), + `_service/http_bridge/mixin.py` (clear the owner with the anchor on an + account-changing reconnect). +- Behavior: on cross-account failover, continuity is preserved by resending full + history instead of an unresolvable anchor. No client-visible protocol change. +- Follow-ups (tracked separately, not in this change): the WebSocket-transport + anchor path (`websocket_session_anchor_injected`), and a proactive + `response.created` watchdog that replays the stored full-history payload if any + anchored request stalls. diff --git a/openspec/changes/fix-cross-account-previous-response-anchor/specs/sticky-session-operations/spec.md b/openspec/changes/fix-cross-account-previous-response-anchor/specs/sticky-session-operations/spec.md new file mode 100644 index 000000000..144b057fd --- /dev/null +++ b/openspec/changes/fix-cross-account-previous-response-anchor/specs/sticky-session-operations/spec.md @@ -0,0 +1,58 @@ +## ADDED Requirements + +### Requirement: Compact previous_response_id anchors are account-scoped + +codex-lb MUST NOT inject a compact `previous_response_id` anchor whose owning account differs from the account that will serve the request. + +The HTTP-bridge compact-anchor injection reduces payload size by replacing +already-stored history with a proxy-supplied `previous_response_id`, and a +`previous_response_id` can only be resumed by the account that created it. The +rule applies to every injection site that runs after the serving account is +bound: the session-level anchor (`session.last_completed_response_id`) and the +owner-forward recovery anchor (`durable_lookup.latest_response_id`, injected +after a rebind that is allowed to land on a different account). + +codex-lb MUST record the account that owns `last_completed_response_id` whenever +that value is set — from a real upstream `response.completed` (the session's +current account) or from a durable-session restore (the durable owner account) — +and keep the two in sync. + +Injection sites that run before the serving account is bound stay covered by the +existing required-continuity-owner pin, which fails the request rather than +serving a proxy-injected anchor on a different account. + +#### Scenario: Anchor injected when the serving account owns it + +- **WHEN** a Codex session follow-up turn is eligible for compact-anchor injection +- **AND** the account that owns `last_completed_response_id` equals the session's + serving account +- **THEN** codex-lb injects `previous_response_id = last_completed_response_id` + and trims the already-stored history prefix + +#### Scenario: Anchor skipped after cross-account failover + +- **WHEN** a Codex session follow-up turn is eligible for compact-anchor injection +- **AND** the account that owns `last_completed_response_id` differs from the + session's serving account (for example the session failed over after the durable + owner account became unavailable) +- **THEN** codex-lb MUST NOT inject the anchor +- **AND** codex-lb resends the full history to the serving account so continuity + is preserved without an unresolvable `previous_response_id` +- **AND** the request MUST NOT stall waiting for a `response.created` that upstream + will never send for an anchor the serving account does not own + +#### Scenario: Owner-forward recovery anchor skipped after a cross-account rebind + +- **WHEN** an owner forward fails and the local recovery rebind binds the session + to an account other than the durable record's owner +- **AND** the durable record still carries a `latest_response_id` the recovery + request would otherwise anchor on +- **THEN** codex-lb MUST NOT inject that anchor +- **AND** the recovery request keeps its full input instead of a trimmed suffix + +#### Scenario: Declined anchors are observable + +- **WHEN** codex-lb declines a compact anchor because the serving account does not + own it +- **THEN** codex-lb logs a `cross_account_anchor_declined` bridge event naming the + injection site, the anchor's owning account, and the full-history-resend outcome diff --git a/openspec/changes/fix-cross-account-previous-response-anchor/tasks.md b/openspec/changes/fix-cross-account-previous-response-anchor/tasks.md new file mode 100644 index 000000000..37e26aacc --- /dev/null +++ b/openspec/changes/fix-cross-account-previous-response-anchor/tasks.md @@ -0,0 +1,25 @@ +# Tasks + +- [x] Add `last_completed_response_account_id` to `_HTTPBridgeSession` +- [x] Record the serving account on the real `response.completed` setter + (`upstream_events.py`) +- [x] Record the durable owner account on the durable-restore setters + (`streaming.py`), and clear it with the anchor on an account-changing + reconnect (`mixin.py`) +- [x] Gate session-level compact-anchor injection on + `last_completed_response_account_id == session.account.id` +- [x] Gate the owner-forward recovery anchor injection on + `durable_lookup.account_id == session.account.id` — the recovery rebind is + explicitly allowed to land on a different account +- [x] Log `cross_account_anchor_declined` at both sites so the wedge family stays + observable from bridge events +- [x] Regression tests: anchor injected when same-account; anchor skipped and + full history resent when cross-account, at both injection sites +- [x] Bridge-level regression: real bridge session, real response-create gate, + upstream that models the account scope of `previous_response_id` (silence + for a foreign anchor) — the turn settles instead of wedging +- [x] Verify full `test_proxy_http_bridge` + bridge integration suites green +- [ ] Follow-up (separate change): proactive `response.created` watchdog that + replays stored full-history payload on stall +- [ ] Follow-up (separate change): audit the WebSocket-transport anchor path + (`websocket_session_anchor_injected`) for the same cross-account exposure diff --git a/tests/integration/test_http_responses_bridge.py b/tests/integration/test_http_responses_bridge.py index 41ee741d2..c03b056a6 100644 --- a/tests/integration/test_http_responses_bridge.py +++ b/tests/integration/test_http_responses_bridge.py @@ -30,7 +30,7 @@ set_request_scope_id, ) from app.core.utils.time import utcnow -from app.db.models import Account, AccountStatus, DashboardSettings, RequestLog +from app.db.models import Account, AccountStatus, DashboardSettings, HttpBridgeSessionState, RequestLog from app.db.session import SessionLocal from app.dependencies import get_proxy_service_for_app from app.modules.proxy._service import support as proxy_support @@ -533,6 +533,25 @@ async def send_text(self, text: str) -> None: self.sent_text.append(text) +class _AccountScopedAnchorUpstreamWebSocket(_FakeBridgeUpstreamWebSocket): + """Upstream that only resolves ``previous_response_id`` values it issued. + + A ``previous_response_id`` is account-scoped upstream: only the account that + created the response can resume it. A ``response.create`` carrying a foreign + anchor is accepted by the socket but never answered with ``response.created``. + Modelling that here makes a cross-account anchor observable as the production + symptom instead of a silent assertion: the turn never settles and the + per-bridge ``response_create_gate`` stays held. + """ + + async def send_text(self, text: str) -> None: + anchor = json.loads(text).get("previous_response_id") + if isinstance(anchor, str) and not anchor.startswith(self.response_id_prefix): + self.sent_text.append(text) + return + await super().send_text(text) + + class _RecordingUpstreamWebSocket(_FakeBridgeUpstreamWebSocket): pass @@ -8272,6 +8291,178 @@ async def fake_connect_responses_websocket( assert degraded_reasons == [] +@pytest.mark.asyncio +async def test_backend_responses_http_bridge_declines_cross_account_anchor_and_settles( + async_client, app_instance, monkeypatch +): + """A restored durable anchor owned by another account must not be injected. + + The durable record still names the owner account, but the owner is out of the + rotation so the bridge session is created on another account. Replaying the + owner's ``previous_response_id`` there would send an anchor upstream cannot + resolve with the history trimmed away: upstream never emits + ``response.created`` and the per-bridge response-create gate wedges. The turn + must go upstream as a full-history resend instead, and it must settle. + """ + + _install_bridge_settings(monkeypatch, enabled=True) + owner_account_id = await _import_account( + async_client, + "acc_cross_account_anchor_owner", + "cross-account-anchor-owner@example.com", + ) + serving_account_id = await _import_account( + async_client, + "acc_cross_account_anchor_serving", + "cross-account-anchor-serving@example.com", + ) + owner_account = await _get_account(owner_account_id) + serving_account = await _get_account(serving_account_id) + serving_chatgpt_account_id = cast(str, serving_account.chatgpt_account_id) + serving_upstream = _AccountScopedAnchorUpstreamWebSocket("resp_cross_account_serving") + service = get_proxy_service_for_app(app_instance) + + 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 headers, access_token, base_url, session + assert account_id_header == serving_chatgpt_account_id + return serving_upstream + + 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) + + # The durable owner account left the rotation, so selection lands on the + # other account while the restored durable record still names the owner. + pause = await async_client.post(f"/api/accounts/{owner_account_id}/pause") + assert pause.status_code == 200, pause.text + + stored_input: list[proxy_module.JsonValue] = [ + {"role": "user", "content": [{"type": "input_text", "text": "first question"}]}, + ] + + def _durable_record( + *, + account_id: str, + latest_response_id: str, + stored_items: list[proxy_module.JsonValue], + ) -> proxy_module.DurableBridgeLookup: + return proxy_module.DurableBridgeLookup( + session_id="durable-cross-account-anchor", + canonical_kind="prompt_cache", + canonical_key="cross-account-anchor-cache-key", + api_key_scope="__anonymous__", + account_id=account_id, + owner_instance_id=None, + owner_epoch=1, + lease_expires_at=None, + state=HttpBridgeSessionState.ACTIVE, + latest_turn_state=None, + latest_response_id=latest_response_id, + latest_input_item_count=len(stored_items), + latest_input_full_fingerprint=proxy_module._fingerprint_input_items(stored_items), + ) + + durable_record = _durable_record( + account_id=owner_account.id, + latest_response_id="resp_cross_account_owner_1", + stored_items=stored_input, + ) + + async def fake_lookup_request_targets(**kwargs): + del kwargs + return durable_record + + monkeypatch.setattr(service._durable_bridge, "lookup_request_targets", fake_lookup_request_targets) + + # Compaction-shaped follow-up: the stored prefix still matches, but the + # suffix carries no prior assistant output, so the account-neutral fresh + # resend projection is unavailable and the restored durable anchor is the + # only continuity candidate the session-level injection can reach for. + compacted_resend: list[proxy_module.JsonValue] = [ + *stored_input, + {"role": "user", "content": [{"type": "input_text", "text": "second question"}]}, + ] + session_id = "cross-account-anchor-session" + first_events, first_headers = await asyncio.wait_for( + _collect_sse_events_with_headers( + async_client, + "/backend-api/codex/responses", + json_body={ + "model": "gpt-5.1", + "instructions": "Return exactly OK.", + "input": compacted_resend, + "stream": True, + }, + headers={"session_id": session_id}, + ), + timeout=_TEST_SYNC_TIMEOUT_SECONDS, + ) + _assert_created_text_delta_completed(first_events) + turn_state = first_headers["x-codex-turn-state"] + + assert len(serving_upstream.sent_text) == 1 + resend_payload = json.loads(serving_upstream.sent_text[0]) + assert "previous_response_id" not in resend_payload + assert resend_payload["input"] == compacted_resend + + bridge_session = next( + candidate for candidate in service._http_bridge_sessions.values() if candidate.account.id == serving_account.id + ) + assert bridge_session.codex_session is True + # The turn settled on the serving account, so the gate is free and the + # session anchor is now owned by the account that actually created it. + assert bridge_session.response_create_gate.locked() is False + assert bridge_session.last_completed_response_id == "resp_cross_account_serving_1" + assert bridge_session.last_completed_response_account_id == serving_account.id + + # Same-account continuity is untouched: once the durable record names the + # account that actually created the response, the very next turn anchors on + # it instead of resending the whole history. + durable_record = _durable_record( + account_id=serving_account.id, + latest_response_id="resp_cross_account_serving_1", + stored_items=compacted_resend, + ) + second_events = await asyncio.wait_for( + _collect_sse_events( + async_client, + "/backend-api/codex/responses", + json_body={ + "model": "gpt-5.1", + "instructions": "Return exactly OK.", + "input": [ + *compacted_resend, + { + "type": "message", + "role": "assistant", + "status": "completed", + "content": [{"type": "output_text", "text": "OK"}], + }, + {"role": "user", "content": [{"type": "input_text", "text": "third question"}]}, + ], + "stream": True, + }, + headers={"session_id": session_id, "x-codex-turn-state": turn_state}, + ), + timeout=_TEST_SYNC_TIMEOUT_SECONDS, + ) + _assert_created_text_delta_completed(second_events) + assert len(serving_upstream.sent_text) == 2 + follow_up_payload = json.loads(serving_upstream.sent_text[1]) + assert follow_up_payload["previous_response_id"] == "resp_cross_account_serving_1" + assert bridge_session.response_create_gate.locked() is False + + @pytest.mark.asyncio async def test_backend_responses_projects_retained_encrypted_reasoning_before_replaying_to_available_account( async_client, diff --git a/tests/unit/test_proxy_http_bridge.py b/tests/unit/test_proxy_http_bridge.py index b0a61903b..0333edea4 100644 --- a/tests/unit/test_proxy_http_bridge.py +++ b/tests/unit/test_proxy_http_bridge.py @@ -3945,6 +3945,7 @@ async def test_http_bridge_precreated_completed_terminal_falls_back_to_unresolve ] assert request_state.response_id == "resp_precreated_completed" assert session.last_completed_response_id == "resp_precreated_completed" + assert session.last_completed_response_account_id == session.account.id assert session.queued_request_count == 0 assert not session.pending_requests assert session.key not in retry_circuits @@ -4070,6 +4071,7 @@ async def test_ordinary_completed_alias_rejection_preserves_successful_response( assert completed["type"] == "response.completed" assert await asyncio.wait_for(request_state.event_queue.get(), timeout=1.0) is None assert session.last_completed_response_id == "resp_ordinary_completed" + assert session.last_completed_response_account_id == session.account.id assert session.upstream_control.reconnect_requested is False assert session.upstream_control.retire_after_drain is False finalize.assert_awaited_once() @@ -8509,6 +8511,155 @@ def fake_prepare( assert prepared_previous_response_ids == [None] +async def _run_session_anchor_owner_stream( + monkeypatch: pytest.MonkeyPatch, + *, + account_id: str, + anchor_owner_account_id: str | None, +) -> list[proxy_service.ResponsesRequest]: + """Drive _stream_via_http_bridge for a trimmable session-anchor turn. + + The stored prefix matches the incoming input (so the trim branch WOULD + apply); the only variable is whether the serving account owns the anchor. + Returns the payloads passed to each prepare call. + """ + service = proxy_service.ProxyService(cast(Any, nullcontext())) + prefix_items: list[proxy_service.JsonValue] = [ + {"role": "user", "content": [{"type": "input_text", "text": "a"}]}, + {"role": "assistant", "content": [{"type": "output_text", "text": "b"}]}, + {"role": "user", "content": [{"type": "input_text", "text": "c"}]}, + ] + payload = proxy_service.ResponsesRequest.model_validate( + { + "model": "gpt-5.4", + "instructions": "hi", + "input": [*prefix_items, {"role": "user", "content": [{"type": "input_text", "text": "d"}]}], + }, + ) + request_state = proxy_service._WebSocketRequestState( + request_id="req-session-anchor-owner", + model="gpt-5.4", + service_tier=None, + reasoning_effort=None, + api_key_reservation=None, + started_at=1.0, + event_queue=asyncio.Queue(), + transport="http", + ) + event_queue = request_state.event_queue + assert event_queue is not None + await event_queue.put(None) + prepared_payloads: list[proxy_service.ResponsesRequest] = [] + + def fake_prepare( + prepared_payload: proxy_service.ResponsesRequest, + _headers: dict[str, str] | Any, + *, + api_key: proxy_service.ApiKeyData | None, + api_key_reservation: proxy_service.ApiKeyUsageReservationData | None, + request_id: str, + client_ip: str | None = None, + ) -> tuple[proxy_service._WebSocketRequestState, str]: + del api_key, api_key_reservation, request_id, client_ip + prepared_payloads.append(prepared_payload) + return request_state, '{"type":"response.create"}' + + session = proxy_service._HTTPBridgeSession( + key=proxy_service._HTTPBridgeSessionKey("session_header", "sid-anchor-owner", None), + headers={"x-codex-session-id": "sid-anchor-owner"}, + affinity=proxy_service._AffinityPolicy( + key="sid-anchor-owner", + kind=proxy_service.StickySessionKind.CODEX_SESSION, + ), + request_model="gpt-5.4", + account=cast(Any, SimpleNamespace(id=account_id, status=AccountStatus.ACTIVE)), + upstream=cast(UpstreamWebSocket, SimpleNamespace(close=AsyncMock())), + upstream_control=proxy_service._WebSocketUpstreamControl(), + pending_requests=deque(), + pending_lock=anyio.Lock(), + response_create_gate=asyncio.Semaphore(1), + queued_request_count=0, + last_used_at=1.0, + idle_ttl_seconds=120.0, + codex_session=True, + last_completed_response_id="resp_session_latest", + last_completed_response_account_id=anchor_owner_account_id, + last_completed_input_count=3, + last_completed_input_prefix_fingerprint=proxy_service._fingerprint_input_items(prefix_items), + ) + + 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_enabled=True, + http_responses_session_bridge_prompt_cache_idle_ttl_seconds=3600, + http_responses_session_bridge_gateway_safe_mode=False, + ) + ) + ), + ), + ) + monkeypatch.setattr(service._durable_bridge, "lookup_request_targets", AsyncMock(return_value=None)) + monkeypatch.setattr(proxy_service, "get_settings", lambda: _make_app_settings()) + monkeypatch.setattr(service, "_prepare_http_bridge_request", fake_prepare) + monkeypatch.setattr(service, "_get_or_create_http_bridge_session", AsyncMock(return_value=session)) + monkeypatch.setattr(service, "_submit_http_bridge_request", AsyncMock()) + monkeypatch.setattr(service, "_detach_http_bridge_request", AsyncMock()) + + async for _chunk in service._stream_via_http_bridge( + payload, + headers={"x-codex-session-id": "sid-anchor-owner"}, + codex_session_affinity=True, + propagate_http_errors=False, + openai_cache_affinity=False, + 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, + ): + pass + return prepared_payloads + + +@pytest.mark.asyncio +async def test_stream_via_http_bridge_injects_session_anchor_when_account_owns_it( + monkeypatch: pytest.MonkeyPatch, +) -> None: + # Serving account owns the anchor -> the compact anchor is injected as normal. + prepared = await _run_session_anchor_owner_stream(monkeypatch, account_id="acc-1", anchor_owner_account_id="acc-1") + # Injection re-prepares the payload, so the final (sent) request carries the anchor. + assert prepared[-1].previous_response_id == "resp_session_latest" + + +@pytest.mark.asyncio +async def test_stream_via_http_bridge_skips_session_anchor_after_cross_account_failover( + monkeypatch: pytest.MonkeyPatch, +) -> None: + # Anchor was created on acc-1 but the session now serves on acc-2 (failover). + # A previous_response_id is account-scoped upstream, so injecting it here would + # send an unresolvable anchor with the history trimmed away -> upstream never + # emits response.created -> the response-create gate wedges. It must be skipped + # and the full history resent instead. + prepared = await _run_session_anchor_owner_stream(monkeypatch, account_id="acc-2", anchor_owner_account_id="acc-1") + assert all(payload.previous_response_id != "resp_session_latest" for payload in prepared) + assert prepared[-1].input == [ + {"role": "user", "content": [{"type": "input_text", "text": "a"}]}, + {"role": "assistant", "content": [{"type": "output_text", "text": "b"}]}, + {"role": "user", "content": [{"type": "input_text", "text": "c"}]}, + {"role": "user", "content": [{"type": "input_text", "text": "d"}]}, + ] + + @pytest.mark.asyncio async def test_stream_via_http_bridge_does_not_inject_durable_previous_response_anchor_for_full_resend_payload( monkeypatch: pytest.MonkeyPatch, @@ -8983,6 +9134,7 @@ def fake_prepare( ) assert creation.kwargs["preferred_account_id"] == "acc-1" assert session.last_completed_response_id == (None if preserves_full_resend else "resp_latest") + assert session.last_completed_response_account_id == (None if preserves_full_resend else "acc-1") if not preserves_full_resend: assert request_state.proxy_injected_previous_response_id is True assert request_state.fresh_upstream_request_is_retry_safe is False @@ -10520,7 +10672,16 @@ async def fake_get_or_create_http_bridge_session(*args: object, **kwargs: object ) raise owner_unavailable key = cast(proxy_service._HTTPBridgeSessionKey, args[0]) - return _make_bridge_session(key=key, key_value=key.affinity_key) + recovery_session = _make_bridge_session(key=key, key_value=key.affinity_key) + preferred_account_id = cast("str | None", kwargs.get("preferred_account_id")) + if preferred_account_id is not None: + # The real creation path pins the recovery session to the required + # continuity owner, so the durable anchor stays on its own account. + recovery_session.account = cast( + Any, + SimpleNamespace(id=preferred_account_id, status=AccountStatus.ACTIVE, plan_type="plus"), + ) + return recovery_session forwarded_payloads: list[proxy_service.ResponsesRequest] = [] @@ -12740,6 +12901,189 @@ async def test_stream_via_http_bridge_owner_forward_recovery_without_pending_sta assert prepared_inputs == [input_items, input_items] +async def _run_owner_forward_recovery_durable_anchor_stream( + monkeypatch: pytest.MonkeyPatch, + *, + durable_owner_account_id: str, + recovery_account_id: str, +) -> list[proxy_service.ResponsesRequest]: + """Drive owner-forward failure -> local rebind with a durable anchor available. + + The bootstrap rebind is explicitly allowed to bind the recovery session to an + account other than the durable owner, so this is the second site where a + proxy-injected ``previous_response_id`` can cross an account boundary. + Returns the payloads handed to each prepare call. + """ + + service = proxy_service.ProxyService(cast(Any, nullcontext())) + started_at = time.monotonic() + stored_items: list[dict[str, Any]] = [{"role": "user", "content": "first question"}] + # Full resend whose suffix carries no prior assistant output, so the + # account-neutral fresh-replay projection is unavailable and the durable + # anchor is the only continuity candidate the recovery path can reach for. + input_items: list[dict[str, Any]] = [*stored_items, {"role": "user", "content": "second question"}] + payload = proxy_service.ResponsesRequest.model_validate( + {"model": "gpt-5.4", "instructions": "hi", "input": input_items} + ) + prepared_payloads: list[proxy_service.ResponsesRequest] = [] + + def fake_prepare( + prepared_payload: proxy_service.ResponsesRequest, + _headers: dict[str, str] | Any, + *, + api_key: proxy_service.ApiKeyData | None, + api_key_reservation: proxy_service.ApiKeyUsageReservationData | None, + request_id: str, + client_ip: str | None = None, + ) -> tuple[proxy_service._WebSocketRequestState, str]: + del api_key, api_key_reservation, request_id, client_ip + prepared_payloads.append(prepared_payload) + state = proxy_service._WebSocketRequestState( + request_id=f"req-{len(prepared_payloads)}", + model="gpt-5.4", + service_tier=None, + reasoning_effort=None, + api_key_reservation=None, + started_at=started_at, + event_queue=asyncio.Queue(), + transport="http", + previous_response_id=prepared_payload.previous_response_id, + ) + return state, '{"type":"response.create"}' + + owner_forward = proxy_service._HTTPBridgeOwnerForward( + owner_instance="instance-b", + owner_endpoint="http://instance-b", + key=proxy_service._HTTPBridgeSessionKey("session_header", "sid-recover", None), + ) + recovery_session = _make_owner_forward_recovery_session() + recovery_session.account = cast(Any, SimpleNamespace(id=recovery_account_id, status=AccountStatus.ACTIVE)) + + async def fake_forward_http_bridge_request_to_owner(**kwargs: object): + del kwargs + raise ProxyResponseError( + 502, + proxy_service.openai_error("bridge_owner_unreachable", "owner instance unreachable"), + ) + yield "" + + async def fake_submit_http_bridge_request( + _session: proxy_service._HTTPBridgeSession, + *, + request_state: proxy_service._WebSocketRequestState, + text_data: str, + queue_limit: int, + ) -> None: + del _session, text_data, queue_limit + event_queue = request_state.event_queue + assert event_queue is not None + await event_queue.put('data: {"type":"response.completed"}\n\n') + await event_queue.put(None) + + 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=proxy_service.DurableBridgeLookup( + session_id="sess-recover", + canonical_kind="session_header", + canonical_key="sid-recover", + api_key_scope="__anonymous__", + account_id=durable_owner_account_id, + owner_instance_id="instance-b", + owner_epoch=1, + lease_expires_at=datetime.now(timezone.utc) + timedelta(seconds=60), + state=HttpBridgeSessionState.ACTIVE, + latest_turn_state=None, + latest_response_id="resp_durable_owner_1", + latest_input_item_count=len(stored_items), + latest_input_full_fingerprint=proxy_service._fingerprint_input_items( + cast(list[proxy_service.JsonValue], stored_items) + ), + ) + ), + ) + monkeypatch.setattr(service, "_http_bridge_can_forward_to_active_owner", AsyncMock(return_value=True)) + monkeypatch.setattr(service, "_http_bridge_has_live_local_session", AsyncMock(return_value=False)) + monkeypatch.setattr(service, "_prepare_http_bridge_request", fake_prepare) + monkeypatch.setattr( + service, + "_get_or_create_http_bridge_session", + AsyncMock(side_effect=[owner_forward, recovery_session]), + ) + monkeypatch.setattr(service, "_forward_http_bridge_request_to_owner", fake_forward_http_bridge_request_to_owner) + monkeypatch.setattr(service, "_submit_http_bridge_request", fake_submit_http_bridge_request) + monkeypatch.setattr(service, "_detach_http_bridge_request", AsyncMock()) + + async for _chunk in service._stream_via_http_bridge( + payload, + headers={"x-codex-session-id": "sid-recover"}, + codex_session_affinity=True, + propagate_http_errors=False, + openai_cache_affinity=False, + api_key=None, + api_key_reservation=None, + suppress_text_done_events=False, + idle_ttl_seconds=120.0, + codex_idle_ttl_seconds=900.0, + max_sessions=8, + queue_limit=4, + ): + pass + return prepared_payloads + + +@pytest.mark.asyncio +async def test_stream_via_http_bridge_owner_forward_recovery_injects_durable_anchor_on_owner_account( + monkeypatch: pytest.MonkeyPatch, +) -> None: + # The rebind stayed on the durable owner account, so the compact anchor is + # still the cheapest correct continuity and must be injected as before. + prepared = await _run_owner_forward_recovery_durable_anchor_stream( + monkeypatch, + durable_owner_account_id="acc-1", + recovery_account_id="acc-1", + ) + assert prepared[-1].previous_response_id == "resp_durable_owner_1" + + +@pytest.mark.asyncio +async def test_stream_via_http_bridge_owner_forward_recovery_skips_cross_account_durable_anchor( + monkeypatch: pytest.MonkeyPatch, +) -> None: + # The rebind landed on another account. Replaying the durable owner's anchor + # there sends a previous_response_id upstream cannot resolve with the history + # trimmed away, so the recovery request must keep the full input instead. + prepared = await _run_owner_forward_recovery_durable_anchor_stream( + monkeypatch, + durable_owner_account_id="acc-1", + recovery_account_id="acc-2", + ) + assert all(prepared_payload.previous_response_id is None for prepared_payload in prepared) + assert prepared[-1].input == [ + {"role": "user", "content": "first question"}, + {"role": "user", "content": "second question"}, + ] + + @pytest.mark.asyncio async def test_stream_via_http_bridge_local_previous_response_rebind_fails_existing_pending_requests( monkeypatch: pytest.MonkeyPatch,