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

from __future__ import annotations

import asyncio
import json
import sys
from collections.abc import Awaitable, Coroutine
from dataclasses import dataclass
from typing import Annotated, Any, Literal, cast

import pytest
from openai.types.responses import (
    ResponseApplyPatchToolCall,
    ResponseCustomToolCall,
    ResponseFileSearchToolCall,
    ResponseFunctionShellToolCall,
    ResponseFunctionShellToolCallOutput,
    ResponseFunctionToolCall,
    ResponseFunctionWebSearch,
    ResponseToolSearchCall,
    ResponseToolSearchOutputItem,
)
from openai.types.responses.response_apply_patch_tool_call import OperationCreateFile
from openai.types.responses.response_code_interpreter_tool_call import (
    ResponseCodeInterpreterToolCall,
)
from openai.types.responses.response_function_shell_tool_call import Action
from openai.types.responses.response_function_tool_call import CallerProgram
from openai.types.responses.response_function_web_search import ActionSearch
from openai.types.responses.response_output_item import (
    ImageGenerationCall,
    McpApprovalRequest,
    McpCall,
    McpListTools,
    Program,
    ProgramOutput,
)
from pydantic import BaseModel, ConfigDict, Field
from typing_extensions import TypedDict

from agents import (
    Agent,
    ApplyPatchTool,
    CodeInterpreterTool,
    CustomTool,
    HostedMCPTool,
    ModelResponse,
    ModelSettings,
    ProgrammaticToolCallingTool,
    RunConfig,
    RunItem,
    Runner,
    RunState,
    ShellTool,
    ToolCallItem,
    ToolCallOutputItem,
    ToolExecutionConfig,
    ToolGuardrailFunctionOutput,
    ToolInputGuardrailData,
    ToolOutputGuardrailData,
    ToolSearchTool,
    Usage,
    UserError,
    function_tool,
    tool_input_guardrail,
    tool_output_guardrail,
)
from agents.exceptions import ModelBehaviorError
from agents.items import ItemHelpers
from agents.memory import SQLiteSession
from agents.models.chatcmpl_converter import Converter as ChatCompletionsConverter
from agents.models.openai_responses import Converter as ResponsesConverter
from agents.run_internal.turn_resolution import process_model_response
from agents.testing import ScriptedModel
from agents.tool_context import ToolContext

from .model_test_helpers import get_exact_output_stream_step
from .test_responses import get_handoff_tool_call, get_text_message

PROGRAM_CALL_ID = "call_program"
FUNCTION_CALL_ID = "call_lookup"
PROGRAM_CALLER = {"type": "program", "caller_id": PROGRAM_CALL_ID}


class InventoryOutput(BaseModel):
    sku: str
    available_units: int


class InventoryAwaitable(Awaitable[InventoryOutput]):
    def __await__(self) -> Any:
        raise NotImplementedError


class InventoryDict(TypedDict):
    sku: str
    available_units: int


@dataclass
class InventoryData:
    sku: str
    available_units: int


class AliasedInventoryOutput(BaseModel):
    model_config = ConfigDict(populate_by_name=True)

    sku: str
    available_units: int = Field(
        validation_alias="inputUnits",
        serialization_alias="availableUnits",
    )


def _program() -> Program:
    return Program(
        id="program_item",
        call_id=PROGRAM_CALL_ID,
        code='lookup_inventory(sku="A-1")',
        fingerprint="fingerprint",
        type="program",
    )


def _function_call() -> ResponseFunctionToolCall:
    return ResponseFunctionToolCall(
        id="function_item",
        call_id=FUNCTION_CALL_ID,
        name="lookup_inventory",
        arguments='{"sku":"A-1"}',
        caller=CallerProgram(type="program", caller_id=PROGRAM_CALL_ID),
        type="function_call",
    )


def _program_output(
    status: Literal["completed", "incomplete"] = "completed",
) -> ProgramOutput:
    return ProgramOutput(
        id="program_output_item",
        call_id=PROGRAM_CALL_ID,
        result='{"sku":"A-1","available_units":42}',
        status=status,
        type="program_output",
    )


def _hosted_program_call_and_tool(
    output_type: str,
    allowed_callers: list[Any] | None,
) -> tuple[Any, Any]:
    if output_type in ("mcp_approval_request", "mcp_call", "mcp_list_tools"):
        mcp_config: dict[str, Any] = {
            "type": "mcp",
            "server_label": "docs_server",
            "server_url": "https://example.com/mcp",
        }
        if allowed_callers is not None:
            mcp_config["allowed_callers"] = allowed_callers
        hosted_tool = HostedMCPTool(tool_config=cast(Any, mcp_config))
        if output_type == "mcp_list_tools":
            return (
                McpListTools.model_construct(
                    id="mcp_list_tools_1",
                    server_label="docs_server",
                    tools=[],
                    type="mcp_list_tools",
                    caller=PROGRAM_CALLER,
                ),
                hosted_tool,
            )
        call_type: Any = McpApprovalRequest if output_type == "mcp_approval_request" else McpCall
        output = call_type.model_construct(
            id=f"{output_type}_1",
            arguments="{}",
            name="search_docs",
            server_label="docs_server",
            type=output_type,
            caller=PROGRAM_CALLER,
        )
        return output, hosted_tool

    code_interpreter_config: dict[str, Any] = {
        "type": "code_interpreter",
        "container": "auto",
    }
    if allowed_callers is not None:
        code_interpreter_config["allowed_callers"] = allowed_callers
    return (
        ResponseCodeInterpreterToolCall.model_construct(
            id="code_interpreter_1",
            container_id="container_1",
            status="completed",
            type="code_interpreter_call",
            caller=PROGRAM_CALLER,
        ),
        CodeInterpreterTool(tool_config=cast(Any, code_interpreter_config)),
    )


def _caller_dict(value: Any) -> dict[str, str]:
    if isinstance(value, dict):
        return cast(dict[str, str], value)
    return cast(dict[str, str], value.model_dump(exclude_none=True))


def _raw_item_type(value: Any) -> str | None:
    if isinstance(value, dict):
        item_type = value.get("type")
        return item_type if isinstance(item_type, str) else None
    item_type = getattr(value, "type", None)
    return item_type if isinstance(item_type, str) else None


def _function_output_raw_items(result: Any) -> list[dict[str, Any]]:
    return [
        cast(dict[str, Any], item.raw_item)
        for item in result.new_items
        if isinstance(item, ToolCallOutputItem)
        and isinstance(item.raw_item, dict)
        and item.raw_item.get("type") == "function_call_output"
    ]


def test_responses_converter_serializes_programmatic_tool_configuration() -> None:
    @function_tool(allowed_callers=["programmatic"])
    def lookup_inventory(sku: str) -> InventoryOutput:
        return InventoryOutput(sku=sku, available_units=42)

    converted = ResponsesConverter.convert_tools(
        tools=[ProgrammaticToolCallingTool(), lookup_inventory],
        handoffs=[],
    )

    assert converted.tools[0] == {"type": "programmatic_tool_calling"}
    function_payload = cast(dict[str, Any], converted.tools[1])
    assert function_payload["allowed_callers"] == ["programmatic"]
    assert function_payload["output_schema"] == lookup_inventory.output_json_schema
    assert function_payload["output_schema"] == {
        "additionalProperties": False,
        "properties": {
            "sku": {"title": "Sku", "type": "string"},
            "available_units": {"title": "Available Units", "type": "integer"},
        },
        "required": ["sku", "available_units"],
        "title": "InventoryOutput",
        "type": "object",
    }
    assert ResponsesConverter.convert_tool_choice("programmatic_tool_calling") == {
        "type": "programmatic_tool_calling"
    }


def test_function_tool_infers_typed_dict_and_dataclass_output_schemas() -> None:
    @function_tool(allowed_callers=["programmatic"])
    def typed_dict_tool() -> InventoryDict:
        return {"sku": "A-1", "available_units": 42}

    @function_tool(allowed_callers=["programmatic"])
    def dataclass_tool() -> InventoryData:
        return InventoryData(sku="A-1", available_units=42)

    assert typed_dict_tool.output_json_schema is not None
    assert typed_dict_tool.output_json_schema["type"] == "object"
    assert typed_dict_tool.output_json_schema["additionalProperties"] is False
    assert dataclass_tool.output_json_schema is not None
    assert dataclass_tool.output_json_schema["type"] == "object"
    assert dataclass_tool.output_json_schema["additionalProperties"] is False


@pytest.mark.parametrize(
    "return_annotation",
    [
        Awaitable[InventoryOutput],
        Coroutine[Any, Any, InventoryOutput],
        InventoryAwaitable,
        Awaitable[InventoryOutput] | InventoryOutput,
        Awaitable,
        Coroutine,
    ],
)
def test_sync_callable_does_not_infer_through_awaitable_output(
    return_annotation: Any,
) -> None:
    class LookupInventory:
        def __call__(self) -> Any:
            raise AssertionError("The handler must not run during tool construction.")

    cast(Any, LookupInventory.__call__).__annotations__["return"] = return_annotation

    with pytest.raises(UserError, match="programmatic function tool return annotation"):
        function_tool(LookupInventory(), allowed_callers=["programmatic"])


@pytest.mark.skipif(sys.version_info < (3, 12), reason="PEP 695 requires Python 3.12")
def test_sync_callable_does_not_infer_through_pep695_awaitable_alias() -> None:
    namespace = {
        "Any": Any,
        "Awaitable": Awaitable,
        "InventoryOutput": InventoryOutput,
    }
    exec(
        "type OutputAwaitable = Awaitable[InventoryOutput]\n"
        "class LookupInventory:\n"
        "    def __call__(self) -> OutputAwaitable:\n"
        "        raise AssertionError\n",
        namespace,
    )

    with pytest.raises(UserError, match="explicit wrapper function"):
        function_tool(namespace["LookupInventory"](), allowed_callers=["programmatic"])


def test_function_tool_treats_annotated_plain_returns_as_untyped() -> None:
    @function_tool(allowed_callers=["programmatic"])
    def string_tool() -> Annotated[str, "plain string"]:
        return "ok"

    @function_tool(allowed_callers=["programmatic"])
    def any_tool() -> Annotated[Any, "untyped value"]:
        return {"status": "ok"}

    @function_tool(allowed_callers=["programmatic"])
    def none_tool() -> Annotated[None, "no value"]:
        return None

    assert string_tool.output_json_schema is None
    assert string_tool._output_type_adapter is None
    assert any_tool.output_json_schema is None
    assert any_tool._output_type_adapter is None
    assert none_tool.output_json_schema is None
    assert none_tool._output_type_adapter is None


def test_function_tool_preserves_annotated_structured_return_metadata() -> None:
    @function_tool(allowed_callers=["programmatic"])
    def lookup_inventory() -> Annotated[
        InventoryOutput,
        Field(description="Inventory result"),
    ]:
        return InventoryOutput(sku="A-1", available_units=42)

    assert lookup_inventory.output_json_schema is not None
    assert lookup_inventory.output_json_schema["description"] == "Inventory result"
    assert lookup_inventory._output_type_adapter is not None


