fix(proxy): handle ClientDisconnect in passthrough body reads (#2033)

## Description

Catch `starlette.requests.ClientDisconnect` when reading request bodies
in passthrough/forwarding handlers. Closes #2019

Without this, a client that disconnects mid-request causes an unhandled
`ClientDisconnect` to propagate through the entire middleware stack,
crashing the ASGI TaskGroup and contributing to proxy instability over
long sessions (memory growth, freeze, unresponsive to SIGTERM).

**Adversarial review uncovered 3 additional unprotected sites** in
`proxy_routes.py` — same pattern (body read before try/except). Now
fixed.

## Type of Change

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

## Changes Made

**Proxy handlers** (6 sites, first commit):
- openai `handle_passthrough`: wrap `await request.body()` in try/except
ClientDisconnect (main crash site)
- openai `_handle_streaming_passthrough`: same protection
- anthropic batch passthrough: same protection
- batch `_google_batch_passthrough`: same protection
- batch `handle_google_batch_passthrough`: same protection
- bedrock fallback-forward path: early-return on ClientDisconnect
instead of attempting verbatim forward

**Proxy routes** (3 sites, second commit — found by adversarial design
scan):
- `_handle_chatgpt_model_metadata` (proxy_routes.py:398)
- `_handle_chatgpt_codex_images` (proxy_routes.py:438)
- `openai_responses_sub` nested handler (proxy_routes.py:597)

All nine sites return HTTP 204 on disconnect to allow the request to
terminate cleanly.

## Testing

- [x] **Existing tests**: 34/34 pass in `test_proxy_handler_helpers.py`
- [x] **Unit tests**: 2 new tests — passthrough + streaming passthrough
disconnect
- [x] **Adversarial concurrency**: 50 threads × 10 iterations = 500
concurrent disconnect requests — zero crashes, all return 204
- [x] **Adversarial edge cases**: minimal request state, regression
check (normal request path unaffected)
- [x] **PBT (Hypothesis)**: 250 random method/path combinations, 3
properties verified:
  - All disconnect requests return 204
  - ClientDisconnect never leaks out of handler
  - Response is always valid HTTP 2xx

```text
# Unit tests
tests/test_proxy_handler_helpers.py::test_handle_passthrough_client_disconnect PASSED
tests/test_proxy_handler_helpers.py::test_handle_streaming_passthrough_client_disconnect PASSED

# PBT (3 properties × 100-250 examples each)
/tmp/pbt_client_disconnect.py::test_disconnect_always_returns_204 PASSED
/tmp/pbt_client_disconnect.py::test_disconnect_does_not_crash_asgi PASSED
/tmp/pbt_client_disconnect.py::test_response_is_valid_http PASSED

# Adversarial
/tmp/adversarial_client_disconnect.py → 500 concurrent requests: 0 errors, all 204
```

- [x] `ruff check` and `ruff format --check` pass on all changed files

## Real Behavior Proof

- Environment: Linux, Python 3.12, headroom main @ a617455
- Exact command / steps: 
  - `uv run pytest tests/test_proxy_handler_helpers.py -v` — 34 passed
- `uv run python /tmp/adversarial_client_disconnect.py` — 500
concurrent, 0 errors
- `uv run python /tmp/pbt_client_disconnect.py` — 250 random inputs, 3/3
properties hold
- `uv run ruff check . && uv run ruff format --check .` — All checks
passed
- Observed result: ClientDisconnect caught gracefully at all 9 sites,
204 returned, no ExceptionGroup crash, no data corruption
- Not tested: Full E2E with real client disconnect (requires integration
test infrastructure). Manual confirmation from issue reporter would
validate the real-world fix.

## Review Readiness

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

---------

Co-authored-by: lennney <lennney@users.noreply.github.com>

headroom/providers/proxy_routes.py

@@ -395,7 +395,13 @@ async def _handle_chatgpt_model_metadata(
     if request.url.query:
         url = f"{url}?{request.url.query}"
 
-    body = await request.body()
+    from starlette.requests import ClientDisconnect
+
+    try:
+        body = await request.body()
+    except ClientDisconnect:
+        logger.debug("Client disconnected during body read for passthrough")
+        return Response(status_code=204)
     try:
         assert proxy.http_client is not None
         resp = await proxy.http_client.request(
@@ -435,7 +441,13 @@ async def _handle_chatgpt_codex_images(
     if request.url.query:
         url = f"{url}?{request.url.query}"
 
-    body = await request.body()
+    from starlette.requests import ClientDisconnect
+
+    try:
+        body = await request.body()
+    except ClientDisconnect:
+        logger.debug("Client disconnected during body read for passthrough")
+        return Response(status_code=204)
     try:
         client = getattr(proxy, "http_client_h1", None) or getattr(proxy, "http_client", None)
         if client is None:
@@ -594,7 +606,13 @@ def register_provider_routes(app: FastAPI, proxy: Any) -> None:
         if request.url.query:
             url = f"{url}?{request.url.query}"
 
-        body = await request.body()
+        from starlette.requests import ClientDisconnect
+
+        try:
+            body = await request.body()
+        except ClientDisconnect:
+            logger.debug("Client disconnected during body read for codex responses passthrough")
+            return Response(status_code=204)
         try:
             assert proxy.http_client is not None
             resp = await proxy.http_client.request(

headroom/proxy/handlers/anthropic.py

@@ -3632,7 +3632,13 @@ class AnthropicHandlerMixin:
             request_id=None,
         )
 
-        body = await request.body()
+        from starlette.requests import ClientDisconnect
+
+        try:
+            body = await request.body()
+        except ClientDisconnect:
+            logger.debug("Client disconnected during body read for anthropic batch passthrough")
+            return Response(status_code=204)
 
         response = await self.http_client.request(  # type: ignore[union-attr]
             method=request.method,

headroom/proxy/handlers/batch.py

@@ -393,7 +393,13 @@ class BatchHandlerMixin:
         from headroom.proxy.helpers import log_outbound_request
 
         if body is None:
-            body_content = await request.body()
+            from starlette.requests import ClientDisconnect
+
+            try:
+                body_content = await request.body()
+            except ClientDisconnect:
+                logger.debug("Client disconnected during body read for google batch passthrough")
+                return Response(status_code=204)
             outbound_source = "passthrough"
             body_mutated = False
         else:
@@ -502,7 +508,13 @@ class BatchHandlerMixin:
             else:
                 url = f"{url}?key={api_key}"
 
-        body = await request.body()
+        from starlette.requests import ClientDisconnect
+
+        try:
+            body = await request.body()
+        except ClientDisconnect:
+            logger.debug("Client disconnected during body read for gemini passthrough")
+            return Response(status_code=204)
 
         response = await self.http_client.request(  # type: ignore[union-attr]
             method=request.method,

headroom/proxy/handlers/bedrock.py

@@ -129,6 +129,11 @@ class BedrockHandlerMixin:
         try:
             body, raw = await read_request_json_with_bytes(request)
         except Exception as err:
+            from starlette.requests import ClientDisconnect
+
+            if isinstance(err, ClientDisconnect):
+                logger.debug("[%s] %s client disconnected during body read", request_id, LOG_TAG)
+                return Response(status_code=204)
             logger.warning(
                 "[%s] %s could not parse body; forwarding verbatim: %s",
                 request_id,

headroom/proxy/handlers/openai.py

@@ -7400,7 +7400,13 @@ class OpenAIHandlerMixin:
             request_id=None,
         )
 
-        body = await request.body()
+        from starlette.requests import ClientDisconnect
+
+        try:
+            body = await request.body()
+        except ClientDisconnect:
+            logger.debug("Client disconnected during body read for passthrough")
+            return Response(status_code=204)
 
         headers = await apply_copilot_api_auth(headers, url=url)
         # Cloudflare bot-management challenges our HTTP/2 fingerprint on
@@ -7575,7 +7581,13 @@ class OpenAIHandlerMixin:
             request_id=None,
         )
 
-        body = await request.body()
+        from starlette.requests import ClientDisconnect
+
+        try:
+            body = await request.body()
+        except ClientDisconnect:
+            logger.debug("Client disconnected during body read for streaming passthrough")
+            return Response(status_code=204)
         headers = await apply_copilot_api_auth(headers, url=url)
         request_id = await self._next_request_id()
         stream_provider = "gemini" if provider == "vertex:google" else "anthropic"

tests/test_proxy_handler_helpers.py

@@ -1006,3 +1006,55 @@ def test_strict_frozen_count_tool_and_function_tail_are_mutable():
         )
         == 3
     )
+
+
+class _ClientDisconnectRequest:
+    """Mock request whose body() raises ClientDisconnect to simulate mid-stream cancel."""
+
+    method = "POST"
+    headers = {"content-type": "application/json"}
+    url = SimpleNamespace(path="/v1/chat/completions", query="")
+
+    async def body(self) -> bytes:
+        from starlette.requests import ClientDisconnect
+
+        raise ClientDisconnect()
+
+
+class _ClientDisconnectStreamRequest:
+    """Mock request for streaming passthrough with ClientDisconnect."""
+
+    method = "POST"
+    headers = {"content-type": "application/json"}
+    url = SimpleNamespace(
+        path="/v1/projects/p/locations/us-central1/publishers/google/models/gemini-2.0-flash:streamGenerateContent",
+        query="alt=sse",
+    )
+
+    async def body(self) -> bytes:
+        from starlette.requests import ClientDisconnect
+
+        raise ClientDisconnect()
+
+
+def test_handle_passthrough_client_disconnect():
+    """ClientDisconnect during body read returns 204 instead of crashing TaskGroup."""
+    handler = object.__new__(OpenAIHandlerMixin)
+    response = asyncio.run(
+        handler.handle_passthrough(_ClientDisconnectRequest(), "https://api.openai.com")
+    )
+    assert response.status_code == 204
+
+
+def test_handle_streaming_passthrough_client_disconnect():
+    """ClientDisconnect during streaming body read returns 204."""
+    handler = object.__new__(OpenAIHandlerMixin)
+    response = asyncio.run(
+        handler.handle_passthrough(
+            _ClientDisconnectStreamRequest(),
+            "https://us-central1-aiplatform.googleapis.com",
+            endpoint_name="streamRawPredict",
+            provider="vertex:google",
+        )
+    )
+    assert response.status_code == 204