import random

from prefect import flow, task
from prefect.utilities.asyncutils import run_sync_in_worker_thread


@task
def my_random():
    return random.random()


@flow
def my_subflow():
    result = my_random()
    return result


@flow
async def my_test_flow():
    subflows = [run_sync_in_worker_thread(deepcopy(my_subflow)) for _ in range(10)]
    results = await asyncio.gather(*subflows)
    print(results)


if __name__ == "__main__":
    asyncio.run(my_test_flow())

is the code, when i run against my local prefect everything is fine, but when i

prefect cloud login

then it falls apart

here's the full stack trace

the tl;dr here is that i have a bunch of subflows i want to start in worker threads on the machine, and i guess i'm just a bit confused the best way to do this

i'm running in EKS and the top level flow is in a pod, i don't necessarily want to spin up a lot of other pods for each of those subflows because they're actually quite small. i'll be running thousands of the parent flow though