Fix TimeSensor dag_version churn by removing start_from_trigger support

providers/standard/docs/sensors/datetime.rst

@@ -41,7 +41,7 @@ TimeSensor
 
 Use the :class:`~airflow.providers.standard.sensors.time_sensor.TimeSensor` to end sensing after time specified. ``TimeSensor`` can be run in deferrable mode, if a Triggerer is available.
 
-Time will be evaluated against ``data_interval_end`` if present for the Dag run, otherwise ``run_after`` will be used.
+Time is evaluated against the wall-clock date in the Dag's timezone at execution time (poke or trigger time), not against ``data_interval_end`` or ``run_after``. For interval-relative behavior, use :class:`~airflow.providers.standard.sensors.time_delta.TimeDeltaSensor`.
 
 .. exampleinclude:: /../src/airflow/providers/standard/example_dags/example_sensors.py
     :language: python

providers/standard/src/airflow/providers/standard/sensors/time.py

@@ -24,7 +24,6 @@
 from airflow.exceptions import AirflowProviderDeprecationWarning
 from airflow.providers.common.compat.sdk import BaseSensorOperator, conf, timezone
 from airflow.providers.standard.triggers.temporal import DateTimeTrigger
-from airflow.triggers.base import StartTriggerArgs
 
 if TYPE_CHECKING:
     from airflow.sdk import Context
@@ -34,6 +33,11 @@ class TimeSensor(BaseSensorOperator):
     """
     Waits until the specified time of the day.
 
+    The time is evaluated against the current date in the Dag's timezone
+    at execution time (poke or deferral), not at Dag-parse time. This avoids
+    Dag version churn that previously occurred when target_datetime was
+    baked into serialized start_trigger_args.
+
     :param target_time: time after which the job succeeds
     :param deferrable: whether to defer execution
 
@@ -43,53 +47,50 @@ class TimeSensor(BaseSensorOperator):
 
     """
 
-    start_trigger_args = StartTriggerArgs(
-        trigger_cls="airflow.providers.standard.triggers.temporal.DateTimeTrigger",
-        trigger_kwargs={"moment": "", "end_from_trigger": False},
-        next_method="execute_complete",
-        next_kwargs=None,
-        timeout=None,
-    )
-    start_from_trigger = False
-
     def __init__(
         self,
         *,
         target_time: datetime.time,
         deferrable: bool = conf.getboolean("operators", "default_deferrable", fallback=False),
-        start_from_trigger: bool = False,
         end_from_trigger: bool = False,
         trigger_kwargs: dict[str, Any] | None = None,
         **kwargs,
     ) -> None:
+        start_from_trigger = kwargs.pop("start_from_trigger", None)
+        if start_from_trigger is not None:
+            warnings.warn(
+                "start_from_trigger is deprecated and no longer supported for TimeSensor. "
+                "It has been ignored. Target time is now always evaluated at execution time. "
+                "Use deferrable=True to defer the sensor instead.",
+                AirflowProviderDeprecationWarning,
+                stacklevel=2,
+            )
         super().__init__(**kwargs)
+        self.target_time = target_time
+        self.deferrable = deferrable
+        self.end_from_trigger = end_from_trigger
+        # Accepted for compatibility only; storing it would reintroduce serialized Dag hash churn.
+        del trigger_kwargs
 
-        # Create a "date-aware" timestamp that will be used as the "target_datetime". This is a requirement
-        # of the DateTimeTrigger
-
-        # Get date considering dag.timezone
+    def _get_target_datetime(self) -> datetime.datetime:
+        """Compute target datetime at execution time, not parse time."""
+        dag_timezone = getattr(getattr(self, "dag", None), "timezone", None) or timezone.utc
+        now_date = datetime.datetime.now(dag_timezone).date()
         aware_time = timezone.coerce_datetime(
-            datetime.datetime.combine(
-                datetime.datetime.now(self.dag.timezone), target_time, self.dag.timezone
-            )
+            datetime.datetime.combine(now_date, self.target_time, dag_timezone)
         )