def test_function_tool_output_type_override_and_raw_schema_are_mutually_exclusive() -> None:
    def unannotated_tool() -> Any:
        return {"sku": "A-1", "available_units": 42}

    tool = function_tool(
        unannotated_tool,
        allowed_callers=["programmatic"],
        output_type=InventoryOutput,
    )

    assert tool.output_json_schema is not None
    assert tool.output_json_schema["title"] == "InventoryOutput"

    with pytest.raises(UserError, match="cannot both be provided"):
        function_tool(
            unannotated_tool,
            allowed_callers=["programmatic"],
            output_type=InventoryOutput,
            output_json_schema={"type": "object"},
        )

    with pytest.raises(UserError, match="output_type must define a strict JSON object"):
        function_tool(
            unannotated_tool,
            allowed_callers=["programmatic"],
            output_type=str,
        )


def test_function_tool_rejects_loose_programmatic_output_annotation() -> None:
    def loose_dict_tool() -> dict[str, Any]:
        return {"sku": "A-1", "available_units": 42}

    with pytest.raises(UserError, match="return annotation must define a strict JSON object"):
        function_tool(loose_dict_tool, allowed_callers=["programmatic"])


def test_function_tool_does_not_infer_non_programmatic_output() -> None:
    @function_tool
    def direct_tool() -> InventoryOutput:
        return InventoryOutput(sku="A-1", available_units=42)

    assert direct_tool.output_json_schema is None


@pytest.mark.parametrize(
    "output_json_schema",
    [
        {"type": "string"},
        {"type": "object", "additionalProperties": True},
    ],
)
def test_function_tool_rejects_non_object_or_non_strict_raw_output_schema(
    output_json_schema: dict[str, Any],
) -> None:
    with pytest.raises(UserError, match="output_json_schema must define a.*object schema"):
        function_tool(
            lambda: "ok",
            allowed_callers=["programmatic"],
            output_json_schema=output_json_schema,
        )


@pytest.mark.asyncio
async def test_function_tool_validates_inferred_output_type() -> None:
    @function_tool(allowed_callers=["programmatic"])
    def invalid_tool() -> InventoryOutput:
        return {"sku": "A-1", "available_units": "many"}  # type: ignore[return-value]

    context = ToolContext(
        None,
        tool_name=invalid_tool.name,
        tool_call_id="invalid",
        tool_arguments="{}",
        tool_call=_function_call(),
    )
    with pytest.raises(UserError, match="does not match its declared output type"):
        await invalid_tool.on_invoke_tool(context, "{}")


@pytest.mark.asyncio
async def test_schema_backed_programmatic_tool_bypasses_default_failure_formatter() -> None:
    @function_tool(allowed_callers=["programmatic"])
    def failing_tool() -> InventoryOutput:
        raise RuntimeError("inventory unavailable")

    context = ToolContext(
        None,
        tool_name=failing_tool.name,
        tool_call_id="failing",
        tool_arguments="{}",
        tool_call=_function_call(),
    )
    with pytest.raises(RuntimeError, match="inventory unavailable"):
        await failing_tool.on_invoke_tool(context, "{}")

    @function_tool(
        allowed_callers=["programmatic"],
        output_json_schema={
            "type": "object",
            "properties": {"error": {"type": "string"}},
            "required": ["error"],
            "additionalProperties": False,
        },
    )
    def failing_declared_schema_tool() -> str:
        raise RuntimeError("declared schema unavailable")

    declared_context = ToolContext(
        None,
        tool_name=failing_declared_schema_tool.name,
        tool_call_id="failing-declared",
        tool_arguments="{}",
        tool_call=_function_call(),
    )
    with pytest.raises(RuntimeError, match="declared schema unavailable"):
        await failing_declared_schema_tool.on_invoke_tool(declared_context, "{}")


@pytest.mark.asyncio
async def test_schema_backed_direct_tool_preserves_argument_error_formatter() -> None:
    @function_tool(allowed_callers=["direct", "programmatic"])
    def failing_tool(sku: str) -> InventoryOutput:
        return InventoryOutput(sku=sku, available_units=42)

    direct_call = ResponseFunctionToolCall(
        id="function_item",
        call_id=FUNCTION_CALL_ID,
        name=failing_tool.name,
        arguments="{}",
        type="function_call",
    )
    context = ToolContext(
        None,
        tool_name=failing_tool.name,
        tool_call_id=FUNCTION_CALL_ID,
        tool_arguments="{}",
        tool_call=direct_call,
    )

    result = await failing_tool.on_invoke_tool(context, "{}")

    assert result.startswith("An error occurred while running the tool. Please try again. Error:")
    assert "Invalid JSON input for tool failing_tool" in result
    assert "sku" not in result


@pytest.mark.asyncio
async def test_runner_preserves_direct_error_for_schema_backed_tool() -> None:
    model = ScriptedModel()
    direct_call = ResponseFunctionToolCall(
        id="function_item",
        call_id=FUNCTION_CALL_ID,
        name="lookup_inventory",
        arguments='{"sku":"A-1"}',
        type="function_call",
    )
    model.extend([[direct_call], [get_text_message("inventory lookup failed")]])

    @function_tool(allowed_callers=["direct", "programmatic"])
    def lookup_inventory(sku: str) -> InventoryOutput:
        raise RuntimeError(f"inventory unavailable for {sku}")

    result = await Runner.run(
        Agent(name="inventory", model=model, tools=[lookup_inventory]),
        "Check inventory",
    )

    function_output = next(
        item for item in result.new_items if isinstance(item, ToolCallOutputItem)
    )
    expected_error = (
        "An error occurred while running the tool. Please try again. "
        "Error: inventory unavailable for A-1"
    )
    assert result.final_output == "inventory lookup failed"
    assert function_output.output == expected_error
    assert cast(dict[str, Any], function_output.raw_item)["output"] == expected_error


@pytest.mark.asyncio
async def test_runner_preserves_direct_default_timeout_for_schema_backed_tool() -> None:
    model = ScriptedModel()
    direct_call = ResponseFunctionToolCall(
        id="function_item",
        call_id=FUNCTION_CALL_ID,
        name="lookup_inventory",
        arguments='{"sku":"A-1"}',
        type="function_call",
    )
    model.extend([[direct_call], [get_text_message("timed out")]])

    @function_tool(allowed_callers=["direct", "programmatic"], timeout=0.01)
    async def lookup_inventory(sku: str) -> InventoryOutput:
        await asyncio.sleep(0.2)
        return InventoryOutput(sku=sku, available_units=42)

    result = await Runner.run(
        Agent(name="inventory", model=model, tools=[lookup_inventory]),
        "Check inventory",
    )

    function_output = next(
        item for item in result.new_items if isinstance(item, ToolCallOutputItem)
    )
    assert result.final_output == "timed out"
    assert isinstance(function_output.output, str)
    assert "timed out" in function_output.output.lower()
    assert cast(dict[str, Any], function_output.raw_item)["output"] == function_output.output


@pytest.mark.asyncio
async def test_schema_backed_function_tool_accepts_conforming_custom_error_output() -> None:
    @function_tool(
        allowed_callers=["programmatic"],
        failure_error_function=lambda _context, _error: json.dumps(
            {"sku": "ERROR", "available_units": 0}
        ),
    )
    def failing_tool() -> InventoryOutput:
        raise RuntimeError("inventory unavailable")

    context = ToolContext(
        None,
        tool_name=failing_tool.name,
        tool_call_id="failing",
        tool_arguments="{}",
    )
    result = await failing_tool.on_invoke_tool(context, "{}")
    output = ItemHelpers.tool_call_output_item(
        _function_call(),
        result,
        output_json_schema=failing_tool.output_json_schema,
        output_type_adapter=failing_tool._output_type_adapter,
    )

    assert json.loads(cast(str, output["output"])) == {
        "sku": "ERROR",
        "available_units": 0,
    }


@pytest.mark.asyncio
async def test_schema_backed_programmatic_tool_accepts_conforming_custom_timeout_output() -> None:
    model = ScriptedModel()
    model.extend(
        [
            [_program(), _function_call()],
            [_program_output(), get_text_message("timeout handled")],
        ]
    )

    @function_tool(
        allowed_callers=["programmatic"],
        timeout=0.01,
        timeout_error_function=lambda _context, _error: json.dumps(
            {"sku": "TIMEOUT", "available_units": 0}
        ),
    )
    async def lookup_inventory(sku: str) -> InventoryOutput:
        await asyncio.sleep(0.2)
        return InventoryOutput(sku=sku, available_units=42)

    result = await Runner.run(
        Agent(
            name="inventory",
            model=model,
            tools=[ProgrammaticToolCallingTool(), lookup_inventory],
        ),
        "Check inventory",
    )

    function_output = next(
        item for item in result.new_items if isinstance(item, ToolCallOutputItem)
    )
    assert result.final_output == "timeout handled"
    assert json.loads(cast(str, cast(dict[str, Any], function_output.raw_item)["output"])) == {
        "sku": "TIMEOUT",
        "available_units": 0,
    }


def test_schema_backed_function_output_rejects_plain_error_text() -> None:
    @function_tool(allowed_callers=["programmatic"])
    def lookup_inventory() -> InventoryOutput:
        return InventoryOutput(sku="A-1", available_units=42)

    with pytest.raises(UserError, match="does not match its declared output schema"):
        ItemHelpers.tool_call_output_item(
            _function_call(),
            "inventory unavailable",
            output_json_schema=lookup_inventory.output_json_schema,
            output_type_adapter=lookup_inventory._output_type_adapter,
        )

    with pytest.raises(UserError, match="requires a JSON object"):
        ItemHelpers.tool_call_output_item(
            _function_call(),
            "inventory unavailable",
            output_json_schema={"type": "object"},
        )


@pytest.mark.asyncio
async def test_function_tool_serializes_typed_output_with_schema_aliases() -> None:
    @function_tool(allowed_callers=["programmatic"])
    def aliased_tool() -> AliasedInventoryOutput:
        return AliasedInventoryOutput(sku="A-1", available_units=42)

    context = ToolContext(
        None,
        tool_name=aliased_tool.name,
        tool_call_id="aliased",
        tool_arguments="{}",
    )
    result = await aliased_tool.on_invoke_tool(context, "{}")
    output = ItemHelpers.tool_call_output_item(
        _function_call(),
        result,
        output_json_schema=aliased_tool.output_json_schema,
        output_type_adapter=aliased_tool._output_type_adapter,
    )

    assert aliased_tool.output_json_schema is not None
    assert "availableUnits" in aliased_tool.output_json_schema["properties"]
    assert json.loads(cast(str, output["output"])) == {
        "sku": "A-1",
        "availableUnits": 42,
    }


