turn_resolution.py

from __future__ import annotations

import asyncio
import inspect
from collections.abc import Awaitable, Callable, Mapping, Sequence
from typing import Any, Literal, cast

from openai.types.responses import (
    ResponseCompactionItem,
    ResponseComputerToolCall,
    ResponseCustomToolCall,
    ResponseFileSearchToolCall,
    ResponseFunctionShellToolCallOutput,
    ResponseFunctionToolCall,
    ResponseFunctionWebSearch,
    ResponseOutputMessage,
)
from openai.types.responses.response_code_interpreter_tool_call import (
    ResponseCodeInterpreterToolCall,
)
from openai.types.responses.response_output_item import (
    ImageGenerationCall,
    LocalShellCall,
    McpApprovalRequest,
    McpCall,
    McpListTools,
)
from openai.types.responses.response_reasoning_item import ResponseReasoningItem

from .._mcp_tool_metadata import collect_mcp_list_tools_metadata
from .._tool_identity import (
    build_function_tool_lookup_map,
    get_function_tool_lookup_key,
    get_function_tool_lookup_key_for_call,
    get_function_tool_lookup_key_for_tool,
    get_tool_call_namespace,
    get_tool_call_qualified_name,
    get_tool_call_trace_name,
    normalize_tool_call_for_function_tool,
    should_allow_bare_name_approval_alias,
)
from ..agent import Agent, ToolsToFinalOutputResult
from ..agent_output import AgentOutputSchemaBase
from ..agent_tool_state import get_agent_tool_state_scope, peek_agent_tool_run_result
from ..exceptions import ModelBehaviorError, UserError
from ..handoffs import Handoff, HandoffInputData, HandoffInputFilter, nest_handoff_history
from ..items import (
    CompactionItem,
    HandoffCallItem,
    HandoffOutputItem,
    ItemHelpers,
    MCPApprovalRequestItem,
    MCPListToolsItem,
    MessageOutputItem,
    ModelResponse,
    ReasoningItem,
    RunItem,
    ToolApprovalItem,
    ToolCallItem,
    ToolCallOutputItem,
    ToolSearchCallItem,
    ToolSearchOutputItem,
    TResponseInputItem,
    coerce_tool_search_call_raw_item,
    coerce_tool_search_output_raw_item,
)
from ..lifecycle import RunHooks
from ..logger import logger
from ..run_config import RunConfig
from ..run_context import AgentHookContext, RunContextWrapper, TContext
from ..run_state import RunState
from ..stream_events import StreamEvent
from ..tool import (
    ApplyPatchTool,
    ComputerTool,
    CustomTool,
    FunctionTool,
    FunctionToolResult,
    HostedMCPTool,
    LocalShellTool,
    ShellTool,
    Tool,
    ToolOrigin,
    ToolOriginType,
    get_function_tool_origin,
)
from ..tool_guardrails import ToolInputGuardrailResult, ToolOutputGuardrailResult
from ..tracing import SpanError, handoff_span
from ..util import _coro, _error_tracing
from ..util._approvals import evaluate_needs_approval_setting
from .agent_bindings import AgentBindings
from .items import (
    REJECTION_MESSAGE,
    apply_patch_rejection_item,
    function_rejection_item,
    shell_rejection_item,
)
from .run_steps import (
    NOT_FINAL_OUTPUT,
    NextStepFinalOutput,
    NextStepHandoff,
    NextStepInterruption,
    NextStepRunAgain,
    ProcessedResponse,
    QueueCompleteSentinel,
    SingleStepResult,
    ToolRunApplyPatchCall,
    ToolRunComputerAction,
    ToolRunCustom,
    ToolRunFunction,
    ToolRunHandoff,
    ToolRunLocalShellCall,
    ToolRunMCPApprovalRequest,
    ToolRunShellCall,
)
from .streaming import stream_step_items_to_queue
from .tool_execution import (
    build_litellm_json_tool_call,
    coerce_apply_patch_operations,
    coerce_shell_call,
    extract_apply_patch_call_id,
    extract_shell_call_id,
    extract_tool_call_id,
    function_needs_approval,
    get_mapping_or_attr,
    index_approval_items_by_call_id,
    is_apply_patch_name,
    parse_apply_patch_custom_input,
    parse_apply_patch_function_args,
    process_hosted_mcp_approvals,
    resolve_approval_rejection_message,
    resolve_enabled_function_tools,
    should_keep_hosted_mcp_item,
)
from .tool_planning import (
    _append_mcp_callback_results,
    _build_plan_for_fresh_turn,
    _build_plan_for_resume_turn,
    _build_tool_output_index,
    _build_tool_result_items,
    _collect_runs_by_approval,
    _collect_tool_interruptions,
    _dedupe_tool_call_items,
    _execute_tool_plan,
    _make_unique_item_appender,
    _select_function_tool_runs_for_resume,
)

__all__ = [
    "execute_final_output_step",
    "execute_final_output",
    "execute_handoffs",
    "check_for_final_output_from_tools",
    "process_model_response",
    "execute_tools_and_side_effects",
    "resolve_interrupted_turn",
    "get_single_step_result_from_response",
    "run_final_output_hooks",
]


async def _maybe_finalize_from_tool_results(
    *,
    public_agent: Agent[TContext],
    original_input: str | list[TResponseInputItem],
    new_response: ModelResponse,
    pre_step_items: list[RunItem],
    new_step_items: list[RunItem],
    function_results: list[FunctionToolResult],
    hooks: RunHooks[TContext],
    context_wrapper: RunContextWrapper[TContext],
    tool_input_guardrail_results: list[ToolInputGuardrailResult],
    tool_output_guardrail_results: list[ToolOutputGuardrailResult],
) -> SingleStepResult | None:
    check_tool_use = await check_for_final_output_from_tools(
        public_agent, function_results, context_wrapper
    )
    if not check_tool_use.is_final_output:
        return None

    if not public_agent.output_type or public_agent.output_type is str:
        check_tool_use.final_output = str(check_tool_use.final_output)

    if check_tool_use.final_output is None:
        logger.error(
            "Model returned a final output of None. Not raising an error because we assume"
            "you know what you're doing."
        )

    return await execute_final_output(
        public_agent=public_agent,
        original_input=original_input,
        new_response=new_response,
        pre_step_items=pre_step_items,
        new_step_items=new_step_items,
        final_output=check_tool_use.final_output,
        hooks=hooks,
        context_wrapper=context_wrapper,
        tool_input_guardrail_results=tool_input_guardrail_results,
        tool_output_guardrail_results=tool_output_guardrail_results,
    )


async def run_final_output_hooks(
    agent: Agent[TContext],
    hooks: RunHooks[TContext],
    context_wrapper: RunContextWrapper[TContext],
    final_output: Any,
) -> None:
    agent_hook_context = AgentHookContext(
        context=context_wrapper.context,
        usage=context_wrapper.usage,
        _approvals=context_wrapper._approvals,
        turn_input=context_wrapper.turn_input,
    )

    await asyncio.gather(
        hooks.on_agent_end(agent_hook_context, agent, final_output),
        agent.hooks.on_end(agent_hook_context, agent, final_output)
        if agent.hooks
        else _coro.noop_coroutine(),
    )


async def execute_final_output_step(
    *,
    public_agent: Agent[Any],
    original_input: str | list[TResponseInputItem],
    new_response: ModelResponse,
    pre_step_items: list[RunItem],
    new_step_items: list[RunItem],
    final_output: Any,
    hooks: RunHooks[Any],
    context_wrapper: RunContextWrapper[Any],
    tool_input_guardrail_results: list[ToolInputGuardrailResult],
    tool_output_guardrail_results: list[ToolOutputGuardrailResult],
    run_final_output_hooks_fn: Callable[
        [Agent[Any], RunHooks[Any], RunContextWrapper[Any], Any], Awaitable[None]
    ]
    | None = None,
) -> SingleStepResult:
    """Finalize a turn once final output is known and run end hooks."""
    final_output_hooks = run_final_output_hooks_fn or run_final_output_hooks
    await final_output_hooks(public_agent, hooks, context_wrapper, final_output)

    return SingleStepResult(
        original_input=original_input,
        model_response=new_response,
        pre_step_items=pre_step_items,
        new_step_items=new_step_items,
        next_step=NextStepFinalOutput(final_output),
        tool_input_guardrail_results=tool_input_guardrail_results,
        tool_output_guardrail_results=tool_output_guardrail_results,
        output_guardrail_results=[],
    )


