openai--openai-agents-python/tests/models/test_deepseek_reasoning_content.py

from typing import Any

import litellm
import pytest
from litellm.types.utils import (
    ChatCompletionMessageToolCall,
    Choices,
    Function,
    Message,
    ModelResponse,
    Usage,
)

from agents.extensions.models.litellm_model import LitellmModel
from agents.model_settings import ModelSettings
from agents.models.chatcmpl_converter import Converter
from agents.models.interface import ModelTracing


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_deepseek_reasoning_content_preserved_in_tool_calls(monkeypatch):
    """
    Ensure DeepSeek reasoning_content is preserved when converting items to messages.

    DeepSeek requires reasoning_content field in assistant messages with tool_calls.
    This test verifies that reasoning content from reasoning items is correctly
    extracted and added to assistant messages during conversion.
    """
    # Capture the messages sent to the model
    captured_calls: list[dict[str, Any]] = []

    async def fake_acompletion(model, messages=None, **kwargs):
        captured_calls.append({"model": model, "messages": messages, **kwargs})

        # First call: model returns reasoning_content + tool_call
        if len(captured_calls) == 1:
            tool_call = ChatCompletionMessageToolCall(
                id="call_123",
                type="function",
                function=Function(name="get_weather", arguments='{"city": "Tokyo"}'),
            )
            msg = Message(
                role="assistant",
                content=None,
                tool_calls=[tool_call],
            )
            # DeepSeek adds reasoning_content to the message
            msg.reasoning_content = "Let me think about getting the weather for Tokyo..."

            choice = Choices(index=0, message=msg)
            return ModelResponse(choices=[choice], usage=Usage(100, 50, 150))

        # Second call: model returns final response
        msg = Message(role="assistant", content="The weather in Tokyo is sunny.")
        choice = Choices(index=0, message=msg)
        return ModelResponse(choices=[choice], usage=Usage(100, 50, 150))

    monkeypatch.setattr(litellm, "acompletion", fake_acompletion)

    model = LitellmModel(model="deepseek/deepseek-reasoner")

    # First call: get the tool call response
    first_response = await model.get_response(
        system_instructions="You are a helpful assistant.",
        input="What's the weather in Tokyo?",
        model_settings=ModelSettings(),
        tools=[],  # We'll simulate the tool response manually
        output_schema=None,
        handoffs=[],
        tracing=ModelTracing.DISABLED,
    )

    assert len(first_response.output) >= 1

    input_items: list[Any] = []
    input_items.append({"role": "user", "content": "What's the weather in Tokyo?"})

    for item in first_response.output:
        if hasattr(item, "model_dump"):
            input_items.append(item.model_dump())
        else:
            input_items.append(item)

    input_items.append(
        {
            "type": "function_call_output",
            "call_id": "call_123",
            "output": "The weather in Tokyo is sunny.",
        }
    )

    messages = Converter.items_to_messages(
        input_items,
        model="deepseek/deepseek-reasoner",
    )

    assistant_messages_with_tool_calls = [
        m
        for m in messages
        if isinstance(m, dict) and m.get("role") == "assistant" and m.get("tool_calls")
    ]

    assert len(assistant_messages_with_tool_calls) > 0
    assistant_msg = assistant_messages_with_tool_calls[0]
    assert "reasoning_content" in assistant_msg


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_deepseek_reasoning_content_in_multi_turn_conversation(monkeypatch):
    """
    Verify reasoning_content is included in assistant messages during multi-turn conversations.

    When DeepSeek returns reasoning_content with tool_calls, subsequent API calls must
    include the reasoning_content field in the assistant message to avoid 400 errors.
    """
    captured_calls: list[dict[str, Any]] = []

    async def fake_acompletion(model, messages=None, **kwargs):
        captured_calls.append({"model": model, "messages": messages, **kwargs})

        # First call: model returns reasoning_content + tool_call
        if len(captured_calls) == 1:
            tool_call = ChatCompletionMessageToolCall(
                id="call_weather_123",
                type="function",
                function=Function(name="get_weather", arguments='{"city": "Tokyo"}'),
            )
            msg = Message(
                role="assistant",
                content=None,
                tool_calls=[tool_call],
            )
            # DeepSeek adds reasoning_content
            msg.reasoning_content = "I need to get the weather for Tokyo first."
            choice = Choices(index=0, message=msg)
            return ModelResponse(choices=[choice], usage=Usage(100, 50, 150))

        # Second call: check if reasoning_content was in the request
        # In real DeepSeek API, this would fail with 400 if reasoning_content is missing
        msg = Message(
            role="assistant", content="Based on my findings, the weather in Tokyo is sunny."
        )
        choice = Choices(index=0, message=msg)
        return ModelResponse(choices=[choice], usage=Usage(100, 50, 150))

    monkeypatch.setattr(litellm, "acompletion", fake_acompletion)

    model = LitellmModel(model="deepseek/deepseek-reasoner")

    # First call
    first_response = await model.get_response(
        system_instructions="You are a helpful assistant.",
        input="What's the weather in Tokyo?",
        model_settings=ModelSettings(),
        tools=[],
        output_schema=None,
        handoffs=[],
        tracing=ModelTracing.DISABLED,
    )

    input_items: list[Any] = []
    input_items.append({"role": "user", "content": "What's the weather in Tokyo?"})

    for item in first_response.output:
        if hasattr(item, "model_dump"):
            input_items.append(item.model_dump())
        else:
            input_items.append(item)

    input_items.append(
        {
            "type": "function_call_output",
            "call_id": "call_weather_123",
            "output": "The weather in Tokyo is sunny and 22°C.",
        }
    )

    await model.get_response(
        system_instructions="You are a helpful assistant.",
        input=input_items,
        model_settings=ModelSettings(),
        tools=[],
        output_schema=None,
        handoffs=[],
        tracing=ModelTracing.DISABLED,
    )

    assert len(captured_calls) == 2

    second_call_messages = captured_calls[1]["messages"]

    assistant_with_tools = None
    for msg in second_call_messages:
        if isinstance(msg, dict) and msg.get("role") == "assistant" and msg.get("tool_calls"):
            assistant_with_tools = msg
            break

    assert assistant_with_tools is not None
    assert "reasoning_content" in assistant_with_tools


