fix(proxy): retry HTTP/2 stream resets instead of 502ing (#1645)

## Description

Under concurrent load with large request bodies, `/v1/messages` returns
**HTTP 502**. A single upstream HTTP/2 stream reset poisons the shared
h2 connection and raises `RemoteProtocolError` (`StreamReset`) /
`LocalProtocolError` on every other in-flight stream:

```
ERROR [hr_...] Request failed: RemoteProtocolError: <StreamReset stream_id:35, error_code:1, remote_reset:True>
ERROR [hr_...] Request failed: LocalProtocolError: 39
INFO  event=proxy_inbound_response ... status=502 duration_ms=78712
```

These are transport errors, but they weren't in the proxy's retry paths
— the non-streaming `_retry_request` caught `(ConnectError,
TimeoutException, HTTPStatusError)` and the streaming connect loop
caught `(ConnectError, ConnectTimeout, PoolTimeout)`. So a stream reset
skipped retry entirely and fell through to the broad handler catch as a
`502`, with no reconnect.

This broadens both retry paths to treat any `httpx.TransportError` —
which includes the h2 `Local`/`RemoteProtocolError` — as retryable, so
the poisoned connection is dropped and the request re-sent on a fresh
one.

Closes #1639

> Scope note: the issue also mentions `HEADROOM_HTTP2` being ignored on
the `headroom install agent run` launch path. That's a separate
config-plumbing gap; I've kept this PR to the 502-cascade fix (which
makes the chain self-recover regardless of the env workaround) and am
happy to follow up on the env plumbing separately.

## Type of Change

- [x] Bug fix (non-breaking change that fixes an issue)

## Changes Made

- `headroom/proxy/server.py` (`_retry_request`): the retry `except` now
catches `(httpx.TransportError, httpx.HTTPStatusError)` instead of
`(ConnectError, TimeoutException, HTTPStatusError)`. `TransportError` is
the common base of ConnectError, the timeout family, and the
protocol/network errors — so h2 stream resets are retried with backoff.
- `headroom/proxy/handlers/streaming.py`: the streaming connect-retry
loop and its terminal handler now catch `httpx.TransportError`. The
retry runs before any body byte is forwarded to the client (only
`build_request` + `send(stream=True)` are inside the loop), so
re-sending is safe. On exhaustion the terminal handler still emits a
clean `event: error` SSE instead of letting the reset bubble up as a
502. The mid-stream handler was left as-is (already covered by its
`except Exception`, and not safe to retry once bytes have been sent).

## Testing

- [x] Unit tests pass (`pytest`)
- [x] Linting passes (`ruff check .`)
- [x] Type checking passes (`mypy headroom`)
- [x] New tests added for new functionality
- [x] Manual testing performed

### Test Output

```text
$ pytest tests/test_h2_stream_reset_retry.py -q
4 passed

$ pytest tests/test_proxy_streaming_resilience.py tests/test_mid_turn_steering.py \
        tests/test_proxy_streaming_ratelimit_headers.py tests/test_streaming_usage_parser.py \
        tests/test_proxy_byte_faithful_forwarding.py -q
87 passed, 1 skipped

$ ruff check <changed files> && ruff format --check <changed files>
All checks passed! / 3 files already formatted

$ mypy headroom/proxy/server.py headroom/proxy/handlers/streaming.py --ignore-missing-imports
Success: no issues found in 2 source files
```

## Real Behavior Proof

- Environment: macOS (arm64), Python 3.14 venv, editable install of this
branch.
- Exact command / steps: ran `pytest
tests/test_h2_stream_reset_retry.py` — the tests drive the real
`_retry_request` and `_stream_response` with `http_client.post` /
`http_client.send` set to raise `httpx.RemoteProtocolError("<StreamReset
...>")` on the first attempt and return a good response on the second.
- Observed result: non-streaming — the request is retried and returns
the `200` response (`post` awaited twice); on unconditional resets it
re-raises after `retry_max_attempts` (no silent hang). Streaming — the
reset on `send()` is retried and the upstream SSE (`message_start`…) is
forwarded with no `connection_error` event (`send` awaited twice); on
repeated resets a clean `event: error` SSE is emitted rather than a
crash/502. Before this change the same `RemoteProtocolError` was
uncaught and propagated to the `502` handler.
- Not tested: a live 10-session concurrent-load repro against a real
Anthropic h2 endpoint — reproduced deterministically at the retry
boundary with an injected `RemoteProtocolError` instead.

## Review Readiness

- [x] I have performed a self-review
- [x] This PR is ready for human review

