"""
Action executors used by the run loop. This module only houses XXXAction classes; helper
functions and approval plumbing live in tool_execution.py.
"""

from __future__ import annotations

import asyncio
import copy
import dataclasses
import inspect
import json
from typing import TYPE_CHECKING, Any, Literal, cast

from openai.types.responses import ResponseComputerToolCall
from openai.types.responses.response_input_item_param import (
    ComputerCallOutputAcknowledgedSafetyCheck,
)
from openai.types.responses.response_input_param import ComputerCallOutput

from .._tool_identity import get_mapping_or_attr, get_tool_trace_name_for_tool
from ..agent import Agent
from ..exceptions import ModelBehaviorError
from ..items import RunItem, ToolCallOutputItem
from ..logger import logger
from ..run_config import RunConfig
from ..run_context import RunContextWrapper
from ..tool import (
    ApplyPatchTool,
    ApplyPatchToolCustomDataContext,
    ComputerToolCustomDataContext,
    CustomTool,
    CustomToolCustomDataContext,
    LocalShellCommandRequest,
    ShellCommandRequest,
    ShellResult,
    resolve_computer,
)
from ..tool_context import ToolContext
from ..tracing import SpanError
from ..util import _coro
from ..util._approvals import evaluate_needs_approval_setting
from ..util._custom_data import maybe_extract_custom_data
from .items import apply_patch_rejection_item, shell_rejection_item
from .tool_execution import (
    coerce_apply_patch_operations,
    coerce_shell_call,
    extract_apply_patch_call_id,
    format_shell_error,
    get_trace_tool_error,
    normalize_apply_patch_result,
    normalize_max_output_length,
    normalize_shell_output,
    normalize_shell_output_entries,
    render_shell_outputs,
    resolve_approval_rejection_message,
    resolve_approval_status,
    serialize_shell_output,
    truncate_shell_outputs,
    with_tool_function_span,
)

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

__all__ = [
    "ComputerAction",
    "LocalShellAction",
    "ShellAction",
    "CustomToolAction",
    "ApplyPatchAction",
]


def _serialize_trace_payload(payload: Any) -> str:
    """Serialize tool payloads for tracing while tolerating non-JSON values."""
    if payload is None:
        return ""
    if isinstance(payload, str):
        return payload
    if hasattr(payload, "model_dump") and callable(payload.model_dump):
        return json.dumps(payload.model_dump(exclude_none=True))
    if dataclasses.is_dataclass(payload) and not isinstance(payload, type):
        return json.dumps(dataclasses.asdict(payload))
    try:
        return json.dumps(payload)
    except TypeError:
        return str(payload)


