fix(proxy): quarantine silent HTTP bridge sessions

app/core/config/settings.py

@@ -293,6 +293,8 @@ class Settings(BaseSettings):
     openai_prompt_cache_key_derivation_enabled: bool = True
     http_responses_session_bridge_enabled: bool = True
     http_responses_session_bridge_request_budget_seconds: float = Field(default=7200.0, gt=0)
+    http_responses_session_bridge_response_created_timeout_seconds: float = Field(default=5.0, gt=0)
+    http_responses_session_bridge_quarantine_seconds: float = Field(default=60.0, ge=0)
     http_responses_session_bridge_idle_ttl_seconds: float = Field(default=120.0, gt=0)
     http_responses_session_bridge_codex_idle_ttl_seconds: float = Field(default=900.0, gt=0)
     http_responses_session_bridge_codex_prewarm_enabled: bool = False

app/modules/proxy/_service/http_bridge/mixin.py

@@ -1479,6 +1479,7 @@ async def close_all_http_bridge_sessions(self) -> None:
             self._http_bridge_sessions.clear()
             self._http_bridge_inflight_sessions.clear()
             self._http_bridge_previous_response_index.clear()
+            self._http_bridge_quarantine_until.clear()
         shutdown_error = ProxyResponseError(
             503,
             openai_error(

app/modules/proxy/_service/http_bridge/protocol.py

@@ -23,6 +23,7 @@ class _HTTPBridgeServiceProtocol(Protocol):
     _http_bridge_inflight_sessions: Any
     _http_bridge_turn_state_index: Any
     _http_bridge_previous_response_index: Any
+    _http_bridge_quarantine_until: Any
     _sessions: Any
     _session_lock: Any
     _pending_lock: Any

app/modules/proxy/_service/http_bridge/request_submit.py

@@ -202,6 +202,7 @@ async def _send_http_bridge_request_text_with_archive_id(
     token = set_request_id(request_state.archive_request_id)
     try:
         await session.upstream.send_text(text_data)
+        request_state.response_create_sent_at = _service_time().monotonic()
     finally:
         reset_request_id(token)
 

app/modules/proxy/_service/support.py

@@ -707,6 +707,7 @@ class _WebSocketRequestState:
     latency_response_create_gate_wait_ms: int | None = None
     latency_bridge_queue_wait_ms: int | None = None
     response_create_gate_wait_started_at: float | None = None
+    response_create_sent_at: float | None = None
     bridge_queue_wait_started_at: float | None = None
     # Monotonic deadline of the original bridge request budget. Retry and
     # recovery paths re-prepare request states with a fresh started_at, so

app/modules/proxy/api.py

@@ -4444,6 +4444,7 @@ async def _stream_responses(
             startup_error,
             headers=rate_limit_headers,
         )
+    lifecycle_stream = stream
     stream = _normalize_public_responses_stream(
         _stream_response_error_events(
             stream,
@@ -4460,11 +4461,15 @@ async def _stream_responses(
             request_id=get_request_id(),
             route_family="responses",
         )
+    response_stream = inject_sse_keepalives(
+        stream,
+        get_settings().sse_keepalive_interval_seconds,
+        keepalive_frame=keepalive_frame,
+    )
     return StreamingResponse(
-        inject_sse_keepalives(
-            stream,
-            get_settings().sse_keepalive_interval_seconds,
-            keepalive_frame=keepalive_frame,
+        _close_response_stream_on_exit(
+            response_stream,
+            lifecycle_stream=lifecycle_stream,
         ),
         media_type="text/event-stream",
         headers={
@@ -4476,6 +4481,113 @@ async def _stream_responses(
     )
 
 
+async def _close_async_iterator(stream: AsyncIterator[str]) -> None:
+    aclose = getattr(stream, "aclose", None)
+    if not callable(aclose):
+        return
+    try:
+        await aclose()
+    except Exception:
+        logger.warning("Failed to close responses stream iterator", exc_info=True)
+
+
+async def _close_response_stream_on_exit(
+    stream: AsyncIterator[str],
+    *,
+    lifecycle_stream: AsyncIterator[str],
+) -> AsyncIterator[str]:
+    try:
+        async for chunk in stream:
+            yield chunk
+    finally:
+        await _close_async_iterator(stream)
+        if lifecycle_stream is not stream:
+            await _close_async_iterator(lifecycle_stream)
+
+
+class _BufferedStreamIterator:
+    def __init__(self, items: Iterable[str], stream: AsyncIterator[str]) -> None:
+        self._items = tuple(items)
+        self._index = 0
+        self._stream = stream
+        self._closed = False
+
+    def __aiter__(self) -> _BufferedStreamIterator:
+        return self
+
+    async def __anext__(self) -> str:
+        if self._closed:
+            raise StopAsyncIteration
+        if self._index < len(self._items):
+            item = self._items[self._index]
+            self._index += 1
+            return item
+        try:
+            return await self._stream.__anext__()
+        except StopAsyncIteration:
+            self._closed = True
+            raise
+
+    async def aclose(self) -> None:
+        if self._closed:
+            return
+        self._closed = True
+        await _close_async_iterator(self._stream)
+
+
+class _FirstTaskStreamIterator:
+    def __init__(self, first_task: asyncio.Task[str], stream: AsyncIterator[str]) -> None:
+        self._first_task = first_task
+        self._stream = stream
+        self._first_pending = True
+        self._closed = False
+
+    def __aiter__(self) -> _FirstTaskStreamIterator:
+        return self
+
+    async def __anext__(self) -> str:
+        if self._closed:
+            raise StopAsyncIteration
+        if self._first_pending:
+            self._first_pending = False
+            try:
+                return await self._first_task
+            except StopAsyncIteration:
+                self._closed = True
+                raise
+            except BaseException:
+                if not self._first_task.done():
+                    self._first_task.cancel()
+                await asyncio.gather(self._first_task, return_exceptions=True)
+                raise
+        try:
+            return await self._stream.__anext__()
+        except StopAsyncIteration:
+            self._closed = True
+            raise
+
+    async def aclose(self) -> None:
+        if self._closed:
+            return
+        self._closed = True
+        if not self._first_task.done():
+            self._first_task.cancel()
+        await asyncio.gather(self._first_task, return_exceptions=True)
+        await _close_async_iterator(self._stream)
+
+
+def _prepend_first(first: str | None, stream: AsyncIterator[str]) -> AsyncIterator[str]:
+    return _BufferedStreamIterator(() if first is None else (first,), stream)
+
+
+def _prepend_items(items: list[str], stream: AsyncIterator[str]) -> AsyncIterator[str]:
+    return _BufferedStreamIterator(items, stream)
+
+
+def _prepend_first_task(first_task: asyncio.Task[str], stream: AsyncIterator[str]) -> AsyncIterator[str]:
+    return _FirstTaskStreamIterator(first_task, stream)
+
+
 def _strip_internal_bridge_headers(headers: Mapping[str, str]) -> dict[str, str]:
     return {key: value for key, value in headers.items() if not key.lower().startswith("x-codex-bridge-")}
 
@@ -4953,13 +5065,6 @@ def _request_state_str(request: Request, name: str) -> str | None:
     return stripped or None
 
 
-async def _prepend_first(first: str | None, stream: AsyncIterator[str]) -> AsyncIterator[str]:
-    if first is not None:
-        yield first
-    async for line in stream:
-        yield line
-
-
 async def _read_first_stream_item(stream: AsyncIterator[str]) -> str:
     return await anext(stream)
 
@@ -5464,29 +5569,6 @@ async def _probe_chat_stream_startup_error(
     return _prepend_items(buffered, stream), None
 
 
-async def _prepend_items(items: list[str], stream: AsyncIterator[str]) -> AsyncIterator[str]:
-    for item in items:
-        yield item
-    async for line in stream:
-        yield line
-
-
-async def _prepend_first_task(first_task: asyncio.Task[str], stream: AsyncIterator[str]) -> AsyncIterator[str]:
-    try:
-        first = await first_task
-    except StopAsyncIteration:
-        return
-    finally:
-        # If the wrapping stream is closed before the first item is consumed
-        # (client disconnect, request teardown), cancel the still-running probe
-        # task so it does not hold the upstream connection open.
-        if not first_task.done():
-            first_task.cancel()
-    yield first
-    async for line in stream:
-        yield line
-
-
 async def _prepend_initial_sse_heartbeat(
     stream: AsyncIterator[str],
     keepalive_frame: str,

app/modules/proxy/service.py

@@ -947,16 +947,12 @@ def __init__(self, repo_factory: ProxyRepoFactory) -> None:
         self._http_bridge_inflight_sessions: dict[_HTTPBridgeSessionKey, asyncio.Future[_HTTPBridgeSession]] = {}
         self._http_bridge_turn_state_index: dict[tuple[str, str | None], _HTTPBridgeSessionKey] = {}
         self._http_bridge_previous_response_index: dict[tuple[str, str | None], _HTTPBridgeSessionKey] = {}
+        self._http_bridge_quarantine_until: dict[_HTTPBridgeSessionKey, float] = {}
         self._websocket_previous_response_account_index: dict[tuple[str, str | None, str | None], str] = {}
         self._websocket_continuity_index: dict[tuple[str, str | None], _WebSocketContinuityState] = {}
         self._background_cleanup_tasks: set[asyncio.Task[None]] = set()
-        # In-memory pin from upstream-issued file_id -> codex-lb account_id.
-        # Used so ``finalize_file`` for a given ``file_id`` is routed to
-        # the same account that handled ``create_file``. Cross-instance
-        # routing is best-effort: if the finalize request lands on a
-        # different replica with no pin, we fall back to a fresh load-
-        # balancer selection. The TTL is short enough (5 min) that we
-        # never hold stale pins after the upstream upload window closes.
+        # Best-effort in-memory pin keeps finalize_file on create_file's account.
+        # Its five-minute TTL bounds stale pins across replicas.
         self._file_account_pins: dict[str, _FilePinEntry] = {}
         self._file_account_pin_lock = asyncio.Lock()
         self._http_bridge_lock = anyio.Lock()

openspec/changes/quarantine-silent-http-bridge-sessions/.openspec.yaml

@@ -0,0 +1,2 @@
+schema: spec-driven
+created: 2026-07-20

openspec/changes/quarantine-silent-http-bridge-sessions/proposal.md

@@ -0,0 +1,43 @@
+## Why
+
+An upstream Responses websocket can accept `response.create` without ever
+emitting `response.created` or a terminal event. The HTTP bridge then keeps the
+request pending and holds the session response-create gate, so later requests
+can appear idle or fail admission even when other accounts have capacity.
+
+Replaying the same request automatically is unsafe because the proxy cannot
+know whether upstream accepted it. Recovery must retire the silent bridge
+without duplicating work, then let a new client retry use a transport that does
+not depend on that bridge.
+
+## What Changes
+
+- Bound the wait from sending `response.create` to receiving
+  `response.created`.
+- Retire and temporarily quarantine a bridge session that exceeds that
+  deadline, failing its current request without internally replaying it.
+- Route the next independent client retry for the quarantined affinity key over
+  direct HTTP while preserving any `previous_response_id` account owner.
+- Ensure closing a downstream streaming response also closes the bridge
+  lifecycle iterator and its upstream websocket, including when only the
+  initial heartbeat was consumed.
+- Keep the timeout and quarantine duration as internal, zero-config settings
+  with conservative defaults.
+
+## Capabilities
+
+### New Capabilities
+
+- None.
+
+### Modified Capabilities
+
+- `responses-api-compat`: defines safe recovery from an accepted-but-silent
+  bridge submission and downstream stream cancellation.
+
+## Impact
+
+- HTTP Responses session bridge lifecycle and transport selection.
+- Codex-native and OpenAI-compatible Responses streaming routes.
+- No database migration, dashboard change, required setup step, or public API
+  schema change.

openspec/changes/quarantine-silent-http-bridge-sessions/specs/responses-api-compat/spec.md

@@ -0,0 +1,63 @@
+## ADDED Requirements
+
+### Requirement: Silent HTTP bridge submissions fail closed without replay
+
+After an HTTP bridge session sends `response.create`, the service MUST bound
+the wait for upstream to emit `response.created`. If the deadline expires, the
+service MUST fail the current request terminally, retire and close the affected
+bridge session, release its pending request and response-create admission
+state, and temporarily quarantine that bridge affinity key.
+
+The service MUST NOT automatically replay the same submitted request within
+the current client call because upstream acceptance is ambiguous. During the
+quarantine window, the next independent client retry for that affinity key MUST
+bypass the HTTP bridge and use direct HTTP. If that retry carries
+`previous_response_id`, it MUST remain pinned to the response owner's account.
+A quarantined key MUST NOT prevent unrelated affinity keys or accounts from
+making progress.
+
+#### Scenario: Silent submission is retired without internal replay
+
+- **WHEN** an HTTP bridge sends `response.create`
+- **AND** upstream emits neither `response.created` nor a terminal event before the configured deadline
+- **THEN** the current request fails with a retryable terminal error
+- **AND** the bridge session is retired, removed from reuse, and closed
+- **AND** its pending queue entry and response-create gate are released
+- **AND** the service does not resend the request during the same client call
+
+#### Scenario: Client retry bypasses a quarantined bridge
+
+- **GIVEN** an affinity key was quarantined after a silent bridge submission
+- **WHEN** the client independently retries before the quarantine expires
+- **THEN** the service sends the retry over direct HTTP instead of creating or reusing an upstream websocket bridge
+- **AND** the direct HTTP response is streamed through the existing Responses contract
+
+#### Scenario: Previous-response fallback preserves account ownership
+
+- **GIVEN** a quarantined retry carries `previous_response_id`
+- **AND** the service can resolve the account that owns that response
+- **WHEN** the retry is sent over direct HTTP
+- **THEN** the retry uses the owning account
+- **AND** the service does not move the response chain to another available account
+
+#### Scenario: Silent bridge does not block unrelated accounts
+
+- **GIVEN** one affinity key has an accepted-but-silent bridge submission on account A
+- **WHEN** independent requests target eligible sessions on other accounts
+- **THEN** those requests can complete without waiting for the silent bridge deadline
+
+### Requirement: Downstream stream cancellation closes bridge lifecycle
+
+When a downstream Responses streaming body is closed, the service MUST close
+the bridge lifecycle iterator and release the associated pending request,
+response-create gate, and upstream websocket resources. This requirement
+applies even when the downstream consumed only the initial SSE heartbeat and
+no upstream `response.created` event.
+
+#### Scenario: Client closes after initial heartbeat
+
+- **GIVEN** a Codex-native HTTP Responses stream has emitted its initial heartbeat
+- **AND** upstream has not emitted `response.created`
+- **WHEN** the downstream client closes the stream
+- **THEN** the pending bridge request and queue admission are released
+- **AND** the bridge session and upstream websocket are closed

openspec/changes/quarantine-silent-http-bridge-sessions/tasks.md

@@ -0,0 +1,21 @@
+## 1. Specification
+
+- [x] Define the bounded `response.created` wait and no-replay safety rule.
+- [x] Define temporary direct-HTTP fallback with continuity account ownership.
+- [x] Define cleanup when the downstream closes after the initial heartbeat.
+
+## 2. Implementation
+
+- [x] Track and bound the post-submit wait for `response.created`.
+- [x] Retire, close, and quarantine silent bridge sessions.
+- [x] Bypass quarantined bridge keys on the next independent client retry.
+- [x] Close the bridge lifecycle iterator when the public stream closes.
+
+## 3. Verification
+
+- [x] Cover silent-session quarantine and direct-HTTP retry.
+- [x] Cover `previous_response_id` account ownership during fallback.
+- [x] Cover unrelated-account concurrency while one bridge is silent.
+- [x] Cover downstream cancellation after the initial heartbeat.
+- [x] Run focused and full bridge/API integration tests, lint, type checks, and
+  OpenSpec validation.