Fix TimeSensor dag_version churn via TimeOfDayTrigger

providers/standard/docs/sensors/datetime.rst

@@ -39,9 +39,14 @@ To run the sensor in deferrable mode, set ``deferrable=True``. See :ref:`deferri
 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.
+Use the :class:`~airflow.providers.standard.sensors.time.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, deferral, or trigger start), not against ``data_interval_end`` or ``run_after``. For interval-relative behavior, use :class:`~airflow.providers.standard.sensors.time_delta.TimeDeltaSensor`.
+
+When ``start_from_trigger=True``, the sensor starts on the triggerer via
+:class:`~airflow.providers.standard.triggers.temporal.TimeOfDayTrigger`. The trigger
+stores only the parse-stable ``target_time`` and ``tz``; the concrete moment is
+resolved when the trigger starts, so repeated Dag parses do not create new Dag versions.
 
 .. exampleinclude:: /../src/airflow/providers/standard/example_dags/example_sensors.py
     :language: python

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

@@ -17,14 +17,17 @@
 # under the License.
 from __future__ import annotations
 
-import dataclasses
 import datetime
 import warnings
 from typing import TYPE_CHECKING, Any
 
 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.providers.standard.triggers.temporal import (
+    DateTimeTrigger,
+    resolve_time_of_day_moment,
+    serializable_timezone,
+)
 from airflow.triggers.base import StartTriggerArgs
 
 if TYPE_CHECKING:
@@ -35,22 +38,32 @@ class TimeSensor(BaseSensorOperator):
     """
     Waits until the specified time of the day.
 
+    The time is evaluated against the wall-clock date in the Dag's timezone at
+    execution time (poke / deferral / trigger start), not at Dag-parse time.
+    This avoids dag_version churn from baking an absolute ``moment`` into
+    serialized ``start_trigger_args``.
+
+    When ``start_from_trigger=True``, the sensor starts directly on the triggerer
+    via :class:`~airflow.providers.standard.triggers.temporal.TimeOfDayTrigger`,
+    which stores only the parse-stable ``target_time`` + tz and resolves
+    the concrete moment when the trigger actually starts.
+
     :param target_time: time after which the job succeeds
     :param deferrable: whether to defer execution
+    :param start_from_trigger: Start the task directly from the triggerer without
+        going into the worker.
+    :param end_from_trigger: End the task directly from the triggerer without
+        going into the worker.
+    :param trigger_kwargs: Accepted for API compatibility with other sensors that
+        support dynamic task mapping into start-from-trigger; not used by TimeSensor.
 
     .. seealso::
         For more information on how to use this sensor, take a look at the guide:
         :ref:`howto/operator:TimeSensor`
 
     """
 
-    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_trigger_args = None
     start_from_trigger = False
 
     def __init__(
@@ -64,37 +77,68 @@ def __init__(
         **kwargs,
     ) -> 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)
+        # Wall-clock only; tzinfo is stripped so serialized target_time is deterministic.
+        if isinstance(target_time, datetime.time) and target_time.tzinfo is not None:
+            self.target_time = target_time.replace(tzinfo=None)
+        else:
+            self.target_time = target_time
         self.deferrable = deferrable
         self.start_from_trigger = start_from_trigger
         self.end_from_trigger = end_from_trigger
+        # Cached for this task attempt so a local-date rollover does not change the target.
+        self._cached_target_datetime: datetime.datetime | None = None
 
         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),
+            dag = self._dag
+            if dag is None:
+                raise ValueError(
+                    "TimeSensor(start_from_trigger=True) requires the sensor to be attached to a Dag "
+                    "so the timezone is known."
+                )
+            # Parse-stable kwargs only (no datetime.now()); moment is resolved when the trigger starts.
+            self.start_trigger_args = StartTriggerArgs(
+                trigger_cls="airflow.providers.standard.triggers.temporal.TimeOfDayTrigger",
+                trigger_kwargs={
+                    "target_time": self.target_time.isoformat(),
+                    "tz": serializable_timezone(dag.timezone),
+                    "end_from_trigger": self.end_from_trigger,
+                },
+                next_method="execute_complete",
+                next_kwargs=None,
+                timeout=None,
             )
 
