Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 7 additions & 1 deletion agent/auxiliary_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -1806,10 +1806,16 @@ def _check_cancelled() -> None:
# Consuming raw events and assembling the final response
# ourselves from ``response.output_item.done`` makes us
# structurally immune to that drift.
from agent.codex_runtime import _consume_codex_event_stream
from agent.codex_runtime import (
_bypass_sdk_request_transform,
_consume_codex_event_stream,
)

stream_kwargs = dict(resp_kwargs)
stream_kwargs["stream"] = True
# #93650: keep bulk wire-format payload out of the SDK's
# GIL-holding request transform on auxiliary calls too.
stream_kwargs = _bypass_sdk_request_transform(stream_kwargs)

def _on_each_event(_event: Any) -> None:
# Re-check timeout/cancellation per event, matching the
Expand Down
74 changes: 74 additions & 0 deletions agent/codex_runtime.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@

import json
import logging
import os
import time
from types import SimpleNamespace
from typing import Any, Callable, Dict, List
Expand Down Expand Up @@ -1496,6 +1497,78 @@ def _sanitize_consumer_codex_request(
return sanitized


# Bulk request fields that carry the conversation payload. Everything else in
# the request is scalar configuration the SDK transform handles in microseconds.
_SDK_TRANSFORM_BYPASS_FIELDS = ("input", "tools")


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 Codex 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.
"""
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()
)
if isinstance(value, list):
return all(_is_plain_json_data(item) for item in value)
return False


def _bypass_sdk_request_transform(stream_kwargs: dict) -> dict:
"""Route bulk payload fields around the SDK's ``maybe_transform`` (#93650).

``responses.create`` re-walks the entire request body against the
``ResponseCreateParams`` union graph before any byte leaves the process.
That walk runs with the GIL held, and #93650 documents it wedging for
12+ hours on a ~1.4 MB conversation — starving every other thread,
including the TTFB/stale watchdogs whose job is to rescue this exact
call. Because the hang is client-side and pre-network, no socket kill
can unblock it.

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 entirely and produces a byte-identical
request. Fields containing anything that is not plain JSON data (e.g.
pydantic models, generators) stay on the typed path, which still needs
the transform. Set HERMES_CODEX_SDK_TRANSFORM=1 to restore the pre-fix
behavior.
"""
if os.environ.get("HERMES_CODEX_SDK_TRANSFORM", "").strip().lower() in {
"1", "true", "yes", "on"
}:
return stream_kwargs

moved = {
field: stream_kwargs[field]
for field in _SDK_TRANSFORM_BYPASS_FIELDS
if isinstance(stream_kwargs.get(field), (dict, list))
and _is_plain_json_data(stream_kwargs[field])
}
if not moved:
return stream_kwargs

bypassed = {
key: value for key, value in stream_kwargs.items() if 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
return bypassed


def run_codex_stream(agent, api_kwargs: dict, client: Any = None, on_first_delta=None):
"""Execute one streaming Responses API request and return the final response.

Expand Down Expand Up @@ -1543,6 +1616,7 @@ def _open_codex_stream(next_api_kwargs: dict[str, Any]):
next_api_kwargs,
)
stream_kwargs["stream"] = True
stream_kwargs = _bypass_sdk_request_transform(stream_kwargs)
return active_client.responses.create(**stream_kwargs)

def _codex_stream_created(_raw_stream: Any) -> None:
Expand Down
2 changes: 2 additions & 0 deletions contributors/emails/kaloyan.chernev@aifnet.com
Original file line number Diff line number Diff line change
@@ -0,0 +1,2 @@
kchernev
# PR #93681 salvage
15 changes: 14 additions & 1 deletion tests/agent/test_auxiliary_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -2830,6 +2830,9 @@ def close(self): pass

def _create(**kwargs):
captured_kwargs.update(kwargs)
# #93650 routes bulk fields through extra_body; fold them back in
# so assertions read the effective wire body the SDK would send.
captured_kwargs.update(kwargs.get("extra_body") or {})
return _FakeCreateStream()

real_client = MagicMock()
Expand Down Expand Up @@ -2913,6 +2916,9 @@ def close(self): pass

