fix(proxy): only queue mid-turn messages for opt-in clients with explicit session header (#1951)

## Description

Mid-turn steering wrongly queues **concurrent independent streams**.
When two streaming `/v1/messages` requests share the same model + system
prompt and arrive concurrently (no `x-headroom-session-id` header), the
proxy misclassifies the second as a "mid-turn message", returns `202
{"event":"headroom_queued"}`, and never forwards it upstream. A standard
Anthropic SDK client that made a *streaming* call receives a non-SSE 202
→ empty event stream → `AssertionError` (`assert
self.__final_message_snapshot is not None` in
`anthropic/lib/streaming/_messages.py`), and fails after retries.

**Root cause.** Without an `x-headroom-session-id` header,
`_get_session_key()` falls back to `md5(model + system[:500])` (mirrors
`prefix_tracker.compute_session_id`). That key is intentionally coarse
and cannot distinguish genuinely concurrent, independent streams that
share a model + system prompt (e.g. a main conversation plus its
background / parallel requests), so the second stream hits `session_key
in self._active_streams` and gets queued.

A queued message is only ever drained back to the client via the custom
`headroom_pending_messages` SSE event, which a standard Anthropic SDK
does not understand — so mid-turn steering is effectively a private
protocol for clients that **opt in** via `x-headroom-session-id`. A
client that never sends the header can never participate in the queue;
for it, the 202 is simply a broken streaming response.

Note: "send a unique header per request" is **not** a workaround — the
same header also drives `prefix_tracker.compute_session_id()`, so
unique-per-stream ids break prompt caching while a shared id keeps
colliding.

Closes #1949

## Type of Change

- [x] Bug fix (non-breaking change that fixes an issue)
- [ ] New feature (non-breaking change that adds functionality)
- [ ] Breaking change (fix or feature that would cause existing
functionality to change)
- [ ] Documentation update
- [ ] Performance improvement
- [ ] Code refactoring (no functional changes)

## Changes Made

- Add `StreamingMixin._should_queue_mid_turn()` helper that gates
mid-turn queuing behind an explicit `x-headroom-session-id` header.
- Header-less concurrent streams are now forwarded upstream normally;
only opt-in (header-bearing) callers can be queued.
- Prefix-tracker / cache-alignment behavior is untouched — the header
still drives `compute_session_id()` exactly as before.

## Testing

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

### Test Output

```text
$ pytest tests/test_mid_turn_steering.py
6 passed
```

New test `test_should_queue_only_with_explicit_session_header`: a
header-less concurrent stream must not queue; an explicit-header opt-in
must. All existing `test_mid_turn_steering.py` cases pass an explicit
header and are unaffected.

## Real Behavior Proof

- Environment: macOS, `headroom-ai` 0.30.0 (installed via `uv tool`),
proxy running `headroom proxy --port 8799 --no-http2 --mode cache`,
upstream = an Anthropic-compatible gateway. Client = Hermes Agent
(Anthropic SDK, streaming) driving a main conversation plus concurrent
background/parallel requests that share the same model + system prompt.
- Exact command / steps:
1. Reproduce on stock 0.30.0: concurrent streaming requests without
`x-headroom-session-id` → second stream returns `202
{"event":"headroom_queued"}` → client raises `AssertionError` in
`anthropic/lib/streaming/_messages.py`.
2. Correlate logs: count of `AssertionError` in the client error log vs
count of `202` in the proxy access log for the window — **48 == 48**,
timestamps line up 1:1.
3. Apply this patch to the running package, restart the proxy, re-run
the same concurrent workload.
- Observed result: after the fix, **0 × 202 / all requests 200**, no new
`AssertionError`, and `cache_hit_pct` stayed ~99% (prefix caching
intact). Header-bearing opt-in clients still queue mid-turn as before.
- Not tested: behavior under a client that deliberately sends a
*changing* `x-headroom-session-id` per request (out of scope —
documented as a caching anti-pattern, not a supported mode).

## Review Readiness

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

## Checklist

- [x] My code follows the project's style guidelines
- [x] I have performed a self-review of my code
- [x] I have commented my code, particularly in hard-to-understand areas
- [ ] I have made corresponding changes to the documentation
- [x] My changes generate no new warnings
- [x] I have added tests that prove my fix is effective or that my
feature works
- [x] New and existing unit tests pass locally with my changes
- [ ] I have updated the CHANGELOG.md if applicable

## Additional Notes

- Docs / CHANGELOG unchanged: this is a proxy-internal correctness fix
with no user-facing config surface.
- The fix is deliberately minimal and conservative — it only narrows
*when* queuing engages (explicit opt-in header), leaving the
prefix-tracker, cache-alignment, and body-rewrite paths byte-for-byte
identical.

---------

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

headroom/proxy/handlers/anthropic.py

@@ -2558,11 +2558,16 @@ class AnthropicHandlerMixin:
                         metadata={"path": pipeline_path, "stream": True},
                     )
                     await _finalize_pre_upstream()
