"""
Tool execution helpers for the run pipeline. This module hosts execution-time helpers,
approval plumbing, and payload coercion. Action classes live in tool_actions.py.
"""

from __future__ import annotations

import asyncio
import copy
import dataclasses
import functools
import inspect
import json
from collections.abc import Awaitable, Callable, Mapping, Sequence
from typing import TYPE_CHECKING, Any, Literal, TypeVar, cast

from openai.types.responses import ResponseFunctionToolCall
from openai.types.responses.response_input_item_param import (
    ComputerCallOutputAcknowledgedSafetyCheck,
)
from openai.types.responses.response_input_param import McpApprovalResponse
from openai.types.responses.response_output_item import McpApprovalRequest

from .._tool_identity import (
    FunctionToolLookupKey,
    NamedToolLookupKey,
    build_function_tool_lookup_map,
    get_function_tool_lookup_key,
    get_function_tool_lookup_key_for_call,
    get_function_tool_trace_name,
    get_tool_call_namespace,
    get_tool_call_trace_name,
    is_deferred_top_level_function_tool,
    normalize_tool_call_for_function_tool,
    should_allow_bare_name_approval_alias,
    tool_trace_name,
)
from ..agent import Agent
from ..agent_tool_state import (
    consume_agent_tool_run_result,
    get_agent_tool_state_scope,
    peek_agent_tool_run_result,
)
from ..editor import ApplyPatchOperation, ApplyPatchResult
from ..exceptions import (
    AgentsException,
    ModelBehaviorError,
    ToolInputGuardrailTripwireTriggered,
    ToolOutputGuardrailTripwireTriggered,
    UserError,
)
from ..items import (
    ItemHelpers,
    MCPApprovalResponseItem,
    RunItem,
    RunItemBase,
    ToolApprovalItem,
    ToolCallOutputItem,
)
from ..logger import logger
from ..model_settings import ModelSettings
from ..run_config import RunConfig, ToolErrorFormatterArgs
from ..run_context import RunContextWrapper
from ..tool import (
    ApplyPatchTool,
    ComputerTool,
    ComputerToolSafetyCheckData,
    FunctionTool,
    FunctionToolCustomDataContext,
    FunctionToolResult,
    ShellActionRequest,
    ShellCallData,
    ShellCallOutcome,
    ShellCommandOutput,
    Tool,
    ToolOrigin,
    get_function_tool_origin,
    invoke_function_tool,
    maybe_invoke_function_tool_failure_error_function,
    resolve_computer,
)
from ..tool_context import ToolContext
from ..tool_guardrails import (
    ToolInputGuardrailData,
    ToolInputGuardrailResult,
    ToolOutputGuardrailData,
    ToolOutputGuardrailResult,
)
from ..tracing import Span, SpanError, function_span, get_current_trace
from ..util import _coro, _error_tracing
from ..util._approvals import evaluate_needs_approval_setting
from ..util._custom_data import maybe_extract_custom_data, merge_custom_data
from ..util._tool_errors import get_trace_tool_error
from ..util._types import MaybeAwaitable
from ._asyncio_progress import get_function_tool_task_progress_deadline
from .agent_bindings import AgentBindings, bind_public_agent
from .approvals import append_approval_error_output
from .items import (
    REJECTION_MESSAGE,
    extract_mcp_request_id,
    extract_mcp_request_id_from_run,
    function_rejection_item,
)
from .run_steps import ToolRunFunction
from .tool_use_tracker import AgentToolUseTracker

if TYPE_CHECKING:
    from ..lifecycle import RunHooks
    from .run_steps import (
        ToolRunApplyPatchCall,
        ToolRunComputerAction,
        ToolRunCustom,
        ToolRunFunction,
        ToolRunLocalShellCall,
        ToolRunShellCall,
    )

__all__ = [
    "maybe_reset_tool_choice",
    "initialize_computer_tools",
    "extract_tool_call_id",
    "coerce_shell_call",
    "parse_apply_patch_custom_input",
    "parse_apply_patch_function_args",
    "extract_apply_patch_call_id",
    "coerce_apply_patch_operation",
    "coerce_apply_patch_operations",
    "normalize_apply_patch_result",
    "is_apply_patch_name",
    "normalize_shell_output",
    "serialize_shell_output",
    "resolve_exit_code",
    "render_shell_outputs",
    "truncate_shell_outputs",
    "normalize_max_output_length",
    "normalize_shell_output_entries",
    "format_shell_error",
    "get_trace_tool_error",
    "with_tool_function_span",
    "build_litellm_json_tool_call",
    "process_hosted_mcp_approvals",
    "collect_manual_mcp_approvals",
    "index_approval_items_by_call_id",
    "should_keep_hosted_mcp_item",
    "resolve_approval_status",
    "resolve_approval_interruption",
    "resolve_approval_rejection_message",
    "function_needs_approval",
    "resolve_enabled_function_tools",
    "execute_function_tool_calls",
    "execute_custom_tool_calls",
    "execute_local_shell_calls",
    "execute_shell_calls",
    "execute_apply_patch_calls",
    "execute_computer_actions",
    "execute_approved_tools",
]

TToolSpanResult = TypeVar("TToolSpanResult")
_FUNCTION_TOOL_CANCELLED_DRAIN_SECONDS = 0.25
_FUNCTION_TOOL_CANCELLED_IMMEDIATE_STEP_LIMIT = 64
_FUNCTION_TOOL_POST_INVOKE_WAIT_SECONDS = 0.1


_FunctionToolFailureSource = Literal["direct", "cancelled_teardown", "post_invoke"]
_FunctionToolSettlementWaiter = Callable[
    [set[asyncio.Task[Any]], asyncio.AbstractEventLoop, float],
    Awaitable[bool],
]
_FunctionToolBackgroundExceptionMessage = Callable[[BaseException], str | None]


@dataclasses.dataclass(frozen=True)
class _FunctionToolFailure:
    """A function-tool failure with ordering metadata for arbitration."""

    error: BaseException
    order: int
    source: _FunctionToolFailureSource = "direct"


@dataclasses.dataclass
class _FunctionToolTaskState:
    """Mutable execution state tracked for each function-tool task in a batch."""

    tool_run: ToolRunFunction
    order: int
    invoke_task: asyncio.Task[Any] | None = None
    in_post_invoke_phase: bool = False


def _background_cleanup_task_exception_message(exc: BaseException) -> str | None:
    """Return the loop-level message for late sibling-cleanup failures."""
    if isinstance(exc, asyncio.CancelledError):
        return None
    if isinstance(exc, Exception):
        return (
            "Background function tool task raised during cancellation cleanup after failure "
            "propagation."
        )
    return "Background function tool task raised a fatal exception."


def _background_post_invoke_task_exception_message(exc: BaseException) -> str | None:
    """Return the loop-level message for late post-invoke failures."""
    del exc
    return "Background function tool post-invoke task raised after failure propagation."


def _parent_cancelled_task_exception_message(exc: BaseException) -> str | None:
    """Return the loop-level message for detached tasks after parent cancellation."""
    if isinstance(exc, Exception):
        return None
    return "Background function tool task raised a fatal exception."


def _consume_function_tool_task_result(
    task: asyncio.Task[Any],
    *,
    message_for_exception: _FunctionToolBackgroundExceptionMessage,
) -> None:
    """Report background task failures according to the provided reporting policy."""
    if task.cancelled():
        return

    exc = task.exception()
    if exc is None:
        return

    message = message_for_exception(exc)
    if message is None:
        return

    task.get_loop().call_exception_handler(
        {
            "message": message,
            "exception": exc,
            "task": task,
        }
    )


def _get_function_tool_failure_priority(error: BaseException) -> int:
    """Return the precedence used to arbitrate concurrent function-tool failures."""
    if isinstance(error, asyncio.CancelledError):
        return 0
    if isinstance(error, Exception):
        return 1
    return 2


def _select_function_tool_failure(
    current_failure: _FunctionToolFailure | None,
    new_failure: _FunctionToolFailure | None,
) -> _FunctionToolFailure | None:
    """Keep the highest-priority failure, breaking ties by tool call order."""
    if current_failure is None:
        return new_failure
    if new_failure is None:
        return current_failure

    current_priority = _get_function_tool_failure_priority(current_failure.error)
    new_priority = _get_function_tool_failure_priority(new_failure.error)
    if new_priority > current_priority:
        return new_failure
    if new_priority == current_priority and new_failure.order < current_failure.order:
        return new_failure
    return current_failure


def _merge_late_function_tool_failure(
    current_failure: _FunctionToolFailure | None,
    late_failure: _FunctionToolFailure | None,
) -> _FunctionToolFailure | None:
    """Merge a late failure into the triggering failure without masking the root cause."""
    if current_failure is None:
        return late_failure
    if late_failure is None:
        return current_failure

    current_priority = _get_function_tool_failure_priority(current_failure.error)
    late_priority = _get_function_tool_failure_priority(late_failure.error)
    if late_priority > current_priority:
        return late_failure
    if late_priority < current_priority:
        return current_failure
    if late_failure.source == "post_invoke" and current_failure.source != "post_invoke":
        return late_failure
    return current_failure


def _cancel_function_tool_tasks(tasks: set[asyncio.Task[Any]]) -> None:
    """Cancel sibling function-tool tasks."""
    for task in tasks:
        task.cancel()


def _attach_function_tool_task_result_callbacks(
    tasks: set[asyncio.Task[Any]],
    *,
    message_for_exception: _FunctionToolBackgroundExceptionMessage,
) -> None:
    """Attach a shared loop-level reporter to a set of background function-tool tasks."""
    callback = functools.partial(
        _consume_function_tool_task_result,
        message_for_exception=message_for_exception,
    )
    for task in tasks:
        task.add_done_callback(callback)


