openai--openai-agents-python/tests/test_run_internal_items.py

from __future__ import annotations

from typing import Any, cast

import pytest
from openai.types.responses import (
    ResponseFunctionToolCall,
    ResponseToolSearchCall,
    ResponseToolSearchOutputItem,
)
from openai.types.responses.response_reasoning_item import ResponseReasoningItem

from agents import Agent
from agents.exceptions import AgentsException
from agents.items import (
    ReasoningItem,
    ToolCallItem,
    ToolSearchCallItem,
    ToolSearchOutputItem,
    TResponseInputItem,
    coerce_tool_search_output_raw_item,
)
from agents.models.fake_id import FAKE_RESPONSES_ID
from agents.result import RunResult
from agents.run_context import RunContextWrapper
from agents.run_internal import items as run_items


def test_drop_orphan_function_calls_preserves_non_mapping_entries() -> None:
    payload: list[Any] = [
        cast(TResponseInputItem, "plain-text-input"),
        cast(TResponseInputItem, {"type": "message", "role": "user", "content": "hello"}),
        cast(
            TResponseInputItem,
            {
                "type": "function_call",
                "call_id": "orphan_call",
                "name": "orphan",
                "arguments": "{}",
            },
        ),
        cast(
            TResponseInputItem,
            {
                "type": "function_call",
                "call_id": "paired_call",
                "name": "paired",
                "arguments": "{}",
            },
        ),
        cast(
            TResponseInputItem,
            {"type": "function_call_output", "call_id": "paired_call", "output": "ok"},
        ),
        cast(TResponseInputItem, {"call_id": "not-a-tool-call"}),
    ]

    filtered = run_items.drop_orphan_function_calls(cast(list[TResponseInputItem], payload))
    filtered_values = cast(list[Any], filtered)
    assert "plain-text-input" in filtered_values
    assert cast(dict[str, Any], filtered[1])["type"] == "message"
    assert any(
        isinstance(entry, dict)
        and entry.get("type") == "function_call"
        and entry.get("call_id") == "paired_call"
        for entry in filtered
    )
    assert not any(
        isinstance(entry, dict)
        and entry.get("type") == "function_call"
        and entry.get("call_id") == "orphan_call"
        for entry in filtered
    )


def test_drop_orphan_function_calls_drops_reasoning_preceding_dropped_tool_call() -> None:
    # Regression: reasoning items tied to a now-dropped orphan tool call would otherwise be
    # forwarded to the API and trigger
    # ``Item 'rs_...' of type 'reasoning' was provided without its required following item``.
    payload: list[Any] = [
        cast(TResponseInputItem, {"role": "user", "content": "hi"}),
        cast(TResponseInputItem, {"type": "reasoning", "id": "rs_orphan_a", "summary": []}),
        cast(TResponseInputItem, {"type": "reasoning", "id": "rs_orphan_b", "summary": []}),
        cast(
            TResponseInputItem,
            {
                "type": "function_call",
                "call_id": "orphan_call",
                "name": "orphan",
                "arguments": "{}",
            },
        ),
        cast(TResponseInputItem, {"type": "reasoning", "id": "rs_paired", "summary": []}),
        cast(
            TResponseInputItem,
            {
                "type": "function_call",
                "call_id": "paired_call",
                "name": "paired",
                "arguments": "{}",
            },
        ),
        cast(
            TResponseInputItem,
            {"type": "function_call_output", "call_id": "paired_call", "output": "ok"},
        ),
    ]

    filtered = run_items.drop_orphan_function_calls(cast(list[TResponseInputItem], payload))

    reasoning_ids = [
        entry.get("id")
        for entry in filtered
        if isinstance(entry, dict) and entry.get("type") == "reasoning"
    ]
    assert reasoning_ids == ["rs_paired"]
    assert not any(
        isinstance(entry, dict)
        and entry.get("type") == "function_call"
        and entry.get("call_id") == "orphan_call"
        for entry in filtered
    )


def test_drop_orphan_function_calls_keeps_lone_reasoning_when_no_tool_calls_dropped() -> None:
    # Server-managed conversations (or compaction) may forward standalone reasoning items whose
    # required following item lives in the server-side conversation. We must not drop those.
    payload: list[Any] = [
        cast(TResponseInputItem, {"type": "reasoning", "id": "rs_lone", "summary": []}),
    ]

    filtered = run_items.drop_orphan_function_calls(cast(list[TResponseInputItem], payload))

    assert filtered == payload