+                    explicit_session_header = request.headers.get("x-headroom-session-id")
                     session_key = self._get_session_key(
                         body,
-                        session_header=request.headers.get("x-headroom-session-id"),
+                        session_header=explicit_session_header,
                     )
-                    if session_key in self._active_streams:
+                    # Only opt-in (header-bearing) callers participate in
+                    # mid-turn steering; see StreamingMixin._should_queue_mid_turn
+                    # for why the coarse md5 fallback must not queue concurrent
+                    # independent streams (it wrongly 202s a streaming caller).
+                    if self._should_queue_mid_turn(session_key, explicit_session_header):
                         from fastapi.responses import JSONResponse
 
                         queued = self._queue_mid_turn_message(session_key, body)

headroom/proxy/handlers/streaming.py

@@ -95,6 +95,27 @@ class StreamingMixin:
         self._mid_turn_queues[session_key].put_nowait(body)
         return {"status": 202, "event": "headroom_queued"}
 
+    def _should_queue_mid_turn(self, session_key: str, explicit_session_header: str | None) -> bool:
+        """Return True only when a follow-up should be queued as a mid-turn message.
+
+        Mid-turn steering is a private Headroom protocol: a queued message is
+        only ever drained back to the client via the custom
+        ``headroom_pending_messages`` SSE event, which a standard Anthropic SDK
+        does not understand. We therefore only engage it for clients that
+        *opt in* by sending an explicit ``x-headroom-session-id`` header.
+
+        Without that header the session identity falls back to
+        ``md5(model + system[:500])`` (see ``_get_session_key`` /
+        ``prefix_tracker.compute_session_id``). That fallback is intentionally
+        coarse and cannot distinguish genuinely concurrent, independent streams
+        that happen to share a model + system prompt (e.g. a main conversation
+        plus its background/parallel requests). Queuing those as mid-turn
+        messages wrongly returns a 202 + JSON body to a caller that issued a
+        *streaming* request, whose stream parser then sees an empty (non-SSE)
+        stream and fails. So only opt-in (header-bearing) callers get queued.
+        """
+        return bool(explicit_session_header) and session_key in self._active_streams
+
     def _cleanup_mid_turn_stream(
         self, session_key: str, *, drain_pending_messages: bool = False
     ) -> list[dict]:

tests/test_mid_turn_steering.py

@@ -42,6 +42,34 @@ class TestMidTurnSteering:
         assert session_key not in mixin._active_streams
         assert session_key not in mixin._mid_turn_queues
 
+    def test_should_queue_only_with_explicit_session_header(self):
+        """Regression: mid-turn queuing must require an explicit session header.
+
+        Without ``x-headroom-session-id`` the session key is a coarse
+        ``md5(model + system[:500])`` shared by concurrent independent streams
+        (e.g. a main conversation plus background/parallel requests). Queuing
+        those wrongly returns a 202 to a streaming caller, whose SDK stream
+        parser then fails on an empty (non-SSE) stream. Only opt-in callers
+        that send the header may be queued.
+        """
+        from headroom.proxy.handlers.streaming import StreamingMixin
+
+        mixin = StreamingMixin()
+        session_key = "shared-md5-key"
+        # An earlier stream on this (fallback) session key is in flight.
+        mixin._active_streams.add(session_key)
+        try:
+            # No explicit header (header-less concurrent stream): must NOT queue,
+            # even though the key collides in _active_streams.
+            assert mixin._should_queue_mid_turn(session_key, None) is False
+            assert mixin._should_queue_mid_turn(session_key, "") is False
+            # Explicit header present: opt-in client, queuing is allowed.
+            assert mixin._should_queue_mid_turn(session_key, session_key) is True
+            # Explicit header but no active stream: nothing to queue behind.
+            assert mixin._should_queue_mid_turn("other-key", "other-key") is False
+        finally:
+            mixin._active_streams.discard(session_key)
+
     def _create_mock_proxy(self):
         proxy = object.__new__(HeadroomProxy)
         proxy.http_client = MagicMock(spec=httpx.AsyncClient)