def _record_completed_function_tool_tasks(
    *,
    completed_tasks: Sequence[asyncio.Task[Any]],
    task_states: Mapping[asyncio.Task[Any], _FunctionToolTaskState],
    results_by_tool_run: dict[int, Any],
    failure_sources_by_task: Mapping[asyncio.Task[Any], _FunctionToolFailureSource] | None = None,
    ignore_cancelled_tasks: set[asyncio.Task[Any]] | None = None,
) -> _FunctionToolFailure | None:
    """Store finished task results and return the preferred failure, if any."""
    failure: _FunctionToolFailure | None = None
    ordered_done_tasks = sorted(completed_tasks, key=lambda task: task_states[task].order)
    ignored_tasks = ignore_cancelled_tasks or set()
    failure_sources = failure_sources_by_task or {}
    for task in ordered_done_tasks:
        task_state = task_states[task]
        tool_run = task_state.tool_run
        try:
            results_by_tool_run[id(tool_run)] = task.result()
        except BaseException as exc:
            if task in ignored_tasks and isinstance(exc, asyncio.CancelledError):
                continue
            failure = _select_function_tool_failure(
                failure,
                _FunctionToolFailure(
                    error=exc,
                    order=task_state.order,
                    source=failure_sources.get(task, "direct"),
                ),
            )
    return failure


def _collect_settled_function_tool_tasks(
    *,
    remaining_tasks: set[asyncio.Task[Any]],
    task_states: Mapping[asyncio.Task[Any], _FunctionToolTaskState],
    results_by_tool_run: dict[int, Any],
    failure_sources_by_task: Mapping[asyncio.Task[Any], _FunctionToolFailureSource] | None = None,
    ignore_cancelled_tasks: set[asyncio.Task[Any]] | None = None,
) -> tuple[_FunctionToolFailure | None, set[asyncio.Task[Any]]]:
    """Remove completed tasks from the pending set and record their outcomes."""
    settled_tasks = {task for task in remaining_tasks if task.done()}
    if not settled_tasks:
        return None, remaining_tasks

    new_failure = _record_completed_function_tool_tasks(
        completed_tasks=list(settled_tasks),
        task_states=task_states,
        results_by_tool_run=results_by_tool_run,
        failure_sources_by_task=failure_sources_by_task,
        ignore_cancelled_tasks=ignore_cancelled_tasks,
    )
    return new_failure, remaining_tasks - settled_tasks


async def _wait_for_cancelled_function_tool_task_progress(
    remaining_tasks: set[asyncio.Task[Any]],
    loop: asyncio.AbstractEventLoop,
    remaining_time: float,
    *,
    task_states: Mapping[asyncio.Task[Any], _FunctionToolTaskState],
) -> tuple[bool, bool]:
    """Wait until a cancelled sibling can make another self-driven step."""
    task_to_invoke_task = {
        tracked_task: task_state.invoke_task
        for tracked_task, task_state in task_states.items()
        if task_state.invoke_task is not None
    }
    progress_deadlines = {
        task: get_function_tool_task_progress_deadline(
            task=task,
            task_to_invoke_task=task_to_invoke_task,
            loop=loop,
        )
        for task in remaining_tasks
    }
    self_progressing_tasks = {
        task: deadline for task, deadline in progress_deadlines.items() if deadline is not None
    }
    if not self_progressing_tasks:
        return False, False

    now = loop.time()
    next_deadline = min(self_progressing_tasks.values())
    delay = max(0.0, next_deadline - now)
    if delay > 0:
        await asyncio.wait(
            set(self_progressing_tasks),
            timeout=min(delay, remaining_time),
            return_when=asyncio.FIRST_COMPLETED,
        )
        return True, False

    await asyncio.sleep(0)
    return True, True


async def _wait_for_function_tool_task_completion(
    remaining_tasks: set[asyncio.Task[Any]],
    _loop: asyncio.AbstractEventLoop,
    remaining_time: float,
) -> bool:
    """Wait briefly for a pending task to finish without forcing cancellation."""
    done_tasks, _ = await asyncio.wait(
        remaining_tasks,
        timeout=remaining_time,
        return_when=asyncio.FIRST_COMPLETED,
    )
    return bool(done_tasks)


async def _settle_pending_function_tool_tasks(
    *,
    pending_tasks: set[asyncio.Task[Any]],
    task_states: Mapping[asyncio.Task[Any], _FunctionToolTaskState],
    results_by_tool_run: dict[int, Any],
    timeout_seconds: float,
    wait_for_pending_tasks: _FunctionToolSettlementWaiter,
    failure_sources_by_task: Mapping[asyncio.Task[Any], _FunctionToolFailureSource] | None = None,
    ignore_cancelled_tasks: set[asyncio.Task[Any]] | None = None,
) -> tuple[_FunctionToolFailure | None, set[asyncio.Task[Any]]]:
    """Wait for pending tasks to settle within a bounded window and collect failures."""
    if not pending_tasks:
        return None, set()

    failure: _FunctionToolFailure | None = None
    remaining_tasks = set(pending_tasks)
    loop = asyncio.get_running_loop()
    deadline = loop.time() + timeout_seconds

    while remaining_tasks:
        new_failure, remaining_tasks = _collect_settled_function_tool_tasks(
            remaining_tasks=remaining_tasks,
            task_states=task_states,
            results_by_tool_run=results_by_tool_run,
            failure_sources_by_task=failure_sources_by_task,
            ignore_cancelled_tasks=ignore_cancelled_tasks,
        )
        failure = _select_function_tool_failure(failure, new_failure)
        if failure is not None and not isinstance(failure.error, Exception):
            break

        remaining_time = deadline - loop.time()
        if not remaining_tasks or remaining_time <= 0:
            break

        should_continue = await wait_for_pending_tasks(remaining_tasks, loop, remaining_time)
        if not should_continue:
            break

    new_failure, remaining_tasks = _collect_settled_function_tool_tasks(
        remaining_tasks=remaining_tasks,
        task_states=task_states,
        results_by_tool_run=results_by_tool_run,
        failure_sources_by_task=failure_sources_by_task,
        ignore_cancelled_tasks=ignore_cancelled_tasks,
    )
    failure = _select_function_tool_failure(failure, new_failure)
    return failure, remaining_tasks


async def _drain_cancelled_function_tool_tasks(
    *,
    pending_tasks: set[asyncio.Task[Any]],
    task_states: Mapping[asyncio.Task[Any], _FunctionToolTaskState],
    results_by_tool_run: dict[int, Any],
    failure_sources_by_task: Mapping[asyncio.Task[Any], _FunctionToolFailureSource] | None = None,
    ignore_cancelled_tasks: set[asyncio.Task[Any]] | None = None,
) -> tuple[_FunctionToolFailure | None, set[asyncio.Task[Any]]]:
    """Drain cancelled siblings while they can continue making self-driven progress."""
    remaining_immediate_steps = _FUNCTION_TOOL_CANCELLED_IMMEDIATE_STEP_LIMIT

    async def _wait_for_progress(
        remaining: set[asyncio.Task[Any]],
        loop: asyncio.AbstractEventLoop,
        remaining_time: float,
    ) -> bool:
        nonlocal remaining_immediate_steps
        if remaining_immediate_steps <= 0:
            return False

        (
            should_continue,
            consumed_immediate_step,
        ) = await _wait_for_cancelled_function_tool_task_progress(
            remaining,
            loop,
            remaining_time,
            task_states=task_states,
        )
        if consumed_immediate_step:
            remaining_immediate_steps -= 1
        return should_continue

    return await _settle_pending_function_tool_tasks(
        pending_tasks=pending_tasks,
        task_states=task_states,
        results_by_tool_run=results_by_tool_run,
        timeout_seconds=_FUNCTION_TOOL_CANCELLED_DRAIN_SECONDS,
        wait_for_pending_tasks=_wait_for_progress,
        failure_sources_by_task=failure_sources_by_task,
        ignore_cancelled_tasks=ignore_cancelled_tasks,
    )


async def _wait_pending_function_tool_tasks_for_timeout(
    *,
    pending_tasks: set[asyncio.Task[Any]],
    task_states: Mapping[asyncio.Task[Any], _FunctionToolTaskState],
    results_by_tool_run: dict[int, Any],
    failure_sources_by_task: Mapping[asyncio.Task[Any], _FunctionToolFailureSource] | None = None,
    timeout_seconds: float,
) -> tuple[_FunctionToolFailure | None, set[asyncio.Task[Any]]]:
    """Wait briefly for post-invoke siblings so in-flight failures can still surface."""
    return await _settle_pending_function_tool_tasks(
        pending_tasks=pending_tasks,
        task_states=task_states,
        results_by_tool_run=results_by_tool_run,
        timeout_seconds=timeout_seconds,
        wait_for_pending_tasks=_wait_for_function_tool_task_completion,
        failure_sources_by_task=failure_sources_by_task,
    )


# --------------------------
# Public helpers
# --------------------------


def maybe_reset_tool_choice(
    agent: Agent[Any],
    tool_use_tracker: AgentToolUseTracker,
    model_settings: ModelSettings,
) -> ModelSettings:
    """Reset tool_choice if the agent was forced to pick a tool previously and should be reset."""
    if agent.reset_tool_choice is True and tool_use_tracker.has_used_tools(agent):
        return dataclasses.replace(model_settings, tool_choice=None)
    return model_settings


async def resolve_enabled_function_tools(
    agent: Agent[Any],
    context_wrapper: RunContextWrapper[Any],
) -> list[FunctionTool]:
    """Resolve enabled function tools without triggering MCP tool discovery."""

    async def _check_tool_enabled(tool: FunctionTool) -> bool:
        attr = tool.is_enabled
        if isinstance(attr, bool):
            return attr
        result = attr(context_wrapper, agent)
        if inspect.isawaitable(result):
            return bool(await result)
        return bool(result)

    function_tools = [tool for tool in agent.tools if isinstance(tool, FunctionTool)]
    if not function_tools:
        return []

    enabled_results = await asyncio.gather(*(_check_tool_enabled(tool) for tool in function_tools))
    return [tool for tool, enabled in zip(function_tools, enabled_results, strict=False) if enabled]


async def initialize_computer_tools(
    *,
    tools: list[Tool],
    context_wrapper: RunContextWrapper[Any],
) -> None:
    """Resolve computer tools ahead of model invocation so each run gets its own instance."""
    computer_tools = [tool for tool in tools if isinstance(tool, ComputerTool)]
    if not computer_tools:
        return

    await asyncio.gather(
        *(resolve_computer(tool=tool, run_context=context_wrapper) for tool in computer_tools)
    )


