import asyncio
from collections.abc import Awaitable, Callable
from typing import Any

from ...job import get_job_context
from ...llm import RealtimeModel, ToolFlag, Toolset, function_tool
from ...log import logger
from ...voice.events import CloseEvent, RunContext, SpeechCreatedEvent
from ...voice.speech_handle import SpeechHandle

END_CALL_DESCRIPTION = """
Ends the current call and disconnects immediately.

Call when:
- The user clearly indicates they are done (e.g., “that’s all, bye”).

Do not call when:
- The user asks to pause, hold, or transfer.
- Intent is unclear.

This is the final action the agent can take.
Once called, no further interaction is possible with the user.
Don't generate any other text or response when the tool is called.
"""


class EndCallTool(Toolset):
    def __init__(
        self,
        *,
        extra_description: str = "",
        delete_room: bool = True,
        end_instructions: str | None = "say goodbye to the user",
        ignore_on_enter: bool = False,
        on_tool_called: Callable[[Toolset.ToolCalledEvent], Awaitable[None]] | None = None,
        on_tool_completed: Callable[[Toolset.ToolCompletedEvent], Awaitable[None]] | None = None,
    ):
        """
        This tool allows the agent to end the call and disconnect from the room.

        Args:
            extra_description: Additional description to add to the end call tool.
            delete_room: Whether to delete the room when the user ends the call. deleting the room disconnects all remote users, including SIP callers.
            end_instructions: Tool output to the LLM for generating the tool response.
            ignore_on_enter: Hide the tool during ``on_enter`` so the model can't end the call while greeting.
            on_tool_called: Callback to call when the tool is called.
            on_tool_completed: Callback to call when the tool is completed.
        """
        end_call_tool = function_tool(
            self._end_call,
            name="end_call",
            description=f"{END_CALL_DESCRIPTION}\n{extra_description}",
            flags=ToolFlag.IGNORE_ON_ENTER if ignore_on_enter else ToolFlag.NONE,
        )
        super().__init__(id="end_call", tools=[end_call_tool])

        self._delete_room = delete_room
        self._extra_description = extra_description

        self._end_instructions = end_instructions
        self._on_tool_called = on_tool_called
        self._on_tool_completed = on_tool_completed

        self._shutdown_session_task: asyncio.Task[None] | None = None

    async def _end_call(self, ctx: RunContext) -> Any | None:
        logger.debug("end_call tool called")
        activity = ctx.session.current_agent._get_activity_or_raise()
        llm_v = activity.llm

        def _on_speech_done(_: SpeechHandle) -> None:
            # read auto_tool_reply_generation from the active realtime session so a fallback
            # adapter reports the model actually in use
            rt_session = activity.realtime_llm_session
            auto_tool_reply = (
                rt_session is not None and rt_session.capabilities.auto_tool_reply_generation
            )
            if not isinstance(llm_v, RealtimeModel) or not auto_tool_reply:
                # tool reply will reuse the same speech handle, so we can shutdown the session
                # directly after this speech handle is done
                ctx.session.shutdown()
            else:
                self._shutdown_session_task = asyncio.create_task(
                    self._delayed_session_shutdown(ctx)
                )

        ctx.speech_handle.add_done_callback(_on_speech_done)
        ctx.session.once("close", self._on_session_close)

        if self._on_tool_called:
            await self._on_tool_called(Toolset.ToolCalledEvent(ctx=ctx, arguments={}))

        completed_ev = Toolset.ToolCompletedEvent(ctx=ctx, output=self._end_instructions)
        if self._on_tool_completed:
            await self._on_tool_completed(completed_ev)

        return completed_ev.output

    async def _delayed_session_shutdown(self, ctx: RunContext) -> None:
        """Shutdown the session after the tool reply is played out"""
        speech_created_fut = asyncio.Future[SpeechHandle]()

        @ctx.session.once("speech_created")
        def _on_speech_created(ev: SpeechCreatedEvent) -> None:
            if not speech_created_fut.done():
                speech_created_fut.set_result(ev.speech_handle)

        try:
            speech_handle = await asyncio.wait_for(speech_created_fut, timeout=5.0)
            await speech_handle
        except asyncio.TimeoutError:
            logger.warning("tool reply timed out, shutting down session")
        finally:
            ctx.session.off("speech_created", _on_speech_created)
            ctx.session.shutdown()

    def _on_session_close(self, ev: CloseEvent) -> None:
        """Close the job process when AgentSession is closed"""
        if self._shutdown_session_task:
            # cleanup
            self._shutdown_session_task.cancel()
            self._shutdown_session_task = None

        job_ctx = get_job_context()

        if self._delete_room:

            async def _on_shutdown() -> None:
                logger.info("deleting the room because the user ended the call")
                await job_ctx.delete_room()

            job_ctx.add_shutdown_callback(_on_shutdown)

        # shutdown the job process
        job_ctx.shutdown(reason=ev.reason.value)