def test_responses_converter_serializes_allowed_callers_for_other_eligible_tools() -> None:
    async def shell_executor(_request: Any) -> str:
        return "ok"

    def custom_executor(_context: Any, _input: str) -> str:
        return "ok"

    class Editor:
        def create_file(self, _operation: Any) -> str:
            return "ok"

        def update_file(self, _operation: Any) -> str:
            return "ok"

        def delete_file(self, _operation: Any) -> str:
            return "ok"

    converted = ResponsesConverter.convert_tools(
        tools=[
            ProgrammaticToolCallingTool(),
            ShellTool(executor=shell_executor, allowed_callers=["programmatic"]),
            ApplyPatchTool(editor=Editor(), allowed_callers=["direct", "programmatic"]),
            CustomTool(
                name="custom",
                description="Custom tool",
                on_invoke_tool=custom_executor,
                allowed_callers=["programmatic"],
            ),
        ],
        handoffs=[],
    )

    tool_payloads = [cast(dict[str, Any], tool) for tool in converted.tools]
    assert tool_payloads[1]["allowed_callers"] == ["programmatic"]
    assert tool_payloads[2]["allowed_callers"] == ["direct", "programmatic"]
    assert tool_payloads[3]["allowed_callers"] == ["programmatic"]


@pytest.mark.parametrize(
    "allowed_callers",
    [
        [],
        ["direct", "direct"],
        ["unsupported"],
    ],
)
def test_tool_construction_rejects_invalid_allowed_callers(
    allowed_callers: list[Any],
) -> None:
    with pytest.raises(UserError, match="allowed_callers"):
        function_tool(lambda: "ok", allowed_callers=allowed_callers)

    with pytest.raises(UserError, match="allowed_callers"):
        ShellTool(executor=lambda _request: "ok", allowed_callers=allowed_callers)

    with pytest.raises(UserError, match="allowed_callers"):
        HostedMCPTool(
            tool_config=cast(
                Any,
                {
                    "type": "mcp",
                    "server_label": "inventory",
                    "server_url": "https://example.com/mcp",
                    "allowed_callers": allowed_callers,
                },
            )
        )

    with pytest.raises(UserError, match="allowed_callers"):
        CodeInterpreterTool(
            tool_config=cast(
                Any,
                {
                    "type": "code_interpreter",
                    "container": "auto",
                    "allowed_callers": allowed_callers,
                },
            )
        )


def test_responses_converter_rejects_incomplete_programmatic_configuration() -> None:
    @function_tool(allowed_callers=["programmatic"])
    def programmatic_only() -> str:
        return "ok"

    with pytest.raises(UserError, match="requires ProgrammaticToolCallingTool"):
        ResponsesConverter.convert_tools(tools=[programmatic_only], handoffs=[])

    with pytest.raises(UserError, match="requires ProgrammaticToolCallingTool"):
        ResponsesConverter.convert_tools(
            tools=[],
            handoffs=[],
            tool_choice="programmatic_tool_calling",
        )

    with pytest.raises(UserError, match="requires at least one tool"):
        ResponsesConverter.convert_tools(
            tools=[ProgrammaticToolCallingTool()],
            handoffs=[],
        )

    with pytest.raises(UserError, match="Only one ProgrammaticToolCallingTool"):
        ResponsesConverter.convert_tools(
            tools=[ProgrammaticToolCallingTool(), ProgrammaticToolCallingTool()],
            handoffs=[],
        )


def test_responses_converter_accepts_mixed_or_tool_search_managed_configuration() -> None:
    @function_tool(allowed_callers=["direct", "programmatic"])
    def mixed_callers() -> str:
        return "ok"

    converted_mixed = ResponsesConverter.convert_tools(tools=[mixed_callers], handoffs=[])
    assert cast(dict[str, Any], converted_mixed.tools[0])["allowed_callers"] == [
        "direct",
        "programmatic",
    ]

    converted_search = ResponsesConverter.convert_tools(
        tools=[ProgrammaticToolCallingTool(), ToolSearchTool()],
        handoffs=[],
        allow_opaque_tool_search_surface=True,
    )
    assert converted_search.tools == [
        {"type": "programmatic_tool_calling"},
        {"type": "tool_search"},
    ]


def test_chat_completions_rejects_programmatic_tool_configuration() -> None:
    @function_tool(allowed_callers=["programmatic"])
    def lookup_inventory() -> InventoryOutput:
        return InventoryOutput(sku="A-1", available_units=42)

    with pytest.raises(UserError, match="only supported with OpenAI Responses models"):
        ChatCompletionsConverter.tool_to_openai(lookup_inventory)

    with pytest.raises(UserError, match="programmatic_tool_calling"):
        ChatCompletionsConverter.convert_tool_choice("programmatic_tool_calling")

    with pytest.raises(UserError, match="Hosted tools are not supported"):
        ChatCompletionsConverter.tool_to_openai(ProgrammaticToolCallingTool())


def test_function_output_preserves_caller_and_uses_declared_json_schema() -> None:
    output = ItemHelpers.tool_call_output_item(
        _function_call(),
        {"sku": "A-1", "available_units": 42},
        output_json_schema={"type": "object"},
    )

    assert json.loads(cast(str, output["output"])) == {
        "sku": "A-1",
        "available_units": 42,
    }
    assert _caller_dict(output["caller"]) == PROGRAM_CALLER

    programmatic_output = ItemHelpers.tool_call_output_item(
        _function_call(),
        {"sku": "A-1", "units": [1, 2]},
    )
    assert json.loads(cast(str, programmatic_output["output"])) == {
        "sku": "A-1",
        "units": [1, 2],
    }
    assert _caller_dict(programmatic_output["caller"]) == PROGRAM_CALLER

    direct_call = ResponseFunctionToolCall(
        id="direct_function_item",
        call_id="direct_call",
        name="lookup_inventory",
        arguments="{}",
        type="function_call",
    )
    legacy_output = ItemHelpers.tool_call_output_item(direct_call, {"sku": "A-1"})
    assert legacy_output["output"] == "{'sku': 'A-1'}"


def test_process_model_response_keeps_program_items_in_order() -> None:
    @function_tool(allowed_callers=["programmatic"])
    def lookup_inventory(sku: str) -> str:
        return sku

    agent = Agent(
        name="inventory",
        tools=[ProgrammaticToolCallingTool(), lookup_inventory],
    )
    response = ModelResponse(
        output=[_program(), _function_call(), _program_output("incomplete")],
        usage=Usage(),
        response_id="response_1",
    )

    processed = process_model_response(
        agent=agent,
        all_tools=agent.tools,
        response=response,
        output_schema=None,
        handoffs=[],
    )

    assert [type(item) for item in processed.new_items] == [
        ToolCallItem,
        ToolCallItem,
        ToolCallOutputItem,
    ]
    assert [_raw_item_type(item.raw_item) for item in processed.new_items] == [
        "program",
        "function_call",
        "program_output",
    ]
    assert processed.tools_used == [
        "programmatic_tool_calling",
        "lookup_inventory",
        "programmatic_tool_calling",
    ]


@pytest.mark.parametrize("call_id", [None, ""])
def test_process_model_response_rejects_program_without_valid_call_id(
    call_id: str | None,
) -> None:
    program: dict[str, Any] = {
        "type": "program",
        "id": "program_item",
        "code": "return 42",
        "fingerprint": "fingerprint",
    }
    if call_id is not None:
        program["call_id"] = call_id

    response = ModelResponse(output=[], usage=Usage(), response_id="response_1")
    response.output = cast(Any, [program])
    agent = Agent(name="inventory", tools=[ProgrammaticToolCallingTool()])

    with pytest.raises(ModelBehaviorError, match="without a valid call_id"):
        process_model_response(
            agent=agent,
            all_tools=agent.tools,
            response=response,
            output_schema=None,
            handoffs=[],
        )


@pytest.mark.parametrize(
    "program_output",
    [_program_output(), _program_output().model_dump(exclude_none=True)],
)
def test_process_model_response_rejects_orphan_program_output(program_output: Any) -> None:
    agent = Agent(name="inventory", tools=[ProgrammaticToolCallingTool()])

    with pytest.raises(ModelBehaviorError, match="does not match a parent program item"):
        process_model_response(
            agent=agent,
            all_tools=agent.tools,
            response=ModelResponse(
                output=[program_output],
                usage=Usage(),
                response_id="response_1",
            ),
            output_schema=None,
            handoffs=[],
        )


def test_process_model_response_accepts_program_output_for_retained_program() -> None:
    agent = Agent(name="inventory", tools=[ProgrammaticToolCallingTool()])
    existing_program = ToolCallItem(raw_item=_program(), agent=agent)

    processed = process_model_response(
        agent=agent,
        all_tools=agent.tools,
        response=ModelResponse(
            output=[_program_output()],
            usage=Usage(),
            response_id="response_1",
        ),
        output_schema=None,
        handoffs=[],
        existing_items=[existing_program],
    )

    assert len(processed.new_items) == 1
    assert isinstance(processed.new_items[0], ToolCallOutputItem)


@pytest.mark.parametrize(
    ("field", "value", "remove_field", "error_match"),
    [
        ("status", None, True, "without a valid status"),
        ("status", "running", False, "without a valid status"),
        ("result", None, True, "without a string result"),
        ("result", 42, False, "without a string result"),
    ],
)
def test_process_model_response_rejects_malformed_program_output(
    field: str,
    value: Any,
    remove_field: bool,
    error_match: str,
) -> None:
    agent = Agent(name="inventory", tools=[ProgrammaticToolCallingTool()])
    program_output = _program_output().model_dump(exclude_none=True)
    if remove_field:
        program_output.pop(field)
    else:
        program_output[field] = value
    response = ModelResponse(output=[], usage=Usage(), response_id="response_1")
    response.output = cast(Any, [_program(), program_output])

    with pytest.raises(ModelBehaviorError, match=error_match):
        process_model_response(
            agent=agent,
            all_tools=agent.tools,
            response=response,
            output_schema=None,
            handoffs=[],
        )


@pytest.mark.parametrize("parent_location", ["existing_items", "current_response"])
@pytest.mark.parametrize("as_mapping", [False, True])
def test_process_model_response_rejects_duplicate_completed_program_output(
    parent_location: str,
    as_mapping: bool,
) -> None:
    agent = Agent(name="inventory", tools=[ProgrammaticToolCallingTool()])
    program: Any = _program()
    completed_output: Any = _program_output()
    duplicate_output: Any = _program_output().model_copy(
        update={"id": "duplicate_program_output_item"}
    )
    if as_mapping:
        program = program.model_dump(exclude_none=True)
        completed_output = completed_output.model_dump(exclude_none=True)
        duplicate_output = duplicate_output.model_dump(exclude_none=True)

    response_output = [duplicate_output]
    existing_items: list[RunItem] = []
    if parent_location == "existing_items":
        existing_items = [
            ToolCallItem(raw_item=program, agent=agent),
            ToolCallOutputItem(
                raw_item=completed_output,
                output=(
                    completed_output["result"]
                    if isinstance(completed_output, dict)
                    else completed_output.result
                ),
                agent=agent,
            ),
        ]
    else:
        response_output = [program, completed_output, duplicate_output]

    with pytest.raises(ModelBehaviorError, match="parent program is already completed"):
        process_model_response(
            agent=agent,
            all_tools=agent.tools,
            response=ModelResponse(
                output=response_output,
                usage=Usage(),
                response_id="response_1",
            ),
            output_schema=None,
            handoffs=[],
            existing_items=existing_items,
        )