class ComputerAction:
    """Execute computer tool actions and emit screenshot outputs with hooks fired."""

    TRACE_TOOL_NAME = "computer"
    """Tracing should expose the GA computer tool alias."""

    @classmethod
    async def execute(
        cls,
        *,
        agent: Agent[Any],
        action: ToolRunComputerAction,
        hooks: RunHooks[Any],
        context_wrapper: RunContextWrapper[Any],
        config: RunConfig,
        acknowledged_safety_checks: list[ComputerCallOutputAcknowledgedSafetyCheck] | None = None,
    ) -> RunItem:
        """Run a computer action, capturing a screenshot and notifying hooks."""
        trace_tool_name = get_tool_trace_name_for_tool(action.computer_tool) or cls.TRACE_TOOL_NAME

        async def _run_action(span: Any | None) -> RunItem:
            if span and config.trace_include_sensitive_data:
                span.span_data.input = _serialize_trace_payload(
                    cls._get_trace_input_payload(action.tool_call)
                )

            computer = await resolve_computer(
                tool=action.computer_tool, run_context=context_wrapper
            )
            agent_hooks = agent.hooks
            await asyncio.gather(
                hooks.on_tool_start(context_wrapper, agent, action.computer_tool),
                (
                    agent_hooks.on_tool_start(context_wrapper, agent, action.computer_tool)
                    if agent_hooks
                    else _coro.noop_coroutine()
                ),
            )

            try:
                output = await cls._execute_action_and_capture(computer, action.tool_call)
            except Exception as exc:
                error_text = format_shell_error(exc)
                trace_error = get_trace_tool_error(
                    trace_include_sensitive_data=config.trace_include_sensitive_data,
                    error_message=error_text,
                )
                if span:
                    span.set_error(
                        SpanError(
                            message="Error running tool",
                            data={
                                "tool_name": trace_tool_name,
                                "error": trace_error,
                            },
                        )
                    )
                logger.error("Failed to execute computer action: %s", exc, exc_info=True)
                raise

            image_url = f"data:image/png;base64,{output}" if output else ""
            raw_item = ComputerCallOutput(
                call_id=action.tool_call.call_id,
                output={
                    "type": "computer_screenshot",
                    "image_url": image_url,
                },
                type="computer_call_output",
                acknowledged_safety_checks=acknowledged_safety_checks,
            )
            custom_data = await maybe_extract_custom_data(
                action.computer_tool.custom_data_extractor,
                ComputerToolCustomDataContext(
                    run_context=context_wrapper,
                    tool=action.computer_tool,
                    tool_call=action.tool_call,
                    output=image_url,
                    raw_item=copy.deepcopy(raw_item),
                ),
            )

            await asyncio.gather(
                hooks.on_tool_end(context_wrapper, agent, action.computer_tool, output),
                (
                    agent_hooks.on_tool_end(context_wrapper, agent, action.computer_tool, output)
                    if agent_hooks
                    else _coro.noop_coroutine()
                ),
            )

            if span and config.trace_include_sensitive_data:
                span.span_data.output = image_url

            return ToolCallOutputItem(
                agent=agent,
                output=image_url,
                raw_item=raw_item,
                custom_data=custom_data,
            )

        return await with_tool_function_span(
            config=config,
            tool_name=trace_tool_name,
            fn=_run_action,
        )

    @classmethod
    async def _execute_action_and_capture(
        cls, computer: Any, tool_call: ResponseComputerToolCall
    ) -> str:
        """Execute computer actions (sync or async drivers) and return the final screenshot."""

        async def maybe_call(method_name: str, *args: Any, **kwargs: Any) -> Any:
            method = getattr(computer, method_name, None)
            if method is None or not callable(method):
                raise ModelBehaviorError(f"Computer driver missing method {method_name}")
            filtered_kwargs = cls._filter_supported_kwargs(
                method_name=method_name,
                method=method,
                kwargs=kwargs,
            )
            result = method(*args, **filtered_kwargs)
            return await result if inspect.isawaitable(result) else result

        last_action_was_screenshot = False
        last_screenshot_result: Any = None
        for action in cls._iter_actions(tool_call):
            action_type = get_mapping_or_attr(action, "type")
            action_keys = cls._normalize_modifier_keys(get_mapping_or_attr(action, "keys"))
            last_action_was_screenshot = False
            if action_type == "click":
                await maybe_call(
                    "click",
                    get_mapping_or_attr(action, "x"),
                    get_mapping_or_attr(action, "y"),
                    get_mapping_or_attr(action, "button"),
                    keys=action_keys,
                )
            elif action_type == "double_click":
                await maybe_call(
                    "double_click",
                    get_mapping_or_attr(action, "x"),
                    get_mapping_or_attr(action, "y"),
                    keys=action_keys,
                )
            elif action_type == "drag":
                path = get_mapping_or_attr(action, "path") or []
                await maybe_call(
                    "drag",
                    [
                        (
                            cast(int, get_mapping_or_attr(point, "x")),
                            cast(int, get_mapping_or_attr(point, "y")),
                        )
                        for point in path
                    ],
                    keys=action_keys,
                )
            elif action_type == "keypress":
                await maybe_call("keypress", get_mapping_or_attr(action, "keys"))
            elif action_type == "move":
                await maybe_call(
                    "move",
                    get_mapping_or_attr(action, "x"),
                    get_mapping_or_attr(action, "y"),
                    keys=action_keys,
                )
            elif action_type == "screenshot":
                last_screenshot_result = await maybe_call("screenshot")
                last_action_was_screenshot = True
            elif action_type == "scroll":
                await maybe_call(
                    "scroll",
                    get_mapping_or_attr(action, "x"),
                    get_mapping_or_attr(action, "y"),
                    get_mapping_or_attr(action, "scroll_x"),
                    get_mapping_or_attr(action, "scroll_y"),
                    keys=action_keys,
                )
            elif action_type == "type":
                await maybe_call("type", get_mapping_or_attr(action, "text"))
            elif action_type == "wait":
                await maybe_call("wait")
            else:
                raise ModelBehaviorError(
                    f"Computer tool returned unknown action type {action_type!r}"
                )

        # Reuse the last screenshot action result when the batch already ended in a capture.
        if last_action_was_screenshot:
            return cast(str, last_screenshot_result)
        screenshot_result = await maybe_call("screenshot")
        return cast(str, screenshot_result)

    @staticmethod
    def _iter_actions(tool_call: ResponseComputerToolCall) -> list[Any]:
        if tool_call.actions:
            return list(tool_call.actions)
        if tool_call.action is not None:
            # The GA tool returns batched actions[], but released preview snapshots and older
            # Responses payloads may still carry a single action field.
            return [tool_call.action]
        return []

    @classmethod
    def _get_trace_input_payload(cls, tool_call: ResponseComputerToolCall) -> Any:
        actions = cls._iter_actions(tool_call)
        if tool_call.actions:
            return [cls._serialize_action_payload(action) for action in actions]
        if actions:
            return cls._serialize_action_payload(actions[0])
        return None

    @staticmethod
    def _serialize_action_payload(action: Any) -> Any:
        if hasattr(action, "model_dump") and callable(action.model_dump):
            return action.model_dump(exclude_none=True)
        if isinstance(action, dict):
            return dict(action)
        if dataclasses.is_dataclass(action) and not isinstance(action, type):
            return dataclasses.asdict(action)
        return action

    @staticmethod
    def _normalize_modifier_keys(keys: Any) -> list[str] | None:
        if not keys:
            return None
        return cast(list[str], keys)

    @classmethod
    def _filter_supported_kwargs(
        cls,
        *,
        method_name: str,
        method: Any,
        kwargs: dict[str, Any],
    ) -> dict[str, Any]:
        filtered_kwargs = {key: value for key, value in kwargs.items() if value is not None}
        if not filtered_kwargs:
            return {}

        supported_kwargs = cls._supported_keyword_arguments(method)
        unsupported_kwargs = [
            key
            for key in filtered_kwargs
            if key not in supported_kwargs and None not in supported_kwargs
        ]
        if unsupported_kwargs:
            logger.warning(
                "Computer driver method %r does not accept keyword argument(s) %s; "
                "dropping them and continuing.",
                method_name,
                ", ".join(sorted(unsupported_kwargs)),
            )
            for key in unsupported_kwargs:
                filtered_kwargs.pop(key, None)

        return filtered_kwargs

    @staticmethod
    def _supported_keyword_arguments(method: Any) -> set[str | None]:
        try:
            signature = inspect.signature(method)
        except (TypeError, ValueError):
            return set()
        supported: set[str | None] = {
            parameter.name
            for parameter in signature.parameters.values()
            if parameter.kind
            in {
                inspect.Parameter.KEYWORD_ONLY,
                inspect.Parameter.POSITIONAL_OR_KEYWORD,
            }
        }
        if any(
            parameter.kind == inspect.Parameter.VAR_KEYWORD
            for parameter in signature.parameters.values()
        ):
            supported.add(None)
        return supported


