Great question. Assuming Prefect 3.x, you can branch on a task’s failure without blocking by using

allow_failure

and having the downstream tasks self-skip based on the upstream state. Key idea: - Submit

task1

as usual. - Wrap its future with

allow_failure(...)

so downstream tasks will still be scheduled even if

task1

fails, and they’ll receive either the value (on success) or a

State

object (on failure). - Have two downstream tasks: one that runs only when

task1

succeeded, and one that runs only when

task1

failed. Each checks the input and raises

Abort

to skip itself when its condition is not met. Example:

from prefect import flow, task, allow_failure
from prefect.states import State, StateType
from prefect.exceptions import Abort

@task
def task1():
    # your logic here
    ...

@task
def task2_only_if_success(x):
    # x is either the actual value from task1 (if success) or a State (if failure)
    if isinstance(x, State):
        # upstream failed -> skip this branch
        raise Abort("task1 failed; skipping task2")
    # normal success-path logic using x
    return "task2 ran"

@task
def task3_only_if_failure(x):
    # run only when upstream failed
    if not isinstance(x, State) or x.type == StateType.COMPLETED:
        raise Abort("task1 succeeded; skipping task3")
    # optional: inspect failure details
    # from prefect.states import get_state_exception
    # err = get_state_exception(x)
    return "task3 ran"

@flow
def my_flow():
    f1 = task1.submit()

    # allow_failure lets downstream tasks run even if f1 failed; they receive either the value or a State
    x = allow_failure(f1)

    # Both are non-blocking submissions; the correct one runs, the other aborts
    s2 = task2_only_if_success.submit(x)
    s3 = task3_only_if_failure.submit(x)
    return s2, s3

Notes: - With your original

task2.submit(wait_for=future1)

, Prefect will automatically cancel

task2

if

task1

fails. The missing piece is triggering the failure path (

task3

) — the pattern above handles that without blocking. - If you truly want to avoid scheduling the “wrong” task at all, you must branch in the flow after waiting for

task1

to finish (which is a blocking decision) or split into separate flows triggered by automations. Within a single flow with non-blocking

submit

, the self-abort pattern is the idiomatic approach. Docs for deeper reference: - States overview: States -

allow_failure

annotation: allow_failure - Task API (including

submit

): Tasks If you share a bit of your real `task2`/`task3` logic, I can help tailor the guard conditions or error inspection (e.g., surfacing the exception from the failed state).

You’ve got the right mental model, with a couple important corrections: - allow_failure is not a task decorator. It’s a wrapper you apply to an upstream future/value when passing it to downstream tasks. It tells Prefect “don’t cancel this downstream if the upstream fails; instead pass the upstream’s State to the downstream.” - The allow_failure wrapper is resolved at task runtime (when Prefect resolves inputs), not at flow-time. So you can’t branch in the flow by checking whether allow_failure(...) is a State without waiting — it won’t be a State yet in the flow context. Given your requirement to avoid blocking and still branch on success vs failure, the idiomatic pattern is: - Submit task1 - Pass allow_failure(f1) into both branch tasks - Each branch task self-skips (Abort) when its condition isn’t met Minimal example:

from prefect import flow, task, allow_failure
from prefect.states import State, StateType
from prefect.exceptions import Abort

@task
def task1():
    ...

@task
def task2_only_if_success(x):
    # x is the actual value on success OR a State on failure
    if isinstance(x, State):
        raise Abort("task1 failed; skipping success branch")
    return "task2 ran"

@task
def task3_only_if_failure(x):
    if not isinstance(x, State) or x.type == StateType.COMPLETED:
        raise Abort("task1 succeeded; skipping failure branch")
    return "task3 ran"

@flow
def my_flow():
    f1 = task1.submit()
    x = allow_failure(f1)
    s2 = task2_only_if_success.submit(x)
    s3 = task3_only_if_failure.submit(x)
    return s2, s3

About your proposed snippet: - Remove

@allow_failure

above

task1

— that’s not valid. - If you keep a decide_what_branch function, make it a plain function in the flow (not a task) and be aware: you cannot decide purely with allow_failure at flow-time without waiting for task1 to finish. If you try to branch in a task (i.e.,

@task def decide_what_branch(...)

), you’d be calling

.submit()