@pytest.mark.parametrize("as_mapping", [False, True])
def test_process_model_response_rejects_program_owned_call_for_completed_retained_program(
    as_mapping: bool,
) -> None:
    @function_tool(allowed_callers=["programmatic"])
    def lookup_inventory(sku: str) -> str:
        return sku

    agent = Agent(
        name="inventory",
        tools=[ProgrammaticToolCallingTool(), lookup_inventory],
    )
    program: Any = _program()
    program_output: Any = _program_output()
    tool_call: Any = _function_call()
    if as_mapping:
        program = program.model_dump(exclude_none=True)
        program_output = program_output.model_dump(exclude_none=True)
        tool_call = tool_call.model_dump(exclude_none=True)

    existing_items: list[RunItem] = [
        ToolCallItem(raw_item=program, agent=agent),
        ToolCallOutputItem(
            raw_item=program_output,
            output=getattr(program_output, "result", None)
            or cast(dict[str, Any], program_output)["result"],
            agent=agent,
        ),
    ]

    with pytest.raises(ModelBehaviorError, match="parent program is already completed"):
        process_model_response(
            agent=agent,
            all_tools=agent.tools,
            response=ModelResponse(output=[tool_call], usage=Usage(), response_id="response_1"),
            output_schema=None,
            handoffs=[],
            existing_items=existing_items,
        )


@pytest.mark.parametrize("as_mapping", [False, True])
@pytest.mark.parametrize("call_before_completed_output", [False, True])
def test_process_model_response_rejects_program_owned_call_when_response_completes_program(
    as_mapping: bool,
    call_before_completed_output: bool,
) -> None:
    @function_tool(allowed_callers=["programmatic"])
    def lookup_inventory(sku: str) -> str:
        return sku

    agent = Agent(
        name="inventory",
        tools=[ProgrammaticToolCallingTool(), lookup_inventory],
    )
    child_items = (
        [_function_call(), _program_output()]
        if call_before_completed_output
        else [_program_output(), _function_call()]
    )
    output: list[Any] = [_program(), *child_items]
    if as_mapping:
        output = [item.model_dump(exclude_none=True) for item in output]

    with pytest.raises(ModelBehaviorError, match="parent program is already completed"):
        process_model_response(
            agent=agent,
            all_tools=agent.tools,
            response=ModelResponse(output=output, usage=Usage(), response_id="response_1"),
            output_schema=None,
            handoffs=[],
        )


@pytest.mark.parametrize("as_mapping", [False, True])
def test_process_model_response_accepts_program_owned_call_for_incomplete_retained_program(
    as_mapping: bool,
) -> None:
    @function_tool(allowed_callers=["programmatic"])
    def lookup_inventory(sku: str) -> str:
        return sku

    agent = Agent(
        name="inventory",
        tools=[ProgrammaticToolCallingTool(), lookup_inventory],
    )
    program: Any = _program()
    program_output: Any = _program_output("incomplete")
    tool_call: Any = _function_call()
    if as_mapping:
        program = program.model_dump(exclude_none=True)
        program_output = program_output.model_dump(exclude_none=True)
        tool_call = tool_call.model_dump(exclude_none=True)

    existing_items: list[RunItem] = [
        ToolCallItem(raw_item=program, agent=agent),
        ToolCallOutputItem(
            raw_item=program_output,
            output=getattr(program_output, "result", None)
            or cast(dict[str, Any], program_output)["result"],
            agent=agent,
        ),
    ]
    processed = process_model_response(
        agent=agent,
        all_tools=agent.tools,
        response=ModelResponse(output=[tool_call], usage=Usage(), response_id="response_1"),
        output_schema=None,
        handoffs=[],
        existing_items=existing_items,
    )

    assert len(processed.new_items) == 1
    assert _raw_item_type(processed.new_items[0].raw_item) == "function_call"


@pytest.mark.parametrize(
    "output",
    [
        _program(),
        _program().model_dump(exclude_none=True),
        _program_output(),
        _program_output().model_dump(exclude_none=True),
    ],
)
def test_process_model_response_rejects_program_items_without_programmatic_tool(
    output: Any,
) -> None:
    agent = Agent(name="inventory")

    with pytest.raises(ModelBehaviorError, match="programmatic_tool_calling tool"):
        process_model_response(
            agent=agent,
            all_tools=agent.tools,
            response=ModelResponse(output=[output], usage=Usage(), response_id="response_1"),
            output_schema=None,
            handoffs=[],
        )


@pytest.mark.asyncio
async def test_runner_rejects_program_item_without_programmatic_tool() -> None:
    model = ScriptedModel()
    model.enqueue([_program()])
    agent = Agent(name="inventory", model=model)

    with pytest.raises(ModelBehaviorError, match="programmatic_tool_calling tool"):
        await Runner.run(agent, "Check inventory")


def test_process_model_response_rejects_program_owned_calls_without_programmatic_tool() -> None:
    @function_tool(allowed_callers=["programmatic"])
    def lookup_inventory(sku: str) -> str:
        return sku

    async def shell_executor(_request: Any) -> str:
        return "ok"

    shell_tool = ShellTool(executor=shell_executor, allowed_callers=["programmatic"])
    shell_call = ResponseFunctionShellToolCall(
        id="shell_item",
        call_id="call_shell",
        action=Action(commands=["echo ok"]),
        status="completed",
        type="shell_call",
        caller=cast(Any, PROGRAM_CALLER),
    )
    custom_tool = CustomTool(
        name="custom",
        description="Custom tool",
        on_invoke_tool=lambda _context, _input: "ok",
        allowed_callers=["programmatic"],
    )
    custom_call = ResponseCustomToolCall(
        id="custom_item",
        call_id="call_custom",
        input="input",
        name="custom",
        type="custom_tool_call",
        caller=cast(Any, PROGRAM_CALLER),
    )

    cases: list[tuple[Any, Any]] = [
        (lookup_inventory, _function_call()),
        (lookup_inventory, _function_call().model_dump(exclude_none=True)),
        (shell_tool, shell_call),
        (shell_tool, shell_call.model_dump(exclude_none=True)),
        (custom_tool, custom_call),
        (custom_tool, custom_call.model_dump(exclude_none=True)),
    ]
    for tool, tool_call in cases:
        agent = Agent(name="tool agent", tools=[tool])
        with pytest.raises(ModelBehaviorError, match="programmatic_tool_calling tool"):
            process_model_response(
                agent=agent,
                all_tools=agent.tools,
                response=ModelResponse(output=[tool_call], usage=Usage(), response_id="response_1"),
                output_schema=None,
                handoffs=[],
            )


@pytest.mark.asyncio
async def test_runner_does_not_execute_program_owned_call_without_programmatic_tool() -> None:
    model = ScriptedModel()
    model.enqueue([_function_call()])
    executed = False

    @function_tool(allowed_callers=["programmatic"])
    def lookup_inventory(sku: str) -> str:
        nonlocal executed
        executed = True
        return sku

    agent = Agent(name="inventory", model=model, tools=[lookup_inventory])

    with pytest.raises(ModelBehaviorError, match="programmatic_tool_calling tool"):
        await Runner.run(agent, "Check inventory")

    assert executed is False


def test_process_model_response_rejects_program_owned_calls_without_parent_program() -> None:
    @function_tool(allowed_callers=["programmatic"])
    def lookup_inventory(sku: str) -> str:
        return sku

    async def shell_executor(_request: Any) -> str:
        return "ok"

    shell_tool = ShellTool(executor=shell_executor, allowed_callers=["programmatic"])
    shell_call = ResponseFunctionShellToolCall(
        id="shell_item",
        call_id="call_shell",
        action=Action(commands=["echo ok"]),
        status="completed",
        type="shell_call",
        caller=cast(Any, PROGRAM_CALLER),
    )
    custom_tool = CustomTool(
        name="custom",
        description="Custom tool",
        on_invoke_tool=lambda _context, _input: "ok",
        allowed_callers=["programmatic"],
    )
    custom_call = ResponseCustomToolCall(
        id="custom_item",
        call_id="call_custom",
        input="input",
        name="custom",
        type="custom_tool_call",
        caller=cast(Any, PROGRAM_CALLER),
    )

    cases: list[tuple[Any, Any]] = [
        (lookup_inventory, _function_call()),
        (lookup_inventory, _function_call().model_dump(exclude_none=True)),
        (shell_tool, shell_call),
        (shell_tool, shell_call.model_dump(exclude_none=True)),
        (custom_tool, custom_call),
        (custom_tool, custom_call.model_dump(exclude_none=True)),
    ]
    for tool, tool_call in cases:
        agent = Agent(name="tool agent", tools=[ProgrammaticToolCallingTool(), tool])
        with pytest.raises(ModelBehaviorError, match="parent program item"):
            process_model_response(
                agent=agent,
                all_tools=agent.tools,
                response=ModelResponse(output=[tool_call], usage=Usage(), response_id="response_1"),
                output_schema=None,
                handoffs=[],
            )


@pytest.mark.parametrize("parent_location", ["current_response", "existing_items"])
@pytest.mark.parametrize("as_mapping", [False, True])
def test_process_model_response_accepts_program_owned_call_with_parent_program(
    parent_location: str,
    as_mapping: bool,
) -> None:
    @function_tool(allowed_callers=["programmatic"])
    def lookup_inventory(sku: str) -> str:
        return sku

    agent = Agent(
        name="inventory",
        tools=[ProgrammaticToolCallingTool(), lookup_inventory],
    )
    tool_call: Any = _function_call()
    parent_program: Any = _program()
    if as_mapping:
        tool_call = tool_call.model_dump(exclude_none=True)
        parent_program = parent_program.model_dump(exclude_none=True)

    response_output: list[Any] = [tool_call]
    existing_items: list[ToolCallItem] = []
    if parent_location == "current_response":
        response_output.insert(0, parent_program)
    else:
        existing_items.append(ToolCallItem(raw_item=parent_program, agent=agent))

    processed = process_model_response(
        agent=agent,
        all_tools=agent.tools,
        response=ModelResponse(output=response_output, usage=Usage(), response_id="response_1"),
        output_schema=None,
        handoffs=[],
        existing_items=existing_items,
    )

    assert "function_call" in {_raw_item_type(item.raw_item) for item in processed.new_items}


