Align skipped-interval callback metrics and simplify guard logic

Keep Dag callback exception tags consistent across callback paths and centralize
skipped-interval precondition handling in one place to reduce duplicate checks.

Author: Tegh25

airflow-core/src/airflow/dag_processing/processor.py

@@ -452,7 +452,19 @@ def _execute_dag_skipped_intervals_callback(
             callback(context)
         except Exception:
             log.exception("Callback failed", dag_id=request.dag_id)
-            stats.incr("dag.callback_exceptions", tags={"dag_id": request.dag_id})
+            stats.incr(
+                "dag.callback_exceptions",
+                tags=prune_dict(
+                    {
+                        "dag_id": request.dag_id,
+                        "team_name": (
+                            DagModel.get_team_name(request.dag_id)
+                            if conf.getboolean("core", "multi_team")
+                            else None
+                        ),
+                    }
+                ),
+            )
 
 
 def _execute_task_callbacks(dagbag: DagBag, request: TaskCallbackRequest, log: FilteringBoundLogger) -> None:

airflow-core/src/airflow/jobs/scheduler_job_runner.py

@@ -2704,26 +2704,25 @@ def _create_dag_runs(
                     active_non_backfill_runs=active_runs_of_dags[dag_model.dag_id],
                 )
 
-                if data_interval is not None:
-                    skipped_summary = self._collect_skipped_intervals(
-                        serdag=serdag,
-                        new_info=next_info,
-                        session=session,
-                        listener_has_impls=listener_has_impls,
-                    )
-                    if skipped_summary is not None:
-                        if listener_has_impls:
-                            skipped_intervals_listener_events.append((serdag.dag_id, skipped_summary))
-                        if serdag.has_on_skipped_intervals_callback:
-                            skip_callback_requests.append(
-                                DagSkippedIntervalsCallbackRequest.from_summary(
-                                    filepath=dag_model.relative_fileloc or "",
-                                    bundle_name=dag_model.bundle_name,
-                                    bundle_version=created_run.bundle_version,
-                                    dag_id=serdag.dag_id,
-                                    summary=skipped_summary,
-                                )
+                skipped_summary = self._collect_skipped_intervals(
+                    serdag=serdag,
+                    new_info=next_info,
+                    session=session,
+                    listener_has_impls=listener_has_impls,
+                )
+                if skipped_summary is not None:
+                    if listener_has_impls:
+                        skipped_intervals_listener_events.append((serdag.dag_id, skipped_summary))
+                    if serdag.has_on_skipped_intervals_callback:
+                        skip_callback_requests.append(
+                            DagSkippedIntervalsCallbackRequest.from_summary(
+                                filepath=dag_model.relative_fileloc or "",
+                                bundle_name=dag_model.bundle_name,
+                                bundle_version=created_run.bundle_version,
+                                dag_id=serdag.dag_id,
+                                summary=skipped_summary,
                             )
+                        )
 
             # Exceptions like ValueError, ParamValidationError, etc. are raised by
             # DagModel.create_dagrun() when dag is misconfigured. The scheduler should not

task-sdk/src/airflow/sdk/execution_time/schema/versions/v2026_10_30.py

@@ -44,7 +44,7 @@ class AddArgBindingsToSupervisorTIRunContext(VersionChange):
 
 
 class AddDagSkippedIntervalsCallbackRequest(VersionChange):
-    """Introduce ``DagSkippedIntervalsCallbackRequest`` in ``CallbackRequest`` union."""
+    """Introduce ``DagSkippedIntervalsCallbackRequest`` in the ``CallbackRequest`` union."""
 
     description = __doc__