def _create(**kwargs):
captured_kwargs.update(kwargs)
# #93650 routes bulk fields through extra_body; fold them back in
# so assertions read the effective wire body the SDK would send.
captured_kwargs.update(kwargs.get("extra_body") or {})
return _FakeCreateStream()

real_client = MagicMock()
Expand Down Expand Up @@ -3028,6 +3034,9 @@ def close(self): pass

def _create(**kwargs):
captured_kwargs.update(kwargs)
# #93650 routes bulk fields through extra_body; fold them back in
# so assertions read the effective wire body the SDK would send.
captured_kwargs.update(kwargs.get("extra_body") or {})
return _FakeCreateStream()

real_client = MagicMock()
Expand Down Expand Up @@ -3312,7 +3321,11 @@ def create(self, **kwargs):
fake_client = SimpleNamespace(responses=FakeResponses())
adapter = _CodexCompletionsAdapter(fake_client, "gpt-5.5")
adapter.create(messages=messages, model="gpt-5.5")
return fake_client.responses.kwargs
# #93650 routes bulk fields through extra_body; fold them back in so
# assertions read the effective wire body the SDK would send.
kwargs = dict(fake_client.responses.kwargs)
kwargs.update(kwargs.pop("extra_body", None) or {})
return kwargs

def test_tool_history_never_leaks_role_tool(self):
messages = [
Expand Down
157 changes: 157 additions & 0 deletions tests/run_agent/test_codex_sdk_transform_bypass.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,157 @@
"""Regression tests for the SDK request-transform bypass (#93650).

``responses.create`` re-walks the whole request body against the
``ResponseCreateParams`` union graph client-side, holding the GIL. #93650
documents that walk wedging for 12+ hours on a ~1.4 MB conversation and
freezing the entire agent — no in-process watchdog can fire while the GIL
is held, and no socket kill helps a pre-network hang. Bulk wire-format
fields are therefore routed through ``extra_body``, which the SDK merges
into the JSON body *after* the transform.
"""

import sys
import types
from types import SimpleNamespace
from unittest.mock import MagicMock

sys.modules.setdefault("fire", types.SimpleNamespace(Fire=lambda *a, **k: None))
sys.modules.setdefault("firecrawl", types.SimpleNamespace(Firecrawl=object))
sys.modules.setdefault("fal_client", types.SimpleNamespace())

from agent.codex_runtime import (
_bypass_sdk_request_transform,
_is_plain_json_data,
)


def _wire_kwargs():
return {
"model": "gpt-5.6-sol",
"instructions": "You are Hermes.",
"input": [
{"role": "user", "content": [{"type": "input_text", "text": "Ping"}]},
{"type": "function_call_output", "call_id": "c1", "output": "ok"},
],
"tools": [{"type": "function", "name": "terminal", "parameters": {}}],
"store": False,
"stream": True,
"timeout": 1800.0,
}


class TestIsPlainJsonData:
def test_accepts_nested_wire_payloads(self):
assert _is_plain_json_data(_wire_kwargs()["input"])

def test_rejects_non_json_leaves(self):
assert not _is_plain_json_data([{"role": "user", "content": object()}])

def test_rejects_non_string_dict_keys(self):
assert not _is_plain_json_data({1: "a"})

def test_rejects_generators(self):
assert not _is_plain_json_data((item for item in ()))


class TestBypassSdkRequestTransform:
def test_moves_bulk_fields_to_extra_body(self):
kwargs = _wire_kwargs()
original_input = kwargs["input"]

bypassed = _bypass_sdk_request_transform(kwargs)

assert "input" not in bypassed
assert "tools" not in bypassed
assert bypassed["extra_body"]["input"] is original_input
assert bypassed["extra_body"]["tools"] == kwargs["tools"]
# Scalar configuration stays on the typed path.
assert bypassed["model"] == "gpt-5.6-sol"
assert bypassed["stream"] is True
assert bypassed["timeout"] == 1800.0
# The caller's mapping is untouched.
assert kwargs["input"] is original_input
assert "extra_body" not in kwargs

def test_merges_with_existing_extra_body_and_keeps_caller_precedence(self):
kwargs = _wire_kwargs()
caller_extra = {"prompt_cache_retention": "24h", "input": "explicit-wins"}
kwargs["extra_body"] = caller_extra

bypassed = _bypass_sdk_request_transform(kwargs)

# An explicit extra_body entry wins, exactly as the SDK's
# post-transform merge would have resolved the collision.
assert bypassed["extra_body"]["input"] == "explicit-wins"
assert bypassed["extra_body"]["prompt_cache_retention"] == "24h"
assert bypassed["extra_body"]["tools"] == kwargs["tools"]
assert caller_extra == {
"prompt_cache_retention": "24h",
"input": "explicit-wins",
}

def test_non_json_field_stays_on_typed_sdk_path(self):
kwargs = _wire_kwargs()
kwargs["input"] = [{"role": "user", "content": object()}]

bypassed = _bypass_sdk_request_transform(kwargs)

assert bypassed["input"] == kwargs["input"]
assert bypassed["extra_body"] == {"tools": kwargs["tools"]}

def test_string_input_stays_in_place(self):
kwargs = _wire_kwargs()
kwargs["input"] = "plain prompt"
kwargs.pop("tools")

bypassed = _bypass_sdk_request_transform(kwargs)

assert bypassed is kwargs

def test_env_escape_hatch_restores_passthrough(self, monkeypatch):
monkeypatch.setenv("HERMES_CODEX_SDK_TRANSFORM", "1")
kwargs = _wire_kwargs()

assert _bypass_sdk_request_transform(kwargs) is kwargs


class TestRunCodexStreamRoutesPayloadViaExtraBody:
def _make_agent(self):
from run_agent import AIAgent

agent = AIAgent(
api_key="test-key",
base_url="https://chatgpt.com/backend-api/codex",
model="gpt-5.6-sol",
quiet_mode=True,
skip_context_files=True,
skip_memory=True,
)
agent._interrupt_requested = False
return agent

def test_create_receives_input_via_extra_body(self):
from agent.codex_runtime import run_codex_stream

agent = self._make_agent()
events = [
SimpleNamespace(
type="response.completed",
response=SimpleNamespace(
id="r1",
status="completed",
output=[],
usage=None,
),
)
]
mock_client = MagicMock()
mock_client.responses.create.return_value = iter(events)

run_codex_stream(agent, _wire_kwargs(), client=mock_client)

create_kwargs = mock_client.responses.create.call_args.kwargs
assert "input" not in create_kwargs
assert "tools" not in create_kwargs
assert create_kwargs["stream"] is True
assert create_kwargs["extra_body"]["input"][0]["role"] == "user"
assert create_kwargs["extra_body"]["tools"][0]["name"] == "terminal"
6 changes: 5 additions & 1 deletion tests/run_agent/test_native_compaction.py
Original file line number Diff line number Diff line change
Expand Up @@ -694,7 +694,11 @@ def _spy(messages, **kw):

class _Responses:
def create(self, **kwargs):
seen["input"] = kwargs.get("input")
# #93650 routes the bulk input around the SDK transform via
# extra_body; accept the payload in either wire shape.
seen["input"] = kwargs.get("input") or (
kwargs.get("extra_body") or {}
).get("input")
raise RuntimeError("stop before network")

class _Client:
Expand Down
7 changes: 6 additions & 1 deletion tests/run_agent/test_run_agent_codex_responses.py
Original file line number Diff line number Diff line change
Expand Up @@ -630,7 +630,12 @@ def _fake_create(**kwargs):
with caplog.at_level("WARNING", logger="agent.codex_runtime"):
agent._run_codex_stream(request)

assert "extra_body" not in captured
# The transform bypass (#93650) re-introduces extra_body to carry the bulk
# payload fields; the guard's contract is that retention itself never
# crosses the wire boundary in either shape.
assert "prompt_cache_retention" not in captured
assert "prompt_cache_retention" not in captured.get("extra_body", {})
assert "input" in captured.get("extra_body", {})
assert request["extra_body"] == {"prompt_cache_retention": "24h"}
assert any(
"Dropped unsupported prompt_cache_retention at consumer Codex wire boundary"
Expand Down
Loading