diff --git a/headroom/providers/hermes.py b/headroom/providers/hermes.py new file mode 100644 index 000000000..2538dfdcf --- /dev/null +++ b/headroom/providers/hermes.py @@ -0,0 +1,214 @@ +"""Hermes Studio scoped coding-agent proxy support. + +Hermes owns authentication and protocol adaptation for its scoped proxy routes. +This module owns the small Headroom integration point: safely compress the chat +portion of those requests before the generic proxy forwards them upstream. +""" + +from __future__ import annotations + +import json +import logging +from typing import Any + +logger = logging.getLogger("headroom.providers.hermes") + +_CHAT_ROLES = frozenset({"user", "assistant"}) +_CLAUDE_TEXT_PART_TYPES = frozenset({"text"}) +# Responses uses protocol-specific text part names; normalize these only while +# passing a message through the generic compressor, then restore them exactly. +_RESPONSES_TEXT_PART_TYPES = frozenset({"text", "input_text", "output_text"}) +_CODEX_RESPONSES_SUFFIX = "/v1/responses" +_CLAUDE_MESSAGES_SUFFIX = "/v1/messages" + + +def is_scoped_coding_agent_path(path: str) -> bool: + """Return whether *path* is a Hermes scoped coding-agent endpoint.""" + return (path.startswith("/api/codex-proxy/") and path.endswith(_CODEX_RESPONSES_SUFFIX)) or ( + path.startswith("/api/claude-code-proxy/") and path.endswith(_CLAUDE_MESSAGES_SUFFIX) + ) + + +def compress_scoped_passthrough_body( + path: str, + body: bytes, + *, + optimize: bool, + bypass: bool, +) -> bytes: + """Compress supported Hermes request bodies, otherwise return *body* unchanged. + + The adapter deliberately understands only Hermes's two scoped routes. It + leaves system, tool, reasoning, and non-dictionary input items untouched; + only user/assistant messages are handed to Headroom's compressor. + """ + if not optimize or bypass or not is_scoped_coding_agent_path(path): + return body + + try: + payload = json.loads(body) + if not isinstance(payload, dict): + return body + model = str(payload.get("model") or "").strip() + if not model: + return body + + if path.startswith("/api/claude-code-proxy/"): + field_name = "messages" + route_name = "claude-code" + else: + field_name = "input" + route_name = "codex" + + raw_items = payload.get(field_name) + if isinstance(raw_items, str) and route_name == "codex": + compressed = _compress_messages( + [{"role": "user", "content": raw_items}], model=model, route_name=route_name + ) + if compressed is None: + return body + payload[field_name] = compressed + return _encode_payload(payload) + + if not isinstance(raw_items, list): + return body + + chat_indices = [ + index + for index, item in enumerate(raw_items) + if _is_compressible_chat_message(item, route_name=route_name) + ] + if not chat_indices: + return body + + chat_messages = [raw_items[index] for index in chat_indices] + messages_for_compression = ( + _normalize_responses_text_parts(chat_messages) + if route_name == "codex" + else chat_messages + ) + compressed = _compress_messages( + messages_for_compression, model=model, route_name=route_name + ) + if compressed is None: + return body + if route_name == "codex": + compressed = _restore_responses_text_parts(chat_messages, compressed) + payload[field_name] = _splice_compressed_messages(raw_items, chat_indices, compressed) + return _encode_payload(payload) + except Exception as exc: # Compression must never block Hermes passthrough. + logger.info("Hermes passthrough compression skipped: %s", exc) + return body + + +def _is_compressible_chat_message(item: Any, *, route_name: str) -> bool: + """Return whether a message has only text content safe for compression.""" + if not isinstance(item, dict) or item.get("role") not in _CHAT_ROLES: + return False + content = item.get("content") + if isinstance(content, str): + return True + if not isinstance(content, list) or not content: + return False + text_part_types = ( + _RESPONSES_TEXT_PART_TYPES if route_name == "codex" else _CLAUDE_TEXT_PART_TYPES + ) + return all( + isinstance(part, dict) + and part.get("type") in text_part_types + and isinstance(part.get("text"), str) + for part in content + ) + + +def _normalize_responses_text_parts(messages: list[dict[str, Any]]) -> list[dict[str, Any]]: + """Adapt Responses text parts to the compressor's generic ``text`` shape.""" + normalized: list[dict[str, Any]] = [] + for message in messages: + content = message.get("content") + if not isinstance(content, list): + normalized.append(message) + continue + normalized.append( + { + **message, + "content": [{**part, "type": "text"} for part in content], + } + ) + return normalized + + +def _restore_responses_text_parts( + original_messages: list[dict[str, Any]], compressed_messages: list[dict[str, Any]] +) -> list[dict[str, Any]]: + """Restore Responses part types and metadata after generic compression. + + If the compressor unexpectedly changes the list/message/block structure, + preserve the original message rather than risk producing an invalid + Responses request. + """ + if len(compressed_messages) != len(original_messages): + return original_messages + + restored: list[dict[str, Any]] = [] + for original, compressed in zip(original_messages, compressed_messages, strict=True): + if not isinstance(compressed, dict): + return original_messages + original_content = original.get("content") + compressed_content = compressed.get("content") + if isinstance(original_content, str): + if not isinstance(compressed_content, str): + return original_messages + restored.append({**original, "content": compressed_content}) + continue + if not isinstance(original_content, list) or not isinstance(compressed_content, list): + return original_messages + if len(compressed_content) != len(original_content): + return original_messages + + restored_parts: list[dict[str, Any]] = [] + for original_part, compressed_part in zip( + original_content, compressed_content, strict=True + ): + if not isinstance(compressed_part, dict) or not isinstance( + compressed_part.get("text"), str + ): + return original_messages + restored_parts.append({**original_part, "text": compressed_part["text"]}) + restored.append({**original, "content": restored_parts}) + return restored + + +def _compress_messages( + messages: list[dict[str, Any]], *, model: str, route_name: str +) -> list[dict[str, Any]] | None: + from headroom import compress as headroom_compress + + before_bytes = len(json.dumps(messages, ensure_ascii=False).encode("utf-8")) + result = headroom_compress(messages=messages, model=model, optimize=True) + compressed = list(result.messages) + after_bytes = len(json.dumps(compressed, ensure_ascii=False).encode("utf-8")) + logger.info( + "Hermes %s passthrough compression: %d -> %d bytes (saved %d)", + route_name, + before_bytes, + after_bytes, + max(0, before_bytes - after_bytes), + ) + return compressed + + +def _splice_compressed_messages( + original_items: list[Any], chat_indices: list[int], compressed_items: list[dict[str, Any]] +) -> list[Any]: + """Restore compressed chat messages to their original slots. + + A defensive fallback retains an original item if a compressor unexpectedly + returns fewer messages than it received. + """ + compressed_by_index = dict(zip(chat_indices, compressed_items, strict=False)) + return [compressed_by_index.get(index, item) for index, item in enumerate(original_items)] + + +def _encode_payload(payload: dict[str, Any]) -> bytes: + return json.dumps(payload, separators=(",", ":"), ensure_ascii=False).encode("utf-8") diff --git a/headroom/proxy/handlers/openai.py b/headroom/proxy/handlers/openai.py index 8401470f1..ec9cf1a4b 100644 --- a/headroom/proxy/handlers/openai.py +++ b/headroom/proxy/handlers/openai.py @@ -7494,6 +7494,21 @@ class OpenAIHandlerMixin: logger.debug("Client disconnected during body read for passthrough") return Response(status_code=204) + # Hermes owns its scoped coding-agent protocols; keep that adapter out of + # the generic OpenAI passthrough handler. + from headroom.providers.hermes import compress_scoped_passthrough_body + + optimize_enabled = bool(getattr(getattr(self, "config", None), "optimize", False)) + original_body = body + body = compress_scoped_passthrough_body( + path, + body, + optimize=optimize_enabled, + bypass=_headroom_bypass_enabled(request.headers), + ) + if body is not original_body: + headers["content-length"] = str(len(body)) + headers = await apply_copilot_api_auth(headers, url=url) # Cloudflare bot-management challenges our HTTP/2 fingerprint on # ChatGPT's sensitive account endpoints (/backend-api/me, diff --git a/pyproject.toml b/pyproject.toml index 7d9894d5f..8010dbff2 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -278,6 +278,7 @@ dev = [ "sentence-transformers>=2.2.0,<6.0", "numpy>=1.24.0", "openpyxl>=3.1.0", # exercises spreadsheet_ingest (.xlsx) in the test suite + "respx>=0.20.0", # HTTP mock transport for passthrough handler tests ] # All optional dependencies (everything you need) # diff --git a/tests/test_hermes_passthrough_compression.py b/tests/test_hermes_passthrough_compression.py new file mode 100644 index 000000000..dc450faef --- /dev/null +++ b/tests/test_hermes_passthrough_compression.py @@ -0,0 +1,484 @@ +"""Regression tests for Hermes Studio scoped coding-agent passthrough compression. + +Verifies the new compression logic in ``handle_passthrough`` that rewrites +``/api/codex-proxy/.../v1/responses`` and ``/api/claude-code-proxy/.../v1/messages`` +request bodies before forwarding upstream. + +Key invariants tested: +- Chat messages (role=user/assistant) are compressed +- Non-chat items (tool, function, reasoning, system) are preserved byte-stable +- Non-dict items in the input array are preserved +- x-headroom-bypass header skips compression entirely +- Malformed/unsupported payloads are forwarded unchanged +""" + +from __future__ import annotations + +import json +from typing import Any + +import httpx +import pytest +import respx + +pytest.importorskip("fastapi") + +from fastapi.testclient import TestClient # noqa: E402 + +from headroom.providers.hermes import ( # noqa: E402 + compress_scoped_passthrough_body, + is_scoped_coding_agent_path, +) +from headroom.proxy.loopback_guard import require_loopback # noqa: E402 +from headroom.proxy.server import ProxyConfig, create_app # noqa: E402 + + +def _make_app(**kwargs: Any): + """Create a minimal test app with loopback guard bypassed.""" + config = ProxyConfig( + optimize=True, + cache_enabled=False, + rate_limit_enabled=False, + cost_tracking_enabled=False, + log_requests=False, + **kwargs, + ) + app = create_app(config) + app.dependency_overrides[require_loopback] = lambda: None + return app + + +def _mock_upstream( + router: respx.MockRouter, upstream_url: str = "https://api.openai.com" +) -> dict[str, Any]: + """Install a mock upstream that captures the forwarded request body.""" + captured: dict[str, Any] = {} + + def _capture(request: httpx.Request) -> httpx.Response: + try: + captured["body"] = json.loads(request.content) + except (json.JSONDecodeError, UnicodeDecodeError): + captured["body"] = request.content + captured["url"] = str(request.url) + captured["method"] = request.method + captured["headers"] = dict(request.headers) + return httpx.Response( + 200, + json={"id": "resp_1", "output": [], "usage": {"input_tokens": 10, "output_tokens": 1}}, + ) + + # Mock any upstream path + router.route(method="POST", url__startswith=upstream_url).mock(side_effect=_capture) + router.route(method="GET", url__startswith=upstream_url).mock( + return_value=httpx.Response(200, json={"data": []}) + ) + return captured + + +# ── Codex proxy (Responses API) ────────────────────────────────────────────── + + +@respx.mock +def test_codex_proxy_preserves_tool_and_function_items() -> None: + """Tool/function/reasoning items are preserved after compression.""" + original_input: list[Any] = [ + {"role": "system", "content": "You are a coding assistant."}, + {"role": "user", "content": "Write a sort function."}, + { + "type": "function_call", + "call_id": "call_1", + "name": "read_file", + "arguments": '{"path":"/tmp/x.py"}', + }, + {"type": "function_call_output", "call_id": "call_1", "output": "def foo(): pass"}, + {"role": "assistant", "content": "I see the file contains a foo function."}, + {"role": "user", "content": "Add error handling."}, + ] + + app = _make_app() + with TestClient(app) as client: + captured = _mock_upstream(respx) + + response = client.post( + "/api/codex-proxy/some-session/v1/responses", + headers={"authorization": "Bearer test-key"}, + json={"model": "gpt-4o-mini", "input": original_input}, + ) + + assert response.status_code == 200 + forwarded_input = captured["body"]["input"] + + # Must have same length (all items preserved) + assert len(forwarded_input) == len(original_input), ( + f"Input length changed: {len(original_input)} -> {len(forwarded_input)}" + ) + + # Non-chat items must be identical + for idx in [0, 2, 3]: # system, function_call, function_call_output + assert forwarded_input[idx] == original_input[idx], ( + f"Item {idx} was mutated: {forwarded_input[idx]} != {original_input[idx]}" + ) + + # Chat items (user/assistant) should still be present (may be compressed) + for idx in [1, 4, 5]: + assert isinstance(forwarded_input[idx], dict), f"Item {idx} is no longer a dict" + assert forwarded_input[idx].get("role") == original_input[idx].get("role"), ( + f"Item {idx} role changed" + ) + + +@respx.mock +def test_codex_proxy_preserves_nondict_items() -> None: + """Non-dict items in the input array survive compression.""" + original_input: list[Any] = [ + {"role": "user", "content": "hello"}, + "a plain string that is not a dict", + 42, + {"role": "assistant", "content": "hi there"}, + ] + + app = _make_app() + with TestClient(app) as client: + captured = _mock_upstream(respx) + + response = client.post( + "/api/codex-proxy/session-1/v1/responses", + headers={"authorization": "Bearer test-key"}, + json={"model": "gpt-4o-mini", "input": original_input}, + ) + + assert response.status_code == 200 + forwarded_input = captured["body"]["input"] + + assert len(forwarded_input) == len(original_input) + # Non-dict items preserved exactly + assert forwarded_input[1] == "a plain string that is not a dict" + assert forwarded_input[2] == 42 + # Dict items still present + assert forwarded_input[0].get("role") == "user" + assert forwarded_input[3].get("role") == "assistant" + + +@respx.mock +def test_codex_proxy_bypass_header_skips_compression() -> None: + """x-headroom-bypass: true prevents any body mutation.""" + original_input = [ + {"role": "user", "content": "compressible " + "text " * 200}, + {"role": "assistant", "content": "response"}, + ] + + app = _make_app() + with TestClient(app) as client: + captured = _mock_upstream(respx) + + response = client.post( + "/api/codex-proxy/session-1/v1/responses", + headers={ + "authorization": "Bearer test-key", + "x-headroom-bypass": "true", + }, + json={"model": "gpt-4o-mini", "input": original_input}, + ) + + assert response.status_code == 200 + # Body must be identical (no compression) + assert captured["body"]["input"] == original_input + + +@respx.mock +def test_codex_proxy_malformed_input_preserved() -> None: + """Malformed input (no model) is forwarded without mutation.""" + original_input = [ + {"role": "user", "content": "hello"}, + ] + + app = _make_app() + with TestClient(app) as client: + captured = _mock_upstream(respx) + + response = client.post( + "/api/codex-proxy/session-1/v1/responses", + headers={"authorization": "Bearer test-key"}, + json={"input": original_input}, # no "model" key + ) + + assert response.status_code == 200 + # Must be forwarded unchanged + assert captured["body"]["input"] == original_input + + +@respx.mock +def test_codex_proxy_compression_applies_to_chat_messages() -> None: + """Chat messages are compressed when model is present and bypass is off.""" + app = _make_app() + with TestClient(app) as client: + captured = _mock_upstream(respx) + + response = client.post( + "/api/codex-proxy/session-1/v1/responses", + headers={"authorization": "Bearer test-key"}, + json={ + "model": "gpt-4o-mini", + "input": [ + {"role": "user", "content": "hello world"}, + {"role": "assistant", "content": "hi"}, + ], + }, + ) + + assert response.status_code == 200 + forwarded_input = captured["body"]["input"] + assert len(forwarded_input) >= 2 # at least original count + # Roles preserved + assert forwarded_input[0].get("role") == "user" + assert forwarded_input[1].get("role") == "assistant" + + +# ── Claude Code proxy (Anthropic Messages API) ─────────────────────────────── + + +@respx.mock +def test_claude_proxy_preserves_tool_use_items() -> None: + """tool_use and tool_result messages are preserved after compression.""" + original_messages = [ + {"role": "user", "content": [{"type": "text", "text": "Read the file."}]}, + { + "role": "assistant", + "content": [ + { + "type": "tool_use", + "id": "toolu_1", + "name": "read_file", + "input": {"path": "/tmp/x.py"}, + } + ], + }, + { + "role": "user", + "content": [ + {"type": "tool_result", "tool_use_id": "toolu_1", "content": "print('hello')"} + ], + }, + {"role": "assistant", "content": [{"type": "text", "text": "The file prints hello."}]}, + ] + + app = _make_app() + with TestClient(app) as client: + captured = _mock_upstream( + respx, upstream_url="https://httpbin.org" + ) # any URL, overridden by header + + response = client.post( + "/api/claude-code-proxy/session-1/v1/messages", + headers={ + "authorization": "Bearer test-key", + "x-headroom-base-url": "https://httpbin.org", + }, + json={"model": "claude-sonnet-4-5-20250929", "messages": original_messages}, + ) + + assert response.status_code == 200 + forwarded_messages = captured["body"]["messages"] + + assert len(forwarded_messages) == len(original_messages) + + # tool_use message preserved + assert forwarded_messages[1]["role"] == "assistant" + assert forwarded_messages[1]["content"][0]["type"] == "tool_use" + + # tool_result message preserved + assert forwarded_messages[2]["role"] == "user" + assert forwarded_messages[2]["content"][0]["type"] == "tool_result" + + +@respx.mock +def test_claude_proxy_bypass_header_skips_compression() -> None: + """x-headroom-bypass: true prevents any body mutation on Claude proxy.""" + original_messages = [ + {"role": "user", "content": "compressible " + "text " * 200}, + {"role": "assistant", "content": "response"}, + ] + + app = _make_app() + with TestClient(app) as client: + captured = _mock_upstream(respx, upstream_url="https://httpbin.org") + + response = client.post( + "/api/claude-code-proxy/session-1/v1/messages", + headers={ + "authorization": "Bearer test-key", + "x-headroom-bypass": "true", + "x-headroom-base-url": "https://httpbin.org", + }, + json={"model": "claude-sonnet-4-5-20250929", "messages": original_messages}, + ) + + assert response.status_code == 200 + assert captured["body"]["messages"] == original_messages + + +@respx.mock +def test_claude_proxy_no_model_forwarded_unchanged() -> None: + """Missing model → forwarded without compression.""" + original_messages = [ + {"role": "user", "content": "hello"}, + ] + + app = _make_app() + with TestClient(app) as client: + captured = _mock_upstream(respx, upstream_url="https://httpbin.org") + + response = client.post( + "/api/claude-code-proxy/session-1/v1/messages", + headers={ + "authorization": "Bearer test-key", + "x-headroom-base-url": "https://httpbin.org", + }, + json={"messages": original_messages}, + ) + + assert response.status_code == 200 + assert captured["body"]["messages"] == original_messages + + +@respx.mock +def test_claude_proxy_compression_applies_to_chat_messages() -> None: + """Chat messages (user/assistant) are compressed.""" + app = _make_app() + with TestClient(app) as client: + captured = _mock_upstream(respx, upstream_url="https://httpbin.org") + + response = client.post( + "/api/claude-code-proxy/session-1/v1/messages", + headers={ + "authorization": "Bearer test-key", + "x-headroom-base-url": "https://httpbin.org", + }, + json={ + "model": "claude-sonnet-4-5-20250929", + "messages": [ + {"role": "user", "content": "hello world"}, + {"role": "assistant", "content": "hi"}, + ], + }, + ) + + assert response.status_code == 200 + forwarded_messages = captured["body"]["messages"] + assert len(forwarded_messages) >= 2 + assert forwarded_messages[0].get("role") == "user" + assert forwarded_messages[1].get("role") == "assistant" + + +# ── Generic passthrough (non-Hermes routes) ────────────────────────────────── + + +@respx.mock +def test_non_hermes_routes_not_affected() -> None: + """Non-Hermes passthrough routes are not touched.""" + app = _make_app() + with TestClient(app) as client: + captured = _mock_upstream(respx) + + response = client.post( + "/v1/chat/completions", + headers={"authorization": "Bearer test-key"}, + json={ + "model": "gpt-4o-mini", + "messages": [{"role": "user", "content": "hello"}], + }, + ) + + assert response.status_code == 200 + # Normal passthrough: body should just have "messages" + assert "messages" in captured["body"] + + +def test_hermes_adapter_ignores_non_hermes_paths() -> None: + assert not is_scoped_coding_agent_path("/v1/responses") + assert not is_scoped_coding_agent_path("/api/codex-proxy/session/v1/chat/completions") + assert is_scoped_coding_agent_path("/api/codex-proxy/session/v1/responses") + assert is_scoped_coding_agent_path("/api/claude-code-proxy/session/v1/messages") + + +def test_hermes_adapter_preserves_structured_messages_when_compressor_changes_output( + monkeypatch: pytest.MonkeyPatch, +) -> None: + def fake_compress(*, messages: list[dict[str, Any]], **_: Any): + class Result: + pass + + result = Result() + result.messages = [{**message, "content": "compressed"} for message in messages] + return result + + monkeypatch.setattr("headroom.compress", fake_compress) + original = { + "model": "claude-sonnet-4-5-20250929", + "messages": [ + {"role": "user", "content": "compress this"}, + {"role": "assistant", "content": [{"type": "tool_use", "name": "read_file"}]}, + {"role": "user", "content": [{"type": "tool_result", "content": "secret"}]}, + ], + } + body = json.dumps(original).encode() + + transformed = compress_scoped_passthrough_body( + "/api/claude-code-proxy/session/v1/messages", body, optimize=True, bypass=False + ) + + messages = json.loads(transformed)["messages"] + assert messages[0]["content"] == "compressed" + assert messages[1] == original["messages"][1] + assert messages[2] == original["messages"][2] + + +def test_codex_adapter_compresses_canonical_responses_input_text( + monkeypatch: pytest.MonkeyPatch, +) -> None: + seen: dict[str, list[dict[str, Any]]] = {} + + def fake_compress(*, messages: list[dict[str, Any]], **_: Any): + class Result: + pass + + seen["messages"] = messages + result = Result() + result.messages = [ + { + **messages[0], + "content": [{**messages[0]["content"][0], "text": "compressed prompt"}], + } + ] + return result + + monkeypatch.setattr("headroom.compress", fake_compress) + original = { + "model": "gpt-5.5", + "input": [ + { + "type": "message", + "role": "user", + "metadata": {"source": "codex"}, + "content": [{"type": "input_text", "text": "long prompt", "annotations": []}], + } + ], + } + body = json.dumps(original).encode() + + transformed = compress_scoped_passthrough_body( + "/api/codex-proxy/session/v1/responses", body, optimize=True, bypass=False + ) + + assert transformed != body + assert seen["messages"][0]["type"] == "message" + assert seen["messages"][0]["content"] == [ + {"type": "text", "text": "long prompt", "annotations": []} + ] + forwarded = json.loads(transformed)["input"][0] + assert forwarded == { + "type": "message", + "role": "user", + "metadata": {"source": "codex"}, + "content": [{"type": "input_text", "text": "compressed prompt", "annotations": []}], + } diff --git a/tests/test_proxy_handler_helpers.py b/tests/test_proxy_handler_helpers.py index 3fa3e23f6..9292f9930 100644 --- a/tests/test_proxy_handler_helpers.py +++ b/tests/test_proxy_handler_helpers.py @@ -279,6 +279,17 @@ def test_headroom_bypass_helper_is_transport_neutral() -> None: assert OpenAIHandlerMixin._headroom_bypass_enabled({"x-headroom-bypass": "true"}) is True +def test_openai_passthrough_without_config_preserves_generic_request() -> None: + handler = object.__new__(OpenAIHandlerMixin) + handler.http_client = _RecordingHttpClient("h2") + request = _PassthroughRequest() + + response = asyncio.run(handler.handle_passthrough(request, "https://api.openai.com")) + + assert response.status_code == 200 + assert json.loads(response.body)["client"] == "h2" + + def test_openai_passthrough_connect_timeout_returns_502() -> None: handler = object.__new__(OpenAIHandlerMixin) handler.http_client = _TimeoutHttpClient()