That message usually means your flow submitted one or more tasks asynchronously with

.submit()

, but then the flow function returned before those task futures were resolved. In Prefect 3.x,

.submit()

returns a

PrefectFuture

. If your flow exits while that future is still running or pending, Prefect warns with:

Finished in state Pending() Please wait for all submitted tasks to complete before exiting your flow...

Why it happens - You called

my_task.submit(...)

- But you did not call

.wait()

or

.result()

on the returned future - So the flow ended before Prefect had a chance to finish the submitted task(s) Typical fix If you do not need the task result:

python
from prefect import flow, task

@task
def say_hello(name):
    print(f"Hello, {name}!")

@flow
def example_flow():
    future = say_hello.submit("Marvin")
    future.wait()

example_flow()

If you do need the result:

python
from prefect import flow, task

@task
def add(x, y):
    return x + y

@flow
def example_flow():
    future = add.submit(1, 2)
    result = future.result()
    print(result)

example_flow()

If you submit multiple tasks:

python
from prefect import flow, task

@task
def work(i):
    print(i)

@flow
def example_flow():
    futures = [work.submit(i) for i in range(5)]
    for future in futures:
        future.wait()

example_flow()

Common pattern that causes this error

python
@flow
def bad_flow():
    my_task.submit()
    # flow exits immediately here

What to check in your code - Are you storing the return value of every

.submit()

call? - Are you calling

.wait()

or

.result()

on all of them? - Are you returning from the flow too early? - If using loops/list comprehensions, are you waiting on the whole list of futures? If you want, paste your flow code and I can point to the exact line causing it.

Your message was too long, here's your message at the allowed limit: @flow(name= Dynamic Workflow Executor ) def dynamic_workflow_executor(workflow_config Dict[str Any] meta Dict[str Any]) Executes a workflow defined by the input JSON configuration Expects workflow_config format like { id my_dynamic_flow_1 description steps [ { id step_a type add params { x 1 y 2} depends_on []} { id step_b type print params { message Result } depends_on [ step_a ]} // Note For passing data see comments below ] } context = get_run_context() flow_run_id = None if isinstance(context FlowRunContext) flow_run_id = str(context flow_run id) if context and context flow_run and context flow_run id else None elif isinstance(context TaskRunContext) flow_run_id = str(context task_run flow_run_id) if context and context task_run and context task_run flow_run_id else None flow_id = workflow_config get( WorkflowName Unnamed Dynamic Flow ) execution_status_og sherlock_report_og = sh get_sherlock_logger( subscriptionId=meta get( subscription_id ) finonId=meta get( finon_id ) docId=meta get( document_id ) job_name= XFlowlogs job_id=meta get( job_id ) conversation_id=meta get( conversation_id ) ) execution_status = deepcopy(execution_status_og) sherlock_report = deepcopy(sherlock_report_og) sherlock_report SubJobType = dynamic_workflow_executor logger = get_run_logger() <http //logger info|logger info>(f --- Starting Dynamic Workflow {flow_id} --- ) steps = workflow_config get( Tasks []) if not steps logger warning( Workflow configuration contains no 'Tasks' Exiting ) return {} results_futures Dict[str PrefectFuture] = {} step_map = {step[ TaskName ] step for step in steps} # For easy lookup # --- Dependency Resolution (using networkx) --- # Build a graph to easily iterate in topological order graph = nx DiGraph() for step in steps # step_id = step['TaskId'] step_id = step[ TaskName ] # TODO Assuming TaskName is unique graph add_node(step_id) for dep_id in step get( DependsOn []) if dep_id in step_map graph add_edge(dep_id step_id) # Edge from dependency TO current step else # Optionally raise an error here logger error( f Excecutor_Flow Dependency '{dep_id}' for step '{step_id}' not found in steps definition Flow may fail ) if not nx is_directed_acyclic_graph(graph) cycles = list(nx simple_cycles(graph)) logger error( f Excecutor_Flow Workflow definition contains cycles {cycles} Cannot execute ) raise ValueError(f Workflow definition contains cycles {cycles} ) # --- Execute tasks in dependency order --- execution_order = list(nx topological_sort(graph)) <http //logger info|logger info>(f Execution order determined {execution_order} ) final_results = {} # Store actual results if needed black_listed_routes = set() for step_id in execution_order step_info = step_map[step_id] step_type = step_info[ TaskType ] step_params = step_info get( Parameters {}) <http //logger info|logger info>(f Executing Step {step_id} (Type {step_type}) ) # Find the corresponding Prefect task function task_func = TASK_REGISTRY get(step_type) if not task_func msg = f Unknown step type '{step_type}' for step '{step_id}' logger error(f Excecutor_Flow {msg} ) raise ValueError(msg) # Gather dependency futures for waiting dependencies = step_info get( DependsOn []) wait_for = [] dependecy_list = [] for dep_id in dependencies if step_map[dep_id][ TaskId ] in black_listed_routes if if_anyone in step_info[ Parameters ] flag = step_info[ Parameters ][ if_anyone ] if flag is False black_listed_routes add(step_info[ TaskId ]) if dep_id in results_futures wait_for append(results_futures[dep_id]) dependecy_list append(dep_id) # Check if the current step is in the blacklisted routes if step_info[ TaskId ] in black_listed_routes <http //logger info|logger info>(f Step '{step_id}'