class LocalShellAction:
    """Execute local shell commands via the LocalShellTool with lifecycle hooks."""

    @classmethod
    async def execute(
        cls,
        *,
        agent: Agent[Any],
        call: ToolRunLocalShellCall,
        hooks: RunHooks[Any],
        context_wrapper: RunContextWrapper[Any],
        config: RunConfig,
    ) -> RunItem:
        """Run a local shell tool call and wrap the result as a ToolCallOutputItem."""
        agent_hooks = agent.hooks
        await asyncio.gather(
            hooks.on_tool_start(context_wrapper, agent, call.local_shell_tool),
            (
                agent_hooks.on_tool_start(context_wrapper, agent, call.local_shell_tool)
                if agent_hooks
                else _coro.noop_coroutine()
            ),
        )

        request = LocalShellCommandRequest(
            ctx_wrapper=context_wrapper,
            data=call.tool_call,
        )
        output = call.local_shell_tool.executor(request)
        result = await output if inspect.isawaitable(output) else output

        await asyncio.gather(
            hooks.on_tool_end(context_wrapper, agent, call.local_shell_tool, result),
            (
                agent_hooks.on_tool_end(context_wrapper, agent, call.local_shell_tool, result)
                if agent_hooks
                else _coro.noop_coroutine()
            ),
        )

        raw_payload: dict[str, Any] = {
            "type": "local_shell_call_output",
            "call_id": call.tool_call.call_id,
            "output": result,
        }
        return ToolCallOutputItem(
            agent=agent,
            output=result,
            raw_item=raw_payload,
        )


