handler.py

# SPDX-License-Identifier: MIT
# Copyright (c) 2026 LlamaIndex Inc.

from __future__ import annotations

import asyncio
import functools
import warnings
from collections.abc import Generator
from typing import TYPE_CHECKING, Any, AsyncGenerator, Awaitable

from workflows.runtime.types.plugin import (
    ExternalRunAdapter,
    as_v2_runtime_compatibility_shim,
)

from .errors import WorkflowCancelledByUser, WorkflowRuntimeError
from .events import Event, InternalDispatchEvent, StopEvent, WorkflowCancelledEvent
from .types import RunResultT

if TYPE_CHECKING:
    from .context import Context
    from .workflow import Workflow


class WorkflowHandler(Awaitable[RunResultT]):
    """
    Stable interface for communicating with a running workflow. Is awaitable and streamable, and supports things like cancellation.
    """

    _ctx: Context

    async def _await_result(self) -> RunResultT:
        stop_event = await self.stop_event_result()
        return stop_event.result if type(stop_event) is StopEvent else stop_event

    def __await__(self) -> Generator[Any, Any, RunResultT]:
        return self._await_result().__await__()

    def __init__(
        self,
        workflow: "Workflow",
        external_adapter: ExternalRunAdapter,
        ctx: "Context[Any] | None" = None,
    ) -> None:
        from .context import Context

        self._workflow = workflow
        self._external_adapter = external_adapter
        # TODO(v3): Remove ctx parameter. The handler will just be the external face.
        self._ctx = (
            ctx
            if ctx is not None
            else Context._create_external(
                workflow=workflow, external_adapter=external_adapter
            )
        )
        self.run_id = external_adapter.run_id
        self._all_events_consumed = False
        self._result: StopEvent | None = None
        self._result_exception: BaseException | None = None
        self._result_task = asyncio.create_task(self._wait_for_result())
        self._result_task.add_done_callback(self._handle_result_task_done)

    async def _wait_for_result(self) -> StopEvent:
        result = await self._external_adapter.get_result()
        self._result = result
        return result

    def _handle_result_task_done(self, task: asyncio.Task[StopEvent]) -> None:
        if task.cancelled():
            return
        try:
            exc = task.exception()
        except asyncio.CancelledError:
            return
        if exc is None:
            return
        self._result_exception = exc
        if isinstance(exc, WorkflowCancelledByUser) and self._result is None:
            # Preserve cancellation in handler state without changing await semantics.
            self._result = WorkflowCancelledEvent()

    @property
    def ctx(self) -> Context:
        """The workflow [Context][workflows.context.context.Context] for this run."""
        return self._ctx

    def get_stop_event(self) -> StopEvent | None:
        """The stop event for this run. Always defined once the future is done. In a future major release, this will be removed, and the result will be the stop event itself."""
        return self._result

    async def stop_event_result(self) -> StopEvent:
        """Get the stop event for this run. Always defined once the future is done. In a future major release, this will be removed, and the result will be the stop event itself."""
        return await self._result_task

    def __str__(self) -> str:
        return f"WorkflowHandler(workflow={self._workflow.workflow_name}, run_id={self.run_id}, result={self._result})"

    def is_done(self) -> bool:
        """Return True when the workflow has completed."""
        return self._result_task.done()

    async def stream_events(
        self, expose_internal: bool = False
    ) -> AsyncGenerator[Event, None]:
        """
        Stream events from the workflow execution as they occur.

        This method provides real-time access to events generated during workflow
        execution, allowing for monitoring and processing of intermediate results.
        Events are yielded in the order they are generated by the workflow.

        The stream includes all events written to the context's streaming queue,
        and terminates when a [StopEvent][workflows.events.StopEvent] is
        encountered, indicating the workflow has completed.

        Args:
            expose_internal (bool): Whether to expose internal events.

        Returns:
            AsyncGenerator[Event, None]: An async generator that yields Event objects
                as they are produced by the workflow.

        Raises:
            ValueError: If the context is not set on the handler.
            WorkflowRuntimeError: If all events have already been consumed by a
                previous call to `stream_events()` on the same handler instance.

        Examples:
            ```python
            handler = workflow.run()

            # Stream and process events in real-time
            async for event in handler.stream_events():
                if isinstance(event, StopEvent):
                    print(f"Workflow completed with result: {event.result}")
                else:
                    print(f"Received event: {event}")

            # Get final result
            result = await handler
            ```

        Note:
            Events can only be streamed once per handler instance. Subsequent
            calls to `stream_events()` will raise a WorkflowRuntimeError.
        """

        # Check if we already consumed all the streamed events
        if self._all_events_consumed:
            msg = "All the streamed events have already been consumed."
            raise WorkflowRuntimeError(msg)

        async for ev in self.ctx.stream_events():
            if isinstance(ev, InternalDispatchEvent) and not expose_internal:
                continue
            yield ev

            if isinstance(ev, StopEvent):
                self._all_events_consumed = True
                break

    def done(self) -> bool:
        """Return True when the workflow has completed."""
        _warn_done_deprecated()
        return self._result_task.done()

    def cancel(self) -> None:
        """Cancel the running workflow."""
        _warn_cancel_deprecated()
        shim = as_v2_runtime_compatibility_shim(self._external_adapter)
        if shim is None:
            raise NotImplementedError(
                "Hard cancel is not supported by this runtime. "
                "Use await handler.cancel_run() for graceful cancellation."
            )
        shim.abort()
        self._result_task.cancel()

    def exception(self) -> BaseException | None:
        """Get the exception for this run. Always defined once the future is done."""
        _warn_exception_deprecated()
        try:
            return self._result_task.exception()
        except asyncio.CancelledError:
            return None

    def cancelled(self) -> bool:
        """Return True when the underlying workflow has been cancelled."""
        _warn_cancelled_deprecated()
        if self._result_task.cancelled():
            return True
        exc = self.exception()
        if exc is not None and isinstance(exc, WorkflowCancelledByUser):
            return True
        stop_event = self.get_stop_event()
        if stop_event is not None and isinstance(stop_event, WorkflowCancelledEvent):
            return True
        return False

    async def cancel_run(self, *, timeout: float = 5.0) -> None:
        """Cancel the running workflow.

        Signals the underlying context to raise
        [WorkflowCancelledByUser][workflows.errors.WorkflowCancelledByUser],
        which will be caught by the workflow and gracefully end the run.

        Examples:
            ```python
            handler = workflow.run()
            await handler.cancel_run()
            ```
        """
        try:
            await self._external_adapter.cancel()
        except Exception:
            pass
        try:
            await asyncio.wait_for(self._result_task, timeout=timeout)
        except asyncio.TimeoutError:
            pass
        except asyncio.CancelledError:
            pass
        except Exception:
            pass

    async def send_event(self, event: Event, step: str | None = None) -> None:
        """Send an event into the workflow.

        Args:
            event: The event to send into the workflow.
            step: Optional step name to target. If None, broadcasts to all.
        """
        self.ctx.send_event(event, step)


@functools.lru_cache(maxsize=1)
def _warn_done_deprecated() -> None:
    warnings.warn(
        "WorkflowHandler.done() is deprecated and will be removed in a future release",
        DeprecationWarning,
        stacklevel=2,
    )


@functools.lru_cache(maxsize=1)
def _warn_cancel_deprecated() -> None:
    warnings.warn(
        "WorkflowHandler.cancel() is deprecated and will be removed in a future release. Prefer to cancel the underlying workflow with await handler.cancel_run(), and then awaiting the result with await handler to obtain the cancellation exception.",
        DeprecationWarning,
        stacklevel=2,
    )


@functools.lru_cache(maxsize=1)
def _warn_exception_deprecated() -> None:
    warnings.warn(
        "WorkflowHandler.exception() is deprecated and will be removed in a future release",
        DeprecationWarning,
        stacklevel=2,
    )


@functools.lru_cache(maxsize=1)
def _warn_cancelled_deprecated() -> None:
    warnings.warn(
        "WorkflowHandler.cancelled() is deprecated and will be removed in a future release",
        DeprecationWarning,
        stacklevel=2,
    )