fix(source-twilio): Resume messages from saved cursor

airbyte-integrations/connectors/source-twilio/components.py

@@ -72,14 +72,21 @@ class TwilioStateMigration(StateMigration):
     """
 
     def migrate(self, stream_state: Mapping[str, Any]) -> Mapping[str, Any]:
+        if self._has_global_messages_state(stream_state):
+            return stream_state["state"]
+
         for state in stream_state.get("states", []):
             state["partition"]["parent_slice"] = {}
         return stream_state
 
     def should_migrate(self, stream_state: Mapping[str, Any]) -> bool:
-        if stream_state and any("parent_slice" not in state["partition"] for state in stream_state.get("states", [])):
-            return True
-        return False
+        return self._has_global_messages_state(stream_state) or (
+            bool(stream_state) and any("parent_slice" not in state["partition"] for state in stream_state.get("states", []))
+        )
+
+    @staticmethod
+    def _has_global_messages_state(stream_state: Mapping[str, Any]) -> bool:
+        return "date_sent" in stream_state.get("state", {})
 
 
 class TwilioAlertsStateMigration(StateMigration):

airbyte-integrations/connectors/source-twilio/manifest.yaml

@@ -569,6 +569,7 @@ definitions:
       $ref: "#/definitions/base_nested_incremental_from_accounts_stream/incremental_sync"
       datetime_format: "%Y-%m-%d %H:%M:%SZ"
       cursor_granularity: PT0.000001S
+      global_substream_cursor: true
       start_datetime:
         type: MinMaxDatetime
         datetime: "{{ max(format_datetime(config.get('start_date', '1970-01-01T00:00:00Z'), '%Y-%m-%d %H:%M:%SZ'), day_delta(-400, format='%Y-%m-%d %H:%M:%SZ')) }}"

airbyte-integrations/connectors/source-twilio/metadata.yaml

@@ -14,7 +14,7 @@ data:
   connectorSubtype: api
   connectorType: source
   definitionId: b9dc6155-672e-42ea-b10d-9f1f1fb95ab1
-  dockerImageTag: 1.0.0
+  dockerImageTag: 1.0.1
   dockerRepository: airbyte/source-twilio
   documentationUrl: https://docs.airbyte.com/integrations/sources/twilio
   githubIssueLabel: source-twilio

airbyte-integrations/connectors/source-twilio/unit_tests/test_streams.py

@@ -133,14 +133,7 @@ def test_calls_includes_date_window_params(self, requests_mock):
                 "/Accounts/AC123/Messages.json",
                 "DateSent>",
                 "DateSent<",
-                {
-                    "states": [
-                        {
-                            "partition": {"subresource_uri": "/2010-04-01/Accounts/AC123/Messages.json"},
-                            "cursor": {"date_sent": "2022-11-13T12:11:10Z"},
-                        }
-                    ]
-                },
+                {"date_sent": "2022-11-13 12:11:10Z"},
                 [
                     ("2022-11-13 12:11:10Z", "2022-11-16 12:03:11Z"),
                 ],
@@ -205,6 +198,40 @@ def _match(req):
         assert all(m.called for m in child_matchers), "Not all date-window URLs were called"
         assert sum(m.call_count for m in child_matchers) == len(windows)
 
+    @freeze_time("2022-11-16 12:03:11+00:00")
+    def test_messages_prefers_global_cursor_over_partition_state(self, requests_mock):
+        requests_mock.get(f"{BASE}/Accounts.json", json=ACCOUNTS_JSON, status_code=200)
+
+        matcher = requests_mock.get(
+            f"{BASE}/Accounts/AC123/Messages.json",
+            json={"messages": [{"sid": "SM1", "date_sent": "2022-11-16 01:00:00Z"}]},
+            status_code=200,
+            additional_matcher=lambda req: parse_qs(urlparse(req.url).query, keep_blank_values=True).get("DateSent>")
+            == ["2022-11-15 00:00:00Z"],
+        )
+        state = (
+            StateBuilder()
+            .with_stream_state(
+                "messages",
+                {
+                    "state": {"date_sent": "2022-11-15 00:00:00Z"},
+                    "states": [
+                        {
+                            "partition": {"subresource_uri": "/2010-04-01/Accounts/AC123/Messages.json"},
+                            "cursor": {"date_sent": "2022-01-01 00:00:00Z"},
+                        }
+                    ],
+                    "use_global_cursor": False,
+                },
+            )
+            .build()
+        )
+
+        records = read_from_stream(TEST_CONFIG, "messages", SyncMode.incremental, state).records
+
+        assert matcher.called
+        assert len(records) == 1
+
     @freeze_time("2022-11-16 12:03:11+00:00")
     def test_alerts_pagination_limit_error_message(self, requests_mock):
         requests_mock.get(

docs/integrations/sources/twilio.md

@@ -144,6 +144,7 @@ For programmatic configuration, use these parameter names:
 
 | Version | Date | Pull Request | Subject |
 | :------ | :--- | :----------- | :------ |
+| 1.0.1 | 2026-05-22 | [*PR_NUMBER_PLACEHOLDER*](https://github.com/airbytehq/airbyte/pull/*PR_NUMBER_PLACEHOLDER*) | Fix `messages` incremental state handling so syncs resume from the saved `date_sent` cursor. |
 | 1.0.0 | 2026-05-21 | [76911](https://github.com/airbytehq/airbyte/pull/76911) | Migrate `services` and `roles` streams from the deprecated Programmable Chat API to the Conversations API. See the [migration guide](./twilio-migrations.md#upgrading-to-100). |
 | 0.17.11 | 2026-05-21 | [78322](https://github.com/airbytehq/airbyte/pull/78322) | Revert concurrency tuning rollout |
 | 0.17.10 | 2026-05-12 | [77988](https://github.com/airbytehq/airbyte/pull/77988) | Improve the Twilio Alerts pagination-limit error message. |