https://github.com/apache/airflow/blob/main/airflow/cli/commands/task_command.py#L145
if TYPE_CHECKING:
            assert session is not None
        
        
        
        nonlocal dag_run
        if dag_run not in session:
            dag_run = session.merge(dag_run, load=False)
        dataset_events = dag_run.consumed_dataset_events
        triggering_events: dict[str, list[DatasetEvent]] = defaultdict(list)
        for event in dataset_events:
            triggering_events[event.dataset.uri].append(event)
        return triggering_events
    try:
        expanded_ti_count: int | None = task.get_mapped_ti_count(task_instance.run_id, session=session)
    except NotMapped:
        expanded_ti_count = None
    
    
    
    context = {
        "conf": conf,
        "dag": dag,
        "dag_run": dag_run,
        "data_interval_end": timezone.coerce_datetime(data_interval.end),
        "data_interval_start": timezone.coerce_datetime(data_interval.start),
        "ds": ds,