def get_mapping_or_attr(target: Any, key: str) -> Any:
    """Allow mapping-or-attribute access so tool payloads can be dicts or objects."""
    if isinstance(target, Mapping):
        return target.get(key)
    return getattr(target, key, None)


def extract_tool_call_id(raw: Any) -> str | None:
    """Return a call ID from tool call payloads or approval items."""
    # OpenAI tool call payloads are documented to include a call_id/id so outputs can be matched.
    # See https://platform.openai.com/docs/guides/function-calling
    # We still guard against missing IDs to avoid hard failures on malformed or non-OpenAI inputs.
    if isinstance(raw, Mapping):
        candidate = raw.get("call_id") or raw.get("id")
        return candidate if isinstance(candidate, str) else None
    candidate = get_mapping_or_attr(raw, "call_id") or get_mapping_or_attr(raw, "id")
    return candidate if isinstance(candidate, str) else None


def extract_shell_call_id(tool_call: Any) -> str:
    """Ensure shell calls include a call_id before executing them."""
    value = extract_tool_call_id(tool_call)
    if not value:
        raise ModelBehaviorError("Shell call is missing call_id.")
    return str(value)


def coerce_shell_call(tool_call: Any) -> ShellCallData:
    """Normalize a shell call payload into ShellCallData for consistent execution."""
    call_id = extract_shell_call_id(tool_call)
    action_payload = get_mapping_or_attr(tool_call, "action")
    if action_payload is None:
        raise ModelBehaviorError("Shell call is missing an action payload.")

    commands_value = get_mapping_or_attr(action_payload, "commands")
    if isinstance(commands_value, str | bytes | bytearray) or not isinstance(
        commands_value, Sequence
    ):
        raise ModelBehaviorError(
            "Shell call action commands must be a sequence of command strings."
        )
    commands: list[str] = []
    for entry in commands_value:
        if entry is None:
            continue
        commands.append(str(entry))
    if not commands:
        raise ModelBehaviorError("Shell call action must include at least one command.")

    timeout_value = (
        get_mapping_or_attr(action_payload, "timeout_ms")
        or get_mapping_or_attr(action_payload, "timeoutMs")
        or get_mapping_or_attr(action_payload, "timeout")
    )
    timeout_ms = int(timeout_value) if isinstance(timeout_value, int | float) else None

    max_length_value = get_mapping_or_attr(action_payload, "max_output_length")
    if max_length_value is None:
        max_length_value = get_mapping_or_attr(action_payload, "maxOutputLength")
    max_output_length = int(max_length_value) if isinstance(max_length_value, int | float) else None

    action = ShellActionRequest(
        commands=commands,
        timeout_ms=timeout_ms,
        max_output_length=max_output_length,
    )

    status_value = get_mapping_or_attr(tool_call, "status")
    status_literal: Literal["in_progress", "completed"] | None = None
    if isinstance(status_value, str):
        lowered = status_value.lower()
        if lowered in {"in_progress", "completed"}:
            status_literal = cast(Literal["in_progress", "completed"], lowered)

    return ShellCallData(call_id=call_id, action=action, status=status_literal, raw=tool_call)


def _parse_apply_patch_json(payload: str, *, label: str) -> dict[str, Any]:
    """Parse apply_patch JSON payloads with consistent error messages."""
    try:
        parsed = json.loads(payload or "{}")
    except json.JSONDecodeError as exc:
        raise ModelBehaviorError(f"Invalid apply_patch {label} JSON: {exc}") from exc
    if not isinstance(parsed, Mapping):
        raise ModelBehaviorError(f"Apply patch {label} must be a JSON object.")
    return dict(parsed)


def parse_apply_patch_custom_input(input_json: str) -> dict[str, Any]:
    """Parse custom apply_patch tool input used by legacy hosted-tool rollouts."""
    parsed = _parse_apply_patch_json(input_json, label="input")
    if "operation" in parsed or "operations" in parsed:
        return parsed
    return {"operation": parsed}


def parse_apply_patch_function_args(arguments: str) -> dict[str, Any]:
    """Parse apply_patch function tool arguments from the model."""
    return _parse_apply_patch_json(arguments, label="arguments")


def extract_apply_patch_call_id(tool_call: Any) -> str:
    """Ensure apply_patch calls include a call_id for approvals and tracing."""
    value = extract_tool_call_id(tool_call)
    if not value:
        raise ModelBehaviorError("Apply patch call is missing call_id.")
    return str(value)


def coerce_apply_patch_operation(
    tool_call: Any, *, context_wrapper: RunContextWrapper[Any]
) -> ApplyPatchOperation:
    """Normalize a single-operation tool payload for legacy callers."""
    operations = coerce_apply_patch_operations(tool_call, context_wrapper=context_wrapper)
    if len(operations) != 1:
        raise ModelBehaviorError(
            f"Apply patch call includes {len(operations)} operations; expected exactly one."
        )
    return operations[0]


def coerce_apply_patch_operations(
    tool_call: Any,
    *,
    context_wrapper: RunContextWrapper[Any],
) -> list[ApplyPatchOperation]:
    """Normalize apply_patch payloads into one or more editor operations."""
    raw_operations = get_mapping_or_attr(tool_call, "operations")
    if isinstance(raw_operations, list):
        operations = [
            _coerce_apply_patch_operation_payload(operation, context_wrapper=context_wrapper)
            for operation in raw_operations
        ]
        if not operations:
            raise ModelBehaviorError("Apply patch call includes no operations.")
        return operations

    raw_operation = get_mapping_or_attr(tool_call, "operation")
    if raw_operation is not None:
        return [
            _coerce_apply_patch_operation_payload(raw_operation, context_wrapper=context_wrapper)
        ]

    raise ModelBehaviorError("Apply patch call is missing an operation payload.")


def _coerce_apply_patch_operation_payload(
    raw_operation: Any, *, context_wrapper: RunContextWrapper[Any]
) -> ApplyPatchOperation:
    """Normalize the tool payload into an ApplyPatchOperation the editor can consume."""
    if raw_operation is None:
        raise ModelBehaviorError("Apply patch call is missing an operation payload.")

    op_type_value = str(get_mapping_or_attr(raw_operation, "type"))
    if op_type_value not in {"create_file", "update_file", "delete_file"}:
        raise ModelBehaviorError(f"Unknown apply_patch operation: {op_type_value}")
    op_type_literal = cast(Literal["create_file", "update_file", "delete_file"], op_type_value)

    path = get_mapping_or_attr(raw_operation, "path")
    if not isinstance(path, str) or not path:
        raise ModelBehaviorError("Apply patch operation is missing a valid path.")

    diff_value = get_mapping_or_attr(raw_operation, "diff")
    if op_type_literal in {"create_file", "update_file"}:
        if not isinstance(diff_value, str) or not diff_value:
            raise ModelBehaviorError(
                f"Apply patch operation {op_type_literal} is missing the required diff payload."
            )
        diff: str | None = diff_value
    else:
        diff = None

    return ApplyPatchOperation(
        type=op_type_literal,
        path=str(path),
        diff=diff,
        ctx_wrapper=context_wrapper,
        move_to=_coerce_apply_patch_move_to(raw_operation),
    )


def _coerce_apply_patch_move_to(raw_operation: Any) -> str | None:
    move_to = get_mapping_or_attr(raw_operation, "move_to")
    if move_to is None:
        return None
    if not isinstance(move_to, str) or not move_to:
        raise ModelBehaviorError("Apply patch operation move_to must be a non-empty path.")
    return move_to


def normalize_apply_patch_result(
    result: ApplyPatchResult | Mapping[str, Any] | str | None,
) -> ApplyPatchResult | None:
    """Coerce editor return values into ApplyPatchResult for consistent handling."""
    if result is None:
        return None
    if isinstance(result, ApplyPatchResult):
        return result
    if isinstance(result, Mapping):
        status = result.get("status")
        output = result.get("output")
        normalized_status = status if status in {"completed", "failed"} else None
        normalized_output = str(output) if output is not None else None
        return ApplyPatchResult(status=normalized_status, output=normalized_output)
    if isinstance(result, str):
        return ApplyPatchResult(output=result)
    return ApplyPatchResult(output=str(result))


def is_apply_patch_name(name: str | None, tool: ApplyPatchTool | None) -> bool:
    """Allow flexible matching for apply_patch so existing names keep working."""
    if not name:
        return False
    candidate = name.strip().lower()
    if candidate.startswith("apply_patch"):
        return True
    if tool and candidate == tool.name.strip().lower():
        return True
    return False


def normalize_shell_output(entry: ShellCommandOutput | Mapping[str, Any]) -> ShellCommandOutput:
    """Normalize shell output into ShellCommandOutput so downstream code sees a stable shape."""
    if isinstance(entry, ShellCommandOutput):
        return entry

    stdout = str(entry.get("stdout", "") or "")
    stderr = str(entry.get("stderr", "") or "")
    command_value = entry.get("command")
    provider_data_value = entry.get("provider_data")
    outcome_value = entry.get("outcome")

    outcome_type: Literal["exit", "timeout"] = "exit"
    exit_code_value: Any | None = None

    if isinstance(outcome_value, Mapping):
        type_value = outcome_value.get("type")
        if type_value == "timeout":
            outcome_type = "timeout"
        elif isinstance(type_value, str):
            outcome_type = "exit"
        exit_code_value = outcome_value.get("exit_code")
    else:
        status_str = str(entry.get("status", "completed") or "completed").lower()
        if status_str == "timeout":
            outcome_type = "timeout"
        if isinstance(outcome_value, str):
            if outcome_value == "failure":
                exit_code_value = 1
            elif outcome_value == "success":
                exit_code_value = 0
        if exit_code_value is None and "exit_code" in entry:
            exit_code_value = entry.get("exit_code")

    outcome = ShellCallOutcome(
        type=outcome_type,
        exit_code=_normalize_exit_code(exit_code_value),
    )

    return ShellCommandOutput(
        stdout=stdout,
        stderr=stderr,
        outcome=outcome,
        command=str(command_value) if command_value is not None else None,
        provider_data=cast(dict[str, Any], provider_data_value)
        if isinstance(provider_data_value, Mapping)
        else provider_data_value,
    )


