import asyncio
from contextlib import asynccontextmanager

if hasattr(asyncio, "timeout"):

    @asynccontextmanager
    async def timeout_context(timeout_seconds):
        with asyncio.timeout(timeout_seconds) as ctx:
            yield ctx

else:

    @asynccontextmanager
    async def timeout_context(timeout_seconds):
        try:
            yield await asyncio.wait_for(asyncio.Future(), timeout=timeout_seconds)
        except asyncio.TimeoutError as e:
            msg = f"Operation timed out after {timeout_seconds} seconds"
            raise TimeoutError(msg) from e


def run_until_complete(coro):
    try:
        asyncio.get_running_loop()
    except RuntimeError:
        # If there's no event loop, create a new one and run the coroutine
        return asyncio.run(coro)
    # If there's already a running event loop, we can't call run_until_complete on it.
    # Instead, run the coroutine in a new thread with a new event loop. Propagate the
    # caller's contextvars context so request-scoped state (e.g. lfx serve request
    # variables and the no-env-fallback flag) stays visible inside the worker thread;
    # a bare ThreadPoolExecutor would otherwise reset every ContextVar to its default.
    import concurrent.futures
    import contextvars

    ctx = contextvars.copy_context()

    def run_in_new_loop():
        new_loop = asyncio.new_event_loop()
        asyncio.set_event_loop(new_loop)
        try:
            return new_loop.run_until_complete(coro)
        finally:
            new_loop.close()

    with concurrent.futures.ThreadPoolExecutor() as executor:
        future = executor.submit(ctx.run, run_in_new_loop)
        return future.result()