async def execute_final_output(
    *,
    public_agent: Agent[Any],
    original_input: str | list[TResponseInputItem],
    new_response: ModelResponse,
    pre_step_items: list[RunItem],
    new_step_items: list[RunItem],
    final_output: Any,
    hooks: RunHooks[Any],
    context_wrapper: RunContextWrapper[Any],
    tool_input_guardrail_results: list[ToolInputGuardrailResult],
    tool_output_guardrail_results: list[ToolOutputGuardrailResult],
    run_final_output_hooks_fn: Callable[
        [Agent[Any], RunHooks[Any], RunContextWrapper[Any], Any], Awaitable[None]
    ]
    | None = None,
) -> SingleStepResult:
    """Convenience wrapper to finalize a turn and run end hooks."""
    return await execute_final_output_step(
        public_agent=public_agent,
        original_input=original_input,
        new_response=new_response,
        pre_step_items=pre_step_items,
        new_step_items=new_step_items,
        final_output=final_output,
        hooks=hooks,
        context_wrapper=context_wrapper,
        tool_input_guardrail_results=tool_input_guardrail_results,
        tool_output_guardrail_results=tool_output_guardrail_results,
        run_final_output_hooks_fn=run_final_output_hooks_fn,
    )


def _resolve_server_managed_handoff_behavior(
    *,
    handoff: Handoff[Any, Agent[Any]],
    from_agent: Agent[Any],
    to_agent: Agent[Any],
    run_config: RunConfig,
    server_manages_conversation: bool,
    input_filter: HandoffInputFilter | None,
    should_nest_history: bool,
) -> tuple[HandoffInputFilter | None, bool]:
    if not server_manages_conversation:
        return input_filter, should_nest_history

    if input_filter is not None:
        raise UserError(
            "Server-managed conversations do not support handoff input filters. "
            "Remove Handoff.input_filter or RunConfig.handoff_input_filter, "
            "or disable conversation_id, previous_response_id, and auto_previous_response_id."
        )

    if not should_nest_history:
        return input_filter, should_nest_history

    logger.warning(
        "Server-managed conversations do not support nest_handoff_history for handoff "
        "%s -> %s. Disabling nested handoff history and continuing with delta-only input.",
        from_agent.name,
        to_agent.name,
    )
    return input_filter, False


async def execute_handoffs(
    *,
    public_agent: Agent[TContext],
    original_input: str | list[TResponseInputItem],
    pre_step_items: list[RunItem],
    new_step_items: list[RunItem],
    new_response: ModelResponse,
    run_handoffs: list[ToolRunHandoff],
    hooks: RunHooks[TContext],
    context_wrapper: RunContextWrapper[TContext],
    run_config: RunConfig,
    server_manages_conversation: bool = False,
    nest_handoff_history_fn: Callable[..., HandoffInputData] | None = None,
) -> SingleStepResult:
    """Execute a handoff and prepare the next turn for the new agent."""

    def nest_history(data: HandoffInputData, mapper: Any | None = None) -> HandoffInputData:
        if nest_handoff_history_fn is None:
            return nest_handoff_history(data, history_mapper=mapper)
        return nest_handoff_history_fn(data, mapper)

    multiple_handoffs = len(run_handoffs) > 1
    if multiple_handoffs:
        output_message = "Multiple handoffs detected, ignoring this one."
        new_step_items.extend(
            [
                ToolCallOutputItem(
                    output=output_message,
                    raw_item=ItemHelpers.tool_call_output_item(handoff.tool_call, output_message),
                    agent=public_agent,
                )
                for handoff in run_handoffs[1:]
            ]
        )

    actual_handoff = run_handoffs[0]
    with handoff_span(from_agent=public_agent.name) as span_handoff:
        handoff = actual_handoff.handoff
        new_agent: Agent[Any] = await handoff.on_invoke_handoff(
            context_wrapper, actual_handoff.tool_call.arguments
        )
        span_handoff.span_data.to_agent = new_agent.name
        if multiple_handoffs:
            requested_agents = [handoff.handoff.agent_name for handoff in run_handoffs]
            span_handoff.set_error(
                SpanError(
                    message="Multiple handoffs requested",
                    data={
                        "requested_agents": requested_agents,
                    },
                )
            )

        new_step_items.append(
            HandoffOutputItem(
                agent=public_agent,
                raw_item=ItemHelpers.tool_call_output_item(
                    actual_handoff.tool_call,
                    handoff.get_transfer_message(new_agent),
                ),
                source_agent=public_agent,
                target_agent=new_agent,
            )
        )

        await asyncio.gather(
            hooks.on_handoff(
                context=context_wrapper,
                from_agent=public_agent,
                to_agent=new_agent,
            ),
            (
                public_agent.hooks.on_handoff(
                    context_wrapper,
                    agent=new_agent,
                    source=public_agent,
                )
                if public_agent.hooks
                else _coro.noop_coroutine()
            ),
        )

        input_filter = handoff.input_filter or (
            run_config.handoff_input_filter if run_config else None
        )
        handoff_nest_setting = handoff.nest_handoff_history
        should_nest_history = (
            handoff_nest_setting
            if handoff_nest_setting is not None
            else run_config.nest_handoff_history
        )
        input_filter, should_nest_history = _resolve_server_managed_handoff_behavior(
            handoff=handoff,
            from_agent=public_agent,
            to_agent=new_agent,
            run_config=run_config,
            server_manages_conversation=server_manages_conversation,
            input_filter=input_filter,
            should_nest_history=should_nest_history,
        )
        handoff_input_data: HandoffInputData | None = None
        session_step_items: list[RunItem] | None = None
        if input_filter or should_nest_history:
            handoff_input_data = HandoffInputData(
                input_history=tuple(original_input)
                if isinstance(original_input, list)
                else original_input,
                pre_handoff_items=tuple(pre_step_items),
                new_items=tuple(new_step_items),
                run_context=context_wrapper,
            )

        if input_filter and handoff_input_data is not None:
            filter_name = getattr(input_filter, "__qualname__", repr(input_filter))
            from_agent = getattr(public_agent, "name", public_agent.__class__.__name__)
            to_agent = getattr(new_agent, "name", new_agent.__class__.__name__)
            logger.debug(
                "Filtering handoff inputs with %s for %s -> %s",
                filter_name,
                from_agent,
                to_agent,
            )
            if not callable(input_filter):
                _error_tracing.attach_error_to_span(
                    span_handoff,
                    SpanError(
                        message="Invalid input filter",
                        data={"details": "not callable()"},
                    ),
                )
                raise UserError(f"Invalid input filter: {input_filter}")
            filtered = input_filter(handoff_input_data)
            if inspect.isawaitable(filtered):
                filtered = await filtered
            if not isinstance(filtered, HandoffInputData):
                _error_tracing.attach_error_to_span(
                    span_handoff,
                    SpanError(
                        message="Invalid input filter result",
                        data={"details": "not a HandoffInputData"},
                    ),
                )
                raise UserError(f"Invalid input filter result: {filtered}")

            original_input = (
                filtered.input_history
                if isinstance(filtered.input_history, str)
                else list(filtered.input_history)
            )
            pre_step_items = list(filtered.pre_handoff_items)
            new_step_items = list(filtered.new_items)
            # For custom input filters, keep full new_items for session history and
            # use input_items for model input when provided.
            if filtered.input_items is not None:
                session_step_items = list(filtered.new_items)
                new_step_items = list(filtered.input_items)
            else:
                session_step_items = None
        elif should_nest_history and handoff_input_data is not None:
            nested = nest_history(handoff_input_data, run_config.handoff_history_mapper)
            original_input = (
                nested.input_history
                if isinstance(nested.input_history, str)
                else list(nested.input_history)
            )
            pre_step_items = list(nested.pre_handoff_items)
            # Keep full new_items for session history.
            session_step_items = list(nested.new_items)
            # Use input_items (filtered) for model input if available.
            if nested.input_items is not None:
                new_step_items = list(nested.input_items)
            else:
                new_step_items = session_step_items
        else:
            # No filtering or nesting - session_step_items not needed.
            session_step_items = None

    return SingleStepResult(
        original_input=original_input,
        model_response=new_response,
        pre_step_items=pre_step_items,
        new_step_items=new_step_items,
        next_step=NextStepHandoff(new_agent),
        tool_input_guardrail_results=[],
        tool_output_guardrail_results=[],
        session_step_items=session_step_items,
    )