## Additional Notes

Retrying a stream reset re-sends the (potentially large) body, but that
is bounded by the existing `retry_max_attempts` + jittered backoff and
only happens before the first client byte — the same contract the
existing connect-error retry already relied on. This is complementary
to, not a replacement for, an operator forcing HTTP/1.1; it makes the
default h2 path self-heal from transient resets.

Co-authored-by: JerrettDavis <mxjerrett@gmail.com>

CHANGELOG.md

@@ -9,6 +9,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## Unreleased
 
 ### Fixed
+- Concurrent large requests no longer 502 on a transient HTTP/2 stream reset.
+  A single upstream `StreamReset` poisons the shared h2 connection and raises
+  `RemoteProtocolError` / `LocalProtocolError` on every in-flight request; those
+  transport errors weren't in the proxy's retry paths, so they collapsed
+  straight to a 502 with no reconnect. The Anthropic non-streaming and streaming
+  retry paths now treat any `httpx.TransportError` (including h2 protocol
+  errors) as retryable before the first client byte, so the bad connection
+  is dropped and the request re-sent on a fresh one
+  ([#1639](https://github.com/headroomlabs-ai/headroom/issues/1639)).
 - `headroom wrap claude` no longer leaves a dead `ANTHROPIC_BASE_URL` in a
   project's `.claude/settings.local.json` after an unclean exit (`SIGKILL`,
   OOM, reboot, or terminal/tmux close via `SIGHUP`, which was not caught).

headroom/proxy/handlers/streaming.py

@@ -1078,7 +1078,12 @@ class StreamingMixin:
                         await asyncio.sleep(delay_with_jitter / 1000)
                         continue
                     break
-                except (httpx.ConnectError, httpx.ConnectTimeout, httpx.PoolTimeout) as e:
+                # Retry any transport-level failure while opening the upstream
+                # stream — including HTTP/2 protocol errors (Local/RemoteProtocol
+                # `StreamReset`) from a poisoned shared h2 connection. This runs
+                # before any body byte is forwarded to the client, so re-sending
+                # on a fresh connection is safe and avoids a 502. (#1639)
+                except httpx.TransportError as e:
                     last_connect_error = e
                     if attempt >= retry_attempts - 1:
                         raise
@@ -1097,7 +1102,10 @@ class StreamingMixin:
 
             if upstream_response is None:
                 raise last_connect_error or RuntimeError("upstream connection did not start")
-        except (httpx.ConnectError, httpx.ConnectTimeout, httpx.PoolTimeout) as e:
+        # Retries exhausted (or a transport failure escaped the loop): emit a
+        # clean SSE error instead of letting an h2 StreamReset bubble up as a
+        # 502. Covers ConnectError/timeouts and Local/RemoteProtocolError. (#1639)
+        except httpx.TransportError as e:
             error_msg = str(e) or repr(e)
             logger.error(f"[{request_id}] Connection error to upstream API: {error_msg}")
 

headroom/proxy/server.py

@@ -1875,7 +1875,12 @@ class HeadroomProxy(
 
                     return response
 
-            except (httpx.ConnectError, httpx.TimeoutException, httpx.HTTPStatusError) as e:
+            # httpx.TransportError covers ConnectError, the timeout family, and —
+            # crucially — the protocol errors (Local/RemoteProtocolError, e.g. an
+            # HTTP/2 `StreamReset`) that a poisoned shared h2 connection raises on
+            # every in-flight request. Retrying drops the bad connection and
+            # re-sends on a fresh one instead of collapsing to a 502. (#1639)
+            except (httpx.TransportError, httpx.HTTPStatusError) as e:
                 last_error = e
 
                 if not self.config.retry_enabled or attempt >= self.config.retry_max_attempts - 1:

tests/test_h2_stream_reset_retry.py

@@ -0,0 +1,144 @@
+"""HTTP/2 stream-reset resilience (issue #1639).
+
+Under concurrent load a single upstream HTTP/2 stream reset poisons the shared
+h2 connection and surfaces as `RemoteProtocolError` / `LocalProtocolError` on
+every in-flight request. Those are transport errors, so the proxy must retry
+them (dropping the bad connection and re-sending on a fresh one) instead of
+collapsing to a 502. These tests drive the real `_retry_request` and
+`_stream_response` paths.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import AsyncMock, MagicMock
+
+import httpx
+import pytest
+
+from headroom.proxy.server import HeadroomProxy
+
+
+def _mock_proxy():
+    proxy = object.__new__(HeadroomProxy)
+    proxy.http_client = MagicMock(spec=httpx.AsyncClient)
+    proxy._config = MagicMock()
+    proxy._config.memory_enabled = False
+    proxy._config.ccr_inject_tool = False
+    proxy._config.retry_enabled = True
+    proxy._config.retry_max_attempts = 2
+    proxy._config.retry_base_delay_ms = 0
+    proxy._config.retry_max_delay_ms = 0
+    proxy.config = proxy._config
+    proxy.memory_handler = None
+    proxy._parse_sse_usage_from_buffer = MagicMock(return_value=None)
+    proxy._finalize_stream_response = AsyncMock(return_value=None)
+    return proxy
+
+
+def _good_stream_response(chunks):
+    resp = AsyncMock()
+    resp.headers = httpx.Headers({"content-type": "text/event-stream"})
+    resp.status_code = 200
+
+    async def aiter_bytes():
+        for chunk in chunks:
+            yield chunk
+
+    resp.aiter_bytes = aiter_bytes
+    resp.aclose = AsyncMock()
+    return resp
+
+
+async def _run_stream(proxy, session_key="k"):
+    return await proxy._stream_response(
+        url="https://api.anthropic.com/v1/messages",
+        headers={"x-api-key": "sk-test"},
+        body={
+            "model": "claude-sonnet-4-20250514",
+            "max_tokens": 100,
+            "stream": True,
+            "messages": [{"role": "user", "content": "hi"}],
+        },
+        provider="anthropic",
+        model="claude-sonnet-4-20250514",
+        request_id="test-1639",
+        original_tokens=10,
+        optimized_tokens=10,
+        tokens_saved=0,
+        transforms_applied=[],
+        tags={},
+        optimization_latency=0.0,
+        session_key=session_key,
+    )
+
+
+@pytest.mark.asyncio
+async def test_retry_request_retries_remote_protocol_error():
+    proxy = _mock_proxy()
+    good = MagicMock()
+    good.status_code = 200
+    good.request = MagicMock()
+    proxy.http_client.post = AsyncMock(
+        side_effect=[httpx.RemoteProtocolError("<StreamReset stream_id:35>"), good]
+    )
+
+    result = await proxy._retry_request(
+        "POST",
+        "https://api.anthropic.com/v1/messages",
+        {"x-api-key": "sk-test"},
+        {"model": "claude-sonnet-4-20250514", "messages": []},
+    )
+
+    assert result is good
+    assert proxy.http_client.post.await_count == 2
+
+
+@pytest.mark.asyncio
+async def test_retry_request_reraises_after_exhaustion():
+    proxy = _mock_proxy()
+    proxy.http_client.post = AsyncMock(side_effect=httpx.RemoteProtocolError("reset"))
+
+    with pytest.raises(httpx.RemoteProtocolError):
+        await proxy._retry_request(
+            "POST",
+            "https://api.anthropic.com/v1/messages",
+            {"x-api-key": "sk-test"},
+            {"model": "claude-sonnet-4-20250514", "messages": []},
+        )
+    assert proxy.http_client.post.await_count == 2
+
+
+@pytest.mark.asyncio
+async def test_stream_retries_h2_stream_reset_then_succeeds():
+    proxy = _mock_proxy()
+    good = _good_stream_response(
+        [
+            b'event: message_start\ndata: {"type":"message_start"}\n\n',
+            b'event: message_stop\ndata: {"type":"message_stop"}\n\n',
+        ]
+    )
+    proxy.http_client.build_request = MagicMock(return_value=MagicMock())
+    proxy.http_client.send = AsyncMock(
+        side_effect=[httpx.RemoteProtocolError("<StreamReset stream_id:35>"), good]
+    )
+
+    result = await _run_stream(proxy)
+    body = b"".join([chunk async for chunk in result.body_iterator])
+
+    assert proxy.http_client.send.await_count == 2
+    assert b"message_start" in body
+    assert b"connection_error" not in body
+
+
+@pytest.mark.asyncio
+async def test_stream_reset_exhaustion_yields_sse_error_not_crash():
+    proxy = _mock_proxy()
+    proxy.http_client.build_request = MagicMock(return_value=MagicMock())
+    proxy.http_client.send = AsyncMock(side_effect=httpx.RemoteProtocolError("reset"))
+
+    result = await _run_stream(proxy)
+    body = b"".join([chunk async for chunk in result.body_iterator])
+
+    assert proxy.http_client.send.await_count == 2
+    assert b"event: error" in body
+    assert b"connection_error" in body