Hi, I am getting started in prefect and some help would be really helpful. I have a flow that runs two async sub-flows, both subflows call tasks that are async with a time sleep. The first task of my 2 concurrent subflows executes in parallel. Then each task after that first task is run sequentially. I can’t seem to figure out the issue.

@flow
async def run_sleep_multi_flow():
    flows = [
        wait_flow_1(),
        wait_flow_2(),
    ]
    await asyncio.gather(*flows)

@flow
async def wait_flow_1():
    await sleep_task_async(5)
    await sleep_task_async(5)
    await sleep_task_async(5)


@flow
async def wait_flow_2():
    await sleep_task_async(5)
    await sleep_task_async(5)
    await sleep_task_async(5)

@task
async def sleep_task_async(time_in_seconds):
    await asyncio.sleep(time_in_seconds)
    return "Done sleep"

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

what version of prefect do you have installed? this looks familiar, but like something we fixed

thank you, version 2.14.15

if there isn't anything blocking you from upgrading to the latest release, it should be fixed there