def test_drop_orphan_function_calls_handles_tool_search_calls() -> None:
    payload: list[Any] = [
        cast(
            TResponseInputItem,
            {
                "type": "tool_search_call",
                "call_id": "tool_search_orphan",
                "arguments": {"query": "orphan"},
                "execution": "server",
                "status": "completed",
            },
        ),
        cast(
            TResponseInputItem,
            {
                "type": "tool_search_call",
                "call_id": "tool_search_keep",
                "arguments": {"query": "keep"},
                "execution": "server",
                "status": "completed",
            },
        ),
        cast(
            TResponseInputItem,
            {
                "type": "tool_search_output",
                "call_id": "tool_search_keep",
                "execution": "server",
                "status": "completed",
                "tools": [],
            },
        ),
    ]

    filtered = run_items.drop_orphan_function_calls(cast(list[TResponseInputItem], payload))

    assert any(
        isinstance(entry, dict)
        and entry.get("type") == "tool_search_call"
        and entry.get("call_id") == "tool_search_keep"
        for entry in filtered
    )
    assert not any(
        isinstance(entry, dict)
        and entry.get("type") == "tool_search_call"
        and entry.get("call_id") == "tool_search_orphan"
        for entry in filtered
    )


def test_drop_orphan_function_calls_preserves_hosted_tool_search_pairs_without_call_ids() -> None:
    payload: list[Any] = [
        cast(
            TResponseInputItem,
            {
                "type": "tool_search_call",
                "call_id": None,
                "arguments": {"query": "keep"},
                "execution": "server",
                "status": "completed",
            },
        ),
        cast(
            TResponseInputItem,
            {
                "type": "tool_search_output",
                "call_id": None,
                "execution": "server",
                "status": "completed",
                "tools": [],
            },
        ),
    ]

    filtered = run_items.drop_orphan_function_calls(cast(list[TResponseInputItem], payload))

    assert len(filtered) == 2
    assert cast(dict[str, Any], filtered[0])["type"] == "tool_search_call"
    assert cast(dict[str, Any], filtered[1])["type"] == "tool_search_output"


def test_drop_orphan_function_calls_matches_latest_anonymous_tool_search_call() -> None:
    payload: list[Any] = [
        cast(
            TResponseInputItem,
            {
                "type": "tool_search_call",
                "call_id": None,
                "arguments": {"query": "orphan"},
                "execution": "server",
                "status": "completed",
            },
        ),
        cast(
            TResponseInputItem,
            {
                "type": "tool_search_call",
                "call_id": None,
                "arguments": {"query": "paired"},
                "execution": "server",
                "status": "completed",
            },
        ),
        cast(
            TResponseInputItem,
            {
                "type": "tool_search_output",
                "call_id": None,
                "execution": "server",
                "status": "completed",
                "tools": [],
            },
        ),
    ]

    filtered = run_items.drop_orphan_function_calls(cast(list[TResponseInputItem], payload))

    assert [cast(dict[str, Any], item)["type"] for item in filtered] == [
        "tool_search_call",
        "tool_search_output",
    ]
    assert cast(dict[str, Any], filtered[0])["arguments"] == {"query": "paired"}


def test_drop_orphan_function_calls_does_not_pair_named_tool_search_with_anonymous_output() -> None:
    payload: list[Any] = [
        cast(
            TResponseInputItem,
            {
                "type": "tool_search_call",
                "call_id": "orphan_search",
                "arguments": {"query": "keep"},
                "execution": "server",
                "status": "completed",
            },
        ),
        cast(
            TResponseInputItem,
            {
                "type": "tool_search_output",
                "call_id": None,
                "execution": "server",
                "status": "completed",
                "tools": [],
            },
        ),
    ]

    filtered = run_items.drop_orphan_function_calls(cast(list[TResponseInputItem], payload))

    assert [cast(dict[str, Any], item)["type"] for item in filtered] == ["tool_search_output"]