@pytest.mark.parametrize("child_type", ["function_call", "program_output"])
@pytest.mark.parametrize("as_mapping", [False, True])
def test_process_model_response_rejects_program_child_before_parent(
    child_type: str,
    as_mapping: bool,
) -> None:
    @function_tool(allowed_callers=["programmatic"])
    def lookup_inventory(sku: str) -> str:
        return sku

    agent = Agent(
        name="inventory",
        tools=[ProgrammaticToolCallingTool(), lookup_inventory],
    )
    child: Any = _function_call() if child_type == "function_call" else _program_output()
    parent: Any = _program()
    if as_mapping:
        child = child.model_dump(exclude_none=True)
        parent = parent.model_dump(exclude_none=True)

    with pytest.raises(ModelBehaviorError, match="parent program item"):
        process_model_response(
            agent=agent,
            all_tools=agent.tools,
            response=ModelResponse(
                output=[child, parent],
                usage=Usage(),
                response_id="response_1",
            ),
            output_schema=None,
            handoffs=[],
        )


@pytest.mark.parametrize("child_type", ["function_call", "program_output"])
@pytest.mark.parametrize("as_mapping", [False, True])
def test_process_model_response_accepts_server_owned_parent_from_submitted_delta(
    child_type: str,
    as_mapping: bool,
) -> None:
    @function_tool(allowed_callers=["programmatic"])
    def lookup_inventory(sku: str) -> str:
        return sku

    agent = Agent(
        name="inventory",
        tools=[ProgrammaticToolCallingTool(), lookup_inventory],
    )
    child: Any = _function_call() if child_type == "function_call" else _program_output()
    if as_mapping:
        child = child.model_dump(exclude_none=True)
    submitted_delta = [
        {
            "type": "function_call_output",
            "call_id": FUNCTION_CALL_ID,
            "output": '{"sku":"A-1","available_units":42}',
            "caller": PROGRAM_CALLER,
        }
    ]

    processed = process_model_response(
        agent=agent,
        all_tools=agent.tools,
        response=ModelResponse(output=[child], usage=Usage(), response_id="response_1"),
        output_schema=None,
        handoffs=[],
        server_manages_conversation=True,
        server_managed_input_items=submitted_delta,
    )

    assert child_type in {_raw_item_type(item.raw_item) for item in processed.new_items}


def test_process_model_response_rejects_server_owned_child_without_parent_evidence() -> None:
    @function_tool(allowed_callers=["programmatic"])
    def lookup_inventory(sku: str) -> str:
        return sku

    agent = Agent(
        name="inventory",
        tools=[ProgrammaticToolCallingTool(), lookup_inventory],
    )

    with pytest.raises(ModelBehaviorError, match="parent program item"):
        process_model_response(
            agent=agent,
            all_tools=agent.tools,
            response=ModelResponse(
                output=[_function_call()],
                usage=Usage(),
                response_id="response_1",
            ),
            output_schema=None,
            handoffs=[],
            server_manages_conversation=True,
        )


@pytest.mark.parametrize("as_mapping", [False, True])
def test_process_model_response_accepts_server_owned_parent_after_incomplete_output(
    as_mapping: bool,
) -> None:
    @function_tool(allowed_callers=["programmatic"])
    def lookup_inventory(sku: str) -> str:
        return sku

    agent = Agent(
        name="inventory",
        tools=[ProgrammaticToolCallingTool(), lookup_inventory],
    )
    prior_output: Any = _program_output("incomplete")
    if as_mapping:
        prior_output = prior_output.model_dump(exclude_none=True)

    processed = process_model_response(
        agent=agent,
        all_tools=agent.tools,
        response=ModelResponse(
            output=[_function_call()],
            usage=Usage(),
            response_id="response_2",
        ),
        output_schema=None,
        handoffs=[],
        existing_items=[
            ToolCallOutputItem(
                raw_item=prior_output,
                output='{"status":"waiting"}',
                agent=agent,
            )
        ],
        server_manages_conversation=True,
    )

    assert _raw_item_type(processed.new_items[0].raw_item) == "function_call"


@pytest.mark.asyncio
async def test_runner_does_not_execute_program_owned_call_without_parent_program() -> None:
    model = ScriptedModel()
    model.enqueue([_function_call()])
    executed = False

    @function_tool(allowed_callers=["programmatic"])
    def lookup_inventory(sku: str) -> str:
        nonlocal executed
        executed = True
        return sku

    agent = Agent(
        name="inventory",
        model=model,
        tools=[ProgrammaticToolCallingTool(), lookup_inventory],
    )

    with pytest.raises(ModelBehaviorError, match="parent program item"):
        await Runner.run(agent, "Check inventory")

    assert executed is False


@pytest.mark.parametrize(
    ("allowed_callers", "caller", "expected_caller"),
    [
        (None, CallerProgram(type="program", caller_id=PROGRAM_CALL_ID), "programmatic"),
        (["direct"], CallerProgram(type="program", caller_id=PROGRAM_CALL_ID), "programmatic"),
        (["programmatic"], None, "direct"),
    ],
)
def test_process_model_response_rejects_disallowed_function_callers(
    allowed_callers: list[Any] | None,
    caller: CallerProgram | None,
    expected_caller: str,
) -> None:
    @function_tool(allowed_callers=allowed_callers)
    def lookup_inventory(sku: str) -> str:
        return sku

    agent = Agent(
        name="inventory",
        tools=[ProgrammaticToolCallingTool(), lookup_inventory],
    )
    tool_call = ResponseFunctionToolCall(
        id="function_item",
        call_id=FUNCTION_CALL_ID,
        name="lookup_inventory",
        arguments='{"sku":"A-1"}',
        caller=caller,
        type="function_call",
    )
    response_output: list[Any] = [_program(), tool_call] if caller is not None else [tool_call]

    with pytest.raises(ModelBehaviorError, match=f"caller {expected_caller}"):
        process_model_response(
            agent=agent,
            all_tools=agent.tools,
            response=ModelResponse(output=response_output, usage=Usage(), response_id="response_1"),
            output_schema=None,
            handoffs=[],
        )


def test_process_model_response_rejects_unknown_function_caller_type() -> None:
    @function_tool
    def lookup_inventory(sku: str) -> str:
        return sku

    agent = Agent(name="inventory", tools=[lookup_inventory])
    tool_call = ResponseFunctionToolCall.model_construct(
        id="function_item",
        call_id=FUNCTION_CALL_ID,
        name="lookup_inventory",
        arguments='{"sku":"A-1"}',
        caller={"type": "unknown", "caller_id": PROGRAM_CALL_ID},
        type="function_call",
    )

    with pytest.raises(ModelBehaviorError, match="unsupported caller type 'unknown'"):
        process_model_response(
            agent=agent,
            all_tools=agent.tools,
            response=ModelResponse(output=[tool_call], usage=Usage(), response_id="response_1"),
            output_schema=None,
            handoffs=[],
        )


def test_process_model_response_rejects_disallowed_non_function_callers() -> None:
    caller = cast(Any, PROGRAM_CALLER)

    async def shell_executor(_request: Any) -> str:
        return "ok"

    shell_tool = ShellTool(executor=shell_executor)
    programmatic_tool = ProgrammaticToolCallingTool()
    shell_call = ResponseFunctionShellToolCall(
        id="shell_item",
        call_id="call_shell",
        action=Action(commands=["echo ok"]),
        status="completed",
        type="shell_call",
        caller=caller,
    )
    with pytest.raises(ModelBehaviorError, match="caller programmatic"):
        process_model_response(
            agent=Agent(name="shell", tools=[programmatic_tool, shell_tool]),
            all_tools=[programmatic_tool, shell_tool],
            response=ModelResponse(
                output=[_program(), shell_call], usage=Usage(), response_id="response_1"
            ),
            output_schema=None,
            handoffs=[],
        )

    custom_tool = CustomTool(
        name="custom",
        description="Custom tool",
        on_invoke_tool=lambda _context, _input: "ok",
    )
    custom_call = ResponseCustomToolCall(
        id="custom_item",
        call_id="call_custom",
        input="input",
        name="custom",
        type="custom_tool_call",
        caller=caller,
    )
    with pytest.raises(ModelBehaviorError, match="caller programmatic"):
        process_model_response(
            agent=Agent(name="custom", tools=[programmatic_tool, custom_tool]),
            all_tools=[programmatic_tool, custom_tool],
            response=ModelResponse(
                output=[_program(), custom_call], usage=Usage(), response_id="response_1"
            ),
            output_schema=None,
            handoffs=[],
        )

    class Editor:
        def create_file(self, _operation: Any) -> str:
            return "ok"

        def update_file(self, _operation: Any) -> str:
            return "ok"

        def delete_file(self, _operation: Any) -> str:
            return "ok"

    apply_patch_tool = ApplyPatchTool(editor=Editor(), allowed_callers=["programmatic"])
    apply_patch_call = ResponseApplyPatchToolCall(
        id="apply_patch_item",
        call_id="call_apply_patch",
        operation=OperationCreateFile(type="create_file", path="example.txt", diff="hello"),
        status="completed",
        type="apply_patch_call",
    )
    with pytest.raises(ModelBehaviorError, match="caller direct"):
        process_model_response(
            agent=Agent(name="apply patch", tools=[programmatic_tool, apply_patch_tool]),
            all_tools=[programmatic_tool, apply_patch_tool],
            response=ModelResponse(
                output=[apply_patch_call], usage=Usage(), response_id="response_1"
            ),
            output_schema=None,
            handoffs=[],
        )


@pytest.mark.parametrize("as_mapping", [False, True])
@pytest.mark.parametrize(
    "tool_call",
    [
        ResponseFileSearchToolCall.model_construct(
            id="file_search_1",
            queries=["inventory"],
            status="completed",
            type="file_search_call",
            caller=PROGRAM_CALLER,
        ),
        ResponseFunctionWebSearch.model_construct(
            id="web_search_1",
            action=ActionSearch(type="search", query="inventory"),
            status="completed",
            type="web_search_call",
            caller=PROGRAM_CALLER,
        ),
        ImageGenerationCall.model_construct(
            id="image_generation_1",
            status="completed",
            type="image_generation_call",
            caller=PROGRAM_CALLER,
        ),
    ],
)
def test_process_model_response_rejects_program_owned_direct_only_hosted_calls(
    tool_call: Any,
    as_mapping: bool,
) -> None:
    output = tool_call.model_dump(exclude_none=True) if as_mapping else tool_call
    agent = Agent(name="hosted", tools=[ProgrammaticToolCallingTool()])

    with pytest.raises(ModelBehaviorError, match="caller programmatic"):
        process_model_response(
            agent=agent,
            all_tools=agent.tools,
            response=ModelResponse(
                output=[_program(), output],
                usage=Usage(),
                response_id="response_1",
            ),
            output_schema=None,
            handoffs=[],
        )


