from types import SimpleNamespace import pytest from agent.codex_responses_adapter import ( _chat_messages_to_responses_input, _format_responses_error, _normalize_codex_response, _preflight_codex_api_kwargs, _preflight_codex_input_items, ) def test_normalize_codex_response_drops_transient_rs_tmp_reasoning_items(): response = SimpleNamespace( status="completed", output=[ SimpleNamespace( type="reasoning", id="rs_tmp_123", encrypted_content="opaque-transient", summary=[], ), SimpleNamespace( type="reasoning", id="rs_456", encrypted_content="opaque-stable", summary=[SimpleNamespace(text="stable summary")], ), SimpleNamespace( type="message", role="assistant", status="completed", content=[SimpleNamespace(type="output_text", text="done")], ), ], ) assistant_message, finish_reason = _normalize_codex_response(response) assert finish_reason == "stop" assert assistant_message.content == "done" assert assistant_message.codex_reasoning_items == [ { "type": "reasoning", "encrypted_content": "opaque-stable", "id": "rs_456", "summary": [{"type": "summary_text", "text": "stable summary"}], } ] def test_normalize_codex_response_treats_summary_only_reasoning_as_incomplete(): """Summary-only reasoning keeps the continuation path for Codex backends. Since #64434, an unrecognized issuer with ``response.status="completed"`` trusts the provider and returns ``stop`` — so this test pins the Codex backend explicitly, where reasoning-only still means "still thinking". """ response = SimpleNamespace( status="completed", output=[ SimpleNamespace( type="reasoning", id="rs_tmp_789", encrypted_content="opaque-transient", summary=[SimpleNamespace(text="still thinking")], ) ], ) assistant_message, finish_reason = _normalize_codex_response( response, issuer_kind="codex_backend" ) assert finish_reason == "incomplete" assert assistant_message.content == "" assert assistant_message.reasoning == "still thinking" assert assistant_message.codex_reasoning_items is None def test_normalize_codex_response_maps_incomplete_content_filter_to_refusal(): response = SimpleNamespace( status="incomplete", incomplete_details=SimpleNamespace(reason="content_filter"), output=[], output_text="", ) assistant_message, finish_reason = _normalize_codex_response(response) assert finish_reason == "content_filter" assert assistant_message.content == "" assert response.output # --------------------------------------------------------------------------- # Server-side built-in tool calls (xAI native web_search, code interpreter, # etc.) come back as discrete ``*_call`` output items that xAI's # /v1/responses surface routinely leaves at ``status="in_progress"`` even # when the overall ``response.status == "completed"``. These must NOT mark # the turn incomplete — otherwise grok-composer-2.5-fast research queries # (which invoke server-side web_search) get misclassified as # ``finish_reason="incomplete"`` and burn 3 fruitless continuation retries # before failing with "Codex response remained incomplete after 3 # continuation attempts". Observed live against grok-composer-2.5-fast on # SuperGrok OAuth (2026-06). # --------------------------------------------------------------------------- def test_normalize_codex_response_ignores_in_progress_server_side_tool_calls(): """A completed response with a final message + lingering in_progress server-side web_search_call items resolves to 'stop', not 'incomplete'.""" response = SimpleNamespace( status="completed", incomplete_details=None, output=[ SimpleNamespace( type="reasoning", id="rs_1", encrypted_content="opaque", summary=[SimpleNamespace(text="researching blades")], ), SimpleNamespace( type="message", role="assistant", status="completed", content=[SimpleNamespace( type="output_text", text="Milwaukee M18 blade 49-16-2734, ~$30 OEM.", )], ), SimpleNamespace(type="web_search_call", status="in_progress"), SimpleNamespace(type="web_search_call", status="in_progress"), SimpleNamespace(type="web_search_call", status="in_progress"), ], ) assistant_message, finish_reason = _normalize_codex_response(response) assert finish_reason == "stop" assert assistant_message.content == "Milwaukee M18 blade 49-16-2734, ~$30 OEM." def test_normalize_codex_response_in_progress_message_still_incomplete(): """Guard scope: an in_progress *message* item (genuine model output that is still streaming) must still mark the turn incomplete — only server-side ``*_call`` items are exempted.""" response = SimpleNamespace( status="completed", incomplete_details=None, output=[ SimpleNamespace( type="message", role="assistant", status="in_progress", content=[SimpleNamespace(type="output_text", text="partial...")], ), ], ) _assistant_message, finish_reason = _normalize_codex_response(response) assert finish_reason == "incomplete" # --------------------------------------------------------------------------- # Replayed assistant message items with an oversized server-assigned ``id`` # (Codex issues 400+ char base64 blobs) must never reach the API — the # Responses endpoint caps input[].id at 64 chars and rejects the whole # request with a non-retryable HTTP 400, permanently bricking the session # (every subsequent turn replays the same bad id). Short ids (msg_...) are # still worth keeping for prefix-cache hits, so this is a length guard, not # a blanket strip. # --------------------------------------------------------------------------- _OVERSIZED_ITEM_ID = "x" * 408 _VALID_ITEM_ID = "msg_abc123" def test_chat_messages_to_responses_input_drops_oversized_message_id(): messages = [ { "role": "assistant", "content": "pong", "codex_message_items": [ { "type": "message", "role": "assistant", "status": "completed", "content": [{"type": "output_text", "text": "pong"}], "id": _OVERSIZED_ITEM_ID, "phase": "final_answer", } ], } ] items = _chat_messages_to_responses_input(messages) message_item = next(item for item in items if item.get("type") == "message") assert "id" not in message_item assert message_item["phase"] == "final_answer" assert message_item["content"] == [{"type": "output_text", "text": "pong"}] def test_chat_messages_to_responses_input_keeps_short_message_id(): messages = [ { "role": "assistant", "content": "pong", "codex_message_items": [ { "type": "message", "role": "assistant", "status": "completed", "content": [{"type": "output_text", "text": "pong"}], "id": _VALID_ITEM_ID, } ], } ] items = _chat_messages_to_responses_input(messages) message_item = next(item for item in items if item.get("type") == "message") assert message_item["id"] == _VALID_ITEM_ID def test_preflight_codex_input_items_drops_oversized_message_id(): items = _preflight_codex_input_items( [ { "type": "message", "role": "assistant", "status": "completed", "content": [{"type": "output_text", "text": "pong"}], "id": _OVERSIZED_ITEM_ID, "phase": "final_answer", } ] ) assert "id" not in items[0] assert items[0]["phase"] == "final_answer" def test_preflight_codex_input_items_keeps_short_message_id(): items = _preflight_codex_input_items( [ { "type": "message", "role": "assistant", "status": "completed", "content": [{"type": "output_text", "text": "pong"}], "id": _VALID_ITEM_ID, } ] ) assert items[0]["id"] == _VALID_ITEM_ID def test_preflight_codex_input_items_drops_short_id_for_github_responses(): items = _preflight_codex_input_items( [ { "type": "message", "role": "assistant", "status": "in_progress", "content": [{"type": "output_text", "text": "pong"}], "id": _VALID_ITEM_ID, "phase": "final_answer", } ], is_github_responses=True, ) assert "id" not in items[0] assert items[0]["status"] == "in_progress" assert items[0]["phase"] == "final_answer" assert items[0]["content"] == [{"type": "output_text", "text": "pong"}] def test_preflight_codex_api_kwargs_drops_oversized_message_id_end_to_end(): kwargs = _preflight_codex_api_kwargs( { "model": "gpt-5.5", "instructions": "You are Hermes.", "input": [ {"role": "user", "content": "ping"}, { "type": "message", "role": "assistant", "status": "completed", "content": [{"type": "output_text", "text": "pong"}], "id": _OVERSIZED_ITEM_ID, "phase": "final_answer", }, ], "tools": [], "store": False, } ) message_item = next(item for item in kwargs["input"] if item.get("type") == "message") assert "id" not in message_item # --------------------------------------------------------------------------- # _preflight_codex_api_kwargs — built-in (provider-executed) tools must pass # through validation. Regression guard for the xAI native web_search # injection: the preflight validator previously rejected any tool whose # ``type != "function"`` with "unsupported type", which would 400 every xAI # turn once the native web_search tool is declared. # --------------------------------------------------------------------------- def test_preflight_passes_native_web_search_tool_through(): kwargs = { "model": "grok-composer-2.5-fast", "instructions": "You are helpful.", "input": [{"role": "user", "content": [{"type": "input_text", "text": "hi"}]}], "store": False, "tools": [ {"type": "function", "name": "read_file", "description": "Read.", "parameters": {"type": "object", "properties": {}}}, {"type": "web_search"}, ], } out = _preflight_codex_api_kwargs(kwargs, allow_stream=True) tools = out["tools"] assert {"type": "web_search"} in tools assert any(t.get("type") == "function" and t.get("name") == "read_file" for t in tools) def test_preflight_still_rejects_unknown_tool_type(): kwargs = { "model": "grok-composer-2.5-fast", "instructions": "You are helpful.", "input": [{"role": "user", "content": [{"type": "input_text", "text": "hi"}]}], "store": False, "tools": [{"type": "totally_made_up_tool"}], } with pytest.raises(ValueError, match="unsupported type"): _preflight_codex_api_kwargs(kwargs, allow_stream=True) # --------------------------------------------------------------------------- # _format_responses_error — adapted from anomalyco/opencode#28757. # Provider failures should surface BOTH the code (rate_limit_exceeded / # context_length_exceeded / internal_error / server_error) and the message, # so consumers can tell rate limits apart from context-length failures and # both apart from generic stream drops. # --------------------------------------------------------------------------- def test_format_responses_error_combines_code_and_message(): err = {"code": "rate_limit_exceeded", "message": "Slow down"} assert _format_responses_error(err, "failed") == "rate_limit_exceeded: Slow down" def test_format_responses_error_message_only(): err = {"message": "Upstream model unavailable"} assert _format_responses_error(err, "failed") == "Upstream model unavailable" def test_format_responses_error_code_only_when_message_empty(): # Some providers/proxies emit a code with an empty message body. We # used to fall back to ``str(error_obj)`` — a dict dump — which leaked # ``{'code': 'internal_error', 'message': ''}`` into chat output. Now # the bare code is surfaced, which is the meaningful field. err = {"code": "internal_error", "message": ""} assert _format_responses_error(err, "failed") == "internal_error" def test_format_responses_error_code_only_when_message_missing(): err = {"code": "server_error"} assert _format_responses_error(err, "failed") == "server_error" def test_format_responses_error_attribute_style_payload(): # SDK objects expose ``code``/``message`` as attributes rather than dict # keys. The helper must accept both shapes since the Responses SDK # returns SimpleNamespace-style objects on ``response.failed``. err = SimpleNamespace(code="context_length_exceeded", message="too long") assert _format_responses_error(err, "failed") == "context_length_exceeded: too long" def test_format_responses_error_falls_back_to_status_when_empty(): assert ( _format_responses_error(None, "failed") == "Responses API returned status 'failed'" ) assert ( _format_responses_error(None, "cancelled") == "Responses API returned status 'cancelled'" ) def test_format_responses_error_stringifies_opaque_payload(): # Last-resort: a provider sent something that isn't a dict and has no # code/message attributes. Surface its repr rather than swallow it # silently — at least it's visible in logs. assert _format_responses_error("opaque sentinel", "failed") == "opaque sentinel" def test_format_responses_error_ignores_non_string_code_message(): # Defensive: a malformed gateway could send numbers/objects in these # fields. We don't want to crash; we want a best-effort string. err = {"code": 500, "message": None} assert _format_responses_error(err, "failed") == "500" def test_normalize_codex_response_failed_includes_code_in_error(): """Regression: response_status == 'failed' should surface the error code, not just the message. Used to leak a bare 'Slow down' string that was indistinguishable from a generic stream truncation.""" # ``output`` non-empty so we don't trip the "no output items" guard # before reaching the failed-status branch. Real failed responses # often DO carry a partial message item alongside the error. response = SimpleNamespace( status="failed", output=[ SimpleNamespace( type="message", role="assistant", status="incomplete", content=[SimpleNamespace(type="output_text", text="partial")], ), ], error={"code": "rate_limit_exceeded", "message": "Slow down"}, ) with pytest.raises(RuntimeError, match=r"^rate_limit_exceeded: Slow down$"): _normalize_codex_response(response) def test_normalize_codex_response_failed_with_message_only(): """Backwards-compat: a failed response with only a message field (no code) should still surface that message verbatim.""" response = SimpleNamespace( status="failed", output=[ SimpleNamespace( type="message", role="assistant", status="incomplete", content=[SimpleNamespace(type="output_text", text="partial")], ), ], error={"message": "model error"}, ) with pytest.raises(RuntimeError, match=r"^model error$"): _normalize_codex_response(response) # --------------------------------------------------------------------------- # Reasoning-channel answer salvage (xAI grok) — grok-4.x on the xAI # /v1/responses surface sometimes emits its final answer inside the # reasoning item, delimited by grok's internal "" tag, with no # ``message`` output item at all. Because those reasoning items carry no # encrypted_content, the interim message replays as nothing and every # continuation request is byte-identical — the turn burns 3 retries and # fails even though the answer was produced. Observed live with grok-4.20 # on xai-oauth (2026-07-13). # --------------------------------------------------------------------------- def _xai_reasoning_only_response(reasoning_text): return SimpleNamespace( status="completed", output=[ SimpleNamespace( type="reasoning", id="rs_1", encrypted_content=None, summary=[SimpleNamespace(text=reasoning_text)], ) ], ) def test_normalize_codex_response_salvages_xai_reasoning_channel_answer(): response = _xai_reasoning_only_response( "The process is still running.\n\nAll good, process running." ) assistant_message, finish_reason = _normalize_codex_response( response, issuer_kind="xai_responses" ) assert finish_reason == "stop" assert assistant_message.content == "All good, process running." assert assistant_message.reasoning == "The process is still running." def test_normalize_codex_response_salvage_strips_closing_tag(): response = _xai_reasoning_only_response( "Thinking.\nThe answer." ) assistant_message, finish_reason = _normalize_codex_response( response, issuer_kind="xai_responses" ) assert finish_reason == "stop" assert assistant_message.content == "The answer." def test_normalize_codex_response_salvage_is_xai_scoped(): """Non-xAI special-cased issuers (Codex backend) keep the reasoning-only → incomplete classification; the Codex backend replays encrypted reasoning, so its continuation genuinely progresses and must not be short-circuited. Pins ``issuer_kind="codex_backend"`` explicitly: with no issuer at all, the unrecognized-backend rule (#64434) trusts ``status="completed"`` and returns ``stop`` — that path is covered by the #64434 regression tests. """ response = _xai_reasoning_only_response( "Thinking.\nThe answer." ) assistant_message, finish_reason = _normalize_codex_response( response, issuer_kind="codex_backend" ) assert finish_reason == "incomplete" assert assistant_message.content == "" def test_normalize_codex_response_xai_reasoning_without_marker_stays_incomplete(): response = _xai_reasoning_only_response("Still thinking, no answer yet.") assistant_message, finish_reason = _normalize_codex_response( response, issuer_kind="xai_responses" ) assert finish_reason == "incomplete" assert assistant_message.content == ""