Any ideas? Is my question unclear maybe? I'd like to have some kind of terminal task that would look upstream along the mapped parents to see if any errors occurred. In the above example I'd like to have for example some print statements that said: There WAS a problem with input: 2 There was no a problem with input: 5 There was no a problem with input: 7

Surely there's a way to do something like this? No? The Schematic tab in the GUI provides information about which mapped runs were skiped because of problems upstream.

from prefect.engine import signals
from prefect.exceptions import PrefectSignal
from prefect.triggers import all_successful, all_failed, always_run
from prefect import task, Flow


l = [2,5,7]
@task(name="Task A")
def task_a(x: int) -> int:
    if x == 2:
        raise signals.FAIL
    return x + 2

@task(name="Task B")
def task_b(x: int) -> int:
    return x + 2

@task(name="check", trigger=always_run)
def checker(x: int=None) -> None:
    if isinstance(x, PrefectSignal):
        print("THINGS ARE NOT OK")
    else:
        print("THINGS ARE OK")


with Flow("x example") as flow:
    la = task_a.map(x=l)
    lb = task_b.map(x=la)
    lc = checker.map(x=lb)
flow.run()

this kind of does what I'm talking about I think, in case it's useful. force the task to always run, and then assume if something upstream broke it's going to give you a PrefectSignal rather than what you were expecting. I'm not sure how to look up the originating input ATM.