@pytest.mark.parametrize(
    "tool_search_item",
    [
        ResponseToolSearchCall.model_construct(
            id="tool_search_call_item",
            arguments={},
            execution="server",
            status="completed",
            type="tool_search_call",
            caller=PROGRAM_CALLER,
        ),
        ResponseToolSearchOutputItem.model_construct(
            id="tool_search_output_item",
            execution="server",
            status="completed",
            tools=[],
            type="tool_search_output",
            caller=PROGRAM_CALLER,
        ),
    ],
)
def test_process_model_response_rejects_program_owned_tool_search_items(
    tool_search_item: Any,
) -> None:
    agent = Agent(
        name="tool search",
        tools=[ProgrammaticToolCallingTool(), ToolSearchTool()],
    )
    with pytest.raises(ModelBehaviorError, match="caller programmatic"):
        process_model_response(
            agent=agent,
            all_tools=agent.tools,
            response=ModelResponse(
                output=[_program(), tool_search_item],
                usage=Usage(),
                response_id="response_1",
            ),
            output_schema=None,
            handoffs=[],
        )


@pytest.mark.parametrize(
    ("output_type", "allowed_callers"),
    [
        ("mcp_approval_request", None),
        ("mcp_approval_request", ["direct"]),
        ("mcp_call", None),
        ("mcp_call", ["direct"]),
        ("mcp_list_tools", None),
        ("mcp_list_tools", ["direct"]),
        ("code_interpreter_call", None),
        ("code_interpreter_call", ["direct"]),
    ],
)
def test_process_model_response_rejects_disallowed_hosted_program_callers(
    output_type: str,
    allowed_callers: list[Any] | None,
) -> None:
    output, hosted_tool = _hosted_program_call_and_tool(output_type, allowed_callers)
    programmatic_tool = ProgrammaticToolCallingTool()
    agent = Agent(name="hosted", tools=[programmatic_tool, hosted_tool])
    with pytest.raises(ModelBehaviorError, match="caller programmatic"):
        process_model_response(
            agent=agent,
            all_tools=agent.tools,
            response=ModelResponse(
                output=[_program(), output],
                usage=Usage(),
                response_id="response_1",
            ),
            output_schema=None,
            handoffs=[],
        )


@pytest.mark.parametrize(
    "output_type",
    ["mcp_approval_request", "mcp_call", "mcp_list_tools", "code_interpreter_call"],
)
def test_process_model_response_accepts_allowed_hosted_program_callers(
    output_type: str,
) -> None:
    output, hosted_tool = _hosted_program_call_and_tool(output_type, ["programmatic"])
    programmatic_tool = ProgrammaticToolCallingTool()
    agent = Agent(name="hosted", tools=[programmatic_tool, hosted_tool])
    processed = process_model_response(
        agent=agent,
        all_tools=agent.tools,
        response=ModelResponse(
            output=[_program(), output],
            usage=Usage(),
            response_id="response_1",
        ),
        output_schema=None,
        handoffs=[],
    )

    assert len(processed.new_items) == 2


@pytest.mark.parametrize("as_mapping", [False, True])
@pytest.mark.parametrize("allowed_callers", [None, ["direct"]])
def test_process_model_response_rejects_disallowed_program_owned_shell_output(
    as_mapping: bool,
    allowed_callers: list[Any] | None,
) -> None:
    shell_output: Any = ResponseFunctionShellToolCallOutput.model_construct(
        id="shell_output_1",
        call_id="shell_call_1",
        status="completed",
        type="shell_call_output",
        output=[],
        caller=PROGRAM_CALLER,
    )
    if as_mapping:
        shell_output = shell_output.model_dump(exclude_none=True)
    shell_tool = ShellTool(executor=lambda _request: "ok", allowed_callers=allowed_callers)
    agent = Agent(
        name="shell",
        tools=[ProgrammaticToolCallingTool(), shell_tool],
    )

    with pytest.raises(ModelBehaviorError, match="caller programmatic"):
        process_model_response(
            agent=agent,
            all_tools=agent.tools,
            response=ModelResponse(
                output=[_program(), shell_output],
                usage=Usage(),
                response_id="response_1",
            ),
            output_schema=None,
            handoffs=[],
        )


@pytest.mark.parametrize("as_mapping", [False, True])
def test_process_model_response_accepts_allowed_program_owned_shell_output(
    as_mapping: bool,
) -> None:
    shell_output: Any = ResponseFunctionShellToolCallOutput.model_construct(
        id="shell_output_1",
        call_id="shell_call_1",
        status="completed",
        type="shell_call_output",
        output=[],
        caller=PROGRAM_CALLER,
    )
    if as_mapping:
        shell_output = shell_output.model_dump(exclude_none=True)
    shell_tool = ShellTool(executor=lambda _request: "ok", allowed_callers=["programmatic"])
    agent = Agent(
        name="shell",
        tools=[ProgrammaticToolCallingTool(), shell_tool],
    )

    processed = process_model_response(
        agent=agent,
        all_tools=agent.tools,
        response=ModelResponse(
            output=[_program(), shell_output],
            usage=Usage(),
            response_id="response_1",
        ),
        output_schema=None,
        handoffs=[],
    )

    assert len(processed.new_items) == 2


@pytest.mark.asyncio
@pytest.mark.parametrize("streamed", [False, True])
async def test_runner_executes_and_replays_programmatic_function_calls(streamed: bool) -> None:
    outputs = [
        [_program(), _function_call()],
        [_program_output(), get_text_message("42 units are available")],
    ]
    model = ScriptedModel(
        [get_exact_output_stream_step(output) for output in outputs] if streamed else outputs
    )

    @function_tool(allowed_callers=["programmatic"])
    def lookup_inventory(sku: str) -> InventoryOutput:
        return InventoryOutput(sku=sku, available_units=42)

    agent = Agent(
        name="inventory",
        model=model,
        tools=[ProgrammaticToolCallingTool(), lookup_inventory],
        model_settings=ModelSettings(tool_choice="programmatic_tool_calling"),
    )

    result: Any
    if streamed:
        result = Runner.run_streamed(agent, "Check inventory")
        events = [event async for event in result.stream_events()]
        assert any(
            getattr(event, "name", None) == "tool_called"
            and _raw_item_type(getattr(getattr(event, "item", None), "raw_item", None)) == "program"
            for event in events
        )
    else:
        result = await Runner.run(agent, "Check inventory")

    assert result.final_output == "42 units are available"
    assert bool(model.calls)
    assert model.calls[0].model_settings.tool_choice == "programmatic_tool_calling"
    assert model.calls[-1].model_settings.tool_choice is None

    function_outputs = [
        item
        for item in result.new_items
        if isinstance(item, ToolCallOutputItem)
        and getattr(item.raw_item, "get", lambda _key: None)("type") == "function_call_output"
    ]
    assert len(function_outputs) == 1
    raw_output = cast(dict[str, Any], function_outputs[0].raw_item)
    assert json.loads(cast(str, raw_output["output"])) == {
        "sku": "A-1",
        "available_units": 42,
    }
    assert _caller_dict(raw_output["caller"]) == PROGRAM_CALLER

    replayed_output = cast(
        dict[str, Any],
        next(
            item
            for item in model.calls[-1].input
            if isinstance(item, dict) and item.get("type") == "function_call_output"
        ),
    )
    assert _caller_dict(replayed_output["caller"]) == PROGRAM_CALLER


@pytest.mark.asyncio
async def test_typed_programmatic_tool_preserves_input_guardrail_rejection() -> None:
    model = ScriptedModel()
    model.extend(
        [
            [_program(), _function_call()],
            [_program_output(), get_text_message("request rejected")],
        ]
    )
    executed = False

    @tool_input_guardrail
    def reject_tool_input(_data: ToolInputGuardrailData) -> ToolGuardrailFunctionOutput:
        return ToolGuardrailFunctionOutput.reject_content("inventory lookup blocked")

    @function_tool(
        allowed_callers=["programmatic"],
        tool_input_guardrails=[reject_tool_input],
    )
    def lookup_inventory(sku: str) -> InventoryOutput:
        nonlocal executed
        executed = True
        return InventoryOutput(sku=sku, available_units=42)

    agent = Agent(
        name="inventory",
        model=model,
        tools=[ProgrammaticToolCallingTool(), lookup_inventory],
    )

    result = await Runner.run(agent, "Check inventory")

    assert executed is False
    assert result.final_output == "request rejected"
    function_outputs = _function_output_raw_items(result)
    assert len(function_outputs) == 1
    assert json.loads(function_outputs[0]["output"]) == {"error": "inventory lookup blocked"}
    assert _caller_dict(function_outputs[0]["caller"]) == PROGRAM_CALLER


@pytest.mark.asyncio
async def test_typed_programmatic_tool_preserves_default_timeout_result() -> None:
    model = ScriptedModel()
    model.extend(
        [
            [_program(), _function_call()],
            [_program_output(), get_text_message("request timed out")],
        ]
    )

    @function_tool(allowed_callers=["programmatic"], timeout=0.01)
    async def lookup_inventory(sku: str) -> InventoryOutput:
        await asyncio.sleep(0.2)
        return InventoryOutput(sku=sku, available_units=42)

    agent = Agent(
        name="inventory",
        model=model,
        tools=[ProgrammaticToolCallingTool(), lookup_inventory],
    )

    result = await Runner.run(agent, "Check inventory")

    assert result.final_output == "request timed out"
    function_outputs = _function_output_raw_items(result)
    assert len(function_outputs) == 1
    timeout_output = json.loads(function_outputs[0]["output"])
    assert isinstance(timeout_output, dict)
    assert "timed out" in timeout_output["error"].lower()
    assert _caller_dict(function_outputs[0]["caller"]) == PROGRAM_CALLER


@pytest.mark.asyncio
async def test_typed_programmatic_tool_preserves_output_guardrail_rejection() -> None:
    model = ScriptedModel()
    model.extend(
        [
            [_program(), _function_call()],
            [_program_output(), get_text_message("request rejected")],
        ]
    )

    @tool_output_guardrail
    def reject_tool_output(_data: ToolOutputGuardrailData) -> ToolGuardrailFunctionOutput:
        return ToolGuardrailFunctionOutput.reject_content("inventory result blocked")

    @function_tool(
        allowed_callers=["programmatic"],
        tool_output_guardrails=[reject_tool_output],
    )
    def lookup_inventory(sku: str) -> InventoryOutput:
        return InventoryOutput(sku=sku, available_units=42)

    agent = Agent(
        name="inventory",
        model=model,
        tools=[ProgrammaticToolCallingTool(), lookup_inventory],
    )

    result = await Runner.run(agent, "Check inventory")

    assert result.final_output == "request rejected"
    function_outputs = _function_output_raw_items(result)
    assert len(function_outputs) == 1
    assert json.loads(function_outputs[0]["output"]) == {"error": "inventory result blocked"}
    assert _caller_dict(function_outputs[0]["caller"]) == PROGRAM_CALLER