+        return timezone.convert_to_utc(aware_time)
 
-        # Now that the dag's timezone has made the datetime timezone aware, we need to convert to UTC
-        self.target_datetime = timezone.convert_to_utc(aware_time)
-        self.deferrable = deferrable
-        self.start_from_trigger = start_from_trigger
-        self.end_from_trigger = end_from_trigger
-
-        if self.start_from_trigger:
-            self.start_trigger_args.trigger_kwargs = dict(
-                moment=self.target_datetime, end_from_trigger=self.end_from_trigger
-            )
+    @property
+    def target_datetime(self) -> datetime.datetime:
+        """Backward-compatible property, now computed at call time."""
+        return self._get_target_datetime()
 
     def execute(self, context: Context) -> None:
         if self.deferrable:
             self.defer(
                 trigger=DateTimeTrigger(
-                    moment=self.target_datetime,  # This needs to be an aware timestamp
+                    moment=self._get_target_datetime(),
                     end_from_trigger=self.end_from_trigger,
                 ),
                 method_name="execute_complete",
@@ -101,10 +102,9 @@ def execute_complete(self, context: Context, event: Any = None) -> None:
         return None
 
     def poke(self, context: Context) -> bool:
-        self.log.info("Checking if the time (%s) has come", self.target_datetime)
-
-        # self.target_date has been converted to UTC, so we do not need to convert timezone
-        return timezone.utcnow() > self.target_datetime
+        target_datetime = self._get_target_datetime()
+        self.log.info("Checking if the time (%s) has come", target_datetime)
+        return timezone.utcnow() > target_datetime
 
 
 class TimeSensorAsync(TimeSensor):

providers/standard/tests/unit/standard/sensors/test_time.py

@@ -23,10 +23,12 @@
 import pytest
 import time_machine
 
-from airflow.models.dag import DAG
+from airflow.exceptions import AirflowProviderDeprecationWarning
 from airflow.providers.common.compat.sdk import TaskDeferred
 from airflow.providers.standard.sensors.time import TimeSensor
 from airflow.providers.standard.triggers.temporal import DateTimeTrigger
+from airflow.sdk import DAG
+from airflow.serialization.serialized_objects import LazyDeserializedDAG
 
 from tests_common.test_utils.compat import timezone
 
@@ -135,3 +137,37 @@ def test_execute_complete_accepts_event(self):
             op.execute_complete(context={}, event={"status": "success"})
         except TypeError as e:
             pytest.fail(f"TypeError raised: {e}")
+
+    def test_start_from_trigger_is_ignored_with_warning(self):
+        with pytest.warns(AirflowProviderDeprecationWarning, match="start_from_trigger is deprecated"):
+            op = TimeSensor(
+                task_id="test",
+                target_time=time(10, 0),
+                start_from_trigger=True,
+                trigger_kwargs={"moment": "2020-01-01T10:00:00Z"},
+            )
+
+        assert getattr(op, "start_from_trigger", False) is False
+        assert not hasattr(op, "_trigger_kwargs")
+
+    def test_ignored_start_from_trigger_keeps_serialized_dag_hash_stable(self):
+        def build_hash() -> str:
+            with pytest.warns(AirflowProviderDeprecationWarning, match="start_from_trigger is deprecated"):
+                with DAG(
+                    dag_id="test_ignored_start_from_trigger_hash",
+                    schedule=None,
+                    start_date=datetime(2020, 1, 1),
+                ) as dag:
+                    TimeSensor(
+                        task_id="test",
+                        target_time=time(10, 0),
+                        start_from_trigger=True,
+                    )
+            return LazyDeserializedDAG.from_dag(dag).hash
+
+        with time_machine.travel("2025-01-01 00:00:00", tick=False):
+            first_hash = build_hash()
+        with time_machine.travel("2025-01-02 00:00:00", tick=False):
+            second_hash = build_hash()
+
+        assert first_hash == second_hash