diff --git a/app/modules/proxy/_service/http_bridge/streaming.py b/app/modules/proxy/_service/http_bridge/streaming.py index 1fce2706ea..3f54d8558f 100644 --- a/app/modules/proxy/_service/http_bridge/streaming.py +++ b/app/modules/proxy/_service/http_bridge/streaming.py @@ -220,6 +220,155 @@ _RESPONSE_CREATE_GATE_RETRY_SLEEP_SECONDS = 10.0 +class _VerifiedDurableFullResend: + """Immutable proof that one payload contains a durable turn's complete context.""" + + _durable_session_id: str + _full_input_fingerprint: str + _latest_response_id: str + _owner_account_id: str + _pending_tool_calls: tuple[tuple[str, str], ...] | None + _stored_input_fingerprint: str + _stored_input_item_count: int + __slots__ = ( + "_durable_session_id", + "_full_input_fingerprint", + "_latest_response_id", + "_owner_account_id", + "_pending_tool_calls", + "_stored_input_fingerprint", + "_stored_input_item_count", + ) + __construction_token = object() + + def __init__( + self, + *, + _token: object, + durable_session_id: str, + owner_account_id: str, + latest_response_id: str, + stored_input_item_count: int, + stored_input_fingerprint: str, + full_input_fingerprint: str, + pending_tool_calls: tuple[tuple[str, str], ...] | None, + ) -> None: + if _token is not self.__construction_token: + raise TypeError("verified durable full resend proofs are created only by the verifier") + object.__setattr__(self, "_durable_session_id", durable_session_id) + object.__setattr__(self, "_owner_account_id", owner_account_id) + object.__setattr__(self, "_latest_response_id", latest_response_id) + object.__setattr__(self, "_stored_input_item_count", stored_input_item_count) + object.__setattr__(self, "_stored_input_fingerprint", stored_input_fingerprint) + object.__setattr__(self, "_full_input_fingerprint", full_input_fingerprint) + object.__setattr__(self, "_pending_tool_calls", pending_tool_calls) + + def __setattr__(self, _name: str, _value: object) -> None: + raise AttributeError("verified durable full resend proofs are immutable") + + def __copy__(self) -> "_VerifiedDurableFullResend": + return self + + def __deepcopy__(self, _memo: dict[int, object]) -> "_VerifiedDurableFullResend": + return self + + def __reduce_ex__(self, _protocol: object) -> str | tuple[Any, ...]: + raise TypeError("verified durable full resend proofs cannot be serialized") + + @property + def stored_input_item_count(self) -> int: + return self._stored_input_item_count + + def matches( + self, + payload: ResponsesRequest, + durable_lookup: DurableBridgeLookup | None, + ) -> bool: + input_items = payload.input + return ( + isinstance(input_items, list) + and durable_lookup is not None + and durable_lookup.session_id == self._durable_session_id + and durable_lookup.account_id == self._owner_account_id + and durable_lookup.latest_response_id == self._latest_response_id + and durable_lookup.latest_input_item_count == self._stored_input_item_count + and durable_lookup.latest_input_full_fingerprint == self._stored_input_fingerprint + and _pending_tool_calls_identity(durable_lookup.latest_pending_tool_calls) == self._pending_tool_calls + and _fingerprint_input_items(cast(list[JsonValue], input_items)) == self._full_input_fingerprint + ) + + @classmethod + def _verify( + cls, + payload: ResponsesRequest, + durable_lookup: DurableBridgeLookup, + ) -> "_VerifiedDurableFullResend | None": + owner_account_id = durable_lookup.account_id + latest_response_id = durable_lookup.latest_response_id + stored_count = durable_lookup.latest_input_item_count + stored_fingerprint = durable_lookup.latest_input_full_fingerprint + if ( + owner_account_id is None + or latest_response_id is None + or stored_count is None + or stored_fingerprint is None + or not _http_bridge_payload_looks_like_full_resend(payload) + or not isinstance(payload.input, list) + or not _input_prefix_matches_stored_context( + payload.input, + stored_count=stored_count, + stored_fingerprint=stored_fingerprint, + ) + ): + return None + input_items = cast(list[JsonValue], payload.input) + replay_projection = project_responses_input_for_account_neutral_fresh_replay( + input_items, + stored_count=stored_count, + ) + pending_tool_calls = durable_lookup.latest_pending_tool_calls + if replay_projection is None: + return None + safe_fresh_context = responses_input_suffix_retains_prior_output( + replay_projection.input_items, + stored_count=replay_projection.stored_prefix_count, + ) or ( + pending_tool_calls is not None + and responses_input_suffix_matches_pending_tool_calls( + replay_projection.input_items, + stored_count=replay_projection.stored_prefix_count, + pending_tool_calls=pending_tool_calls, + ) + ) + if not safe_fresh_context: + return None + return cls( + _token=cls.__construction_token, + durable_session_id=durable_lookup.session_id, + owner_account_id=owner_account_id, + latest_response_id=latest_response_id, + stored_input_item_count=stored_count, + stored_input_fingerprint=stored_fingerprint, + full_input_fingerprint=_fingerprint_input_items(input_items), + pending_tool_calls=_pending_tool_calls_identity(pending_tool_calls), + ) + + +def _pending_tool_calls_identity( + pending_tool_calls: Mapping[str, str] | None, +) -> tuple[tuple[str, str], ...] | None: + return None if pending_tool_calls is None else tuple(sorted(pending_tool_calls.items())) + + +def _verify_durable_full_resend( + payload: ResponsesRequest, + durable_lookup: DurableBridgeLookup | None, +) -> _VerifiedDurableFullResend | None: + if durable_lookup is None or durable_lookup.account_id is None or durable_lookup.latest_response_id is None: + return None + return _VerifiedDurableFullResend._verify(payload, durable_lookup) + + def _http_bridge_payload_is_account_neutral_fresh_replay(payload: ResponsesRequest) -> bool: return responses_payload_is_account_neutral_fresh_replay(payload.to_payload()) @@ -907,6 +1056,8 @@ def prepare_bridge_request( durable_full_resend_is_account_neutral: bool | None = None durable_full_resend_has_safe_fresh_context = False durable_full_resend_retains_prior_output = False + durable_full_resend_proof = _verify_durable_full_resend(payload, durable_lookup) + durable_full_resend_fresh_bridge_proof: _VerifiedDurableFullResend | None = None force_local_recovery_creation = False payload_looks_like_full_resend = _http_bridge_payload_looks_like_full_resend(payload) @@ -1020,17 +1171,36 @@ def classify_durable_full_resend( and payload_looks_like_full_resend and durable_full_resend_has_safe_fresh_context ): - # The client already supplied a complete fresh request. Adding - # a durable anchor here can strand it on the new WebSocket. - _log_http_bridge_event( - "fresh_reattach_full_resend_preserved", - bridge_session_key, - account_id=durable_lookup.account_id, - model=payload.model, - detail="outcome=client_unanchored_full_resend", - cache_key_family=bridge_session_key.affinity_kind, - model_class=_extract_model_class(payload.model) if payload.model else None, - ) + if durable_full_resend_proof is not None and durable_full_resend_proof.matches(payload, durable_lookup): + durable_full_resend_fresh_bridge_proof = durable_full_resend_proof + # The client already supplied a proved complete fresh + # request. Adding a durable anchor here can strand it on + # the new WebSocket. + _log_http_bridge_event( + "fresh_reattach_full_resend_preserved", + bridge_session_key, + account_id=durable_lookup.account_id, + model=payload.model, + detail="outcome=client_unanchored_full_resend", + cache_key_family=bridge_session_key.affinity_kind, + model_class=_extract_model_class(payload.model) if payload.model else None, + ) + else: + effective_payload = payload.model_copy( + update={"previous_response_id": durable_lookup.latest_response_id} + ) + proxy_injected_previous_response_id = True + _fresh_request_state, fresh_upstream_request_text = prepare_bridge_request(payload) + del _fresh_request_state + _log_http_bridge_event( + "fresh_reattach_anchor_injected", + bridge_session_key, + account_id=None, + model=payload.model, + detail=f"response_id={durable_lookup.latest_response_id}", + cache_key_family=bridge_session_key.affinity_kind, + model_class=_extract_model_class(payload.model) if payload.model else None, + ) elif fresh_reattach_can_use_durable_anchor: effective_payload = payload.model_copy( update={"previous_response_id": durable_lookup.latest_response_id} @@ -1055,6 +1225,30 @@ def classify_durable_full_resend( affinity = _AffinityPolicy() incoming_turn_state_header = None session_header_fallback_key = None + owner_bound_full_resend_ignores_broad_session = ( + not forwarded_request + and durable_full_resend_fresh_bridge_proof is not None + and durable_full_resend_fresh_bridge_proof.matches(payload, durable_lookup) + and affinity.codex_session_source == "session_header" + ) + if owner_bound_full_resend_ignores_broad_session: + # The durable owner remains required through request_state below. + # Remove only the broad client alias that can resolve a stale raw + # compatibility row; keep CODEX_SESSION semantics so the new + # bridge can anchor later incremental turns to its fresh response. + affinity = _AffinityPolicy(kind=StickySessionKind.CODEX_SESSION) + incoming_session_header = None + session_header_fallback_key = None + _log_http_bridge_event( + "fresh_reattach_broad_session_owner_ignored", + bridge_session_key, + account_id=durable_lookup.account_id if durable_lookup is not None else None, + model=payload.model, + detail=(f"stored_items={durable_full_resend_fresh_bridge_proof.stored_input_item_count}"), + cache_key_family=bridge_session_key.affinity_kind, + model_class=_extract_model_class(payload.model) if payload.model else None, + owner_check_applied=True, + ) if effective_payload.previous_response_id is not None and isinstance(effective_payload.input, list): previous_response_input_items = cast(list[JsonValue], effective_payload.input) trimmed_input_items = _trim_http_bridge_previous_response_input_items(previous_response_input_items) @@ -1179,7 +1373,9 @@ def classify_durable_full_resend( settings = _service_get_settings() request_deadline = request_state.started_at + _http_bridge_request_budget_seconds(settings) session_creation_headers = ( - without_http_bridge_session_affinity_headers(headers) if account_neutral_recovery else dict(headers) + without_http_bridge_session_affinity_headers(headers) + if account_neutral_recovery or owner_bound_full_resend_ignores_broad_session + else dict(headers) ) fresh_replay_excluded_account_ids: set[str] = set() unanchored_fork_spill_attempted = False diff --git a/openspec/changes/reconcile-durable-full-resend-owner/.openspec.yaml b/openspec/changes/reconcile-durable-full-resend-owner/.openspec.yaml new file mode 100644 index 0000000000..3a038210f2 --- /dev/null +++ b/openspec/changes/reconcile-durable-full-resend-owner/.openspec.yaml @@ -0,0 +1,2 @@ +schema: spec-driven +created: 2026-07-25 diff --git a/openspec/changes/reconcile-durable-full-resend-owner/design.md b/openspec/changes/reconcile-durable-full-resend-owner/design.md new file mode 100644 index 0000000000..1848a5b687 --- /dev/null +++ b/openspec/changes/reconcile-durable-full-resend-owner/design.md @@ -0,0 +1,98 @@ +## Context + +PR #1486 makes a fingerprint-verified complete resend, including an exact +response-bound pending-tool-call settlement, start a fresh upstream bridge +without the stale durable `previous_response_id`. Account selection still +receives the original session-header affinity, though. During rolling upgrades, +a raw legacy `CODEX_SESSION` row may represent older hard turn-state ownership +on account A while the more specific durable bridge row owns this task on +account B. The load balancer correctly refuses to choose between those sources, +but the resulting retryable error cannot converge because neither persisted row +changes. + +## Goals / Non-Goals + +**Goals:** + +- Let a verified complete resend establish a fresh bridge on its durable owner. +- Remove only the broad session-header source that causes the deterministic + selection conflict. +- Keep subsequent incremental continuity on the newly established bridge. +- Make the eligibility proof immutable, request-bound, and unavailable through + ordinary construction or hydration. + +**Non-Goals:** + +- Prefer one conflicting turn-state, previous-response, file, or other specific + owner over another. +- Add or widen an account-movement path. Existing separately proved + account-neutral replay after a genuine owner-unavailable result remains + unchanged. +- Delete or rebind the broad legacy row, which may still own sibling work. +- Retry anything after upstream dispatch may have started. + +## Decisions + +### 1. Bind the bypass to a sealed local proof + +The verifier requires a durable owner, latest response ID, positive stored input +count, full fingerprint, exact raw-prefix match, and either retained prior +assistant output followed by fresh input or an exact call/output settlement of +the response-bound pending-tool-call manifest. The proof records the durable +session, owner, response, stored metadata, pending-tool-call manifest identity, +and full input fingerprint. Its normal constructor is sealed inside the +verifier closure, its fields are immutable, and serialization is rejected. +Before use, it is matched again against the current payload and durable lookup +so mutation or state substitution invalidates the proof. + +The proof is request-local. It is never accepted from a caller, serialized, +persisted, cloned into another request, or hydrated from the database. + +### 2. Remove only broad legacy selection provenance + +When the proved full resend is about to create a fresh owner-bound bridge from a +session header, the service removes downstream session and turn aliases from +the new upstream connection and replaces selection affinity with a +`CODEX_SESSION` policy that has no client key or legacy source. The durable +canonical bridge key and durable owner account remain unchanged, so selection +cannot move to another account. + +The broad sticky row is left intact for sibling traffic. Specific durable +turn-state, previous-response, and file-owner checks occur before this step and +remain hard conflicts. + +This reconciliation does not itself rebind the request to another account. +Existing account-neutral full-resend recovery after a genuine +owner-unavailable result remains separately gated by its own projection and +replay-safety checks. + +### 3. Preserve Codex bridge semantics after creation + +The selection policy retains `CODEX_SESSION` kind even though it drops the +stale client key. The created session therefore remains a Codex continuity +session and can inject the response ID established by the successful fresh +request for later incremental turns. + +## Risks / Trade-offs + +- The new upstream connection no longer receives the stale session header on + this one recovery path. The durable canonical key still owns internal routing, + and the complete request supplies the upstream context. +- Python cannot prevent hostile reflection through `object.__new__`, but + ordinary construction, mutation, copying with altered fields, and + serialization are closed; every use also revalidates the payload and durable + identity. +- An incomplete resend may still surface a continuity conflict. It remains + fail-closed because dropping either owner would risk context or account-bound + state. + +## Example + +Durable session `S` records owner B, response `resp_old`, two stored input items, +their fingerprint, and any pending tool calls bound to that response. A raw +legacy sticky row for the shared session header still points at A. The client +resends the two stored items plus either retained completed assistant output and +a new user message or the exact pending call/output settlement. codex-lb proves +the complete resend, opens the fresh bridge on B without `resp_old`, and omits +the stale broad header from selection and the upstream handshake. The raw row +remains on A for unrelated sibling traffic. diff --git a/openspec/changes/reconcile-durable-full-resend-owner/proposal.md b/openspec/changes/reconcile-durable-full-resend-owner/proposal.md new file mode 100644 index 0000000000..0f2db3b769 --- /dev/null +++ b/openspec/changes/reconcile-durable-full-resend-owner/proposal.md @@ -0,0 +1,41 @@ +## Why + +The fresh durable full-resend path can still fail before upstream dispatch when +a broad legacy session-header sticky row points at a different account than the +durable bridge owner. codex-lb returns `continuity_owner_conflict` as a +retryable 503, while the client repeats the same request and the two persisted +owners remain unchanged. This creates a deterministic retry loop even though +the request already contains fingerprint-verified complete context and can +safely start a fresh bridge on the durable owner. + +## What Changes + +- Represent complete durable full-resend eligibility with an immutable, + request-bound internal proof created only by the count, fingerprint, and + retained-output or response-bound pending-tool-call checks. +- For that proved fresh reattach only, stop consulting and forwarding the broad + legacy session-header alias while retaining the durable canonical key and + owner account as hard constraints. +- Preserve normal Codex session behavior on the replacement bridge so later + incremental turns can use its newly established response anchor. +- Keep incomplete resends, conflicting specific aliases, and file-owner + conflicts fail-closed; do not add or widen an account-movement path. + +## Capabilities + +### New Capabilities + +None. + +### Modified Capabilities + +- `responses-api-compat`: Reconcile a stale broad session mapping during a + verified owner-bound fresh reattach without moving accounts. + +## Impact + +- Affected code: HTTP bridge durable full-resend verification and fresh session + affinity preparation. +- Affected surface: hard Codex session reattach after the live upstream bridge + is gone and a legacy raw session row disagrees with the durable owner. +- No new cross-account replay, schema, setting, dependency, or post-send retry. diff --git a/openspec/changes/reconcile-durable-full-resend-owner/specs/responses-api-compat/spec.md b/openspec/changes/reconcile-durable-full-resend-owner/specs/responses-api-compat/spec.md new file mode 100644 index 0000000000..206c1fbe2a --- /dev/null +++ b/openspec/changes/reconcile-durable-full-resend-owner/specs/responses-api-compat/spec.md @@ -0,0 +1,76 @@ +## MODIFIED Requirements + +### Requirement: HTTP bridge MUST support fresh upstream reattach from durable continuity + +When a bridged HTTP request arrives for a valid hard continuity key but no live +local session or active remote owner remains, the service MUST retain the +durable owner account for routing. If the client supplies a full resend whose +stored prefix matches the durable input count and fingerprint and whose suffix +either proves that prior assistant output is retained before the new input or +exactly settles the response-bound pending-tool-call manifest, the service MUST +submit that complete payload on the fresh upstream bridge without injecting the +durable `previous_response_id`. Otherwise, when the request needs the durable +response anchor to represent prior context, the service MUST preserve existing +durable-anchor reattach behavior. The service MUST NOT replay a request after +uncertain upstream acceptance. + +Complete full-resend eligibility MUST be represented by an immutable +request-local proof that binds the current payload fingerprint to the durable +session ID, owner account, latest response ID, stored count, stored fingerprint, +and pending-tool-call manifest identity. The proof MUST be created only by the +count, fingerprint, and retained-output or response-bound pending-tool-call +verifier; it MUST NOT be accepted from a caller, persisted, deserialized, or +ordinarily constructed, and mutation or durable-state substitution MUST +invalidate it. + +When that proof authorizes a fresh owner-bound bridge and the incoming affinity +comes from a broad session header, the service MUST omit the downstream +session/turn aliases from the fresh upstream connection and MUST NOT consult +the broad legacy sticky row during account selection. It MUST retain the +durable canonical bridge key, require the durable owner account, preserve Codex +session behavior for subsequent turns, and leave the broad sticky row +unchanged. Conflicting specific turn-state, previous-response, bridge, or file +owners MUST still fail closed. This broad-alias reconciliation MUST NOT itself +rebind the request to another account; existing account-neutral full-resend +recovery after a genuine owner-unavailable result remains governed by its +separate replay-safety requirements. + +#### Scenario: stale broad session owner does not loop a verified full resend + +- **GIVEN** a hard durable session is owned by account B and records a latest + response ID, positive input count, and full fingerprint +- **AND** no live local bridge or active remote owner remains +- **AND** a broad legacy session-header sticky row points at account A +- **WHEN** the client sends a full resend whose stored prefix matches both + durable values and whose suffix retains completed prior output before fresh + input or exactly settles the response-bound pending-tool-call manifest +- **THEN** the service opens the fresh bridge on account B +- **AND** it submits the complete payload without `previous_response_id` +- **AND** it does not consult or rewrite the broad legacy row +- **AND** it omits downstream session and turn aliases from the fresh upstream + connection + +#### Scenario: recovered bridge keeps incremental continuity + +- **GIVEN** a verified full resend established a fresh owner-bound bridge +- **WHEN** that bridge completes and a later incremental turn arrives +- **THEN** the later turn remains on the same account +- **AND** the bridge may use the newly established response anchor + +#### Scenario: incomplete resend cannot bypass a broad owner conflict + +- **GIVEN** a durable owner conflicts with a broad legacy session owner +- **WHEN** the input prefix does not match, retained prior output is absent, the + payload or durable identity changes after verification, or the request is + incremental +- **THEN** no full-resend proof authorizes reconciliation +- **AND** the request retains existing anchor and fail-closed owner behavior + +#### Scenario: specific owner conflict remains fail-closed + +- **GIVEN** a verified full resend also contains a turn-state, + previous-response, bridge, or file owner that conflicts with the durable + owner +- **WHEN** continuity is resolved +- **THEN** the request fails with `continuity_owner_conflict` +- **AND** the broad-session reconciliation does not choose either account diff --git a/openspec/changes/reconcile-durable-full-resend-owner/tasks.md b/openspec/changes/reconcile-durable-full-resend-owner/tasks.md new file mode 100644 index 0000000000..2e3383e223 --- /dev/null +++ b/openspec/changes/reconcile-durable-full-resend-owner/tasks.md @@ -0,0 +1,20 @@ +## 1. Specification + +- [x] 1.1 Define owner-bound reconciliation for verified durable full resends. +- [x] 1.2 Define proof integrity and preserved fail-closed boundaries. + +## 2. Implementation + +- [x] 2.1 Add a sealed, immutable, request-bound full-resend proof. +- [x] 2.2 Drop only broad session-header selection provenance and forwarded + aliases while preserving durable owner routing and Codex session behavior. + +## 3. Verification + +- [x] 3.1 Add proof construction, mutation, state-substitution, and + response-bound pending-tool-call regressions. +- [x] 3.2 Add bridge regressions for stale broad owner reconciliation, + incremental fail-closed behavior, same-account routing, and stripped + upstream aliases. +- [x] 3.3 Run focused bridge tests, Ruff, type checking, and strict OpenSpec + validation. diff --git a/tests/integration/test_http_responses_bridge.py b/tests/integration/test_http_responses_bridge.py index c2139f2e64..34cf81cad7 100644 --- a/tests/integration/test_http_responses_bridge.py +++ b/tests/integration/test_http_responses_bridge.py @@ -45,6 +45,7 @@ AccountSelection, CatalogOmissionQuotaAdmission, ) +from app.modules.proxy.sticky_repository import StickySessionsRepository from app.modules.usage.repository import AdditionalUsageRepository pytestmark = pytest.mark.integration @@ -7595,6 +7596,170 @@ async def fake_connect_responses_websocket( assert owner_miss["preferred_account_is_continuity_owner"] is True +@pytest.mark.asyncio +async def test_backend_responses_verified_full_resend_ignores_stale_broad_owner_on_durable_account( + async_client, + app_instance, + monkeypatch, +): + _install_bridge_settings(monkeypatch, enabled=True) + owner_account_id = await _import_account( + async_client, + "acc_backend_durable_full_resend_owner", + "backend-durable-full-resend-owner@example.com", + ) + stale_account_id = await _import_account( + async_client, + "acc_backend_durable_full_resend_stale", + "backend-durable-full-resend-stale@example.com", + ) + owner_account = await _get_account(owner_account_id) + stale_account = await _get_account(stale_account_id) + owner_chatgpt_account_id = cast(str, owner_account.chatgpt_account_id) + service = get_proxy_service_for_app(app_instance) + session_id = "backend-durable-full-resend-session" + historical_input: list[proxy_module.JsonValue] = [ + { + "role": "user", + "content": [{"type": "input_text", "text": "first question"}], + } + ] + claimed = await service._durable_bridge.claim_live_session( + session_key_kind="session_header", + session_key_value=session_id, + api_key_id=None, + instance_id="instance-a", + lease_ttl_seconds=60.0, + account_id=owner_account.id, + model="gpt-5.1", + service_tier=None, + latest_turn_state="http_turn_durable_full_resend", + latest_response_id="resp_durable_full_resend_previous", + allow_takeover=True, + ) + renewed = await service._durable_bridge.renew_live_session( + session_id=claimed.session_id, + api_key_id=None, + instance_id="instance-a", + owner_epoch=claimed.owner_epoch, + lease_ttl_seconds=60.0, + latest_turn_state="http_turn_durable_full_resend", + latest_response_id="resp_durable_full_resend_previous", + latest_input_item_count=len(historical_input), + latest_input_full_fingerprint=proxy_module._fingerprint_input_items(historical_input), + ) + assert renewed is not None + released = await service._durable_bridge.release_live_session( + session_id=claimed.session_id, + instance_id="instance-a", + owner_epoch=claimed.owner_epoch, + draining=False, + ) + assert released is not None + assert released.account_id == owner_account.id + + async with SessionLocal() as session: + await StickySessionsRepository(session).upsert( + session_id, + stale_account.id, + kind=proxy_module.StickySessionKind.CODEX_SESSION, + ) + + upstream = _FakeBridgeUpstreamWebSocket("resp_durable_full_resend") + connect_calls: list[tuple[dict[str, str], str]] = [] + + 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 + connect_calls.append((dict(headers), account_id_header)) + return 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) + + full_resend = [ + *historical_input, + { + "type": "message", + "role": "assistant", + "status": "completed", + "content": [{"type": "output_text", "text": "first answer"}], + }, + { + "role": "user", + "content": [{"type": "input_text", "text": "second question"}], + }, + ] + first_events = await _collect_sse_events( + async_client, + "/backend-api/codex/responses", + json_body={ + "model": "gpt-5.1", + "instructions": "Return exactly OK.", + "input": full_resend, + "stream": True, + }, + headers={"session_id": session_id, "x-request-trace": "keep-me"}, + ) + second_events = await _collect_sse_events( + async_client, + "/backend-api/codex/responses", + json_body={ + "model": "gpt-5.1", + "instructions": "Return exactly OK.", + "input": "third question", + "stream": True, + }, + headers={"session_id": session_id}, + ) + + _assert_created_text_delta_completed(first_events) + _assert_created_text_delta_completed(second_events) + assert connect_calls[0][1] == owner_chatgpt_account_id + assert len(connect_calls) == 1 + connect_headers = {key.lower(): value for key, value in connect_calls[0][0].items()} + assert connect_headers["x-request-trace"] == "keep-me" + assert ( + not { + "session_id", + "session-id", + "thread-id", + "x-codex-conversation-id", + "x-codex-session-id", + "x-codex-turn-state", + } + & connect_headers.keys() + ) + assert len(upstream.sent_text) == 2 + replay_payload = json.loads(upstream.sent_text[0]) + assert "previous_response_id" not in replay_payload + assert replay_payload["input"] == full_resend + bridge_key = proxy_module._HTTPBridgeSessionKey("session_header", session_id, None) + bridge_session = service._http_bridge_sessions[bridge_key] + assert bridge_session.account.id == owner_account.id + assert bridge_session.codex_session is True + assert bridge_session.affinity.kind == proxy_module.StickySessionKind.CODEX_SESSION + assert bridge_session.affinity.key is None + async with SessionLocal() as session: + assert ( + await StickySessionsRepository(session).get_account_id( + session_id, + kind=proxy_module.StickySessionKind.CODEX_SESSION, + ) + == stale_account.id + ) + + @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..37490ff867 100644 --- a/tests/unit/test_proxy_http_bridge.py +++ b/tests/unit/test_proxy_http_bridge.py @@ -1,9 +1,11 @@ from __future__ import annotations import asyncio +import copy import inspect import json import logging +import pickle import time from collections import deque from contextlib import nullcontext @@ -7486,6 +7488,7 @@ def fake_prepare( "_http_bridge_payload_is_account_neutral_fresh_replay", account_neutral_classifier, ) + get_or_create = AsyncMock(return_value=session) monkeypatch.setattr(service, "_prepare_http_bridge_request", fake_prepare) monkeypatch.setattr( service, @@ -7493,7 +7496,6 @@ def fake_prepare( AsyncMock(return_value=forwardable_owner), ) monkeypatch.setattr(service, "_resolve_websocket_previous_response_owner", AsyncMock(return_value="acc-1")) - get_or_create = AsyncMock(return_value=session) monkeypatch.setattr(service, "_get_or_create_http_bridge_session", get_or_create) monkeypatch.setattr(service, "_submit_http_bridge_request", AsyncMock()) monkeypatch.setattr(service, "_detach_http_bridge_request", AsyncMock()) @@ -7552,6 +7554,172 @@ def fake_prepare( assert request_state.proxy_injected_previous_response_id is True assert request_state.fresh_upstream_request_is_retry_safe is False account_neutral_classifier.assert_not_called() + create_call = get_or_create.await_args + assert create_call is not None + create_kwargs = create_call.kwargs + create_headers = {key.lower(): value for key, value in create_kwargs["headers"].items()} + create_affinity = cast(proxy_service._AffinityPolicy, create_kwargs["affinity"]) + if preserves_full_resend and not forwardable_owner: + assert "x-codex-session-id" not in create_headers + assert create_affinity.kind == proxy_service.StickySessionKind.CODEX_SESSION + assert create_affinity.key is None + assert create_affinity.codex_session_source is None + assert create_kwargs["session_header_fallback_key"] is None + assert create_kwargs["preferred_account_id"] == "acc-1" + assert create_kwargs["preferred_account_has_continuity_provenance"] is True + else: + assert create_headers["x-codex-session-id"] == "sid-123" + assert create_affinity.key == "sid-123" + assert create_affinity.codex_session_source == "session_header" + + +def test_verified_durable_full_resend_proof_is_sealed_immutable_and_request_bound() -> None: + stored_input_items: list[proxy_service.JsonValue] = [ + { + "type": "additional_tools", + "role": "developer", + "tools": [{"type": "custom", "name": "shell"}], + }, + {"role": "user", "content": "hello"}, + ] + full_input = [ + *stored_input_items, + {"role": "assistant", "content": "hello back"}, + {"role": "user", "content": "follow up"}, + ] + payload = proxy_service.ResponsesRequest.model_validate( + { + "model": "gpt-5.4", + "instructions": "hi", + "input": full_input, + } + ) + durable_lookup = proxy_service.DurableBridgeLookup( + session_id="sess-proof", + canonical_kind="session_header", + canonical_key="sid-proof", + api_key_scope="__anonymous__", + account_id="acc-proof", + owner_instance_id=None, + owner_epoch=3, + lease_expires_at=None, + state=HttpBridgeSessionState.CLOSED, + latest_turn_state="http_turn_proof", + latest_response_id="resp-proof", + latest_input_item_count=len(stored_input_items), + latest_input_full_fingerprint=proxy_service._fingerprint_input_items(stored_input_items), + model="gpt-5.4", + ) + + with pytest.raises(TypeError, match="created only by the verifier"): + http_bridge_streaming_module._VerifiedDurableFullResend( + _token=object(), + durable_session_id=durable_lookup.session_id, + owner_account_id=cast(str, durable_lookup.account_id), + latest_response_id=cast(str, durable_lookup.latest_response_id), + stored_input_item_count=len(stored_input_items), + stored_input_fingerprint=cast(str, durable_lookup.latest_input_full_fingerprint), + full_input_fingerprint=proxy_service._fingerprint_input_items(full_input), + pending_tool_calls=None, + ) + + proof = http_bridge_streaming_module._verify_durable_full_resend(payload, durable_lookup) + assert proof is not None + assert proof.matches(payload, durable_lookup) is True + assert copy.copy(proof) is proof + assert copy.deepcopy(proof) is proof + with pytest.raises(AttributeError, match="immutable"): + proof._owner_account_id = "acc-forged" # type: ignore[misc] + with pytest.raises(TypeError, match="cannot be serialized"): + pickle.dumps(proof) + + changed_payload = payload.model_copy( + update={ + "input": [ + *stored_input_items, + {"role": "assistant", "content": "different output"}, + {"role": "user", "content": "follow up"}, + ] + } + ) + assert proof.matches(changed_payload, durable_lookup) is False + substituted_durable_lookups = ( + replace(durable_lookup, session_id="sess-other"), + replace(durable_lookup, account_id="acc-other"), + replace(durable_lookup, latest_response_id="resp-other"), + replace(durable_lookup, latest_input_item_count=len(stored_input_items) + 1), + replace(durable_lookup, latest_input_full_fingerprint="fingerprint-other"), + replace(durable_lookup, latest_pending_tool_calls={"call-other": "function_call"}), + ) + assert all(proof.matches(payload, lookup) is False for lookup in substituted_durable_lookups) + + incomplete_payload = payload.model_copy( + update={"input": [*stored_input_items, {"role": "user", "content": "follow up"}]} + ) + assert http_bridge_streaming_module._verify_durable_full_resend(incomplete_payload, durable_lookup) is None + + +def test_verified_durable_full_resend_accepts_response_bound_pending_tool_calls() -> None: + stored_input_items: list[proxy_service.JsonValue] = [ + {"role": "user", "content": "look that up"}, + ] + full_input: list[proxy_service.JsonValue] = [ + *stored_input_items, + { + "type": "function_call", + "call_id": "call-1", + "name": "lookup", + "arguments": "{}", + }, + { + "type": "function_call_output", + "call_id": "call-1", + "output": "result", + }, + ] + payload = proxy_service.ResponsesRequest.model_validate( + { + "model": "gpt-5.4", + "instructions": "hi", + "input": full_input, + } + ) + durable_lookup = proxy_service.DurableBridgeLookup( + session_id="sess-tool-proof", + canonical_kind="session_header", + canonical_key="sid-tool-proof", + api_key_scope="__anonymous__", + account_id="acc-proof", + owner_instance_id=None, + owner_epoch=3, + lease_expires_at=None, + state=HttpBridgeSessionState.CLOSED, + latest_turn_state="http_turn_tool_proof", + latest_response_id="resp-tool-proof", + latest_input_item_count=len(stored_input_items), + latest_input_full_fingerprint=proxy_service._fingerprint_input_items(stored_input_items), + model="gpt-5.4", + latest_pending_tool_calls={"call-1": "function_call"}, + ) + + proof = http_bridge_streaming_module._verify_durable_full_resend(payload, durable_lookup) + + assert proof is not None + assert proof.matches(payload, durable_lookup) is True + assert ( + proof.matches( + payload, + replace(durable_lookup, latest_pending_tool_calls={"call-other": "function_call"}), + ) + is False + ) + assert ( + http_bridge_streaming_module._verify_durable_full_resend( + payload, + replace(durable_lookup, latest_pending_tool_calls=None), + ) + is None + ) @pytest.mark.asyncio