Honour wait_for_past_depends_before_skipping when a task is told to skip

A task with depends_on_past=True and wait_for_past_depends_before_skipping=True
was skipped at once when an upstream ShortCircuitOperator or branch operator
decided to skip it, even though its previous run's task instance had failed.

Two things stood between the parameter and the scheduler. SkipMixin raised
DownstreamTasksSkipped for every direct downstream task, so the execution API
wrote SKIPPED before any dependency ran. And the scheduler deps only honoured
DepContext.wait_for_past_depends_before_skipping, which nothing sets in
Airflow 3; the task attribute was never read, and SerializedMappedOperator
did not expose it.

SkipMixin now leaves the direct downstream tasks that set both flags out of
the state change, keeping them in the XCom decision, so NotPreviouslySkippedDep
decides. The deps read the task attribute as well as the context flag, and
PrevDagrunDep writes the past-depends-met mark through the scheduler's session
instead of a fresh one that commits inside the scheduling loop. A task that
sets only one of the two flags keeps today's immediate skip, as its docstring
says, and tasks reached through another skipped task are still skipped at once.

Author: luc-pimentel

airflow-core/src/airflow/serialization/definitions/mappedoperator.py

@@ -223,6 +223,10 @@ def ignore_first_depends_on_past(self) -> bool:
     def wait_for_downstream(self) -> bool:
         return self._get_partial_kwargs_or_operator_default("wait_for_downstream")
 
+    @property
+    def wait_for_past_depends_before_skipping(self) -> bool:
+        return self._get_partial_kwargs_or_operator_default("wait_for_past_depends_before_skipping")
+
     @property
     def retries(self) -> int:
         return self._get_partial_kwargs_or_operator_default("retries")

airflow-core/src/airflow/ti_deps/deps/not_previously_skipped_dep.py

@@ -130,7 +130,9 @@ def _skip(
         # If the parent SkipMixin has run, and the XCom result stored indicates this
         # ti should be skipped, set ti.state to SKIPPED and fail the rule so that the
         # ti does not execute.
-        if dep_context.wait_for_past_depends_before_skipping:
+        if TYPE_CHECKING:
+            assert ti.task
+        if dep_context.wait_for_past_depends_before_skipping or ti.task.wait_for_past_depends_before_skipping:
             past_depends_met = ti.xcom_pull(
                 task_ids=ti.task_id, key=PAST_DEPENDS_MET, session=session, default=False
             )

airflow-core/src/airflow/ti_deps/deps/prev_dagrun_dep.py

@@ -50,9 +50,12 @@ class PrevDagrunDep(BaseTIDep):
     IS_TASK_DEP = True
 
     @staticmethod
-    def _push_past_deps_met_xcom_if_needed(ti: TI, dep_context):
-        if dep_context.wait_for_past_depends_before_skipping:
-            ti.xcom_push(key=PAST_DEPENDS_MET, value=True)
+    def _push_past_deps_met_xcom_if_needed(ti: TI, dep_context, *, session: Session):
+        if TYPE_CHECKING:
+            assert ti.task
+        if dep_context.wait_for_past_depends_before_skipping or ti.task.wait_for_past_depends_before_skipping:
+            # The scheduler's session: a fresh one would commit inside the scheduling loop.
+            ti.xcom_push(key=PAST_DEPENDS_MET, value=True, session=session)
 
     @staticmethod
     def _has_tis(dagrun: DagRun, task_id: str, *, session: Session) -> bool:
@@ -134,13 +137,13 @@ def _get_dep_statuses(self, ti: TI, dep_context, *, session: Session):
         if TYPE_CHECKING:
             assert ti.task
         if dep_context.ignore_depends_on_past:
-            self._push_past_deps_met_xcom_if_needed(ti, dep_context)
+            self._push_past_deps_met_xcom_if_needed(ti, dep_context, session=session)
             reason = "The context specified that the state of past DAGs could be ignored."
             yield self._passing_status(reason=reason)
             return
 
         if not ti.task.depends_on_past:
-            self._push_past_deps_met_xcom_if_needed(ti, dep_context)
+            self._push_past_deps_met_xcom_if_needed(ti, dep_context, session=session)
             yield self._passing_status(reason="The task did not have depends_on_past set.")
             return
 
@@ -157,7 +160,7 @@ def _get_dep_statuses(self, ti: TI, dep_context, *, session: Session):
                 return
 
         if not dr:
-            self._push_past_deps_met_xcom_if_needed(ti, dep_context)
+            self._push_past_deps_met_xcom_if_needed(ti, dep_context, session=session)
             yield self._passing_status(reason="This task instance does not belong to a DAG.")
             return
 
@@ -169,7 +172,7 @@ def _get_dep_statuses(self, ti: TI, dep_context, *, session: Session):
             last_dagrun = DagRun.get_previous_dagrun(dr, session=session)
         # First ever run for this DAG.
         if not last_dagrun:
-            self._push_past_deps_met_xcom_if_needed(ti, dep_context)
+            self._push_past_deps_met_xcom_if_needed(ti, dep_context, session=session)
             yield self._passing_status(reason="This task instance was the first task instance for its task.")
             return
 
@@ -180,14 +183,14 @@ def _get_dep_statuses(self, ti: TI, dep_context, *, session: Session):
             and last_dagrun.logical_date is not None
             and last_dagrun.logical_date < ti.task.start_date
         ):
