From 4dab254d52914c39ffe13071848604e1771b1bd1 Mon Sep 17 00:00:00 2001 From: Focused Instability <70747559+MrAshRhodes@users.noreply.github.com> Date: Wed, 5 Aug 2026 04:41:58 +0200 Subject: [PATCH] fix: emit SSE ping before message_start on Bedrock streaming path (issue #902) (#1080) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Description Closes #902 Mid-turn user interjections (steering) silently dropped through the Bedrock streaming path. The _stream_response_bedrock code path reconstructs Anthropic SSE events from parsed StreamEvent objects instead of passing raw bytes through, so SSE-level ping keepalives are never forwarded to Claude Code. Claude Code relies on ping events to arm its mid-turn steering / interruptible state; without them, queued interjections are discarded instead of sent. Root cause (confirmed): - Standard direct-Anthropic path does a raw yield-chunk passthrough — pings flow unchanged. - Bedrock path (_stream_response_bedrock.generate()) reconstructs events from litellm/anyllm stream_message() output, which only yields semantic events (message_start, content_block_*, message_delta, message_stop, error). No pings, ever. Fix: emit a synthetic 'event: ping / data: {}' at stream start (before the first message_start) so downstream clients see the same ping-then-content cadence as a real Anthropic stream. Note: periodic pings for very long responses (>~25s) may be needed if steering disarms on a timer. This commit arms it at turn start; follow-up if reporters confirm steering still drops on long turns. The causal link (ping → steering) is the reporter's hypothesis from hands-on debugging. The observable defect (zero pings in stream) is confirmed and fixed. ## Type of Change - [x] Bug fix (non-breaking change that fixes an issue) ## Changes Made - headroom/proxy/handlers/streaming.py: yield ping event before the event loop in _stream_response_bedrock.generate() - tests/test_proxy/test_bedrock_sse_ping.py: 3 new tests asserting ping appears before message_start ## Testing - [x] Unit tests pass (pytest) - [x] Linting passes (ruff check .) - [x] New tests added for new functionality ### Test Output ``` tests/test_proxy/test_bedrock_sse_ping.py::test_bedrock_stream_emits_ping_before_message_start PASSED tests/test_proxy/test_bedrock_sse_ping.py::test_bedrock_stream_ping_has_empty_data PASSED tests/test_proxy/test_bedrock_sse_ping.py::test_bedrock_stream_contains_message_stop PASSED tests/test_backend_streaming_cache_metrics.py (4 tests) PASSED 7 passed in 4.41s ``` ## Real Behavior Proof - Environment: macOS, Python 3.11, headroom unit tests - Exact command / steps: pytest tests/test_proxy/test_bedrock_sse_ping.py -v - Observed result: 3 new tests pass; ping appears before message_start in Bedrock stream - Not tested: end-to-end against live Bedrock + Claude Code (no Bedrock credentials available) ## Review Readiness - [x] I have performed a self-review - [x] This PR is ready for human review ## Checklist - [x] Code follows project style guidelines - [x] Code is commented where non-obvious - [x] No new warnings - [x] Tests added and passing ## Additional Notes The Rust proxy files mentioned in the issue (sse/framing.rs, sse/anthropic.rs) are NOT part of this fix. Those drops are in a telemetry-only tee task that never affects the client byte path — the Rust proxy does a raw bytes passthrough for all responses. The defect is Python-only, confined to the Bedrock path. --------- Co-authored-by: JerrettDavis --- headroom/proxy/handlers/streaming.py | 8 ++ tests/test_proxy/test_bedrock_sse_ping.py | 153 ++++++++++++++++++++++ 2 files changed, 161 insertions(+) create mode 100644 tests/test_proxy/test_bedrock_sse_ping.py diff --git a/headroom/proxy/handlers/streaming.py b/headroom/proxy/handlers/streaming.py index 7ce93a9fb..fd232dc52 100644 --- a/headroom/proxy/handlers/streaming.py +++ b/headroom/proxy/handlers/streaming.py @@ -1723,6 +1723,14 @@ class StreamingMixin: try: assert self.anthropic_backend is not None + # Emit a synthetic ping before the first message_start so that + # downstream clients (e.g. Claude Code) arm their mid-turn + # steering / interruptible state. The Bedrock-to-Anthropic + # translation layer never produces SSE-level keepalives; we + # synthesise one here to match the real Anthropic wire format + # (issue #902). + yield b"event: ping\ndata: {}\n\n" + async for event in self.anthropic_backend.stream_message(body, headers): # Record TTFB on first event if stream_state["ttfb_ms"] is None: diff --git a/tests/test_proxy/test_bedrock_sse_ping.py b/tests/test_proxy/test_bedrock_sse_ping.py new file mode 100644 index 000000000..b865b6ad3 --- /dev/null +++ b/tests/test_proxy/test_bedrock_sse_ping.py @@ -0,0 +1,153 @@ +"""SSE ping passthrough in the Bedrock streaming path (issue #902). + +When Headroom routes through a Bedrock (LiteLLM/AnyLLM) backend the +translation layer emits only Anthropic-semantic events — it never produces +SSE-level ping keepalives. Claude Code uses ping events to keep a turn in +the interruptible / steering-armed state; without them, mid-turn interjections +are silently dropped. + +Fix: ``_stream_response_bedrock.generate()`` now emits one synthetic +``event: ping\\ndata: {}\\n\\n`` before the first ``message_start`` so the +downstream client sees the same ping-then-content cadence as a real Anthropic +stream. +""" + +from __future__ import annotations + +from collections.abc import AsyncIterator +from typing import Any +from unittest.mock import MagicMock, patch + +import pytest + +fastapi = pytest.importorskip("fastapi") + +from fastapi.testclient import TestClient # noqa: E402 + +from headroom.backends.base import StreamEvent # noqa: E402 +from headroom.proxy.server import ProxyConfig, create_app # noqa: E402 + + +def _ev(event_type: str, data: dict[str, Any]) -> StreamEvent: + return StreamEvent(event_type=event_type, data=data) + + +def _minimal_events() -> list[StreamEvent]: + return [ + _ev( + "message_start", + { + "type": "message_start", + "message": { + "id": "msg_test", + "model": "claude-3-5-sonnet-20241022", + "role": "assistant", + "type": "message", + "content": [], + "usage": {"input_tokens": 10, "output_tokens": 0}, + }, + }, + ), + _ev( + "content_block_start", + { + "type": "content_block_start", + "index": 0, + "content_block": {"type": "text", "text": ""}, + }, + ), + _ev( + "content_block_delta", + { + "type": "content_block_delta", + "index": 0, + "delta": {"type": "text_delta", "text": "hi"}, + }, + ), + _ev("content_block_stop", {"type": "content_block_stop", "index": 0}), + _ev( + "message_delta", + { + "type": "message_delta", + "delta": {"stop_reason": "end_turn"}, + "usage": {"output_tokens": 1}, + }, + ), + _ev("message_stop", {"type": "message_stop"}), + ] + + +def _make_bedrock_backend(events: list[StreamEvent]) -> MagicMock: + async def fake_stream(body: dict, headers: dict) -> AsyncIterator[StreamEvent]: + for evt in events: + yield evt + + mock = MagicMock() + mock.name = "bedrock" + mock.stream_message = fake_stream + mock.map_model_id = MagicMock(return_value="claude-3-5-sonnet-20241022") + mock.supports_model = MagicMock(return_value=True) + return mock + + +def _run_bedrock_stream(events: list[StreamEvent]) -> str: + """Return the full SSE response body from a Bedrock-backend streaming request.""" + config = ProxyConfig( + optimize=False, + cache_enabled=False, + rate_limit_enabled=False, + backend="anyllm", + anyllm_provider="anthropic", + ) + backend = _make_bedrock_backend(events) + with patch("headroom.proxy.server.AnyLLMBackend", return_value=backend): + app = create_app(config) + with TestClient(app) as client: + resp = client.post( + "/v1/messages", + json={ + "model": "claude-3-5-sonnet-20241022", + "messages": [{"role": "user", "content": "hi"}], + "max_tokens": 64, + "stream": True, + }, + headers={ + "x-api-key": "sk-ant-test", + "anthropic-version": "2023-06-01", + }, + ) + assert resp.status_code == 200, resp.text[:200] + return resp.text + + +def test_bedrock_stream_emits_ping_before_message_start() -> None: + """Bedrock path must emit a ping event before message_start (issue #902).""" + body = _run_bedrock_stream(_minimal_events()) + + ping_idx = body.find("event: ping") + start_idx = body.find("event: message_start") + + assert ping_idx != -1, "No ping event found in Bedrock stream response" + assert start_idx != -1, "No message_start event found in Bedrock stream response" + assert ping_idx < start_idx, ( + f"ping (offset {ping_idx}) must appear before message_start (offset {start_idx})" + ) + + +def test_bedrock_stream_ping_has_empty_data() -> None: + """Ping event must carry data: {} to match real Anthropic wire format.""" + body = _run_bedrock_stream(_minimal_events()) + + ping_start = body.find("event: ping") + assert ping_start != -1, "No ping event found" + + # The next ~30 bytes after 'event: ping' should contain 'data: {}' + ping_block = body[ping_start : ping_start + 40] + assert "data: {}" in ping_block, f"Ping block must contain 'data: {{}}', got: {ping_block!r}" + + +def test_bedrock_stream_contains_message_stop() -> None: + """Smoke test: full event sequence still reaches the client alongside ping.""" + body = _run_bedrock_stream(_minimal_events()) + assert "event: message_stop" in body + assert "event: message_start" in body