async def check_for_final_output_from_tools(
    agent: Agent[TContext],
    tool_results: list[FunctionToolResult],
    context_wrapper: RunContextWrapper[TContext],
) -> ToolsToFinalOutputResult:
    """Determine if tool results should produce a final output."""
    if not tool_results:
        return NOT_FINAL_OUTPUT

    if agent.tool_use_behavior == "run_llm_again":
        return NOT_FINAL_OUTPUT
    elif agent.tool_use_behavior == "stop_on_first_tool":
        return ToolsToFinalOutputResult(is_final_output=True, final_output=tool_results[0].output)
    elif isinstance(agent.tool_use_behavior, dict):
        names = agent.tool_use_behavior.get("stop_at_tool_names", [])
        for tool_result in tool_results:
            if tool_result.tool.name in names or tool_result.tool.qualified_name in names:
                return ToolsToFinalOutputResult(
                    is_final_output=True, final_output=tool_result.output
                )
        return ToolsToFinalOutputResult(is_final_output=False, final_output=None)
    elif callable(agent.tool_use_behavior):
        if inspect.iscoroutinefunction(agent.tool_use_behavior):
            return await cast(
                Awaitable[ToolsToFinalOutputResult],
                agent.tool_use_behavior(context_wrapper, tool_results),
            )
        return cast(
            ToolsToFinalOutputResult, agent.tool_use_behavior(context_wrapper, tool_results)
        )

    logger.error("Invalid tool_use_behavior: %s", agent.tool_use_behavior)
    raise UserError(f"Invalid tool_use_behavior: {agent.tool_use_behavior}")


async def execute_tools_and_side_effects(
    *,
    bindings: AgentBindings[TContext],
    original_input: str | list[TResponseInputItem],
    pre_step_items: list[RunItem],
    new_response: ModelResponse,
    processed_response: ProcessedResponse,
    output_schema: AgentOutputSchemaBase | None,
    hooks: RunHooks[TContext],
    context_wrapper: RunContextWrapper[TContext],
    run_config: RunConfig,
    server_manages_conversation: bool = False,
) -> SingleStepResult:
    """Run one turn of the loop, coordinating tools, approvals, guardrails, and handoffs."""
    public_agent = bindings.public_agent

    execute_final_output_call = execute_final_output
    execute_handoffs_call = execute_handoffs

    pre_step_items = list(pre_step_items)
    approval_items_by_call_id = index_approval_items_by_call_id(pre_step_items)

    plan = _build_plan_for_fresh_turn(
        processed_response=processed_response,
        agent=public_agent,
        context_wrapper=context_wrapper,
        approval_items_by_call_id=approval_items_by_call_id,
    )

    new_step_items = _dedupe_tool_call_items(
        existing_items=pre_step_items,
        new_items=processed_response.new_items,
    )

    (
        function_results,
        tool_input_guardrail_results,
        tool_output_guardrail_results,
        computer_results,
        custom_tool_results,
        shell_results,
        apply_patch_results,
        local_shell_results,
    ) = await _execute_tool_plan(
        plan=plan,
        bindings=bindings,
        hooks=hooks,
        context_wrapper=context_wrapper,
        run_config=run_config,
    )
    new_step_items.extend(
        _build_tool_result_items(
            function_results=function_results,
            computer_results=computer_results,
            custom_tool_results=custom_tool_results,
            shell_results=shell_results,
            apply_patch_results=apply_patch_results,
            local_shell_results=local_shell_results,
        )
    )

    interruptions = _collect_tool_interruptions(
        function_results=function_results,
        custom_tool_results=custom_tool_results,
        shell_results=shell_results,
        apply_patch_results=apply_patch_results,
    )
    if plan.approved_mcp_responses:
        new_step_items.extend(plan.approved_mcp_responses)
    if plan.pending_interruptions:
        interruptions.extend(plan.pending_interruptions)
        new_step_items.extend(plan.pending_interruptions)

    processed_response.interruptions = interruptions

    if interruptions:
        return SingleStepResult(
            original_input=original_input,
            model_response=new_response,
            pre_step_items=pre_step_items,
            new_step_items=new_step_items,
            next_step=NextStepInterruption(interruptions=interruptions),
            tool_input_guardrail_results=tool_input_guardrail_results,
            tool_output_guardrail_results=tool_output_guardrail_results,
            processed_response=processed_response,
        )

    await _append_mcp_callback_results(
        agent=public_agent,
        requests=plan.mcp_requests_with_callback,
        context_wrapper=context_wrapper,
        append_item=new_step_items.append,
    )

    if run_handoffs := processed_response.handoffs:
        return await execute_handoffs_call(
            public_agent=public_agent,
            original_input=original_input,
            pre_step_items=pre_step_items,
            new_step_items=new_step_items,
            new_response=new_response,
            run_handoffs=run_handoffs,
            hooks=hooks,
            context_wrapper=context_wrapper,
            run_config=run_config,
            server_manages_conversation=server_manages_conversation,
        )

    tool_final_output = await _maybe_finalize_from_tool_results(
        public_agent=public_agent,
        original_input=original_input,
        new_response=new_response,
        pre_step_items=pre_step_items,
        new_step_items=new_step_items,
        function_results=function_results,
        hooks=hooks,
        context_wrapper=context_wrapper,
        tool_input_guardrail_results=tool_input_guardrail_results,
        tool_output_guardrail_results=tool_output_guardrail_results,
    )
    if tool_final_output is not None:
        return tool_final_output

    message_items = [item for item in new_step_items if isinstance(item, MessageOutputItem)]
    potential_final_output_text = (
        ItemHelpers.extract_text(message_items[-1].raw_item) if message_items else None
    )

    if not processed_response.has_tools_or_approvals_to_run():
        has_tool_activity_without_message = not message_items and bool(
            processed_response.tools_used
        )
        if not has_tool_activity_without_message:
            if output_schema and not output_schema.is_plain_text() and potential_final_output_text:
                final_output = output_schema.validate_json(potential_final_output_text)
                return await execute_final_output_call(
                    public_agent=public_agent,
                    original_input=original_input,
                    new_response=new_response,
                    pre_step_items=pre_step_items,
                    new_step_items=new_step_items,
                    final_output=final_output,
                    hooks=hooks,
                    context_wrapper=context_wrapper,
                    tool_input_guardrail_results=tool_input_guardrail_results,
                    tool_output_guardrail_results=tool_output_guardrail_results,
                )
            if not output_schema or output_schema.is_plain_text():
                return await execute_final_output_call(
                    public_agent=public_agent,
                    original_input=original_input,
                    new_response=new_response,
                    pre_step_items=pre_step_items,
                    new_step_items=new_step_items,
                    final_output=potential_final_output_text or "",
                    hooks=hooks,
                    context_wrapper=context_wrapper,
                    tool_input_guardrail_results=tool_input_guardrail_results,
                    tool_output_guardrail_results=tool_output_guardrail_results,
                )

    return SingleStepResult(
        original_input=original_input,
        model_response=new_response,
        pre_step_items=pre_step_items,
        new_step_items=new_step_items,
        next_step=NextStepRunAgain(),
        tool_input_guardrail_results=tool_input_guardrail_results,
        tool_output_guardrail_results=tool_output_guardrail_results,
    )