@pytest.mark.asyncio
@pytest.mark.parametrize("streaming", [False, True], ids=["nonstreaming", "streaming"])
@pytest.mark.parametrize("serialize_state", [False, True], ids=["in-memory", "serialized"])
@pytest.mark.parametrize(
    "rejection_message",
    [None, 'Denied: "東京"'],
    ids=["default-rejection", "custom-rejection"],
)
async def test_typed_programmatic_tool_preserves_approval_rejection(
    streaming: bool,
    serialize_state: bool,
    rejection_message: str | None,
) -> None:
    outputs = [
        [_program(), _function_call()],
        [_program_output(), get_text_message("request rejected")],
    ]
    model = ScriptedModel(
        [get_exact_output_stream_step(output) for output in outputs] if streaming else outputs
    )

    @function_tool(allowed_callers=["programmatic"], needs_approval=True)
    def lookup_inventory(sku: str) -> InventoryOutput:
        return InventoryOutput(sku=sku, available_units=42)

    agent = Agent(
        name="inventory",
        model=model,
        tools=[ProgrammaticToolCallingTool(), lookup_inventory],
    )
    first_result: Any
    if streaming:
        first_result = Runner.run_streamed(agent, "Check inventory")
        async for _event in first_result.stream_events():
            pass
    else:
        first_result = await Runner.run(agent, "Check inventory")
    assert len(first_result.interruptions) == 1

    state = first_result.to_state()
    if serialize_state:
        state = await RunState.from_json(agent, state.to_json())
    state.reject(state.get_interruptions()[0], rejection_message=rejection_message)
    result: Any
    if streaming:
        result = Runner.run_streamed(agent, state)
        async for _event in result.stream_events():
            pass
    else:
        result = await Runner.run(agent, state)

    assert result.final_output == "request rejected"
    function_outputs = _function_output_raw_items(result)
    assert len(function_outputs) == 1
    expected_message = rejection_message or "Tool execution was not approved."
    assert json.loads(function_outputs[0]["output"]) == {"error": expected_message}
    assert _caller_dict(function_outputs[0]["caller"]) == PROGRAM_CALLER
    assert bool(model.calls)
    replayed_output = cast(
        dict[str, Any],
        next(
            item
            for item in model.calls[-1].input
            if isinstance(item, dict) and item.get("type") == "function_call_output"
        ),
    )
    assert json.loads(replayed_output["output"]) == {"error": expected_message}


@pytest.mark.asyncio
async def test_rebuilt_mapping_programmatic_approval_preserves_caller() -> None:
    model = ScriptedModel()
    model.extend(
        [
            [_program(), _function_call()],
            [_program_output(), get_text_message("done")],
        ]
    )
    executed = False

    @function_tool(allowed_callers=["programmatic"], needs_approval=True)
    def lookup_inventory(sku: str) -> InventoryOutput:
        nonlocal executed
        executed = True
        return InventoryOutput(sku=sku, available_units=42)

    agent = Agent(
        name="inventory",
        model=model,
        tools=[ProgrammaticToolCallingTool(), lookup_inventory],
    )
    first_result = await Runner.run(agent, "Check inventory")
    state = first_result.to_state()
    approval = state.get_interruptions()[0]
    approval.raw_item = cast(Any, approval.raw_item).model_dump(exclude_none=True)
    assert state._last_processed_response is not None
    state._last_processed_response.functions.clear()
    state.approve(approval)

    result = await Runner.run(agent, state)

    assert executed is True
    assert result.final_output == "done"
    function_outputs = _function_output_raw_items(result)
    assert len(function_outputs) == 1
    assert _caller_dict(function_outputs[0]["caller"]) == PROGRAM_CALLER


@pytest.mark.asyncio
async def test_rebuilt_mapping_programmatic_approval_rechecks_caller_permissions() -> None:
    model = ScriptedModel()
    model.enqueue([_program(), _function_call()])
    executed = False

    @function_tool(allowed_callers=["programmatic"], needs_approval=True)
    def lookup_inventory(sku: str) -> InventoryOutput:
        nonlocal executed
        executed = True
        return InventoryOutput(sku=sku, available_units=42)

    agent = Agent(
        name="inventory",
        model=model,
        tools=[ProgrammaticToolCallingTool(), lookup_inventory],
    )
    first_result = await Runner.run(agent, "Check inventory")
    state = first_result.to_state()
    approval = state.get_interruptions()[0]
    approval.raw_item = cast(Any, approval.raw_item).model_dump(exclude_none=True)
    lookup_inventory.allowed_callers = ["direct"]
    assert state._last_processed_response is not None
    state._last_processed_response.functions.clear()
    state.approve(approval)

    with pytest.raises(ModelBehaviorError, match="caller programmatic"):
        await Runner.run(agent, state)

    assert executed is False


@pytest.mark.asyncio
@pytest.mark.parametrize("parent_state", ["missing", "completed"])
async def test_rebuilt_programmatic_approval_requires_active_parent(parent_state: str) -> None:
    model = ScriptedModel()
    model.extend(
        [
            [_program(), _function_call()],
            [_program_output(), get_text_message("done")],
        ]
    )
    executed = False

    @function_tool(allowed_callers=["programmatic"], needs_approval=True)
    def lookup_inventory(sku: str) -> InventoryOutput:
        nonlocal executed
        executed = True
        return InventoryOutput(sku=sku, available_units=42)

    agent = Agent(
        name="inventory",
        model=model,
        tools=[ProgrammaticToolCallingTool(), lookup_inventory],
    )
    first_result = await Runner.run(agent, "Check inventory")
    state = first_result.to_state()
    approval = state.get_interruptions()[0]
    approval.raw_item = cast(Any, approval.raw_item).model_dump(exclude_none=True)
    assert state._last_processed_response is not None
    state._last_processed_response.functions.clear()
    assert state._model_responses
    if parent_state == "missing":
        state._generated_items = [
            item for item in state._generated_items if _raw_item_type(item.raw_item) != "program"
        ]
        state._last_processed_response.new_items = [
            item
            for item in state._last_processed_response.new_items
            if _raw_item_type(item.raw_item) != "program"
        ]
        state._model_responses[-1].output = [
            item for item in state._model_responses[-1].output if _raw_item_type(item) != "program"
        ]
        expected_error = "parent program item"
    else:
        state._model_responses[-1].output.append(_program_output())
        expected_error = "already completed"
    state.approve(approval)

    with pytest.raises(ModelBehaviorError, match=expected_error):
        await Runner.run(agent, state)

    assert executed is False


@pytest.mark.asyncio
async def test_typed_programmatic_tool_preserves_pre_approval_guardrail_rejection() -> None:
    model = ScriptedModel()
    model.extend(
        [
            [_program(), _function_call()],
            [_program_output(), get_text_message("request rejected")],
        ]
    )

    @tool_input_guardrail
    def reject_tool_input(_data: ToolInputGuardrailData) -> ToolGuardrailFunctionOutput:
        return ToolGuardrailFunctionOutput.reject_content("inventory lookup blocked")

    @function_tool(
        allowed_callers=["programmatic"],
        needs_approval=True,
        tool_input_guardrails=[reject_tool_input],
    )
    def lookup_inventory(sku: str) -> InventoryOutput:
        return InventoryOutput(sku=sku, available_units=42)

    agent = Agent(
        name="inventory",
        model=model,
        tools=[ProgrammaticToolCallingTool(), lookup_inventory],
    )
    run_config = RunConfig(
        tool_execution=ToolExecutionConfig(pre_approval_tool_input_guardrails=True)
    )

    result = await Runner.run(agent, "Check inventory", run_config=run_config)

    assert result.final_output == "request rejected"
    function_outputs = _function_output_raw_items(result)
    assert len(function_outputs) == 1
    assert json.loads(function_outputs[0]["output"]) == {"error": "inventory lookup blocked"}
    assert _caller_dict(function_outputs[0]["caller"]) == PROGRAM_CALLER


@pytest.mark.asyncio
async def test_runner_handles_multiple_pauses_from_one_program() -> None:
    model = ScriptedModel()
    second_call = ResponseFunctionToolCall(
        id="function_item_2",
        call_id="call_lookup_2",
        name="lookup_inventory",
        arguments='{"sku":"B-2"}',
        caller=CallerProgram(type="program", caller_id=PROGRAM_CALL_ID),
        type="function_call",
    )
    model.extend(
        [
            [_program(), _function_call()],
            [_program_output("incomplete"), second_call],
            [
                ProgramOutput(
                    id="program_output_item_2",
                    call_id=PROGRAM_CALL_ID,
                    result='{"total":84}',
                    status="completed",
                    type="program_output",
                ),
                get_text_message("84 units are available"),
            ],
        ]
    )

    calls: list[str] = []

    @function_tool(allowed_callers=["programmatic"])
    def lookup_inventory(sku: str) -> InventoryOutput:
        calls.append(sku)
        return InventoryOutput(sku=sku, available_units=42)

    agent = Agent(
        name="inventory",
        model=model,
        tools=[ProgrammaticToolCallingTool(), lookup_inventory],
        model_settings=ModelSettings(tool_choice="programmatic_tool_calling"),
    )

    result = await Runner.run(agent, "Check two SKUs")

    assert result.final_output == "84 units are available"
    assert calls == ["A-1", "B-2"]
    assert model.calls[-1].model_settings.tool_choice is None
    assert len([item for item in result.new_items if isinstance(item, ToolCallOutputItem)]) == 4


@pytest.mark.asyncio
async def test_runner_executes_programmatic_batch_calls_concurrently() -> None:
    model = ScriptedModel()
    batch_calls = [
        ResponseFunctionToolCall(
            id=f"function_item_{index}",
            call_id=f"call_lookup_{index}",
            name="lookup_inventory",
            arguments=json.dumps({"sku": f"SKU-{index}"}),
            caller=CallerProgram(type="program", caller_id=PROGRAM_CALL_ID),
            type="function_call",
        )
        for index in range(9)
    ]
    model.extend(
        [
            [_program(), *batch_calls],
            [_program_output(), get_text_message("batch complete")],
        ]
    )

    active_calls = 0
    max_active_calls = 0

    @function_tool(allowed_callers=["programmatic"])
    async def lookup_inventory(sku: str) -> InventoryOutput:
        nonlocal active_calls, max_active_calls
        active_calls += 1
        max_active_calls = max(max_active_calls, active_calls)
        await asyncio.sleep(0.01)
        active_calls -= 1
        return InventoryOutput(sku=sku, available_units=42)

    agent = Agent(
        name="inventory",
        model=model,
        tools=[ProgrammaticToolCallingTool(), lookup_inventory],
    )

    result = await Runner.run(agent, "Check nine SKUs")

    function_outputs = [
        item
        for item in result.new_items
        if isinstance(item, ToolCallOutputItem)
        and isinstance(item.raw_item, dict)
        and item.raw_item.get("type") == "function_call_output"
    ]
    assert result.final_output == "batch complete"
    assert len(result.raw_responses) == 2
    assert len(function_outputs) == 9
    assert max_active_calls == 9
    assert all(
        _caller_dict(cast(dict[str, Any], item.raw_item)["caller"]) == PROGRAM_CALLER
        for item in function_outputs
    )


