@NhanNguyen thanks for your question!

Have you maybe consider to:

• Define a Global Error Task: Add a handle_error task in the workflow that runs conditionally on task failure.
• Set Failure Conditions: In the UI or through JSON, configure the handle_error task with a "run if any upstream task fails" condition.

{
    "tasks": [
        {
            "task_key": "task_a",
            "libraries": [],
            "timeout_seconds": 3600
        },
        {
            "task_key": "task_b",
            "depends_on": [{"task_key": "task_a"}],
            "libraries": [],
            "timeout_seconds": 3600
        },
        {
            "task_key": "handle_error",
            "depends_on": [
                {"task_key": "task_a"},
                {"task_key": "task_b"}
            ],
            "condition_task_state": "FAILED",
            "libraries": []
        }
    ]
}

• Implement Custom Error Logging: Write a Python script in handle_error that aggregates errors from upstream tasks for logging or notification.