def serialize_shell_output(output: ShellCommandOutput) -> dict[str, Any]:
    """Serialize ShellCommandOutput for persistence or cross-run transmission."""
    payload: dict[str, Any] = {
        "stdout": output.stdout,
        "stderr": output.stderr,
        "status": output.status,
        "outcome": {"type": output.outcome.type},
    }
    if output.outcome.type == "exit":
        payload["outcome"]["exit_code"] = output.outcome.exit_code
        if output.outcome.exit_code is not None:
            payload["exit_code"] = output.outcome.exit_code
    if output.command is not None:
        payload["command"] = output.command
    if output.provider_data:
        payload["provider_data"] = output.provider_data
    return payload


def resolve_exit_code(raw_exit_code: Any, outcome_status: str | None) -> int:
    """Fallback logic to produce an exit code when providers omit one."""
    normalized = _normalize_exit_code(raw_exit_code)
    if normalized is not None:
        return normalized

    normalized_status = (outcome_status or "").lower()
    if normalized_status == "success":
        return 0
    if normalized_status == "failure":
        return 1
    return 0


def render_shell_outputs(outputs: Sequence[ShellCommandOutput]) -> str:
    """Render shell outputs into human-readable text for tool responses."""
    if not outputs:
        return "(no output)"

    rendered_chunks: list[str] = []
    for result in outputs:
        chunk_lines: list[str] = []
        if result.command:
            chunk_lines.append(f"$ {result.command}")

        stdout = result.stdout.rstrip("\n")
        stderr = result.stderr.rstrip("\n")

        if stdout:
            chunk_lines.append(stdout)
        if stderr:
            if stdout:
                chunk_lines.append("")
            chunk_lines.append("stderr:")
            chunk_lines.append(stderr)

        if result.exit_code not in (None, 0):
            chunk_lines.append(f"exit code: {result.exit_code}")
        if result.status == "timeout":
            chunk_lines.append("status: timeout")

        chunk = "\n".join(chunk_lines).strip()
        rendered_chunks.append(chunk if chunk else "(no output)")

    return "\n\n".join(rendered_chunks)


def truncate_shell_outputs(
    outputs: Sequence[ShellCommandOutput], max_length: int
) -> list[ShellCommandOutput]:
    """Truncate shell output streams to a maximum combined length."""
    if max_length <= 0:
        return [
            ShellCommandOutput(
                stdout="",
                stderr="",
                outcome=output.outcome,
                command=output.command,
                provider_data=output.provider_data,
            )
            for output in outputs
        ]

    remaining = max_length
    truncated: list[ShellCommandOutput] = []
    for output in outputs:
        stdout = ""
        stderr = ""
        if remaining > 0 and output.stdout:
            stdout = output.stdout[:remaining]
            remaining -= len(stdout)
        if remaining > 0 and output.stderr:
            stderr = output.stderr[:remaining]
            remaining -= len(stderr)
        truncated.append(
            ShellCommandOutput(
                stdout=stdout,
                stderr=stderr,
                outcome=output.outcome,
                command=output.command,
                provider_data=output.provider_data,
            )
        )

    return truncated


def normalize_shell_output_entries(
    entries: Sequence[Mapping[str, Any]],
) -> list[dict[str, Any]]:
    """Normalize raw shell output entries into the model-facing payload."""
    structured_output: list[dict[str, Any]] = []
    for entry in entries:
        sanitized = dict(entry)
        status_value = sanitized.pop("status", None)
        sanitized.pop("provider_data", None)
        raw_exit_code = sanitized.pop("exit_code", None)
        sanitized.pop("command", None)
        outcome_value = sanitized.get("outcome")
        if isinstance(outcome_value, str):
            resolved_type = "exit"
            if status_value == "timeout":
                resolved_type = "timeout"
            outcome_payload: dict[str, Any] = {"type": resolved_type}
            if resolved_type == "exit":
                outcome_payload["exit_code"] = resolve_exit_code(raw_exit_code, outcome_value)
            sanitized["outcome"] = outcome_payload
        elif isinstance(outcome_value, dict):
            outcome_payload = dict(outcome_value)
            outcome_status = outcome_payload.pop("status", None)
            outcome_type = outcome_payload.get("type")
            if outcome_type != "timeout":
                status_str = outcome_status if isinstance(outcome_status, str) else None
                outcome_payload.setdefault(
                    "exit_code",
                    resolve_exit_code(raw_exit_code, status_str),
                )
            sanitized["outcome"] = outcome_payload
        structured_output.append(sanitized)
    return structured_output


def normalize_max_output_length(value: int | None) -> int | None:
    """Clamp negative max output lengths to zero while preserving None."""
    if value is None:
        return None
    return max(0, value)


def format_shell_error(error: Exception | BaseException | Any) -> str:
    """Best-effort stringify of shell errors to keep tool failures readable."""
    if isinstance(error, Exception):
        message = str(error)
        return message or error.__class__.__name__
    try:
        return str(error)
    except Exception:  # pragma: no cover - fallback only
        return repr(error)


async def with_tool_function_span(
    *,
    config: RunConfig,
    tool_name: str,
    fn: Callable[[Span[Any] | None], MaybeAwaitable[TToolSpanResult]],
) -> TToolSpanResult:
    """Execute a tool callback in a function span when tracing is active."""
    if config.tracing_disabled or get_current_trace() is None:
        result = fn(None)
        if inspect.isawaitable(result):
            return await result
        direct_result: object = result
        return cast(TToolSpanResult, direct_result)

    with function_span(tool_name) as span:
        result = fn(span)
        if inspect.isawaitable(result):
            return await result
        span_result: object = result
        return cast(TToolSpanResult, span_result)


def build_litellm_json_tool_call(output: ResponseFunctionToolCall) -> FunctionTool:
    """Wrap a JSON string result in a FunctionTool so LiteLLM can stream it."""

    async def on_invoke_tool(_ctx: ToolContext[Any], value: Any) -> Any:
        """Deserialize JSON strings so LiteLLM callers receive structured data."""
        if isinstance(value, str):
            return json.loads(value)
        return value

    return FunctionTool(
        name=output.name,
        description=output.name,
        params_json_schema={},
        on_invoke_tool=on_invoke_tool,
        strict_json_schema=True,
        is_enabled=True,
        _emit_tool_origin=False,
    )


async def resolve_approval_status(
    *,
    tool_name: str,
    call_id: str,
    raw_item: Any,
    agent: Agent[Any],
    context_wrapper: RunContextWrapper[Any],
    tool_namespace: str | None = None,
    tool_lookup_key: FunctionToolLookupKey | None = None,
    tool_origin: ToolOrigin | None = None,
    on_approval: Callable[[RunContextWrapper[Any], ToolApprovalItem], Any] | None = None,
) -> tuple[bool | None, ToolApprovalItem]:
    """Build approval item, run on_approval hook if needed, and return latest approval status."""
    approval_item = ToolApprovalItem(
        agent=agent,
        raw_item=raw_item,
        tool_name=tool_name,
        tool_namespace=tool_namespace,
        tool_origin=tool_origin,
        tool_lookup_key=tool_lookup_key,
    )
    approval_status = context_wrapper.get_approval_status(
        tool_name,
        call_id,
        tool_namespace=tool_namespace,
        existing_pending=approval_item,
        tool_lookup_key=tool_lookup_key,
    )
    if approval_status is None and on_approval:
        decision_result = on_approval(context_wrapper, approval_item)
        if inspect.isawaitable(decision_result):
            decision_result = await decision_result
        if isinstance(decision_result, Mapping):
            if decision_result.get("approve") is True:
                context_wrapper.approve_tool(approval_item)
            elif decision_result.get("approve") is False:
                reason = decision_result.get("reason")
                rejection_message = reason if isinstance(reason, str) and reason else None
                context_wrapper.reject_tool(
                    approval_item,
                    rejection_message=rejection_message,
                )
        approval_status = context_wrapper.get_approval_status(
            tool_name,
            call_id,
            tool_namespace=tool_namespace,
            existing_pending=approval_item,
            tool_lookup_key=tool_lookup_key,
        )
    return approval_status, approval_item


def resolve_approval_interruption(
    approval_status: bool | None,
    approval_item: ToolApprovalItem,
    *,
    rejection_factory: Callable[[], RunItem],
) -> RunItem | ToolApprovalItem | None:
    """Return a rejection or pending approval item when approval is required."""
    if approval_status is False:
        return rejection_factory()
    if approval_status is not True:
        return approval_item
    return None


async def resolve_approval_rejection_message(
    *,
    context_wrapper: RunContextWrapper[Any],
    run_config: RunConfig,
    tool_type: Literal["function", "computer", "shell", "apply_patch", "custom"],
    tool_name: str,
    call_id: str,
    tool_namespace: str | None = None,
    tool_lookup_key: FunctionToolLookupKey | None = None,
    existing_pending: ToolApprovalItem | None = None,
) -> str:
    """Resolve model-visible output text for approval rejections."""
    explicit_message = context_wrapper.get_rejection_message(
        tool_name,
        call_id,
        tool_namespace=tool_namespace,
        tool_lookup_key=tool_lookup_key,
        existing_pending=existing_pending,
    )
    if explicit_message is not None:
        return explicit_message

    formatter = run_config.tool_error_formatter
    if formatter is None:
        return REJECTION_MESSAGE

    try:
        maybe_message = formatter(
            ToolErrorFormatterArgs(
                kind="approval_rejected",
                tool_type=tool_type,
                tool_name=tool_name,
                call_id=call_id,
                default_message=REJECTION_MESSAGE,
                run_context=context_wrapper,
            )
        )
        message = await maybe_message if inspect.isawaitable(maybe_message) else maybe_message
    except Exception as exc:
        logger.error("Tool error formatter failed for %s: %s", tool_name, exc)
        return REJECTION_MESSAGE

    if message is None:
        return REJECTION_MESSAGE

    if not isinstance(message, str):
        logger.error(
            "Tool error formatter returned non-string for %s: %s",
            tool_name,
            type(message).__name__,
        )
        return REJECTION_MESSAGE

    return message


