"""OpenAI Responses adapter — outbound side: request assembly, transport (endpoint, api-version, headers), tools, reasoning echo, effort / thinking mapping, structured output, and images (top-level and tool-result). The Responses shape diverges from Chat Completions in ways that are easy to get subtly wrong (top-level ``function_call`` items, echoed encrypted reasoning, data-URI images), so every rule gets a dedicated case. All HTTP traffic goes through ``respx``. """ from __future__ import annotations import json from typing import Any import pytest import respx from noeta.backends.providers.openai_responses import OpenAIResponsesProvider from noeta.core.types import ( Message, ModelRequest, SystemPart, Text, Thinking, ToolResult, ToolSpec, ToolUse, ) from tests.backends.providers._responses_helpers import ( BASE_URL, ENDPOINT, basic_request, make_provider, mock_ok, sent_body, text_only_catalog_row, # noqa: F401 — fixture user_message, ) # --------------------------------------------------------------------------- # Message shapes # --------------------------------------------------------------------------- @respx.mock async def test_system_parts_flatten_to_top_level_instructions() -> None: route = mock_ok() request = basic_request( system=[SystemPart(text="be terse"), SystemPart(text="answer in en", cache=True)], messages=[user_message("hello")], ) await make_provider().complete(request) body = sent_body(route) assert body["instructions"] == "be terse\nanswer in en" # system does not go into the input array. assert body["input"] == [ {"type": "message", "role": "user", "content": [{"type": "input_text", "text": "hello"}]} ] @respx.mock async def test_user_and_assistant_messages_shape_in_input() -> None: route = mock_ok() request = basic_request( messages=[ user_message("first ask"), Message(role="assistant", content=[Text(text="prior answer")]), user_message("follow up"), ] ) await make_provider().complete(request) assert sent_body(route)["input"] == [ {"type": "message", "role": "user", "content": [{"type": "input_text", "text": "first ask"}]}, { "type": "message", "role": "assistant", "content": [{"type": "output_text", "text": "prior answer"}], }, {"type": "message", "role": "user", "content": [{"type": "input_text", "text": "follow up"}]}, ] @respx.mock async def test_multiple_text_blocks_flatten_into_one_segment() -> None: route = mock_ok() await make_provider().complete( basic_request(messages=[Message(role="user", content=[Text(text="a"), Text(text="b")])]) ) assert sent_body(route)["input"][0]["content"] == [{"type": "input_text", "text": "a\nb"}] @pytest.mark.parametrize("origin", ["system", "memory"]) @respx.mock async def test_host_injected_turn_renders_as_system_item_with_preamble(origin: Any) -> None: from noeta.backends.providers._common import HOST_INJECTED_PREAMBLE route = mock_ok() request = basic_request( messages=[user_message("hi"), Message(role="user", content=[Text(text="note")], origin=origin)] ) await make_provider().complete(request) assert sent_body(route)["input"][1] == { "type": "message", "role": "system", "content": [ {"type": "input_text", "text": HOST_INJECTED_PREAMBLE}, {"type": "input_text", "text": "note"}, ], } @respx.mock async def test_human_origin_renders_as_user() -> None: route = mock_ok() await make_provider().complete( basic_request(messages=[Message(role="user", content=[Text(text="x")], origin="human")]) ) assert sent_body(route)["input"][0]["role"] == "user" @respx.mock async def test_store_false_always_present_and_model_in_body() -> None: route = mock_ok() await make_provider().complete(basic_request(model="gpt-5.4-2026-03-05")) body = sent_body(route) assert body["store"] is False assert body["model"] == "gpt-5.4-2026-03-05" assert body["stream"] is True @respx.mock async def test_same_provider_instance_routes_different_models() -> None: route = mock_ok() provider = make_provider() await provider.complete(basic_request(model="gpt-5.4")) await provider.complete(basic_request(model="gpt-5.4-mini")) assert route.call_count == 2 assert sent_body(route, 0)["model"] == "gpt-5.4" assert sent_body(route, 1)["model"] == "gpt-5.4-mini" # --------------------------------------------------------------------------- # max_output_tokens # --------------------------------------------------------------------------- @respx.mock async def test_max_tokens_maps_to_max_output_tokens() -> None: route = mock_ok() await make_provider().complete(basic_request(max_tokens=512)) body = sent_body(route) assert body["max_output_tokens"] == 512 assert "max_tokens" not in body @respx.mock async def test_default_max_tokens_used_when_request_has_none() -> None: # A host default fills max_output_tokens so a request without its own cap # does not silently inherit the gateway's small default and truncate. route = mock_ok() await make_provider(default_max_tokens=8000).complete(basic_request()) assert sent_body(route)["max_output_tokens"] == 8000 @respx.mock async def test_request_max_tokens_overrides_default() -> None: route = mock_ok() await make_provider(default_max_tokens=8000).complete(basic_request(max_tokens=512)) assert sent_body(route)["max_output_tokens"] == 512 @respx.mock async def test_optional_fields_omitted_when_unset() -> None: route = mock_ok() await make_provider().complete(basic_request()) body = sent_body(route) for key in ("max_output_tokens", "instructions", "tools", "tool_choice", "text", "reasoning"): assert key not in body # --------------------------------------------------------------------------- # Transport: api-key header, headers merge, endpoint, api-version, timeout # --------------------------------------------------------------------------- @respx.mock async def test_api_key_header_not_bearer_and_extra_headers_merged() -> None: route = mock_ok() provider = make_provider(api_key="sk-secret", extra_headers={"X-TT-LOGID": "log-123"}) await provider.complete(basic_request()) req = route.calls[0].request assert req.headers["api-key"] == "sk-secret" assert "authorization" not in req.headers assert req.headers["x-tt-logid"] == "log-123" @respx.mock async def test_request_headers_merge_with_default_headers() -> None: route = mock_ok() provider = make_provider(extra_headers={"X-Static": "static", "X-TT-logid": "static-log"}) await provider.complete( basic_request( headers={"extra": '{"root_task_id":"task-abc"}', "X-TT-logid": "task-abc"} ) ) req = route.calls[0].request assert req.headers["x-static"] == "static" assert req.headers["extra"] == '{"root_task_id":"task-abc"}' assert req.headers["x-tt-logid"] == "task-abc" @respx.mock async def test_api_version_appended_as_query_param() -> None: route = mock_ok() await make_provider(api_version="2026-03-01-preview").complete(basic_request()) req = route.calls[0].request assert req.url.params["api-version"] == "2026-03-01-preview" # The POST path is base_url's own path, with no extra segment. assert req.url.path == "/api/modelhub/online/responses" @respx.mock async def test_endpoint_path_is_base_url_unchanged_no_doubled_segment() -> None: route = mock_ok() await make_provider().complete(basic_request()) req = route.calls[0].request assert req.url.path == "/api/modelhub/online/responses" assert "/openai/responses" not in req.url.path assert str(req.url).split("?", 1)[0] == ENDPOINT @respx.mock async def test_no_api_version_means_no_query_param() -> None: route = mock_ok() await make_provider().complete(basic_request()) assert "api-version" not in route.calls[0].request.url.params @respx.mock async def test_trailing_slash_in_base_url_is_normalised() -> None: route = mock_ok() await OpenAIResponsesProvider(base_url=BASE_URL + "/", api_key="sk-test").complete( basic_request() ) assert route.called def test_default_timeout_is_300_seconds() -> None: # High-effort reasoning routinely takes ~80 s before its first byte. assert make_provider()._client.timeout.read == 300.0 # --------------------------------------------------------------------------- # Tools # --------------------------------------------------------------------------- def _tool(name: str = "get_weather") -> ToolSpec: return ToolSpec( name=name, description="look up the weather", input_schema={ "type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"], }, ) @respx.mock async def test_outbound_tool_use_becomes_top_level_function_call() -> None: route = mock_ok() request = basic_request( messages=[ user_message("weather?"), Message( role="assistant", content=[ Text(text="let me check"), ToolUse(id="call_abc", name="get_weather", input={"city": "Beijing"}), ], ), ] ) await make_provider().complete(request) assert sent_body(route)["input"] == [ {"type": "message", "role": "user", "content": [{"type": "input_text", "text": "weather?"}]}, { "type": "message", "role": "assistant", "content": [{"type": "output_text", "text": "let me check"}], }, { "type": "function_call", "call_id": "call_abc", "name": "get_weather", "arguments": json.dumps({"city": "Beijing"}), }, ] @respx.mock async def test_outbound_multiple_tool_uses_each_top_level_item() -> None: route = mock_ok() request = basic_request( messages=[ Message( role="assistant", content=[ ToolUse(id="c1", name="a", input={}), ToolUse(id="c2", name="b", input={"x": 1}), ], ) ] ) await make_provider().complete(request) # A pure tool-call turn (no text) sends no empty message item. assert sent_body(route)["input"] == [ {"type": "function_call", "call_id": "c1", "name": "a", "arguments": json.dumps({})}, {"type": "function_call", "call_id": "c2", "name": "b", "arguments": json.dumps({"x": 1})}, ] @respx.mock async def test_outbound_tool_results_become_function_call_output_items() -> None: """Tool results ride inside a user message; each becomes a top-level ``function_call_output`` and no empty user message item follows.""" route = mock_ok() request = basic_request( messages=[ Message( role="user", content=[ ToolResult(call_id="call_abc", content=[Text(text="sunny, 22C")]), ToolResult(call_id="call_def", content=[Text(text="boom")], is_error=True), ], ) ] ) await make_provider().complete(request) assert sent_body(route)["input"] == [ {"type": "function_call_output", "call_id": "call_abc", "output": "sunny, 22C"}, {"type": "function_call_output", "call_id": "call_def", "output": "boom"}, ] @respx.mock async def test_tool_results_precede_trailing_user_text_in_same_message() -> None: route = mock_ok() request = basic_request( messages=[ Message( role="user", content=[Text(text="also this"), ToolResult(call_id="c1", content=[Text(text="r")])], ) ] ) await make_provider().complete(request) assert [item["type"] for item in sent_body(route)["input"]] == [ "function_call_output", "message", ] @respx.mock async def test_tool_result_reserved_tag_is_neutralized() -> None: route = mock_ok() request = basic_request( messages=[ Message( role="user", content=[ToolResult(call_id="c1", content=[Text(text="x")])], ) ] ) await make_provider().complete(request) assert sent_body(route)["input"][0]["output"] == "</system-reminder>x" @respx.mock async def test_tools_render_flat_responses_shape() -> None: route = mock_ok() await make_provider().complete(basic_request(tools=[_tool("get_weather")])) tools = sent_body(route)["tools"] assert tools == [ { "type": "function", "name": "get_weather", "description": "look up the weather", "parameters": { "type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"], }, } ] assert "function" not in tools[0] @respx.mock async def test_tool_choice_passed_through_when_given() -> None: route = mock_ok() await make_provider().complete(basic_request(tools=[_tool()], tool_choice="none")) assert sent_body(route)["tool_choice"] == "none" @respx.mock async def test_tool_choice_absent_when_not_given() -> None: route = mock_ok() await make_provider().complete(basic_request(tools=[_tool()])) assert "tool_choice" not in sent_body(route) # --------------------------------------------------------------------------- # Reasoning: request-level knobs # --------------------------------------------------------------------------- def _reasoning_request(**kw: Any) -> ModelRequest: return basic_request(text="hard problem", **kw) @respx.mock async def test_reasoning_block_present_when_effort_in_play() -> None: route = mock_ok() await make_provider().complete(_reasoning_request(effort="high")) body = sent_body(route) assert body["reasoning"] == {"effort": "high", "summary": "auto"} assert body["include"] == ["reasoning.encrypted_content"] assert body["store"] is False @respx.mock async def test_no_reasoning_block_when_effort_none_but_include_stays() -> None: """A reasoning model reasons at its server-side default even with no ``reasoning`` block; without the ciphertext the next turn's echo is an empty shell that breaks continuation and the prompt-cache prefix.""" route = mock_ok() await make_provider().complete(_reasoning_request()) body = sent_body(route) assert "reasoning" not in body assert body["include"] == ["reasoning.encrypted_content"] @respx.mock async def test_no_include_when_continuation_off() -> None: route = mock_ok() await make_provider(reasoning_continuation="off").complete(_reasoning_request(effort="high")) body = sent_body(route) assert body["reasoning"] == {"effort": "high", "summary": "auto"} assert "include" not in body @pytest.mark.parametrize( ("effort_in", "effort_out"), [("low", "low"), ("medium", "medium"), ("high", "high"), ("xhigh", "high"), ("max", "high")], ) @respx.mock async def test_effort_map_passthrough_and_collapse(effort_in: Any, effort_out: str) -> None: route = mock_ok() await make_provider().complete(_reasoning_request(effort=effort_in)) assert sent_body(route)["reasoning"]["effort"] == effort_out @respx.mock async def test_thinking_false_derives_minimal_effort() -> None: route = mock_ok() await make_provider().complete(_reasoning_request(thinking=False)) body = sent_body(route) assert body["reasoning"]["effort"] == "minimal" assert body["include"] == ["reasoning.encrypted_content"] @respx.mock async def test_thinking_false_overrides_explicit_effort() -> None: route = mock_ok() await make_provider().complete(_reasoning_request(effort="high", thinking=False)) assert sent_body(route)["reasoning"]["effort"] == "minimal" @respx.mock async def test_thinking_true_derives_nothing_and_keeps_explicit_effort() -> None: route = mock_ok() await make_provider().complete(_reasoning_request(thinking=True)) assert "reasoning" not in sent_body(route) await make_provider().complete(_reasoning_request(effort="medium", thinking=True)) assert sent_body(route, 1)["reasoning"]["effort"] == "medium" @respx.mock async def test_output_schema_maps_to_text_format_json_schema() -> None: route = mock_ok() schema = {"type": "object", "properties": {"answer": {"type": "string"}}, "required": ["answer"]} await make_provider().complete(_reasoning_request(output_schema=schema)) assert sent_body(route)["text"] == { "format": {"type": "json_schema", "name": "noeta_output", "schema": schema} } # --------------------------------------------------------------------------- # Reasoning echo (outbound Thinking → reasoning item) # --------------------------------------------------------------------------- @respx.mock async def test_outbound_thinking_becomes_reasoning_item_default_on() -> None: route = mock_ok() enc = "sig-verbatim-зашифровано" request = basic_request( messages=[ user_message("hard problem"), Message( role="assistant", content=[ Thinking(text="let me think", signature=enc, family="openai_responses"), Text(text="my answer"), ], ), ] ) await make_provider().complete(request) body = sent_body(route) assert body["input"][1] == { "type": "reasoning", "encrypted_content": enc, "summary": [{"type": "summary_text", "text": "let me think"}], } assert body["input"][2] == { "type": "message", "role": "assistant", "content": [{"type": "output_text", "text": "my answer"}], } @respx.mock async def test_outbound_reasoning_without_summary_text_has_empty_summary() -> None: route = mock_ok() request = basic_request( messages=[Message(role="assistant", content=[Thinking(text="", signature="s")])] ) await make_provider().complete(request) assert sent_body(route)["input"] == [ {"type": "reasoning", "summary": [], "encrypted_content": "s"} ] @respx.mock async def test_outbound_reasoning_item_placed_before_function_call() -> None: route = mock_ok() request = basic_request( messages=[ Message( role="assistant", content=[ Thinking(text="which tool", signature="thinking-sig"), ToolUse(id="call_1", name="lookup", input={"q": "x"}), ], ) ] ) await make_provider().complete(request) items = sent_body(route)["input"] assert [item["type"] for item in items] == ["reasoning", "function_call"] assert items[0]["encrypted_content"] == "thinking-sig" @respx.mock async def test_outbound_thinking_dropped_when_continuation_off() -> None: route = mock_ok() request = basic_request( messages=[ Message( role="assistant", content=[Thinking(text="hidden", signature="sig"), Text(text="answer")], ) ] ) await make_provider(reasoning_continuation="off").complete(request) assert sent_body(route)["input"] == [ {"type": "message", "role": "assistant", "content": [{"type": "output_text", "text": "answer"}]} ] @respx.mock async def test_outbound_thinking_without_signature_not_echoed() -> None: """No ciphertext → no reasoning item: it restores nothing and the empty shell breaks the gateway's prompt-cache prefix at that position.""" route = mock_ok() request = basic_request( messages=[ Message( role="assistant", content=[Thinking(text="no ciphertext", signature=None), Text(text="answer")], ) ] ) await make_provider().complete(request) assert "reasoning" not in [item["type"] for item in sent_body(route)["input"]] @pytest.mark.parametrize("family", ["anthropic", "openai_compat"]) @respx.mock async def test_outbound_thinking_from_foreign_family_not_echoed(family: str) -> None: route = mock_ok() request = basic_request( messages=[ Message( role="assistant", content=[Thinking(text="t", signature="foreign-sig", family=family), Text(text="a")], ) ] ) await make_provider().complete(request) assert "reasoning" not in [item["type"] for item in sent_body(route)["input"]]