class ShellAction:
    """Execute shell calls, handling approvals and normalizing outputs."""

    @classmethod
    async def execute(
        cls,
        *,
        agent: Agent[Any],
        call: ToolRunShellCall,
        hooks: RunHooks[Any],
        context_wrapper: RunContextWrapper[Any],
        config: RunConfig,
    ) -> RunItem:
        """Run a shell tool call and return a normalized ToolCallOutputItem."""
        shell_call = coerce_shell_call(call.tool_call)
        shell_tool = call.shell_tool
        agent_hooks = agent.hooks

        async def _run_call(span: Any | None) -> RunItem:
            if span and config.trace_include_sensitive_data:
                span.span_data.input = _serialize_trace_payload(
                    dataclasses.asdict(shell_call.action)
                )

            needs_approval_result = await evaluate_needs_approval_setting(
                shell_tool.needs_approval, context_wrapper, shell_call.action, shell_call.call_id
            )

            if needs_approval_result:
                approval_status, approval_item = await resolve_approval_status(
                    tool_name=shell_tool.name,
                    call_id=shell_call.call_id,
                    raw_item=call.tool_call,
                    agent=agent,
                    context_wrapper=context_wrapper,
                    on_approval=shell_tool.on_approval,
                )

                if approval_status is False:
                    rejection_message = await resolve_approval_rejection_message(
                        context_wrapper=context_wrapper,
                        run_config=config,
                        tool_type="shell",
                        tool_name=shell_tool.name,
                        call_id=shell_call.call_id,
                    )
                    return shell_rejection_item(
                        agent,
                        shell_call.call_id,
                        rejection_message=rejection_message,
                    )

                if approval_status is not True:
                    return approval_item

            await asyncio.gather(
                hooks.on_tool_start(context_wrapper, agent, shell_tool),
                (
                    agent_hooks.on_tool_start(context_wrapper, agent, shell_tool)
                    if agent_hooks
                    else _coro.noop_coroutine()
                ),
            )
            request = ShellCommandRequest(ctx_wrapper=context_wrapper, data=shell_call)
            status: Literal["completed", "failed"] = "completed"
            output_text = ""
            shell_output_payload: list[dict[str, Any]] | None = None
            provider_meta: dict[str, Any] | None = None
            max_output_length: int | None = None
            requested_max_output_length = normalize_max_output_length(
                shell_call.action.max_output_length
            )

            try:
                executor = call.shell_tool.executor
                if executor is None:
                    raise ModelBehaviorError("Shell tool has no local executor configured.")
                executor_result = executor(request)
                result = (
                    await executor_result
                    if inspect.isawaitable(executor_result)
                    else executor_result
                )

                if isinstance(result, ShellResult):
                    normalized = [normalize_shell_output(entry) for entry in result.output]
                    result_max_output_length = normalize_max_output_length(result.max_output_length)
                    if result_max_output_length is None:
                        max_output_length = requested_max_output_length
                    elif requested_max_output_length is None:
                        max_output_length = result_max_output_length
                    else:
                        max_output_length = min(
                            result_max_output_length, requested_max_output_length
                        )
                    if max_output_length is not None:
                        normalized = truncate_shell_outputs(normalized, max_output_length)
                    output_text = render_shell_outputs(normalized)
                    if max_output_length is not None:
                        output_text = output_text[:max_output_length]
                    shell_output_payload = [serialize_shell_output(entry) for entry in normalized]
                    provider_meta = dict(result.provider_data or {})
                else:
                    output_text = str(result)
                    if requested_max_output_length is not None:
                        max_output_length = requested_max_output_length
                        output_text = output_text[:max_output_length]
            except Exception as exc:
                status = "failed"
                output_text = format_shell_error(exc)
                trace_error = get_trace_tool_error(
                    trace_include_sensitive_data=config.trace_include_sensitive_data,
                    error_message=output_text,
                )
                if span:
                    span.set_error(
                        SpanError(
                            message="Error running tool",
                            data={
                                "tool_name": shell_tool.name,
                                "error": trace_error,
                            },
                        )
                    )
                if requested_max_output_length is not None:
                    max_output_length = requested_max_output_length
                    output_text = output_text[:max_output_length]
                logger.error("Shell executor failed: %s", exc, exc_info=True)

            await asyncio.gather(
                hooks.on_tool_end(context_wrapper, agent, call.shell_tool, output_text),
                (
                    agent_hooks.on_tool_end(context_wrapper, agent, call.shell_tool, output_text)
                    if agent_hooks
                    else _coro.noop_coroutine()
                ),
            )

            raw_entries: list[dict[str, Any]] | None = None
            if shell_output_payload:
                raw_entries = shell_output_payload
            elif output_text:
                raw_entries = [
                    {
                        "stdout": output_text,
                        "stderr": "",
                        "status": status,
                        "outcome": "success" if status == "completed" else "failure",
                    }
                ]

            structured_output = normalize_shell_output_entries(raw_entries) if raw_entries else []

            raw_item: dict[str, Any] = {
                "type": "shell_call_output",
                "call_id": shell_call.call_id,
                "output": structured_output,
                "status": status,
            }
            if max_output_length is not None:
                raw_item["max_output_length"] = max_output_length
            if raw_entries:
                raw_item["shell_output"] = raw_entries
            if provider_meta:
                raw_item["provider_data"] = provider_meta

            if span and config.trace_include_sensitive_data:
                span.span_data.output = output_text

            return ToolCallOutputItem(
                agent=agent,
                output=output_text,
                raw_item=raw_item,
            )

        return await with_tool_function_span(
            config=config,
            tool_name=shell_tool.name,
            fn=_run_call,
        )


