Tasks marked with allow_failure runs even though upstream tasks are pending

[petlofg]

Bug summary

Description

Observed behavior

Tasks that are annotated with allow_failure runs if their direct upstream task is pending, if in turn it's upstream task has failed:

Tasks that don't specify allow_failure don't have this behavior and are correctly not started.

Expected behavior

Downstream tasks with allow_failure should not be executed if their parent is pending, in the same way as it works for tasks without allow_failure.

Reproduction

"""Minimal reproduction of allow_failure behavior with chained task failures."""

from prefect import allow_failure, flow, task


@task
def task_1_fails():
    raise RuntimeError("Task 1 always fails")


@task
def task_2_depends_on_1():
    print("Task 2 ran")
    return "task_2_result"


@task
def task_3_allow_failure():
    print("Task 3 ran (allow_failure) — THIS SHOULD NOT RUN if task 2 never ran")
    return "task_3_result"


@task
def task_4_no_allow_failure():
    print("Task 4 ran (no allow_failure) — this should not run")
    return "task_4_result"


@flow
def repro_flow():
    t1 = task_1_fails.submit()
    t2 = task_2_depends_on_1.submit(wait_for=[t1])
    t3 = task_3_allow_failure.submit(wait_for=[allow_failure(t2)])
    t4 = task_4_no_allow_failure.submit(wait_for=[t2])

    for label, future in [("t1", t1), ("t2", t2), ("t3", t3), ("t4", t4)]:
        future.wait()
        print(f"{label}: state={future.state.name}, type={future.state.type}")


if __name__ == "__main__":
    repro_flow()

Run the above on a local Prefect Server. The output shows that task_3_allow_failure executes but task_4_no_allow_failure don't.

Console output

12:41:10.203 | INFO    | prefect - Starting temporary server on http://127.0.0.1:8845
See https://docs.prefect.io/v3/concepts/server#how-to-guides for more information on running a dedicated Prefect server.
12:41:44.053 | INFO    | Flow run 'antique-koala' - Beginning flow run 'antique-koala' for flow 'repro-flow'
12:41:44.114 | ERROR   | Task run 'task_1_fails-18b' - Task run failed with exception: RuntimeError('Task 1 always fails')
Traceback (most recent call last):
  File "/home/petlofg/git/analytics/.venv/lib/python3.12/site-packages/prefect/task_engine.py", line 1021, in run_context
    yield self
  File "/home/petlofg/git/analytics/.venv/lib/python3.12/site-packages/prefect/task_engine.py", line 1683, in run_task_sync
    engine.call_task_fn(txn)
  File "/home/petlofg/git/analytics/.venv/lib/python3.12/site-packages/prefect/task_engine.py", line 1038, in call_task_fn
    result = call_with_parameters(self.task.fn, parameters)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/petlofg/git/analytics/.venv/lib/python3.12/site-packages/prefect/utilities/callables/__init__.py", line 348, in call_with_parameters
    return fn(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^
  File "/home/petlofg/git/analytics/minimal_repro.py", line 8, in task_1_fails
    raise RuntimeError("Task 1 always fails")
RuntimeError: Task 1 always fails
12:41:44.124 | ERROR   | Task run 'task_1_fails-18b' - Finished in state Failed('Task run encountered an exception RuntimeError: Task 1 always fails')
12:41:44.127 | ERROR   | Task run 'task_2_depends_on_1-08e' - Finished in state NotReady("Upstream task run '01a0335c-a28d-70b8-a2ed-ca6aadacbfc2' did not reach a 'COMPLETED' state.", type=PENDING)
t1: state=Failed, type=StateType.FAILED12:41:44.131 | ERROR   | Task run 'task_4_no_allow_failure-a6a' - Finished in state NotReady("Upstream task run '01a0335c-a290-771a-8ae6-c311af7ed839' did not reach a 'COMPLETED' state.", type=PENDING)
t2: state=NotReady, type=StateType.PENDING
Task 3 ran (allow_failure) — THIS SHOULD NOT RUN if task 2 never ran
12:41:44.137 | INFO    | Task run 'task_3_allow_failure-ca6' - Finished in state Completed()
t3: state=Completed, type=StateType.COMPLETED
t4: state=NotReady, type=StateType.PENDING
12:41:45.099 | INFO    | Flow run 'antique-koala' - Finished in state Completed()

The output line Task 3 ran (allow_failure) — THIS SHOULD NOT RUN if task 2 never ran show that Task 3 ran even though Task 2 is pending/not ready. Note that no output is show for Task 4 which is correct.

Version info

Version:              3.8.3
API version:          0.8.4
Python version:       3.12.11
Git commit:           d8f54b5c
Built:                Thu, Aug 13, 2026 06:55 PM
OS/Arch:              linux/x86_64
Profile:              default
Server type:          ephemeral
Pydantic version:     2.13.4
Server:
  Database:           sqlite
  SQLite version:     3.47.1
Integrations:
  prefect-azure:      0.4.12
  prefect-docker:     0.7.3
  prefect-shell:      0.3.6
  prefect-snowflake:  0.28.8
  prefect-dbt:        0.7.25

Additional context

We started noticing this behavior after upgrading from version 3.6.29 to 3.8.3. We've also tested this running on a self-hosted Prefect server which produces the same behavior. Could PR #21305 have caused this to start happening?

[qdivan]

I reproduced this on current main (ce79dd3) with the four-task chain from the report: the states are FAILED/Failed, PENDING/NotReady, COMPLETED/Completed, and PENDING/NotReady respectively, so the allow_failure-wrapped downstream task executes while the ordinary downstream task is withheld.

Before preparing a PR, I found that this is also the behavior explicitly covered by tests/test_tasks.py::TestTaskWaitFor::test_allow_failure_with_mapped_tasks_in_wait_for (and introduced as the desired behavior by #21305 for failure-derived NotReady states). I ran that test together with the direct-failure allow_failure wait-for control; both pass on current main.

The requested change would therefore intentionally reverse the current allow_failure contract for failure-derived NotReady states, rather than be a localized regression repair. Could a maintainer confirm that this semantic change is desired (and whether the mapped-task behavior from #21305 should change too) before I prepare a patch?

[desertaxle]

Thanks for the clear reproduction @petlofg! We introduced this behavior intentionally in #21305 so cleanup tasks can run after chained failures, but I agree the current behavior is confusing.

A task that will never execute is recorded as PENDING/NotReady. That means we lose info on whether it was blocked by a FAILED, CANCELLED, or CRASHED upstream. Once this happens, allow_failure can't tell why an upstream node entered NotReady.

I don't think we should reverse the existing semantics because that would be a breaking change. Instead, we should preserve the blocking cause and distinguish direct failure from blocked-by-upstream. We can then add an opt-in strict form, such as allow_failure(t2, include_blocked=False), for users who only want to continue when t2 itself ran and failed. The existing default can remain transitive for compatibility.

I can start working on a fix for this, but let me know if you think the proposed changes wouldn't work for your use case.

[petlofg]

Sounds good to me! Thanks for the quick response!