fix(proxy): bound upstream streams that never send response headers

app/core/clients/http.py

@@ -4,6 +4,7 @@
 import contextlib
 import logging
 import os
+import socket
 import ssl
 import time
 from collections.abc import AsyncIterator, Mapping
@@ -12,6 +13,7 @@
 
 import aiohttp
 import certifi
+from aiohappyeyeballs.types import AddrInfoType
 from aiohttp_retry import RetryClient
 from aiohttp_socks import ProxyConnector
 
@@ -49,6 +51,16 @@ class _ManagedHttpClient:
 _last_generationless_network_rotation_at: float | None = None
 _GENERATIONLESS_NETWORK_ROTATION_COOLDOWN_SECONDS = 1.0
 
+# Pooled upstream connections outlive the request that opened them, so a socket
+# dropped by an intermediary (NAT rebind, tunnel reconnect, route change) is
+# otherwise only discovered when an application-level timeout fires. Probes turn
+# that silent black hole into a transport error the failover paths already
+# handle. Idle/interval/count are chosen to declare a dead peer in ~90s, which
+# matches the pooled keepalive window below.
+_TCP_KEEPALIVE_IDLE_SECONDS = 30
+_TCP_KEEPALIVE_INTERVAL_SECONDS = 10
+_TCP_KEEPALIVE_PROBE_COUNT = 6
+
 
 def _socks_proxy_config(environ: Mapping[str, str | None] = os.environ) -> _SocksProxyConfig | None:
     request_method_set = bool(environ.get("REQUEST_METHOD"))
@@ -95,6 +107,46 @@ def _build_ssl_context() -> ssl.SSLContext:
     return context
 
 
+def _apply_tcp_keepalive(sock: socket.socket) -> None:
+    """Enable OS keepalive probes on an upstream socket.
+
+    Probe tuning is best-effort by design: ``TCP_KEEPIDLE`` is Linux-only,
+    macOS spells the same knob ``TCP_KEEPALIVE``, and other platforms may
+    expose neither. Failing client construction over a missing socket option
+    would trade a rare hang for a certain outage, so unsupported knobs are
+    skipped and only the enable step is required.
+    """
+    try:
+        sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
+    except OSError:
+        logger.debug("Upstream socket rejected SO_KEEPALIVE", exc_info=True)
+        return
+    for option_name, option_value in (
+        ("TCP_KEEPIDLE", _TCP_KEEPALIVE_IDLE_SECONDS),
+        ("TCP_KEEPALIVE", _TCP_KEEPALIVE_IDLE_SECONDS),
+        ("TCP_KEEPINTVL", _TCP_KEEPALIVE_INTERVAL_SECONDS),
+        ("TCP_KEEPCNT", _TCP_KEEPALIVE_PROBE_COUNT),
+    ):
+        option = getattr(socket, option_name, None)
+        if option is None:
+            continue
+        try:
+            sock.setsockopt(socket.IPPROTO_TCP, option, option_value)
+        except OSError:
+            logger.debug("Upstream socket rejected %s", option_name, exc_info=True)
+
+
+def _keepalive_socket_factory(addr_info: AddrInfoType) -> socket.socket:
+    family, socket_type, proto = addr_info[0], addr_info[1], addr_info[2]
+    sock = socket.socket(family=family, type=socket_type, proto=proto)
+    try:
+        _apply_tcp_keepalive(sock)
+    except BaseException:
+        sock.close()
+        raise
+    return sock
+
+
 class HttpClientLease:
     def __init__(self, managed_client: _ManagedHttpClient) -> None:
         self.client = managed_client.client
@@ -133,6 +185,7 @@ async def _build_http_client() -> HttpClient:
             limit_per_host=settings.http_connector_limit_per_host,
             ssl=ssl_context,
             rdns=socks_config.rdns,
+            socket_factory=_keepalive_socket_factory,
         )
     else:
         connector = aiohttp.TCPConnector(
@@ -146,6 +199,7 @@ async def _build_http_client() -> HttpClient:
             # around across turns instead.
             keepalive_timeout=90,
             ttl_dns_cache=300,
+            socket_factory=_keepalive_socket_factory,
         )
     session = aiohttp.ClientSession(
         connector=connector,
@@ -158,10 +212,16 @@ async def _build_http_client() -> HttpClient:
                 socks_config.connector_url,
                 ssl=ssl_context,
                 rdns=socks_config.rdns,
+                socket_factory=_keepalive_socket_factory,
             )
             ws_trust_env = False
         else:
