fix: normalize timezone-aware cursor values to UTC before serializing state

airbyte_cdk/sources/streams/concurrent/state_converters/datetime_stream_state_converter.py

@@ -42,6 +42,16 @@ def increment(self, timestamp: datetime) -> datetime: ...
     @abstractmethod
     def parse_timestamp(self, timestamp: Any) -> datetime: ...
 
+    @staticmethod
+    def _to_utc(timestamp: datetime) -> datetime:
+        # Serialized state must represent an absolute instant in UTC. A timezone-aware
+        # datetime carrying a non-UTC offset would otherwise be formatted with its local
+        # wall-clock components, relabeling e.g. 21:59:38+05:30 as 21:59:38Z. Naive values
+        # are treated as UTC to match parse()/from_datetime() conventions.
+        if timestamp.tzinfo is None:
+            return timestamp.replace(tzinfo=timezone.utc)
+        return timestamp.astimezone(timezone.utc)
+
     @abstractmethod
     def output_format(self, timestamp: datetime) -> Any: ...
 
@@ -176,7 +186,7 @@ def output_format(self, timestamp: datetime) -> str:
         Returns:
             str: ISO8601/RFC3339 formatted string with milliseconds.
         """
-        dt = AirbyteDateTime.from_datetime(timestamp)
+        dt = AirbyteDateTime.from_datetime(self._to_utc(timestamp))
         # Always include milliseconds, even if zero
         millis = dt.microsecond // 1000 if dt.microsecond else 0
         return f"{dt.year:04d}-{dt.month:02d}-{dt.day:02d}T{dt.hour:02d}:{dt.minute:02d}:{dt.second:02d}.{millis:03d}Z"
@@ -212,7 +222,7 @@ def __init__(
         self._parser = DatetimeParser()
 
     def output_format(self, timestamp: datetime) -> str:
-        return self._parser.format(timestamp, self._datetime_format)
+        return self._parser.format(self._to_utc(timestamp), self._datetime_format)
 
     def parse_timestamp(self, timestamp: str) -> datetime:
         for datetime_format in self._input_datetime_formats:

unit_tests/sources/streams/concurrent/test_datetime_state_converter.py

@@ -2,7 +2,7 @@
 # Copyright (c) 2023 Airbyte, Inc., all rights reserved.
 #
 
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
 
 import pytest
 
@@ -406,3 +406,52 @@ def test_given_when_parse_timestamp_then_eventually_fallback_on_output_format():
     parsed_datetime = converter.parse_timestamp("2024-01-01T02:00:00")
 
     assert parsed_datetime == datetime(2024, 1, 1, 2, 0, 0, tzinfo=timezone.utc)
+
+
+@pytest.mark.parametrize(
+    "converter,timestamp,expected_output",
+    [
+        pytest.param(
+            CustomFormatConcurrentStreamStateConverter(datetime_format="%Y-%m-%dT%H:%M:%SZ"),
+            datetime(2026, 7, 23, 21, 59, 38, tzinfo=timezone(timedelta(hours=5, minutes=30))),
+            "2026-07-23T16:29:38Z",
+            id="custom-format-positive-offset-normalized-to-utc",
+        ),
+        pytest.param(
+            CustomFormatConcurrentStreamStateConverter(datetime_format="%Y-%m-%dT%H:%M:%SZ"),
+            datetime(2026, 7, 23, 12, 1, 25, tzinfo=timezone(timedelta(hours=-5))),
+            "2026-07-23T17:01:25Z",
+            id="custom-format-negative-offset-normalized-to-utc",
+        ),
+        pytest.param(
+            CustomFormatConcurrentStreamStateConverter(datetime_format="%Y-%m-%dT%H:%M:%SZ"),
+            datetime(2026, 7, 23, 16, 29, 38, tzinfo=timezone.utc),
+            "2026-07-23T16:29:38Z",
+            id="custom-format-utc-unchanged",
+        ),
+        pytest.param(
+            CustomFormatConcurrentStreamStateConverter(datetime_format="%Y-%m-%dT%H:%M:%SZ"),
+            datetime(2026, 7, 23, 16, 29, 38),
+            "2026-07-23T16:29:38Z",
+            id="custom-format-naive-treated-as-utc",
+        ),
+        pytest.param(
+            IsoMillisConcurrentStreamStateConverter(),
+            datetime(2026, 7, 23, 21, 59, 38, tzinfo=timezone(timedelta(hours=5, minutes=30))),
+            "2026-07-23T16:29:38.000Z",
+            id="iso-millis-positive-offset-normalized-to-utc",
+        ),
+        pytest.param(
+            IsoMillisConcurrentStreamStateConverter(),
+            datetime(2026, 7, 23, 16, 29, 38),
+            "2026-07-23T16:29:38.000Z",
+            id="iso-millis-naive-treated-as-utc",
+        ),
+    ],
+)
+def test_output_format_normalizes_offset_aware_datetime_to_utc(
+    converter, timestamp, expected_output
+):
+    """A cursor value carrying a non-UTC offset must be serialized as its absolute UTC
+    instant, not its local wall clock relabeled with `Z`."""
+    assert converter.output_format(timestamp) == expected_output