def test_deepseek_reasoning_content_with_openai_chatcompletions_path():
    """
    Verify reasoning_content works when using OpenAIChatCompletionsModel.

    This ensures the fix works for both LiteLLM and OpenAI ChatCompletions code paths.
    """
    from agents.models.chatcmpl_converter import Converter

    input_items: list[Any] = [
        {"role": "user", "content": "What's the weather in Paris?"},
        {
            "id": "__fake_id__",
            "summary": [{"text": "I need to check the weather in Paris.", "type": "summary_text"}],
            "type": "reasoning",
            "content": None,
            "encrypted_content": None,
            "status": None,
            "provider_data": {"model": "deepseek-reasoner", "response_id": "chatcmpl-test"},
        },
        {
            "arguments": '{"city": "Paris"}',
            "call_id": "call_weather_456",
            "name": "get_weather",
            "type": "function_call",
            "id": "__fake_id__",
            "status": None,
            "provider_data": {"model": "deepseek-reasoner"},
        },
        {
            "type": "function_call_output",
            "call_id": "call_weather_456",
            "output": "The weather in Paris is cloudy and 15°C.",
        },
    ]

    messages = Converter.items_to_messages(
        input_items,
        model="deepseek-reasoner",
    )

    assistant_with_tools = None
    for msg in messages:
        if isinstance(msg, dict) and msg.get("role") == "assistant" and msg.get("tool_calls"):
            assistant_with_tools = msg
            break

    assert assistant_with_tools is not None
    assert "reasoning_content" in assistant_with_tools
    # Use type: ignore since reasoning_content is a dynamic field not in OpenAI's TypedDict
    assert assistant_with_tools["reasoning_content"] == "I need to check the weather in Paris."  # type: ignore[typeddict-item]


def test_reasoning_content_from_other_provider_not_attached_to_deepseek():
    """
    Verify reasoning_content from non-DeepSeek providers is NOT attached to DeepSeek messages.

    When switching models mid-conversation (e.g., from Claude to DeepSeek), reasoning items
    that originated from Claude should not have their summaries attached as reasoning_content
    to DeepSeek assistant messages, as this would leak unrelated reasoning and may trigger
    DeepSeek 400 errors.
    """
    from agents.models.chatcmpl_converter import Converter

    input_items: list[Any] = [
        {"role": "user", "content": "What's the weather in Paris?"},
        {
            "id": "__fake_id__",
            "summary": [{"text": "Claude's reasoning about the weather.", "type": "summary_text"}],
            "type": "reasoning",
            "content": None,
            "encrypted_content": None,
            "status": None,
            # this one came from Claude, not DeepSeek
            "provider_data": {"model": "claude-sonnet-4-20250514", "response_id": "chatcmpl-test"},
        },
        {
            "arguments": '{"city": "Paris"}',
            "call_id": "call_weather_789",
            "name": "get_weather",
            "type": "function_call",
            "id": "__fake_id__",
            "status": None,
            "provider_data": {"model": "claude-sonnet-4-20250514"},
        },
        {
            "type": "function_call_output",
            "call_id": "call_weather_789",
            "output": "The weather in Paris is cloudy.",
        },
    ]

    messages = Converter.items_to_messages(
        input_items,
        model="deepseek-reasoner",
    )

    assistant_with_tools = None
    for msg in messages:
        if isinstance(msg, dict) and msg.get("role") == "assistant" and msg.get("tool_calls"):
            assistant_with_tools = msg
            break

    assert assistant_with_tools is not None
    # reasoning_content should NOT be present since the reasoning came from Claude, not DeepSeek
    assert "reasoning_content" not in assistant_with_tools


def test_reasoning_content_without_provider_data_attached_for_backward_compat():
    """
    Verify reasoning_content from items without provider_data is attached for backward compat.

    For older items that don't have provider_data (before provider tracking was added),
    we should still attach reasoning_content to maintain backward compatibility.
    """
    from agents.models.chatcmpl_converter import Converter

    # Reasoning item without provider_data (older format)
    input_items: list[Any] = [
        {"role": "user", "content": "What's the weather in Tokyo?"},
        {
            "id": "__fake_id__",
            "summary": [{"text": "Reasoning without provider info.", "type": "summary_text"}],
            "type": "reasoning",
            "content": None,
            "encrypted_content": None,
            "status": None,
            # No provider_data
        },
        {
            "arguments": '{"city": "Tokyo"}',
            "call_id": "call_weather_101",
            "name": "get_weather",
            "type": "function_call",
            "id": "__fake_id__",
            "status": None,
        },
        {
            "type": "function_call_output",
            "call_id": "call_weather_101",
            "output": "The weather in Tokyo is sunny.",
        },
    ]

    messages = Converter.items_to_messages(
        input_items,
        model="deepseek-reasoner",
    )

    assistant_with_tools = None
    for msg in messages:
        if isinstance(msg, dict) and msg.get("role") == "assistant" and msg.get("tool_calls"):
            assistant_with_tools = msg
            break

    assert assistant_with_tools is not None
    # reasoning_content SHOULD be present for backward compatibility
    assert "reasoning_content" in assistant_with_tools
    assert assistant_with_tools["reasoning_content"] == "Reasoning without provider info."  # type: ignore[typeddict-item]