https://github.com/apache/airflow/blob/main/airflow/cli/commands/task_command.py#L145
if self.state == DagRunState.RUNNING:
            get_listener_manager().hook.on_dag_run_running(dag_run=self, msg=msg)
        elif self.state == DagRunState.SUCCESS:
            get_listener_manager().hook.on_dag_run_success(dag_run=self, msg=msg)
        elif self.state == DagRunState.FAILED:
            get_listener_manager().hook.on_dag_run_failed(dag_run=self, msg=msg)
        
        
        
    def _get_ready_tis(
        self,
        schedulable_tis: list[TI],
        finished_tis: list[TI],
        session: Session,
    ) -> tuple[list[TI], bool, bool]:
        old_states = {}
        ready_tis: list[TI] = []
        changed_tis = False
        if not schedulable_tis:
            return ready_tis, changed_tis, False
        
        
        additional_tis: list[TI] = []
        dep_context = DepContext(
            flag_upstream_failed=True,
            ignore_unmapped_tasks=True,  
            finished_tis=finished_tis,
        )
        def _expand_mapped_task_if_needed(ti: TI) -> Iterable[TI] | None: