fix(agent): keep moved chat fields as slot placeholders so the wire body is byte-identical; one shared escape hatch
The cherry-picked helper deleted 'tools' from the typed kwargs, so the SDK's post-transform extra_body merge appended it after the caller's extra_body keys — equal dict, different bytes (byte-keyed prompt caches would miss). keep_slots=True leaves [] placeholders that the merge overwrites in place. Drop the invented HERMES_CHAT_SDK_TRANSFORM env var; the pre-existing HERMES_CODEX_SDK_TRANSFORM hatch from #93650 now disables both API families. Tests trimmed to the two invariants (byte-identity incl. caller extra_body precedence; escape hatch).
This commit is contained in:
@@ -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)
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user