from inside a task, which is not recommended. If you truly need to schedule exactly one branch (not both) without blocking, you’ll need to move the branching across flow runs (e.g., run task1 in one flow then use Automations on its terminal state to trigger the appropriate downstream flow). For reference: - States: States - allow_failure: allow_failure - Tasks API (submit): Tasks If you share a bit of task2/task3’s signatures and how you want their inputs shaped, I can tailor the guard logic precisely.

Short answer: not automatically for a Pydantic model. In Prefect 3.x, Prefect will automatically resolve futures that are passed as task/flow inputs directly or inside standard Python containers (dict/list/tuple/set). It does not reliably traverse arbitrary objects like a Pydantic BaseModel to replace nested PrefectFuture values with results. So your

M(task1=future1, task2=future2)

would arrive in

task_requiring_strings

still containing PrefectFuture objects, and those

isinstance(..., str)

checks would fail. Good patterns to keep things async without calling

.result()

early: 1) Let Prefect handle dependency resolution by passing futures directly (or inside standard containers) - Prefect will schedule the downstream task, add edges to the upstream task runs, and only start the downstream once upstream futures are completed. No need to call

.result()

yourself.

@task
def task1() -> str:
    return "a"

@task
def task2() -> str:
    return "b"

@task
def task_requiring_strings(s1: str, s2: str):
    assert isinstance(s1, str)
    assert isinstance(s2, str)

@flow
def main():
    f1 = task1.submit()
    f2 = task2.submit()
    # Prefect resolves futures to results before the task starts
    task_requiring_strings.submit(f1, f2)

If you need to pass a complex structure, prefer plain containers:

from prefect.futures import resolve_futures_to_results

@task
def meta_task():
    return {
        "task1": task1.submit(),
        "task2": task2.submit(),
    }

@task
def consumer(payload: dict):
    payload = resolve_futures_to_results(payload)
    assert isinstance(payload["task1"], str)
    assert isinstance(payload["task2"], str)

2) If you must use a Pydantic model as the “envelope,” resolve inside the consumer - Convert to a dict, resolve futures, then rebuild the model.

from pydantic import BaseModel
from prefect.futures import resolve_futures_to_results

class M(BaseModel):
    task1: str
    task2: str

@task
def meta_task() -> M:
    f1 = task1.submit()
    f2 = task2.submit()
    # Note: Pydantic will accept these but they remain PrefectFutures here
    return M(task1=f1, task2=f2)

@task
def task_requiring_strings(m: M):
    # Resolve to real values at task start
    data = resolve_futures_to_results(m.model_dump())
    m_resolved = M(**data)
    assert isinstance(m_resolved.task1, str)
    assert isinstance(m_resolved.task2, str)

Notes: - This keeps everything asynchronous: you’re not calling

.result()

in the flow body; Prefect schedules dependencies and only blocks at the point the consumer task starts (which is when you actually need the values). - If you want to keep futures typed in the model, you could declare fields as

PrefectFuture[str]

and set Pydantic config to allow arbitrary types, then use

resolve_futures_to_results

(plus converting to/from a dict) before you actually need strings. However, the plain-container approach is simpler and aligns with Prefect’s automatic traversal. If you share a small snippet of your exact flow/task signatures and your Pydantic model, I can suggest the minimal change needed.

✅ thought for 176.5 seconds

No —

model_dump()

does not resolve Prefect futures. It just converts your model to a dict; any

PrefectFuture

values remain futures in that dict. To get real values, you need to explicitly resolve them, e.g.:

from prefect.futures import resolve_futures_to_results

data = resolve_futures_to_results(m.model_dump())
m_resolved = M(**data)

Alternatively, avoid the Pydantic wrapper for the handoff and pass futures directly (or inside plain dict/list/tuple). Prefect will resolve those to results before invoking the downstream task. Helpful references: - prefect.futures API (resolve_futures_to_results, resolve_futures_to_states) - Source for resolve_futures_to_results

For a process worker there isn’t really a “container” to shell into — it’s just a Python process on the host where you started the worker, using that host’s environment (plus any env you pass via job variables). The easiest way to inspect the exact runtime is to run a small diagnostic task/flow and look at the logs. Try this probe (run on the same work pool/queue as the failing run):

from prefect import flow, task, get_run_logger
from prefect.utilities.processutils import get_sys_executable, run_process
import os, sys, json, site

