Here’s a quick rundown for Prefect 3.x. Terminal vs non-terminal task state - Terminal states: COMPLETED, FAILED, CRASHED, CANCELLED - Non-terminal states: PENDING, SCHEDULED, RUNNING, PAUSED, CANCELLING - You don’t “mark” a task as terminal yourself — the engine does. You can inspect it via the state. Examples:

from prefect import task, flow

@task
def t():
    return 42

@flow
def f():
    fut = t.submit()
    st = fut.state              # a State object
    print(st.is_final())        # False initially
    fut.wait()                  # block until the task finishes
    print(fut.state.is_final()) # True
    print(fut.state.type)       # e.g. StateType.COMPLETED

How a flow decides when to finish - By default, a flow will not finalize until all tasks it submitted reach terminal states. - Returning futures from a flow also causes the engine to wait for them before finalizing the flow’s state. Waiting for a task to complete without crashing the flow on failure Use one of these patterns: 1) Just wait (don’t need the result)

@flow
def f():
    fut = t.submit()
    fut.wait()                  # waits; does not raise on failure
    if fut.state.is_failed():
        # handle failure, but the flow continues
        ...

2) Get the result but suppress raising

@flow
def f():
    fut = t.submit()
    value_or_exc = fut.result(raise_on_failure=False)
    if isinstance(value_or_exc, Exception):
        # handle failure; flow does not crash
        ...
    else:
        # success path
        ...

3) Wait for multiple tasks and handle partial failures

from prefect.futures import wait

@task
def maybe_fail(i):
    if i % 2 == 0:
        raise ValueError(f"boom on {i}")
    return i

@flow
def f():
    futs = maybe_fail.map(range(5))
    done, not_done = wait(futs)  # block until all finish (or use timeout=...)
    successes, failures = [], []

    for fut in done:
        if fut.state.is_completed():
            successes.append(fut.result())
        else:
            failures.append({
                "id": fut.task_run_id,
                "state": fut.state.type.value,
                "exception": fut.result(raise_on_failure=False),
                "message": fut.state.message,
            })

    return {"successes": successes, "failures": failures}

Handy APIs to remember -

PrefectFuture.wait()

— wait for completion; does not raise -

PrefectFuture.result(timeout=None, raise_on_failure=True)

— get value, optionally suppress raise -

prefect.futures.wait(futures, timeout=None)

— wait for many; returns (done, not_done) -

future.state.is_final()

,

is_completed()

,

is_failed()

, etc. — quick checks for state categories If you want the flow to keep running even if some tasks fail, avoid calling

future.result()

with the default

raise_on_failure=True

unless you wrap it in try/except, or use

raise_on_failure=False

and handle the exception object yourself.