"""Tests for the ResponsesApiTransport (Codex).""" import json import pytest from types import SimpleNamespace from agent.transports import get_transport from agent.transports.types import NormalizedResponse @pytest.fixture def transport(): import agent.transports.codex # noqa: F401 return get_transport("codex_responses") class TestCodexTransportBasic: def test_convert_tools(self, transport): tools = [{ "type": "function", "function": { "name": "terminal", "description": "Run a command", "parameters": {"type": "object", "properties": {"command": {"type": "string"}}}, } }] result = transport.convert_tools(tools) assert len(result) == 1 assert result[0]["type"] == "function" assert result[0]["name"] == "terminal" class TestCodexBuildKwargs: @pytest.mark.parametrize("model", ["gpt-6-astra", "openai/gpt-6-astra"]) @pytest.mark.parametrize("effort, expected", [ ("low", "low"), ("medium", "medium"), ("high", "high"), ("xhigh", "xhigh"), ("max", "max"), ("ultra", "max"), ("minimal", "low"), ("none", None), ]) def test_astra_copilot_forwards_configured_reasoning(self, transport, model, effort, expected): from agent.reasoning_params import ReasoningParamsMixin from hermes_constants import resolve_reasoning_config reasoning = resolve_reasoning_config({"agent": {"reasoning_effort": effort}}, model) agent = SimpleNamespace(model=model, reasoning_config=reasoning) kw = transport.build_kwargs( model=model, messages=[{"role": "user", "content": "Hi"}], provider="github-copilot", is_github_responses=True, reasoning_config=reasoning, github_reasoning_extra=ReasoningParamsMixin._github_models_reasoning_extra_body(agent), ) assert kw.get("reasoning") == ({"effort": expected} if expected else None) def test_astra_direct_request_applies_model_contract_after_overrides(self, transport): kw = transport.build_kwargs( model="gpt-6-astra", messages=[{"role": "user", "content": "Hi"}], tools=[], base_url="https://api.openai.com/v1", reasoning_config={"enabled": False, "effort": "none"}, request_overrides={ "temperature": 0.4, "top_p": 0.9, "top_logprobs": 5, "logprobs": True, "reasoning": {"effort": "none"}, "include": ["reasoning.encrypted_content", "message.output_text.logprobs"], "prompt_cache_retention": "24h", }, ) assert kw["reasoning"]["effort"] == "low" assert kw["include"] == ["reasoning.encrypted_content"] for unsupported in ("temperature", "top_p", "top_logprobs", "logprobs", "prompt_cache_retention"): assert unsupported not in kw @pytest.mark.parametrize("effort", ["none", "minimal"]) def test_astra_normalizes_unsupported_low_efforts_after_overrides(self, transport, effort): kw = transport.build_kwargs( model="gpt-6-astra", messages=[{"role": "user", "content": "Hi"}], tools=[], base_url="https://api.openai.com/v1", request_overrides={"reasoning": {"effort": effort}}, ) assert kw["reasoning"]["effort"] == "low" @pytest.mark.parametrize("effort", ["low", "medium", "high", "xhigh", "max"]) def test_astra_accepts_complete_effort_ladder(self, transport, effort): kw = transport.build_kwargs( model="gpt-6-astra", messages=[{"role": "user", "content": "Hi"}], tools=[], base_url="https://api.openai.com/v1", reasoning_config={"enabled": True, "effort": effort}, ) assert kw["reasoning"]["effort"] == effort @pytest.mark.parametrize("base_url", ["https://responses.example.com/v1", "https://evil.api.openai.com/v1"]) def test_astra_contract_is_exact_host_only(self, transport, base_url): """Proxies and lookalike subdomains keep the generic Responses contract (effort passes through).""" kw = transport.build_kwargs( model="gpt-6-astra", messages=[{"role": "user", "content": "Hi"}], tools=[], base_url=base_url, reasoning_config={"enabled": True, "effort": "none"}, request_overrides={"temperature": 0.4}, ) assert kw["reasoning"]["effort"] == "none" assert kw["temperature"] == 0.4 @pytest.mark.parametrize( "base_url,is_codex", [ ("https://api.openai.com/v1", False), ("https://chatgpt.com/backend-api/codex", True), ("https://responses.example.com/v1", False), ], ) def test_prompt_cache_options_dropped_from_overrides( self, transport, monkeypatch, caplog, base_url, is_codex ): """``prompt_cache_options`` has no Responses.create() kwarg, so a top-level copy from request_overrides raises TypeError before any request is sent — on the official API, the Codex backend, and custom Responses endpoints alike. The ``extra_body`` escape hatch the warning points to still reaches the wire, and the warning fires once per process rather than every turn.""" import agent.transports.codex as codex_mod monkeypatch.setattr(codex_mod, "_PROMPT_CACHE_OPTIONS_DROP_WARNED", False) caplog.set_level("WARNING", logger=codex_mod.logger.name) for _ in range(2): kw = transport.build_kwargs( model="gpt-6-astra", messages=[{"role": "user", "content": "Hi"}], tools=[], base_url=base_url, is_codex_backend=is_codex, request_overrides={ "prompt_cache_options": {"ttl": "30m"}, "store": True, "extra_body": {"prompt_cache_options": {"ttl": "30m"}}, }, ) assert "prompt_cache_options" not in kw assert kw["store"] is True assert kw["extra_body"]["prompt_cache_options"] == {"ttl": "30m"} assert caplog.text.count("Dropped prompt_cache_options") == 1 assert "extra_body" in caplog.text def test_900k_context_variant_suffix_stripped_on_wire(self, transport): """``-900k`` large-context picker variants are Hermes-side aliases — the Codex backend only knows the base slug, so build_kwargs must strip the suffix from the wire model id.""" messages = [{"role": "user", "content": "Hi"}] kw = transport.build_kwargs( model="gpt-5.6-sol-900k", messages=messages, tools=[], params={"is_codex_backend": True}, ) assert kw["model"] == "gpt-5.6-sol" def test_base_slug_model_id_unchanged_on_wire(self, transport): messages = [{"role": "user", "content": "Hi"}] kw = transport.build_kwargs( model="gpt-5.6-sol", messages=messages, tools=[], params={"is_codex_backend": True}, ) assert kw["model"] == "gpt-5.6-sol" def test_cache_key_is_content_addressed_not_session_id(self, transport): """prompt_cache_key is content-addressed from the static prefix (instructions + tools), not the session_id. This keeps recurring cron jobs — whose session_id carries a per-fire timestamp — on a stable warm cache key. The key is a 'pck_' hash and must NOT equal session_id.""" messages = [{"role": "user", "content": "Hi"}] kw = transport.build_kwargs( model="gpt-5.4", messages=messages, tools=[], session_id="cron_job42_20260624_143000", ) pck = kw.get("prompt_cache_key", "") assert pck.startswith("pck_") assert pck != "cron_job42_20260624_143000" def test_cache_key_stable_across_session_ids(self, transport): """Same static prefix + different session_id (e.g. two cron fires of the same job) must yield the same prompt_cache_key — the whole point of the fix: repeated fires reuse the warm prefix instead of going cold.""" messages = [{"role": "user", "content": "Hi"}] kw1 = transport.build_kwargs( model="gpt-5.4", messages=messages, tools=[], session_id="cron_job42_20260624_143000", ) kw2 = transport.build_kwargs( model="gpt-5.4", messages=messages, tools=[], session_id="cron_job42_20260624_143500", ) assert kw1["prompt_cache_key"] == kw2["prompt_cache_key"] def test_cache_key_differs_across_unrelated_sessions(self, transport): """#78941: two unrelated sessions (different users/conversations) sharing the same static prefix must NOT collapse onto the same prompt_cache_key — session_id scopes the hash unless it is a cron per-fire id, which is normalized to its stable job prefix instead.""" messages = [{"role": "user", "content": "Hi"}] kw1 = transport.build_kwargs( model="gpt-5.4", messages=messages, tools=[], session_id="session_alice_1", ) kw2 = transport.build_kwargs( model="gpt-5.4", messages=messages, tools=[], session_id="session_bob_1", ) assert kw1["prompt_cache_key"] != kw2["prompt_cache_key"] def test_github_responses_drops_message_item_id_end_to_end(self, transport): # #32716: Copilot binds codex_message_items ids to a backend # "connection" that doesn't survive credential rotation, a gateway # restart, or load-balancer churn — replaying a stale id gets HTTP # 401 "input item ID does not belong to this connection", even for # ids well under the #27038 64-char length cap. build_kwargs must # thread is_github_responses through to the input converter so the # id never reaches the request. messages = [ {"role": "system", "content": "You are Hermes."}, { "role": "assistant", "content": "pong", "codex_message_items": [ { "type": "message", "role": "assistant", "status": "in_progress", "content": [{"type": "output_text", "text": "pong"}], "id": "msg_short_but_connection_scoped", "phase": "final_answer", } ], }, ] kw = transport.build_kwargs( model="gpt-5.5", messages=messages, tools=[], is_github_responses=True, ) message_item = next(item for item in kw["input"] if item.get("type") == "message") assert "id" not in message_item assert message_item["phase"] == "final_answer" assert message_item["status"] == "in_progress" assert message_item["content"] == [{"type": "output_text", "text": "pong"}] def test_non_github_responses_keeps_message_item_id_end_to_end(self, transport): messages = [ {"role": "system", "content": "You are Hermes."}, { "role": "assistant", "content": "pong", "codex_message_items": [ { "type": "message", "role": "assistant", "status": "completed", "content": [{"type": "output_text", "text": "pong"}], "id": "msg_short_id", } ], }, ] kw = transport.build_kwargs( model="gpt-5.5", messages=messages, tools=[], is_codex_backend=True, ) message_item = next(item for item in kw["input"] if item.get("type") == "message") assert message_item["id"] == "msg_short_id" def test_codex_backend_drops_foreign_message_item_id_end_to_end(self, transport): messages = [ { "role": "assistant", "content": "pong", "codex_message_items": [ { "type": "message", "role": "assistant", "status": "completed", "content": [{"type": "output_text", "text": "pong"}], "id": "123e4567-e89b-12d3-a456-426614174000", "phase": "final_answer", } ], }, ] kw = transport.build_kwargs( model="gpt-5.5", messages=messages, tools=[], is_codex_backend=True, ) message_item = next(item for item in kw["input"] if item.get("type") == "message") assert "id" not in message_item assert message_item["phase"] == "final_answer" @pytest.mark.parametrize(("is_codex_backend", "expected_user", "expected_assistant"), [ (True, [{"type": "input_text", "text": "hi"}], [{"type": "output_text", "text": "pong"}]), (False, "hi", "pong"), # other Responses routes keep the string shorthand they always sent ], ids=["codex-typed-parts", "other-route-string"]) def test_codex_backend_sends_typed_text_parts_for_string_content( self, transport, is_codex_backend, expected_user, expected_assistant, ): """#51512 (no-replay atom): the ChatGPT Codex backend 400s ``{"detail": "Unsupported content type"}`` on a role message whose ``content`` is a plain string, even with no reasoning replay in the request. Text must go out as typed ``input_text``/``output_text`` parts; the preflight the real call runs through must keep them.""" messages = [ {"role": "user", "content": "hi"}, {"role": "assistant", "content": "pong"}, {"role": "user", "content": "hi"}, ] kw = transport.build_kwargs(model="gpt-5.5", messages=messages, tools=[], is_codex_backend=is_codex_backend) kw = transport.preflight_kwargs(kw, sanitize_harmony_tokens=is_codex_backend) assert [item["content"] for item in kw["input"]] == [expected_user, expected_assistant, expected_user] @pytest.mark.parametrize("model", [ "gpt-5.5", "openai.gpt-5.5-pro", "openai/gpt-5.1-codex-2026-01-01", ]) def test_extended_cache_models_set_24h_prompt_cache_retention(self, transport, model): messages = [{"role": "user", "content": "Hi"}] kw = transport.build_kwargs( model=model, messages=messages, tools=[], session_id="test-session", base_url="https://bedrock-mantle.us-west-2.api.aws/v1", ) assert kw["prompt_cache_retention"] == "24h" @pytest.mark.parametrize("model", ["gpt-4o", "o3"]) def test_prompt_cache_retention_omitted_for_other_model_families(self, transport, model): kw = transport.build_kwargs( model=model, messages=[{"role": "user", "content": "Hi"}], tools=[], session_id="test-session", base_url="https://bedrock-mantle.us-west-2.api.aws/v1", ) assert "prompt_cache_retention" not in kw @pytest.mark.parametrize("base_url", [ "https://api.openai.com/v1", "https://example.openai.azure.com/openai/v1", "https://api.x.ai/v1", "https://models.github.ai/inference", "https://api.githubcopilot.com", "https://chatgpt.com/backend-api/codex", "https://responses.example.com/v1", "https://bedrock-mantle.us-west-2.api.aws.example/v1", "https://example.com/bedrock-mantle.us-west-2.api.aws/v1", ]) def test_prompt_cache_retention_omitted_for_non_mantle_endpoints(self, transport, base_url): kw = transport.build_kwargs( model="gpt-5.4", messages=[{"role": "user", "content": "Hi"}], tools=[], base_url=base_url, ) assert "prompt_cache_retention" not in kw def test_xai_responses_sends_cache_key_via_extra_body(self, transport): """xAI's Responses API documents ``prompt_cache_key`` as the body-level cache-routing key (the ``x-grok-conv-id`` header is Chat-Completions-only). Passing it via ``extra_body`` is robust against openai SDK builds whose ``Responses.stream()`` kwarg signature ever drops the field — the body field still serializes and reaches xAI either way. The ``x-grok-conv-id`` header is kept as a belt-and-braces fallback so cache routing survives even when the body field would be stripped by an intermediate proxy. Ref: https://docs.x.ai/developers/advanced-api-usage/prompt-caching/maximizing-cache-hits """ messages = [{"role": "user", "content": "Hi"}] kw = transport.build_kwargs( model="grok-4.3", messages=messages, tools=[], session_id="conv-xai-1", is_xai_responses=True, ) assert "prompt_cache_key" not in kw # Body-level prompt_cache_key is content-addressed (pck_ hash), not the # raw session_id, so recurring cron fires stay on a stable warm key. eb_pck = kw.get("extra_body", {}).get("prompt_cache_key", "") assert eb_pck.startswith("pck_") assert eb_pck != "conv-xai-1" # x-grok-conv-id stays the session/transcript id, not the cache key. assert kw.get("extra_headers", {}).get("x-grok-conv-id") == "conv-xai-1" def test_xai_responses_extra_body_preserves_caller_fields(self, transport): """When the caller already supplies ``extra_body`` (e.g. via request_overrides), the xAI cache-key injection must merge into the existing dict instead of overwriting it. Caller-supplied ``prompt_cache_key`` wins (setdefault semantics) so user overrides aren't silently clobbered by the transport.""" messages = [{"role": "user", "content": "Hi"}] kw = transport.build_kwargs( model="grok-4.3", messages=messages, tools=[], session_id="conv-xai-1", is_xai_responses=True, request_overrides={"extra_body": {"prompt_cache_key": "caller-override", "other_field": 42}}, ) eb = kw.get("extra_body", {}) assert eb.get("prompt_cache_key") == "caller-override" assert eb.get("other_field") == 42 # ── Azure Foundry post-tool reasoning suppression ────────────────── # # Foundry's Responses surface accepts the initial function-call request # and ordinary multi-turn continuity, but rejects the post-tool follow-up # payload when a replayed encrypted ``reasoning`` item sits alongside # ``function_call`` / ``function_call_output`` (HTTP 400 invalid_payload). # Suppression is scoped to that follow-up turn only. @staticmethod def _reasoning_item(): return {"type": "reasoning", "encrypted_content": "sealed", "summary": []} @classmethod def _post_tool_messages(cls): """user → assistant(tool_calls + reasoning) → tool result.""" return [ {"role": "user", "content": "Create a marker"}, { "role": "assistant", "content": "", "codex_reasoning_items": [cls._reasoning_item()], "tool_calls": [ { "id": "call_marker", "type": "function", "function": {"name": "write_marker", "arguments": "{}"}, } ], }, { "role": "tool", "tool_call_id": "call_marker", "content": "marker written", }, ] AZURE_FOUNDRY_BASE_URL = ( "https://placeholder.services.ai.azure.com/" "api/projects/placeholder/openai/v1" ) def test_post_tool_replay_preserves_reasoning_for_default_responses(self, transport): """Non-Azure Responses endpoints keep post-tool reasoning replay.""" kw = transport.build_kwargs( model="gpt-5.4", messages=self._post_tool_messages(), tools=[], replay_encrypted_reasoning=True, ) item_types = [item.get("type") for item in kw["input"] if isinstance(item, dict)] assert "reasoning" in item_types assert "function_call" in item_types assert "function_call_output" in item_types assert kw.get("include") == ["reasoning.encrypted_content"] @classmethod def _two_turn_messages(cls): """Two answered turns, each assistant row sealed by its own response, then a fresh user message.""" return [ {"role": "user", "content": "first"}, { "role": "assistant", "content": "one", "codex_reasoning_items": [{"type": "reasoning", "encrypted_content": "sealed-1", "summary": []}], }, {"role": "user", "content": "second"}, { "role": "assistant", "content": "two", "codex_reasoning_items": [{"type": "reasoning", "encrypted_content": "sealed-2", "summary": []}], }, {"role": "user", "content": "third"}, ] def test_azure_foundry_new_turn_replays_only_newest_reasoning(self, transport): """Two sealed prior responses on the wire is the 400 "Conflicting authenticated continuation identities" shape (#105369); one is accepted.""" kw = transport.build_kwargs( model="gpt-6-astra", messages=self._two_turn_messages(), tools=[], provider="azure-foundry", base_url="https://placeholder.openai.azure.com/openai/v1", replay_encrypted_reasoning=True, ) reasoning = [item for item in kw["input"] if item.get("type") == "reasoning"] assert [item["encrypted_content"] for item in reasoning] == ["sealed-2"] assert kw.get("include") == ["reasoning.encrypted_content"] assistant_text = [item for item in kw["input"] if item.get("role") == "assistant"] assert len(assistant_text) == 2 @pytest.mark.parametrize("base_url", [ "https://placeholder.openai.azure.com/openai/v1", "https://placeholder.services.ai.azure.com/api/projects/placeholder/openai/v1", ]) def test_azure_host_without_provider_replays_only_newest_reasoning(self, transport, base_url): """Both Azure hosts are detected without ``provider``; the rejection is not gateway-specific.""" kw = transport.build_kwargs( model="gpt-6-astra", messages=self._two_turn_messages(), tools=[], base_url=base_url, replay_encrypted_reasoning=True, ) reasoning = [item for item in kw["input"] if item.get("type") == "reasoning"] assert [item["encrypted_content"] for item in reasoning] == ["sealed-2"] def test_azure_trimmed_reasoning_turn_still_drops_its_message_id(self, transport): """The older turn's reasoning is trimmed for Azure (#105369) but its ``msg_*`` id is still bound to a ``rs_*`` id that is no longer on the wire; the id must go with it (#97427). The newest turn's id is dropped too (its reasoning replays without id); a reasoning-free turn keeps its id.""" def _turn(text, *, reasoning): msg = { "role": "assistant", "content": text, "codex_message_items": [{ "type": "message", "role": "assistant", "status": "completed", "id": f"msg_{text}", "content": [{"type": "output_text", "text": text}], }], } if reasoning: msg["codex_reasoning_items"] = [{"type": "reasoning", "id": f"rs_{text}", "encrypted_content": f"sealed-{text}", "summary": []}] return msg messages = [ {"role": "user", "content": "first"}, _turn("old", reasoning=True), {"role": "user", "content": "second"}, _turn("plain", reasoning=False), {"role": "user", "content": "third"}, _turn("new", reasoning=True), {"role": "user", "content": "fourth"}, ] kw = transport.build_kwargs( model="gpt-6-astra", messages=messages, tools=[], base_url="https://placeholder.openai.azure.com/openai/v1", replay_encrypted_reasoning=True, ) reasoning = [i for i in kw["input"] if i.get("type") == "reasoning"] assert [i["encrypted_content"] for i in reasoning] == ["sealed-new"] by_text = {i["content"][0]["text"]: i for i in kw["input"] if i.get("type") == "message" and i.get("role") == "assistant"} assert "id" not in by_text["old"] and "id" not in by_text["new"] assert by_text["plain"]["id"] == "msg_plain" assert "codex_reasoning_items" in messages[1] # canonical history untouched def test_default_responses_new_turn_replays_all_reasoning(self, transport): """Non-Azure Responses endpoints keep cross-turn reasoning replay.""" kw = transport.build_kwargs( model="gpt-5.4", messages=self._two_turn_messages(), tools=[], replay_encrypted_reasoning=True, ) reasoning = [item for item in kw["input"] if item.get("type") == "reasoning"] assert [item["encrypted_content"] for item in reasoning] == ["sealed-1", "sealed-2"] def test_normalize_response_stamps_wire_model_and_replay_drops_it_for_another_model(self, transport): """The wire model from build_kwargs (``-900k`` stripped) is what normalize_response stamps, and a later build_kwargs for another model on the same endpoint replays none of it (sealed blob → 400).""" transport.build_kwargs(model="gpt-5.6-sol-900k", messages=[{"role": "user", "content": "hi"}], tools=[]) normalized = transport.normalize_response(SimpleNamespace( status="completed", output=[ SimpleNamespace(type="reasoning", id="rs_1", encrypted_content="sealed-sol", summary=[]), SimpleNamespace(type="message", role="assistant", status="completed", id="msg_1", content=[SimpleNamespace(type="output_text", text="done")]), ], )) captured = normalized.codex_reasoning_items[0] assert captured["_issuer_model"] == "gpt-5.6-sol" history = [ {"role": "user", "content": "hi"}, {"role": "assistant", "content": "done", "codex_reasoning_items": [captured]}, {"role": "user", "content": "next"}, ] same = transport.build_kwargs(model="gpt-5.6-sol-900k", messages=history, tools=[], replay_encrypted_reasoning=True) other = transport.build_kwargs(model="gpt-5.7", messages=history, tools=[], replay_encrypted_reasoning=True) assert [i["encrypted_content"] for i in same["input"] if i.get("type") == "reasoning"] == ["sealed-sol"] assert not any(i.get("type") == "reasoning" for i in other["input"]) def test_newest_reasoning_only_keeps_compaction_checkpoints(self): from agent.transports.codex import _newest_reasoning_only checkpoint = {"type": "compaction", "encrypted_content": "ckpt", "summary": []} messages = [ {"role": "user", "content": "first"}, {"role": "assistant", "content": "one", "codex_reasoning_items": [checkpoint, self._reasoning_item()]}, {"role": "user", "content": "second"}, {"role": "assistant", "content": "two", "codex_reasoning_items": [self._reasoning_item()]}, {"role": "user", "content": "third"}, ] pruned = _newest_reasoning_only(messages) assert pruned[1]["codex_reasoning_items"] == [checkpoint] assert pruned[3]["codex_reasoning_items"] == [self._reasoning_item()] assert len(messages[1]["codex_reasoning_items"]) == 2 def test_azure_foundry_post_tool_replay_suppresses_reasoning_items(self, transport): """The rejected payload shape drops reasoning, keeps tool continuity.""" kw = transport.build_kwargs( model="gpt-5.4", messages=self._post_tool_messages(), tools=[], provider="azure-foundry", base_url=self.AZURE_FOUNDRY_BASE_URL, replay_encrypted_reasoning=True, ) item_types = [item.get("type") for item in kw["input"] if isinstance(item, dict)] assert "reasoning" not in item_types assert "function_call" in item_types assert "function_call_output" in item_types assert kw.get("include") == [] def test_azure_foundry_detected_by_host_without_provider(self, transport): """Foundry detection works on the endpoint host alone.""" kw = transport.build_kwargs( model="gpt-5.4", messages=self._post_tool_messages(), tools=[], base_url=self.AZURE_FOUNDRY_BASE_URL, replay_encrypted_reasoning=True, ) item_types = [item.get("type") for item in kw["input"] if isinstance(item, dict)] assert "reasoning" not in item_types @pytest.mark.parametrize( "base_url", [ "https://proxy.example.com/.services.ai.azure.com/openai/v1", "https://openrouter.ai/api/v1?upstream=.services.ai.azure.com", "https://services.ai.azure.com.evil.example/v1", ], ) def test_non_foundry_host_lookalikes_keep_reasoning(self, transport, base_url): """A Foundry domain in a path/query/suffix is not a Foundry endpoint. Guards against the substring-match false positive: these URLs all contain the Foundry domain but are served by someone else, and suppressing their reasoning replay would silently degrade cross-turn coherence on an unrelated provider. """ kw = transport.build_kwargs( model="gpt-5.4", messages=self._post_tool_messages(), tools=[], base_url=base_url, replay_encrypted_reasoning=True, ) item_types = [item.get("type") for item in kw["input"] if isinstance(item, dict)] assert "reasoning" in item_types assert kw.get("include") == ["reasoning.encrypted_content"] def test_azure_foundry_non_tool_follow_up_preserves_reasoning_items(self, transport): """Ordinary (non-tool) Azure Foundry continuity is unchanged. A plain assistant reasoning turn followed by another user message has no tool continuity, so the encrypted reasoning item must still be replayed — Foundry only rejects the post-tool payload. """ messages = [ {"role": "user", "content": "Explain recursion"}, { "role": "assistant", "content": "Recursion is when a function calls itself.", "codex_reasoning_items": [self._reasoning_item()], }, {"role": "user", "content": "Give an example"}, ] kw = transport.build_kwargs( model="gpt-5.4", messages=messages, tools=[], provider="azure-foundry", base_url=self.AZURE_FOUNDRY_BASE_URL, replay_encrypted_reasoning=True, ) item_types = [item.get("type") for item in kw["input"] if isinstance(item, dict)] assert "reasoning" in item_types assert "function_call" not in item_types assert "function_call_output" not in item_types assert kw.get("include") == ["reasoning.encrypted_content"] def test_azure_foundry_user_turn_after_completed_tool_call_keeps_reasoning( self, transport ): """Suppression must not stick for the rest of the conversation. The tool call completed and the assistant already answered; this turn is a plain user follow-up whose payload ends on a user message, which Foundry accepts. A predicate that scanned the whole history for any tool call plus any tool result would suppress reasoning here — and on every later turn — which is the all-turns behavior this scoping exists to avoid. """ messages = self._post_tool_messages() + [ { "role": "assistant", "content": "Marker created.", "codex_reasoning_items": [self._reasoning_item()], }, {"role": "user", "content": "Now explain recursion"}, ] kw = transport.build_kwargs( model="gpt-5.4", messages=messages, tools=[], provider="azure-foundry", base_url=self.AZURE_FOUNDRY_BASE_URL, replay_encrypted_reasoning=True, ) item_types = [item.get("type") for item in kw["input"] if isinstance(item, dict)] assert "reasoning" in item_types assert "function_call" in item_types assert "function_call_output" in item_types assert kw.get("include") == ["reasoning.encrypted_content"] def test_azure_foundry_parallel_tool_results_suppress_reasoning(self, transport): """A trailing run of parallel tool results is still the rejected shape.""" messages = [ {"role": "user", "content": "Read both files"}, { "role": "assistant", "content": "", "codex_reasoning_items": [self._reasoning_item()], "tool_calls": [ { "id": "call_a", "type": "function", "function": {"name": "read_file", "arguments": "{}"}, }, { "id": "call_b", "type": "function", "function": {"name": "read_file", "arguments": "{}"}, }, ], }, {"role": "tool", "tool_call_id": "call_a", "content": "a"}, {"role": "tool", "tool_call_id": "call_b", "content": "b"}, ] kw = transport.build_kwargs( model="gpt-5.4", messages=messages, tools=[], provider="azure-foundry", base_url=self.AZURE_FOUNDRY_BASE_URL, replay_encrypted_reasoning=True, ) item_types = [item.get("type") for item in kw["input"] if isinstance(item, dict)] assert "reasoning" not in item_types assert item_types.count("function_call_output") == 2 def test_azure_foundry_respects_caller_replay_disabled(self, transport): """An explicit replay_encrypted_reasoning=False is not re-enabled.""" kw = transport.build_kwargs( model="gpt-5.4", messages=self._post_tool_messages(), tools=[], provider="azure-foundry", base_url=self.AZURE_FOUNDRY_BASE_URL, replay_encrypted_reasoning=False, ) item_types = [item.get("type") for item in kw["input"] if isinstance(item, dict)] assert "reasoning" not in item_types assert kw.get("include") == [] @pytest.mark.parametrize( "tool_call,tool_call_id", [ # Responses histories carry the function call id in call_id while # ``id`` holds the response item id. Resumed legacy sessions and # host-fed histories still use this shape. ({"id": "fc_item_a", "call_id": "call_a"}, "call_a"), # Plain chat-completions shape: id IS the call id. ({"id": "call_a"}, "call_a"), # Bare fc_ id with no call_id — the converter derives call_. ({"id": "fc_a"}, "call_a"), # Composite stored id, on either side of the pairing. ({"id": "call_a|fc_a"}, "call_a"), ({"id": "call_a"}, "call_a|fc_a"), # call_id present, no id at all. ({"call_id": "call_a"}, "call_a"), ], ) def test_azure_foundry_suppresses_across_tool_call_id_shapes( self, transport, tool_call, tool_call_id ): """Every id shape the converter can pair must be detected. The converter resolves a function call's identity as ``call_id`` -> embedded ``id`` -> derived from an ``fc_`` item id, and splits composite ``"call_x|fc_y"`` ids. A predicate that matched only ``tool_calls[*].id`` would miss the id=fc_ / call_id=call_ shape: the converter still emits paired function_call / function_call_output, so the exact payload Foundry rejects would ship with the reasoning item intact. """ messages = [ {"role": "user", "content": "Create a marker"}, { "role": "assistant", "content": "", "codex_reasoning_items": [self._reasoning_item()], "tool_calls": [ {**tool_call, "type": "function", "function": {"name": "write_marker", "arguments": "{}"}} ], }, {"role": "tool", "tool_call_id": tool_call_id, "content": "marker written"}, ] kw = transport.build_kwargs( model="gpt-5.4", messages=messages, tools=[], provider="azure-foundry", base_url=self.AZURE_FOUNDRY_BASE_URL, replay_encrypted_reasoning=True, ) item_types = [item.get("type") for item in kw["input"] if isinstance(item, dict)] # The converter paired them, so this is the rejected shape. assert "function_call" in item_types assert "function_call_output" in item_types assert "reasoning" not in item_types assert kw.get("include") == [] def test_azure_foundry_unpaired_tool_result_keeps_reasoning(self, transport): """A tool result that pairs with nothing is not the rejected shape.""" messages = [ {"role": "user", "content": "Create a marker"}, { "role": "assistant", "content": "", "codex_reasoning_items": [self._reasoning_item()], "tool_calls": [ { "id": "fc_item_x", "call_id": "call_x", "type": "function", "function": {"name": "write_marker", "arguments": "{}"}, } ], }, {"role": "tool", "tool_call_id": "call_unrelated", "content": "?"}, ] kw = transport.build_kwargs( model="gpt-5.4", messages=messages, tools=[], provider="azure-foundry", base_url=self.AZURE_FOUNDRY_BASE_URL, replay_encrypted_reasoning=True, ) item_types = [item.get("type") for item in kw["input"] if isinstance(item, dict)] assert "reasoning" in item_types assert kw.get("include") == ["reasoning.encrypted_content"] def test_xai_top_level_override_also_governs_extra_body(self, transport): """A caller's top-level request_overrides={"prompt_cache_key": ...} must win in extra_body.prompt_cache_key too -- the field xAI actually reads -- instead of being silently outrun by the auto-derived content-hash cache_key (#78941).""" messages = [{"role": "user", "content": "Hi"}] kw = transport.build_kwargs( model="grok-4.3", messages=messages, tools=[], session_id="conv-xai-1", is_xai_responses=True, request_overrides={"prompt_cache_key": "caller-top-level"}, ) assert kw["prompt_cache_key"] == "caller-top-level" assert kw["extra_body"]["prompt_cache_key"] == "caller-top-level" @pytest.mark.parametrize("length", [64, 65]) def test_codex_cache_scope_boundary(self, transport, length): session_id = "s" * length kw = transport.build_kwargs( model="gpt-5.4", messages=[{"role": "user", "content": "Hi"}], tools=[], session_id=session_id, is_codex_backend=True, request_overrides={"extra_headers": {"x-test": "1"}}, ) headers = kw["extra_headers"] assert headers["x-test"] == "1" # session_id header carries the raw physical id untouched regardless # of length (#57012); x-client-request-id mirrors the body's # effective (already-bounded) prompt_cache_key. assert headers["session_id"] == session_id assert headers["x-client-request-id"] == kw["prompt_cache_key"] assert len(headers["x-client-request-id"]) <= 64 def test_codex_cache_scope_headers_normalize_cron_session_id(self, transport): """x-client-request-id shares a cache scope across cron re-fires of the same job (cron per-fire timestamp stripped, same as prompt_cache_key), while session_id stays the raw per-fire physical id (#57012).""" first_run = transport.build_kwargs( model="gpt-5.4", messages=[{"role": "user", "content": "Hi"}], tools=[], session_id="cron_job42_20260801_090000", is_codex_backend=True, )["extra_headers"] second_run = transport.build_kwargs( model="gpt-5.4", messages=[{"role": "user", "content": "Hi"}], tools=[], session_id="cron_job42_20260802_090000", is_codex_backend=True, )["extra_headers"] other_job = transport.build_kwargs( model="gpt-5.4", messages=[{"role": "user", "content": "Hi"}], tools=[], session_id="cron_job99_20260801_090000", is_codex_backend=True, )["extra_headers"] assert first_run["session_id"] == "cron_job42_20260801_090000" assert second_run["session_id"] == "cron_job42_20260802_090000" assert first_run["x-client-request-id"].startswith("pck_") assert first_run["x-client-request-id"] == second_run["x-client-request-id"] assert first_run["x-client-request-id"] != other_job["x-client-request-id"] def test_xai_injects_native_web_search_when_client_web_search_present(self, transport, monkeypatch): """When the active/configured search backend is xAI, swap client ``web_search`` for Grok's native built-in so server-side search completes (otherwise the turn stalls as incomplete → 3 retries). Non-conflicting client tools are preserved. """ import agent.transports.codex as codex_mod monkeypatch.setattr(codex_mod, "_xai_prefers_native_web_search", lambda: True) messages = [{"role": "user", "content": "Find current prices."}] kw = transport.build_kwargs( model="grok-composer-2.5-fast", messages=messages, tools=[ {"type": "function", "function": { "name": "read_file", "description": "Read a file.", "parameters": {"type": "object", "properties": {"path": {"type": "string"}}}}}, {"type": "function", "function": { "name": "web_search", "description": "Search the web.", "parameters": {"type": "object", "properties": {"query": {"type": "string"}}}}}, ], is_xai_responses=True, ) tool_types = [t.get("type") for t in kw.get("tools", [])] assert "web_search" in tool_types, kw.get("tools") # Non-conflicting client-side tools are preserved. names = [t.get("name") for t in kw.get("tools", []) if t.get("type") == "function"] assert "read_file" in names assert "web_search" not in names assert "hermes_web_search" not in names def test_xai_renames_client_web_search_when_firecrawl_configured(self, transport, monkeypatch): """Configured Firecrawl (or any non-xai backend) must keep Hermes dispatch — rename the wire tool so Grok cannot hijack ``web_search``. """ import agent.transports.codex as codex_mod monkeypatch.setattr(codex_mod, "_xai_prefers_native_web_search", lambda: False) messages = [{"role": "user", "content": "Find current prices."}] kw = transport.build_kwargs( model="grok-4.5", messages=messages, tools=[ {"type": "function", "function": { "name": "read_file", "description": "Read a file.", "parameters": {"type": "object", "properties": {"path": {"type": "string"}}}}}, {"type": "function", "function": { "name": "web_search", "description": "Search the web.", "parameters": {"type": "object", "properties": {"query": {"type": "string"}}}}}, ], is_xai_responses=True, ) tools = kw.get("tools", []) assert not any(t.get("type") == "web_search" for t in tools), tools names = [t.get("name") for t in tools if t.get("type") == "function"] assert "read_file" in names assert "hermes_web_search" in names assert "web_search" not in names def test_xai_does_not_inject_native_web_search_without_client_web_search(self, transport): """The native ``web_search`` built-in is a 1:1 swap for an already-requested client ``web_search`` — NOT an additive grant. A turn whose toolset has no ``web_search`` (user never enabled the web toolset) must not get Grok server-side search force-injected, which would silently bypass Hermes's web-provider config and tool-trace plumbing for every xai-oauth turn. """ messages = [{"role": "user", "content": "Read this file."}] kw = transport.build_kwargs( model="grok-composer-2.5-fast", messages=messages, tools=[{"type": "function", "function": { "name": "read_file", "description": "Read a file.", "parameters": {"type": "object", "properties": {"path": {"type": "string"}}}}}], is_xai_responses=True, ) tools = kw.get("tools", []) assert not any(t.get("type") == "web_search" for t in tools), tools names = [t.get("name") for t in tools if t.get("type") == "function"] assert "read_file" in names def test_non_xai_path_does_not_inject_native_web_search(self, transport): """Native web_search injection is scoped to xAI — Codex/GitHub paths keep the client-side web_search function untouched.""" messages = [{"role": "user", "content": "Search."}] kw = transport.build_kwargs( model="gpt-5.4", messages=messages, tools=[{"type": "function", "function": { "name": "web_search", "description": "Search the web.", "parameters": {"type": "object", "properties": {"query": {"type": "string"}}}}}], is_xai_responses=False, ) tools = kw.get("tools", []) assert not any(t.get("type") == "web_search" for t in tools) assert any( t.get("type") == "function" and t.get("name") == "web_search" for t in tools ) # --- OpenAI Codex native web-search swap --- # The Codex Responses endpoint exposes the same server-executed # ``web_search`` built-in as xAI, so selecting the ``openai-native`` # backend performs the same 1:1 swap. Unlike xAI there is no alias path: # an unselected or non-Codex request keeps the client-side function, so a # custom OpenAI-compatible endpoint never receives a tool it cannot host. def test_openai_native_swaps_client_web_search_for_builtin(self, transport, monkeypatch): """Selecting ``openai-native`` replaces the client ``web_search`` function with the provider-executed built-in. ``web_extract`` is a separate capability and must survive — native search covers search only. """ import agent.transports.codex as codex_mod monkeypatch.setattr(codex_mod, "_openai_prefers_native_web_search", lambda: True) kw = transport.build_kwargs( model="gpt-5.6-sol", messages=[{"role": "user", "content": "Find current prices."}], tools=[ {"type": "function", "function": { "name": "read_file", "description": "Read a file.", "parameters": {"type": "object", "properties": {"path": {"type": "string"}}}}}, {"type": "function", "function": { "name": "web_search", "description": "Search the web.", "parameters": {"type": "object", "properties": {"query": {"type": "string"}}}}}, {"type": "function", "function": { "name": "web_extract", "description": "Extract a page.", "parameters": {"type": "object", "properties": {"url": {"type": "string"}}}}}, ], is_codex_backend=True, ) tools = kw.get("tools", []) assert any(t.get("type") == "web_search" for t in tools), tools names = [t.get("name") for t in tools if t.get("type") == "function"] assert "web_search" not in names assert "read_file" in names assert "web_extract" in names def test_openai_native_not_selected_keeps_client_web_search(self, transport, monkeypatch): """A Codex turn that has not selected ``openai-native`` keeps Hermes dispatch — the built-in must never be granted additively.""" import agent.transports.codex as codex_mod monkeypatch.setattr(codex_mod, "_openai_prefers_native_web_search", lambda: False) kw = transport.build_kwargs( model="gpt-5.6-sol", messages=[{"role": "user", "content": "Find current prices."}], tools=[{"type": "function", "function": { "name": "web_search", "description": "Search the web.", "parameters": {"type": "object", "properties": {"query": {"type": "string"}}}}}], is_codex_backend=True, ) tools = kw.get("tools", []) assert not any(t.get("type") == "web_search" for t in tools), tools names = [t.get("name") for t in tools if t.get("type") == "function"] assert "web_search" in names def test_openai_native_ignored_on_non_codex_transport(self, transport, monkeypatch): """The provider-executed built-in only exists on ``chatgpt.com/backend-api/codex``. A custom OpenAI-compatible endpoint must keep the client tool even when the search backend says native. """ import agent.transports.codex as codex_mod monkeypatch.setattr(codex_mod, "_openai_prefers_native_web_search", lambda: True) kw = transport.build_kwargs( model="gpt-5.6-sol", messages=[{"role": "user", "content": "Find current prices."}], tools=[{"type": "function", "function": { "name": "web_search", "description": "Search the web.", "parameters": {"type": "object", "properties": {"query": {"type": "string"}}}}}], is_codex_backend=False, ) tools = kw.get("tools", []) assert not any(t.get("type") == "web_search" for t in tools), tools names = [t.get("name") for t in tools if t.get("type") == "function"] assert "web_search" in names # --- Grok reasoning-effort capability allowlist --- # api.x.ai 400s with "Model X does not support parameter reasoningEffort" # on grok-4 / grok-4-fast / grok-3 / grok-code-fast / grok-4.20-0309-*. # Those models reason natively but don't expose the dial. The transport # must omit the `reasoning` key for them. As of May 2026 we DO request # ``reasoning.encrypted_content`` back from xAI on every model — # see test_xai_reasoning_effort_passed for the rationale. def test_xai_grok_4_20_0309_variants_omit_reasoning_effort(self, transport): """grok-4.20-0309-(non-)reasoning reject the effort dial. Counterintuitively, only grok-4.20-multi-agent-0309 accepts it. """ messages = [{"role": "user", "content": "Hi"}] for model in ("grok-4.20-0309-reasoning", "grok-4.20-0309-non-reasoning"): kw = transport.build_kwargs( model=model, messages=messages, tools=[], is_xai_responses=True, reasoning_config={"effort": "high"}, ) assert "reasoning" not in kw, f"{model} must not receive reasoning" class TestResponsesReservedToolAliases: """Responses providers may reserve web_search / search_files as function names. The transport aliases them on the wire and maps them back.""" @pytest.fixture def transport(self): from agent.transports.codex import ResponsesApiTransport return ResponsesApiTransport() _TOOLS = [ {"type": "function", "function": { "name": "search_files", "description": "Search files.", "parameters": {"type": "object", "properties": {"pattern": {"type": "string"}}}}}, {"type": "function", "function": { "name": "web_search", "description": "Search the web.", "parameters": {"type": "object", "properties": {"query": {"type": "string"}}}}}, {"type": "function", "function": { "name": "read_file", "description": "Read a file.", "parameters": {"type": "object", "properties": {"path": {"type": "string"}}}}}, ] _PERPLEXITY_ONLY_TOOLS = [ {"type": "function", "function": { "name": name, "description": f"Client {name}.", "parameters": {"type": "object", "properties": {}}}} for name in ("fetch_url", "people_search", "finance_search") ] def _names(self, kw): return [t.get("name") for t in kw.get("tools", []) if t.get("type") == "function"] def test_builtin_opencode_go_aliases_reserved_names(self, transport): kw = transport.build_kwargs( model="grok-4.5", messages=[{"role": "user", "content": "hi"}], tools=list(self._TOOLS), provider="opencode-go", base_url="https://opencode.ai/zen/go/v1", ) names = self._names(kw) assert "hermes_search_files" in names assert "hermes_web_search" in names assert "search_files" not in names assert "web_search" not in names assert "read_file" in names # non-reserved untouched def test_custom_opencode_family_provider_aliases_reserved_names(self, transport): """Custom opencode-go-* providers get the same aliasing (#85589).""" kw = transport.build_kwargs( model="grok-4.5", messages=[{"role": "user", "content": "hi"}], tools=list(self._TOOLS), provider="opencode-go-bridge", base_url="https://opencode.ai/zen/go/v1", ) names = self._names(kw) assert "hermes_search_files" in names assert "search_files" not in names def test_opencode_host_match_without_family_provider(self, transport): """An arbitrary custom provider pointing at opencode.ai still aliases.""" kw = transport.build_kwargs( model="gpt-5.6-luna", messages=[{"role": "user", "content": "hi"}], tools=list(self._TOOLS), provider="my-oc-proxy", base_url="https://opencode.ai/zen/go/v1", ) names = self._names(kw) assert "hermes_search_files" in names assert "hermes_web_search" in names def test_non_opencode_backend_keeps_original_names(self, transport): kw = transport.build_kwargs( model="gpt-5.4", messages=[{"role": "user", "content": "hi"}], tools=list(self._TOOLS), provider="openai-codex", base_url="https://api.openai.com/v1", ) names = self._names(kw) assert "search_files" in names assert "web_search" in names assert "hermes_search_files" not in names def test_perplexity_agent_api_aliases_reserved_names(self, transport, monkeypatch): kw = transport.build_kwargs( model="perplexity/sonar", messages=[{"role": "user", "content": "hi"}], tools=list(self._TOOLS) + list(self._PERPLEXITY_ONLY_TOOLS), provider="custom", base_url="https://api.perplexity.ai/v1", ) names = self._names(kw) assert "hermes_search_files" in names assert "hermes_web_search" in names assert "search_files" not in names assert "web_search" not in names assert "hermes_fetch_url" in names assert "hermes_people_search" in names assert "hermes_finance_search" in names assert "fetch_url" not in names assert "people_search" not in names assert "finance_search" not in names assert "read_file" in names assert transport._last_wire_aliases == { "hermes_search_files": "search_files", "hermes_web_search": "web_search", "hermes_fetch_url": "fetch_url", "hermes_people_search": "people_search", "hermes_finance_search": "finance_search", } msg = SimpleNamespace( content=None, reasoning=None, tool_calls=[SimpleNamespace( id="call_1", call_id="call_1", response_item_id="fc_1", function=SimpleNamespace( name="hermes_search_files", arguments='{"pattern":"README"}', ), )], codex_reasoning_items=None, codex_message_items=None, reasoning_details=None, ) monkeypatch.setattr( "agent.codex_responses_adapter._normalize_codex_response", lambda resp, issuer_kind=None, issuer_model=None: (msg, "tool_calls"), ) normalized = transport.normalize_response(SimpleNamespace(output=[], status="completed")) assert [tc.name for tc in normalized.tool_calls] == ["search_files"] def test_perplexity_lookalike_host_keeps_original_names(self, transport): for base_url in ( "https://sub.api.perplexity.ai/v1", "https://api.perplexity.ai.example.com/v1", "https://example.com/api.perplexity.ai/v1?host=api.perplexity.ai", ): kw = transport.build_kwargs( model="perplexity/sonar", messages=[{"role": "user", "content": "hi"}], tools=list(self._TOOLS), provider="custom", base_url=base_url, ) names = self._names(kw) assert "search_files" in names assert "web_search" in names assert "hermes_search_files" not in names def test_normalize_maps_reserved_aliases_back(self, transport, monkeypatch): msg = SimpleNamespace( content=None, reasoning=None, tool_calls=[ SimpleNamespace( id="call_1", call_id="call_1", response_item_id="fc_1", function=SimpleNamespace( name="hermes_search_files", arguments='{"pattern":"README"}', ), ), SimpleNamespace( id="call_2", call_id="call_2", response_item_id="fc_2", function=SimpleNamespace( name="hermes_web_search", arguments='{"query":"hermes"}', ), ), ], codex_reasoning_items=None, codex_message_items=None, reasoning_details=None, ) response = SimpleNamespace(output=[], status="completed") monkeypatch.setattr( "agent.codex_responses_adapter._normalize_codex_response", lambda resp, issuer_kind=None, issuer_model=None: (msg, "tool_calls"), ) normalized = transport.normalize_response(response) names = [tc.name for tc in normalized.tool_calls] assert names == ["search_files", "web_search"] class TestXaiReservedToolSearchAlias: """xAI reserves ``tool_search`` for Grok's native Tool Search and rejects the client declaration with HTTP 400 (#95003). The transport aliases the progressive-disclosure bridge on the wire and maps it back on dispatch.""" @pytest.fixture def transport(self): from agent.transports.codex import ResponsesApiTransport return ResponsesApiTransport() _TOOLS = [ {"type": "function", "function": { "name": "tool_search", "description": "Search deferred tools.", "parameters": {"type": "object", "properties": {"query": {"type": "string"}}}}}, {"type": "function", "function": { "name": "tool_describe", "description": "Describe a deferred tool.", "parameters": {"type": "object", "properties": {"name": {"type": "string"}}}}}, {"type": "function", "function": { "name": "read_file", "description": "Read a file.", "parameters": {"type": "object", "properties": {"path": {"type": "string"}}}}}, ] def _names(self, kw): return [t.get("name") for t in kw.get("tools", []) if t.get("type") == "function"] def test_xai_aliases_reserved_tool_search(self, transport): kw = transport.build_kwargs( model="grok-4.6", messages=[{"role": "user", "content": "hi"}], tools=list(self._TOOLS), is_xai_responses=True, ) names = self._names(kw) assert "hermes_tool_search" in names assert "tool_search" not in names # Only ``tool_search`` is reserved — the sibling bridge tools and # ordinary tools go out untouched. assert "tool_describe" in names assert "read_file" in names def test_openai_responses_aliases_reserved_tool_search(self, transport): """OpenAI Responses reserves the ``tool_search`` namespace for its native Tool Search (#83122): both the ChatGPT Codex backend and api.openai.com get the bridge under ``hermes_tool_search``.""" for extra in ({"is_codex_backend": True}, {"base_url": "https://api.openai.com/v1"}): kw = transport.build_kwargs( model="gpt-5.4", messages=[{"role": "user", "content": "hi"}], tools=list(self._TOOLS), **extra, ) names = self._names(kw) assert "hermes_tool_search" in names and "tool_search" not in names, extra assert transport._last_wire_aliases == {"hermes_tool_search": "tool_search"} def test_other_responses_backend_keeps_tool_search_name(self, transport): kw = transport.build_kwargs( model="gpt-5.4", messages=[{"role": "user", "content": "hi"}], tools=list(self._TOOLS), base_url="https://responses-proxy.example.invalid/v1", ) names = self._names(kw) assert "tool_search" in names assert "hermes_tool_search" not in names def test_alias_composes_with_native_web_search_swap(self, transport, monkeypatch): """The bridge alias must survive the xAI web_search branch (#48108).""" import agent.transports.codex as codex_mod monkeypatch.setattr(codex_mod, "_xai_prefers_native_web_search", lambda: True) tools = list(self._TOOLS) + [ {"type": "function", "function": { "name": "web_search", "description": "Search the web.", "parameters": {"type": "object", "properties": {"query": {"type": "string"}}}}}, ] kw = transport.build_kwargs( model="grok-4.6", messages=[{"role": "user", "content": "hi"}], tools=tools, is_xai_responses=True, ) assert any(t.get("type") == "web_search" for t in kw.get("tools", [])) names = self._names(kw) assert "hermes_tool_search" in names assert "tool_search" not in names def test_normalize_maps_tool_search_alias_back(self, transport, monkeypatch): msg = SimpleNamespace( content=None, reasoning=None, tool_calls=[ SimpleNamespace( id="call_1", call_id="call_1", response_item_id="fc_1", function=SimpleNamespace( name="hermes_tool_search", arguments='{"query":"create github issue"}', ), ), ], codex_reasoning_items=None, codex_message_items=None, reasoning_details=None, ) response = SimpleNamespace(output=[], status="completed") monkeypatch.setattr( "agent.codex_responses_adapter._normalize_codex_response", lambda resp, issuer_kind=None, issuer_model=None: (msg, "tool_calls"), ) # Pair the response with a real request so provenance is recorded. transport.build_kwargs( model="grok-4.6", messages=[{"role": "user", "content": "hi"}], tools=list(self._TOOLS), is_xai_responses=True, ) assert transport._last_wire_aliases == {"hermes_tool_search": "tool_search"} normalized = transport.normalize_response(response) assert [tc.name for tc in normalized.tool_calls] == ["tool_search"] def _normalize_named_call(self, transport, monkeypatch, wire_name): msg = SimpleNamespace( content=None, reasoning=None, tool_calls=[ SimpleNamespace( id="call_1", call_id="call_1", response_item_id="fc_1", function=SimpleNamespace(name=wire_name, arguments="{}"), ), ], codex_reasoning_items=None, codex_message_items=None, reasoning_details=None, ) response = SimpleNamespace(output=[], status="completed") monkeypatch.setattr( "agent.codex_responses_adapter._normalize_codex_response", lambda resp, issuer_kind=None, issuer_model=None: (msg, "tool_calls"), ) return transport.normalize_response(response) def test_no_alias_emitted_means_no_reverse_rewrite(self, transport, monkeypatch): """Provenance contract (#95003 review): a request that emitted no aliases must not have a real ``hermes_tool_search`` tool rewritten.""" real_tool = {"type": "function", "function": { "name": "hermes_tool_search", "description": "A real MCP tool.", "parameters": {"type": "object", "properties": {}}}} transport.build_kwargs( model="grok-4.6", messages=[{"role": "user", "content": "hi"}], tools=[real_tool], is_xai_responses=True, ) assert transport._last_wire_aliases == {} normalized = self._normalize_named_call( transport, monkeypatch, "hermes_tool_search" ) assert [tc.name for tc in normalized.tool_calls] == ["hermes_tool_search"] def test_alias_collision_takes_suffix_no_duplicates(self, transport, monkeypatch): """A real tool already named ``hermes_tool_search`` keeps its wire name; the bridge is suffixed and both round-trip independently.""" tools = [ {"type": "function", "function": { "name": "hermes_tool_search", "description": "Real tool.", "parameters": {"type": "object", "properties": {}}}}, {"type": "function", "function": { "name": "tool_search", "description": "Bridge.", "parameters": {"type": "object", "properties": {}}}}, ] kw = transport.build_kwargs( model="grok-4.6", messages=[{"role": "user", "content": "hi"}], tools=tools, is_xai_responses=True, ) names = self._names(kw) assert names == ["hermes_tool_search", "hermes_tool_search_2"] assert len(names) == len(set(names)) assert transport._last_wire_aliases == {"hermes_tool_search_2": "tool_search"} # Bridge alias maps back; the real tool's name is untouched. normalized = self._normalize_named_call( transport, monkeypatch, "hermes_tool_search_2" ) assert [tc.name for tc in normalized.tool_calls] == ["tool_search"] normalized2 = self._normalize_named_call( transport, monkeypatch, "hermes_tool_search" ) assert [tc.name for tc in normalized2.tool_calls] == ["hermes_tool_search"] def test_legacy_fallback_without_provenance(self, transport, monkeypatch): """Normalize-only call sites (no build_kwargs on this instance) keep the historical unconditional reverse mapping.""" assert transport._last_wire_aliases is None normalized = self._normalize_named_call( transport, monkeypatch, "hermes_tool_search" ) assert [tc.name for tc in normalized.tool_calls] == ["tool_search"] class TestXaiWebSearchBackendPreference: """``_xai_prefers_native_web_search`` must honor web backend config.""" def test_explicit_firecrawl_prefers_client(self, monkeypatch): import agent.transports.codex as codex_mod monkeypatch.setattr( "agent.web_search_registry.get_active_search_provider", lambda: SimpleNamespace(name="firecrawl"), ) assert codex_mod._xai_prefers_native_web_search() is False def test_explicit_search_backend_xai_prefers_native(self, monkeypatch): import agent.transports.codex as codex_mod monkeypatch.setattr( "agent.web_search_registry.get_active_search_provider", lambda: SimpleNamespace(name="xai"), ) assert codex_mod._xai_prefers_native_web_search() is True def test_no_provider_legacy_fallback_xai(self, monkeypatch): """When no provider is registered, fall back to _get_search_backend.""" import agent.transports.codex as codex_mod monkeypatch.setattr( "agent.web_search_registry.get_active_search_provider", lambda: None, ) monkeypatch.setattr( "tools.web_tools._get_search_backend", lambda: "xai", ) assert codex_mod._xai_prefers_native_web_search() is True def test_no_provider_legacy_fallback_non_xai(self, monkeypatch): """When no provider is registered and backend isn't xai, keep client.""" import agent.transports.codex as codex_mod monkeypatch.setattr( "agent.web_search_registry.get_active_search_provider", lambda: None, ) monkeypatch.setattr( "tools.web_tools._get_search_backend", lambda: "firecrawl", ) assert codex_mod._xai_prefers_native_web_search() is False class TestCodexValidateResponse: def test_none_response(self, transport): assert transport.validate_response(None) is False def test_valid_output(self, transport): r = SimpleNamespace(output=[{"type": "message", "content": []}]) assert transport.validate_response(r) is True @pytest.mark.parametrize("reason", ["max_output_tokens", "length", "", None]) def test_empty_output_other_incomplete_reasons_remain_invalid(self, transport, reason): r = SimpleNamespace( status="incomplete", incomplete_details=SimpleNamespace(reason=reason), output=[], output_text="", ) assert transport.validate_response(r) is False class TestCodexMapFinishReason: def test_completed(self, transport): assert transport.map_finish_reason("completed") == "stop" class TestCodexNormalizeResponse: def test_text_response(self, transport): """Normalize a simple text Codex response.""" r = SimpleNamespace( output=[ SimpleNamespace( type="message", role="assistant", content=[SimpleNamespace(type="output_text", text="Hello world")], status="completed", ), ], status="completed", incomplete_details=None, usage=SimpleNamespace(input_tokens=10, output_tokens=5, input_tokens_details=None, output_tokens_details=None), ) nr = transport.normalize_response(r) assert isinstance(nr, NormalizedResponse) assert nr.content == "Hello world" assert nr.finish_reason == "stop" def test_message_items_preserved_in_provider_data(self, transport): """Codex assistant message item ids/phases must survive transport normalization.""" r = SimpleNamespace( output=[ SimpleNamespace( type="message", role="assistant", id="msg_abc", phase="final_answer", content=[SimpleNamespace(type="output_text", text="Hello world")], status="completed", ), ], status="completed", incomplete_details=None, usage=SimpleNamespace(input_tokens=10, output_tokens=5, input_tokens_details=None, output_tokens_details=None), ) nr = transport.normalize_response(r) assert nr.codex_message_items == [ { "type": "message", "role": "assistant", "status": "completed", "content": [{"type": "output_text", "text": "Hello world"}], "id": "msg_abc", "phase": "final_answer", } ] def test_tool_call_response(self, transport): """Normalize a Codex response with tool calls.""" r = SimpleNamespace( output=[ SimpleNamespace( type="function_call", call_id="call_abc123", name="terminal", arguments=json.dumps({"command": "ls"}), id="fc_abc123", status="completed", ), ], status="completed", incomplete_details=None, usage=SimpleNamespace(input_tokens=10, output_tokens=20, input_tokens_details=None, output_tokens_details=None), ) nr = transport.normalize_response(r) assert nr.finish_reason == "tool_calls" assert len(nr.tool_calls) == 1 tc = nr.tool_calls[0] assert tc.name == "terminal" assert '"command"' in tc.arguments class TestCodexTransportTimeout: """Forward per-request timeout from build_kwargs to the SDK kwargs.""" def test_positive_timeout_preserved(self, transport): kw = transport.build_kwargs( model="gpt-5.5", messages=[{"role": "user", "content": "hi"}], tools=[], timeout=600.0, ) assert kw.get("timeout") == 600.0 def test_inf_timeout_dropped(self, transport): kw = transport.build_kwargs( model="gpt-5.5", messages=[{"role": "user", "content": "hi"}], tools=[], timeout=float("inf"), ) assert "timeout" not in kw class TestCodexTransportXaiReasoningEffort: @pytest.fixture def transport(self): from agent.transports.codex import ResponsesApiTransport return ResponsesApiTransport() def test_grok_46_preserves_xhigh(self, transport): kw = transport.build_kwargs( model="grok-4.6", messages=[{"role": "user", "content": "hi"}], tools=[], is_xai_responses=True, reasoning_config={"effort": "xhigh"}, ) assert kw["reasoning"]["effort"] == "xhigh" @pytest.mark.parametrize("effort", ["max", "ultra"]) def test_grok_46_clamps_hermes_aliases_to_model_ceiling(self, transport, effort): """Hermes ladder aliases mean "this model's ceiling" — on grok-4.6 that is xhigh, not one rung below it (#87279).""" kw = transport.build_kwargs( model="x-ai/grok-4.6-latest", messages=[{"role": "user", "content": "hi"}], tools=[], is_xai_responses=True, reasoning_config={"effort": effort}, ) assert kw["reasoning"]["effort"] == "xhigh" @pytest.mark.parametrize("effort", ["max", "ultra"]) def test_older_grok_clamps_aliases_to_high(self, transport, effort): """Older Grok tops out at high; above-ceiling aliases land there.""" kw = transport.build_kwargs( model="grok-4.5", messages=[{"role": "user", "content": "hi"}], tools=[], is_xai_responses=True, reasoning_config={"effort": effort}, ) assert kw["reasoning"]["effort"] == "high" def test_older_grok_clamps_xhigh_to_high(self, transport): kw = transport.build_kwargs( model="grok-4.5", messages=[{"role": "user", "content": "hi"}], tools=[], is_xai_responses=True, reasoning_config={"effort": "xhigh"}, ) assert kw["reasoning"]["effort"] == "high" class TestCodexTransportXaiServiceTierStrip: """xAI Responses API rejects ``service_tier`` (#28490). ``resolve_fast_mode_overrides`` only returns ``service_tier`` for OpenAI fast-eligible models, so on paper the field should never reach a Grok request. But ``self.service_tier`` lingers across model switches and can also be set directly via ``agent.service_tier`` in config.yaml — both leak paths plumb through ``request_overrides`` and would 400 against xAI's ``/v1/responses``. Strip defensively when targeting xAI. """ @pytest.fixture def transport(self): from agent.transports.codex import ResponsesApiTransport return ResponsesApiTransport() def test_xai_strips_service_tier_from_request_overrides(self, transport): """Headline #28490 case: service_tier=priority leaks through request_overrides, must not reach the xAI request body.""" kw = transport.build_kwargs( model="grok-4.3", messages=[{"role": "user", "content": "hi"}], tools=[], is_xai_responses=True, request_overrides={"service_tier": "priority"}, ) assert "service_tier" not in kw, ( f"service_tier must be stripped on xAI requests, " f"got {kw.get('service_tier')!r}" ) def test_grok_46_preserves_priority_service_tier(self, transport): kw = transport.build_kwargs( model="x-ai/grok-4.6-latest", messages=[{"role": "user", "content": "hi"}], tools=[], is_xai_responses=True, request_overrides={"service_tier": "priority"}, ) assert kw.get("service_tier") == "priority" def test_grok_46_strips_non_priority_service_tier(self, transport): kw = transport.build_kwargs( model="grok-4.6", messages=[{"role": "user", "content": "hi"}], tools=[], is_xai_responses=True, request_overrides={"service_tier": "unsupported"}, ) assert "service_tier" not in kw def test_non_xai_codex_preserves_service_tier(self, transport): """The strip is xAI-only — native Codex DOES accept service_tier=priority (OpenAI Priority Processing). Stripping it elsewhere would silently disable the user's fast-mode opt-in. """ kw = transport.build_kwargs( model="gpt-5.5", messages=[{"role": "user", "content": "hi"}], tools=[], is_xai_responses=False, is_codex_backend=True, request_overrides={"service_tier": "priority"}, ) assert kw.get("service_tier") == "priority", ( "non-xAI codex_responses providers must keep service_tier" ) def test_github_responses_preserves_service_tier(self, transport): """GitHub Models (Copilot) is another codex_responses surface that should not be affected by the xAI strip.""" kw = transport.build_kwargs( model="gpt-5.5", messages=[{"role": "user", "content": "hi"}], tools=[], is_github_responses=True, request_overrides={"service_tier": "priority"}, ) assert kw.get("service_tier") == "priority" class TestPreflightSlashEnumStrip: """xAI Responses safety-net: strip slash-containing enum values when the model name indicates a Grok target (#28490). Native Codex accepts ``/``-containing enums; xAI rejects them with HTTP 400 "Invalid arguments passed to the model". The main agent loop and the auxiliary client already sanitize at request-build time; this preflight catches any future code path that bypasses those — gated on model name so we don't unnecessarily strip on non-xAI providers. """ def _make_kwargs(self, model: str, enum_values: list[str]) -> dict: return { "model": model, "instructions": "test", "input": [{"role": "user", "content": "hi"}], "tools": [ { "type": "function", "name": "pick_model", "description": "pick a model", "parameters": { "type": "object", "properties": { "model_id": { "type": "string", "enum": enum_values, }, }, }, }, ], } def test_grok_model_strips_slash_enum_values(self): """When the model name is Grok-family, slash-containing enum values are stripped so xAI doesn't 400 on the tool schema.""" from agent.codex_responses_adapter import _preflight_codex_api_kwargs kwargs = self._make_kwargs( "grok-4.3", ["Qwen/Qwen3.5-0.8B", "openai/gpt-oss-20b", "plain-id"], ) result = _preflight_codex_api_kwargs(kwargs) # The enum keyword itself is stripped (per strip_slash_enum's # semantics — it removes the constraint entirely when any value # contains /). params = result["tools"][0]["parameters"] assert "enum" not in params["properties"]["model_id"], ( "slash-containing enum must be stripped on Grok" ) def test_aggregator_prefixed_grok_also_strips(self): """Aggregator-prefixed (x-ai/grok-*) names hit the same path.""" from agent.codex_responses_adapter import _preflight_codex_api_kwargs kwargs = self._make_kwargs( "x-ai/grok-4.3", ["Qwen/Qwen3.5-0.8B"], ) result = _preflight_codex_api_kwargs(kwargs) assert "enum" not in result["tools"][0]["parameters"]["properties"]["model_id"] def test_non_grok_model_preserves_slash_enum_values(self): """Native Codex / GitHub Models DO accept slash-containing enums. The safety-net must NOT strip there or we silently degrade tool-schema constraints on every codex_responses provider that isn't xAI.""" from agent.codex_responses_adapter import _preflight_codex_api_kwargs kwargs = self._make_kwargs( "gpt-5.5", ["Qwen/Qwen3.5-0.8B", "plain-id"], ) result = _preflight_codex_api_kwargs(kwargs) params = result["tools"][0]["parameters"] # The enum must survive on non-xAI providers. assert params["properties"]["model_id"].get("enum") == [ "Qwen/Qwen3.5-0.8B", "plain-id" ] def test_text_verbosity_reaches_responses_body_only_when_configured(transport): """``agent.text_verbosity`` maps to top-level ``text.verbosity`` on Responses routes (#20203). Unset/empty sends nothing (never flips the provider default), xAI never gets it (its /responses rejects unknown top-level fields), and the chat_completions transport has no such field at all. """ from agent.transports import chat_completions # noqa: F401 (registers the sibling) msgs = [{"role": "user", "content": "hi"}] assert transport.build_kwargs(model="gpt-5.1", messages=msgs, text_verbosity="low")["text"] == {"verbosity": "low"} for unset in (None, ""): assert "text" not in transport.build_kwargs(model="gpt-5.1", messages=msgs, text_verbosity=unset) assert "text" not in transport.build_kwargs( model="grok-4", messages=msgs, text_verbosity="low", is_xai_responses=True, base_url="https://api.x.ai/v1", ) chat = get_transport("chat_completions").build_kwargs(model="gpt-5.1", messages=msgs, text_verbosity="low") assert "text" not in chat and "text" not in (chat.get("extra_body") or {}) class TestOpenAIReasoningWireProjection: """Explicit ``reasoning_effort: none`` and non-reasoning OpenAI models on the Responses wire (#75227, #76255): a disable is sent as ``effort: none`` where the model accepts it — omitting the field leaves the model's default effort on — and chat-era models on api.openai.com, which 400 on any ``reasoning`` key, get no ``reasoning`` field at all.""" OPENAI = "https://api.openai.com/v1" def _reasoning(self, transport, model, reasoning_config, base_url=OPENAI): kw = transport.build_kwargs(model=model, messages=[{"role": "user", "content": "Hi"}], tools=[], base_url=base_url, reasoning_config=reasoning_config) return kw.get("reasoning") def test_explicit_none_is_sent_and_unset_keeps_the_default(self, transport): assert self._reasoning(transport, "gpt-5.6-sol", {"enabled": False}) == {"effort": "none"} assert self._reasoning(transport, "gpt-5.6-sol", None) == {"effort": "medium", "summary": "auto"} # Astra's vocabulary has no ``none``: nothing to send, never an escalated level. assert self._reasoning(transport, "gpt-6-astra", {"enabled": False}) is None def test_disable_the_route_cannot_express_is_reported_once(self, transport, caplog): """#75227: a disable the vocabulary cannot carry (Astra has no ``none``) is reported as an unsupported configuration — the model's default effort stays on — instead of silently omitted; once per model.""" import logging from agent.transports import codex as codex_transport codex_transport._UNPROJECTABLE_DISABLE_WARNED.discard("gpt-6-astra") with caplog.at_level(logging.WARNING, logger="agent.transports.codex"): for _ in range(2): assert self._reasoning(transport, "gpt-6-astra", {"enabled": False}) is None warned = [r.getMessage() for r in caplog.records if r.name == "agent.transports.codex" and r.levelno >= logging.WARNING] assert len(warned) == 1 and "gpt-6-astra" in warned[0], caplog.text @pytest.mark.parametrize("model", ["gpt-4o-mini", "gpt-4.1-mini", "openai/gpt-4o", "ft:gpt-4o-mini:acme::abc1"]) def test_chat_era_openai_models_get_no_reasoning_field_on_the_official_origin(self, transport, model): for rc in (None, {"enabled": True, "effort": "high"}, {"enabled": False}): assert self._reasoning(transport, model, rc) is None, (model, rc) # Reasoning models on the same origin and the same id on a relay keep the dial (the relay may translate). assert self._reasoning(transport, "o4-mini", None) == {"effort": "medium", "summary": "auto"} assert self._reasoning(transport, model, {"enabled": True, "effort": "high"}, base_url="https://relay.example.com/v1") == {"effort": "high", "summary": "auto"}