"""Tests for the ChatCompletionsTransport.""" import json from types import SimpleNamespace import httpx import pytest from openai import OpenAI from agent.transports import get_transport from agent.transports.types import NormalizedResponse @pytest.fixture def transport(): import agent.transports.chat_completions # noqa: F401 return get_transport("chat_completions") class TestChatCompletionsBasic: @pytest.mark.parametrize("provider", ["nous", "openrouter"]) def test_gpt56_ultra_uses_max_wire_effort(self, transport, provider): from providers import get_provider_profile profile = get_provider_profile(provider) kw = transport.build_kwargs( model="openai/gpt-5.6-sol", messages=[{"role": "user", "content": "Hi"}], tools=[], reasoning_config={"enabled": True, "effort": "ultra"}, supports_reasoning=True, provider_profile=profile, provider_name=provider, base_url=profile.base_url, ) assert kw["extra_body"]["reasoning"] == {"enabled": True, "effort": "max"} def test_convert_messages_no_codex_leaks(self, transport): msgs = [{"role": "user", "content": "hi"}] result = transport.convert_messages(msgs) assert result is msgs # no copy needed def _msg_with_extra_content(self): return [ {"role": "assistant", "content": "ok", "tool_calls": [{"id": "call_1", "type": "function", "extra_content": {"google": {"thought_signature": "SIG_123"}}, "function": {"name": "t", "arguments": "{}"}}]}, ] def test_convert_messages_strips_timestamp(self, transport): """Internal per-message ``timestamp`` metadata (stamped by ``_apply_persist_user_message_override`` to preserve platform event time without embedding it in content, and persisted to the SQLite store) is not part of the OpenAI Chat Completions schema. Strict providers like Mistral / Fireworks-backed endpoints reject it with HTTP 422 'Extra inputs are not permitted, field: messages[N].timestamp'. Regression test for #47868. """ msgs = [ {"role": "user", "content": "hi", "timestamp": 1781976577.0}, ] result = transport.convert_messages(msgs) assert "timestamp" not in result[0] assert result[0]["content"] == "hi" assert result[0]["role"] == "user" # Original list untouched (deepcopy-on-demand) assert msgs[0]["timestamp"] == 1781976577.0 def test_convert_messages_no_copy_without_timestamp(self, transport): """A timestamp-free message list needs no sanitize pass and is returned by identity (preserves the deepcopy-on-demand contract).""" msgs = [{"role": "user", "content": "hi"}] assert transport.convert_messages(msgs) is msgs def test_convert_messages_strips_internal_scaffolding_markers(self, transport): """Hermes-internal ``_``-prefixed markers must never reach the wire. The empty-response recovery path appends synthetic messages tagged with ``_empty_recovery_synthetic``; permissive providers ignore the unknown key, but strict gateways (opencode-go, codex.nekos.me) reject the request, poisoning every later turn in the session. """ msgs = [ {"role": "user", "content": "run the task"}, {"role": "assistant", "content": "(empty)", "_empty_recovery_synthetic": True}, {"role": "user", "content": "continue", "_empty_recovery_synthetic": True}, {"role": "assistant", "content": "done", "_thinking_prefill": True, "_empty_terminal_sentinel": True}, ] result = transport.convert_messages(msgs) for m in result: assert not any(k.startswith("_") for k in m), m # Visible content preserved assert result[1]["content"] == "(empty)" assert result[2]["content"] == "continue" # Original list untouched (deepcopy-on-demand) assert msgs[1]["_empty_recovery_synthetic"] is True def test_convert_messages_copy_on_write_for_dirty_history(self, transport): """Dirty provider metadata should not force a full-history deepcopy.""" clean_tool_call = { "id": "call_clean", "type": "function", "function": {"name": "safe", "arguments": "{}"}, } msgs = [ {"role": "user", "content": "hi", "metadata": {"large": ["shared"]}}, { "role": "assistant", "content": "ok", "tool_calls": [ clean_tool_call, { "id": "call_dirty", "call_id": "call_dirty", "response_item_id": "fc_dirty", "extra_content": {"google": {"thought_signature": "SIG"}}, "type": "function", "function": {"name": "t", "arguments": "{}"}, }, ], }, ] result = transport.convert_messages(msgs, model="gpt-4o") assert result is not msgs assert result[0] is msgs[0] assert result[1] is not msgs[1] assert result[1]["tool_calls"] is not msgs[1]["tool_calls"] assert result[1]["tool_calls"][0] is clean_tool_call assert result[1]["tool_calls"][1] is not msgs[1]["tool_calls"][1] assert "call_id" not in result[1]["tool_calls"][1] assert "response_item_id" not in result[1]["tool_calls"][1] assert "extra_content" not in result[1]["tool_calls"][1] assert "call_id" in msgs[1]["tool_calls"][1] assert "extra_content" in msgs[1]["tool_calls"][1] class TestChatCompletionsBuildKwargs: def test_basic_kwargs(self, transport): msgs = [{"role": "user", "content": "Hello"}] kw = transport.build_kwargs(model="gpt-4o", messages=msgs, timeout=30.0) assert kw["model"] == "gpt-4o" assert kw["messages"][0]["content"] == "Hello" assert kw["timeout"] == 30.0 def test_tools_included(self, transport): msgs = [{"role": "user", "content": "Hi"}] tools = [{"type": "function", "function": {"name": "test", "parameters": {}}}] kw = transport.build_kwargs(model="gpt-4o", messages=msgs, tools=tools) assert kw["tools"] == tools def test_openrouter_provider_prefs(self, transport): from providers import get_provider_profile profile = get_provider_profile("openrouter") msgs = [{"role": "user", "content": "Hi"}] kw = transport.build_kwargs( model="gpt-4o", messages=msgs, provider_profile=profile, provider_preferences={"only": ["openai"]}, ) assert kw["extra_body"]["provider"] == {"only": ["openai"]} def test_nous_tags(self, transport): from agent.portal_tags import nous_portal_tags from providers import get_provider_profile profile = get_provider_profile("nous") msgs = [{"role": "user", "content": "Hi"}] kw = transport.build_kwargs(model="gpt-4o", messages=msgs, provider_profile=profile) assert kw["extra_body"]["tags"] == nous_portal_tags() def test_reasoning_default(self, transport): msgs = [{"role": "user", "content": "Hi"}] kw = transport.build_kwargs( model="gpt-4o", messages=msgs, supports_reasoning=True, ) assert kw["extra_body"]["reasoning"] == {"enabled": True, "effort": "medium"} def test_nous_omits_disabled_reasoning(self, transport): from providers import get_provider_profile profile = get_provider_profile("nous") msgs = [{"role": "user", "content": "Hi"}] kw = transport.build_kwargs( model="gpt-4o", messages=msgs, provider_profile=profile, supports_reasoning=True, reasoning_config={"enabled": False}, ) # Nous rejects enabled=false; reasoning omitted entirely assert "reasoning" not in kw.get("extra_body", {}) def test_ollama_num_ctx(self, transport): from providers import get_provider_profile profile = get_provider_profile("custom") msgs = [{"role": "user", "content": "Hi"}] kw = transport.build_kwargs( model="llama3", messages=msgs, provider_profile=profile, ollama_num_ctx=32768, ) assert kw["extra_body"]["options"]["num_ctx"] == 32768 def test_custom_think_false(self, transport): from providers import get_provider_profile profile = get_provider_profile("custom") msgs = [{"role": "user", "content": "Hi"}] kw = transport.build_kwargs( model="qwen3", messages=msgs, provider_profile=profile, reasoning_config={"effort": "none"}, ) assert kw["extra_body"]["think"] is False def test_gemini_openai_compat_flash_reasoning_maps_to_nested_google_thinking_config(self, transport): msgs = [{"role": "user", "content": "Hi"}] kw = transport.build_kwargs( model="gemini-3-flash-preview", messages=msgs, provider_name="gemini", base_url="https://generativelanguage.googleapis.com/v1beta/openai", reasoning_config={"enabled": True, "effort": "high"}, ) assert "thinking_config" not in kw["extra_body"] assert kw["extra_body"]["extra_body"]["google"]["thinking_config"] == { "include_thoughts": True, "thinking_level": "high", } def test_omit_temperature(self, transport): """Omit temperature is set via ProviderProfile with OMIT_TEMPERATURE sentinel.""" from providers.base import ProviderProfile, OMIT_TEMPERATURE msgs = [{"role": "user", "content": "Hi"}] kw = transport.build_kwargs( model="gpt-4o", messages=msgs, provider_profile=ProviderProfile(name="_t", fixed_temperature=OMIT_TEMPERATURE), ) assert "temperature" not in kw class TestChatCompletionsKimi: """Regression tests for the Kimi/Moonshot quirks migrated into the transport.""" def test_moonshot_tool_schemas_are_sanitized_by_model_name(self, transport): """Aggregator routes (Nous, OpenRouter) hit Moonshot by model name, not base URL.""" tools = [ { "type": "function", "function": { "name": "search", "description": "Search", "parameters": { "type": "object", "properties": { "q": {"description": "query"}, # missing type }, }, }, }, ] kw = transport.build_kwargs( model="moonshotai/kimi-k2.6", messages=[{"role": "user", "content": "Hi"}], tools=tools, max_tokens_param_fn=lambda n: {"max_tokens": n}, ) assert kw["tools"][0]["function"]["parameters"]["properties"]["q"]["type"] == "string" def test_moonshot_outgoing_schema_carries_required_array(self, transport): """Moonshot 400s on object schemas without an explicit `required` array (#66835). Assert the wire-level tool schema — what actually leaves the transport — carries `required: []` on a zero-required-param tool.""" tools = [ { "type": "function", "function": { "name": "browser_snapshot", "description": "Snapshot", "parameters": {"type": "object", "properties": {}}, }, }, ] kw = transport.build_kwargs( model="moonshotai/kimi-k3", messages=[{"role": "user", "content": "Hi"}], tools=tools, max_tokens_param_fn=lambda n: {"max_tokens": n}, ) assert kw["tools"][0]["function"]["parameters"]["required"] == [] def test_non_moonshot_tools_are_not_mutated(self, transport): """Other models don't go through the Moonshot sanitizer.""" original_params = { "type": "object", "properties": {"q": {"description": "query"}}, # missing type } tools = [ { "type": "function", "function": { "name": "search", "description": "Search", "parameters": original_params, }, }, ] kw = transport.build_kwargs( model="anthropic/claude-sonnet-4.6", messages=[{"role": "user", "content": "Hi"}], tools=tools, max_tokens_param_fn=lambda n: {"max_tokens": n}, ) # The parameters dict is passed through untouched (no synthetic type) assert "type" not in kw["tools"][0]["function"]["parameters"]["properties"]["q"] class TestChatCompletionsLmStudioReasoning: """LM Studio publishes per-model reasoning ``allowed_options``. When the user requests an effort the model can't honor (e.g. ``high`` on a toggle-style ``["off","on"]`` model), the transport omits ``reasoning_effort`` so LM Studio falls back to the model's default — silently downgrading "high" to "low" would mislead the user. """ def test_omits_effort_when_high_not_allowed_toggle(self, transport): kw = transport.build_kwargs( model="gpt-oss", messages=[{"role": "user", "content": "Hi"}], is_lmstudio=True, supports_reasoning=True, reasoning_config={"effort": "high"}, lmstudio_reasoning_options=["off", "on"], ) assert "reasoning_effort" not in kw def test_passes_through_when_effort_allowed(self, transport): kw = transport.build_kwargs( model="gpt-oss", messages=[{"role": "user", "content": "Hi"}], is_lmstudio=True, supports_reasoning=True, reasoning_config={"effort": "high"}, lmstudio_reasoning_options=["off", "low", "medium", "high"], ) assert kw["reasoning_effort"] == "high" class TestChatCompletionsValidate: def test_none(self, transport): assert transport.validate_response(None) is False def test_valid(self, transport): r = SimpleNamespace(choices=[SimpleNamespace(message=SimpleNamespace(content="hi"))]) assert transport.validate_response(r) is True class TestChatCompletionsNormalize: def test_text_response(self, transport): r = SimpleNamespace( choices=[SimpleNamespace( message=SimpleNamespace(content="Hello", tool_calls=None, reasoning_content=None), finish_reason="stop", )], usage=SimpleNamespace(prompt_tokens=10, completion_tokens=5, total_tokens=15), ) nr = transport.normalize_response(r) assert isinstance(nr, NormalizedResponse) assert nr.content == "Hello" assert nr.finish_reason == "stop" assert nr.tool_calls is None def test_tool_call_response(self, transport): tc = SimpleNamespace( id="call_123", function=SimpleNamespace(name="terminal", arguments='{"command": "ls"}'), ) r = SimpleNamespace( choices=[SimpleNamespace( message=SimpleNamespace(content=None, tool_calls=[tc], reasoning_content=None), finish_reason="tool_calls", )], usage=SimpleNamespace(prompt_tokens=10, completion_tokens=20, total_tokens=30), ) nr = transport.normalize_response(r) assert len(nr.tool_calls) == 1 assert nr.tool_calls[0].name == "terminal" assert nr.tool_calls[0].id == "call_123" def test_empty_reasoning_content_preserved(self, transport): """DeepSeek can require an explicit empty reasoning_content replay field.""" r = SimpleNamespace( choices=[SimpleNamespace( message=SimpleNamespace( content=None, tool_calls=None, reasoning=None, reasoning_content="", ), finish_reason="stop", )], usage=None, ) nr = transport.normalize_response(r) assert nr.provider_data == {"reasoning_content": ""} assert nr.reasoning_content == "" def test_refusal_none_is_noop(self, transport): """The common case: ``refusal`` is None → behavior unchanged.""" r = SimpleNamespace( choices=[SimpleNamespace( message=SimpleNamespace( content="hello", tool_calls=None, reasoning_content=None, refusal=None, ), finish_reason="stop", )], usage=None, ) nr = transport.normalize_response(r) assert nr.finish_reason == "stop" assert nr.content == "hello" assert nr.provider_data is None class TestChatCompletionsCacheStats: def test_no_usage(self, transport): r = SimpleNamespace(usage=None) assert transport.extract_cache_stats(r) is None def test_deepseek_native_top_level_cache_hit_tokens(self, transport): """DeepSeek's native API (api.deepseek.com) reports cache hits as top-level prompt_cache_hit_tokens, not the OpenAI nested shape — the extractor must read it or direct DeepSeek sessions show 0% cache hit rate (#61871).""" r = SimpleNamespace( usage=SimpleNamespace( prompt_tokens_details=None, prompt_cache_hit_tokens=1500, prompt_cache_miss_tokens=500, ) ) result = transport.extract_cache_stats(r) assert result == {"cached_tokens": 1500, "creation_tokens": 0} class TestChatCompletionsGeminiNativeExtraBodyStrip: """Profile extra_body (e.g. Nous portal tags) must not reach a native Gemini endpoint — Google's REST API rejects unknown fields with HTTP 400. """ def _nous_profile(self): from providers import get_provider_profile return get_provider_profile("nous") def test_tags_stripped_when_endpoint_is_native_gemini(self, transport): kw = transport.build_kwargs( "anthropic/claude-sonnet-4.6", [{"role": "user", "content": "hi"}], None, provider_profile=self._nous_profile(), base_url="https://generativelanguage.googleapis.com/v1beta", session_id="s1", max_tokens=None, ) eb = kw.get("extra_body") assert not eb or "tags" not in eb def test_tags_preserved_on_nous_endpoint(self, transport): kw = transport.build_kwargs( "hermes-3-405b", [{"role": "user", "content": "hi"}], None, provider_profile=self._nous_profile(), base_url="https://inference.nousresearch.com/v1", session_id="s1", max_tokens=None, ) eb = kw.get("extra_body") assert eb and "tags" in eb def test_tags_pass_through_on_gemini_openai_compat(self, transport): # /openai compat endpoint is not "native" — unchanged behavior. kw = transport.build_kwargs( "anthropic/claude-sonnet-4.6", [{"role": "user", "content": "hi"}], None, provider_profile=self._nous_profile(), base_url="https://generativelanguage.googleapis.com/v1beta/openai", session_id="s1", max_tokens=None, ) eb = kw.get("extra_body") assert eb and "tags" in eb class TestPromptCacheKeyCapability: """Chat Completions cache routing is opt-in and body-safe.""" @staticmethod def _messages(instructions="You are stable."): return [ {"role": "system", "content": instructions}, {"role": "user", "content": "hello"}, ] @staticmethod def _tools(name="lookup"): return [{ "type": "function", "function": { "name": name, "description": "Look something up.", "parameters": {"type": "object", "properties": {}}, }, }] def _request_body(self, kwargs, *, stream=False): captured = {} def handler(request): captured.update(json.loads(request.content)) if stream: return httpx.Response( 200, headers={"content-type": "text/event-stream"}, content=( 'data: {"id":"chatcmpl_1","object":"chat.completion.chunk",' '"choices":[{"index":0,"delta":{"content":"ok"},' '"finish_reason":null}]}\n\n' "data: [DONE]\n\n" ), ) return httpx.Response(200, json={ "id": "chatcmpl_1", "object": "chat.completion", "created": 0, "model": kwargs["model"], "choices": [{ "index": 0, "message": {"role": "assistant", "content": "ok"}, "finish_reason": "stop", }], }) with httpx.Client(transport=httpx.MockTransport(handler)) as http_client: client = OpenAI( api_key="test-key", base_url="https://cache-capable.test/v1", http_client=http_client, ) result = client.chat.completions.create(**kwargs, stream=stream) if stream: list(result) return captured def test_profile_capability_emits_content_key_in_nonstream_request_body(self, transport): from providers.base import ProviderProfile kwargs = transport.build_kwargs( model="cache-model", messages=self._messages(), tools=self._tools(), session_id="cron_job_2026-07-15T10:00:00Z", provider_profile=ProviderProfile( name="cache-capable", supports_prompt_cache_key=True, ), ) body = self._request_body(kwargs) assert body["prompt_cache_key"].startswith("pck_") assert body["prompt_cache_key"] == kwargs["prompt_cache_key"] def test_legacy_capability_emits_same_key_in_streaming_request_body(self, transport): kwargs = transport.build_kwargs( model="cache-model", messages=self._messages(), tools=self._tools(), session_id="cron_job_2026-07-15T10:05:00Z", supports_prompt_cache_key=True, ) body = self._request_body(kwargs, stream=True) assert body["prompt_cache_key"] == kwargs["prompt_cache_key"] def test_openai_api_base_url_implies_capability(self, transport): """api.openai.com gets the key WITHOUT an explicit flag (exact host).""" kwargs = transport.build_kwargs( model="gpt-cache-model", messages=self._messages(), tools=self._tools(), session_id="cron_job_2026-07-15T10:07:00Z", base_url="https://api.openai.com/v1", ) assert kwargs["prompt_cache_key"].startswith("pck_") @pytest.mark.parametrize( "base_url", [ "https://myproxy.example.com/api.openai.com/v1", # host embedded in path "https://api.openai.com.evil.example/v1", # prefix-spoofed host "https://eastus.api.cognitive.microsoft.com/openai/v1", # Azure ], ) def test_non_openai_hosts_do_not_imply_capability(self, transport, base_url): kwargs = transport.build_kwargs( model="strict-model", messages=self._messages(), tools=self._tools(), session_id="cron_job_2026-07-15T10:08:00Z", base_url=base_url, ) assert "prompt_cache_key" not in kwargs @pytest.mark.parametrize("provider", [None, "anthropic", "custom"]) def test_default_off_never_leaks_unknown_body_field(self, transport, provider): from providers import get_provider_profile kwargs = transport.build_kwargs( model="strict-model", messages=self._messages(), tools=self._tools(), session_id="cron_job_2026-07-15T10:00:00Z", provider_profile=(get_provider_profile(provider) if provider else None), ) body = self._request_body(kwargs) assert "prompt_cache_key" not in kwargs assert "prompt_cache_key" not in body def test_explicit_top_level_and_extra_body_overrides_are_preserved(self, transport): from providers.base import ProviderProfile profile = ProviderProfile(name="cache-capable", supports_prompt_cache_key=True) top_level = transport.build_kwargs( model="cache-model", messages=self._messages(), tools=self._tools(), provider_profile=profile, request_overrides={"prompt_cache_key": "caller-top-level"}, ) in_extra_body = transport.build_kwargs( model="cache-model", messages=self._messages(), tools=self._tools(), provider_profile=profile, request_overrides={"extra_body": {"prompt_cache_key": "caller-extra-body"}}, ) assert top_level["prompt_cache_key"] == "caller-top-level" assert "prompt_cache_key" not in top_level.get("extra_body", {}) assert "prompt_cache_key" not in in_extra_body assert in_extra_body["extra_body"]["prompt_cache_key"] == "caller-extra-body" def test_cron_ids_share_static_prefix_key_and_content_changes_invalidate(self, transport): def key(session_id, *, instructions="You are stable.", tool_name="lookup"): return transport.build_kwargs( model="cache-model", messages=self._messages(instructions), tools=self._tools(tool_name), session_id=session_id, supports_prompt_cache_key=True, )["prompt_cache_key"] first = key("cron_job_2026-07-15T10:00:00Z") second = key("cron_job_2026-07-15T10:05:00Z") assert first == second assert first != key("cron_job_2026-07-15T10:05:00Z", instructions="You are different.") assert first != key("cron_job_2026-07-15T10:05:00Z", tool_name="search")