From 9d5a2fbbaecf503a5b9e9d12afb98019af46a095 Mon Sep 17 00:00:00 2001 From: BillyNour <298208912+BillyNour@users.noreply.github.com> Date: Wed, 5 Aug 2026 00:24:25 +1000 Subject: [PATCH] fix(agent): clamp aux Ultra effort for Codex and xAI Responses Map Hermes Ultra (and sibling above-high labels) to each Responses backend's accepted ceiling on the auxiliary Codex adapter path, and stamp current_issuer_kind so foreign encrypted reasoning is not replayed. --- agent/auxiliary_client.py | 30 ++++++++- tests/agent/test_auxiliary_client.py | 96 +++++++++++++++++++++++++++- 2 files changed, 122 insertions(+), 4 deletions(-) diff --git a/agent/auxiliary_client.py b/agent/auxiliary_client.py index 5b38c0ab82a63..e7ab1e7085b93 100644 --- a/agent/auxiliary_client.py +++ b/agent/auxiliary_client.py @@ -1327,8 +1327,23 @@ def create(self, **kwargs) -> Any: # build_kwargs, so they need the same guard applied independently. _host_for_input = str(getattr(self._client, "base_url", "") or "") _is_github_for_input = base_url_host_matches(_host_for_input, "githubcopilot.com") + _is_xai_for_input = ( + base_url_host_matches(_host_for_input, "x.ai") + or base_url_host_matches(_host_for_input, "api.x.ai") + ) + # Match main Codex transport issuer stamps so encrypted reasoning + # from a foreign Responses endpoint is not replayed here. + if _is_xai_for_input: + _issuer_kind = "xai_responses" + elif _is_github_for_input: + _issuer_kind = "github_responses" + else: + _issuer_kind = "codex_backend" input_items = _chat_messages_to_responses_input( - replay_messages, is_github_responses=_is_github_for_input, + replay_messages, + is_github_responses=_is_github_for_input, + is_xai_responses=_is_xai_for_input, + current_issuer_kind=_issuer_kind, ) resp_kwargs: Dict[str, Any] = { @@ -1371,8 +1386,17 @@ def create(self, **kwargs) -> Any: # Codex backend, which rejects e.g. {"effort": null} # with a 400. effort = reasoning_cfg.get("effort") or "medium" - # Codex backend rejects "minimal"; clamp to "low" to - # match the main-agent Codex transport behavior. + # Match agent/transports/codex.py effort clamps. + _effort_clamp = {"minimal": "low"} + if "gpt-5.6" in (model or "").lower(): + # Ultra is Hermes' GPT-5.6 product label; wire is max. + _effort_clamp["ultra"] = "max" + if _is_xai_for_input: + # xAI Responses tops out at high. + _effort_clamp.update( + {"xhigh": "high", "max": "high", "ultra": "high"} + ) + effort = _effort_clamp.get(str(effort).strip().lower(), effort) if effort == "minimal": effort = "low" resp_kwargs["reasoning"] = { diff --git a/tests/agent/test_auxiliary_client.py b/tests/agent/test_auxiliary_client.py index ce10a46c9b2e4..c6294d7f8e435 100644 --- a/tests/agent/test_auxiliary_client.py +++ b/tests/agent/test_auxiliary_client.py @@ -2790,7 +2790,7 @@ class TestCodexAdapterReasoningTranslation: """ @staticmethod - def _build_adapter(): + def _build_adapter(base_url="https://chatgpt.com/backend-api/codex"): """Build a _CodexCompletionsAdapter with a mocked responses.create().""" from agent.auxiliary_client import _CodexCompletionsAdapter from types import SimpleNamespace @@ -2829,12 +2829,106 @@ def _create(**kwargs): return _FakeCreateStream() real_client = MagicMock() + real_client.base_url = base_url real_client.responses.create = _create adapter = _CodexCompletionsAdapter(real_client, "gpt-5.3-codex") return adapter, captured_kwargs + def test_gpt_56_reasoning_effort_ultra_clamped_to_max(self): + """GPT-5.6 Codex rejects Hermes' Ultra label; emit its wire maximum.""" + adapter, captured = self._build_adapter() + adapter.create( + model="gpt-5.6-sol", + messages=[{"role": "user", "content": "hi"}], + extra_body={"reasoning": {"effort": "ultra"}}, + ) + assert captured.get("reasoning") == {"effort": "max", "summary": "auto"} + assert captured.get("include") == ["reasoning.encrypted_content"] + + def test_xai_reasoning_effort_ultra_clamped_to_high(self): + """xAI Responses rejects generic Ultra; emit the endpoint ceiling.""" + adapter, captured = self._build_adapter(base_url="https://api.x.ai/v1") + adapter.create( + model="grok-4.5", + messages=[{"role": "user", "content": "hi"}], + extra_body={"reasoning": {"effort": "ultra"}}, + ) + assert captured.get("reasoning") == {"effort": "high", "summary": "auto"} + assert captured.get("include") == ["reasoning.encrypted_content"] + + def test_xai_reasoning_effort_xhigh_and_max_clamped_to_high(self): + """xAI Responses also clamps sibling above-high labels to high.""" + for label in ("xhigh", "max"): + adapter, captured = self._build_adapter(base_url="https://api.x.ai/v1") + adapter.create( + model="grok-4.5", + messages=[{"role": "user", "content": "hi"}], + extra_body={"reasoning": {"effort": label}}, + ) + assert captured.get("reasoning") == { + "effort": "high", + "summary": "auto", + }, label + + def test_xai_auxiliary_drops_codex_issued_encrypted_reasoning(self): + """A GPT fallback must not poison the next MoA/xAI aggregator turn.""" + adapter, captured = self._build_adapter(base_url="https://api.x.ai/v1") + adapter.create( + model="grok-4.5", + messages=[ + {"role": "user", "content": "first"}, + { + "role": "assistant", + "content": "fallback answer", + "codex_reasoning_items": [ + { + "type": "reasoning", + "encrypted_content": "codex-sealed-blob", + "summary": [], + "_issuer_kind": "codex_backend", + } + ], + }, + {"role": "user", "content": "next"}, + ], + ) + assert not [ + item for item in captured["input"] + if item.get("type") == "reasoning" + ] + + def test_xai_auxiliary_keeps_xai_issued_encrypted_reasoning(self): + """The foreign-issuer guard must preserve normal xAI continuity.""" + adapter, captured = self._build_adapter(base_url="https://api.x.ai/v1") + adapter.create( + model="grok-4.5", + messages=[ + {"role": "user", "content": "first"}, + { + "role": "assistant", + "content": "grok answer", + "codex_reasoning_items": [ + { + "type": "reasoning", + "encrypted_content": "xai-sealed-blob", + "summary": [], + "_issuer_kind": "xai_responses", + } + ], + }, + {"role": "user", "content": "next"}, + ], + ) + reasoning = [ + item for item in captured["input"] + if item.get("type") == "reasoning" + ] + assert len(reasoning) == 1 + assert reasoning[0]["encrypted_content"] == "xai-sealed-blob" + assert "_issuer_kind" not in reasoning[0] + def test_reasoning_effort_low_passed_through(self): adapter, captured = self._build_adapter() adapter.create(