Fix time zone normalization bug in interval schedules (#19301)

Author: zzstoatzz

src/prefect/server/schemas/schedules.py

@@ -165,50 +165,33 @@ def _get_dates_generator(
 
         if sys.version_info >= (3, 13):
             # `pendulum` is not supported in Python 3.13, so we use `whenever` instead
-            from whenever import ZonedDateTime
+            from whenever import PlainDateTime, ZonedDateTime
 
             if start is None:
                 start = ZonedDateTime.now("UTC").py_datetime()
 
-            if self.anchor_date.tzinfo is None:
-                from whenever import PlainDateTime
-
-                anchor_zdt = PlainDateTime.from_py_datetime(self.anchor_date).assume_tz(
-                    "UTC"
-                )
-            elif isinstance(self.anchor_date.tzinfo, ZoneInfo):
-                anchor_zdt = ZonedDateTime.from_py_datetime(self.anchor_date).to_tz(
-                    self.timezone or "UTC"
+            target_timezone = self.timezone or "UTC"
+
+            def to_local_zdt(dt: datetime.datetime | None) -> ZonedDateTime | None:
+                if dt is None:
+                    return None
+                if dt.tzinfo is None:
+                    return PlainDateTime.from_py_datetime(dt).assume_tz(target_timezone)
+                if isinstance(dt.tzinfo, ZoneInfo):
+                    return ZonedDateTime.from_py_datetime(dt).to_tz(target_timezone)
+                # For offset-based tzinfo instances (e.g. datetime.timezone(+09:00)),
+                # use astimezone to preserve the instant, then convert to ZonedDateTime.
+                return ZonedDateTime.from_py_datetime(
+                    dt.astimezone(ZoneInfo(target_timezone))
                 )
-            else:
-                # This case handles rogue tzinfo objects that `whenever` doesn't play will with
-                anchor_zdt = ZonedDateTime.from_py_datetime(
-                    self.anchor_date.replace(
-                        tzinfo=ZoneInfo(self.anchor_date.tzname() or "UTC")
-                    )
-                ).to_tz(self.timezone or "UTC")
 
-            if start.tzinfo is None:
-                local_start = PlainDateTime.from_py_datetime(start).assume_tz("UTC")
-            elif isinstance(start.tzinfo, ZoneInfo):
-                local_start = ZonedDateTime.from_py_datetime(start).to_tz(
-                    self.timezone or "UTC"
-                )
-            else:
-                local_start = ZonedDateTime.from_py_datetime(
-                    start.replace(tzinfo=ZoneInfo(start.tzname() or "UTC"))
-                ).to_tz(self.timezone or "UTC")
-
-            if end is None:
-                local_end = None
-            elif isinstance(end.tzinfo, ZoneInfo):
-                local_end = ZonedDateTime.from_py_datetime(end).to_tz(
-                    self.timezone or "UTC"
-                )
-            else:
-                local_end = ZonedDateTime.from_py_datetime(
-                    end.replace(tzinfo=ZoneInfo(end.tzname() or "UTC"))
-                ).to_tz(self.timezone or "UTC")
+            anchor_zdt = to_local_zdt(self.anchor_date)
+            assert anchor_zdt is not None
+
+            local_start = to_local_zdt(start)
+            assert local_start is not None
+
+            local_end = to_local_zdt(end)
 
             offset = (
                 local_start - anchor_zdt

tests/server/schemas/test_schedules.py

@@ -261,6 +261,44 @@ async def test_get_dates_from_offset_naive_end(self):
             datetime(2022, 1, 3, tzinfo=ZoneInfo("UTC")),
         ]
 
+    @pytest.mark.skipif(
+        sys.version_info < (3, 13),
+        reason="Bug only affects Python 3.13+ with whenever library",
+    )
+    async def test_offset_anchor_preserves_local_hour(self):
+        """
+        Regression test for timezone bug from pendulum -> datetime/zoneinfo refactor.
+        When an anchor_date has a fixed-offset tzinfo (e.g. timezone(timedelta(hours=9)))
+        alongside a named timezone (e.g. "Asia/Tokyo"), the scheduler must preserve the
+        instant rather than misinterpreting it as UTC.
+        """
+        from datetime import timezone
+
+        anchor = datetime(
+            2024,
+            7,
+            1,
+            1,
+            15,
+            tzinfo=timezone(timedelta(hours=9)),  # Asia/Tokyo offset
+        )
+        clock = IntervalSchedule(
+            interval=timedelta(days=1),
+            anchor_date=anchor,
+            timezone="Asia/Tokyo",
+        )
+
+        start = datetime(2025, 10, 23, tzinfo=ZoneInfo("UTC"))
+        dates = await clock.get_dates(n=3, start=start)
+        expected_first = datetime(2025, 10, 24, 1, 15, tzinfo=ZoneInfo("Asia/Tokyo"))
+
+        assert dates[0] == expected_first
+        assert all(
+            date.astimezone(ZoneInfo("Asia/Tokyo")).time() == expected_first.time()
+            for date in dates
+        )
+        assert all(date.tzinfo.key == "Asia/Tokyo" for date in dates)
+
 
 class TestCreateCronSchedule:
     def test_create_cron_schedule(self):