async def function_needs_approval(
    function_tool: FunctionTool,
    context_wrapper: RunContextWrapper[Any],
    tool_call: ResponseFunctionToolCall,
) -> bool:
    """Evaluate a function tool's needs_approval setting with parsed args."""
    parsed_args: dict[str, Any] = {}
    if callable(function_tool.needs_approval):
        try:
            parsed_args = json.loads(tool_call.arguments or "{}")
        except json.JSONDecodeError:
            parsed_args = {}
    needs_approval = await evaluate_needs_approval_setting(
        function_tool.needs_approval,
        context_wrapper,
        parsed_args,
        tool_call.call_id,
    )
    return bool(needs_approval)


def process_hosted_mcp_approvals(
    *,
    original_pre_step_items: Sequence[RunItem],
    mcp_approval_requests: Sequence[Any],
    context_wrapper: RunContextWrapper[Any],
    agent: Agent[Any],
    append_item: Callable[[RunItem], None],
) -> tuple[list[ToolApprovalItem], set[str]]:
    """Filter hosted MCP outputs and merge manual approvals so only coherent items remain."""
    hosted_mcp_approvals_by_id: dict[str, ToolApprovalItem] = {}
    for item in original_pre_step_items:
        if not isinstance(item, ToolApprovalItem):
            continue
        raw = item.raw_item
        if not _is_hosted_mcp_approval_request(raw):
            continue
        request_id = extract_mcp_request_id(raw)
        if request_id:
            hosted_mcp_approvals_by_id[request_id] = item

    pending_hosted_mcp_approvals: list[ToolApprovalItem] = []
    pending_hosted_mcp_approval_ids: set[str] = set()

    for mcp_run in mcp_approval_requests:
        request_id = extract_mcp_request_id_from_run(mcp_run)
        # MCP approval requests are documented to include an id used as approval_request_id.
        # See https://platform.openai.com/docs/guides/tools-connectors-mcp#approvals
        approval_item = hosted_mcp_approvals_by_id.get(request_id) if request_id else None
        if not approval_item or not request_id:
            continue

        tool_name = RunContextWrapper._resolve_tool_name(approval_item)
        approved = context_wrapper.get_approval_status(
            tool_name=tool_name,
            call_id=request_id,
            existing_pending=approval_item,
        )

        if approved is not None:
            raw_item: McpApprovalResponse = {
                "type": "mcp_approval_response",
                "approval_request_id": request_id,
                "approve": approved,
            }
            rejection_message = context_wrapper.get_rejection_message(
                tool_name=tool_name,
                call_id=request_id,
                existing_pending=approval_item,
            )
            if approved is False and rejection_message is not None:
                raw_item["reason"] = rejection_message
            response_item = MCPApprovalResponseItem(raw_item=raw_item, agent=agent)
            append_item(response_item)
            continue

        if approval_item not in pending_hosted_mcp_approvals:
            pending_hosted_mcp_approvals.append(approval_item)
        pending_hosted_mcp_approval_ids.add(request_id)
        append_item(approval_item)

    return pending_hosted_mcp_approvals, pending_hosted_mcp_approval_ids


def collect_manual_mcp_approvals(
    *,
    agent: Agent[Any],
    requests: Sequence[Any],
    context_wrapper: RunContextWrapper[Any],
    existing_pending_by_call_id: Mapping[str, ToolApprovalItem] | None = None,
) -> tuple[list[MCPApprovalResponseItem], list[ToolApprovalItem]]:
    """Bridge hosted MCP approval requests with manual approvals to keep state consistent."""
    pending_lookup = existing_pending_by_call_id or {}
    approved: list[MCPApprovalResponseItem] = []
    pending: list[ToolApprovalItem] = []
    seen_request_ids: set[str] = set()

    for request in requests:
        request_item = get_mapping_or_attr(request, "request_item")
        request_id = extract_mcp_request_id_from_run(request)
        # The Responses API returns mcp_approval_request items with an id to correlate approvals.
        # See https://platform.openai.com/docs/guides/tools-connectors-mcp#approvals
        if request_id and request_id in seen_request_ids:
            continue
        if request_id:
            seen_request_ids.add(request_id)

        tool_name = RunContextWrapper._to_str_or_none(getattr(request_item, "name", None))
        tool_name = tool_name or get_mapping_or_attr(request, "mcp_tool").name

        existing_pending = pending_lookup.get(request_id or "")
        approval_status = context_wrapper.get_approval_status(
            tool_name, request_id or "", existing_pending=existing_pending
        )

        if approval_status is not None and request_id:
            approval_response_raw: McpApprovalResponse = {
                "type": "mcp_approval_response",
                "approval_request_id": request_id,
                "approve": approval_status,
            }
            rejection_message = context_wrapper.get_rejection_message(
                tool_name,
                request_id,
                existing_pending=existing_pending,
            )
            if approval_status is False and rejection_message is not None:
                approval_response_raw["reason"] = rejection_message
            approved.append(MCPApprovalResponseItem(raw_item=approval_response_raw, agent=agent))
            continue

        if approval_status is not None:
            continue

        pending.append(
            existing_pending
            or ToolApprovalItem(
                agent=agent,
                raw_item=request_item,
                tool_name=tool_name,
            )
        )

    return approved, pending


def index_approval_items_by_call_id(items: Sequence[RunItem]) -> dict[str, ToolApprovalItem]:
    """Build a mapping of tool call IDs to pending approval items."""
    approvals: dict[str, ToolApprovalItem] = {}
    for item in items:
        if not isinstance(item, ToolApprovalItem):
            continue
        call_id = extract_tool_call_id(item.raw_item)
        if call_id:
            approvals[call_id] = item
    return approvals


def should_keep_hosted_mcp_item(
    item: RunItem,
    *,
    pending_hosted_mcp_approvals: Sequence[ToolApprovalItem],
    pending_hosted_mcp_approval_ids: set[str],
) -> bool:
    """Keep only hosted MCP approvals that match pending requests from the provider."""
    if not isinstance(item, ToolApprovalItem):
        return True
    if not _is_hosted_mcp_approval_request(item.raw_item):
        return False
    request_id = extract_mcp_request_id(item.raw_item)
    return item in pending_hosted_mcp_approvals or (
        request_id is not None and request_id in pending_hosted_mcp_approval_ids
    )