async def resolve_interrupted_turn(
    *,
    bindings: AgentBindings[TContext],
    original_input: str | list[TResponseInputItem],
    original_pre_step_items: list[RunItem],
    new_response: ModelResponse,
    processed_response: ProcessedResponse,
    hooks: RunHooks[TContext],
    context_wrapper: RunContextWrapper[TContext],
    run_config: RunConfig,
    server_manages_conversation: bool = False,
    run_state: RunState | None = None,
    nest_handoff_history_fn: Callable[..., HandoffInputData] | None = None,
) -> SingleStepResult:
    """Continue a turn that was previously interrupted waiting for tool approval."""
    public_agent = bindings.public_agent
    execution_agent = bindings.execution_agent

    execute_handoffs_call = execute_handoffs

    def nest_history(data: HandoffInputData, mapper: Any | None = None) -> HandoffInputData:
        if nest_handoff_history_fn is None:
            return nest_handoff_history(data, history_mapper=mapper)
        return nest_handoff_history_fn(data, mapper)

    def _pending_approvals_from_state() -> list[ToolApprovalItem]:
        if (
            run_state is not None
            and hasattr(run_state, "_current_step")
            and isinstance(run_state._current_step, NextStepInterruption)
        ):
            return [
                item
                for item in run_state._current_step.interruptions
                if isinstance(item, ToolApprovalItem)
            ]
        return [item for item in original_pre_step_items if isinstance(item, ToolApprovalItem)]

    async def _record_function_rejection(
        call_id: str | None,
        tool_call: ResponseFunctionToolCall,
        function_tool: FunctionTool,
    ) -> None:
        if isinstance(call_id, str) and call_id in rejected_function_call_ids:
            return
        rejection_message = REJECTION_MESSAGE
        if call_id:
            tool_namespace = get_tool_call_namespace(tool_call)
            rejection_message = await resolve_approval_rejection_message(
                context_wrapper=context_wrapper,
                run_config=run_config,
                tool_type="function",
                tool_name=get_tool_call_trace_name(tool_call) or function_tool.name,
                call_id=call_id,
                tool_namespace=tool_namespace,
                tool_lookup_key=get_function_tool_lookup_key_for_tool(function_tool),
                existing_pending=approval_items_by_call_id.get(call_id),
            )
        rejected_function_outputs.append(
            function_rejection_item(
                public_agent,
                tool_call,
                rejection_message=rejection_message,
                scope_id=tool_state_scope_id,
                tool_origin=get_function_tool_origin(function_tool),
            )
        )
        if isinstance(call_id, str):
            rejected_function_call_ids.add(call_id)

    async def _function_requires_approval(run: ToolRunFunction) -> bool:
        call_id = run.tool_call.call_id
        if call_id and call_id in approval_items_by_call_id:
            return True

        try:
            return await function_needs_approval(
                run.function_tool,
                context_wrapper,
                run.tool_call,
            )
        except UserError:
            raise
        except Exception:
            return True

    try:
        context_wrapper.turn_input = ItemHelpers.input_to_new_input_list(original_input)
    except Exception:
        context_wrapper.turn_input = []

    pending_approval_items = _pending_approvals_from_state()
    approval_items_by_call_id = index_approval_items_by_call_id(pending_approval_items)
    tool_state_scope_id = get_agent_tool_state_scope(context_wrapper)

    rejected_function_outputs: list[RunItem] = []
    rejected_function_call_ids: set[str] = set()
    rerun_function_call_ids: set[str] = set()
    pending_interruptions: list[ToolApprovalItem] = []
    pending_interruption_keys: set[str] = set()

    output_index = _build_tool_output_index(original_pre_step_items)

    def _has_output_item(call_id: str, expected_type: str) -> bool:
        return (expected_type, call_id) in output_index

    def _shell_call_id_from_run(run: ToolRunShellCall) -> str:
        return extract_shell_call_id(run.tool_call)

    def _apply_patch_call_id_from_run(run: ToolRunApplyPatchCall) -> str:
        return extract_apply_patch_call_id(run.tool_call)

    def _custom_call_id_from_run(run: ToolRunCustom) -> str:
        call_id = extract_tool_call_id(run.tool_call)
        if not call_id:
            raise ModelBehaviorError("Custom tool call is missing call_id.")
        return call_id

    def _computer_call_id_from_run(run: ToolRunComputerAction) -> str:
        call_id = extract_tool_call_id(run.tool_call)
        if not call_id:
            raise ModelBehaviorError("Computer action is missing call_id.")
        return call_id

    def _shell_tool_name(run: ToolRunShellCall) -> str:
        return run.shell_tool.name

    def _apply_patch_tool_name(run: ToolRunApplyPatchCall) -> str:
        return run.apply_patch_tool.name

    def _custom_tool_name(run: ToolRunCustom) -> str:
        return run.custom_tool.name

    async def _build_shell_rejection(run: ToolRunShellCall, call_id: str) -> RunItem:
        rejection_message = await resolve_approval_rejection_message(
            context_wrapper=context_wrapper,
            run_config=run_config,
            tool_type="shell",
            tool_name=run.shell_tool.name,
            call_id=call_id,
        )
        return cast(
            RunItem,
            shell_rejection_item(
                public_agent,
                call_id,
                rejection_message=rejection_message,
            ),
        )

    async def _build_apply_patch_rejection(run: ToolRunApplyPatchCall, call_id: str) -> RunItem:
        rejection_message = await resolve_approval_rejection_message(
            context_wrapper=context_wrapper,
            run_config=run_config,
            tool_type="apply_patch",
            tool_name=run.apply_patch_tool.name,
            call_id=call_id,
        )
        return cast(
            RunItem,
            apply_patch_rejection_item(
                public_agent,
                call_id,
                output_type="apply_patch_call_output",
                rejection_message=rejection_message,
            ),
        )

    async def _build_custom_rejection(run: ToolRunCustom, call_id: str) -> RunItem:
        rejection_message = await resolve_approval_rejection_message(
            context_wrapper=context_wrapper,
            run_config=run_config,
            tool_type="custom",
            tool_name=run.custom_tool.name,
            call_id=call_id,
        )
        return ToolCallOutputItem(
            agent=public_agent,
            output=rejection_message,
            raw_item=cast(
                Any,
                {
                    "type": "custom_tool_call_output",
                    "call_id": call_id,
                    "output": rejection_message,
                },
            ),
        )

    async def _shell_needs_approval(run: ToolRunShellCall) -> bool:
        shell_call = coerce_shell_call(run.tool_call)
        return await evaluate_needs_approval_setting(
            run.shell_tool.needs_approval,
            context_wrapper,
            shell_call.action,
            shell_call.call_id,
        )

    async def _apply_patch_needs_approval(run: ToolRunApplyPatchCall) -> bool:
        operations = coerce_apply_patch_operations(
            run.tool_call,
            context_wrapper=context_wrapper,
        )
        call_id = extract_apply_patch_call_id(run.tool_call)
        for operation in operations:
            if await evaluate_needs_approval_setting(
                run.apply_patch_tool.needs_approval, context_wrapper, operation, call_id
            ):
                return True
        return False

    async def _custom_tool_needs_approval(run: ToolRunCustom) -> bool:
        tool_input = get_mapping_or_attr(run.tool_call, "input")
        call_id = _custom_call_id_from_run(run)
        if not isinstance(tool_input, str):
            raise ModelBehaviorError("Custom tool call is missing input.")
        return await evaluate_needs_approval_setting(
            run.custom_tool.runtime_needs_approval(),
            context_wrapper,
            tool_input,
            call_id,
        )

    def _shell_output_exists(call_id: str) -> bool:
        return _has_output_item(call_id, "shell_call_output")

    def _apply_patch_output_exists(call_id: str) -> bool:
        return _has_output_item(call_id, "apply_patch_call_output")

    def _custom_tool_output_exists(call_id: str) -> bool:
        return _has_output_item(call_id, "custom_tool_call_output")

    def _computer_output_exists(call_id: str) -> bool:
        return _has_output_item(call_id, "computer_call_output")

    def _nested_interruptions_status(
        interruptions: Sequence[ToolApprovalItem],
    ) -> Literal["approved", "pending", "rejected"]:
        has_pending = False
        for interruption in interruptions:
            call_id = extract_tool_call_id(interruption.raw_item)
            if not call_id:
                has_pending = True
                continue
            status = context_wrapper.get_approval_status(
                interruption.tool_name or "",
                call_id,
                tool_namespace=interruption.tool_namespace,
                existing_pending=interruption,
            )
            if status is False:
                return "rejected"
            if status is None:
                has_pending = True
        return "pending" if has_pending else "approved"

    def _function_output_exists(run: ToolRunFunction) -> bool:
        call_id = extract_tool_call_id(run.tool_call)
        if not call_id:
            return False

        pending_run_result = peek_agent_tool_run_result(
            run.tool_call,
            scope_id=tool_state_scope_id,
        )
        if pending_run_result and getattr(pending_run_result, "interruptions", None):
            status = _nested_interruptions_status(pending_run_result.interruptions)
            if status in ("approved", "rejected"):
                rerun_function_call_ids.add(call_id)
                return False
            return True

        return _has_output_item(call_id, "function_call_output")

    def _add_pending_interruption(item: ToolApprovalItem | None) -> None:
        if item is None:
            return
        call_id = extract_tool_call_id(item.raw_item)
        key = call_id or f"raw:{id(item.raw_item)}"
        if key in pending_interruption_keys:
            return
        pending_interruption_keys.add(key)
        pending_interruptions.append(item)

    def _allow_legacy_name_agent_match() -> bool:
        schema_version = getattr(run_state, "_schema_version", None)
        if not isinstance(schema_version, str):
            return False
        try:
            version_parts = tuple(int(part) for part in schema_version.split("."))
        except ValueError:
            return False
        # Schema 1.6 and earlier only serialized approval owners by agent name. With duplicate-name
        # agents, deserialization can legitimately resolve the approval to a sibling instance, so
        # resume must accept a same-name match for those legacy snapshots. Schema 1.7+ persists
        # duplicate-name identities, so newer snapshots should continue requiring object identity.
        return version_parts < (1, 7)

    allow_legacy_name_agent_match = _allow_legacy_name_agent_match()

    def _approval_matches_agent(approval: ToolApprovalItem) -> bool:
        approval_agent = approval.agent
        if approval_agent is None:
            return False
        if approval_agent is public_agent:
            return True
        return allow_legacy_name_agent_match and approval_agent.name == public_agent.name

    available_function_tools = await resolve_enabled_function_tools(
        execution_agent,
        context_wrapper,
    )
    approval_rebuild_function_tools = available_function_tools
    if pending_approval_items and execution_agent.mcp_servers:
        approval_rebuild_function_tools = [
            tool
            for tool in await execution_agent.get_all_tools(context_wrapper)
            if isinstance(tool, FunctionTool)
        ]

    async def _rebuild_function_runs_from_approvals() -> list[ToolRunFunction]:
        if not pending_approval_items:
            return []
        tool_map = build_function_tool_lookup_map(approval_rebuild_function_tools)
        existing_pending_call_ids: set[str] = set()
        for existing_pending in pending_interruptions:
            if isinstance(existing_pending, ToolApprovalItem):
                existing_call_id = extract_tool_call_id(existing_pending.raw_item)
                if existing_call_id:
                    existing_pending_call_ids.add(existing_call_id)
        rebuilt_runs: list[ToolRunFunction] = []

        def _add_unmatched_pending(approval: ToolApprovalItem) -> None:
            call_id = extract_tool_call_id(approval.raw_item)
            if not call_id:
                _add_pending_interruption(approval)
                return
            tool_name = approval.tool_name or ""
            approval_status = context_wrapper.get_approval_status(
                tool_name,
                call_id,
                tool_namespace=approval.tool_namespace,
                existing_pending=approval,
            )
            if approval_status is None:
                _add_pending_interruption(approval)

        for approval in pending_approval_items:
            if not isinstance(approval, ToolApprovalItem):
                continue
            if not _approval_matches_agent(approval):
                _add_unmatched_pending(approval)
                continue
            raw = approval.raw_item
            raw_type = get_mapping_or_attr(raw, "type")
            if raw_type != "function_call":
                _add_unmatched_pending(approval)
                continue
            name = get_mapping_or_attr(raw, "name")
            namespace = get_tool_call_namespace(raw)
            if namespace is None and isinstance(approval.tool_namespace, str):
                namespace = approval.tool_namespace
            approval_key = getattr(approval, "tool_lookup_key", None)
            if approval_key is None:
                approval_key = get_function_tool_lookup_key(name, namespace)
            resolved_tool = tool_map.get(approval_key) if approval_key is not None else None
            if not (isinstance(name, str) and resolved_tool is not None):
                _add_unmatched_pending(approval)
                continue

            rebuilt_call_id: str | None
            arguments: str | None
            tool_call: ResponseFunctionToolCall
            if isinstance(raw, ResponseFunctionToolCall):
                rebuilt_call_id = raw.call_id
                arguments = raw.arguments
                tool_call = raw
            else:
                rebuilt_call_id = extract_tool_call_id(raw)
                arguments = get_mapping_or_attr(raw, "arguments") or "{}"
                status = get_mapping_or_attr(raw, "status")
                if not (isinstance(rebuilt_call_id, str) and isinstance(arguments, str)):
                    _add_unmatched_pending(approval)
                    continue
                valid_status: Literal["in_progress", "completed", "incomplete"] | None = None
                if isinstance(status, str) and status in (
                    "in_progress",
                    "completed",
                    "incomplete",
                ):
                    valid_status = status  # type: ignore[assignment]
                tool_call_payload: dict[str, Any] = {
                    "type": "function_call",
                    "name": name,
                    "call_id": rebuilt_call_id,
                    "arguments": arguments,
                    "status": valid_status,
                }
                if namespace is not None:
                    tool_call_payload["namespace"] = namespace
                tool_call = ResponseFunctionToolCall(**tool_call_payload)
            tool_call = cast(
                ResponseFunctionToolCall,
                normalize_tool_call_for_function_tool(tool_call, resolved_tool),
            )

            if not (isinstance(rebuilt_call_id, str) and isinstance(arguments, str)):
                _add_unmatched_pending(approval)
                continue

            approval_status = context_wrapper.get_approval_status(
                name,
                rebuilt_call_id,
                tool_namespace=namespace,
                existing_pending=approval,
            )
            if approval_status is False:
                await _record_function_rejection(
                    rebuilt_call_id,
                    tool_call,
                    resolved_tool,
                )
                continue
            if approval_status is None:
                if rebuilt_call_id not in existing_pending_call_ids:
                    _add_pending_interruption(approval)
                    existing_pending_call_ids.add(rebuilt_call_id)
                continue
            rebuilt_runs.append(ToolRunFunction(function_tool=resolved_tool, tool_call=tool_call))
        return rebuilt_runs

    function_tool_runs = await _select_function_tool_runs_for_resume(
        processed_response.functions,
        approval_items_by_call_id=approval_items_by_call_id,
        context_wrapper=context_wrapper,
        needs_approval_checker=_function_requires_approval,
        output_exists_checker=_function_output_exists,
        record_rejection=_record_function_rejection,
        pending_interruption_adder=_add_pending_interruption,
        pending_item_builder=lambda run: ToolApprovalItem(
            agent=public_agent,
            raw_item=run.tool_call,
            tool_name=run.function_tool.name,
            tool_namespace=get_tool_call_namespace(run.tool_call),
            tool_origin=get_function_tool_origin(run.function_tool),
            tool_lookup_key=get_function_tool_lookup_key_for_call(run.tool_call),
            _allow_bare_name_alias=should_allow_bare_name_approval_alias(
                run.function_tool,
                available_function_tools,
            ),
        ),
    )

    rebuilt_function_tool_runs = await _rebuild_function_runs_from_approvals()
    if rebuilt_function_tool_runs:
        existing_call_ids: set[str] = set()
        for run in function_tool_runs:
            call_id = extract_tool_call_id(run.tool_call)
            if call_id:
                existing_call_ids.add(call_id)
        for run in rebuilt_function_tool_runs:
            call_id = extract_tool_call_id(run.tool_call)
            if call_id and call_id in existing_call_ids:
                continue
            function_tool_runs.append(run)
            if call_id:
                existing_call_ids.add(call_id)

    pending_computer_actions: list[ToolRunComputerAction] = []
    for action in processed_response.computer_actions:
        call_id = _computer_call_id_from_run(action)
        if _computer_output_exists(call_id):
            continue
        pending_computer_actions.append(action)

    approved_shell_calls, rejected_shell_results = await _collect_runs_by_approval(
        processed_response.shell_calls,
        call_id_extractor=_shell_call_id_from_run,
        tool_name_resolver=_shell_tool_name,
        rejection_builder=_build_shell_rejection,
        context_wrapper=context_wrapper,
        approval_items_by_call_id=approval_items_by_call_id,
        agent=public_agent,
        pending_interruption_adder=_add_pending_interruption,
        needs_approval_checker=_shell_needs_approval,
        output_exists_checker=_shell_output_exists,
    )

    approved_apply_patch_calls, rejected_apply_patch_results = await _collect_runs_by_approval(
        processed_response.apply_patch_calls,
        call_id_extractor=_apply_patch_call_id_from_run,
        tool_name_resolver=_apply_patch_tool_name,
        rejection_builder=_build_apply_patch_rejection,
        context_wrapper=context_wrapper,
        approval_items_by_call_id=approval_items_by_call_id,
        agent=public_agent,
        pending_interruption_adder=_add_pending_interruption,
        needs_approval_checker=_apply_patch_needs_approval,
        output_exists_checker=_apply_patch_output_exists,
    )

    approved_custom_tool_calls, rejected_custom_tool_results = await _collect_runs_by_approval(
        processed_response.custom_tool_calls,
        call_id_extractor=_custom_call_id_from_run,
        tool_name_resolver=_custom_tool_name,
        rejection_builder=_build_custom_rejection,
        context_wrapper=context_wrapper,
        approval_items_by_call_id=approval_items_by_call_id,
        agent=public_agent,
        pending_interruption_adder=_add_pending_interruption,
        needs_approval_checker=_custom_tool_needs_approval,
        output_exists_checker=_custom_tool_output_exists,
    )

    plan = _build_plan_for_resume_turn(
        processed_response=processed_response,
        agent=public_agent,
        context_wrapper=context_wrapper,
        approval_items_by_call_id=approval_items_by_call_id,
        pending_interruptions=pending_interruptions,
        pending_interruption_adder=_add_pending_interruption,
        function_runs=function_tool_runs,
        computer_actions=pending_computer_actions,
        custom_tool_calls=approved_custom_tool_calls,
        shell_calls=approved_shell_calls,
        apply_patch_calls=approved_apply_patch_calls,
    )

    (
        function_results,
        tool_input_guardrail_results,
        tool_output_guardrail_results,
        computer_results,
        custom_tool_results,
        shell_results,
        apply_patch_results,
        _local_shell_results,
    ) = await _execute_tool_plan(
        plan=plan,
        bindings=bindings,
        hooks=hooks,
        context_wrapper=context_wrapper,
        run_config=run_config,
    )

    for interruption in _collect_tool_interruptions(
        function_results=function_results,
        custom_tool_results=custom_tool_results,
        shell_results=[],
        apply_patch_results=[],
    ):
        _add_pending_interruption(interruption)

    new_items, append_if_new = _make_unique_item_appender(original_pre_step_items)

    for item in _build_tool_result_items(
        function_results=function_results,
        computer_results=computer_results,
        custom_tool_results=custom_tool_results,
        shell_results=shell_results,
        apply_patch_results=apply_patch_results,
        local_shell_results=[],
    ):
        append_if_new(item)
    for rejection_item in rejected_function_outputs:
        append_if_new(rejection_item)
    for pending_item in pending_interruptions:
        if pending_item:
            append_if_new(pending_item)
    for shell_rejection in rejected_shell_results:
        append_if_new(shell_rejection)
    for custom_tool_rejection in rejected_custom_tool_results:
        append_if_new(custom_tool_rejection)
    for apply_patch_rejection in rejected_apply_patch_results:
        append_if_new(apply_patch_rejection)
    for approved_response in plan.approved_mcp_responses:
        append_if_new(approved_response)

    processed_response.interruptions = pending_interruptions
    if pending_interruptions:
        return SingleStepResult(
            original_input=original_input,
            model_response=new_response,
            pre_step_items=original_pre_step_items,
            new_step_items=new_items,
            next_step=NextStepInterruption(
                interruptions=[item for item in pending_interruptions if item]
            ),
            tool_input_guardrail_results=tool_input_guardrail_results,
            tool_output_guardrail_results=tool_output_guardrail_results,
            processed_response=processed_response,
        )

    await _append_mcp_callback_results(
        agent=public_agent,
        requests=plan.mcp_requests_with_callback,
        context_wrapper=context_wrapper,
        append_item=append_if_new,
    )

    (
        pending_hosted_mcp_approvals,
        pending_hosted_mcp_approval_ids,
    ) = process_hosted_mcp_approvals(
        original_pre_step_items=original_pre_step_items,
        mcp_approval_requests=processed_response.mcp_approval_requests,
        context_wrapper=context_wrapper,
        agent=public_agent,
        append_item=append_if_new,
    )

    pre_step_items = [
        item
        for item in original_pre_step_items
        if should_keep_hosted_mcp_item(
            item,
            pending_hosted_mcp_approvals=pending_hosted_mcp_approvals,
            pending_hosted_mcp_approval_ids=pending_hosted_mcp_approval_ids,
        )
    ]

    if rejected_function_call_ids:
        pre_step_items = [
            item
            for item in pre_step_items
            if not (
                item.type == "tool_call_output_item"
                and (
                    extract_tool_call_id(getattr(item, "raw_item", None))
                    in rejected_function_call_ids
                )
            )
        ]

    if rerun_function_call_ids:
        pre_step_items = [
            item
            for item in pre_step_items
            if not (
                item.type == "tool_call_output_item"
                and (
                    extract_tool_call_id(getattr(item, "raw_item", None)) in rerun_function_call_ids
                )
            )
        ]

    executed_handoff_call_ids: set[str] = set()
    for item in original_pre_step_items:
        if isinstance(item, HandoffCallItem):
            handoff_call_id = extract_tool_call_id(item.raw_item)
            if handoff_call_id:
                executed_handoff_call_ids.add(handoff_call_id)

    pending_handoffs = [
        handoff
        for handoff in processed_response.handoffs
        if not handoff.tool_call.call_id
        or handoff.tool_call.call_id not in executed_handoff_call_ids
    ]

    if pending_handoffs:
        return await execute_handoffs_call(
            public_agent=public_agent,
            original_input=original_input,
            pre_step_items=pre_step_items,
            new_step_items=new_items,
            new_response=new_response,
            run_handoffs=pending_handoffs,
            hooks=hooks,
            context_wrapper=context_wrapper,
            run_config=run_config,
            server_manages_conversation=server_manages_conversation,
            nest_handoff_history_fn=nest_history,
        )

    tool_final_output = await _maybe_finalize_from_tool_results(
        public_agent=public_agent,
        original_input=original_input,
        new_response=new_response,
        pre_step_items=pre_step_items,
        new_step_items=new_items,
        function_results=function_results,
        hooks=hooks,
        context_wrapper=context_wrapper,
        tool_input_guardrail_results=tool_input_guardrail_results,
        tool_output_guardrail_results=tool_output_guardrail_results,
    )
    if tool_final_output is not None:
        return tool_final_output

    return SingleStepResult(
        original_input=original_input,
        model_response=new_response,
        pre_step_items=pre_step_items,
        new_step_items=new_items,
        next_step=NextStepRunAgain(),
        tool_input_guardrail_results=tool_input_guardrail_results,
        tool_output_guardrail_results=tool_output_guardrail_results,
    )


