vinez/headroom
1
0
Fork 0
mirror of https://github.com/headroomlabs-ai/headroom.git synced 2026-08-10 14:27:00 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions

fix(proxy): compress Hermes scoped coding-agent passthrough (#1815)

Browse source 
## Description

Compress Hermes Studio scoped coding-agent passthrough requests in the
generic OpenAI passthrough handler. Hermes can route scoped Claude Code
and Codex traffic through Headroom while preserving its own proxy paths;
this PR keeps Hermes responsible for scoped proxy
authentication/provider adaptation while still applying Headroom
compression to supported chat payloads before forwarding.

The compression remains narrow-scoped:
- Only chat messages with `user` or `assistant` roles are compressed.
- Tool, function, reasoning, and system items are preserved byte-stable.
- Non-dict items in the Responses `input` array are preserved and
spliced back.

## Type of Change

- [ ] Bug fix (non-breaking change that fixes an issue)
- [x] 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

- Detect `/api/codex-proxy/.../v1/responses` paths and compress
supported Responses `input` chat items before forwarding.
- Detect `/api/claude-code-proxy/.../v1/messages` paths and compress
supported Anthropic `messages` payloads before forwarding.
- Preserve bypass, malformed payload, missing-model, tool/function,
reasoning/system, and non-dict passthrough behavior.
- Add regression coverage in
`tests/test_hermes_passthrough_compression.py`.

## Testing

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

### Test Output

```text
$ python -m pytest tests/test_hermes_passthrough_compression.py -v
test_codex_proxy_preserves_tool_and_function_items PASSED
test_codex_proxy_preserves_nondict_items PASSED
test_codex_proxy_bypass_header_skips_compression PASSED
test_codex_proxy_malformed_input_preserved PASSED
test_codex_proxy_compression_applies_to_chat_messages PASSED
test_claude_proxy_preserves_tool_use_items PASSED
test_claude_proxy_bypass_header_skips_compression PASSED
test_claude_proxy_no_model_forwarded_unchanged PASSED
test_claude_proxy_compression_applies_to_chat_messages PASSED
test_non_hermes_routes_not_affected PASSED
```

## Real Behavior Proof

- Environment: Author-reported local test environment for
`headroom/proxy/handlers/openai.py` and
`tests/test_hermes_passthrough_compression.py`.
- Exact command / steps: `python -m pytest
tests/test_hermes_passthrough_compression.py -v`.
- Observed result: The 10 Hermes passthrough regression tests passed,
covering Codex and Claude scoped proxy routes plus preservation/bypass
cases.
- Not tested: End-to-end Hermes Studio traffic against a live upstream
service is not covered by this PR body evidence.

## 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
- [ ] 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

## Screenshots (if applicable)

N/A.

## Additional Notes

Generated with Claude Code. The unchecked checklist items are not
required for this narrow proxy-handler test change.

---------

Co-authored-by: x1051445024 <你的GitHub注册邮箱>
Co-authored-by: Claude <noreply@anthropic.com>
Co-authored-by: JerrettDavis <mxjerrett@gmail.com>
This commit is contained in:
牧濑红莉栖（BOT） 2026-07-14 06:22:46 +08:00 • committed by GitHub
parent 8870b6971f
commit 09d1ef45be
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
5 changed files with 725 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

214
headroom/providers/hermes.py Normal file
View file

@ -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")

15
headroom/proxy/handlers/openai.py
View file

@ -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,

1
pyproject.toml
View file

@ -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)
#

484
tests/test_hermes_passthrough_compression.py Normal file
View file

@ -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": []}],
}

11
tests/test_proxy_handler_helpers.py
View file

@ -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()