class CustomToolAction:
    """Execute Responses custom tool calls and return custom_tool_call_output items."""

    @classmethod
    async def execute(
        cls,
        *,
        agent: Agent[Any],
        call: ToolRunCustom,
        hooks: RunHooks[Any],
        context_wrapper: RunContextWrapper[Any],
        config: RunConfig,
    ) -> RunItem:
        custom_tool: CustomTool = call.custom_tool
        agent_hooks = agent.hooks
        call_id = get_mapping_or_attr(call.tool_call, "call_id")
        tool_input = get_mapping_or_attr(call.tool_call, "input")
        if not isinstance(call_id, str):
            raise ModelBehaviorError("Custom tool call is missing call_id.")
        if not isinstance(tool_input, str):
            raise ModelBehaviorError("Custom tool call is missing input.")

        tool_context = ToolContext.from_agent_context(
            context_wrapper,
            call_id,
            tool_name=custom_tool.name,
            tool_arguments=tool_input,
            agent=agent,
            run_config=config,
        )

        async def _run_call(span: Any | None) -> RunItem:
            if span and config.trace_include_sensitive_data:
                span.span_data.input = tool_input

            needs_approval_result = await evaluate_needs_approval_setting(
                custom_tool.runtime_needs_approval(), context_wrapper, tool_input, call_id
            )

            if needs_approval_result:
                approval_status, approval_item = await resolve_approval_status(
                    tool_name=custom_tool.name,
                    call_id=call_id,
                    raw_item=call.tool_call,
                    agent=agent,
                    context_wrapper=context_wrapper,
                    on_approval=custom_tool.runtime_on_approval(),
                )

                if approval_status is False:
                    rejection_message = await resolve_approval_rejection_message(
                        context_wrapper=context_wrapper,
                        run_config=config,
                        tool_type="custom",
                        tool_name=custom_tool.name,
                        call_id=call_id,
                    )
                    return cls._tool_output_item(agent, call_id, rejection_message)

                if approval_status is not True:
                    return approval_item

            await asyncio.gather(
                hooks.on_tool_start(tool_context, agent, custom_tool),
                (
                    agent_hooks.on_tool_start(tool_context, agent, custom_tool)
                    if agent_hooks
                    else _coro.noop_coroutine()
                ),
            )

            try:
                result = custom_tool.on_invoke_tool(tool_context, tool_input)
                result = await result if inspect.isawaitable(result) else result
                output_text = cls._normalize_output(result)
            except Exception as exc:
                output_text = format_shell_error(exc)
                trace_error = get_trace_tool_error(
                    trace_include_sensitive_data=config.trace_include_sensitive_data,
                    error_message=output_text,
                )
                if span:
                    span.set_error(
                        SpanError(
                            message="Error running tool",
                            data={
                                "tool_name": custom_tool.name,
                                "error": trace_error,
                            },
                        )
                    )
                logger.error("Custom tool failed: %s", exc, exc_info=True)

            raw_item = cls._raw_tool_output_item(call_id, output_text)
            custom_data = await maybe_extract_custom_data(
                custom_tool.custom_data_extractor,
                CustomToolCustomDataContext(
                    tool_context=tool_context,
                    tool=custom_tool,
                    input=tool_input,
                    output=output_text,
                    raw_item=copy.deepcopy(raw_item),
                ),
            )

            await asyncio.gather(
                hooks.on_tool_end(tool_context, agent, custom_tool, output_text),
                (
                    agent_hooks.on_tool_end(tool_context, agent, custom_tool, output_text)
                    if agent_hooks
                    else _coro.noop_coroutine()
                ),
            )

            if span and config.trace_include_sensitive_data:
                span.span_data.output = output_text
            return cls._tool_output_item(
                agent,
                call_id,
                output_text,
                raw_item=raw_item,
                custom_data=custom_data,
            )

        return await with_tool_function_span(
            config=config,
            tool_name=custom_tool.name,
            fn=_run_call,
        )

    @staticmethod
    def _normalize_output(output: Any) -> str:
        return output if isinstance(output, str) else str(output)

    @staticmethod
    def _raw_tool_output_item(call_id: str, output: str) -> dict[str, Any]:
        return {
            "type": "custom_tool_call_output",
            "call_id": call_id,
            "output": output,
        }

    @classmethod
    def _tool_output_item(
        cls,
        agent: Agent[Any],
        call_id: str,
        output: str,
        *,
        raw_item: dict[str, Any] | None = None,
        custom_data: dict[str, Any] | None = None,
    ) -> ToolCallOutputItem:
        return ToolCallOutputItem(
            agent=agent,
            output=output,
            raw_item=cast(Any, raw_item or cls._raw_tool_output_item(call_id, output)),
            custom_data=custom_data,
        )


