xnano.utils.dispatch

def run_awaitable(awaitable: Awaitable[Any]) -> Any:
    """Drive an async hook result to completion on a free event loop.

    Args:
        awaitable: A coroutine or other awaitable returned by a hook.

    Returns:
        The awaitable's result.

    Raises:
        RuntimeError: If called while an asyncio event loop is already
            running on this thread (the sync run loop cannot nest
            ``asyncio.run``).
        TypeError: If ``awaitable`` is not a coroutine object.
    """
    # Async hooks are optional. Keep asyncio and its comparatively large
    # import graph out of the synchronous startup and first-render path.
    import asyncio

    try:
        asyncio.get_running_loop()
    except RuntimeError:
        if inspect.iscoroutine(awaitable):
            return asyncio.run(cast(Coroutine[Any, Any, Any], awaitable))

        async def _drain() -> Any:
            return await awaitable

        return asyncio.run(_drain())
    raise RuntimeError(
        "async @on_* hooks cannot run while an asyncio event loop is "
        "already active on this thread; call Runtime.live() from a "
        "sync context, or wrap it with asyncio.to_thread(...)."
    )

def invoke_hook(handler: Any, bound_self: Any, ctx: "Context[Any]") -> Any:
    """Invoke ``handler`` with the right arity, awaiting async results.

    Uncaught exceptions (other than ``Exit`` / ``KeyboardInterrupt`` /
    ``SystemExit``) are logged at ERROR and re-raised so the run loop
    can restore the host terminal on the way out.
    """
    name = getattr(handler, "__qualname__", repr(handler))
    try:
        result = _call_hook(handler, bound_self, ctx)
        if inspect.isawaitable(result):
            return run_awaitable(result)
        return result
    except Exit:
        raise
    except (KeyboardInterrupt, SystemExit):
        raise
    except Exception:
        _logger.exception("Uncaught exception in hook %s", name)
        raise