Fix TimeSensor start_from_trigger behavior

providers/standard/docs/sensors/datetime.rst

@@ -41,7 +41,14 @@ 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.
+The target moment is computed from the current wall-clock date in the Dag's timezone combined with
+``target_time``, evaluated fresh each time the sensor pokes or defers. It is not derived from
+``data_interval_end`` or ``run_after``; for interval-relative behavior use
+:class:`~airflow.providers.standard.sensors.time_delta.TimeDeltaSensor`.
+
+``start_from_trigger`` is not supported on ``TimeSensor``: the target moment can only be computed
+correctly at task-execution time, not at Dag-parse time, so it cannot be handed to the triggerer in
+advance without going stale. Passing ``start_from_trigger=True`` emits a deprecation warning and behaves as if it were not set.
 
 .. exampleinclude:: /../src/airflow/providers/standard/example_dags/example_sensors.py
     :language: python

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

@@ -17,7 +17,6 @@
 # under the License.
 from __future__ import annotations
 
-import dataclasses
 import datetime
 import warnings
 from typing import TYPE_CHECKING, Any
@@ -51,7 +50,6 @@ class TimeSensor(BaseSensorOperator):
         next_kwargs=None,
         timeout=None,
     )
-    start_from_trigger = False
 
     def __init__(
         self,
@@ -65,31 +63,39 @@ def __init__(
     ) -> None:
         super().__init__(**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
-        aware_time = timezone.coerce_datetime(
-            datetime.datetime.combine(
-                datetime.datetime.now(self.dag.timezone), target_time, self.dag.timezone
-            )
-        )
-
-        # 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.target_time = target_time
         self.deferrable = deferrable
         self.start_from_trigger = start_from_trigger
         self.end_from_trigger = end_from_trigger
 
-        if self.start_from_trigger:
-            # Replaced rather than mutated: ``start_trigger_args`` is a class attribute, so
-            # assigning through it would overwrite the arguments of every other task built
-            # from this operator.
-            self.start_trigger_args = dataclasses.replace(
-                self.start_trigger_args,
-                trigger_kwargs=dict(moment=self.target_datetime, end_from_trigger=self.end_from_trigger),
+    @property
+    def start_from_trigger(self) -> bool:
+        """Always False: TimeSensor cannot start from trigger. Kept for backward compatibility."""
+        return False
+
+    @start_from_trigger.setter
+    def start_from_trigger(self, value: bool) -> None:
+        if value:
+            warnings.warn(
+                "start_from_trigger is deprecated for TimeSensor and is now ignored. The target "
+                "moment is computed fresh from the current wall-clock time on every Dag parse, so "
+                "baking it into the serialized trigger arguments made the serialized Dag hash change "
+                "on every parse. Use deferrable=True instead, which computes the target moment at "
+                "task execution time and does not have this problem.",
+                AirflowProviderDeprecationWarning,
+                stacklevel=2,
             )
 
+    @property
+    def target_datetime(self) -> datetime.datetime:
+        """Compute the target moment on demand, in the dag's timezone."""
+        aware_time = timezone.coerce_datetime(
+            datetime.datetime.combine(
+                datetime.datetime.now(self.dag.timezone), self.target_time, self.dag.timezone
+            )
+        )
+        return timezone.convert_to_utc(aware_time)
+
     def execute(self, context: Context) -> None:
         if self.deferrable:
             self.defer(
@@ -106,10 +112,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.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,6 +23,7 @@
 import pytest
 import time_machine
 
+from airflow.exceptions import AirflowProviderDeprecationWarning
 from airflow.models.dag import DAG
 from airflow.providers.common.compat.sdk import TaskDeferred
 from airflow.providers.standard.sensors.time import TimeSensor
@@ -136,21 +137,31 @@ def test_execute_complete_accepts_event(self):
         except TypeError as e:
             pytest.fail(f"TypeError raised: {e}")
 
-    def test_start_trigger_args_are_not_shared_between_tasks(self):
-        """Each task must carry its own trigger arguments.
-
-        ``start_trigger_args`` is a class attribute, so assigning through it made every task
-        built from this operator advertise the moment of whichever was constructed last.
-        """
+    def test_start_from_trigger_is_deprecated_and_ignored(self):
+        with DAG(
+            dag_id="test_start_from_trigger_deprecated",
+            schedule=None,
+            start_date=datetime(2020, 1, 1, 13, 0),
+        ):
+            with pytest.warns(AirflowProviderDeprecationWarning, match="start_from_trigger is deprecated"):
+                op = TimeSensor(task_id="test", target_time=time(10, 0), start_from_trigger=True)
+        assert op.start_from_trigger is False
+        with pytest.warns(AirflowProviderDeprecationWarning, match="start_from_trigger is deprecated"):
+            op.start_from_trigger = True
+        assert op.start_from_trigger is False
+
+    def test_target_datetime_recomputed_on_each_access(self):
         with DAG(
-            dag_id="test_start_trigger_args_not_shared",
+            dag_id="test_target_datetime_recomputed",
             schedule=None,
             start_date=datetime(2020, 1, 1),
         ):
-            early = TimeSensor(task_id="early", target_time=time(1, 0), start_from_trigger=True)
-            late = TimeSensor(task_id="late", target_time=time(23, 0), start_from_trigger=True)
+            op = TimeSensor(task_id="test", target_time=time(10, 0))
+
+        with time_machine.travel("2025-06-01 00:00:00", tick=False):
+            first = op.target_datetime
+        with time_machine.travel("2025-06-02 00:00:00", tick=False):
+            second = op.target_datetime
 
-        assert early.start_trigger_args is not late.start_trigger_args
-        assert early.start_trigger_args.trigger_kwargs["moment"] == early.target_datetime
-        assert late.start_trigger_args.trigger_kwargs["moment"] == late.target_datetime
-        assert early.target_datetime != late.target_datetime
+        assert first.date() == pendulum.datetime(2025, 6, 1).date()
+        assert second.date() == pendulum.datetime(2025, 6, 2).date()