Fix deferrable sensors not respecting soft_fail on timeout (#61132)

When a deferrable sensor with soft_fail=True times out, the task
fails with AirflowSensorTimeout instead of being marked as SKIPPED.
This is a regression from Airflow 2.x behavior.

The issue was in resume_execution() where TaskDeferralTimeout was
converted to AirflowSensorTimeout before checking soft_fail. This
fix uses nested exception handling to check soft_fail and never_fail
before the conversion, ensuring timeouts are properly skipped.

closes: #61130

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>

Author: nathadfield

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

@@ -249,13 +249,18 @@ def run_duration() -> float:
         return xcom_value
 
     def resume_execution(self, next_method: str, next_kwargs: dict[str, Any] | None, context: Context):
+        # Use nested try/except to convert TaskDeferralTimeout to AirflowSensorTimeout
+        # while still allowing soft_fail/never_fail to handle both exception types.
         try:
-            return super().resume_execution(next_method, next_kwargs, context)
-        except TaskDeferralTimeout as e:
-            raise AirflowSensorTimeout(*e.args) from e
+            try:
+                return super().resume_execution(next_method, next_kwargs, context)
+            except TaskDeferralTimeout as e:
+                raise AirflowSensorTimeout(*e.args) from e
         except (AirflowException, TaskDeferralError) as e:
             if self.soft_fail:
-                raise AirflowSkipException(str(e)) from e
+                raise AirflowSkipException("Skipping due to soft_fail is set to True.") from e
+            if self.never_fail:
+                raise AirflowSkipException("Skipping due to never_fail is set to True.") from e
             raise
 
     def _get_next_poke_interval(

task-sdk/tests/task_sdk/bases/test_sensor.py

@@ -36,6 +36,7 @@
     AirflowSensorTimeout,
     AirflowSkipException,
     AirflowTaskTimeout,
+    TaskDeferralError,
 )
 from airflow.sdk.execution_time.comms import RescheduleTask, TaskRescheduleStartDate
 from airflow.sdk.timezone import datetime
@@ -684,3 +685,59 @@ def test_fail_after_resuming_deferred_sensor(self, soft_fail, expected_exception
         async_sensor = DummyAsyncSensor(task_id="dummy_async_sensor", soft_fail=soft_fail)
         with pytest.raises(expected_exception):
             async_sensor.resume_execution("execute_complete", None, {})
+
+    @pytest.mark.parametrize(
+        ("soft_fail", "expected_exception"),
+        [
+            (True, AirflowSkipException),
+            (False, AirflowSensorTimeout),
+        ],
+    )
+    def test_timeout_after_resuming_deferred_sensor_with_soft_fail(self, soft_fail, expected_exception):
+        """Test that deferrable sensors with soft_fail skip on timeout instead of failing."""
+        async_sensor = DummyAsyncSensor(task_id="dummy_async_sensor", soft_fail=soft_fail)
+        with pytest.raises(expected_exception):
+            async_sensor.resume_execution(
+                next_method="__fail__",
+                next_kwargs={"error": TriggerFailureReason.TRIGGER_TIMEOUT},
+                context={},
+            )
+
+    def test_timeout_after_resuming_deferred_sensor_with_never_fail(self):
+        """Test that deferrable sensors with never_fail skip on timeout."""
+        async_sensor = DummyAsyncSensor(task_id="dummy_async_sensor", never_fail=True)
+        with pytest.raises(AirflowSkipException):
+            async_sensor.resume_execution(
+                next_method="__fail__",
+                next_kwargs={"error": TriggerFailureReason.TRIGGER_TIMEOUT},
+                context={},
+            )
+
+    @pytest.mark.parametrize(
+        ("soft_fail", "expected_exception"),
+        [
+            (True, AirflowSkipException),
+            (False, TaskDeferralError),
+        ],
+    )
+    def test_trigger_failure_after_resuming_deferred_sensor_with_soft_fail(
+        self, soft_fail, expected_exception
+    ):
+        """Test that deferrable sensors with soft_fail skip on trigger failure instead of failing."""
+        async_sensor = DummyAsyncSensor(task_id="dummy_async_sensor", soft_fail=soft_fail)
+        with pytest.raises(expected_exception):
+            async_sensor.resume_execution(
+                next_method="__fail__",
+                next_kwargs={"error": TriggerFailureReason.TRIGGER_FAILURE},
+                context={},
+            )
+
+    def test_trigger_failure_after_resuming_deferred_sensor_with_never_fail(self):
+        """Test that deferrable sensors with never_fail skip on trigger failure."""
+        async_sensor = DummyAsyncSensor(task_id="dummy_async_sensor", never_fail=True)
+        with pytest.raises(AirflowSkipException):
+            async_sensor.resume_execution(
+                next_method="__fail__",
+                next_kwargs={"error": TriggerFailureReason.TRIGGER_FAILURE},
+                context={},
+            )