https://github.com/apache/airflow/blob/main/airflow/cli/commands/task_command.py#L145
cls,
        ti: TaskInstance | TaskInstancePydantic,
        error: None | str | Exception | KeyboardInterrupt,
        test_mode: bool | None = None,
        context: Context | None = None,
        force_fail: bool = False,
        session: Session = NEW_SESSION,
    ):
        """Handle Failure for the TaskInstance."""
        get_listener_manager().hook.on_task_instance_failed(
            previous_state=TaskInstanceState.RUNNING, task_instance=ti, session=session
        )
        if error:
            if isinstance(error, BaseException):
                tb = TaskInstance.get_truncated_error_traceback(error, truncate_to=ti._execute_task)