class _FunctionToolBatchExecutor:
    """Own the mutable state needed to execute and arbitrate a function-tool batch."""

    def __init__(
        self,
        *,
        bindings: AgentBindings[Any],
        tool_runs: list[ToolRunFunction],
        hooks: RunHooks[Any],
        context_wrapper: RunContextWrapper[Any],
        config: RunConfig,
        isolate_parallel_failures: bool | None,
    ) -> None:
        self.execution_agent = bindings.execution_agent
        self.public_agent = bindings.public_agent
        self.tool_runs = tool_runs
        self.hooks = hooks
        self.context_wrapper = context_wrapper
        self.config = config
        self.isolate_parallel_failures = (
            len(tool_runs) > 1 if isolate_parallel_failures is None else isolate_parallel_failures
        )
        self.tool_input_guardrail_results: list[ToolInputGuardrailResult] = []
        self.tool_output_guardrail_results: list[ToolOutputGuardrailResult] = []
        self.tool_state_scope_id = get_agent_tool_state_scope(context_wrapper)
        self.task_states: dict[asyncio.Task[Any], _FunctionToolTaskState] = {}
        self.teardown_cancelled_tasks: set[asyncio.Task[Any]] = set()
        self.results_by_tool_run: dict[int, Any] = {}
        self.custom_data_by_tool_run: dict[int, dict[str, Any]] = {}
        self.pending_tasks: set[asyncio.Task[Any]] = set()
        self.propagating_failure: BaseException | None = None
        self.available_function_tools: list[FunctionTool] = []
        self.max_function_tool_concurrency = (
            config.tool_execution.max_function_tool_concurrency if config.tool_execution else None
        )

    async def execute(
        self,
    ) -> tuple[
        list[FunctionToolResult], list[ToolInputGuardrailResult], list[ToolOutputGuardrailResult]
    ]:
        self.available_function_tools = await resolve_enabled_function_tools(
            self.execution_agent,
            self.context_wrapper,
        )
        enabled_function_tool_ids = {id(tool) for tool in self.available_function_tools}
        configured_function_tool_ids = {
            id(tool) for tool in self.execution_agent.tools if isinstance(tool, FunctionTool)
        }
        for tool_run in self.tool_runs:
            function_tool = tool_run.function_tool
            function_tool_id = id(function_tool)
            if (
                function_tool_id in configured_function_tool_ids
                and function_tool_id not in enabled_function_tool_ids
            ):
                raise ModelBehaviorError(
                    f"Tool {function_tool.name} is currently disabled for agent "
                    f"{self.public_agent.name}."
                )
            if function_tool_id not in enabled_function_tool_ids:
                self.available_function_tools.append(tool_run.function_tool)
                enabled_function_tool_ids.add(function_tool_id)
        pending_tool_runs = list(enumerate(self.tool_runs))
        self._fill_tool_task_slots(pending_tool_runs)

        try:
            await self._drain_pending_tasks(pending_tool_runs)
        except asyncio.CancelledError as exc:
            if self.propagating_failure is exc:
                raise
            self._cancel_pending_tasks_for_parent_cancellation()
            raise

        return (
            self._build_function_tool_results(),
            self.tool_input_guardrail_results,
            self.tool_output_guardrail_results,
        )

    def _fill_tool_task_slots(self, pending_tool_runs: list[tuple[int, ToolRunFunction]]) -> None:
        max_concurrency = self.max_function_tool_concurrency
        available_slots = (
            len(pending_tool_runs)
            if max_concurrency is None
            else max_concurrency - len(self.pending_tasks)
        )
        while available_slots > 0 and pending_tool_runs:
            order, tool_run = pending_tool_runs.pop(0)
            self._create_tool_task(tool_run, order)
            available_slots -= 1

    def _create_tool_task(self, tool_run: ToolRunFunction, order: int) -> None:
        task_state = _FunctionToolTaskState(tool_run=tool_run, order=order)
        task = asyncio.create_task(
            self._run_single_tool(
                task_state=task_state,
                func_tool=tool_run.function_tool,
                tool_call=tool_run.tool_call,
            )
        )
        self.task_states[task] = task_state
        self.pending_tasks.add(task)

    async def _drain_pending_tasks(
        self,
        pending_tool_runs: list[tuple[int, ToolRunFunction]],
    ) -> None:
        while self.pending_tasks:
            done_tasks, self.pending_tasks = await asyncio.wait(
                self.pending_tasks,
                return_when=asyncio.FIRST_COMPLETED,
            )
            failure = _record_completed_function_tool_tasks(
                completed_tasks=list(done_tasks),
                task_states=self.task_states,
                results_by_tool_run=self.results_by_tool_run,
            )
            if failure is not None:
                await self._raise_failure_after_draining_siblings(failure)
            self._fill_tool_task_slots(pending_tool_runs)

    async def _raise_failure_after_draining_siblings(
        self,
        failure: _FunctionToolFailure,
    ) -> None:
        cancellable_tasks, post_invoke_tasks = self._partition_pending_tasks()
        self.teardown_cancelled_tasks.update(cancellable_tasks)
        _cancel_function_tool_tasks(cancellable_tasks)

        late_failure, remaining_cancelled_tasks = await self._drain_cancelled_tasks(
            cancellable_tasks
        )
        post_invoke_failure, remaining_post_invoke_tasks = await self._wait_post_invoke_tasks(
            post_invoke_tasks
        )

        _attach_function_tool_task_result_callbacks(
            remaining_cancelled_tasks,
            message_for_exception=_background_cleanup_task_exception_message,
        )
        _attach_function_tool_task_result_callbacks(
            remaining_post_invoke_tasks,
            message_for_exception=_background_post_invoke_task_exception_message,
        )

        merged_failure = _merge_late_function_tool_failure(failure, late_failure)
        merged_failure = _merge_late_function_tool_failure(merged_failure, post_invoke_failure)
        assert merged_failure is not None
        self.pending_tasks = set()
        self.propagating_failure = merged_failure.error
        raise merged_failure.error

    def _partition_pending_tasks(self) -> tuple[set[asyncio.Task[Any]], set[asyncio.Task[Any]]]:
        cancellable_tasks = {
            task for task in self.pending_tasks if not self.task_states[task].in_post_invoke_phase
        }
        return cancellable_tasks, self.pending_tasks - cancellable_tasks

    async def _drain_cancelled_tasks(
        self,
        tasks: set[asyncio.Task[Any]],
    ) -> tuple[_FunctionToolFailure | None, set[asyncio.Task[Any]]]:
        late_failure_sources: dict[asyncio.Task[Any], _FunctionToolFailureSource] = dict.fromkeys(
            tasks,
            "cancelled_teardown",
        )
        return await _drain_cancelled_function_tool_tasks(
            pending_tasks=tasks,
            task_states=self.task_states,
            results_by_tool_run=self.results_by_tool_run,
            failure_sources_by_task=late_failure_sources,
            ignore_cancelled_tasks=tasks,
        )

    async def _wait_post_invoke_tasks(
        self,
        tasks: set[asyncio.Task[Any]],
    ) -> tuple[_FunctionToolFailure | None, set[asyncio.Task[Any]]]:
        post_invoke_failure_sources: dict[asyncio.Task[Any], _FunctionToolFailureSource] = (
            dict.fromkeys(tasks, "post_invoke")
        )
        return await _wait_pending_function_tool_tasks_for_timeout(
            pending_tasks=tasks,
            task_states=self.task_states,
            results_by_tool_run=self.results_by_tool_run,
            failure_sources_by_task=post_invoke_failure_sources,
            timeout_seconds=_FUNCTION_TOOL_POST_INVOKE_WAIT_SECONDS,
        )

    def _cancel_pending_tasks_for_parent_cancellation(self) -> None:
        self.teardown_cancelled_tasks.update(self.pending_tasks)
        _cancel_function_tool_tasks(self.pending_tasks)
        _attach_function_tool_task_result_callbacks(
            self.pending_tasks,
            message_for_exception=_parent_cancelled_task_exception_message,
        )

    async def _run_single_tool(
        self,
        *,
        task_state: _FunctionToolTaskState,
        func_tool: FunctionTool,
        tool_call: ResponseFunctionToolCall,
    ) -> Any:
        raw_tool_call = tool_call
        outer_task = asyncio.current_task()
        task_state.in_post_invoke_phase = False

        tool_call = cast(
            ResponseFunctionToolCall,
            normalize_tool_call_for_function_tool(tool_call, func_tool),
        )
        trace_tool_name = (
            get_tool_call_trace_name(tool_call)
            or get_function_tool_trace_name(func_tool)
            or func_tool.name
        )
        with function_span(trace_tool_name) as span_fn:
            tool_context_namespace = get_tool_call_namespace(raw_tool_call)
            if tool_context_namespace is None:
                tool_context_namespace = get_tool_call_namespace(tool_call)
            tool_context = ToolContext.from_agent_context(
                self.context_wrapper,
                tool_call.call_id,
                tool_call=raw_tool_call,
                tool_namespace=tool_context_namespace,
                agent=self.public_agent,
                run_config=self.config,
            )
            agent_hooks = self.public_agent.hooks
            if self.config.trace_include_sensitive_data:
                span_fn.span_data.input = tool_call.arguments

            try:
                approval_result = await self._maybe_execute_tool_approval(
                    func_tool=func_tool,
                    tool_call=tool_call,
                    raw_tool_call=raw_tool_call,
                    span_fn=span_fn,
                )
                if approval_result is not None:
                    result = approval_result
                else:
                    result = await self._execute_single_tool_body(
                        outer_task=outer_task,
                        task_state=task_state,
                        func_tool=func_tool,
                        tool_call=tool_call,
                        tool_context=tool_context,
                        agent_hooks=agent_hooks,
                    )
            except Exception as e:
                trace_error = get_trace_tool_error(
                    trace_include_sensitive_data=self.config.trace_include_sensitive_data,
                    error_message=str(e),
                )
                _error_tracing.attach_error_to_current_span(
                    SpanError(
                        message="Error running tool",
                        data={"tool_name": func_tool.name, "error": trace_error},
                    )
                )
                if isinstance(e, AgentsException):
                    raise e
                raise UserError(f"Error running tool {func_tool.name}: {e}") from e

            if self.config.trace_include_sensitive_data:
                span_fn.span_data.output = result
            return result

    async def _maybe_execute_tool_approval(
        self,
        *,
        func_tool: FunctionTool,
        tool_call: ResponseFunctionToolCall,
        raw_tool_call: ResponseFunctionToolCall,
        span_fn: Span[Any],
    ) -> Any | None:
        needs_approval_result = await function_needs_approval(
            func_tool,
            self.context_wrapper,
            tool_call,
        )
        if not needs_approval_result:
            return None

        tool_namespace = get_tool_call_namespace(raw_tool_call)
        if tool_namespace is None and is_deferred_top_level_function_tool(func_tool):
            tool_namespace = func_tool.name
        tool_lookup_key = get_function_tool_lookup_key_for_call(raw_tool_call)
        if is_deferred_top_level_function_tool(func_tool):
            tool_lookup_key = ("deferred_top_level", func_tool.name)
        approval_status = self.context_wrapper.get_approval_status(
            func_tool.name,
            tool_call.call_id,
            tool_namespace=tool_namespace,
            tool_lookup_key=tool_lookup_key,
        )
        if approval_status is None:
            if self._should_run_pre_approval_tool_input_guardrails():
                tool_context_namespace = get_tool_call_namespace(raw_tool_call)
                if tool_context_namespace is None:
                    tool_context_namespace = get_tool_call_namespace(tool_call)
                tool_context = ToolContext.from_agent_context(
                    self.context_wrapper,
                    tool_call.call_id,
                    tool_call=raw_tool_call,
                    tool_namespace=tool_context_namespace,
                    agent=self.public_agent,
                    run_config=self.config,
                )
                rejected_message = await _execute_tool_input_guardrails(
                    func_tool=func_tool,
                    tool_context=tool_context,
                    agent=self.public_agent,
                    tool_input_guardrail_results=self.tool_input_guardrail_results,
                )
                if rejected_message is not None:
                    return FunctionToolResult(
                        tool=func_tool,
                        output=rejected_message,
                        run_item=function_rejection_item(
                            self.public_agent,
                            tool_call,
                            rejection_message=rejected_message,
                            scope_id=self.tool_state_scope_id,
                            tool_origin=get_function_tool_origin(func_tool),
                        ),
                    )
            approval_item = ToolApprovalItem(
                agent=self.public_agent,
                raw_item=raw_tool_call,
                tool_name=func_tool.name,
                tool_namespace=tool_namespace,
                tool_origin=get_function_tool_origin(func_tool),
                tool_lookup_key=tool_lookup_key,
                _allow_bare_name_alias=should_allow_bare_name_approval_alias(
                    func_tool,
                    self.available_function_tools,
                ),
            )
            return FunctionToolResult(tool=func_tool, output=None, run_item=approval_item)

        if approval_status is not False:
            return None

        rejection_message = await resolve_approval_rejection_message(
            context_wrapper=self.context_wrapper,
            run_config=self.config,
            tool_type="function",
            tool_name=tool_trace_name(func_tool.name, tool_namespace) or func_tool.name,
            call_id=tool_call.call_id,
            tool_namespace=tool_namespace,
            tool_lookup_key=tool_lookup_key,
        )
        span_fn.set_error(
            SpanError(
                message=rejection_message,
                data={
                    "tool_name": func_tool.name,
                    "error": (
                        f"Tool execution for {tool_call.call_id} was manually rejected by user."
                    ),
                },
            )
        )
        span_fn.span_data.output = rejection_message
        return FunctionToolResult(
            tool=func_tool,
            output=rejection_message,
            run_item=function_rejection_item(
                self.public_agent,
                tool_call,
                rejection_message=rejection_message,
                scope_id=self.tool_state_scope_id,
                tool_origin=get_function_tool_origin(func_tool),
            ),
        )

    async def _execute_single_tool_body(
        self,
        *,
        outer_task: asyncio.Task[Any] | None,
        task_state: _FunctionToolTaskState,
        func_tool: FunctionTool,
        tool_call: ResponseFunctionToolCall,
        tool_context: ToolContext[Any],
        agent_hooks: Any,
    ) -> Any:
        rejected_message = await _execute_tool_input_guardrails(
            func_tool=func_tool,
            tool_context=tool_context,
            agent=self.public_agent,
            tool_input_guardrail_results=self.tool_input_guardrail_results,
        )
        if rejected_message is not None:
            return rejected_message

        await asyncio.gather(
            self.hooks.on_tool_start(tool_context, self.public_agent, func_tool),
            (
                agent_hooks.on_tool_start(tool_context, self.public_agent, func_tool)
                if agent_hooks
                else _coro.noop_coroutine()
            ),
        )

        invoke_task = asyncio.create_task(
            self._invoke_tool_and_run_post_invoke(
                outer_task=outer_task,
                task_state=task_state,
                func_tool=func_tool,
                tool_call=tool_call,
                tool_context=tool_context,
                agent_hooks=agent_hooks,
            )
        )
        task_state.invoke_task = invoke_task
        return await self._await_invoke_task(outer_task=outer_task, invoke_task=invoke_task)

    def _should_run_pre_approval_tool_input_guardrails(self) -> bool:
        tool_execution = self.config.tool_execution
        if tool_execution is None:
            return False
        return tool_execution.pre_approval_tool_input_guardrails

    async def _invoke_tool_and_run_post_invoke(
        self,
        *,
        outer_task: asyncio.Task[Any] | None,
        task_state: _FunctionToolTaskState,
        func_tool: FunctionTool,
        tool_call: ResponseFunctionToolCall,
        tool_context: ToolContext[Any],
        agent_hooks: Any,
    ) -> Any:
        try:
            real_result = await invoke_function_tool(
                function_tool=func_tool,
                context=tool_context,
                arguments=tool_call.arguments,
            )
        except asyncio.CancelledError as e:
            if outer_task in self.teardown_cancelled_tasks:
                raise

            result = await maybe_invoke_function_tool_failure_error_function(
                function_tool=func_tool,
                context=tool_context,
                error=e,
            )
            if result is None:
                raise

            trace_error = get_trace_tool_error(
                trace_include_sensitive_data=self.config.trace_include_sensitive_data,
                error_message=str(e),
            )
            _error_tracing.attach_error_to_current_span(
                SpanError(
                    message="Tool execution cancelled",
                    data={"tool_name": func_tool.name, "error": trace_error},
                )
            )
            real_result = result

        task_state.in_post_invoke_phase = True

        final_result = await _execute_tool_output_guardrails(
            func_tool=func_tool,
            tool_context=tool_context,
            agent=self.public_agent,
            real_result=real_result,
            tool_output_guardrail_results=self.tool_output_guardrail_results,
        )
        raw_output_item = ItemHelpers.tool_call_output_item(tool_call, final_result)
        extracted_custom_data = await maybe_extract_custom_data(
            func_tool.custom_data_extractor,
            FunctionToolCustomDataContext(
                tool_context=tool_context,
                tool=func_tool,
                output=final_result,
                raw_item=copy.deepcopy(raw_output_item),
            ),
        )
        custom_data = merge_custom_data(tool_context._custom_data, extracted_custom_data)
        if custom_data:
            self.custom_data_by_tool_run[id(task_state.tool_run)] = custom_data

        await asyncio.gather(
            self.hooks.on_tool_end(tool_context, self.public_agent, func_tool, final_result),
            (
                agent_hooks.on_tool_end(tool_context, self.public_agent, func_tool, final_result)
                if agent_hooks
                else _coro.noop_coroutine()
            ),
        )
        return final_result

    async def _await_invoke_task(
        self,
        *,
        outer_task: asyncio.Task[Any] | None,
        invoke_task: asyncio.Task[Any],
    ) -> Any:
        try:
            return await asyncio.shield(invoke_task)
        except asyncio.CancelledError as cancel_exc:
            sibling_failure_cancelled = (
                outer_task is not None and outer_task in self.teardown_cancelled_tasks
            )
            if not invoke_task.done():
                invoke_task.cancel()
            if sibling_failure_cancelled:
                invoke_results = await asyncio.gather(invoke_task, return_exceptions=True)
                invoke_failure = invoke_results[0] if invoke_results else None
                if isinstance(invoke_failure, BaseException) and not isinstance(
                    invoke_failure, asyncio.CancelledError
                ):
                    raise invoke_failure from cancel_exc
            elif invoke_task.done():
                if not invoke_task.cancelled():
                    invoke_failure = invoke_task.exception()
                    if isinstance(invoke_failure, BaseException) and not isinstance(
                        invoke_failure, Exception
                    ):
                        raise invoke_failure from cancel_exc
            else:
                invoke_task.add_done_callback(
                    functools.partial(
                        _consume_function_tool_task_result,
                        message_for_exception=_parent_cancelled_task_exception_message,
                    )
                )
            raise

    def _get_nested_tool_interruptions(
        self,
        nested_run_result: Any | None,
    ) -> list[ToolApprovalItem]:
        """Extract nested approval interruptions from an agent tool run result."""
        if nested_run_result is None or not hasattr(nested_run_result, "interruptions"):
            return []
        return cast(list[ToolApprovalItem], nested_run_result.interruptions)

    def _consume_nested_tool_run_result(
        self,
        tool_run: ToolRunFunction,
    ) -> tuple[Any | None, list[ToolApprovalItem]]:
        """Consume stored nested run state for a tool call and return its interruptions."""
        nested_run_result = consume_agent_tool_run_result(
            tool_run.tool_call,
            scope_id=self.tool_state_scope_id,
        )
        return nested_run_result, self._get_nested_tool_interruptions(nested_run_result)

    def _resolve_nested_tool_run_result(
        self,
        tool_run: ToolRunFunction,
    ) -> tuple[Any | None, list[ToolApprovalItem]]:
        """Load nested run state, preserving unresolved interruptions until they are handled."""
        nested_run_result = peek_agent_tool_run_result(
            tool_run.tool_call,
            scope_id=self.tool_state_scope_id,
        )
        nested_interruptions = self._get_nested_tool_interruptions(nested_run_result)
        if nested_run_result is None or not nested_interruptions:
            nested_run_result, nested_interruptions = self._consume_nested_tool_run_result(tool_run)
        return nested_run_result, nested_interruptions

    def _build_function_tool_results(self) -> list[FunctionToolResult]:
        function_tool_results: list[FunctionToolResult] = []
        for tool_run in self.tool_runs:
            result = self.results_by_tool_run[id(tool_run)]
            if isinstance(result, FunctionToolResult):
                nested_run_result, nested_interruptions = self._consume_nested_tool_run_result(
                    tool_run
                )
                if nested_run_result:
                    result.agent_run_result = nested_run_result
                    if nested_interruptions:
                        result.interruptions = nested_interruptions

                function_tool_results.append(result)
                continue

            nested_run_result, nested_interruptions = self._resolve_nested_tool_run_result(tool_run)

            run_item: RunItem | None
            if not nested_interruptions:
                run_item = ToolCallOutputItem(
                    output=result,
                    raw_item=ItemHelpers.tool_call_output_item(tool_run.tool_call, result),
                    agent=self.public_agent,
                    tool_origin=get_function_tool_origin(tool_run.function_tool),
                    custom_data=self.custom_data_by_tool_run.get(id(tool_run)),
                )
            else:
                # Skip tool output until nested interruptions are resolved.
                run_item = None

            function_tool_results.append(
                FunctionToolResult(
                    tool=tool_run.function_tool,
                    output=result,
                    run_item=run_item,
                    interruptions=nested_interruptions,
                    agent_run_result=nested_run_result,
                )
            )

        return function_tool_results