@task
async def dump_runtime_env(package_name: str | None = None):
    logger = get_run_logger()
    <http://logger.info|logger.info>(f"sys.executable = {sys.executable}")
    <http://logger.info|logger.info>(f"get_sys_executable() = {get_sys_executable()}")
    <http://logger.info|logger.info>(f"sys.version = {sys.version}")
    <http://logger.info|logger.info>("sys.path =\n" + "\n".join(f"  - {p}" for p in sys.path))
    try:
        <http://logger.info|logger.info>(f"site.getsitepackages() = {site.getsitepackages()}")
    except Exception:
        pass
    # Show python/pip from the runtime
    await run_process(["python", "-V"], stream_output=True)
    await run_process(["python", "-m", "pip", "list"], stream_output=True)
    if package_name:
        await run_process(["python", "-m", "pip", "show", package_name], stream_output=True)
    # Dump env vars
    <http://logger.info|logger.info>("Environment variables:\n" + json.dumps(dict(os.environ), indent=2))

@flow
def env_probe(package_name: str | None = None):
    dump_runtime_env(package_name)

if __name__ == "__main__":
    env_probe("the_module_you_expect")

This will show: - Which Python interpreter the flow is actually using - The import path (

sys.path

) - What packages pip sees in that environment - Env vars that might affect imports (e.g.,

PYTHONPATH

) - Any job variables that were injected Other quick checks and tips: - Verify the work pool base job template and deployment job variables: - Inspect pool:

prefect work-pool inspect "<your-pool-name>"

- Ensure any

env

overrides (e.g.,

PYTHONPATH

,

VIRTUAL_ENV

,

PATH

) and

working_dir

are what you expect - Make sure you started the worker from the same environment where the module is installed. For example, if you installed into a venv/conda env, start the worker from that env so it inherits the same Python and site-packages: - Activate env, then run:

prefect worker start -p "<your-pool>" -q "<your-queue>"

- If the worker runs as a service (systemd, etc.), confirm the service uses the venv’s

prefect

and has the right `PATH`/`VIRTUAL_ENV`. A common cause of ModuleNotFoundError is “package installed in your shell’s venv, worker started with system Python.” - If you’re importing your own local package/module, either install it into the worker’s env (

pip install -e .

) or set

PYTHONPATH

via deployment/job variables so the project root is on the import path. If, instead, you’re on Docker or Kubernetes workers: - Docker: you can

docker exec -it <running-container-id> /bin/sh

to inspect; ensure your image includes the dependency. - Kubernetes:

kubectl logs

and

kubectl exec -it <pod> -- sh

in the job pod (while it’s running). Docs that may help: - Workers overview: Workers (Prefect 3.x) - Process worker API: prefect.workers.process - Runtime context (for inspecting env/context during runs): Runtime context - Customize job variables (env, working_dir): Customize job variables - Logging customization: Logging customization If you can share: - How you start the worker (command, service file, venv/conda?)

- The failing module name - Output from the env_probe run above …I can point to the exact fix.

Great questions — and thanks for the candid feedback. First: why Pydantic here at all? - Advantages vs plain dict/list: - Validation and coercion: guarantees your downstream gets the right types/constraints - Clear contracts and editor support: typed fields, schemas - Safer refactors: fields are discoverable and validated But: Prefect’s automatic future resolution only traverses standard containers (dict/list/tuple/set) when wiring task inputs. A Pydantic model is treated as an opaque object, so futures inside it are not automatically resolved. If you want “automatic” resolution without manually calling resolve_futures_to_results in your consumer, here are practical alternatives that keep things async and preserve your model: 1) Build the Pydantic model in a task that receives resolved values - Let Prefect resolve futures on the edges, then construct the model inside a task. This yields a Future[M] whose value is already fully resolved.

from pydantic import BaseModel
from prefect import flow, task

class M(BaseModel):
    task1: str
    task2: str

@task
def task1() -> str: ...
@task
def task2() -> str: ...

@task
def build_model(task1_val: str, task2_val: str) -> M:
    return M(task1=task1_val, task2=task2_val)

@task
def consumer(m: M):
    # m is fully-resolved here
    ...

@flow
def main():
    f1 = task1.submit()
    f2 = task2.submit()
    m_future = build_model.submit(f1, f2)  # Prefect resolves f1/f2 automatically
    consumer.submit(m_future)