class ApplyPatchAction:
    """Execute apply_patch operations with approvals and editor integration."""

    @classmethod
    async def execute(
        cls,
        *,
        agent: Agent[Any],
        call: ToolRunApplyPatchCall,
        hooks: RunHooks[Any],
        context_wrapper: RunContextWrapper[Any],
        config: RunConfig,
    ) -> RunItem:
        """Run an apply_patch call and serialize the editor result for the model."""
        apply_patch_tool: ApplyPatchTool = call.apply_patch_tool
        agent_hooks = agent.hooks
        operations = coerce_apply_patch_operations(
            call.tool_call,
            context_wrapper=context_wrapper,
        )
        call_id = extract_apply_patch_call_id(call.tool_call)

        async def _run_call(span: Any | None) -> RunItem:
            if span and config.trace_include_sensitive_data:
                span.span_data.input = _serialize_trace_payload(
                    [
                        {
                            "type": operation.type,
                            "path": operation.path,
                            "diff": operation.diff,
                        }
                        for operation in operations
                    ]
                )

            needs_approval_result = False
            for operation in operations:
                if await evaluate_needs_approval_setting(
                    apply_patch_tool.needs_approval, context_wrapper, operation, call_id
                ):
                    needs_approval_result = True
                    break

            if needs_approval_result:
                approval_status, approval_item = await resolve_approval_status(
                    tool_name=apply_patch_tool.name,
                    call_id=call_id,
                    raw_item=call.tool_call,
                    agent=agent,
                    context_wrapper=context_wrapper,
                    on_approval=apply_patch_tool.on_approval,
                )

                if approval_status is False:
                    rejection_message = await resolve_approval_rejection_message(
                        context_wrapper=context_wrapper,
                        run_config=config,
                        tool_type="apply_patch",
                        tool_name=apply_patch_tool.name,
                        call_id=call_id,
                    )
                    return apply_patch_rejection_item(
                        agent,
                        call_id,
                        output_type="apply_patch_call_output",
                        rejection_message=rejection_message,
                    )

                if approval_status is not True:
                    return approval_item

            await asyncio.gather(
                hooks.on_tool_start(context_wrapper, agent, apply_patch_tool),
                (
                    agent_hooks.on_tool_start(context_wrapper, agent, apply_patch_tool)
                    if agent_hooks
                    else _coro.noop_coroutine()
                ),
            )

            status: Literal["completed", "failed"] = "completed"
            output_text = ""

            try:
                operation_outputs: list[str] = []
                editor = apply_patch_tool.editor
                for operation in operations:
                    if operation.type == "create_file":
                        result = editor.create_file(operation)
                    elif operation.type == "update_file":
                        result = editor.update_file(operation)
                    elif operation.type == "delete_file":
                        result = editor.delete_file(operation)
                    else:  # pragma: no cover - validated in coerce_apply_patch_operations
                        raise ModelBehaviorError(
                            f"Unsupported apply_patch operation: {operation.type}"
                        )

                    awaited = await result if inspect.isawaitable(result) else result
                    normalized = normalize_apply_patch_result(awaited)
                    if normalized:
                        if normalized.status == "failed":
                            status = "failed"
                        elif normalized.status == "completed" and status != "failed":
                            status = "completed"
                        if normalized.output:
                            operation_outputs.append(normalized.output)
                output_text = "\n".join(operation_outputs)
            except Exception as exc:
                status = "failed"
                output_text = format_shell_error(exc)
                trace_error = get_trace_tool_error(
                    trace_include_sensitive_data=config.trace_include_sensitive_data,
                    error_message=output_text,
                )
                if span:
                    span.set_error(
                        SpanError(
                            message="Error running tool",
                            data={
                                "tool_name": apply_patch_tool.name,
                                "error": trace_error,
                            },
                        )
                    )
                logger.error("Apply patch editor failed: %s", exc, exc_info=True)

            raw_item: dict[str, Any] = {
                "type": "apply_patch_call_output",
                "call_id": call_id,
                "status": status,
            }
            if output_text:
                raw_item["output"] = output_text

            custom_data = await maybe_extract_custom_data(
                apply_patch_tool.custom_data_extractor,
                ApplyPatchToolCustomDataContext(
                    run_context=context_wrapper,
                    tool=apply_patch_tool,
                    operations=operations,
                    output=output_text,
                    status=status,
                    raw_item=copy.deepcopy(raw_item),
                ),
            )

            await asyncio.gather(
                hooks.on_tool_end(context_wrapper, agent, apply_patch_tool, output_text),
                (
                    agent_hooks.on_tool_end(context_wrapper, agent, apply_patch_tool, output_text)
                    if agent_hooks
                    else _coro.noop_coroutine()
                ),
            )

            if span and config.trace_include_sensitive_data:
                span.span_data.output = output_text

            return ToolCallOutputItem(
                agent=agent,
                output=output_text,
                raw_item=raw_item,
                custom_data=custom_data,
            )

        return await with_tool_function_span(
            config=config,
            tool_name=apply_patch_tool.name,
            fn=_run_call,
        )


__all__ = [
    "ComputerAction",
    "LocalShellAction",
    "ShellAction",
    "CustomToolAction",
    "ApplyPatchAction",
]