@pytest.mark.asyncio
async def test_previous_response_id_continuation_sends_only_program_function_output() -> None:
    model = ScriptedModel()
    model.extend(
        [
            [_program(), _function_call()],
            [_program_output(), get_text_message("done")],
        ]
    )

    @function_tool(allowed_callers=["programmatic"])
    def lookup_inventory(sku: str) -> InventoryOutput:
        return InventoryOutput(sku=sku, available_units=42)

    agent = Agent(
        name="inventory",
        model=model,
        tools=[ProgrammaticToolCallingTool(), lookup_inventory],
    )

    result = await Runner.run(agent, "Check inventory", auto_previous_response_id=True)

    assert result.final_output == "done"
    assert model.calls[-1].previous_response_id == "resp-789"
    last_input = model.calls[-1].input
    assert isinstance(last_input, list)
    assert len(last_input) == 1
    function_output = cast(dict[str, Any], last_input[0])
    assert function_output["type"] == "function_call_output"
    assert _caller_dict(function_output["caller"]) == PROGRAM_CALLER


@pytest.mark.asyncio
@pytest.mark.parametrize("streamed", [False, True])
async def test_previous_response_id_continuation_accepts_server_owned_program_output(
    streamed: bool,
) -> None:
    output = [_program_output(), get_text_message("done")]
    model = ScriptedModel([get_exact_output_stream_step(output) if streamed else output])

    @function_tool(allowed_callers=["programmatic"])
    def lookup_inventory(sku: str) -> InventoryOutput:
        return InventoryOutput(sku=sku, available_units=42)

    agent = Agent(
        name="inventory",
        model=model,
        tools=[ProgrammaticToolCallingTool(), lookup_inventory],
    )
    submitted_delta = [
        {
            "type": "function_call_output",
            "call_id": FUNCTION_CALL_ID,
            "output": '{"sku":"A-1","available_units":42}',
            "caller": PROGRAM_CALLER,
        }
    ]

    result: Any
    if streamed:
        result = Runner.run_streamed(
            agent,
            cast(Any, submitted_delta),
            previous_response_id="response_with_program_parent",
        )
        _events = [event async for event in result.stream_events()]
    else:
        result = await Runner.run(
            agent,
            cast(Any, submitted_delta),
            previous_response_id="response_with_program_parent",
        )

    assert result.final_output == "done"
    assert model.calls[-1].previous_response_id == "response_with_program_parent"


@pytest.mark.asyncio
async def test_previous_response_id_continuation_accepts_repeated_program_pause() -> None:
    model = ScriptedModel()
    model.extend(
        [
            [_function_call()],
            [_program_output(), get_text_message("done")],
        ]
    )
    executed = False

    @function_tool(allowed_callers=["programmatic"])
    def lookup_inventory(sku: str) -> InventoryOutput:
        nonlocal executed
        executed = True
        return InventoryOutput(sku=sku, available_units=42)

    agent = Agent(
        name="inventory",
        model=model,
        tools=[ProgrammaticToolCallingTool(), lookup_inventory],
    )
    submitted_delta = [
        {
            "type": "function_call_output",
            "call_id": "call_previous_lookup",
            "output": '{"sku":"A-0","available_units":21}',
            "caller": PROGRAM_CALLER,
        }
    ]

    result = await Runner.run(
        agent,
        cast(Any, submitted_delta),
        previous_response_id="response_with_program_parent",
    )

    assert executed is True
    assert result.final_output == "done"
    assert len(result.raw_responses) == 2


@pytest.mark.asyncio
@pytest.mark.parametrize("parent_source", ["caller", "incomplete_program_output"])
async def test_run_state_round_trip_preserves_server_owned_program_parent(
    parent_source: str,
) -> None:
    model = ScriptedModel()
    model.extend(
        [
            [_function_call()],
            [_program_output(), get_text_message("done")],
        ]
    )
    executed = False

    @function_tool(allowed_callers=["programmatic"], needs_approval=True)
    def lookup_inventory(sku: str) -> InventoryOutput:
        nonlocal executed
        executed = True
        return InventoryOutput(sku=sku, available_units=42)

    agent = Agent(
        name="inventory",
        model=model,
        tools=[ProgrammaticToolCallingTool(), lookup_inventory],
    )
    if parent_source == "caller":
        submitted_delta = [
            {
                "type": "function_call_output",
                "call_id": "call_previous_lookup",
                "output": '{"sku":"A-0","available_units":21}',
                "caller": PROGRAM_CALLER,
            }
        ]
    else:
        submitted_delta = [_program_output("incomplete").model_dump(exclude_none=True)]

    first_result = await Runner.run(
        agent,
        cast(Any, submitted_delta),
        previous_response_id="response_with_program_parent",
    )
    assert len(first_result.interruptions) == 1

    restored_state = await RunState.from_json(agent, first_result.to_state().to_json())
    approval = restored_state.get_interruptions()[0]
    restored_state.approve(approval)
    result = await Runner.run(agent, restored_state)

    assert executed is True
    assert result.final_output == "done"
    assert model.calls[-1].previous_response_id == "resp-789"


@pytest.mark.asyncio
async def test_sqlite_session_round_trip_preserves_program_history_and_caller() -> None:
    model = ScriptedModel()
    model.extend(
        [
            [_program(), _function_call()],
            [_program_output(), get_text_message("done")],
        ]
    )

    @function_tool(allowed_callers=["programmatic"])
    def lookup_inventory(sku: str) -> InventoryOutput:
        return InventoryOutput(sku=sku, available_units=42)

    agent = Agent(
        name="inventory",
        model=model,
        tools=[ProgrammaticToolCallingTool(), lookup_inventory],
    )
    session = SQLiteSession("programmatic-tool-calling")
    try:
        result = await Runner.run(agent, "Check inventory", session=session)
        assert result.final_output == "done"

        session_items = await session.get_items()
        assert [_raw_item_type(item) for item in session_items] == [
            None,
            "program",
            "function_call",
            "function_call_output",
            "program_output",
            "message",
        ]
        function_output = next(
            cast(dict[str, Any], item)
            for item in session_items
            if _raw_item_type(item) == "function_call_output"
        )
        assert _caller_dict(function_output["caller"]) == PROGRAM_CALLER
    finally:
        session.close()


@pytest.mark.asyncio
async def test_nested_handoff_summarizes_complete_programmatic_transcript() -> None:
    model = ScriptedModel()
    delegate = Agent(name="delegate", model=model)
    model.extend(
        [
            [_program(), _function_call()],
            [_program_output(), get_handoff_tool_call(delegate)],
            [get_text_message("done")],
        ]
    )

    @function_tool(allowed_callers=["programmatic"])
    def lookup_inventory(sku: str) -> InventoryOutput:
        return InventoryOutput(sku=sku, available_units=42)

    triage = Agent(
        name="triage",
        model=model,
        handoffs=[delegate],
        tools=[ProgrammaticToolCallingTool(), lookup_inventory],
    )
    captured_inputs: list[list[Any]] = []

    def capture_model_input(data: Any) -> Any:
        captured_inputs.append(list(data.model_data.input))
        return data.model_data

    session = SQLiteSession("programmatic-tool-calling-nested-handoff")
    try:
        result = await Runner.run(
            triage,
            "Check inventory and delegate the final response.",
            run_config=RunConfig(
                nest_handoff_history=True,
                call_model_input_filter=capture_model_input,
            ),
            session=session,
        )

        assert result.final_output == "done"
        handoff_input = captured_inputs[-1]
        handoff_types = [_raw_item_type(item) for item in handoff_input]
        assert not {
            "program",
            "function_call",
            "function_call_output",
            "program_output",
        }.intersection(handoff_types)

        summary_text = "\n".join(
            cast(str, item.get("content"))
            for item in handoff_input
            if isinstance(item, dict) and isinstance(item.get("content"), str)
        )
        assert '"type": "program"' in summary_text
        assert '"type": "function_call"' in summary_text
        assert '"type": "function_call_output"' in summary_text
        assert '"type": "program_output"' in summary_text

        session_items = await session.get_items()
        assert [_raw_item_type(item) for item in session_items] == [
            None,
            "program",
            "function_call",
            "function_call_output",
            "program_output",
            "function_call",
            "function_call_output",
            "message",
        ]
    finally:
        session.close()


@pytest.mark.asyncio
async def test_non_function_programmatic_outputs_preserve_caller() -> None:
    caller = cast(Any, PROGRAM_CALLER)

    async def run_tool(tool: Any, tool_call: Any) -> dict[str, Any]:
        model = ScriptedModel()
        model.extend([[_program(), tool_call], [get_text_message("done")]])
        agent = Agent(
            name="tool agent",
            model=model,
            tools=[ProgrammaticToolCallingTool(), tool],
        )
        result = await Runner.run(agent, "Run the tool")
        return next(
            cast(dict[str, Any], item.raw_item)
            for item in result.new_items
            if isinstance(item, ToolCallOutputItem)
        )

    async def shell_executor(_request: Any) -> str:
        return "shell done"

    shell_output = await run_tool(
        ShellTool(executor=shell_executor, allowed_callers=["programmatic"]),
        ResponseFunctionShellToolCall(
            id="shell_item",
            call_id="call_shell",
            action=Action(commands=["echo ok"]),
            status="completed",
            type="shell_call",
            caller=caller,
        ),
    )
    assert _caller_dict(shell_output["caller"]) == PROGRAM_CALLER

    custom_output = await run_tool(
        CustomTool(
            name="custom",
            description="Custom tool",
            on_invoke_tool=lambda _context, _input: "custom done",
            allowed_callers=["programmatic"],
        ),
        ResponseCustomToolCall(
            id="custom_item",
            call_id="call_custom",
            input="input",
            name="custom",
            type="custom_tool_call",
            caller=caller,
        ),
    )
    assert _caller_dict(custom_output["caller"]) == PROGRAM_CALLER

    class Editor:
        def create_file(self, _operation: Any) -> str:
            return "patch done"

        def update_file(self, _operation: Any) -> str:
            return "patch done"

        def delete_file(self, _operation: Any) -> str:
            return "patch done"

    apply_patch_output = await run_tool(
        ApplyPatchTool(editor=Editor(), allowed_callers=["programmatic"]),
        {
            "id": "apply_patch_item",
            "call_id": "call_apply_patch",
            "operation": {"type": "create_file", "path": "example.txt", "diff": "hello"},
            "status": "completed",
            "type": "apply_patch_call",
            "caller": caller,
        },
    )
    assert _caller_dict(apply_patch_output["caller"]) == PROGRAM_CALLER