diff --git a/agent/sdk_transform_bypass.py b/agent/sdk_transform_bypass.py index bcbcf02a43..1078819534 100644 --- a/agent/sdk_transform_bypass.py +++ b/agent/sdk_transform_bypass.py @@ -1,21 +1,15 @@ """Route bulk request payloads around the OpenAI SDK's request transform (#93650). -Both the Responses API (``responses.create``) and chat-completions -(``chat.completions.create``) re-walk the entire request body against their -TypedDict/union param graph client-side, before any byte leaves the process. -That walk holds the GIL, and #93650 documents it wedging for 12+ hours on a -~1.4 MB conversation — starving every other thread, including the TTFB and -stale-call watchdogs whose whole job is to rescue this exact call. Because the -hang is client-side and pre-network, no socket kill can unblock it. +``responses.create`` and ``chat.completions.create`` both re-walk the whole request +body against their TypedDict/union param graph client-side, with the GIL held, +before any byte leaves the process. #93650 documents that walk wedging for 12+ +hours on a ~1.4 MB conversation — starving the TTFB/stale watchdogs whose job is +to rescue this exact call; the hang is pre-network, so no socket kill helps. -Hermes assembles these payloads from JSON round-trips, so they are already in -wire format and the walk has nothing to convert. The SDK merges ``extra_body`` -into the JSON body *after* the transform -(``_base_client._build_request``), so moving the already-wire-format bulk -fields there skips the walk and produces a byte-identical request. - -This module is the shared home for the helpers; ``agent.codex_runtime`` -re-exports them so existing import paths keep working. +Hermes assembles these payloads from JSON round-trips, so they are already wire +format and the walk has nothing to convert. The SDK merges ``extra_body`` into +the JSON body *after* the transform (``_base_client._build_request``), so moving +the bulk fields there skips the walk and yields the same request bytes. """ from __future__ import annotations @@ -23,112 +17,64 @@ from __future__ import annotations import os from typing import Any -# Bulk request fields that carry the conversation payload, per API family. -# Everything else in a request is scalar configuration the SDK transform -# handles in microseconds. +# Bulk request fields carrying the conversation payload, per API family. Everything +# else is scalar configuration the SDK transform handles in microseconds. RESPONSES_BYPASS_FIELDS = ("input", "tools") CHAT_COMPLETIONS_BYPASS_FIELDS = ("messages", "tools") +# One hatch for both API families (established by #93650); restores the typed SDK path. +ESCAPE_HATCH_ENV = "HERMES_CODEX_SDK_TRANSFORM" + def _is_plain_json_data(value: Any) -> bool: - """True when ``value`` is composed purely of JSON wire types. - - The SDK's request transform exists to convert typed params (TypedDict - key aliases, pydantic models, ``PropertyInfo`` formats) into wire - format. Hermes assembles these payloads from JSON round-trips, so they - are already wire format — but that is only provable when every node is - a plain JSON type. Anything else must keep the typed SDK path. - """ + """True when ``value`` is purely JSON wire types; pydantic models / generators must keep the typed SDK path.""" if value is None or isinstance(value, (str, int, float, bool)): return True if isinstance(value, dict): - return all( - isinstance(key, str) and _is_plain_json_data(item) - for key, item in value.items() - ) + return all(isinstance(key, str) and _is_plain_json_data(item) for key, item in value.items()) if isinstance(value, list): return all(_is_plain_json_data(item) for item in value) return False -def _env_flag(name: str) -> bool: - return os.environ.get(name, "").strip().lower() in {"1", "true", "yes", "on"} - - def bypass_sdk_request_transform( request_kwargs: dict, + fields: tuple[str, ...] = RESPONSES_BYPASS_FIELDS, *, - fields: tuple = RESPONSES_BYPASS_FIELDS, - escape_hatch_env: str = "HERMES_CODEX_SDK_TRANSFORM", - required_empty: tuple = (), + keep_slots: bool = False, ) -> dict: """Move wire-format bulk ``fields`` into ``extra_body``. - Returns ``request_kwargs`` unchanged when there is nothing safe to move, - so a caller can always use the result unconditionally. Fields holding - anything that is not plain JSON data (pydantic models, generators) stay - on the typed path, which still needs the transform. - - ``required_empty`` names fields the SDK declares ``@required_args`` and - would reject as missing: those are kept in the typed kwargs as an empty - list, and the ``extra_body`` copy overwrites them in the JSON body. Set - ``escape_hatch_env`` to restore the pre-fix behaviour. + Returns ``request_kwargs`` itself when nothing is safe to move, so callers use + the result unconditionally. ``keep_slots`` leaves an empty-list placeholder in + the typed kwargs for each moved field: it satisfies ``@required_args`` + (``messages`` on chat.completions) and keeps the field's position in the JSON + body, so the bytes — and therefore any byte-keyed prompt cache — are unchanged. """ - if _env_flag(escape_hatch_env): + if os.environ.get(ESCAPE_HATCH_ENV, "").strip().lower() in {"1", "true", "yes", "on"}: return request_kwargs - - moved = { - field: request_kwargs[field] - for field in fields - if isinstance(request_kwargs.get(field), (dict, list)) - and _is_plain_json_data(request_kwargs[field]) - } + moved = {f: request_kwargs[f] for f in fields + if isinstance(request_kwargs.get(f), (dict, list)) and _is_plain_json_data(request_kwargs[f])} if not moved: return request_kwargs - - bypassed = { - key: value for key, value in request_kwargs.items() if key not in moved - } - for field in required_empty: - if field in moved: - # The SDK rejects the call outright if a @required_args parameter - # is absent; an empty list satisfies the signature and the - # extra_body entry replaces it in the body the server sees. - bypassed[field] = [] + bypassed = {key: ([] if keep_slots else value) if key in moved else value + for key, value in request_kwargs.items() if keep_slots or key not in moved} extra_body = bypassed.get("extra_body") merged = dict(extra_body) if isinstance(extra_body, dict) else {} - for field, value in moved.items(): - # An explicit caller-provided extra_body entry keeps precedence, - # matching what the SDK's post-transform merge would have done. - merged.setdefault(field, value) - bypassed["extra_body"] = merged + # An explicit caller-provided extra_body entry keeps precedence (SDK post-transform merge). + bypassed["extra_body"] = {**merged, **{f: v for f, v in moved.items() if f not in merged}} return bypassed -def is_openai_sdk_completions(client: Any) -> bool: - """True when ``client.chat.completions`` is the real OpenAI SDK object. +def bypass_chat_sdk_request_transform(request_kwargs: dict, client: Any) -> dict: + """Chat-completions bypass, gated on the real OpenAI SDK. - Only the SDK performs the transform this bypass exists to skip, and only - the SDK merges ``extra_body`` into the body afterwards. Hermes also drives - chat-completions-shaped facades that are NOT the SDK — the in-process MoA - aggregator (``agent/moa_loop.py``) most importantly, plus the stand-ins - the test suite injects — and handing those an ``extra_body`` they never - merge would silently drop the conversation. Gate on the real thing. + Only the SDK performs the transform and only the SDK merges ``extra_body`` + afterwards. Hermes also drives chat-shaped facades that are NOT the SDK (the + in-process MoA aggregator, test stand-ins); handing those an ``extra_body`` they + never merge would silently send an empty conversation. """ completions = getattr(getattr(client, "chat", None), "completions", None) - if completions is None: - return False - return type(completions).__module__.startswith("openai.") - - -def bypass_chat_sdk_request_transform(request_kwargs: dict, client: Any) -> dict: - """``bypass_sdk_request_transform`` for chat-completions, SDK-gated.""" - if not is_openai_sdk_completions(client): + if completions is None or not type(completions).__module__.startswith("openai."): return request_kwargs - return bypass_sdk_request_transform( - request_kwargs, - fields=CHAT_COMPLETIONS_BYPASS_FIELDS, - escape_hatch_env="HERMES_CHAT_SDK_TRANSFORM", - # ``messages`` is @required_args on chat.completions.create. - required_empty=("messages",), - ) + return bypass_sdk_request_transform(request_kwargs, CHAT_COMPLETIONS_BYPASS_FIELDS, keep_slots=True) diff --git a/tests/agent/test_chat_sdk_transform_bypass.py b/tests/agent/test_chat_sdk_transform_bypass.py index f93bd445fc..685030a43e 100644 --- a/tests/agent/test_chat_sdk_transform_bypass.py +++ b/tests/agent/test_chat_sdk_transform_bypass.py @@ -1,18 +1,12 @@ -"""Chat-completions request-transform bypass (#93650). +"""Chat-completions request-transform bypass (#93650 extended to chat.completions). ``chat.completions.create`` re-walks the whole request body against the -``CompletionCreateParams`` union graph client-side, with the GIL held, before -any byte leaves the process. #93650 documents that class of walk wedging for -12+ hours on a ~1.4 MB conversation, where no in-process watchdog can fire and -no socket kill helps because the hang is pre-network. #93773 fixed it for -``responses.create`` only; these tests cover extending the same, already -reviewed mechanism to the default chat path. - -The safety argument is that the request the server receives is unchanged, so -the byte-identity test below is the important one. +``CompletionCreateParams`` union client-side, GIL held, before any byte leaves +the process. The bypass moves the already-wire-format bulk fields into +``extra_body``; its whole safety argument is that the server receives the same +bytes, so that is what these tests pin. """ -import json import sys import types @@ -22,12 +16,8 @@ sys.modules.setdefault("fal_client", types.SimpleNamespace()) import httpx import openai -import pytest -from agent.sdk_transform_bypass import ( - bypass_chat_sdk_request_transform, - is_openai_sdk_completions, -) +from agent.sdk_transform_bypass import ESCAPE_HATCH_ENV, bypass_chat_sdk_request_transform _SSE = ( b'data: {"id":"1","object":"chat.completion.chunk","created":1,"model":"m",' @@ -36,213 +26,73 @@ _SSE = ( ) -def _wire_body(tools: bool = True) -> dict: - """A production-shaped chat body: list content parts, an image part, a - tool_calls turn, a tool result, and function tool schemas.""" - body = { +def _wire_body() -> dict: + """Production-shaped chat body: content parts incl. an image, a tool_calls turn, a tool + result, function tool schemas, and a caller-populated extra_body (reasoning/provider).""" + return { "model": "hermes-4-70b", "messages": [ {"role": "system", "content": "You are Hermes."}, - { - "role": "user", - "content": [ - {"type": "text", "text": "look at this"}, - { - "type": "image_url", - "image_url": {"url": "https://e.example/i.png", "detail": "low"}, - }, - ], - }, - { - "role": "assistant", - "content": None, - "tool_calls": [ - { - "id": "c1", - "type": "function", - "function": {"name": "terminal", "arguments": '{"cmd":"ls"}'}, - } - ], - }, + {"role": "user", "content": [ + {"type": "text", "text": "look at this"}, + {"type": "image_url", "image_url": {"url": "https://e.example/i.png", "detail": "low"}}, + ]}, + {"role": "assistant", "content": None, "tool_calls": [ + {"id": "c1", "type": "function", "function": {"name": "terminal", "arguments": '{"cmd":"ls"}'}}, + ]}, {"role": "tool", "tool_call_id": "c1", "content": "total 0"}, ], + "tools": [ + {"type": "function", "function": {"name": f"tool_{i}", "description": "d", + "parameters": {"type": "object", "properties": {"p": {"type": "string"}}, "required": ["p"]}}} + for i in range(3) + ], + "tool_choice": "auto", "stream": True, "temperature": 0.7, "stream_options": {"include_usage": True}, + "extra_body": {"reasoning": {"effort": "high"}, "provider": {"order": ["nous"]}}, } - if tools: - body["tools"] = [ - { - "type": "function", - "function": { - "name": f"tool_{i}", - "description": "d", - "parameters": { - "type": "object", - "properties": {"p": {"type": "string"}}, - "required": ["p"], - }, - }, - } - for i in range(30) - ] - return body class _Recorder: - """A real openai.OpenAI client whose transport records the request body.""" + """A real openai.OpenAI client whose transport records the request bytes.""" def __init__(self): self.content: bytes | None = None self.client = openai.OpenAI( - api_key="k", - base_url="https://chat.invalid/v1", + api_key="k", base_url="https://chat.invalid/v1", http_client=httpx.Client(transport=httpx.MockTransport(self._handle)), ) def _handle(self, request: httpx.Request) -> httpx.Response: self.content = request.content - return httpx.Response( - 200, content=_SSE, headers={"content-type": "text/event-stream"} - ) + return httpx.Response(200, content=_SSE, headers={"content-type": "text/event-stream"}) def send(self, kwargs: dict) -> bytes: - stream = self.client.chat.completions.create(**kwargs) - for _ in stream: + for _ in self.client.chat.completions.create(**kwargs): pass assert self.content is not None return self.content -class _FacadeCompletions: - """Stands in for the MoA aggregator: chat-completions shaped, not the SDK.""" +def test_bulk_fields_ride_in_extra_body_and_the_wire_bytes_are_identical(): + """Same bytes → same server behaviour and the same byte-keyed prompt-cache prefix.""" + recorder = _Recorder() + body = _wire_body() - def create(self, **kwargs): # pragma: no cover - never called here - raise AssertionError("not exercised") + moved = bypass_chat_sdk_request_transform(dict(body), recorder.client) + + assert moved["messages"] == [] and moved["tools"] == [] + assert moved["extra_body"]["messages"] == body["messages"] + assert moved["extra_body"]["tools"] == body["tools"] + assert moved["extra_body"]["reasoning"] == body["extra_body"]["reasoning"] + assert recorder.send(moved) == recorder.send(dict(body)) -class _FacadeClient: - def __init__(self): - self.chat = types.SimpleNamespace(completions=_FacadeCompletions()) +def test_escape_hatch_restores_the_typed_sdk_path(monkeypatch): + monkeypatch.setenv(ESCAPE_HATCH_ENV, "1") + recorder = _Recorder() + kwargs = _wire_body() - -class TestByteIdentity: - def test_request_body_is_byte_identical_with_and_without_the_bypass(self): - """The whole safety argument: the server sees the same request.""" - recorder = _Recorder() - body = _wire_body() - - plain = recorder.send(dict(body)) - bypassed = recorder.send(bypass_chat_sdk_request_transform(dict(body), recorder.client)) - - assert bypassed == plain - # And it really did take the bypass, rather than silently no-opping. - moved = bypass_chat_sdk_request_transform(dict(body), recorder.client) - assert moved["messages"] == [] - assert moved["extra_body"]["messages"] == body["messages"] - assert moved["extra_body"]["tools"] == body["tools"] - - def test_the_wire_payload_survives_a_json_round_trip_unchanged(self): - recorder = _Recorder() - body = _wire_body() - sent = json.loads(recorder.send(bypass_chat_sdk_request_transform(dict(body), recorder.client))) - - assert sent["messages"] == body["messages"] - assert sent["tools"] == body["tools"] - assert sent["temperature"] == 0.7 - assert sent["stream_options"] == {"include_usage": True} - - -class TestGuards: - def test_a_non_sdk_client_is_left_completely_alone(self): - """MoA's facade never merges extra_body, so moving the conversation - there would silently send an empty message list.""" - kwargs = _wire_body() - - result = bypass_chat_sdk_request_transform(kwargs, _FacadeClient()) - - assert result is kwargs - assert "extra_body" not in result - assert result["messages"][0]["role"] == "system" - - def test_is_openai_sdk_completions_discriminates(self): - assert is_openai_sdk_completions(_Recorder().client) - assert not is_openai_sdk_completions(_FacadeClient()) - assert not is_openai_sdk_completions(object()) - - def test_non_plain_json_messages_stay_on_the_typed_path(self): - """The transform exists to convert typed params; anything that is not - already wire data still needs it.""" - recorder = _Recorder() - kwargs = _wire_body() - kwargs["messages"][1]["content"] = object() - - result = bypass_chat_sdk_request_transform(kwargs, recorder.client) - - assert result["messages"] is kwargs["messages"] - assert "extra_body" not in result or "messages" not in result["extra_body"] - - def test_a_tool_free_request_does_not_gain_an_empty_tools_key(self): - recorder = _Recorder() - body = _wire_body(tools=False) - - sent = json.loads(recorder.send(bypass_chat_sdk_request_transform(dict(body), recorder.client))) - - assert "tools" not in sent - - def test_caller_supplied_extra_body_keeps_precedence(self): - """The chat path already populates extra_body from custom providers, - reasoning config and Nous Portal; those entries must win.""" - recorder = _Recorder() - body = _wire_body() - body["extra_body"] = {"provider": {"order": ["nous"]}, "messages": ["caller wins"]} - - result = bypass_chat_sdk_request_transform(dict(body), recorder.client) - - assert result["extra_body"]["messages"] == ["caller wins"] - assert result["extra_body"]["provider"] == {"order": ["nous"]} - - def test_env_escape_hatch_restores_the_pre_fix_kwargs(self, monkeypatch): - monkeypatch.setenv("HERMES_CHAT_SDK_TRANSFORM", "1") - recorder = _Recorder() - kwargs = _wire_body() - - assert bypass_chat_sdk_request_transform(kwargs, recorder.client) is kwargs - - @pytest.mark.parametrize("value", ["0", "false", "no", "off", ""]) - def test_escape_hatch_stays_off_for_falsey_values(self, monkeypatch, value): - monkeypatch.setenv("HERMES_CHAT_SDK_TRANSFORM", value) - recorder = _Recorder() - - result = bypass_chat_sdk_request_transform(_wire_body(), recorder.client) - - assert result["messages"] == [] - - -class TestResponsesPathUnchanged: - def test_the_codex_import_path_and_behaviour_are_preserved(self): - """agent/auxiliary_client.py and the existing codex test import these - names from agent.codex_runtime; the move must not break them.""" - from agent.codex_runtime import ( - _SDK_TRANSFORM_BYPASS_FIELDS, - _bypass_sdk_request_transform, - _is_plain_json_data, - ) - - assert _SDK_TRANSFORM_BYPASS_FIELDS == ("input", "tools") - assert _is_plain_json_data([{"role": "user", "content": "hi"}]) - assert not _is_plain_json_data([{"role": "user", "content": object()}]) - - kwargs = { - "model": "gpt-5.6-sol", - "input": [{"role": "user", "content": [{"type": "input_text", "text": "hi"}]}], - "tools": [{"type": "function", "name": "terminal", "parameters": {}}], - "stream": True, - } - bypassed = _bypass_sdk_request_transform(kwargs) - - # Responses keeps its historical shape: the fields are REMOVED from the - # typed kwargs entirely (input is not @required_args there). - assert "input" not in bypassed - assert "tools" not in bypassed - assert bypassed["extra_body"]["input"] == kwargs["input"] + assert bypass_chat_sdk_request_transform(kwargs, recorder.client) is kwargs