-            ws_connector = aiohttp.TCPConnector(ssl=ssl_context, keepalive_timeout=90, ttl_dns_cache=300)
+            ws_connector = aiohttp.TCPConnector(
+                ssl=ssl_context,
+                keepalive_timeout=90,
+                ttl_dns_cache=300,
+                socket_factory=_keepalive_socket_factory,
+            )
             ws_trust_env = settings.upstream_websocket_trust_env
         try:
             websocket_session = aiohttp.ClientSession(

app/core/clients/proxy.py

@@ -2780,16 +2780,34 @@ async def _stream_responses_with_session(
         pre_request_started_at,
         time.monotonic(),
     )
+    # sock_read carries the idle budget into the phase before response headers
+    # exist. Without it, a connection that is established but never answered is
+    # bounded only by the request budget, which is hours long, while it holds a
+    # per-session response-create gate that later turns queue behind.
     timeout = aiohttp.ClientTimeout(
         total=remaining_request_timeout,
         sock_connect=effective_connect_timeout,
-        sock_read=None,
+        sock_read=effective_idle_timeout,
     )
     started_at = time.monotonic()
 
     async def _stream_via_http(
         current_headers: Mapping[str, str],
         current_timeout: aiohttp.ClientTimeout,
+    ) -> AsyncIterator[str]:
+        try:
+            async for event_block in _stream_via_http_attempt(current_headers, current_timeout):
+                yield event_block
+        except aiohttp.SocketTimeoutError as exc:
+            # A socket read timeout means the connection was established and
+            # then produced nothing. That is an idle stream, not a transport
+            # failure, so it joins the idle-timeout path instead of being
+            # reported as an unavailable upstream.
+            raise StreamIdleTimeoutError() from exc
+
+    async def _stream_via_http_attempt(
+        current_headers: Mapping[str, str],
+        current_timeout: aiohttp.ClientTimeout,
     ) -> AsyncIterator[str]:
         nonlocal status_code, last_stream_activity_at, error_code, error_message, seen_terminal
 
@@ -3051,7 +3069,7 @@ async def _stream_via_http_after_websocket_rejection(
         timeout = aiohttp.ClientTimeout(
             total=remaining_request_timeout,
             sock_connect=effective_connect_timeout,
-            sock_read=None,
+            sock_read=effective_idle_timeout,
         )
         started_at = time.monotonic()
         _maybe_log_upstream_request_start(

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

@@ -1123,6 +1123,7 @@ async def _process_parsed_http_bridge_upstream_event(
                 release_create_gate = False
 
             _archive_http_bridge_upstream_text(session, original_text, matched_request_state)
+            pending_request_count = len(session.pending_requests)
 
             if matched_request_state is not None:
                 now = _service_time().monotonic()
@@ -1346,6 +1347,22 @@ async def _process_parsed_http_bridge_upstream_event(
             session.upstream_control.reconnect_requested = True
             return
 
+        if status_request_state is None and pending_request_count:
+            # The bridge multiplexes one upstream connection across pending
+            # requests. An event that reaches none of them is dropped here, and
+            # whatever was waiting for it waits until a timeout fires, so the
+            # drop needs to be visible rather than inferred from a missing
+            # downstream response.
+            logger.warning(
+                "HTTP bridge upstream event matched no pending request account_id=%s bridge_kind=%s "
+                "event_type=%s has_response_id=%s pending_count=%d",
+                session.account.id,
+                session.key.affinity_kind,
+                event_type or "unknown",
+                response_id is not None,
+                pending_request_count,
+            )
+
         if status_request_state is not None and event_type not in {
             "response.completed",
             "response.failed",

openspec/changes/bound-stalled-upstream-streams/context.md

@@ -0,0 +1,74 @@
+# Context
+
+## The incident this change is built from
+
+2026-08-05, single-tenant deployment behind a Cloudflare tunnel. Codex CLI reported
+`stream disconnected before completion: idle timeout waiting for SSE` repeatedly, while
+the proxy's own error rate over the preceding three days was 13 failures in 1976
+requests.
+
+Log evidence at the time of the failure:
+
+```
+http_bridge_startup_wait_timeout stage=response_create_gate
+  bridge_key=sha256:8721fd39e627 available=0 pending_count=1 queued_count=5
+  pending_request_ids=d27c4984-… pending_request_ages_seconds=1548.9
+```
+
+`d27c4984` was submitted at 16:29:18Z. It logged `session_anchor_injected` and
+`store_context_input_trimmed`, then nothing: no `response.created`, no terminal event,
+no `request_logs` row (rows are written on completion). Meanwhile the same account
+served `gpt-5.6-terra` and `claude-opus-5` traffic with 4–13s latencies throughout,
+which rules out quota exhaustion, account health, and the network path to the proxy.
+
+The five queued requests behind it were retrying gate acquisition every 10 seconds and
+would have kept doing so until the 7200s request budget expired.
+
+## Why the idle timeout did not help
+
+`stream_idle_timeout_seconds` guards `_iter_sse_events`, which only runs once response
+headers exist. The failing request never got that far, so the only applicable bound was
+`http_responses_stream_request_budget_seconds` — also 7200s by default. Between the
+`upstream_connect_timeout_seconds` handshake bound (8s) and the request budget (2h)
+there was no bound at all.
+
+Carrying the idle timeout into `sock_read` closes that hole without introducing a
+fourth timeout for operators to reason about: the socket read that waits for response
+headers is bounded by the same number that bounds every later read.
+
+The resulting `aiohttp.SocketTimeoutError` is mapped to `StreamIdleTimeoutError` at the
+stream boundary rather than being classified further down, because the generic
+`aiohttp.ClientError` handler runs first and would otherwise report a silent
+established connection as an unavailable upstream. The classifier's existing tie-break
+between idle and request-budget expiry is deliberately left untouched.
+
+## Why keepalive probes matter here
+
+`keepalive_timeout=90` on the connector governs how long an *idle pooled* connection is
+retained; it says nothing about whether that connection is still alive. Without
+`SO_KEEPALIVE`, a socket dropped by a NAT or tunnel is only discovered when the
+application writes and eventually times out. Enabling probes turns a silent black hole
+into a transport error the existing failover already handles.
+
+Probe tuning is deliberately best-effort. `TCP_KEEPIDLE` is Linux-only; macOS exposes
+`TCP_KEEPALIVE` with different semantics; other platforms may expose neither. Failing
+client construction over a missing socket option would trade a rare hang for a certain
+outage.
+
+## What this change deliberately leaves alone
+
+The incident has a second half: the silent request held the per-session
+`response_create_gate` while it waited. `main` already covers that through
+`_http_bridge_pending_state_is_stale`, which uses `last_upstream_activity_at` as the
+silence clock and retires a holder that stopped progressing. The deployment where this
+was observed runs an older build without it.
+
+Retirement is a cleanup of the symptom; this change removes the condition that produces
+the symptom, so the two are complementary and stay in separate changes.
+
+## Scope note
+
+The unroutable-event logging is observability only. It was added because the incident
+could not distinguish "upstream went silent" from "event arrived and was not routed to
+its waiting request" — the bridge drops unmatched events without a trace today. If that
+log ever fires in practice, the routing gap it exposes is a separate change.

openspec/changes/bound-stalled-upstream-streams/proposal.md

@@ -0,0 +1,60 @@
+## Why
+
+Nothing bounds an upstream streaming request between the TCP handshake and the first
+response byte.
+
+The shared session is built with `ClientTimeout(total=None)`, streaming requests pass
+`sock_read=None`, and `upstream_connect_timeout_seconds` covers only the handshake. Once
+a connection is established, a peer that never sends response headers is bounded solely
+by `http_responses_stream_request_budget_seconds`, which defaults to 7200s. The stream
+idle timeout does not apply, because it guards `_iter_sse_events`, which only runs after
+headers exist.
+
+That gap is expensive because the waiting request holds its session's
+`response_create_gate` (an `asyncio.Semaphore(1)`). Observed in a single-tenant
+deployment on 2026-08-05: one bridged request submitted at 16:29:18Z produced no
+upstream event and no request-log row, while five later turns on the same session
+retried gate acquisition every 10 seconds for 26 minutes and the Codex client gave up
+with `stream disconnected before completion: idle timeout waiting for SSE`. Unrelated
+traffic on the same account completed normally throughout.
+
+Two conditions make the silent case likely and hard to see:
+
+1. **A dead pooled socket is indistinguishable from a slow model.** Connectors retain
+   idle connections for 90 seconds and enable no TCP keepalive probes, so a connection
+   dropped by an intermediary (NAT rebind, tunnel reconnect, route change) is only
+   discovered when the application layer eventually gives up.
+2. **A dropped event leaves no trace.** A bridge session multiplexes one upstream
+   connection across its pending requests; an event that matches none of them is
+   discarded silently, so "upstream went quiet" and "the event arrived and was not
+   routed" look identical in operations.
+
+## What Changes
+
+- The configured stream idle timeout MUST also bound the phase before response headers
+  arrive, so "connected but silent" is treated the same as "streaming then silent". No
+  new setting: `sock_read` carries the timeout that already exists.
+- A socket read timeout MUST be reported as `stream_idle_timeout` rather than as an
+  unavailable upstream, so the existing idle retry and failover paths apply.
+- Upstream TCP connectors MUST enable OS-level keepalive probes, so a connection killed
+  by an intermediary surfaces as a transport error instead of an indefinite wait. The
+  existing pooled-reuse guarantees (`keepalive_timeout >= 90`, `ttl_dns_cache >= 300`)
+  are unchanged.
+- Upstream events that match no pending request MUST be logged with a stable
+  low-cardinality reason while work is still waiting on the session.
+
+## Impact
+
+- Affected specs: `outbound-http-clients`, `proxy-runtime-observability`
+- Affected code: `app/core/clients/http.py`, `app/core/clients/proxy.py`,
+  `app/modules/proxy/_service/http_bridge/upstream_events.py`
+- No new settings, no migration, no dashboard surface. Behavior changes only in failure
+  paths that previously had no bound.
+- A deployment that relies on upstream taking longer than `stream_idle_timeout_seconds`
+  to send response headers would now see those attempts fail fast. That window is
+  operator-configurable and defaults to two hours, so the practical blast radius is
+  limited to genuinely dead connections.
+- The stuck-gate retirement side of this failure is already handled on `main` by
+  `_http_bridge_pending_state_is_stale`, which uses `last_upstream_activity_at` as the
+  silence clock. This change is deliberately limited to the transport gap that lets a
+  request go silent in the first place.

openspec/changes/bound-stalled-upstream-streams/specs/outbound-http-clients/spec.md

@@ -0,0 +1,55 @@
+## ADDED Requirements
+
+### Requirement: Upstream streaming requests are bounded before the first response byte
+
+A streaming upstream request MUST reach response headers within the effective stream
+idle timeout. The bound applies from the moment the request is issued, so a connection
+that is established but never answered fails on the same budget as a stream that stops
+mid-flight.
+
+Exceeding the bound MUST be reported with the existing `stream_idle_timeout` error code
+and failure detail, MUST release every resource the attempt holds — including the
+per-session response-create gate and any account lease — and MUST be eligible for the
+same retry and failover handling as an idle timeout observed after the first byte.
+
+Non-streaming control calls (token refresh, usage fetch, compaction) keep their own
+timeouts and are unaffected.
+
+#### Scenario: Established connection never returns response headers
+
+- **GIVEN** an upstream connection that completes its TCP and TLS handshake
+- **AND** the peer sends no response headers
+- **WHEN** the effective stream idle timeout elapses
+- **THEN** the attempt fails with `stream_idle_timeout`
+- **AND** the failure is recorded before the request budget would have expired
+
+#### Scenario: Response headers inside the bound stream normally
+
+- **GIVEN** an upstream request whose response headers arrive before the idle timeout
+- **WHEN** the stream then produces events with gaps shorter than the idle timeout
+- **THEN** the request completes normally
+- **AND** the pre-header bound does not truncate the stream
+
+## MODIFIED Requirements
+
+### Requirement: Upstream connectors persist across interactive turn gaps
+
+The shared upstream TCP connectors MUST configure connection keepalive of at least 90 seconds and a DNS cache TTL of at least 300 seconds, so consecutive interactive requests reuse pooled connections and resolved names instead of re-handshaking per turn.
+
+Because pooled connections outlive the requests that opened them, the connectors MUST
+also enable OS-level TCP keepalive probes on upstream sockets, so a connection dropped
+by an intermediary is reported as a transport error rather than waiting for an
+application-level timeout. Probe tuning beyond enabling keepalive is best-effort:
+platforms that do not expose the per-socket knobs MUST still enable keepalive and MUST
+NOT fail client construction.
+
+#### Scenario: Connector construction pins reuse settings
+
+- **WHEN** the shared HTTP client initializes its direct TCP connectors
+- **THEN** they are constructed with `keepalive_timeout >= 90` and `ttl_dns_cache >= 300`
+
+#### Scenario: Pooled sockets carry keepalive probes
+
+- **WHEN** the shared HTTP client creates an upstream socket
+- **THEN** `SO_KEEPALIVE` is enabled on that socket
+- **AND** client construction succeeds even when per-socket probe tuning is unavailable