Pros: no manual resolve calls, preserves Pydantic benefits. This is the most “automatic” approach with Prefect’s semantics. 2) Use a small “resolver” task between producer and consumer - If you really want the producer to construct M with futures inside, add a dedicated resolver task that returns M with concrete values. It keeps resolution inside the Prefect graph (not in your consumer code) and stays async.

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

class M(BaseModel):
    task1: str
    task2: str

@task
def meta_task() -> M:
    f1 = task1.submit()
    f2 = task2.submit()
    return M(task1=f1, task2=f2)  # contains futures

@task
def resolve_model(m: M) -> M:
    data = resolve_futures_to_results(m.model_dump())
    return M(**data)

@task
def consumer(m: M): ...

@flow
def main():
    m_future = meta_task.submit()
    m_resolved = resolve_model.submit(m_future)  # “automatic” via a task hop
    consumer.submit(m_resolved)

Pros: still no .result() in your flow; clean separation of concerns; downstream tasks always get a validated M. 3) Use a NamedTuple (automatic traversal) then convert to Pydantic at the boundary - If you want a structured return with automatic traversal but not a dict, NamedTuple works because it’s a tuple and Prefect traverses tuples.

from typing import NamedTuple
from pydantic import BaseModel
from prefect import flow, task

class M(BaseModel):
    task1: str
    task2: str

class MTuple(NamedTuple):
    task1: str
    task2: str

@task
def meta_task() -> MTuple:
    f1 = task1.submit()
    f2 = task2.submit()
    return MTuple(f1, f2)  # futures ok; Prefect will resolve inside the tuple

@task
def consumer(m: M):
    ...

@flow
def main():
    mt_future = meta_task.submit()
    # Build M where values are resolved
    @task
    def to_model(mt: MTuple) -> M:
        return M(task1=mt.task1, task2=mt.task2)

    m_future = to_model.submit(mt_future)
    consumer.submit(m_future)

Pros: automatic traversal with strong structure; convert to Pydantic once values are concrete. Why not rely on Pydantic for traversal? - Prefect resolves futures on task boundaries by walking standard containers. Pydantic models aren’t traversed. There isn’t a hook to make Prefect automatically call model_dump() for arbitrary user classes during scheduling. Docs:

- Futures API (resolve helpers): prefect.futures - Source for resolve_futures_to_results: futures.py If none of the above fit your taste, can you share what you disliked in the prior approach? For example: - You don’t want an extra conversion step in the consumer? - You want the producer to define the contract and be done? With that preference clarified, I can tailor the pattern (e.g., builder task vs resolver task vs NamedTuple bridge) to minimize boilerplate in the spot you care about.

Short answer: because you’re orchestrating from inside a task. Tasks don’t keep an active orchestration context once they return. In your example, the “main” task returns after submitting r3; when the task-runner for that task exits, it cancels any still-running inner tasks. That’s why r3 shows “cancelled by task runner.” Since “main” itself finished executing normally (it didn’t wait on r3 and didn’t raise), it’s marked Completed. Why this happens - Prefect treats flows as the orchestration boundary. Flows keep the task runner alive and propagate child task failures to the parent flow state. - A task can submit other tasks, but when the parent task returns, its task-runner context shuts down and will cancel unfinished inner tasks; failures of those inner tasks will not automatically change the parent task’s state unless the parent awaited their results. What to do instead - Preferred: Make “main” a flow, not a task. The flow will keep running until r3 completes and will reflect r3’s failure in the flow state.

from prefect import flow, task

@task
def t1(): ...
@task
def t2(i): ...
@task
def t3(i): ...

@flow
def main():
    r1 = t1.submit()
    r2 = t2.submit(r1)
    r3 = t3.submit(r2)
    return r3  # The flow will wait; if r3 fails, the flow fails

- If you must orchestrate inside a task, explicitly wait before returning so failures propagate and the runner doesn’t cancel the child:

from prefect import task
from prefect.futures import wait  # or just use r3.result()

@task
def main():
    r1 = t1.submit()
    r2 = t2.submit(r1)
    r3 = t3.submit(r2)

    # Ensure r3 completes before this task exits
    r3.result()           # propagates exceptions to this task
    return r3.result()    # or return the concrete value
    # Alternatively: wait([r3])

Reference for waiting on futures: - prefect.futures (wait, as_completed, resolve_futures_to_results) In short: orchestrate in flows; tasks should do work. If you orchestrate in a task, you must wait on the futures before returning or they’ll be cancelled when the task ends.