def test_drop_orphan_function_calls_keeps_reasoning_chain_before_non_dropped_item() -> None:
    payload: list[Any] = [
        cast(TResponseInputItem, {"type": "reasoning", "id": "rs_1", "summary": []}),
        cast(TResponseInputItem, {"type": "reasoning", "id": "rs_2", "summary": []}),
        cast(TResponseInputItem, {"type": "message", "role": "assistant", "content": []}),
        cast(
            TResponseInputItem,
            {
                "type": "function_call",
                "call_id": "orphan_call",
                "name": "orphan",
                "arguments": "{}",
            },
        ),
    ]

    filtered = run_items.drop_orphan_function_calls(cast(list[TResponseInputItem], payload))

    assert [cast(dict[str, Any], item)["id"] for item in filtered[:2]] == ["rs_1", "rs_2"]
    assert [cast(dict[str, Any], item)["type"] for item in filtered] == [
        "reasoning",
        "reasoning",
        "message",
    ]


def test_normalize_and_ensure_input_item_format_keep_non_dict_entries() -> None:
    item = cast(TResponseInputItem, "raw-item")
    assert run_items.ensure_input_item_format(item) == item
    assert run_items.normalize_input_items_for_api([item]) == [item]


def test_fingerprint_input_item_handles_edge_cases(monkeypatch: pytest.MonkeyPatch) -> None:
    assert run_items.fingerprint_input_item(None) is None

    fingerprint = run_items.fingerprint_input_item(
        cast(
            TResponseInputItem, {"id": "id-1", "type": "message", "role": "user", "content": "hi"}
        ),
        ignore_ids_for_matching=True,
    )
    assert fingerprint is not None
    assert '"id"' not in fingerprint

    class _BrokenModelDump:
        def model_dump(self, *_args: Any, **kwargs: Any) -> dict[str, Any]:
            if "warnings" in kwargs:
                raise TypeError("warnings arg unsupported")
            raise RuntimeError("still broken")

    assert run_items.fingerprint_input_item(_BrokenModelDump()) is None
    assert run_items._model_dump_without_warnings(object()) is None

    class _Opaque:
        pass

    monkeypatch.setattr(
        run_items,
        "ensure_input_item_format",
        lambda _item: {"id": "internal-id", "type": "message", "role": "user", "content": "x"},
    )
    opaque_fingerprint = run_items.fingerprint_input_item(_Opaque(), ignore_ids_for_matching=True)
    assert opaque_fingerprint is not None
    assert '"id"' not in opaque_fingerprint