-            self._push_past_deps_met_xcom_if_needed(ti, dep_context)
+            self._push_past_deps_met_xcom_if_needed(ti, dep_context, session=session)
             yield self._passing_status(reason="This task instance was the first task instance for its task.")
             return
 
         if not self._has_tis(last_dagrun, ti.task_id, session=session):
             if ti.task.ignore_first_depends_on_past:
                 if not self._has_any_prior_tis(ti, session=session):
-                    self._push_past_deps_met_xcom_if_needed(ti, dep_context)
+                    self._push_past_deps_met_xcom_if_needed(ti, dep_context, session=session)
                     yield self._passing_status(
                         reason="ignore_first_depends_on_past is true for this task "
                         "and it is the first task instance for its task."
@@ -220,4 +223,4 @@ def _get_dep_statuses(self, ti: TI, dep_context, *, session: Session):
             )
             return
 
-        self._push_past_deps_met_xcom_if_needed(ti, dep_context)
+        self._push_past_deps_met_xcom_if_needed(ti, dep_context, session=session)

airflow-core/src/airflow/ti_deps/deps/trigger_rule_dep.py

@@ -366,9 +366,9 @@ def _evaluate_setup_constraint(
                         new_state = TaskInstanceState.REMOVED
 
             if new_state is not None:
-                if (
-                    new_state == TaskInstanceState.SKIPPED
-                    and dep_context.wait_for_past_depends_before_skipping
+                if new_state == TaskInstanceState.SKIPPED and (
+                    dep_context.wait_for_past_depends_before_skipping
+                    or task.wait_for_past_depends_before_skipping
                 ):
                     past_depends_met = ti.xcom_pull(
                         task_ids=ti.task_id, key=PAST_DEPENDS_MET, session=session, default=False
@@ -527,9 +527,9 @@ def _evaluate_direct_relatives() -> Iterator[TIDepStatus]:
                         # if at least one setup ran, we'll let it run
                         new_state = TaskInstanceState.UPSTREAM_FAILED
             if new_state is not None:
-                if (
-                    new_state == TaskInstanceState.SKIPPED
-                    and dep_context.wait_for_past_depends_before_skipping
+                if new_state == TaskInstanceState.SKIPPED and (
+                    dep_context.wait_for_past_depends_before_skipping
+                    or task.wait_for_past_depends_before_skipping
                 ):
                     past_depends_met = ti.xcom_pull(
                         task_ids=ti.task_id, key=PAST_DEPENDS_MET, session=session, default=False

airflow-core/tests/unit/ti_deps/deps/test_prev_dagrun_dep.py

@@ -330,7 +330,7 @@ def test_dagrun_dep(mock_get_previous_dagrun, mock_get_previous_scheduled_dagrun
 
     assert actual_dep_met == expected_dep_met
     if past_depends_met_xcom_sent:
-        ti.xcom_push.assert_called_with(key="past_depends_met", value=True)
+        ti.xcom_push.assert_called_with(key="past_depends_met", value=True, session=ANY)
     else:
         ti.xcom_push.assert_not_called()
 
@@ -346,4 +346,4 @@ def test_past_depends_met_xcom_sent_when_the_task_asks_to_wait():
     ti = Mock(task=task, task_id=task.task_id, **{"xcom_push.return_value": None})
 
     assert PrevDagrunDep().is_met(ti=ti, dep_context=DepContext(ignore_depends_on_past=True))
-    ti.xcom_push.assert_called_once_with(key="past_depends_met", value=True)
+    ti.xcom_push.assert_called_once_with(key="past_depends_met", value=True, session=ANY)

task-sdk/src/airflow/sdk/bases/skipmixin.py

@@ -55,6 +55,24 @@ def _set_state_to_skipped(
         if tasks and map_index == -1:
             raise DownstreamTasksSkipped(tasks=tasks)
 
+    def _tasks_to_skip_now(self, tasks: Sequence[Operator]) -> list[Operator]:
+        """
+        Drop the direct downstream tasks that wait for their past depends before being skipped.
+
+        They stay in the XCom decision, so the scheduler skips them once the previous run's task
+        instance has succeeded or been skipped (``NotPreviouslySkippedDep``).
+        """
+        direct_downstream_ids: set[str] = getattr(self, "downstream_task_ids", set())
+        return [
+            t
+            for t in tasks
+            if not (
+                t.task_id in direct_downstream_ids
+                and t.depends_on_past
+                and t.wait_for_past_depends_before_skipping
+            )
+        ]
+
     def skip(
         self,
         ti: RuntimeTaskInstanceProtocol,
@@ -79,7 +97,7 @@ def skip(
                 value={XCOM_SKIPMIXIN_SKIPPED: task_ids_list},
             )
 
-        self._set_state_to_skipped(task_ids_list, ti.map_index)
+        self._set_state_to_skipped([t.task_id for t in self._tasks_to_skip_now(task_list)], ti.map_index)
 
     def skip_all_except(
         self,
@@ -156,7 +174,9 @@ def skip_all_except(
                 branch_task_id_set.update(dag.get_task(branch_task_id).get_flat_relative_ids(upstream=False))
 
             skip_tasks = [
-                (t.task_id, ti.map_index) for t in downstream_tasks if t.task_id not in branch_task_id_set
+                (t.task_id, ti.map_index)
+                for t in self._tasks_to_skip_now(downstream_tasks)
+                if t.task_id not in branch_task_id_set
             ]
 
             follow_task_ids = [t.task_id for t in downstream_tasks if t.task_id in branch_task_id_set]