diff --git a/app/modules/proxy/_service/http_bridge/helpers.py b/app/modules/proxy/_service/http_bridge/helpers.py index 9501800a1..221b4f02a 100644 --- a/app/modules/proxy/_service/http_bridge/helpers.py +++ b/app/modules/proxy/_service/http_bridge/helpers.py @@ -1297,12 +1297,28 @@ def _http_bridge_session_reusable_for_lookup( require_preferred_account: bool, service_tier_supported: bool, allow_closed_admission_handoff: bool, + session_key_quarantined: bool, ) -> bool: + if session.quarantined and not session_key_quarantined: + # The quarantine registry TTL is the source of truth + # (``_http_bridge_session_key_quarantined``): pruning drops the + # registry entry without touching the per-session flag, so a live + # session that outlives its quarantine window must become reusable + # again instead of staying rejected forever. Reset the stale flag so + # session state agrees with the registry. + session.quarantined = False live_or_retained = _http_bridge_session_account_active(session) and ( not session.closed or (allow_closed_admission_handoff and _http_bridge_session_has_admission_waiter(session)) ) return ( live_or_retained + # A quarantined key has proven silent/wedged; a new request must take + # the fresh session path instead of re-attaching. The registry verdict + # is authoritative for the key in both directions: a freshly created + # replacement session (flag still False) under a still-quarantined key + # is not reusable either, so a concurrent full-resend cannot restore + # the suppressed durable anchor through session hydration. + and not session_key_quarantined and _http_bridge_session_allows_api_key(session, api_key) and _http_bridge_session_reusable_for_request( session=session, diff --git a/app/modules/proxy/_service/http_bridge/mixin.py b/app/modules/proxy/_service/http_bridge/mixin.py index b5d7b3389..ceec07ef4 100644 --- a/app/modules/proxy/_service/http_bridge/mixin.py +++ b/app/modules/proxy/_service/http_bridge/mixin.py @@ -125,6 +125,9 @@ from app.modules.proxy._service.http_bridge.owner_forwarding import _HTTPBridgeOwnerForwardingMixin from app.modules.proxy._service.http_bridge.protocol import _HTTPBridgeServiceProtocol from app.modules.proxy._service.http_bridge.proxy_failover import _HTTPBridgePreDispatchFailover +from app.modules.proxy._service.http_bridge.quarantine import ( + _http_bridge_session_key_quarantined, +) from app.modules.proxy._service.http_bridge.request_submit import _HTTPBridgeRequestSubmitMixin from app.modules.proxy._service.http_bridge.service_stubs import ( _await_cancelled_task, @@ -683,6 +686,7 @@ def bind_account_neutral_recovery_owner(session: _HTTPBridgeSession) -> None: require_preferred_account=require_preferred_account, service_tier_supported=_http_bridge_compatible(existing, request_model, request_service_tier), allow_closed_admission_handoff=retained_handoff, + session_key_quarantined=_http_bridge_session_key_quarantined(self, existing.key), ) fork_key = _http_bridge_parallel_fork_key( key=key, diff --git a/app/modules/proxy/_service/http_bridge/owner_forwarding.py b/app/modules/proxy/_service/http_bridge/owner_forwarding.py index 51cdbda98..2155d8ebe 100644 --- a/app/modules/proxy/_service/http_bridge/owner_forwarding.py +++ b/app/modules/proxy/_service/http_bridge/owner_forwarding.py @@ -68,6 +68,9 @@ _normalized_http_bridge_instance_ring, _sticky_key_from_turn_state_header, ) +from app.modules.proxy._service.http_bridge.quarantine import ( + _http_bridge_session_key_quarantined, +) from app.modules.proxy._service.http_bridge.service_stubs import ( _headers_with_authorization, _partial_output_proxy_error_event_block, @@ -218,6 +221,16 @@ async def _http_bridge_has_live_local_session( session = self._http_bridge_sessions.get(candidate_key) if session is None or session.closed or not _http_bridge_session_account_active(session): continue + if _http_bridge_session_key_quarantined(self, session.key): + # A session under a quarantined key (#1534) is rejected and + # detached at lookup time, so for durable-anchor selection + # it must count as absent: a delta-only payload then keeps + # the durable anchor on the fresh session instead of + # silently losing its prior context. The registry verdict + # is authoritative for the key — a freshly created + # replacement session (flag still False) under a + # still-quarantined key counts as absent too. + continue if _durable_recovery_supersedes_local_session(durable_lookup, session): _drop_superseded_local_recovery_aliases_locked( self, diff --git a/app/modules/proxy/_service/http_bridge/quarantine.py b/app/modules/proxy/_service/http_bridge/quarantine.py new file mode 100644 index 000000000..ff47ea8b9 --- /dev/null +++ b/app/modules/proxy/_service/http_bridge/quarantine.py @@ -0,0 +1,189 @@ +from __future__ import annotations + +import logging +import time +from dataclasses import dataclass +from typing import Any + +from app.modules.proxy._service.http_bridge.helpers import _log_http_bridge_event +from app.modules.proxy._service.support import ( + _REQUEST_TRANSPORT_HTTP, + _HTTPBridgeSession, + _HTTPBridgeSessionKey, + _WebSocketRequestState, +) +from app.modules.proxy.affinity import _extract_model_class + +logger = logging.getLogger("app.modules.proxy.service") + +# Quarantine is a bounded, in-memory, session-scoped (never account-scoped) +# marker for HTTP bridge session keys that have proven silent/wedged: a later +# request must not re-attach to them and must take the existing fresh +# session/no-anchor path instead (#1534). It complements — and never replaces +# — the in-flight recovery machinery: the eventless watchdog and bounded +# replay (#1394) recover the request that is currently stuck, the fenced +# durable-anchor clear (#1563) stops a *fully eventless* full-resend anchor +# from being re-injected, and the durable retry circuit backs off in-place +# retries. Quarantine covers what those leave open: the reattached stream +# that delivers response events but never gets ``response.created`` (the +# ``response_event_count == 0`` gates in the stale/eventless detection never +# trip on it), and the repeated-wedge case where consecutive eventless +# timeouts keep rebuilding the same reattach. +_HTTP_BRIDGE_QUARANTINE_TTL_SECONDS = 600.0 +_HTTP_BRIDGE_QUARANTINE_EVENTLESS_TIMEOUT_THRESHOLD = 2 +_HTTP_BRIDGE_QUARANTINE_MAX_ENTRIES = 1024 + +_HTTP_BRIDGE_QUARANTINE_WEDGED_REATTACH_REASON = "reattach_missing_response_created" +_HTTP_BRIDGE_QUARANTINE_REPEATED_EVENTLESS_REASON = "repeated_eventless_timeout" + + +@dataclass(slots=True) +class _HTTPBridgeQuarantineEntry: + quarantined_until: float = 0.0 + consecutive_eventless_timeouts: int = 0 + last_touched_monotonic: float = 0.0 + reason: str | None = None + + +def _http_bridge_quarantine_registry( + service: Any, +) -> dict[_HTTPBridgeSessionKey, _HTTPBridgeQuarantineEntry]: + registry = getattr(service, "_http_bridge_quarantined_keys", None) + if registry is None: + registry = {} + service._http_bridge_quarantined_keys = registry + return registry + + +def _prune_http_bridge_quarantine_registry( + registry: dict[_HTTPBridgeSessionKey, _HTTPBridgeQuarantineEntry], + now: float, +) -> None: + expiry = now - _HTTP_BRIDGE_QUARANTINE_TTL_SECONDS + for key, entry in list(registry.items()): + if entry.last_touched_monotonic <= expiry and entry.quarantined_until <= now: + registry.pop(key, None) + overflow = len(registry) - _HTTP_BRIDGE_QUARANTINE_MAX_ENTRIES + if overflow > 0: + for stale_key in sorted(registry, key=lambda candidate: registry[candidate].last_touched_monotonic)[:overflow]: + registry.pop(stale_key, None) + + +def _http_bridge_request_state_wedged_reattach(request_state: _WebSocketRequestState) -> bool: + """Identify the #1534 wedge shape on a request that is being failed/retired. + + A reattached stream (proxy-injected ``previous_response_id``) whose + ``response.create`` was sent and that observed upstream response events, + but whose ``response.created`` was never assigned. This is only evaluated + when the request is already being failed or its session retired — never + against a live owned turn — so legitimate long event gaps (for example + deferred-reasoning streams) can never trip it, and any request whose + ``response.created`` was observed (``response_id`` or created latency set) + is excluded by construction. + """ + return ( + getattr(request_state, "transport", None) == _REQUEST_TRANSPORT_HTTP + and not getattr(request_state, "skip_request_log", False) + and getattr(request_state, "proxy_injected_previous_response_id", False) + and getattr(request_state, "response_create_sent_at", None) is not None + and getattr(request_state, "response_id", None) is None + and getattr(request_state, "latency_response_created_ms", None) is None + and getattr(request_state, "response_event_count", 0) > 0 + ) + + +def _http_bridge_session_key_quarantined(service: Any, key: _HTTPBridgeSessionKey) -> bool: + registry = _http_bridge_quarantine_registry(service) + now = time.monotonic() + _prune_http_bridge_quarantine_registry(registry, now) + entry = registry.get(key) + return entry is not None and entry.quarantined_until > now + + +def _quarantine_http_bridge_session(service: Any, session: _HTTPBridgeSession, *, reason: str) -> None: + """Quarantine a bridge session that has proven silent/wedged. + + Session-scoped only: no account-health writes happen here, and the entry + is bounded by TTL, a registry size cap, and the healthy-completion clear. + """ + now = time.monotonic() + registry = _http_bridge_quarantine_registry(service) + entry = registry.setdefault(session.key, _HTTPBridgeQuarantineEntry()) + already_quarantined = entry.quarantined_until > now + entry.quarantined_until = max(entry.quarantined_until, now + _HTTP_BRIDGE_QUARANTINE_TTL_SECONDS) + entry.last_touched_monotonic = now + entry.reason = reason + _prune_http_bridge_quarantine_registry(registry, now) + session.quarantined = True + if already_quarantined: + return + _log_http_bridge_event( + "session_quarantined", + session.key, + account_id=session.account.id, + model=session.request_model, + detail=f"reason={reason}, ttl_seconds={_HTTP_BRIDGE_QUARANTINE_TTL_SECONDS:.0f}", + cache_key_family=session.key.affinity_kind, + model_class=_extract_model_class(session.request_model) if session.request_model else None, + ) + + +def _record_http_bridge_quarantine_wedged_pending( + service: Any, + session: _HTTPBridgeSession, + request_states: Any, +) -> bool: + """Quarantine the session when a failed/retired pending request proves the wedge shape.""" + if not any(_http_bridge_request_state_wedged_reattach(request_state) for request_state in request_states): + return False + _quarantine_http_bridge_session( + service, + session, + reason=_HTTP_BRIDGE_QUARANTINE_WEDGED_REATTACH_REASON, + ) + return True + + +def _record_http_bridge_quarantine_eventless_timeout(service: Any, session: _HTTPBridgeSession) -> None: + """Count a ``missing_response_created_timeout`` retire; quarantine on repeats. + + The first eventless timeout is left to the merged recovery machinery + (bounded pre-created retry, fenced durable-anchor clear). A second + consecutive one for the same session key proves that path is also + rebuilding a wedged attach, so later requests must stop re-attaching. + """ + now = time.monotonic() + registry = _http_bridge_quarantine_registry(service) + # Prune before touching the entry: a strike whose TTL already lapsed must + # not be resurrected into a "consecutive" second strike hours later. + _prune_http_bridge_quarantine_registry(registry, now) + entry = registry.setdefault(session.key, _HTTPBridgeQuarantineEntry()) + entry.consecutive_eventless_timeouts += 1 + entry.last_touched_monotonic = now + if entry.consecutive_eventless_timeouts < _HTTP_BRIDGE_QUARANTINE_EVENTLESS_TIMEOUT_THRESHOLD: + return + _quarantine_http_bridge_session( + service, + session, + reason=_HTTP_BRIDGE_QUARANTINE_REPEATED_EVENTLESS_REASON, + ) + + +def _clear_http_bridge_quarantine(service: Any, session: _HTTPBridgeSession) -> None: + """A completed response on this key disproves the wedge; drop all state.""" + registry = _http_bridge_quarantine_registry(service) + session.quarantined = False + entry = registry.pop(session.key, None) + if entry is None: + return + if entry.quarantined_until <= time.monotonic(): + return + _log_http_bridge_event( + "session_quarantine_cleared", + session.key, + account_id=session.account.id, + model=session.request_model, + detail=f"reason={entry.reason}", + cache_key_family=session.key.affinity_kind, + model_class=_extract_model_class(session.request_model) if session.request_model else None, + ) diff --git a/app/modules/proxy/_service/http_bridge/request_submit.py b/app/modules/proxy/_service/http_bridge/request_submit.py index c5901c28e..83dc8d56f 100644 --- a/app/modules/proxy/_service/http_bridge/request_submit.py +++ b/app/modules/proxy/_service/http_bridge/request_submit.py @@ -84,6 +84,9 @@ _register_http_bridge_turn_state_aliases_locked, _release_http_bridge_unanchored_handoff, ) +from app.modules.proxy._service.http_bridge.quarantine import ( + _record_http_bridge_quarantine_wedged_pending, +) from app.modules.proxy._service.http_bridge.service_stubs import ( _call_with_supported_optional_kwargs, _classify_upstream_close, @@ -1747,6 +1750,10 @@ async def _fail_stale_http_bridge_pending_requests( stale_requests.append(request_state) if not stale_requests: return + # A stale gate holder that streamed response events without ever + # receiving ``response.created`` proves the reattach wedge (#1534) + # even when the session itself survives with other active requests. + _record_http_bridge_quarantine_wedged_pending(self, session, stale_requests) if response_events_seen == 0: await self._record_http_bridge_retry_circuit_failure(session, detail=detail) await self._fail_pending_websocket_requests( @@ -1818,6 +1825,14 @@ async def _retire_stale_pending_http_bridge_session( retry_circuit_detail: str | None = None, response_events_seen: int | None = None, ) -> None: + async with session.pending_lock: + retired_request_states = list(session.pending_requests) + # Direct retirement (for example the all-stale stuck-gate path, where + # the wedged reattach is the only pending request) cancels the reader + # and fails the pendings without passing the partial-cleanup hook or + # the reader-failure funnel, so evaluate the wedge shape (#1534) here + # too; recording is idempotent for callers that already quarantined. + _record_http_bridge_quarantine_wedged_pending(self, session, retired_request_states) if response_events_seen is None or response_events_seen == 0: await self._record_http_bridge_retry_circuit_failure( session, diff --git a/app/modules/proxy/_service/http_bridge/streaming.py b/app/modules/proxy/_service/http_bridge/streaming.py index da677d8c3..17d83897e 100644 --- a/app/modules/proxy/_service/http_bridge/streaming.py +++ b/app/modules/proxy/_service/http_bridge/streaming.py @@ -101,6 +101,9 @@ _reserve_http_bridge_unanchored_handoff, _trim_http_bridge_previous_response_input_items, ) +from app.modules.proxy._service.http_bridge.quarantine import ( + _http_bridge_session_key_quarantined, +) from app.modules.proxy._service.http_bridge.service_stubs import ( _build_rewritten_stream_response_failed_event, _codex_keepalive_frame, @@ -1194,6 +1197,12 @@ async def release_unowned_bridge_lifecycle( 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) + # Set when the quarantine check below suppresses the durable-anchor + # injection for a full-resend payload; the session hydration and the + # session-level anchor injection further down must honor it so the + # dispatch genuinely goes unanchored instead of rebuilding the same + # wedged reattach through the session-state side door. + fresh_reattach_anchor_suppressed_quarantined = False def classify_durable_full_resend( lookup: DurableBridgeLookup, @@ -1411,6 +1420,31 @@ def classify_durable_full_resend( and durable_lookup.latest_response_id is not None and (not payload_looks_like_full_resend or durable_anchor_trimmable) ) + if payload_looks_like_full_resend and _http_bridge_session_key_quarantined(self, bridge_session_key): + # The previous attach on this key proved silent/wedged + # (#1534). The client's own payload already carries the full + # conversation, so send it unanchored on the fresh path + # instead of rebuilding the same reattach. Delta-only + # payloads keep the anchor: it is their only way to convey + # prior context (same boundary as the fenced anchor clear). + # Evaluated independently of the fresh-reattach eligibility + # above: even when that gate is already false (for example a + # conversation-scoped payload, a live alias session, or an + # active-owner forward that later falls back to a local + # rebind), the suppression must still reach the session + # hydration, session-level injection, and recovery injection + # paths below. + fresh_reattach_can_use_durable_anchor = False + fresh_reattach_anchor_suppressed_quarantined = True + _log_http_bridge_event( + "fresh_reattach_anchor_skipped_quarantined", + bridge_session_key, + account_id=durable_lookup.account_id, + 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, + ) if ( fresh_reattach_can_use_durable_anchor and payload_looks_like_full_resend @@ -2164,6 +2198,10 @@ def switch_to_account_neutral_replay() -> None: recovery_anchor_input_fingerprint: str | None = None if ( not owner_forward_fresh_replay + # The quarantine skip (#1534) applies to the local recovery + # rebind as well: re-injecting the suppressed anchor here + # would rebuild the same wedged reattach. + and not fresh_reattach_anchor_suppressed_quarantined and not durable_full_resend_has_safe_fresh_context and recovery_payload.previous_response_id is None and durable_lookup is not None @@ -2293,7 +2331,12 @@ def switch_to_account_neutral_replay() -> None: return session = session_or_forward if ( - not durable_full_resend_has_safe_fresh_context + # A quarantine-suppressed anchor (#1534) must not be rehydrated + # into the session either: doing so would let the session-level + # injection below re-add the exact anchor the quarantine skipped + # and trim the prefix, rebuilding the wedged reattach. + not fresh_reattach_anchor_suppressed_quarantined + and not durable_full_resend_has_safe_fresh_context and durable_full_resend_anchor_count is not None and durable_full_resend_anchor_fingerprint is not None and durable_lookup is not None @@ -2339,6 +2382,9 @@ def switch_to_account_neutral_replay() -> None: ) and (not _http_bridge_payload_looks_like_full_resend(effective_payload) or session_anchor_trimmable) if ( session.codex_session + # Honor the quarantine decision end to end: the durable anchor + # skipped above must not come back as a session-level injection. + and not fresh_reattach_anchor_suppressed_quarantined and not proxy_injected_previous_response_id and effective_payload.previous_response_id is None and session.last_completed_response_id is not None diff --git a/app/modules/proxy/_service/http_bridge/upstream_events.py b/app/modules/proxy/_service/http_bridge/upstream_events.py index e9fe0c660..0444b55df 100644 --- a/app/modules/proxy/_service/http_bridge/upstream_events.py +++ b/app/modules/proxy/_service/http_bridge/upstream_events.py @@ -57,6 +57,11 @@ _normalize_http_bridge_error_event, _record_http_bridge_stuck_retire, ) +from app.modules.proxy._service.http_bridge.quarantine import ( + _clear_http_bridge_quarantine, + _record_http_bridge_quarantine_eventless_timeout, + _record_http_bridge_quarantine_wedged_pending, +) from app.modules.proxy._service.http_bridge.service_stubs import ( _assign_websocket_response_id, _await_cancelled_task, @@ -677,6 +682,13 @@ async def _fail_http_bridge_reader_and_maybe_retire( (getattr(request_state, "response_event_count", 0) for request_state in session.pending_requests), default=0, ) + pending_request_states = list(session.pending_requests) + # The #1534 wedge shape: a reattached stream that streamed response + # events whose ``response.created`` was never assigned. The eventless + # watchdog and the durable-anchor clear both key on + # ``response_event_count == 0`` and never trip on it, so quarantine + # the session here so later requests stop re-attaching to it. + _record_http_bridge_quarantine_wedged_pending(self, session, pending_request_states) observed_close_code = ( upstream_close_code if upstream_close_code is not None else session.last_upstream_close_code ) @@ -920,6 +932,7 @@ async def _relay_http_bridge_upstream_messages( # generic reader-crash account penalty path. if expired_proxy_injected_anchor: await _clear_durable_http_bridge_response_anchor(self, session) + _record_http_bridge_quarantine_eventless_timeout(self, session) session.closed = True await self._fail_http_bridge_reader_and_maybe_retire( session, @@ -935,6 +948,10 @@ async def _relay_http_bridge_upstream_messages( # to persist the durable-anchor invalidation. if expired_proxy_injected_anchor: await _clear_durable_http_bridge_response_anchor(self, session) + # Count the eventless retire toward the repeated- + # wedge quarantine; the first strike still goes + # through the bounded pre-created recovery below. + _record_http_bridge_quarantine_eventless_timeout(self, session) _record_http_bridge_stuck_retire( reason=_HTTP_BRIDGE_MISSING_RESPONSE_CREATED_TIMEOUT_DETAIL, session=session, @@ -2035,6 +2052,7 @@ async def _process_parsed_http_bridge_upstream_event( and not terminal_request_state.skip_request_log ): await self._clear_http_bridge_retry_circuit(session) + _clear_http_bridge_quarantine(self, session) normalize_error_event = ( terminal_request_state is None or terminal_request_state.enforce_openai_sdk_contract diff --git a/app/modules/proxy/_service/support.py b/app/modules/proxy/_service/support.py index 7c26e6f81..134f10abc 100644 --- a/app/modules/proxy/_service/support.py +++ b/app/modules/proxy/_service/support.py @@ -1030,6 +1030,11 @@ class _HTTPBridgeSession: last_upstream_close_code: int | None = None last_upstream_close_generation: int = 0 closed: bool = False + # Set when the session proved silent/wedged (reattached stream with + # response events but no ``response.created``, or repeated eventless + # timeouts). A quarantined session must never be selected for reuse or + # re-attach; later requests take the fresh session/no-anchor path. + quarantined: bool = False # Set while a reader handoff is replacing the socket. Idle pruning must # retain the registered session during this short transition even though # ``closed`` is fail-closed for normal request reuse. diff --git a/openspec/changes/quarantine-silent-bridge-sessions/.openspec.yaml b/openspec/changes/quarantine-silent-bridge-sessions/.openspec.yaml new file mode 100644 index 000000000..84cfc1245 --- /dev/null +++ b/openspec/changes/quarantine-silent-bridge-sessions/.openspec.yaml @@ -0,0 +1,2 @@ +schema: spec-driven +created: 2026-08-06 diff --git a/openspec/changes/quarantine-silent-bridge-sessions/proposal.md b/openspec/changes/quarantine-silent-bridge-sessions/proposal.md new file mode 100644 index 000000000..242db57e7 --- /dev/null +++ b/openspec/changes/quarantine-silent-bridge-sessions/proposal.md @@ -0,0 +1,45 @@ +# Quarantine silent HTTP bridge sessions + +## Summary + +An HTTP bridge session that has proven silent/wedged must stop attracting new attach attempts. Two shapes prove it: a reattached stream (proxy-injected `previous_response_id`) that delivers upstream response events but never gets `response.created` assigned, and a session key that hits repeated eventless `missing_response_created_timeout` retires. Mark such a session key quarantined — bounded, in-memory, session-scoped, never account-scoped — so that later requests neither reuse the session nor rebuild the same durable-anchor reattach, and instead complete on the existing fresh session/no-anchor path that production evidence in [#1534](https://github.com/Soju06/codex-lb/issues/1534) shows already works. + +This is the maintainer-chosen direction from [#1405](https://github.com/Soju06/codex-lb/pull/1405) (quarantine, chosen over the account-health penalty alternative #1574), revived post-#1394/#1563 as a first-party takeover with the original review findings addressed. + +## Why + +The merged silent-upstream machinery recovers the request that is currently stuck, and one flavor of the durable state behind it: + +- `recover-codex-desktop-idle-bridge` / #1394 bound the eventless `response.create` wait, added the bounded pre-created retry, and back off repeated failures through the durable retry circuit. +- `invalidate-durable-bridge-anchor-after-stuck-timeout` / #1563 clears the durable anchor after a *fully eventless* timeout, for full-resend-shaped payloads. + +Both key on `response_event_count == 0`. The production wedge in #1534 is different: the reattached stream delivered response events whose `response.created` was never assigned. That shape never trips the eventless deadline (it disarms once events flow), never reaches the fenced anchor clear, and refuses in-place replay because model output was seen — so the request fails terminally, the durable anchor survives, and the next turn rebuilds the identical anchored reattach to the same wedged state. The client can only recover by starting a fresh session; the recoverable path exists but the running session can never reach it. + +The original #1405 attacked this with a 5s `response.created` timeout plus a raw-HTTP fallback window. The maintainer review found the timeout arms only while `response_event_count == 0` (so the observed production wedge never trips it) and that 5s false-positives retire healthy sessions. This change keeps #1405's core idea — a bounded per-key quarantine window — and re-triggers it on proof rather than on a racy timeout: the wedge shape is only ever evaluated when a request is already being failed or its session retired, never against a live owned turn, so deferred-reasoning streams with long legitimate event gaps (the P1 documented on #1580) cannot be quarantined; any request whose `response.created` was observed is excluded by construction. + +## What Changes + +- Add a bounded in-memory quarantine registry keyed by HTTP bridge session key (the same key the retry circuit uses). No new settings; fix-class, default-on, zero-config. +- Trigger 1 (immediate): when a pending request being failed or retired proves the wedge shape — HTTP transport, proxy-injected `previous_response_id`, `response.create` sent, upstream response events observed, `response.created` never assigned — quarantine the session key. Hooked at the reader failure/retire funnel and the stale-gate-holder cleanup. +- Trigger 2 (repeats): count consecutive eventless `missing_response_created_timeout` retires per session key; quarantine at the second. The first stays on the merged #1394 recovery path (bounded retry, #1563 anchor clear). +- Effect 1: a quarantined live/retained session is excluded from re-attach/session-reuse selection (`_http_bridge_session_reusable_for_lookup`), so a new request detaches it and creates a fresh session instead. +- Effect 2: while a key is quarantined, the fresh-reattach durable-anchor injection is skipped for full-resend-shaped payloads: the client's own payload already carries the whole conversation, so it goes upstream unanchored on the existing fresh path. Delta-only payloads keep the anchor — it is their only way to convey prior context — matching the scope boundary of `invalidate-durable-bridge-anchor-after-stuck-timeout`. +- Recovery/expiry (bounded, no leak): a completed response on the key clears the quarantine and its strike counter; otherwise entries expire after a 600-second TTL (aligned with the retry circuit's max backoff) and the registry is pruned and size-capped. In-memory only — no durable rows, no janitor involvement, and process restart clears it. +- Account-neutral: quarantine decisions never write account health and never move or exclude accounts; durable continuity ownership is preserved on the fresh path. +- Observability: low-cardinality `session_quarantined`, `session_quarantine_cleared`, and `fresh_reattach_anchor_skipped_quarantined` bridge events. + +## Capabilities + +### New Capabilities +- None. + +### Modified Capabilities +- `responses-api-compat`: An HTTP bridge session that proved silent/wedged (reattached stream with response events but no `response.created`, or repeated eventless timeouts) is quarantined for a bounded window: later requests do not reuse it and full-resend reattaches skip the durable anchor, taking the existing fresh path instead. + +## Non-Goals + +- No change to the eventless watchdog, its deadline, the bounded pre-created retry, or the durable retry circuit (#1394) — quarantine acts only on *later* requests, after those have run for the in-flight one. +- No change to the fenced durable-anchor clear (#1563); quarantine does not write the durable session row at all. +- No change for delta-only payloads' anchor injection: same boundary as #1563, their proxy-injected anchor is load-bearing and stays. A wedged delta-only client still benefits from Trigger/Effect 1 (no reuse of the wedged session) and from #1394's bounded failure, but its reattach anchor is not stripped. +- No account-health, routing, or account-selection changes. +- No new settings, no persistence, no dashboard surface. diff --git a/openspec/changes/quarantine-silent-bridge-sessions/specs/responses-api-compat/spec.md b/openspec/changes/quarantine-silent-bridge-sessions/specs/responses-api-compat/spec.md new file mode 100644 index 000000000..cc6c66892 --- /dev/null +++ b/openspec/changes/quarantine-silent-bridge-sessions/specs/responses-api-compat/spec.md @@ -0,0 +1,80 @@ +# responses-api-compat Delta + +## ADDED Requirements + +### Requirement: Silent HTTP bridge sessions are quarantined from re-attach and reuse + +When an HTTP bridge session proves silent/wedged, the proxy MUST quarantine its session key for a bounded window so later requests stop attaching to it. A session proves silent/wedged when either (a) a pending request being failed or retired carried a proxy-injected `previous_response_id`, had sent `response.create`, observed upstream response events, and never had `response.created` assigned, or (b) the session key hits two consecutive eventless `missing_response_created_timeout` retires. This holds for every path that fails or retires the request — partial stale-holder cleanup, the reader-failure funnel, and direct all-stale session retirement alike. The quarantine MUST be evaluated only when a request is already being failed or its session retired — never against a live owned turn — so a stream whose `response.created` was observed (including deferred-reasoning streams with long event gaps) MUST NOT be quarantined, and mere event silence during an owned live turn MUST NOT trigger quarantine by itself. + +While a session key is quarantined: an existing session under that key MUST NOT be selected for reuse (a new request detaches it and proceeds on a fresh session), and for durable-anchor selection a quarantined session that is still open MUST count as absent, exactly as if it were already gone. The quarantine registry verdict is authoritative for the key: any session under the key while the quarantine window is active — including a freshly created replacement whose own completion has not yet cleared the quarantine — is equally excluded from reuse and equally absent for anchor selection. A fresh reattach whose incoming payload already looks like a full conversation resend MUST NOT receive a proxy-injected durable anchor through any injection point — the fresh-reattach injection, session-state hydration of the durable anchor, or the session-level injection — so the dispatch goes upstream genuinely unanchored with the client's own untrimmed payload. A payload that does not look like a full resend (a genuine delta-only continuation) MUST still receive the durable anchor, because it has no other way to convey prior conversation state. + +Quarantine state MUST be bounded and self-recovering: it is in-memory and session-scoped, expires by TTL (a live session that outlives its quarantine window MUST become reusable again), is cleared when a response completes on the same session key, and MUST NOT write account health or alter account selection. + +#### Scenario: Reattach streams events but response.created is never assigned (#1534) + +- **GIVEN** a durable HTTP bridge session with a stored anchor whose fresh reattach injected a proxy-owned `previous_response_id` +- **AND** the reattached upstream stream delivers response events but `response.created` is never assigned +- **WHEN** the stream fails or the session is retired with that request still pending +- **THEN** the request fails terminally as before +- **AND** the session key is quarantined with reason `reattach_missing_response_created` + +#### Scenario: All-stale direct retirement still quarantines the key + +- **GIVEN** a wedged reattach (proxy-injected `previous_response_id`, `response.create` sent, response events observed, `response.created` never assigned) that is the ONLY stale pending request on its session +- **WHEN** the stuck-gate watchdog retires the session directly instead of failing the stale holder individually +- **THEN** the session key is quarantined with reason `reattach_missing_response_created` +- **AND** the next request takes the fresh no-anchor path instead of rebuilding the identical anchored reattach + +#### Scenario: Next request after the wedge completes on the fresh path + +- **GIVEN** a session key quarantined after a reattach that streamed events without `response.created` +- **WHEN** a later request arrives for the same key with a full-conversation-resend payload and no client `previous_response_id` +- **THEN** the proxy does not inject the durable anchor for that request +- **AND** the request is sent upstream unanchored with the client's own full payload +- **AND** the request can complete normally instead of rebuilding the identical wedged reattach + +#### Scenario: Suppressed anchor does not come back through session state + +- **GIVEN** a quarantined session key and a full-conversation-resend payload whose stored durable prefix is trimmable but whose fresh suffix does not retain the prior output +- **WHEN** the fresh-reattach durable-anchor injection is skipped because of the quarantine +- **THEN** the durable anchor is not rehydrated into the fresh session's completed-response state +- **AND** the session-level injection does not re-add the same anchor or trim the stored prefix +- **AND** the dispatch goes upstream genuinely unanchored with the client's untrimmed payload +- **AND** the suppression applies even when the fresh-reattach injection was already ineligible for other reasons (for example a conversation-scoped payload, a live alias session, or an active-owner forward that falls back to a local rebind) + +#### Scenario: Quarantined session is excluded from reuse selection + +- **GIVEN** a session marked quarantined that is still live or retained for admission handoff +- **WHEN** a new request looks up that session key +- **THEN** the session is not considered reusable +- **AND** the request proceeds on a fresh session instead +- **AND** a replacement session created under the same still-quarantined key is likewise not reusable until a completion or the TTL clears the quarantine + +#### Scenario: Repeated eventless timeouts quarantine the key + +- **GIVEN** a session key whose pending request already retired once with the eventless `missing_response_created_timeout` +- **WHEN** a subsequent attach on the same key retires with the same eventless timeout before any response completes on the key +- **THEN** the session key is quarantined with reason `repeated_eventless_timeout` +- **AND** the first timeout alone does not quarantine the key + +#### Scenario: Deferred-reasoning live turn is never quarantined + +- **GIVEN** an owned live turn whose `response.created` was observed and whose events flow with long gaps (deferred reasoning) +- **WHEN** its stream later fails or its session is retired +- **THEN** the session key is not quarantined +- **AND** later requests keep the existing reuse and anchor-injection behavior + +#### Scenario: Delta-only payloads keep their anchor while quarantined + +- **GIVEN** a quarantined session key — including one whose quarantined session is still open with other active requests +- **WHEN** a later request arrives whose payload does not look like a full conversation resend +- **THEN** the still-open quarantined session counts as absent for durable-anchor selection +- **AND** the durable anchor is still injected for that request, preserving the client's only way to convey prior context + +#### Scenario: Quarantine is bounded and self-clearing + +- **GIVEN** a quarantined session key +- **WHEN** a response completes on that session key, or the quarantine TTL elapses +- **THEN** the quarantine (and its eventless strike counter) is cleared +- **AND** a session that survived the quarantine window is reusable again instead of staying rejected forever +- **AND** no durable row, janitor work, or account-health write was involved at any point diff --git a/openspec/changes/quarantine-silent-bridge-sessions/tasks.md b/openspec/changes/quarantine-silent-bridge-sessions/tasks.md new file mode 100644 index 000000000..56209810f --- /dev/null +++ b/openspec/changes/quarantine-silent-bridge-sessions/tasks.md @@ -0,0 +1,34 @@ +## 1. Spec + +- [x] 1.1 Add the `responses-api-compat` delta requirement: silent/wedged bridge sessions (reattached stream with response events but no `response.created`, or two consecutive eventless-timeout retires) are quarantined — excluded from reuse and from full-resend anchor re-injection — with bounded, account-neutral, self-clearing state; deferred-reasoning live turns and delta-only anchors are explicitly protected. + +## 2. Implementation + +- [x] 2.1 Add `app/modules/proxy/_service/http_bridge/quarantine.py`: bounded in-memory registry keyed by `_HTTPBridgeSessionKey` (TTL 600s, size cap, prune-on-touch), the wedge-shape predicate, the two trigger recorders, the key-quarantined check, and the completion clear. No settings; module constants only. +- [x] 2.2 Trigger 1 hooks: evaluate the wedge shape over pending requests in `_fail_http_bridge_reader_and_maybe_retire` (upstream_events.py) and over the stale set in `_fail_stale_http_bridge_pending_requests` (request_submit.py). +- [x] 2.3 Trigger 2 hooks: record an eventless strike in both `missing_response_created_timeout` branches of the reader loop (force-retire and ordinary), quarantining at the second consecutive strike; the first stays on the merged #1394 recovery path. +- [x] 2.4 Effect 1: `_http_bridge_session_reusable_for_lookup` (helpers.py) rejects quarantined sessions; `_HTTPBridgeSession.quarantined` flag added in support.py. +- [x] 2.5 Effect 2: in streaming.py, skip `fresh_reattach_can_use_durable_anchor` for full-resend payloads when the key is quarantined, emitting `fresh_reattach_anchor_skipped_quarantined`; delta-only payloads keep the anchor (same boundary as `invalidate-durable-bridge-anchor-after-stuck-timeout`). +- [x] 2.6 Recovery: clear quarantine and strikes on `response.completed` alongside the retry-circuit clear (upstream_events.py); TTL expiry and size cap bound everything else. No durable rows and no account-health writes anywhere in the path. + +## 3. Coverage + +- [x] 3.1 Unit tests (`tests/unit/test_proxy_http_bridge.py`): wedge-shape predicate truth table (including created-assigned, created-latency, fully-eventless, non-injected, websocket, internal shapes), registry TTL expiry and size bound, two-strike eventless threshold with completion reset, reuse-gate exclusion, reader-failure trigger (positive and deferred-reasoning negative), stale-gate-holder trigger, and a reader-loop regression proving the real `missing_response_created_timeout` path records exactly one strike without quarantining. +- [x] 3.2 Integration regression (`tests/integration/test_http_responses_bridge.py`) modeling #1534: reattach injects the durable anchor, upstream streams reasoning deltas but never `response.created`, the turn fails and the key is quarantined; the next full-resend request is sent unanchored on a fresh path and completes, and the completed response clears the quarantine. Verified the test fails when quarantine is neutralized (third attempt rebuilds the identical wedged reattach). + +## 4. Review follow-ups (local codex review) + +- [x] 4.1 Quarantine the direct all-stale retirement path: `_retire_stale_pending_http_bridge_session` (request_submit.py) evaluates the wedge shape over its pending snapshot, closing the bypass where the wedged reattach is the only stale holder. Unit + integration regressions (stuck-gate direct retirement quarantines the key). +- [x] 4.2 Anchor selection treats quarantined live sessions as absent: `_http_bridge_has_live_local_session` (owner_forwarding.py) skips effectively-quarantined sessions so delta-only payloads keep the durable anchor instead of losing context on the fresh session. Unit regression. +- [x] 4.3 Carry the quarantine anchor-skip decision end to end (streaming.py): the suppressed durable anchor is neither rehydrated into session state nor re-added by the session-level (or owner-forward recovery) injection; the dispatch goes genuinely unanchored. Integration regression (trimmable prefix, unsafe suffix). +- [x] 4.4 Expire the per-session quarantine flag by TTL: `_http_bridge_session_reusable_for_lookup` (helpers.py) consults the registry verdict (`_http_bridge_session_key_quarantined`) as the source of truth and resets the stale flag, so a surviving session becomes reusable again after expiry. Unit regression. +- [x] 4.5 Registry verdict is authoritative for the key in both directions (second-round review): a freshly created replacement session (flag still False) under a still-quarantined key is excluded from reuse and counts as absent for anchor selection, so concurrent full-resends cannot restore the suppressed anchor. Unit regressions. +- [x] 4.6 Evaluate the quarantine suppression independently of fresh-reattach eligibility (third-round review): a quarantined full-resend whose reattach gate is already false (conversation-scoped payload, live alias session, owner forward falling back locally) still dispatches unanchored — the flag reaches hydration/session-level/recovery injection regardless. Unit regression (live alias session shape). + +## 5. Verification + +- [x] 5.1 `uv run ruff check app tests` and `uv run ruff format --check` on touched files. +- [x] 5.2 `uv run ty check`. +- [x] 5.3 `uv run python scripts/check_proxy_architecture.py` (no budget raised; new logic lives in un-ratcheted `quarantine.py`). +- [x] 5.4 Targeted suites: `tests/unit/test_proxy_http_bridge.py`, `tests/unit/test_http_bridge_cancel_drain.py`, `tests/unit/test_http_bridge_safe_continuity.py`, `tests/unit/test_responses_streaming_timeout_hardening.py`, `tests/integration/test_http_responses_bridge.py`. +- [x] 5.5 `openspec validate quarantine-silent-bridge-sessions --strict` and `openspec validate --specs`. diff --git a/tests/integration/test_http_responses_bridge.py b/tests/integration/test_http_responses_bridge.py index 41ee741d2..93328b6c4 100644 --- a/tests/integration/test_http_responses_bridge.py +++ b/tests/integration/test_http_responses_bridge.py @@ -34,6 +34,7 @@ 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 +from app.modules.proxy._service.http_bridge import quarantine as http_bridge_quarantine_module from app.modules.proxy._service.http_bridge import streaming as http_bridge_streaming_module from app.modules.proxy._service.http_bridge.helpers import ( _release_http_bridge_unanchored_handoff, @@ -9661,6 +9662,104 @@ async def test_http_bridge_stale_gate_retires_after_leading_rate_limit_telemetry assert upstream.closed is True +@pytest.mark.asyncio +async def test_http_bridge_stale_gate_direct_retirement_quarantines_wedged_reattach( + app_instance, + monkeypatch, +): + """Regression for the #1534 quarantine bypass: when the silent reattach is + the ONLY stale pending request, the stuck-gate watchdog retires the whole + session directly (no partial cleanup, no reader-failure funnel). That + direct retirement must still quarantine the key, or the next request + rebuilds the identical anchored wedge.""" + app_settings = _make_app_settings( + enabled=True, + admission_wait_timeout_seconds=0.001, + ) + app_settings.http_responses_session_bridge_stuck_gate_retire_after_seconds = 0.01 + _install_proxy_settings( + monkeypatch, + app_settings=app_settings, + dashboard_settings=_make_dashboard_settings(), + ) + service = get_proxy_service_for_app(app_instance) + http_bridge_quarantine_module._http_bridge_quarantine_registry(service).clear() + upstream = _SilentUpstreamWebSocket() + key = proxy_module._HTTPBridgeSessionKey("session_header", "quarantine-direct-retire-all-stale", None) + gate = asyncio.Semaphore(1) + await gate.acquire() + wedged_reattach = proxy_module._WebSocketRequestState( + request_id="req-wedged-direct-retire", + model="gpt-5.6-sol", + service_tier=None, + reasoning_effort="high", + api_key_reservation=None, + started_at=time.monotonic() - 1.0, + transport="http", + response_create_gate=gate, + response_create_gate_acquired=True, + awaiting_response_created=True, + event_queue=asyncio.Queue(), + ) + # The #1534 wedge shape: a proxy-injected reattach whose response.create + # was sent and that streamed response events, but whose response.created + # was never assigned. + wedged_reattach.proxy_injected_previous_response_id = True + wedged_reattach.previous_response_id = "resp_wedged_direct_retire" + wedged_reattach.response_create_sent_at = time.monotonic() - 1.0 + wedged_reattach.response_event_count = 3 + wedged_reattach.last_upstream_activity_at = time.monotonic() - 1.0 + session = proxy_module._HTTPBridgeSession( + key=key, + headers={}, + affinity=proxy_module._AffinityPolicy(key="quarantine-direct-retire-all-stale"), + request_model="gpt-5.6-sol", + account=cast(Account, SimpleNamespace(id="acct-quarantine-direct-retire", status=AccountStatus.ACTIVE)), + upstream=cast(proxy_module.UpstreamWebSocket, upstream), + upstream_control=proxy_module._WebSocketUpstreamControl(), + pending_requests=deque([wedged_reattach]), + pending_lock=anyio.Lock(), + response_create_gate=gate, + queued_request_count=1, + last_used_at=time.monotonic(), + idle_ttl_seconds=120.0, + ) + service._http_bridge_sessions[key] = session + + waiter = proxy_module._WebSocketRequestState( + request_id="req-waiting-behind-wedged-reattach", + model="gpt-5.6-sol", + service_tier=None, + reasoning_effort="high", + api_key_reservation=None, + started_at=time.monotonic(), + transport="http", + downstream_visible=True, + ) + try: + with pytest.raises(proxy_module.ProxyResponseError) as exc_info: + await service._acquire_request_state_response_create_admission( + waiter, + response_create_gate=gate, + bridge_session=session, + ) + finally: + if gate.locked(): + gate.release() + + assert exc_info.value.payload["error"]["code"] == "response_create_gate_timeout" + assert session.closed is True + assert key not in service._http_bridge_sessions + assert upstream.closed is True + # The direct all-stale retirement must record the quarantine so the next + # request takes the fresh no-anchor path instead of re-attaching. + assert session.quarantined is True + assert http_bridge_quarantine_module._http_bridge_session_key_quarantined(service, key) is True + entry = http_bridge_quarantine_module._http_bridge_quarantine_registry(service)[key] + assert entry.reason == "reattach_missing_response_created" + http_bridge_quarantine_module._http_bridge_quarantine_registry(service).clear() + + @pytest.mark.asyncio async def test_codex_responses_http_bridge_replaces_retired_gate_without_client_retry( async_client, @@ -14092,3 +14191,365 @@ async def test_prepare_http_bridge_request_preserves_existing_client_metadata(ap assert first_request_state.request_id.startswith("ws_") assert second_request_state.request_id.startswith("ws_") assert first_request_state.request_id != second_request_state.request_id + + +class _EventsWithoutCreatedUpstreamWebSocket(_FakeBridgeUpstreamWebSocket): + """Streams response events but never ``response.created``, then closes. + + Models the #1534 production wedge: a reattached HTTP-bridge stream that + delivers upstream response events whose ``response.created`` is never + assigned, so the turn can only end without a completed response. + """ + + async def send_text(self, text: str) -> None: + self.sent_text.append(text) + for delta in ("thinking", " harder"): + await self._messages.put( + _FakeUpstreamMessage( + "text", + text=json.dumps( + {"type": "response.reasoning_summary_text.delta", "delta": delta}, + separators=(",", ":"), + ), + ) + ) + await self._messages.put(_FakeUpstreamMessage("close", close_code=1000)) + + +@pytest.mark.asyncio +async def test_v1_responses_http_bridge_quarantines_reattach_that_streams_without_response_created( + async_client, app_instance, monkeypatch +): + """Regression for #1534: a reattach that streams events but never gets + ``response.created`` must quarantine the session so the next request does + not rebuild the identical anchored reattach and instead completes on the + fresh no-anchor path.""" + _install_bridge_settings(monkeypatch, enabled=True) + account_id = await _import_account( + async_client, + "acc_http_bridge_quarantine_silent", + "http-bridge-quarantine-silent@example.com", + ) + account = await _get_account(account_id) + service = get_proxy_service_for_app(app_instance) + http_bridge_quarantine_module._http_bridge_quarantine_registry(service).clear() + first_upstream = _ClosingInterruptedCustomToolUpstreamWebSocket("resp_quarantine_source") + wedged_upstream = _EventsWithoutCreatedUpstreamWebSocket("resp_quarantine_wedge") + fresh_upstream = _FakeBridgeUpstreamWebSocket("resp_quarantine_fresh") + upstreams = [first_upstream, wedged_upstream, fresh_upstream] + connect_count = 0 + + async def fake_select_account_with_budget(self, deadline, **kwargs): + del self, deadline, kwargs + return AccountSelection(account=account, error_message=None, error_code=None) + + async def fake_ensure_fresh_with_budget(self, target, *, force=False, timeout_seconds): + del self, force, timeout_seconds + return target + + async def fake_connect_responses_websocket( + headers, + access_token, + account_id_header, + *, + base_url=None, + session=None, + ): + nonlocal connect_count + del headers, access_token, account_id_header, base_url, session + connect_count += 1 + return upstreams[connect_count - 1] + + monkeypatch.setattr(proxy_module.ProxyService, "_select_account_with_budget", fake_select_account_with_budget) + monkeypatch.setattr(proxy_module.ProxyService, "_ensure_fresh_with_budget", fake_ensure_fresh_with_budget) + monkeypatch.setattr(proxy_module, "connect_responses_websocket", fake_connect_responses_websocket) + + session_headers = {"x-codex-session-id": "quarantine-silent-reattach"} + historical_input = [ + {"role": "user", "content": [{"type": "input_text", "text": "leading question"}]}, + { + "type": "additional_tools", + "role": "developer", + "tools": [{"type": "custom", "name": "shell"}], + }, + { + "type": "message", + "role": "developer", + "content": [{"type": "input_text", "text": "canonical Lite instructions"}], + }, + { + "role": "user", + "content": [{"type": "input_text", "text": "first question"}], + }, + { + "type": "custom_tool_call", + "call_id": "call_historical_shell", + "name": "shell", + "input": "printf historical", + }, + { + "role": "developer", + "content": [{"type": "input_text", "text": "historical control"}], + }, + { + "type": "custom_tool_call_output", + "call_id": "call_historical_shell", + "output": "historical", + }, + ] + first = await asyncio.wait_for( + async_client.post( + "/v1/responses", + json={ + "model": "gpt-5.1", + "instructions": "Return exactly OK.", + "input": historical_input, + }, + headers=session_headers, + ), + timeout=_TEST_SYNC_TIMEOUT_SECONDS, + ) + assert first.status_code == 200, first.text + + full_resend = [ + *historical_input, + { + "type": "custom_tool_call", + "call_id": "call_custom_shell", + "name": "shell", + "input": "pwd", + }, + { + "type": "custom_tool_call_output", + "call_id": "call_custom_shell", + "output": "/workspace", + }, + ] + second = await asyncio.wait_for( + async_client.post( + "/v1/responses", + json={ + "model": "gpt-5.1", + "instructions": "Return exactly OK.", + "input": full_resend, + }, + headers=session_headers, + ), + timeout=_TEST_SYNC_TIMEOUT_SECONDS, + ) + + # The reattach injected the durable anchor and then wedged: events flowed + # but response.created never arrived, so the turn fails terminally. + assert second.status_code != 200 + assert len(wedged_upstream.sent_text) == 1 + wedged_payload = json.loads(wedged_upstream.sent_text[0]) + assert wedged_payload["previous_response_id"] == "resp_bridge_custom_1" + quarantined_entries = [ + entry + for entry in http_bridge_quarantine_module._http_bridge_quarantine_registry(service).values() + if entry.quarantined_until > time.monotonic() + ] + assert len(quarantined_entries) == 1 + assert quarantined_entries[0].reason == "reattach_missing_response_created" + + third = await asyncio.wait_for( + async_client.post( + "/v1/responses", + json={ + "model": "gpt-5.1", + "instructions": "Return exactly OK.", + "input": full_resend, + }, + headers=session_headers, + ), + timeout=_TEST_SYNC_TIMEOUT_SECONDS, + ) + + # The quarantined key must not rebuild the identical anchored reattach: + # the client's own full resend goes upstream unanchored and completes. + assert third.status_code == 200, third.text + assert third.json()["id"] == "resp_quarantine_fresh_1" + assert connect_count == 3 + assert len(fresh_upstream.sent_text) == 1 + fresh_payload = json.loads(fresh_upstream.sent_text[0]) + assert "previous_response_id" not in fresh_payload + assert fresh_payload["input"] == full_resend + # The completed response on the fresh path clears the quarantine again. + assert not [ + entry + for entry in http_bridge_quarantine_module._http_bridge_quarantine_registry(service).values() + if entry.quarantined_until > time.monotonic() + ] + + +@pytest.mark.asyncio +async def test_v1_responses_http_bridge_quarantined_unsafe_full_resend_dispatches_unanchored( + async_client, app_instance, monkeypatch +): + """Regression for the #1534 session-state side door: a quarantined + full-resend whose durable prefix is trimmable but whose fresh suffix does + NOT retain the prior output must go upstream genuinely unanchored. Before + the fix, the early durable-anchor injection was suppressed but session + hydration restored ``last_completed_response_id`` and the session-level + injection re-added the same anchor and trimmed the prefix — rebuilding the + wedge despite the ``fresh_reattach_anchor_skipped_quarantined`` log.""" + _install_bridge_settings(monkeypatch, enabled=True) + account_id = await _import_account( + async_client, + "acc_http_bridge_quarantine_unsafe_suffix", + "http-bridge-quarantine-unsafe-suffix@example.com", + ) + account = await _get_account(account_id) + service = get_proxy_service_for_app(app_instance) + http_bridge_quarantine_module._http_bridge_quarantine_registry(service).clear() + first_upstream = _ClosingInterruptedCustomToolUpstreamWebSocket("resp_quarantine_unsafe_source") + wedged_upstream = _EventsWithoutCreatedUpstreamWebSocket("resp_quarantine_unsafe_wedge") + fresh_upstream = _FakeBridgeUpstreamWebSocket("resp_quarantine_unsafe_fresh") + upstreams = [first_upstream, wedged_upstream, fresh_upstream] + connect_count = 0 + + async def fake_select_account_with_budget(self, deadline, **kwargs): + del self, deadline, kwargs + return AccountSelection(account=account, error_message=None, error_code=None) + + async def fake_ensure_fresh_with_budget(self, target, *, force=False, timeout_seconds): + del self, force, timeout_seconds + return target + + async def fake_connect_responses_websocket( + headers, + access_token, + account_id_header, + *, + base_url=None, + session=None, + ): + nonlocal connect_count + del headers, access_token, account_id_header, base_url, session + connect_count += 1 + return upstreams[connect_count - 1] + + monkeypatch.setattr(proxy_module.ProxyService, "_select_account_with_budget", fake_select_account_with_budget) + monkeypatch.setattr(proxy_module.ProxyService, "_ensure_fresh_with_budget", fake_ensure_fresh_with_budget) + monkeypatch.setattr(proxy_module, "connect_responses_websocket", fake_connect_responses_websocket) + + # The turn-state header makes the bridge session a true Codex continuity + # session (``session.codex_session``), which is what arms the session-level + # anchor injection this regression guards against. + session_headers = { + "x-codex-session-id": "quarantine-unsafe-suffix-reattach", + "x-codex-turn-state": "quarantine-unsafe-suffix-turn", + } + historical_input = [ + {"role": "user", "content": [{"type": "input_text", "text": "leading question"}]}, + { + "type": "additional_tools", + "role": "developer", + "tools": [{"type": "custom", "name": "shell"}], + }, + { + "type": "message", + "role": "developer", + "content": [{"type": "input_text", "text": "canonical Lite instructions"}], + }, + { + "role": "user", + "content": [{"type": "input_text", "text": "first question"}], + }, + { + "type": "custom_tool_call", + "call_id": "call_historical_shell", + "name": "shell", + "input": "printf historical", + }, + { + "role": "developer", + "content": [{"type": "input_text", "text": "historical control"}], + }, + { + "type": "custom_tool_call_output", + "call_id": "call_historical_shell", + "output": "historical", + }, + ] + first = await asyncio.wait_for( + async_client.post( + "/v1/responses", + json={ + "model": "gpt-5.1", + "instructions": "Return exactly OK.", + "input": historical_input, + }, + headers=session_headers, + ), + timeout=_TEST_SYNC_TIMEOUT_SECONDS, + ) + assert first.status_code == 200, first.text + + full_resend = [ + *historical_input, + { + "type": "custom_tool_call", + "call_id": "call_custom_shell", + "name": "shell", + "input": "pwd", + }, + { + "type": "custom_tool_call_output", + "call_id": "call_custom_shell", + "output": "/workspace", + }, + ] + second = await asyncio.wait_for( + async_client.post( + "/v1/responses", + json={ + "model": "gpt-5.1", + "instructions": "Return exactly OK.", + "input": full_resend, + }, + headers=session_headers, + ), + timeout=_TEST_SYNC_TIMEOUT_SECONDS, + ) + + # The reattach injected the durable anchor and then wedged: the key is now + # quarantined. + assert second.status_code != 200 + assert len(wedged_upstream.sent_text) == 1 + assert json.loads(wedged_upstream.sent_text[0])["previous_response_id"] == "resp_bridge_custom_1" + assert [ + entry + for entry in http_bridge_quarantine_module._http_bridge_quarantine_registry(service).values() + if entry.quarantined_until > time.monotonic() + ] + + # Full resend whose durable prefix is trimmable but whose fresh suffix is + # a plain user turn: it neither retains the prior output nor matches the + # pending tool calls, so the safe-fresh-context proof fails. + unsafe_suffix_resend = [ + *historical_input, + {"role": "user", "content": [{"type": "input_text", "text": "follow-up without prior output"}]}, + ] + third = await asyncio.wait_for( + async_client.post( + "/v1/responses", + json={ + "model": "gpt-5.1", + "instructions": "Return exactly OK.", + "input": unsafe_suffix_resend, + }, + headers=session_headers, + ), + timeout=_TEST_SYNC_TIMEOUT_SECONDS, + ) + + # The dispatch must be genuinely unanchored: no early durable injection, + # no session-level re-injection of the same anchor, no prefix trim. + assert third.status_code == 200, third.text + assert third.json()["id"] == "resp_quarantine_unsafe_fresh_1" + assert connect_count == 3 + assert len(fresh_upstream.sent_text) == 1 + fresh_payload = json.loads(fresh_upstream.sent_text[0]) + assert "previous_response_id" not in fresh_payload + assert fresh_payload["input"] == unsafe_suffix_resend diff --git a/tests/unit/test_proxy_http_bridge.py b/tests/unit/test_proxy_http_bridge.py index b0a61903b..68aede235 100644 --- a/tests/unit/test_proxy_http_bridge.py +++ b/tests/unit/test_proxy_http_bridge.py @@ -40,6 +40,7 @@ from app.modules.proxy._service import support as proxy_support_module from app.modules.proxy._service.http_bridge import helpers as http_bridge_helpers_module from app.modules.proxy._service.http_bridge import mixin as http_bridge_mixin_module +from app.modules.proxy._service.http_bridge import quarantine as http_bridge_quarantine_module from app.modules.proxy._service.http_bridge import request_submit as http_bridge_request_submit_module from app.modules.proxy._service.http_bridge import retry_circuit as http_bridge_retry_circuit_module from app.modules.proxy._service.http_bridge import streaming as http_bridge_streaming_module @@ -23374,3 +23375,661 @@ async def fake_stream_events(*args: object, **kwargs: object): assert chunks == ['data: {"type":"response.completed"}\n\n'] assert preferred_account_ids == ["acc-capped", None] assert request_state.preferred_account_id is None + + +# --- Silent-session quarantine (fix for #1534; first-party takeover of #1405) --- + + +def _make_wedged_reattach_request_state( + *, + request_id: str = "req-wedged-reattach", + response_event_count: int = 3, +) -> proxy_service._WebSocketRequestState: + request_state = _make_eventless_http_bridge_owner(request_id=request_id, sent_at=100.0) + request_state.proxy_injected_previous_response_id = True + request_state.response_event_count = response_event_count + return request_state + + +@pytest.mark.parametrize( + ("mutate", "expected"), + [ + pytest.param(lambda state: None, True, id="wedged-reattach"), + pytest.param(lambda state: setattr(state, "response_id", "resp_created"), False, id="created-assigned"), + pytest.param( + lambda state: setattr(state, "latency_response_created_ms", 1200), + False, + id="created-latency-observed", + ), + pytest.param(lambda state: setattr(state, "response_event_count", 0), False, id="fully-eventless"), + pytest.param( + lambda state: setattr(state, "proxy_injected_previous_response_id", False), + False, + id="not-a-proxy-injected-reattach", + ), + pytest.param(lambda state: setattr(state, "response_create_sent_at", None), False, id="create-never-sent"), + pytest.param(lambda state: setattr(state, "transport", "websocket"), False, id="websocket-transport"), + pytest.param(lambda state: setattr(state, "skip_request_log", True), False, id="internal-request"), + ], +) +def test_http_bridge_wedged_reattach_predicate(mutate, expected: bool) -> None: + request_state = _make_wedged_reattach_request_state() + mutate(request_state) + assert http_bridge_quarantine_module._http_bridge_request_state_wedged_reattach(request_state) is expected + + +def test_http_bridge_quarantine_marks_key_and_expires_by_ttl(caplog: pytest.LogCaptureFixture) -> None: + service = SimpleNamespace() + session = _make_bridge_session(key_value="quarantine-ttl") + + assert http_bridge_quarantine_module._http_bridge_session_key_quarantined(service, session.key) is False + + with caplog.at_level(logging.INFO, logger="app.modules.proxy.service"): + http_bridge_quarantine_module._quarantine_http_bridge_session( + service, + session, + reason="reattach_missing_response_created", + ) + + assert session.quarantined is True + assert http_bridge_quarantine_module._http_bridge_session_key_quarantined(service, session.key) is True + assert "http_bridge_event event=session_quarantined" in caplog.text + assert "reason=reattach_missing_response_created" in caplog.text + + # Expire the entry: the key leaves quarantine and the registry is pruned. + entry = http_bridge_quarantine_module._http_bridge_quarantine_registry(service)[session.key] + entry.quarantined_until = time.monotonic() - 1.0 + entry.last_touched_monotonic = ( + time.monotonic() - http_bridge_quarantine_module._HTTP_BRIDGE_QUARANTINE_TTL_SECONDS - 1.0 + ) + assert http_bridge_quarantine_module._http_bridge_session_key_quarantined(service, session.key) is False + assert session.key not in http_bridge_quarantine_module._http_bridge_quarantine_registry(service) + + +def test_http_bridge_quarantine_registry_is_size_bounded() -> None: + service = SimpleNamespace() + max_entries = http_bridge_quarantine_module._HTTP_BRIDGE_QUARANTINE_MAX_ENTRIES + for index in range(max_entries + 8): + session = _make_bridge_session(key_value=f"quarantine-bound-{index}") + http_bridge_quarantine_module._quarantine_http_bridge_session( + service, + session, + reason="repeated_eventless_timeout", + ) + assert len(http_bridge_quarantine_module._http_bridge_quarantine_registry(service)) <= max_entries + + +def test_http_bridge_quarantine_eventless_strikes_require_threshold(caplog: pytest.LogCaptureFixture) -> None: + service = SimpleNamespace() + session = _make_bridge_session(key_value="quarantine-strikes") + + http_bridge_quarantine_module._record_http_bridge_quarantine_eventless_timeout(service, session) + assert http_bridge_quarantine_module._http_bridge_session_key_quarantined(service, session.key) is False + assert session.quarantined is False + + with caplog.at_level(logging.INFO, logger="app.modules.proxy.service"): + http_bridge_quarantine_module._record_http_bridge_quarantine_eventless_timeout(service, session) + assert http_bridge_quarantine_module._http_bridge_session_key_quarantined(service, session.key) is True + assert session.quarantined is True + assert "reason=repeated_eventless_timeout" in caplog.text + + +def test_http_bridge_quarantine_cleared_by_completed_response(caplog: pytest.LogCaptureFixture) -> None: + service = SimpleNamespace() + session = _make_bridge_session(key_value="quarantine-clear") + http_bridge_quarantine_module._record_http_bridge_quarantine_eventless_timeout(service, session) + + # A healthy completion resets the strike counter before the threshold. + http_bridge_quarantine_module._clear_http_bridge_quarantine(service, session) + http_bridge_quarantine_module._record_http_bridge_quarantine_eventless_timeout(service, session) + assert http_bridge_quarantine_module._http_bridge_session_key_quarantined(service, session.key) is False + + http_bridge_quarantine_module._record_http_bridge_quarantine_eventless_timeout(service, session) + assert http_bridge_quarantine_module._http_bridge_session_key_quarantined(service, session.key) is True + with caplog.at_level(logging.INFO, logger="app.modules.proxy.service"): + http_bridge_quarantine_module._clear_http_bridge_quarantine(service, session) + assert http_bridge_quarantine_module._http_bridge_session_key_quarantined(service, session.key) is False + assert "http_bridge_event event=session_quarantine_cleared" in caplog.text + + +def test_http_bridge_quarantine_expired_strike_is_not_resurrected() -> None: + service = SimpleNamespace() + session = _make_bridge_session(key_value="quarantine-stale-strike") + + http_bridge_quarantine_module._record_http_bridge_quarantine_eventless_timeout(service, session) + entry = http_bridge_quarantine_module._http_bridge_quarantine_registry(service)[session.key] + # Age the first strike past its TTL: the next timeout must count as a + # fresh first strike, not a "consecutive" second one. + entry.last_touched_monotonic = ( + time.monotonic() - http_bridge_quarantine_module._HTTP_BRIDGE_QUARANTINE_TTL_SECONDS - 1.0 + ) + + http_bridge_quarantine_module._record_http_bridge_quarantine_eventless_timeout(service, session) + assert http_bridge_quarantine_module._http_bridge_session_key_quarantined(service, session.key) is False + fresh_entry = http_bridge_quarantine_module._http_bridge_quarantine_registry(service)[session.key] + assert fresh_entry.consecutive_eventless_timeouts == 1 + + +def test_http_bridge_quarantine_clear_restores_session_reusability() -> None: + service = SimpleNamespace() + session = _make_bridge_session(key_value="quarantine-clear-flag") + http_bridge_quarantine_module._quarantine_http_bridge_session( + service, + session, + reason="reattach_missing_response_created", + ) + assert session.quarantined is True + + # A completed response disproves the wedge: the surviving session must be + # reusable again, not permanently rejected by the session-object flag. + http_bridge_quarantine_module._clear_http_bridge_quarantine(service, session) + assert session.quarantined is False + assert http_bridge_quarantine_module._http_bridge_session_key_quarantined(service, session.key) is False + + +def test_http_bridge_session_reusable_for_lookup_excludes_quarantined() -> None: + service = SimpleNamespace() + session = _make_bridge_session(key_value="quarantine-reuse") + + def reusable() -> bool: + return http_bridge_helpers_module._http_bridge_session_reusable_for_lookup( + session=session, + key=session.key, + api_key=None, + incoming_turn_state=None, + previous_response_id=None, + preferred_account_id=None, + require_preferred_account=False, + service_tier_supported=True, + allow_closed_admission_handoff=False, + session_key_quarantined=http_bridge_quarantine_module._http_bridge_session_key_quarantined( + service, session.key + ), + ) + + assert reusable() is True + http_bridge_quarantine_module._quarantine_http_bridge_session( + service, + session, + reason="reattach_missing_response_created", + ) + assert reusable() is False + + +def test_http_bridge_session_reusable_for_lookup_recovers_after_quarantine_ttl() -> None: + """Registry pruning never resets ``session.quarantined``; the lookup must + treat the registry TTL as the source of truth so a live session does not + stay rejected forever past expiry.""" + service = SimpleNamespace() + session = _make_bridge_session(key_value="quarantine-ttl-reuse") + http_bridge_quarantine_module._quarantine_http_bridge_session( + service, + session, + reason="reattach_missing_response_created", + ) + + def reusable() -> bool: + return http_bridge_helpers_module._http_bridge_session_reusable_for_lookup( + session=session, + key=session.key, + api_key=None, + incoming_turn_state=None, + previous_response_id=None, + preferred_account_id=None, + require_preferred_account=False, + service_tier_supported=True, + allow_closed_admission_handoff=False, + session_key_quarantined=http_bridge_quarantine_module._http_bridge_session_key_quarantined( + service, session.key + ), + ) + + assert reusable() is False + + # Expire the registry entry the way the TTL prune does: the entry vanishes + # but the per-session flag survives untouched. + entry = http_bridge_quarantine_module._http_bridge_quarantine_registry(service)[session.key] + entry.quarantined_until = time.monotonic() - 1.0 + entry.last_touched_monotonic = ( + time.monotonic() - http_bridge_quarantine_module._HTTP_BRIDGE_QUARANTINE_TTL_SECONDS - 1.0 + ) + assert http_bridge_quarantine_module._http_bridge_session_key_quarantined(service, session.key) is False + assert session.quarantined is True + + assert reusable() is True + # The lookup also resets the stale flag so session state agrees with the + # registry again. + assert session.quarantined is False + + +def test_http_bridge_session_reusable_for_lookup_rejects_fresh_session_under_quarantined_key() -> None: + """The registry verdict is authoritative for the key: a freshly created + replacement session (per-session flag still False) under a + still-quarantined key must not be reusable, or a concurrent full-resend + could restore the suppressed durable anchor through session hydration.""" + service = SimpleNamespace() + wedged = _make_bridge_session(key_value="quarantine-key-authority") + http_bridge_quarantine_module._quarantine_http_bridge_session( + service, + wedged, + reason="reattach_missing_response_created", + ) + replacement = _make_bridge_session(key_value="quarantine-key-authority") + assert replacement.key == wedged.key + assert replacement.quarantined is False + + def reusable() -> bool: + return http_bridge_helpers_module._http_bridge_session_reusable_for_lookup( + session=replacement, + key=replacement.key, + api_key=None, + incoming_turn_state=None, + previous_response_id=None, + preferred_account_id=None, + require_preferred_account=False, + service_tier_supported=True, + allow_closed_admission_handoff=False, + session_key_quarantined=http_bridge_quarantine_module._http_bridge_session_key_quarantined( + service, replacement.key + ), + ) + + assert reusable() is False + + # A completed response on the key clears the quarantine and the + # replacement becomes reusable. + http_bridge_quarantine_module._clear_http_bridge_quarantine(service, replacement) + assert reusable() is True + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("created_assigned", "expect_quarantined"), + [ + pytest.param(False, True, id="wedged-reattach-quarantined"), + pytest.param(True, False, id="created-assigned-never-quarantined"), + ], +) +async def test_http_bridge_reader_failure_quarantines_wedged_reattach_session( + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, + created_assigned: bool, + expect_quarantined: bool, +) -> None: + service = proxy_service.ProxyService(cast(Any, nullcontext())) + request_state = _make_wedged_reattach_request_state() + if created_assigned: + # The deferred-reasoning-style live turn: response.created was + # observed and events keep flowing with long gaps. Its retirement + # must never quarantine the session. + request_state.response_id = "resp_deferred_reasoning" + request_state.latency_response_created_ms = 900 + session = _make_bridge_session( + key_value="quarantine-reader-failure", + pending_requests=deque([request_state]), + queued_request_count=1, + ) + monkeypatch.setattr(service, "_fail_pending_websocket_requests", AsyncMock()) + monkeypatch.setattr(service, "_retire_stale_pending_http_bridge_session", AsyncMock()) + + with caplog.at_level(logging.INFO, logger="app.modules.proxy.service"): + await service._fail_http_bridge_reader_and_maybe_retire( + session, + error_code="stream_incomplete", + error_message="upstream closed mid-reattach", + ) + + assert session.quarantined is expect_quarantined + assert ( + http_bridge_quarantine_module._http_bridge_session_key_quarantined(service, session.key) is expect_quarantined + ) + assert ("http_bridge_event event=session_quarantined" in caplog.text) is expect_quarantined + + +@pytest.mark.asyncio +async def test_fail_stale_http_bridge_pending_requests_quarantines_wedged_gate_holder( + monkeypatch: pytest.MonkeyPatch, +) -> None: + service = proxy_service.ProxyService(cast(Any, nullcontext())) + wedged = _make_wedged_reattach_request_state(request_id="req-stale-wedged") + session = _make_bridge_session( + key_value="quarantine-stale-gate", + pending_requests=deque([wedged]), + queued_request_count=1, + ) + monkeypatch.setattr(service, "_fail_pending_websocket_requests", AsyncMock()) + record_failure = AsyncMock() + monkeypatch.setattr(service, "_record_http_bridge_retry_circuit_failure", record_failure) + + await service._fail_stale_http_bridge_pending_requests( + session, + [wedged], + detail="response_create_gate_timeout_stuck_pending", + ) + + assert session.quarantined is True + assert http_bridge_quarantine_module._http_bridge_session_key_quarantined(service, session.key) is True + # The wedged holder saw response events, so the eventless retry circuit + # is deliberately not charged for it. + record_failure.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_http_bridge_missing_created_timeout_records_eventless_quarantine_strike( + monkeypatch: pytest.MonkeyPatch, +) -> None: + service = proxy_service.ProxyService(cast(Any, nullcontext())) + upstream = _SilentEventlessUpstream() + session = _make_bridge_session(key_value="quarantine-eventless-strike") + session.upstream = cast(UpstreamWebSocket, upstream) + service._http_bridge_sessions[session.key] = session + settings = _make_app_settings( + sse_keepalive_interval_seconds=0.0, + stream_idle_timeout_seconds=60.0, + http_responses_session_bridge_request_budget_seconds=60.0, + http_responses_session_bridge_stuck_gate_retire_after_seconds=0.02, + ) + monkeypatch.setattr(proxy_service, "get_settings", lambda: settings) + monkeypatch.setattr(service, "_retry_http_bridge_precreated_request", AsyncMock(return_value=False)) + monkeypatch.setattr(service, "_handle_stream_error", AsyncMock()) + monkeypatch.setattr(service, "_write_request_log", AsyncMock()) + + reader_task = asyncio.create_task(service._relay_http_bridge_upstream_messages(session)) + await asyncio.wait_for(upstream.first_receive_started.wait(), timeout=0.5) + + gate = session.response_create_gate + await gate.acquire() + owner = _make_eventless_http_bridge_owner(request_id="req-quarantine-strike", sent_at=0.0) + owner.started_at = time.monotonic() - 30.0 + owner.response_create_sent_at = None + owner.response_create_gate = gate + owner.request_text = '{"type":"response.create","model":"gpt-5.6-sol","input":"hello"}' + async with session.pending_lock: + session.pending_requests.append(owner) + session.queued_request_count = 1 + + await http_bridge_request_submit_module._send_http_bridge_request_text_with_archive_id( + session, + owner, + owner.request_text, + ) + await asyncio.wait_for(reader_task, timeout=1.0) + + entry = http_bridge_quarantine_module._http_bridge_quarantine_registry(service)[session.key] + assert entry.consecutive_eventless_timeouts == 1 + # A single eventless timeout stays on the merged recovery path. + assert http_bridge_quarantine_module._http_bridge_session_key_quarantined(service, session.key) is False + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("created_assigned", "expect_quarantined"), + [ + pytest.param(False, True, id="wedged-reattach-quarantined"), + pytest.param(True, False, id="created-assigned-never-quarantined"), + ], +) +async def test_retire_stale_pending_http_bridge_session_quarantines_wedged_reattach( + monkeypatch: pytest.MonkeyPatch, + created_assigned: bool, + expect_quarantined: bool, +) -> None: + """Direct session retirement (the all-stale stuck-gate path) bypasses both + the partial-cleanup hook and the reader-failure funnel; it must still + quarantine the key when the retired pending proves the wedge shape.""" + service = proxy_service.ProxyService(cast(Any, nullcontext())) + wedged = _make_wedged_reattach_request_state(request_id="req-direct-retire-wedged") + if created_assigned: + wedged.response_id = "resp_created_direct_retire" + wedged.latency_response_created_ms = 700 + session = _make_bridge_session( + key_value="quarantine-direct-retire", + pending_requests=deque([wedged]), + queued_request_count=1, + ) + monkeypatch.setattr(service, "_close_http_bridge_session_bounded", AsyncMock()) + monkeypatch.setattr(service, "_record_http_bridge_retry_circuit_failure", AsyncMock()) + + await service._retire_stale_pending_http_bridge_session( + session, + detail="response_create_gate_timeout_stuck_pending", + response_events_seen=wedged.response_event_count, + ) + + assert session.quarantined is expect_quarantined + assert ( + http_bridge_quarantine_module._http_bridge_session_key_quarantined(service, session.key) is expect_quarantined + ) + + +@pytest.mark.asyncio +async def test_stream_http_bridge_quarantined_full_resend_stays_unanchored_when_reattach_gate_already_false( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """The quarantine suppression must be evaluated independently of the + fresh-reattach eligibility gate: when that gate is already false (here a + live alias session), a quarantined full-resend must still dispatch + unanchored instead of restoring the wedged durable anchor through session + hydration and session-level injection.""" + service = proxy_service.ProxyService(cast(Any, nullcontext())) + historical_input = [ + {"role": "user", "content": [{"type": "input_text", "text": "leading question"}]}, + { + "type": "additional_tools", + "role": "developer", + "tools": [{"type": "custom", "name": "shell"}], + }, + { + "type": "message", + "role": "developer", + "content": [{"type": "input_text", "text": "canonical Lite instructions"}], + }, + {"role": "user", "content": [{"type": "input_text", "text": "first question"}]}, + { + "type": "custom_tool_call", + "call_id": "call_historical_shell", + "name": "shell", + "input": "printf historical", + }, + {"role": "developer", "content": [{"type": "input_text", "text": "historical control"}]}, + { + "type": "custom_tool_call_output", + "call_id": "call_historical_shell", + "output": "historical", + }, + ] + # Full resend with a trimmable durable prefix and a fresh suffix that does + # NOT retain the prior output (plain user turn). + payload = proxy_service.ResponsesRequest.model_validate( + { + "model": "gpt-5.6-sol", + "instructions": "test", + "input": [ + *historical_input, + {"role": "user", "content": [{"type": "input_text", "text": "follow-up without prior output"}]}, + ], + } + ) + bridge_key = proxy_service._HTTPBridgeSessionKey("session_header", "quarantine-eligibility-bypass", None) + prefix_fingerprint = http_bridge_streaming_module._fingerprint_input_items( + cast(list[Any], payload.input)[: len(historical_input)] + ) + durable_lookup = proxy_service.DurableBridgeLookup( + session_id="durable-quarantine-bypass", + canonical_kind=bridge_key.affinity_kind, + canonical_key=bridge_key.affinity_key, + api_key_scope="__anonymous__", + account_id="acc-bridge", + owner_instance_id="instance-a", + owner_epoch=1, + lease_expires_at=proxy_service.utcnow() + timedelta(seconds=60), + state=HttpBridgeSessionState.ACTIVE, + latest_turn_state=None, + latest_response_id="resp_wedged_anchor", + model="gpt-5.6-sol", + latest_input_item_count=len(historical_input), + latest_input_full_fingerprint=prefix_fingerprint, + ) + quarantined_session = _make_bridge_session(key=bridge_key, key_value=bridge_key.affinity_key) + http_bridge_quarantine_module._quarantine_http_bridge_session( + service, + quarantined_session, + reason="reattach_missing_response_created", + ) + fresh_session = _make_bridge_session(key=bridge_key, key_value=bridge_key.affinity_key) + fresh_session.codex_session = True + + 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, + **prepare_kwargs: object, + ) -> tuple[proxy_service._WebSocketRequestState, str]: + del api_key, api_key_reservation, request_id, client_ip, prepare_kwargs + prepared_payloads.append(prepared_payload) + request_state = proxy_service._WebSocketRequestState( + request_id=f"req-quarantine-bypass-{len(prepared_payloads)}", + model=prepared_payload.model, + service_tier=None, + reasoning_effort=None, + api_key_reservation=None, + started_at=time.monotonic(), + event_queue=asyncio.Queue(), + transport="http", + ) + request_state.previous_response_id = prepared_payload.previous_response_id + return request_state, json.dumps(dict(prepared_payload.to_payload()), separators=(",", ":")) + + async def fake_get_or_create( + key: proxy_service._HTTPBridgeSessionKey, + **kwargs: object, + ) -> proxy_service._HTTPBridgeSession: + del kwargs + assert key == bridge_key + return fresh_session + + dispatched_text: list[str] = [] + + async def fake_stream_events(*args: object, **kwargs: object): + dispatched_text.append(cast(str, kwargs["text_data"])) + yield 'data: {"type":"response.completed"}\n\n' + + dashboard_settings = SimpleNamespace( + sticky_threads_enabled=False, + openai_cache_affinity_max_age_seconds=1800, + ) + runtime_config = SimpleNamespace( + enabled=True, + idle_ttl_seconds=120.0, + codex_idle_ttl_seconds=1800.0, + max_sessions=8, + queue_limit=4, + prompt_cache_idle_ttl_seconds=120.0, + gateway_safe_mode=False, + ) + monkeypatch.setattr( + http_bridge_streaming_module, + "_service_get_settings_cache", + lambda: SimpleNamespace(get=AsyncMock(return_value=dashboard_settings)), + ) + monkeypatch.setattr(http_bridge_streaming_module, "_service_get_settings", _make_app_settings) + monkeypatch.setattr(http_bridge_streaming_module, "_http_bridge_runtime_config", lambda *args: runtime_config) + monkeypatch.setattr(service, "_resolve_file_account_for_responses", AsyncMock(return_value=None)) + monkeypatch.setattr(service._durable_bridge, "lookup_request_targets", AsyncMock(return_value=durable_lookup)) + # The bypass shape: a live (alias) session makes the fresh-reattach + # durable-anchor gate false before the quarantine is ever consulted. + monkeypatch.setattr(service, "_http_bridge_has_live_local_session", AsyncMock(return_value=True)) + monkeypatch.setattr(service, "_http_bridge_can_forward_to_active_owner", AsyncMock(return_value=False)) + monkeypatch.setattr(service, "_prepare_http_bridge_request", fake_prepare) + monkeypatch.setattr(service, "_get_or_create_http_bridge_session", fake_get_or_create) + monkeypatch.setattr(service, "_stream_http_bridge_session_events", fake_stream_events) + + stream = service._stream_http_bridge_or_retry( + payload, + {"x-codex-session-id": bridge_key.affinity_key}, + codex_session_affinity=True, + propagate_http_errors=True, + openai_cache_affinity=False, + api_key=None, + api_key_reservation=None, + suppress_text_done_events=False, + ) + chunks = [chunk async for chunk in stream] + + assert chunks == ['data: {"type":"response.completed"}\n\n'] + assert len(dispatched_text) == 1 + dispatched_payload = json.loads(dispatched_text[0]) + # Genuinely unanchored: the suppressed durable anchor did not come back + # through session hydration or session-level injection, and the client's + # payload was not prefix-trimmed against the durable stored context. + assert "previous_response_id" not in dispatched_payload + assert len(dispatched_payload["input"]) == len(historical_input) + 1 + assert fresh_session.last_completed_response_id is None + + +@pytest.mark.asyncio +async def test_http_bridge_has_live_local_session_treats_quarantined_as_absent() -> None: + """A quarantined session is detached at lookup time, so anchor selection + must treat it as absent: a delta-only fresh reattach then keeps the + durable anchor exactly as if the session were already gone.""" + service = proxy_service.ProxyService(cast(Any, nullcontext())) + session = _make_bridge_session(key_value="quarantine-live-lookup") + service._http_bridge_sessions[session.key] = session + + assert ( + await service._http_bridge_has_live_local_session( + key=session.key, + incoming_turn_state=None, + api_key=None, + ) + is True + ) + + http_bridge_quarantine_module._quarantine_http_bridge_session( + service, + session, + reason="reattach_missing_response_created", + ) + assert ( + await service._http_bridge_has_live_local_session( + key=session.key, + incoming_turn_state=None, + api_key=None, + ) + is False + ) + + # The registry verdict is authoritative for the key: a freshly created + # replacement session (per-session flag still False) under the + # still-quarantined key counts as absent too. + replacement = _make_bridge_session(key_value="quarantine-live-lookup") + assert replacement.quarantined is False + service._http_bridge_sessions[session.key] = replacement + assert ( + await service._http_bridge_has_live_local_session( + key=session.key, + incoming_turn_state=None, + api_key=None, + ) + is False + ) + service._http_bridge_sessions[session.key] = session + + # Registry expiry ends the quarantine even though pruning left the + # per-session flag set: the session counts as live again. + entry = http_bridge_quarantine_module._http_bridge_quarantine_registry(service)[session.key] + entry.quarantined_until = time.monotonic() - 1.0 + assert ( + await service._http_bridge_has_live_local_session( + key=session.key, + incoming_turn_state=None, + api_key=None, + ) + is True + )