def test_fingerprint_input_item_returns_none_when_serialization_fails(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    def _raise_json_error(*_args: Any, **_kwargs: Any) -> str:
        raise RuntimeError("boom")

    monkeypatch.setattr(cast(Any, run_items).json, "dumps", _raise_json_error)

    assert run_items.fingerprint_input_item({"type": "message", "role": "user"}) is None


def test_strip_metadata_and_reasoning_id_helpers_keep_non_matching_items() -> None:
    raw = cast(TResponseInputItem, "raw-item")
    non_reasoning = cast(TResponseInputItem, {"type": "message", "id": "msg_1"})
    reasoning_without_id = cast(TResponseInputItem, {"type": "reasoning", "summary": []})

    assert run_items.strip_internal_input_item_metadata(raw) == raw
    assert run_items._without_reasoning_item_id(raw) == raw
    assert run_items._without_reasoning_item_id(non_reasoning) == non_reasoning
    assert run_items._without_reasoning_item_id(reasoning_without_id) == reasoning_without_id


def test_deduplicate_input_items_handles_fake_ids_and_approval_request_ids() -> None:
    items: list[Any] = [
        cast(
            TResponseInputItem,
            {
                "type": "function_call_output",
                "id": FAKE_RESPONSES_ID,
                "call_id": "call-1",
                "output": "first",
            },
        ),
        cast(
            TResponseInputItem,
            {
                "type": "function_call_output",
                "id": FAKE_RESPONSES_ID,
                "call_id": "call-1",
                "output": "latest",
            },
        ),
        cast(
            TResponseInputItem,
            {
                "type": "mcp_approval_response",
                "approval_request_id": "req-1",
                "approve": True,
            },
        ),
        cast(
            TResponseInputItem,
            {
                "type": "mcp_approval_response",
                "approval_request_id": "req-1",
                "approve": False,
            },
        ),
        cast(TResponseInputItem, "plain"),
    ]

    deduplicated = run_items.deduplicate_input_items(cast(list[TResponseInputItem], items))
    assert len(deduplicated) == 3
    assert cast(list[Any], deduplicated)[-1] == "plain"

    latest = run_items.deduplicate_input_items_preferring_latest(
        cast(list[TResponseInputItem], items[:2])
    )
    assert len(latest) == 1
    latest_output = cast(dict[str, Any], latest[0])
    assert latest_output["output"] == "latest"


def test_extract_mcp_request_id_supports_dicts_and_objects() -> None:
    assert (
        run_items.extract_mcp_request_id(
            {"provider_data": {"id": "provider-id"}, "id": "fallback-id"}
        )
        == "provider-id"
    )
    assert run_items.extract_mcp_request_id({"call_id": "call-id"}) == "call-id"

    class _WithProviderData:
        provider_data = {"id": "from-provider"}

    assert run_items.extract_mcp_request_id(_WithProviderData()) == "from-provider"

    class _BrokenObject:
        @property
        def provider_data(self) -> dict[str, Any]:
            raise RuntimeError("boom")

        def __getattr__(self, _name: str) -> Any:
            raise RuntimeError("boom")

    assert run_items.extract_mcp_request_id(_BrokenObject()) is None


def test_extract_mcp_request_id_from_run_variants() -> None:
    class _Run:
        def __init__(self, request_item: Any = None, requestItem: Any = None) -> None:
            self.request_item = request_item
            self.requestItem = requestItem

    class _RequestObject:
        provider_data = {"id": "provider-object"}
        id = "object-id"
        call_id = "object-call-id"

    assert (
        run_items.extract_mcp_request_id_from_run(
            _Run(request_item={"provider_data": {"id": "provider-dict"}, "id": "fallback"})
        )
        == "provider-dict"
    )
    assert (
        run_items.extract_mcp_request_id_from_run(_Run(request_item={"id": "dict-id"})) == "dict-id"
    )
    assert (
        run_items.extract_mcp_request_id_from_run(_Run(request_item=_RequestObject()))
        == "provider-object"
    )
    assert (
        run_items.extract_mcp_request_id_from_run(_Run(requestItem={"call_id": "camel-call"}))
        == "camel-call"
    )


def test_run_item_to_input_item_preserves_reasoning_item_ids_by_default() -> None:
    agent = Agent(name="A")
    reasoning = ReasoningItem(
        agent=agent,
        raw_item=ResponseReasoningItem(
            type="reasoning",
            id="rs_123",
            summary=[],
        ),
    )

    result = run_items.run_item_to_input_item(reasoning)

    assert isinstance(result, dict)
    assert result.get("type") == "reasoning"
    assert result.get("id") == "rs_123"


def test_run_item_to_input_item_omits_reasoning_item_ids_when_configured() -> None:
    agent = Agent(name="A")
    reasoning = ReasoningItem(
        agent=agent,
        raw_item=ResponseReasoningItem(
            type="reasoning",
            id="rs_456",
            summary=[],
        ),
    )

    result = run_items.run_item_to_input_item(reasoning, "omit")

    assert isinstance(result, dict)
    assert result.get("type") == "reasoning"
    assert "id" not in result


def test_run_item_to_input_item_preserves_tool_search_items() -> None:
    agent = Agent(name="A")
    tool_search_call = ToolSearchCallItem(
        agent=agent,
        raw_item={"type": "tool_search_call", "queries": [{"search_term": "profile"}]},
    )
    tool_search_output = ToolSearchOutputItem(
        agent=agent,
        raw_item={"type": "tool_search_output", "results": [{"text": "Customer profile"}]},
    )

    converted_call = run_items.run_item_to_input_item(tool_search_call)
    converted_output = run_items.run_item_to_input_item(tool_search_output)

    assert isinstance(converted_call, dict)
    assert converted_call["type"] == "tool_search_call"
    assert isinstance(converted_output, dict)
    assert converted_output["type"] == "tool_search_output"


def test_run_item_to_input_item_strips_tool_search_created_by() -> None:
    agent = Agent(name="A")
    tool_search_call = ToolSearchCallItem(
        agent=agent,
        raw_item=ResponseToolSearchCall(
            id="tsc_123",
            type="tool_search_call",
            arguments={"query": "profile"},
            execution="client",
            status="completed",
            created_by="server",
        ),
    )
    tool_search_output = ToolSearchOutputItem(
        agent=agent,
        raw_item=ResponseToolSearchOutputItem(
            id="tso_123",
            type="tool_search_output",
            execution="client",
            status="completed",
            tools=[],
            created_by="server",
        ),
    )

    converted_call = run_items.run_item_to_input_item(tool_search_call)
    converted_output = run_items.run_item_to_input_item(tool_search_output)

    assert isinstance(converted_call, dict)
    assert converted_call["type"] == "tool_search_call"
    assert "created_by" not in converted_call
    assert isinstance(converted_output, dict)
    assert converted_output["type"] == "tool_search_output"
    assert "created_by" not in converted_output


def test_run_item_to_input_item_omits_tool_call_metadata() -> None:
    agent = Agent(name="A")
    tool_call = ToolCallItem(
        agent=agent,
        raw_item=ResponseFunctionToolCall(
            id="fc_123",
            call_id="call_123",
            name="lookup_account",
            arguments="{}",
            type="function_call",
            status="completed",
        ),
        description="Lookup customer records.",
        title="Lookup Account",
    )

    result = run_items.run_item_to_input_item(tool_call)
    result_dict = cast(dict[str, Any], result)

    assert isinstance(result, dict)
    assert result_dict["type"] == "function_call"
    assert "description" not in result_dict
    assert "title" not in result_dict


def test_normalize_input_items_for_api_strips_internal_tool_call_metadata() -> None:
    item = cast(
        TResponseInputItem,
        {
            "type": "function_call",
            "call_id": "call_123",
            "name": "lookup_account",
            "arguments": "{}",
            run_items.TOOL_CALL_SESSION_DESCRIPTION_KEY: "Lookup customer records.",
            run_items.TOOL_CALL_SESSION_TITLE_KEY: "Lookup Account",
        },
    )

    normalized = run_items.normalize_input_items_for_api([item])
    normalized_item = cast(dict[str, Any], normalized[0])

    assert run_items.TOOL_CALL_SESSION_DESCRIPTION_KEY not in normalized_item
    assert run_items.TOOL_CALL_SESSION_TITLE_KEY not in normalized_item


def test_fingerprint_input_item_ignores_internal_tool_call_metadata() -> None:
    base_item = cast(
        TResponseInputItem,
        {
            "type": "function_call",
            "call_id": "call_123",
            "name": "lookup_account",
            "arguments": "{}",
        },
    )
    with_metadata = cast(
        TResponseInputItem,
        {
            **cast(dict[str, Any], base_item),
            run_items.TOOL_CALL_SESSION_DESCRIPTION_KEY: "Lookup customer records.",
            run_items.TOOL_CALL_SESSION_TITLE_KEY: "Lookup Account",
        },
    )

    assert run_items.fingerprint_input_item(base_item) == run_items.fingerprint_input_item(
        with_metadata
    )


def test_run_result_to_input_list_preserves_tool_search_items() -> None:
    agent = Agent(name="A")
    result = RunResult(
        input="Find CRM tools",
        new_items=[
            ToolSearchCallItem(
                agent=agent,
                raw_item={"type": "tool_search_call", "queries": [{"search_term": "profile"}]},
            ),
            ToolSearchOutputItem(
                agent=agent,
                raw_item={"type": "tool_search_output", "results": [{"text": "Customer profile"}]},
            ),
        ],
        raw_responses=[],
        final_output="done",
        input_guardrail_results=[],
        output_guardrail_results=[],
        tool_input_guardrail_results=[],
        tool_output_guardrail_results=[],
        context_wrapper=RunContextWrapper(context=None),
        _last_agent=agent,
    )

    input_items = result.to_input_list()

    assert len(input_items) == 3
    assert cast(dict[str, Any], input_items[1])["type"] == "tool_search_call"
    assert cast(dict[str, Any], input_items[2])["type"] == "tool_search_output"


def test_coerce_tool_search_output_raw_item_rejects_legacy_type() -> None:
    with pytest.raises(AgentsException, match="Unexpected tool search output item type"):
        coerce_tool_search_output_raw_item({"type": "tool_search_result", "results": []})