Thanks for the clear MRE — your code is conceptually fine. The failure you’re seeing is due to a known serialization bug that appears when a task tries to launch another background task (i.e., calling

task.delay()

from inside a running task), or when a flow is executed from a background task and then calls

.delay()

. The error you posted matches what we’ve seen elsewhere: - Launching a task from a task causes 'MockValSer' serialization error - Related context for the same Pydantic error in other nested/edge cases: serialization on cancel and flow-run cleanup/cancel What’s “wrong” with the code - Nothing logically — it’s hitting the nested-background-task serialization bug. - In Prefect 3.x today, background task submissions should be initiated from a flow (not from another task). Submitting background tasks from inside tasks is not reliable due to the serialization of the run context. Recommended patterns that work today 1) Move background orchestration into a flow - Have a flow submit background tasks via

.delay()

and await with

.result_async()

(as you’re already doing in other places). - Serve only the leaf tasks with the Task Worker. Working version of your example:

from prefect import flow, task
from prefect.task_worker import serve
import time
import asyncio

@task
def dummy_task(input: str):
    print(f"Running dummy task with input: {input}")
    time.sleep(1)
    return f"processed: {input}"

@flow
async def background_flow(input: str):
    sub_future = dummy_task.with_options(name=f"dummy-{input}").delay(input)
    result = await sub_future.result_async()
    return result

if __name__ == "__main__":
    # Serve only the task(s)
    serve(dummy_task, limit=10)

And from your initiator:

# Run as a flow (ideally via a deployment / worker)
result = background_flow("test")
print(result)

2) If you must start from a “task”, call a flow from the task and do

.delay()

inside that flow - Not my first choice, but works more reliably because the flow boundary owns orchestration:

from prefect import flow, task

@flow
async def orchestrate(input: str):
    fut = dummy_task.delay(input)
    return await fut.result_async()

@task
def background_task(input: str):
    # Call the flow instead of calling .delay() directly here
    return orchestrate(input)

- Serve only

dummy_task

with the Task Worker; run the flow via a standard Prefect worker/deployment. 3) Fan-out directly from flows (preferred for load distribution) - From a flow, launch many background tasks across your Task Workers:

@flow
async def router(inputs: list[str]):
    futures = [
        dummy_task.with_options(name=f"dummy-{i}").delay(x)
        for i, x in enumerate(inputs)
    ]
    results = await asyncio.gather(*[f.result_async() for f in futures])
    return results

- Run multiple Task Worker processes, each

serve

-ing the specific tasks they can handle. Prefect will assign pending background tasks to any worker serving that task. Keep using

.with_options(name=...)

for dynamic run names. Notes on routing and distribution - Task Workers pick up only the tasks they are serving (i.e., the functions you pass to

serve(...)

). To route work, split task functions by capability and start workers that serve different sets. Running multiple workers that serve the same task horizontally scales throughput. - Tags are great for concurrency limits but they do not route background tasks to specific Task Workers; routing is based on which task function a worker serves. - Use

limit

in

serve(..., limit=N)

to cap concurrency per worker. Next steps and questions - Are you able to move the nested

.delay()

to a flow boundary as shown above? That should remove the serialization error.

- If you need more explicit routing rules (“who serves what”), can you share how you decide which worker should get which jobs? We can suggest a task sharding pattern that aligns with Task Workers’ “serve specific functions” model. References - Background tasks docs: Background tasks in Prefect 3 - Known issue for nested task submission: Issue #16422

im also curious about the use case

@task
def background_task(input: str):
    print(f"Running background task with input: {input}")
    sub_task = dummy_task.delay(input)
    return sub_task.result()

why

delay

dummy_task if you block and resolve it immediately with

.result()

?

Interestingly, this is not set to True for flow context serialization..

I don't know why it says "prefect-cloud": I'm running the server locally.

What is quite weird, is that I have done a very similar pattern in my code that worked already quite well: delay -> background_task -> Flow -> mix of background_tasks / submitted tasks / simple calls But now I just get that error whatever I do.. which makes me crazy for 24h 😅

Yes it does throw

let me send you the files:

running them with

PYTHONPATH=./ uv run python dummy_task.py

and

PYTHONPATH=./ uv run python call_task.py

in my project

but the bug only manifest when you do the normal/correct thing and have separate files

i should have a PR to fix this shortly

If I can abuse from you: this bug marked as solved, was not properly fixed and the race condition is still there.. I reproduce the issue a lot, randomly. I wrote a comment to explain my point but I'm not sure it anyone look at closed bug or if I should open a new one: https://github.com/PrefectHQ/prefect/issues/18278#issuecomment-3244381547 This is quite annoying, again when using background tasks 🙂