def process_model_response(
    *,
    agent: Agent[Any],
    all_tools: list[Tool],
    response: ModelResponse,
    output_schema: AgentOutputSchemaBase | None,
    handoffs: list[Handoff],
    existing_items: Sequence[RunItem] | None = None,
) -> ProcessedResponse:
    items: list[RunItem] = []

    run_handoffs = []
    functions = []
    computer_actions = []
    custom_tool_calls = []
    local_shell_calls = []
    shell_calls = []
    apply_patch_calls = []
    mcp_approval_requests = []
    tools_used: list[str] = []
    handoff_map = {handoff.tool_name: handoff for handoff in handoffs}
    function_map = build_function_tool_lookup_map(
        [tool for tool in all_tools if isinstance(tool, FunctionTool)]
    )
    custom_tool_map = {tool.name: tool for tool in all_tools if isinstance(tool, CustomTool)}
    computer_tool = next((tool for tool in all_tools if isinstance(tool, ComputerTool)), None)
    local_shell_tool = next((tool for tool in all_tools if isinstance(tool, LocalShellTool)), None)
    shell_tool = next((tool for tool in all_tools if isinstance(tool, ShellTool)), None)
    apply_patch_tool = next((tool for tool in all_tools if isinstance(tool, ApplyPatchTool)), None)
    hosted_mcp_server_map = {
        tool.tool_config["server_label"]: tool
        for tool in all_tools
        if isinstance(tool, HostedMCPTool)
    }
    hosted_mcp_tool_metadata = collect_mcp_list_tools_metadata(existing_items or ())
    hosted_mcp_tool_metadata.update(collect_mcp_list_tools_metadata(response.output))

    def _dump_output_item(raw_item: Any) -> dict[str, Any]:
        if isinstance(raw_item, dict):
            return dict(raw_item)
        if hasattr(raw_item, "model_dump"):
            dumped = cast(Any, raw_item).model_dump(exclude_unset=True)
            if isinstance(dumped, Mapping):
                return dict(dumped)
            return {"type": get_mapping_or_attr(raw_item, "type")}
        return {
            "type": get_mapping_or_attr(raw_item, "type"),
            "id": get_mapping_or_attr(raw_item, "id"),
        }

    for output in response.output:
        output_type = get_mapping_or_attr(output, "type")
        logger.debug(
            "Processing output item type=%s class=%s",
            output_type,
            output.__class__.__name__ if hasattr(output, "__class__") else type(output),
        )
        if output_type == "shell_call":
            if isinstance(output, dict):
                shell_call_raw = dict(output)
            elif hasattr(output, "model_dump"):
                shell_call_raw = cast(Any, output).model_dump(exclude_unset=True)
            else:
                shell_call_raw = {
                    "type": "shell_call",
                    "id": get_mapping_or_attr(output, "id"),
                    "call_id": get_mapping_or_attr(output, "call_id"),
                    "status": get_mapping_or_attr(output, "status"),
                    "action": get_mapping_or_attr(output, "action"),
                    "environment": get_mapping_or_attr(output, "environment"),
                    "created_by": get_mapping_or_attr(output, "created_by"),
                }
            shell_call_raw.pop("created_by", None)
            items.append(ToolCallItem(raw_item=cast(Any, shell_call_raw), agent=agent))
            if not shell_tool:
                tools_used.append("shell")
                _error_tracing.attach_error_to_current_span(
                    SpanError(
                        message="Shell tool not found",
                        data={},
                    )
                )
                raise ModelBehaviorError("Model produced shell call without a shell tool.")
            tools_used.append(shell_tool.name)
            shell_environment = shell_tool.environment
            if shell_environment is None or shell_environment["type"] != "local":
                logger.debug(
                    "Skipping local shell execution for hosted shell tool %s", shell_tool.name
                )
                continue
            if shell_tool.executor is None:
                _error_tracing.attach_error_to_current_span(
                    SpanError(
                        message="Local shell executor not found",
                        data={},
                    )
                )
                raise ModelBehaviorError(
                    "Model produced local shell call without a local shell executor."
                )
            call_identifier = get_mapping_or_attr(output, "call_id")
            logger.debug("Queuing shell_call %s", call_identifier)
            shell_calls.append(ToolRunShellCall(tool_call=output, shell_tool=shell_tool))
            continue
        if output_type == "shell_call_output" and isinstance(
            output, dict | ResponseFunctionShellToolCallOutput
        ):
            tools_used.append(shell_tool.name if shell_tool else "shell")
            if isinstance(output, dict):
                shell_output_raw = dict(output)
            else:
                shell_output_raw = output.model_dump(exclude_unset=True)
            shell_output_raw.pop("created_by", None)
            shell_outputs = shell_output_raw.get("output")
            if isinstance(shell_outputs, list):
                for shell_output in shell_outputs:
                    if isinstance(shell_output, dict):
                        shell_output.pop("created_by", None)
            items.append(
                ToolCallOutputItem(
                    raw_item=cast(Any, shell_output_raw),
                    output=shell_output_raw.get("output"),
                    agent=agent,
                )
            )
            continue
        if output_type == "apply_patch_call":
            if isinstance(output, dict):
                apply_patch_call_raw = dict(output)
            elif hasattr(output, "model_dump"):
                apply_patch_call_raw = cast(Any, output).model_dump(exclude_unset=True)
            else:
                apply_patch_call_raw = {
                    "type": "apply_patch_call",
                    "id": get_mapping_or_attr(output, "id"),
                    "call_id": get_mapping_or_attr(output, "call_id"),
                    "status": get_mapping_or_attr(output, "status"),
                    "operation": get_mapping_or_attr(output, "operation"),
                    "created_by": get_mapping_or_attr(output, "created_by"),
                }
            apply_patch_call_raw.pop("created_by", None)
            items.append(ToolCallItem(raw_item=cast(Any, apply_patch_call_raw), agent=agent))
            if apply_patch_tool:
                tools_used.append(apply_patch_tool.name)
                call_identifier = get_mapping_or_attr(apply_patch_call_raw, "call_id")
                logger.debug("Queuing apply_patch_call %s", call_identifier)
                apply_patch_calls.append(
                    ToolRunApplyPatchCall(
                        tool_call=apply_patch_call_raw,
                        apply_patch_tool=apply_patch_tool,
                    )
                )
            else:
                tools_used.append("apply_patch")
                _error_tracing.attach_error_to_current_span(
                    SpanError(
                        message="Apply patch tool not found",
                        data={},
                    )
                )
                raise ModelBehaviorError(
                    "Model produced apply_patch call without an apply_patch tool."
                )
            continue
        if output_type == "compaction":
            if isinstance(output, dict):
                compaction_raw = dict(output)
            elif isinstance(output, ResponseCompactionItem):
                compaction_raw = output.model_dump(exclude_unset=True)
            else:
                logger.warning("Unexpected compaction output type, ignoring: %s", type(output))
                continue
            compaction_raw.pop("created_by", None)
            items.append(
                CompactionItem(agent=agent, raw_item=cast(TResponseInputItem, compaction_raw))
            )
            continue
        if output_type == "tool_search_call":
            tool_search_call_raw = coerce_tool_search_call_raw_item(output)
            if get_mapping_or_attr(tool_search_call_raw, "execution") == "client":
                raise ModelBehaviorError(
                    "Client-executed tool_search calls are not supported by the standard "
                    "agent runner. Handle the tool_search_call yourself and return a matching "
                    "tool_search_output item with the same call_id."
                )
            items.append(ToolSearchCallItem(raw_item=tool_search_call_raw, agent=agent))
            tools_used.append("tool_search")
            continue
        if output_type == "tool_search_output":
            items.append(
                ToolSearchOutputItem(
                    raw_item=coerce_tool_search_output_raw_item(output),
                    agent=agent,
                )
            )
            tools_used.append("tool_search")
            continue
        if isinstance(output, ResponseOutputMessage):
            items.append(MessageOutputItem(raw_item=output, agent=agent))
        elif isinstance(output, ResponseFileSearchToolCall):
            items.append(ToolCallItem(raw_item=output, agent=agent))
            tools_used.append("file_search")
        elif isinstance(output, ResponseFunctionWebSearch):
            items.append(ToolCallItem(raw_item=output, agent=agent))
            tools_used.append("web_search")
        elif isinstance(output, ResponseReasoningItem):
            items.append(ReasoningItem(raw_item=output, agent=agent))
        elif isinstance(output, ResponseComputerToolCall):
            items.append(ToolCallItem(raw_item=output, agent=agent))
            if not computer_tool:
                tools_used.append("computer")
                _error_tracing.attach_error_to_current_span(
                    SpanError(
                        message="Computer tool not found",
                        data={},
                    )
                )
                raise ModelBehaviorError("Model produced computer action without a computer tool.")
            tools_used.append(computer_tool.name)
            computer_actions.append(
                ToolRunComputerAction(tool_call=output, computer_tool=computer_tool)
            )
        elif isinstance(output, McpApprovalRequest):
            items.append(MCPApprovalRequestItem(raw_item=output, agent=agent))
            if output.server_label not in hosted_mcp_server_map:
                _error_tracing.attach_error_to_current_span(
                    SpanError(
                        message="MCP server label not found",
                        data={"server_label": output.server_label},
                    )
                )
                raise ModelBehaviorError(f"MCP server label {output.server_label} not found")
            server = hosted_mcp_server_map[output.server_label]
            mcp_approval_requests.append(
                ToolRunMCPApprovalRequest(
                    request_item=output,
                    mcp_tool=server,
                )
            )
            if not server.on_approval_request:
                logger.debug(
                    "Hosted MCP server %s has no on_approval_request hook; approvals will be "
                    "surfaced as interruptions for the caller to handle.",
                    output.server_label,
                )
        elif isinstance(output, McpListTools):
            items.append(MCPListToolsItem(raw_item=output, agent=agent))
        elif isinstance(output, McpCall):
            metadata = hosted_mcp_tool_metadata.get((output.server_label, output.name))
            items.append(
                ToolCallItem(
                    raw_item=output,
                    agent=agent,
                    description=metadata.description if metadata is not None else None,
                    title=metadata.title if metadata is not None else None,
                    tool_origin=ToolOrigin(
                        type=ToolOriginType.MCP,
                        mcp_server_name=output.server_label,
                    ),
                )
            )
            tools_used.append("mcp")
        elif isinstance(output, ImageGenerationCall):
            items.append(ToolCallItem(raw_item=output, agent=agent))
            tools_used.append("image_generation")
        elif isinstance(output, ResponseCodeInterpreterToolCall):
            items.append(ToolCallItem(raw_item=output, agent=agent))
            tools_used.append("code_interpreter")
        elif isinstance(output, LocalShellCall):
            items.append(ToolCallItem(raw_item=output, agent=agent))
            if local_shell_tool:
                tools_used.append("local_shell")
                local_shell_calls.append(
                    ToolRunLocalShellCall(tool_call=output, local_shell_tool=local_shell_tool)
                )
            elif shell_tool:
                tools_used.append(shell_tool.name)
                shell_calls.append(ToolRunShellCall(tool_call=output, shell_tool=shell_tool))
            else:
                tools_used.append("local_shell")
                _error_tracing.attach_error_to_current_span(
                    SpanError(
                        message="Local shell tool not found",
                        data={},
                    )
                )
                raise ModelBehaviorError(
                    "Model produced local shell call without a local shell tool."
                )
        elif isinstance(output, ResponseCustomToolCall):
            custom_tool = custom_tool_map.get(output.name)
            if custom_tool is not None:
                items.append(ToolCallItem(raw_item=cast(Any, output), agent=agent))
                tools_used.append(custom_tool.name)
                custom_tool_calls.append(ToolRunCustom(tool_call=output, custom_tool=custom_tool))
            elif is_apply_patch_name(output.name, apply_patch_tool):
                parsed_operation = parse_apply_patch_custom_input(output.input)
                pseudo_call = {
                    "type": "apply_patch_call",
                    "call_id": output.call_id,
                    **parsed_operation,
                }
                items.append(ToolCallItem(raw_item=cast(Any, pseudo_call), agent=agent))
                if apply_patch_tool:
                    tools_used.append(apply_patch_tool.name)
                    apply_patch_calls.append(
                        ToolRunApplyPatchCall(
                            tool_call=pseudo_call,
                            apply_patch_tool=apply_patch_tool,
                        )
                    )
                else:
                    tools_used.append("apply_patch")
                    _error_tracing.attach_error_to_current_span(
                        SpanError(
                            message="Apply patch tool not found",
                            data={},
                        )
                    )
                    raise ModelBehaviorError(
                        "Model produced apply_patch call without an apply_patch tool."
                    )
            else:
                items.append(ToolCallItem(raw_item=cast(Any, output), agent=agent))
                _error_tracing.attach_error_to_current_span(
                    SpanError(
                        message="Custom tool not found",
                        data={"tool_name": output.name},
                    )
                )
                raise ModelBehaviorError(f"Tool {output.name} not found in agent {agent.name}")
        elif (
            isinstance(output, ResponseFunctionToolCall)
            and is_apply_patch_name(output.name, apply_patch_tool)
            and get_function_tool_lookup_key_for_call(output) not in function_map
        ):
            parsed_operation = parse_apply_patch_function_args(output.arguments)
            pseudo_call = {
                "type": "apply_patch_call",
                "call_id": output.call_id,
                "operation": parsed_operation,
            }
            items.append(ToolCallItem(raw_item=cast(Any, pseudo_call), agent=agent))
            if apply_patch_tool:
                tools_used.append(apply_patch_tool.name)
                apply_patch_calls.append(
                    ToolRunApplyPatchCall(tool_call=pseudo_call, apply_patch_tool=apply_patch_tool)
                )
            else:
                tools_used.append("apply_patch")
                _error_tracing.attach_error_to_current_span(
                    SpanError(
                        message="Apply patch tool not found",
                        data={},
                    )
                )
                raise ModelBehaviorError(
                    "Model produced apply_patch call without an apply_patch tool."
                )
            continue

        elif not isinstance(output, ResponseFunctionToolCall):
            logger.warning("Unexpected output type, ignoring: %s", type(output))
            continue

        if not isinstance(output, ResponseFunctionToolCall):
            continue

        tools_used.append(get_tool_call_trace_name(output) or output.name)
        qualified_output_name = get_tool_call_qualified_name(output)

        if qualified_output_name == output.name and output.name in handoff_map:
            items.append(HandoffCallItem(raw_item=output, agent=agent))
            handoff = ToolRunHandoff(
                tool_call=output,
                handoff=handoff_map[output.name],
            )
            run_handoffs.append(handoff)
        else:
            lookup_key = get_function_tool_lookup_key_for_call(output)
            func_tool = function_map.get(lookup_key) if lookup_key is not None else None
            if func_tool is None:
                if output_schema is not None and output.name == "json_tool_call":
                    synthetic_tool = build_litellm_json_tool_call(output)
                    items.append(
                        ToolCallItem(
                            raw_item=output,
                            agent=agent,
                            description=synthetic_tool.description,
                            tool_origin=get_function_tool_origin(synthetic_tool),
                        )
                    )
                    functions.append(
                        ToolRunFunction(
                            tool_call=output,
                            function_tool=synthetic_tool,
                        )
                    )
                    continue
                _error_tracing.attach_error_to_current_span(
                    SpanError(
                        message="Tool not found",
                        data={"tool_name": qualified_output_name or output.name},
                    )
                )
                error = (
                    f"Tool {qualified_output_name or output.name} not found in agent {agent.name}"
                )
                raise ModelBehaviorError(error)

            items.append(
                ToolCallItem(
                    raw_item=output,
                    agent=agent,
                    description=func_tool.description,
                    title=func_tool._mcp_title,
                    tool_origin=get_function_tool_origin(func_tool),
                )
            )
            functions.append(
                ToolRunFunction(
                    tool_call=output,
                    function_tool=func_tool,
                )
            )

    return ProcessedResponse(
        new_items=items,
        handoffs=run_handoffs,
        functions=functions,
        computer_actions=computer_actions,
        custom_tool_calls=custom_tool_calls,
        local_shell_calls=local_shell_calls,
        shell_calls=shell_calls,
        apply_patch_calls=apply_patch_calls,
        tools_used=tools_used,
        mcp_approval_requests=mcp_approval_requests,
        interruptions=[],
    )