+    def _resolve_target_datetime(self) -> datetime.datetime:
+        """Compute the UTC moment for target_time on today's date in the Dag timezone."""
+        dag = self._dag
+        # Unattached sensors (unit tests / early construction) use UTC.
+        tz: str | int | datetime.tzinfo = "UTC" if dag is None else dag.timezone
+        return resolve_time_of_day_moment(self.target_time, tz=tz)
+
+    def _get_target_datetime(self) -> datetime.datetime:
+        """Return the target moment, computing and caching it once per attempt."""
+        if self._cached_target_datetime is None:
+            self._cached_target_datetime = self._resolve_target_datetime()
+        return self._cached_target_datetime
+
+    @property
+    def target_datetime(self) -> datetime.datetime:
+        """
+        Resolved target datetime in UTC.
+
+        Computed on first access (or first execute/poke) from ``target_time`` and
+        the Dag timezone, then cached for the life of this instance. Two reads on
+        either side of midnight therefore return the *same* date for a given
+        attempt; a fresh task instance re-resolves against "today".
+        """
+        return self._get_target_datetime()
+
     def execute(self, context: Context) -> None:
+        moment = self._get_target_datetime()
         if self.deferrable:
             self.defer(
                 trigger=DateTimeTrigger(
-                    moment=self.target_datetime,  # This needs to be an aware timestamp
+                    moment=moment,
                     end_from_trigger=self.end_from_trigger,
                 ),
                 method_name="execute_complete",
@@ -106,10 +150,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/src/airflow/providers/standard/triggers/temporal.py

@@ -22,11 +22,130 @@
 from typing import Any
 
 import pendulum
+from pendulum.tz.timezone import FixedTimezone, Timezone
 
 from airflow.providers.common.compat.sdk import timezone
 from airflow.triggers.base import BaseTrigger, TaskSuccessEvent, TriggerEvent
 
 
+def _parse_timezone(value: str | int | datetime.tzinfo) -> Timezone | FixedTimezone:
+    """Return a pendulum timezone from an IANA name, fixed offset (seconds), or existing tzinfo."""
+    if isinstance(value, (Timezone, FixedTimezone)):
+        return value
+    if isinstance(value, datetime.tzinfo):
+        # Generic tzinfo (zoneinfo, datetime.timezone): rebuild a pendulum zone from its
+        # IANA name or fixed offset so pendulum's DST arithmetic gets a native zone type.
+        return pendulum.timezone(serializable_timezone(value))
+    return pendulum.timezone(value)
+
+
+def serializable_timezone(tzinfo: datetime.tzinfo | None) -> str | int:
+    """
+    Encode a tzinfo as a value that round-trips through ``pendulum.timezone`` / parse_timezone.
+
+    Named zones become their IANA name (e.g. ``Asia/Singapore``). Fixed-offset zones become
+    the offset in seconds (int), which is what Airflow's own timezone serializer uses.
+    UTC / zero-offset is always the string ``UTC`` for stable serialization.
+    """
+    if tzinfo is None:
+        return "UTC"
+    if isinstance(tzinfo, FixedTimezone):
+        if tzinfo.offset == 0:
+            return "UTC"
+        return tzinfo.offset
+    name = getattr(tzinfo, "name", None) or getattr(tzinfo, "key", None) or getattr(tzinfo, "zone", None)
+    if name:
+        if name in ("UTC", "utc", "+00:00"):
+            return "UTC"
+        return name
+    offset = tzinfo.utcoffset(None)
+    if offset is not None:
+        total = int(offset.total_seconds())
+        return "UTC" if total == 0 else total
+    return "UTC"
+
+
+def _coerce_target_time(target_time: datetime.time | str) -> datetime.time:
+    """Accept ``datetime.time`` or an ISO time string (Airflow serializes time as str)."""
+    if isinstance(target_time, str):
+        return datetime.time.fromisoformat(target_time)
+    if isinstance(target_time, datetime.time):
+        # Drop tzinfo so storage / comparison is wall-clock-only; zone is separate.
+        if target_time.tzinfo is not None:
+            return target_time.replace(tzinfo=None)
+        return target_time
+    raise TypeError(f"Expected datetime.time or str for target_time. Got {type(target_time)}")
+
+
+def resolve_time_of_day_moment(
+    target_time: datetime.time | str,
+    *,
+    tz: str | int | datetime.tzinfo = "UTC",
+    as_of: datetime.datetime | None = None,
+) -> pendulum.DateTime:
+    """
+    Resolve ``target_time`` on "today" in ``tz`` to a UTC-aware moment.
+
+    Semantics:
+
+    - **Already passed today**: still returns today's occurrence (caller succeeds immediately).
+      Does *not* roll forward to the next day.
+    - **Non-existent local time** (spring-forward gap, e.g. 02:30 America/New_York on DST start):
+      shifts forward to the next valid local time (e.g. 03:30).
+    - **Ambiguous local time** (fall-back overlap, e.g. 01:30 on DST end): uses ``fold=0``
+      (the first occurrence).
+    - Moment is computed from ``as_of`` (default: now) so callers can cache per attempt and
+      avoid midnight drift when re-checking within the same run.
+    """
+    wall_time = _coerce_target_time(target_time)
+    tzinfo = _parse_timezone(tz)
+
+    if as_of is None:
+        as_of = timezone.utcnow()
+    local_now = pendulum.instance(as_of).in_timezone(tzinfo)
+
+    # pendulum.datetime shifts non-existent (gap) times forward to the next valid wall time.
+    moment_local = pendulum.datetime(
+        local_now.year,
+        local_now.month,
+        local_now.day,
+        wall_time.hour,
+        wall_time.minute,
+        wall_time.second,
+        wall_time.microsecond,
+        tz=tzinfo,
+    )
+
+    # If the wall clock was preserved, check for ambiguous (fold) times and prefer fold=0.
+    if (
+        moment_local.hour,
+        moment_local.minute,
+        moment_local.second,
+        moment_local.microsecond,
+    ) == (
+        wall_time.hour,
+        wall_time.minute,
+        wall_time.second,
+        wall_time.microsecond,
+    ):
+        dt0 = datetime.datetime(
+            local_now.year,
+            local_now.month,
+            local_now.day,
+            wall_time.hour,
+            wall_time.minute,
+            wall_time.second,
+            wall_time.microsecond,
+            tzinfo=tzinfo,
+            fold=0,
+        )
+        dt1 = dt0.replace(fold=1)
+        if dt0.utcoffset() != dt1.utcoffset():
+            moment_local = pendulum.instance(dt0)
+
+    return timezone.convert_to_utc(moment_local)
+
+
 class DateTimeTrigger(BaseTrigger):
     """
     Trigger based on a datetime.
@@ -86,6 +205,58 @@ async def run(self) -> AsyncIterator[TriggerEvent]:
             yield TriggerEvent(self.moment)
 
 
+class TimeOfDayTrigger(DateTimeTrigger):
+    """
+    Trigger that fires once the wall-clock reaches ``target_time`` in ``tz``.
+
+    The concrete moment is resolved at construction (triggerer start), not at
+    Dag-parse time. That keeps ``start_trigger_args`` parse-stable while
+    preserving ``TimeSensor(start_from_trigger=True)``.
+
+    ``serialize()`` includes the resolved ``moment`` so a reconstruct cannot
+    re-resolve "today" after midnight. ``start_trigger_args`` still omit
+    ``moment`` so Dag serialization stays parse-stable.
+
+    ``target_time`` is accepted as an ISO time string so trigger kwargs remain
+    JSON/serde-safe (``datetime.time`` is not accepted by Airflow's trigger serde).
+
+    :param target_time: wall-clock time of day (``datetime.time`` or ISO time string)
+    :param tz: IANA name (str) or fixed offset in seconds (int); must round-trip
+        through ``pendulum.timezone``. Named ``tz`` to avoid shadowing the
+        ``timezone`` module imported from the SDK compat layer.
+    :param end_from_trigger: whether the trigger should mark the task successful after
+        the time condition is reached
+    :param moment: optional pre-resolved UTC moment; used on serialize reconstruct
+    """
+
+    def __init__(
+        self,
+        target_time: datetime.time | str,
+        *,
+        tz: str | int = "UTC",
+        end_from_trigger: bool = False,
+        moment: datetime.datetime | None = None,
+    ) -> None:
+        wall = _coerce_target_time(target_time)
+        super().__init__(
+            moment=moment if moment is not None else resolve_time_of_day_moment(wall, tz=tz),
+            end_from_trigger=end_from_trigger,
+        )
+        self.target_time: str = wall.isoformat()
+        self.tz: str | int = tz
+
+    def serialize(self) -> tuple[str, dict[str, Any]]:
+        return (
+            "airflow.providers.standard.triggers.temporal.TimeOfDayTrigger",
+            {
+                "target_time": self.target_time,
+                "tz": self.tz,
+                "end_from_trigger": self.end_from_trigger,
+                "moment": self.moment,
+            },
+        )
+
+
 class TimeDeltaTrigger(DateTimeTrigger):
     """
     Create DateTimeTriggers based on delays.