async def execute_function_tool_calls(
    *,
    bindings: AgentBindings[Any],
    tool_runs: list[ToolRunFunction],
    hooks: RunHooks[Any],
    context_wrapper: RunContextWrapper[Any],
    config: RunConfig,
    isolate_parallel_failures: bool | None = None,
) -> tuple[
    list[FunctionToolResult], list[ToolInputGuardrailResult], list[ToolOutputGuardrailResult]
]:
    """Execute function tool calls with approvals, guardrails, and hooks."""
    return await _FunctionToolBatchExecutor(
        bindings=bindings,
        tool_runs=tool_runs,
        hooks=hooks,
        context_wrapper=context_wrapper,
        config=config,
        isolate_parallel_failures=isolate_parallel_failures,
    ).execute()


async def execute_custom_tool_calls(
    *,
    public_agent: Agent[Any],
    calls: list[ToolRunCustom],
    context_wrapper: RunContextWrapper[Any],
    hooks: RunHooks[Any],
    config: RunConfig,
) -> list[RunItem]:
    """Run Responses custom tool calls serially and wrap outputs."""
    from .tool_actions import CustomToolAction

    results: list[RunItem] = []
    for call in calls:
        results.append(
            await CustomToolAction.execute(
                agent=public_agent,
                call=call,
                hooks=hooks,
                context_wrapper=context_wrapper,
                config=config,
            )
        )
    return results


async def execute_local_shell_calls(
    *,
    public_agent: Agent[Any],
    calls: list[ToolRunLocalShellCall],
    context_wrapper: RunContextWrapper[Any],
    hooks: RunHooks[Any],
    config: RunConfig,
) -> list[RunItem]:
    """Run local shell tool calls serially and wrap outputs."""
    from .tool_actions import LocalShellAction

    results: list[RunItem] = []
    for call in calls:
        results.append(
            await LocalShellAction.execute(
                agent=public_agent,
                call=call,
                hooks=hooks,
                context_wrapper=context_wrapper,
                config=config,
            )
        )
    return results


async def execute_shell_calls(
    *,
    public_agent: Agent[Any],
    calls: list[ToolRunShellCall],
    context_wrapper: RunContextWrapper[Any],
    hooks: RunHooks[Any],
    config: RunConfig,
) -> list[RunItem]:
    """Run shell tool calls serially and wrap outputs."""
    from .tool_actions import ShellAction

    results: list[RunItem] = []
    for call in calls:
        results.append(
            await ShellAction.execute(
                agent=public_agent,
                call=call,
                hooks=hooks,
                context_wrapper=context_wrapper,
                config=config,
            )
        )
    return results


