Great question — and your observation is correct. In Prefect 3.x, a task submitted with

.submit()

will not affect the flow’s outcome unless its future is resolved somewhere. A future is “resolved” when you: - call

.result()

on it, - pass it as an input to another task/flow, - return it (or a structure containing it) from the flow, or - explicitly resolve it with a helper. There isn’t a global setting to “auto-fail the flow if any submitted task fails.” You need to resolve those futures. Here are the cleanest ways to do that without manually pulling states: Option 1: Return the futures from the flow - Prefect will automatically wait on and resolve any futures in the flow’s return value, and the flow will fail if any task failed.

from prefect import flow, task

@task
def do_work(x):
    if x == 3:
        raise ValueError("boom")
    return x * 2

@flow
def my_flow():
    futures = [do_work.submit(i) for i in range(5)]
    return futures  # Flow will fail if any of these tasks failed

Option 2: Explicitly resolve futures at the end - This resolves nested structures and raises on the first failure, causing the flow to fail.

from prefect import flow, task
from prefect.futures import resolve_futures_to_results

@task
def do_work(x):
    if x == 3:
        raise ValueError("boom")
    return x * 2

@flow
def my_flow():
    futures = [do_work.submit(i) for i in range(5)]
    # Ensure any failure propagates to the flow
    resolve_futures_to_results(futures)
    return "done"

Option 3: Fail-fast while processing results as they complete - Useful if you want to start handling successes immediately and stop on first failure.

from prefect import flow, task
from prefect.futures import as_completed

@task
def do_work(x):
    if x == 3:
        raise ValueError("boom")
    return x * 2

@flow
def my_flow():
    futures = [do_work.submit(i) for i in range(5)]
    for fut in as_completed(futures):
        # Raises on failure, which fails the flow
        _ = fut.result()
    return "done"

Option 4: Wait for all, then raise if any failed

from prefect import flow, task
from prefect.futures import wait

@task
def do_work(x):
    if x == 3:
        raise ValueError("boom")
    return x * 2

@flow
def my_flow():
    futures = [do_work.submit(i) for i in range(5)]
    done, _ = wait(futures)
    # Calling .result() on each done future will raise on failures
    for f in done:
        _ = f.result()
    return "done"

Quick summary: - No automatic failure for “fire-and-forget” futures. - To get automatic propagation, either return the futures or call

resolve_futures_to_results(futures)

before the flow ends. - Passing a future into another task/flow will also implicitly wait and propagate failure. Which behavior do you want: fail-fast on the first error or wait for all tasks and then fail if any failed? I can tailor an example to your exact pattern.

Yes — you have similar options for subflows in Prefect 3.x. It depends on how you call them: - Call the subflow directly (default): parent fails immediately if the subflow fails

from prefect import flow

@flow
def child():
    raise RuntimeError("boom")

@flow
def parent():
    child()  # raises immediately if child fails; parent flow fails

- Defer failure to the end (inspect/aggregate states): - Ask the subflow for its State with

return_state=True

, then return those states (the “aggregate rule” will fail the parent if any returned state is not completed)

from prefect import flow

@flow
def child_ok():
    return 1

@flow
def child_fail():
    raise ValueError("nope")

@flow
def parent():
    s1 = child_ok(return_state=True)
    s2 = child_fail(return_state=True)
    # Returning states triggers the aggregate rule → parent fails if any failed
    return [s1, s2]

- Run subflows concurrently and fail if any failed: - Use the experimental

.submit()

on flows (returns a PrefectFlowRunFuture) and resolve them before finishing

from prefect import flow
from prefect.futures import resolve_futures_to_results

@flow
def child(x):
    if x == 3:
        raise ValueError("boom")
    return x

@flow
def parent():
    futs = [child.submit(i) for i in range(5)]
    # Raises if any subflow failed → parent fails
    resolve_futures_to_results(futs)
    return "done"

Notes: - Returning futures also works: if you return the flow futures from the parent, Prefect will resolve them when determining the parent’s final state and fail the parent if any subflow failed. - If you want to continue running but still fail the parent at the end, prefer

return_state=True

+ return the states (or use

resolve_futures_to_results

on subflow futures). -

.submit()

for flows is experimental and returns a

PrefectFlowRunFuture

with `.result()`/`.wait()` semantics similar to task futures. Do you typically call subflows directly, or are you submitting them concurrently and not awaiting them until later? I can tailor the pattern to your exact setup.

Great question. In Prefect 3.x you’ve got a few clean ways to wait for a batch of task futures without raising, and then detect failures. Fastest/cleanest for “bulk wait and inspect” is to use the

PrefectFutureList.result(raise_on_failure=False)

API (mapping returns a PrefectFutureList). It blocks until all futures finish, does not raise on failures, and preserves original order. Example:

from prefect import flow, task

@task
def work(x):
    if x == 3:
        raise ValueError("boom")
    return x * 10

@flow
def run():
    futures = work.map([1, 2, 3, 4])

    # Wait for all without raising; failed items return Exception objects
    results = futures.result(raise_on_failure=False)

    # Check for failures
    any_failed = any(isinstance(r, BaseException) for r in results)

    # (Optional) separate successes and failures
    successes = [r for r in results if not isinstance(r, BaseException)]
    failures = [(i, r) for i, r in enumerate(results) if isinstance(r, BaseException)]

    print("any_failed:", any_failed)
    print("successes:", successes)
    print("failures:", failures)

run()

Notes: -

results

will contain real return values for successful tasks and Exception objects for failed tasks. - Order matches your input. If you don’t have a PrefectFutureList (i.e., you have a plain list of futures), you can wrap it:

from prefect.futures import PrefectFutureList

results = PrefectFutureList(list_of_futures).result(raise_on_failure=False)

Alternative: wait + check states (no result deserialization) -

wait()

never raises and returns which futures are done vs. still running. -

resolve_futures_to_states()

converts futures to their State objects so you can check

is_failed()

/

is_completed()

etc.

from prefect import flow, task
from prefect.futures import wait, resolve_futures_to_states
from prefect.states import get_state_exception

@task
def work(x):
    if x == 3:
        raise ValueError("boom")
    return x * 10

@flow
def run():
    futures = work.map([1, 2, 3, 4])

    # Wait for completion (no raise)
    done, not_done = wait(futures)

    # Inspect states (works even if results aren't persisted)
    states = resolve_futures_to_states(futures)

    any_failed = any(s.is_failed() or s.is_crashed() for s in states)
    failures = []
    for i, s in enumerate(states):
        if s.is_failed() or s.is_crashed():
            failures.append((i, get_state_exception(s)))  # extract the exception safely

    print("any_failed:", any_failed)
    print("failures:", failures)

run()

Quick guidance: - Use

futures.result(raise_on_failure=False)

when you want to block until all complete and collect results in input order without raising. - Use

wait()

+ state inspection when you want to avoid pulling results (e.g., large payloads) and just know which failed/succeeded. - Avoid

resolve_futures_to_results

if you don’t want an exception on any failure—it raises as soon as it encounters one. If you share how you’re getting your futures (mapped tasks vs. a list you built elsewhere), I can tailor the snippet to your exact case.