diff --git a/CHANGELOG.md b/CHANGELOG.md index ff255c832..64f8a8757 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -104,6 +104,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 * **cli:** the startup banner no longer advertises `HEADROOM_COMPRESSION_STABLE_AFTER_TURN` and `HEADROOM_STALE_READ_COMPRESS_AFTER_TURNS` as tuning knobs. Both were read only to render the `Performance Tuning` banner section and were never wired into the compression path, so setting them changed the banner but had no effect on behavior. The banner now surfaces only the embedding sidecar, which is a real, consumed setting. * **memory/embedder:** cap CPU thread oversubscription in the local torch/sentence-transformers embedder. Concurrent encodes previously each fanned out to ~`os.cpu_count()` BLAS/OpenMP threads, so under load the memory path starved the asyncio event loop and spiked `/livez` latency to several seconds. CPU encodes now run on a dedicated, size-limited executor whose workers each pin their thread pool, bounding total embedding threads to `HEADROOM_EMBED_CONCURRENCY` × `HEADROOM_EMBED_NUM_THREADS` (defaults `min(4, cpu)` × 1). The ONNX embedder already capped its threads; this brings the torch path to parity ([#198](https://github.com/headroomlabs-ai/headroom/issues/198)). * **proxy:** Buffered passthrough routes (e.g. `GET /v1/models`) no longer return an opaque HTTP 502 when an OpenAI-compatible upstream closes a pooled keep-alive connection mid-response (`httpx.RemoteProtocolError` / "incomplete chunked read"). Headroom now retries the request once on a fresh connection — mirroring a direct `curl` — and only returns a clear `upstream_protocol_error` 502 if the upstream is genuinely sending an incomplete response ([#1112](https://github.com/chopratejas/headroom/issues/1112)). +* **ccr:** buffered Anthropic CCR re-streaming now preserves adaptive-thinking response shape, including empty `thinking` blocks, `signature_delta`, `redacted_thinking.data`, verbatim `stop_reason` values such as `refusal`, and `stop_details`. * **cursor:** `headroom wrap cursor` no longer injects the `rtk` custom-instructions block into `.cursorrules` when rtk's own native Cursor hook registers successfully. rtk supports a real hook for Cursor via `rtk init --agent cursor` (the same mechanism headroom already uses for Claude Code), which rewrites shell commands transparently — the injected `.cursorrules` text duplicated that guidance for no benefit. `wrap cursor` now tries the native hook first and only falls back to injecting `.cursorrules` if hook registration fails (#756). * **proxy:** The Headroom dashboard no longer tunnels `GET /favicon.ico` to the wrapped upstream provider. No route matched that path, so it fell through to the proxy's catch-all passthrough route and was forwarded to the configured Anthropic/OpenAI/etc. backend — burning a real upstream request (and possibly failing auth) for a browser's automatic favicon fetch on `/dashboard`. A dedicated `/favicon.ico` route now answers with `204 No Content` directly, registered ahead of the passthrough catch-all (#1787). diff --git a/headroom/ccr/response_handler.py b/headroom/ccr/response_handler.py index 05112d0da..ade3fb5da 100644 --- a/headroom/ccr/response_handler.py +++ b/headroom/ccr/response_handler.py @@ -734,62 +734,103 @@ class StreamingCCRHandler: "usage": {}, } - current_text = "" - current_tool: dict[str, Any] | None = None + blocks_by_index: dict[int, dict[str, Any]] = {} + current_block: dict[str, Any] | None = None for event in events: event_type = event.get("type", "") if event_type == "content_block_start": block = event.get("content_block", {}) - if block.get("type") == "text": - current_text = block.get("text", "") - elif block.get("type") == "tool_use": - current_tool = { - "type": "tool_use", - "id": block.get("id", ""), - "name": block.get("name", ""), - "input": {}, - } - - elif event_type == "content_block_delta": - delta = event.get("delta", {}) - if delta.get("type") == "text_delta": - current_text += delta.get("text", "") - elif delta.get("type") == "input_json_delta": - # Accumulate JSON for tool input - if current_tool is not None: - partial = delta.get("partial_json", "") - # This is tricky - partial JSON needs accumulation - # For simplicity, we'll try to parse when complete - current_tool["_partial_json"] = ( - current_tool.get("_partial_json", "") + partial - ) - - elif event_type == "content_block_stop": - if current_text: - response["content"].append( + block_index = event.get("index", len(blocks_by_index)) + btype = block.get("type") + current_block = {"type": btype} + if btype == "text": + current_block["text"] = block.get("text", "") + elif btype == "tool_use": + current_block.update( { - "type": "text", - "text": current_text, + "id": block.get("id", ""), + "name": block.get("name", ""), + "input": {}, } ) - current_text = "" - if current_tool: - # Parse accumulated JSON - partial = current_tool.pop("_partial_json", "") - if partial: - try: - current_tool["input"] = json.loads(partial) - except json.JSONDecodeError: - current_tool["input"] = {} - response["content"].append(current_tool) - current_tool = None + elif btype == "thinking": + current_block["thinking_buffer"] = block.get("thinking", "") + if "signature" in block: + current_block["signature"] = block["signature"] + elif btype == "redacted_thinking": + if "data" in block: + current_block["data"] = block["data"] + elif btype: + current_block = dict(block) + blocks_by_index[block_index] = current_block + + elif event_type == "content_block_delta": + idx = event.get("index") + target = (blocks_by_index.get(idx) if idx is not None else None) or current_block + if target is None: + continue + delta = event.get("delta", {}) + dtype = delta.get("type") + if dtype == "text_delta": + target["text"] = target.get("text", "") + delta.get("text", "") + elif dtype == "input_json_delta": + if target.get("type") == "tool_use": + partial = delta.get("partial_json", "") + target["_partial_json"] = target.get("_partial_json", "") + partial + elif dtype == "thinking_delta": + target["thinking_buffer"] = target.get("thinking_buffer", "") + delta.get( + "thinking", "" + ) + elif dtype == "signature_delta": + if "signature" in delta: + target["signature"] = delta["signature"] + elif dtype == "citations_delta": + citation = delta.get("citation") + if citation is not None: + target.setdefault("citations", []).append(citation) + + elif event_type == "content_block_stop": + idx = event.get("index") + target = (blocks_by_index.get(idx) if idx is not None else None) or current_block + if target is not None: + if target.get("type") == "tool_use" and "_partial_json" in target: + partial = target.pop("_partial_json", "") + if partial: + try: + target["input"] = json.loads(partial) + except json.JSONDecodeError: + target["input"] = {} + if target.get("type") == "thinking" and "thinking_buffer" in target: + target["thinking"] = target.pop("thinking_buffer") + if target not in response["content"]: + response["content"].append(target) + current_block = None + + elif event_type == "message_start": + msg = event.get("message", {}) + if "id" in msg: + response["id"] = msg["id"] + if "model" in msg: + response["model"] = msg["model"] + if "role" in msg: + response["role"] = msg["role"] + if "stop_reason" in msg: + response["stop_reason"] = msg["stop_reason"] + if "stop_details" in msg: + response["stop_details"] = msg["stop_details"] + if msg.get("usage"): + response["usage"].update(msg["usage"]) elif event_type == "message_delta": delta = event.get("delta", {}) if "stop_reason" in delta: response["stop_reason"] = delta["stop_reason"] + if "stop_details" in delta: + response["stop_details"] = delta["stop_details"] + if event.get("usage"): + response["usage"].update(event["usage"]) elif event_type == "message_stop": pass @@ -859,11 +900,10 @@ class StreamingCCRHandler: to chunk the response more granularly. """ if self.provider == "anthropic": - # Anthropic SSE format - yield b"event: message_start\n" - yield f"data: {json.dumps({'type': 'message_start', 'message': response})}\n\n".encode() - yield b"event: message_stop\n" - yield b'data: {"type": "message_stop"}\n\n' + from headroom.proxy.handlers.streaming import StreamingMixin + + for chunk in StreamingMixin()._response_to_sse(response, "anthropic"): + yield chunk else: # OpenAI SSE format yield f"data: {json.dumps(response)}\n\n".encode() diff --git a/headroom/proxy/handlers/streaming.py b/headroom/proxy/handlers/streaming.py index 27b51fa0f..51efd402e 100644 --- a/headroom/proxy/handlers/streaming.py +++ b/headroom/proxy/handlers/streaming.py @@ -384,6 +384,8 @@ class StreamingMixin: response["model"] = msg.get("model") response["role"] = msg.get("role", "assistant") response["stop_reason"] = msg.get("stop_reason") + if "stop_details" in msg: + response["stop_details"] = msg["stop_details"] if msg.get("usage"): response["usage"].update(msg["usage"]) @@ -482,8 +484,10 @@ class StreamingMixin: elif event_type == "message_delta": delta = data.get("delta", {}) - if delta.get("stop_reason"): + if "stop_reason" in delta: response["stop_reason"] = delta["stop_reason"] + if "stop_details" in delta: + response["stop_details"] = delta["stop_details"] if data.get("usage"): response["usage"].update(data["usage"]) @@ -627,9 +631,14 @@ class StreamingMixin: events.append(f"event: content_block_stop\ndata: {json.dumps(block_stop)}\n\n".encode()) # message_delta + msg_delta_payload: dict[str, Any] = {} + if "stop_reason" in response: + msg_delta_payload["stop_reason"] = response["stop_reason"] + if "stop_details" in response: + msg_delta_payload["stop_details"] = response["stop_details"] msg_delta = { "type": "message_delta", - "delta": {"stop_reason": response.get("stop_reason", "end_turn")}, + "delta": msg_delta_payload, "usage": {"output_tokens": response.get("usage", {}).get("output_tokens", 0)}, } events.append(f"event: message_delta\ndata: {json.dumps(msg_delta)}\n\n".encode()) diff --git a/tests/test_ccr_response_handler_extra.py b/tests/test_ccr_response_handler_extra.py index 68afa16a7..c85e4dd94 100644 --- a/tests/test_ccr_response_handler_extra.py +++ b/tests/test_ccr_response_handler_extra.py @@ -30,6 +30,19 @@ async def _async_iter(items: list[bytes]): yield item +def _sse_json_events(chunks: list[bytes]) -> list[dict[str, Any]]: + events = [] + for chunk in chunks: + for line in chunk.decode("utf-8").splitlines(): + if not line.startswith("data: "): + continue + payload = line[len("data: ") :] + if payload == "[DONE]": + continue + events.append(json.loads(payload)) + return events + + def test_extract_tool_calls_google_and_invalid_shapes() -> None: handler = CCRResponseHandler() google_response = { @@ -202,23 +215,43 @@ def test_streaming_buffer_and_parse_sse_helpers() -> None: # Per SSE spec each event is terminated by `\n\n`. The byte-buffer # parser introduced in PR-A8 requires the spec terminator so partial # multi-byte UTF-8 reads don't corrupt event boundaries. + stop_details = {"type": "refusal", "message": "policy refusal"} anthropic_data = b"\n\n".join( [ + b'data: {"type":"message_start","message":{"id":"msg_1","model":"claude-fable-5","role":"assistant","usage":{"input_tokens":7}}}', + b'data: {"type":"content_block_start","index":0,"content_block":{"type":"thinking","thinking":""}}', + b'data: {"type":"content_block_delta","index":0,"delta":{"type":"thinking_delta","thinking":""}}', + b'data: {"type":"content_block_delta","index":0,"delta":{"type":"signature_delta","signature":"sig_fable"}}', + b'data: {"type":"content_block_stop","index":0}', + b'data: {"type":"content_block_start","index":1,"content_block":{"type":"redacted_thinking","data":"ENC:abc"}}', + b'data: {"type":"content_block_stop","index":1}', b'data: {"type":"content_block_start","content_block":{"type":"text","text":"Hel"}}', b'data: {"type":"content_block_delta","delta":{"type":"text_delta","text":"lo"}}', b'data: {"type":"content_block_stop"}', b'data: {"type":"content_block_start","content_block":{"type":"tool_use","id":"tool_1","name":"headroom_retrieve"}}', b'data: {"type":"content_block_delta","delta":{"type":"input_json_delta","partial_json":"{\\"hash\\":\\"abc\\"}"}}', b'data: {"type":"content_block_stop"}', - b'data: {"type":"message_delta","delta":{"stop_reason":"tool_use"}}', + ( + b'data: {"type":"message_delta","delta":{"stop_reason":"refusal",' + b'"stop_details":{"type":"refusal","message":"policy refusal"}},' + b'"usage":{"output_tokens":3}}' + ), b"data: [DONE]\n\n", ] ) parsed = handler._parse_sse_stream(anthropic_data) - assert parsed["content"][0] == {"type": "text", "text": "Hello"} - assert parsed["content"][1]["name"] == "headroom_retrieve" - assert parsed["content"][1]["input"] == {"hash": "abc"} - assert parsed["stop_reason"] == "tool_use" + assert parsed["content"][0] == { + "type": "thinking", + "signature": "sig_fable", + "thinking": "", + } + assert parsed["content"][1] == {"type": "redacted_thinking", "data": "ENC:abc"} + assert parsed["content"][2] == {"type": "text", "text": "Hello"} + assert parsed["content"][3]["name"] == "headroom_retrieve" + assert parsed["content"][3]["input"] == {"hash": "abc"} + assert parsed["stop_reason"] == "refusal" + assert parsed["stop_details"] == stop_details + assert parsed["usage"]["output_tokens"] == 3 openai_handler = StreamingCCRHandler(CCRResponseHandler(), provider="openai") parsed_openai = openai_handler._reconstruct_openai_response( @@ -349,9 +382,46 @@ async def test_streaming_handler_falls_back_to_buffer_on_processing_error( async def test_response_to_sse_formats() -> None: anthropic = StreamingCCRHandler(CCRResponseHandler(), provider="anthropic") anthropic_chunks = [chunk async for chunk in anthropic._response_to_sse({"content": []})] - assert anthropic_chunks[0] == b"event: message_start\n" - assert anthropic_chunks[-1] == b'data: {"type": "message_stop"}\n\n' + assert anthropic_chunks[0].startswith(b"event: message_start\n") + assert anthropic_chunks[-1] == b'event: message_stop\ndata: {"type": "message_stop"}\n\n' openai = StreamingCCRHandler(CCRResponseHandler(), provider="openai") openai_chunks = [chunk async for chunk in openai._response_to_sse({"choices": []})] assert openai_chunks == [b'data: {"choices": []}\n\n', b"data: [DONE]\n\n"] + + +@pytest.mark.asyncio +async def test_response_to_sse_preserves_anthropic_shape() -> None: + handler = StreamingCCRHandler(CCRResponseHandler(), provider="anthropic") + stop_details = {"type": "refusal", "message": "policy refusal"} + response = { + "id": "msg_1", + "model": "claude-fable-5", + "role": "assistant", + "content": [ + {"type": "thinking", "thinking": "", "signature": "sig_fable"}, + {"type": "redacted_thinking", "data": "ENC:abc"}, + {"type": "text", "text": "done"}, + ], + "stop_reason": "refusal", + "stop_details": stop_details, + "usage": {"input_tokens": 7, "output_tokens": 3}, + } + + chunks = [chunk async for chunk in handler._response_to_sse(response)] + events = _sse_json_events(chunks) + message_delta = next(event for event in events if event["type"] == "message_delta") + + assert message_delta["delta"]["stop_reason"] == "refusal" + assert message_delta["delta"]["stop_details"] == stop_details + assert any(event.get("delta", {}).get("type") == "signature_delta" for event in events) + assert any( + event.get("content_block", {}).get("type") == "redacted_thinking" for event in events + ) + + parsed = handler._parse_sse_stream(b"".join(chunks)) + assert parsed["content"][0]["signature"] == "sig_fable" + assert parsed["content"][0]["thinking"] == "" + assert parsed["content"][1]["data"] == "ENC:abc" + assert parsed["stop_reason"] == "refusal" + assert parsed["stop_details"] == stop_details diff --git a/tests/test_sse_thinking_blocks.py b/tests/test_sse_thinking_blocks.py index faee266c3..46ca62cab 100644 --- a/tests/test_sse_thinking_blocks.py +++ b/tests/test_sse_thinking_blocks.py @@ -218,6 +218,7 @@ def test_response_to_sse_preserves_server_tool_use_blocks() -> None: def test_response_to_sse_preserves_thinking_redacted_and_citations() -> None: parser = _Parser() redacted_blob = "ENC:" + ("y" * 200) + stop_details = {"type": "refusal", "message": "policy refusal"} response = { "id": "msg_2", "model": "claude-opus-4", @@ -237,7 +238,8 @@ def test_response_to_sse_preserves_thinking_redacted_and_citations() -> None: }, {"type": "redacted_thinking", "data": redacted_blob}, ], - "stop_reason": "end_turn", + "stop_reason": "refusal", + "stop_details": stop_details, "usage": {"input_tokens": 10, "output_tokens": 3}, } @@ -255,6 +257,22 @@ def test_response_to_sse_preserves_thinking_redacted_and_citations() -> None: assert round_tripped["content"][0]["signature"] == "sig_123" assert round_tripped["content"][1]["citations"][0]["cited_text"] == "abc" assert round_tripped["content"][2]["data"] == redacted_blob + assert round_tripped["stop_reason"] == "refusal" + assert round_tripped["stop_details"] == stop_details + + +def test_response_to_sse_does_not_default_missing_stop_reason() -> None: + parser = _Parser() + sse_text = b"".join(parser._response_to_sse({"content": []}, "anthropic")).decode("utf-8") + events = [ + json.loads(line[len("data: ") :]) + for line in sse_text.splitlines() + if line.startswith("data: ") + ] + message_delta = next(event for event in events if event["type"] == "message_delta") + + assert message_delta["delta"] == {} + assert "end_turn" not in sse_text def test_response_to_sse_rejects_unknown_content_block() -> None: