@task
def main_task(name: str):
    print(f"Running task for {name}")
    raise ValueError("Simulated failure!")

@task
def clean_up(name: str, err: typing.Optional[FlyteError] = None):
    if err:
        print(f"Cleaning up resources for {name} due to error: {err.message}")
        print(f"Failed Node ID: {err.failed_node_id}")
    else:
        print(f"Cleaning up resources for {name}")

@workflow(on_failure=clean_up)
def my_workflow(name: str):
    main_task(name=name)

do you see any issue here? err object is not getting propagated to cleanup function when main_task is failed. Getting else clause in output.