async def execute_apply_patch_calls(
    *,
    public_agent: Agent[Any],
    calls: list[ToolRunApplyPatchCall],
    context_wrapper: RunContextWrapper[Any],
    hooks: RunHooks[Any],
    config: RunConfig,
) -> list[RunItem]:
    """Run apply_patch tool calls serially and normalize outputs."""
    from .tool_actions import ApplyPatchAction

    results: list[RunItem] = []
    for call in calls:
        results.append(
            await ApplyPatchAction.execute(
                agent=public_agent,
                call=call,
                hooks=hooks,
                context_wrapper=context_wrapper,
                config=config,
            )
        )
    return results


async def execute_computer_actions(
    *,
    public_agent: Agent[Any],
    actions: list[ToolRunComputerAction],
    hooks: RunHooks[Any],
    context_wrapper: RunContextWrapper[Any],
    config: RunConfig,
) -> list[RunItem]:
    """Run computer actions serially and emit screenshot outputs."""
    from .tool_actions import ComputerAction

    results: list[RunItem] = []
    for action in actions:
        acknowledged: list[ComputerCallOutputAcknowledgedSafetyCheck] | None = None
        if action.tool_call.pending_safety_checks and action.computer_tool.on_safety_check:
            acknowledged = []
            for check in action.tool_call.pending_safety_checks:
                data = ComputerToolSafetyCheckData(
                    ctx_wrapper=context_wrapper,
                    agent=public_agent,
                    tool_call=action.tool_call,
                    safety_check=check,
                )
                maybe = action.computer_tool.on_safety_check(data)
                ack = await maybe if inspect.isawaitable(maybe) else maybe
                if ack:
                    acknowledged.append(
                        ComputerCallOutputAcknowledgedSafetyCheck(
                            id=check.id,
                            code=check.code,
                            message=check.message,
                        )
                    )
                else:
                    raise UserError("Computer tool safety check was not acknowledged")

        results.append(
            await ComputerAction.execute(
                agent=public_agent,
                action=action,
                hooks=hooks,
                context_wrapper=context_wrapper,
                config=config,
                acknowledged_safety_checks=acknowledged,
            )
        )

    return results


async def execute_approved_tools(
    *,
    agent: Agent[Any],
    interruptions: list[Any],
    context_wrapper: RunContextWrapper[Any],
    generated_items: list[RunItem],
    run_config: RunConfig,
    hooks: RunHooks[Any],
    all_tools: list[Tool] | None = None,
) -> None:
    """Execute tools that have been approved after an interruption (HITL resume path)."""
    tool_runs: list[ToolRunFunction] = []
    tool_map: dict[NamedToolLookupKey, Tool] = cast(
        dict[NamedToolLookupKey, Tool],
        build_function_tool_lookup_map(
            [tool for tool in all_tools or [] if isinstance(tool, FunctionTool)]
        ),
    )
    for tool in all_tools or []:
        if isinstance(tool, FunctionTool):
            continue
        if hasattr(tool, "name"):
            tool_name = getattr(tool, "name", None)
            if isinstance(tool_name, str) and tool_name:
                tool_map[tool_name] = tool

    def _append_error(
        message: str,
        *,
        tool_call: Any,
        tool_name: str,
        call_id: str,
        tool_origin: ToolOrigin | None = None,
    ) -> None:
        append_approval_error_output(
            message=message,
            tool_call=tool_call,
            tool_name=tool_name,
            call_id=call_id,
            generated_items=generated_items,
            agent=agent,
            tool_origin=tool_origin,
        )

    async def _resolve_tool_run(
        interruption: Any,
    ) -> tuple[ResponseFunctionToolCall, FunctionTool, str, str] | None:
        tool_call = interruption.raw_item
        tool_name = interruption.name or RunContextWrapper._resolve_tool_name(interruption)
        tool_namespace = getattr(interruption, "tool_namespace", None)
        tool_lookup_key = getattr(
            interruption, "tool_lookup_key", None
        ) or get_function_tool_lookup_key(
            tool_name,
            tool_namespace,
        )
        approval_key = tool_lookup_key
        display_tool_name = tool_trace_name(tool_name, tool_namespace) or tool_name or "unknown"
        if not tool_name:
            _append_error(
                message="Tool approval item missing tool name.",
                tool_call=tool_call,
                tool_name="unknown",
                call_id="unknown",
            )
            return None

        call_id = extract_tool_call_id(tool_call)
        if not call_id:
            resolved_tool = tool_map.get(approval_key) if approval_key is not None else None
            if resolved_tool is None and tool_namespace is None:
                resolved_tool = tool_map.get(tool_name)
            _append_error(
                message="Tool approval item missing call ID.",
                tool_call=tool_call,
                tool_name=tool_name,
                call_id="unknown",
                tool_origin=(
                    get_function_tool_origin(resolved_tool)
                    if isinstance(resolved_tool, FunctionTool)
                    else None
                ),
            )
            return None

        resolved_tool = tool_map.get(approval_key) if approval_key is not None else None
        if resolved_tool is None and tool_namespace is None:
            resolved_tool = tool_map.get(tool_name)
        approval_status = context_wrapper.get_approval_status(
            tool_name,
            call_id,
            tool_namespace=tool_namespace,
            existing_pending=interruption,
            tool_lookup_key=tool_lookup_key,
        )
        if approval_status is False:
            message = REJECTION_MESSAGE
            if isinstance(resolved_tool, FunctionTool):
                message = await resolve_approval_rejection_message(
                    context_wrapper=context_wrapper,
                    run_config=run_config,
                    tool_type="function",
                    tool_name=display_tool_name,
                    call_id=call_id,
                    tool_namespace=tool_namespace,
                    tool_lookup_key=tool_lookup_key,
                    existing_pending=interruption,
                )
            _append_error(
                message=message,
                tool_call=tool_call,
                tool_name=tool_name,
                call_id=call_id,
                tool_origin=(
                    get_function_tool_origin(resolved_tool)
                    if isinstance(resolved_tool, FunctionTool)
                    else None
                ),
            )
            return None

        if approval_status is not True:
            _append_error(
                message="Tool approval status unclear.",
                tool_call=tool_call,
                tool_name=tool_name,
                call_id=call_id,
                tool_origin=(
                    get_function_tool_origin(resolved_tool)
                    if isinstance(resolved_tool, FunctionTool)
                    else None
                ),
            )
            return None

        tool = resolved_tool
        if tool is None:
            _append_error(
                message=f"Tool '{display_tool_name}' not found.",
                tool_call=tool_call,
                tool_name=tool_name,
                call_id=call_id,
            )
            return None

        if not isinstance(tool, FunctionTool):
            _append_error(
                message=f"Tool '{display_tool_name}' is not a function tool.",
                tool_call=tool_call,
                tool_name=tool_name,
                call_id=call_id,
            )
            return None

        if not isinstance(tool_call, ResponseFunctionToolCall):
            _append_error(
                message=(
                    f"Tool '{tool_name}' approval item has invalid raw_item type for execution."
                ),
                tool_call=tool_call,
                tool_name=tool_name,
                call_id=call_id,
            )
            return None

        return tool_call, tool, tool_name, call_id

    for interruption in interruptions:
        resolved = await _resolve_tool_run(interruption)
        if resolved is None:
            continue
        tool_call, tool, tool_name, _ = resolved
        tool_runs.append(ToolRunFunction(function_tool=tool, tool_call=tool_call))

    if tool_runs:
        function_results, _, _ = await execute_function_tool_calls(
            bindings=bind_public_agent(agent),
            tool_runs=tool_runs,
            hooks=hooks,
            context_wrapper=context_wrapper,
            config=run_config,
        )
        for result in function_results:
            if isinstance(result.run_item, RunItemBase):
                generated_items.append(result.run_item)


# --------------------------
# Private helpers
# --------------------------


async def _execute_tool_input_guardrails(
    *,
    func_tool: FunctionTool,
    tool_context: ToolContext[Any],
    agent: Agent[Any],
    tool_input_guardrail_results: list[ToolInputGuardrailResult],
) -> str | None:
    """Execute input guardrails for a tool call and return a rejection message if any."""
    if not func_tool.tool_input_guardrails:
        return None

    for guardrail in func_tool.tool_input_guardrails:
        gr_out = await guardrail.run(
            ToolInputGuardrailData(
                context=tool_context,
                agent=agent,
            )
        )

        tool_input_guardrail_results.append(
            ToolInputGuardrailResult(
                guardrail=guardrail,
                output=gr_out,
            )
        )

        if gr_out.behavior["type"] == "raise_exception":
            raise ToolInputGuardrailTripwireTriggered(guardrail=guardrail, output=gr_out)
        elif gr_out.behavior["type"] == "reject_content":
            return gr_out.behavior["message"]

    return None


async def _execute_tool_output_guardrails(
    *,
    func_tool: FunctionTool,
    tool_context: ToolContext[Any],
    agent: Agent[Any],
    real_result: Any,
    tool_output_guardrail_results: list[ToolOutputGuardrailResult],
) -> Any:
    """Execute output guardrails for a tool call and return the final result."""
    if not func_tool.tool_output_guardrails:
        return real_result

    final_result = real_result
    for output_guardrail in func_tool.tool_output_guardrails:
        gr_out = await output_guardrail.run(
            ToolOutputGuardrailData(
                context=tool_context,
                agent=agent,
                output=real_result,
            )
        )

        tool_output_guardrail_results.append(
            ToolOutputGuardrailResult(
                guardrail=output_guardrail,
                output=gr_out,
            )
        )

        if gr_out.behavior["type"] == "raise_exception":
            raise ToolOutputGuardrailTripwireTriggered(guardrail=output_guardrail, output=gr_out)
        elif gr_out.behavior["type"] == "reject_content":
            final_result = gr_out.behavior["message"]
            break

    return final_result


def _normalize_exit_code(value: Any) -> int | None:
    """Convert arbitrary exit code types into an int if possible."""
    if value is None:
        return None
    try:
        return int(value)
    except (TypeError, ValueError):
        return None


def _is_hosted_mcp_approval_request(raw_item: Any) -> bool:
    """Detect hosted MCP approval request payloads emitted by the provider."""
    if isinstance(raw_item, McpApprovalRequest):
        return True
    if not isinstance(raw_item, dict):
        return False
    provider_data = raw_item.get("provider_data", {})
    return (
        raw_item.get("type") == "hosted_tool_call"
        and provider_data.get("type") == "mcp_approval_request"
    )
