"""Shared async utilities for coverage step files — thread-safe event loop isolation."""

import asyncio
import threading


def _run(coro):
    """Run *coro* to completion in a **fresh, thread-local** event loop.

    Every call creates a brand new loop and runs the coroutine on it.
    After completion, the loop is closed and the previous asyncio state
    is restored.  This means tests can run in any order, in any thread,
    without ever sharing or corrupting each other's event loop state.

    Slipcover instruments the actual function calls inside the
    coroutine, so coverage is tracked regardless of which loop
    executes them.
    """
    # Remember what was there before
    old_loop = None
    try:
        old_loop = asyncio.get_running_loop()
    except RuntimeError:
        pass

    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    try:
        return loop.run_until_complete(coro)
    finally:
        _cancel_pending(loop)
        loop.close()
        # Restore the old loop if one was running
        if old_loop and not old_loop.is_closed():
            asyncio.set_event_loop(old_loop)


def _cancel_pending(loop):
    try:
        pending = asyncio.all_tasks(loop)
        if not pending:
            return
        for t in pending:
            t.cancel()
        loop.run_until_complete(
            asyncio.gather(*pending, return_exceptions=True),
        )
    except Exception:
        pass


async def _catch(fn, *args, **kwargs):
    try:
        return await fn(*args, **kwargs)
    except Exception as exc:
        return f"{type(exc).__name__}: {exc}"