async def get_single_step_result_from_response(
    *,
    bindings: AgentBindings[TContext],
    all_tools: list[Tool],
    original_input: str | list[TResponseInputItem],
    pre_step_items: list[RunItem],
    new_response: ModelResponse,
    output_schema: AgentOutputSchemaBase | None,
    handoffs: list[Handoff],
    hooks: RunHooks[TContext],
    context_wrapper: RunContextWrapper[TContext],
    run_config: RunConfig,
    tool_use_tracker,
    server_manages_conversation: bool = False,
    event_queue: asyncio.Queue[StreamEvent | QueueCompleteSentinel] | None = None,
    before_side_effects: Callable[[], Awaitable[None]] | None = None,
) -> SingleStepResult:
    item_agent = bindings.public_agent
    processed_response = process_model_response(
        agent=item_agent,
        all_tools=all_tools,
        response=new_response,
        output_schema=output_schema,
        handoffs=handoffs,
        existing_items=pre_step_items,
    )

    if before_side_effects is not None:
        await before_side_effects()

    tool_use_tracker.record_processed_response(item_agent, processed_response)

    # Expose the conversation history so tools can inspect prior items via
    # ToolContext.conversation_history (addresses #904).
    context_wrapper._generated_items = list(pre_step_items)

    if event_queue is not None and processed_response.new_items:
        handoff_items = [
            item for item in processed_response.new_items if isinstance(item, HandoffCallItem)
        ]
        if handoff_items:
            stream_step_items_to_queue(cast(list[RunItem], handoff_items), event_queue)

    return await execute_tools_and_side_effects(
        bindings=bindings,
        original_input=original_input,
        pre_step_items=pre_step_items,
        new_response=new_response,
        processed_response=processed_response,
        output_schema=output_schema,
        hooks=hooks,
        context_wrapper=context_wrapper,
        run_config=run_config,
        server_manages_conversation=server_manages_conversation,
    )