feat(declarative): add RESET_SYNC_TOKEN recovery for SyncTokenCursor (2/2)

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -2884,6 +2884,7 @@ definitions:
           - RETRY
           - IGNORE
           - RESET_PAGINATION
+          - RESET_SYNC_TOKEN
           - RATE_LIMITED
           - REFRESH_TOKEN_THEN_RETRY
           - REDUCE_PAGE_SIZE
@@ -2894,6 +2895,7 @@ definitions:
           - RETRY
           - IGNORE
           - RESET_PAGINATION
+          - RESET_SYNC_TOKEN
           - RATE_LIMITED
           - REFRESH_TOKEN_THEN_RETRY
           - REDUCE_PAGE_SIZE

airbyte_cdk/sources/declarative/incremental/sync_token_cursor.py

@@ -197,6 +197,30 @@ def on_page_completed(self, page: PageCompleted) -> None:
                 .replace("+00:00", "Z"),
             }
 
+    def restart_without_token(
+        self, stream_slice: StreamSlice, reason: Optional[str] = None
+    ) -> StreamSlice:
+        if "token" not in stream_slice.cursor_slice:
+            message = (
+                f"Stream {self._stream_name}: the API rejected a read without a change token. "
+                "Check the RESET_SYNC_TOKEN response filter."
+            )
+            raise AirbyteTracedException(
+                internal_message=message,
+                message=message,
+                failure_type=FailureType.system_error,
+            )
+
+        partition_key = self._serializer.to_partition_key(stream_slice.partition or {})
+        with self._lock:
+            self._pending.pop(partition_key, None)
+
+        return StreamSlice(
+            partition=stream_slice.partition,
+            cursor_slice={},
+            extra_fields=stream_slice.extra_fields,
+        )
+
     def _coerce_token(self, value: Any, path: list[str]) -> Optional[str]:
         if value is None or value == "":
             return None

airbyte_cdk/sources/declarative/models/declarative_component_schema.py

@@ -614,6 +614,7 @@ class Action(Enum):
     RETRY = "RETRY"
     IGNORE = "IGNORE"
     RESET_PAGINATION = "RESET_PAGINATION"
+    RESET_SYNC_TOKEN = "RESET_SYNC_TOKEN"
     RATE_LIMITED = "RATE_LIMITED"
     REFRESH_TOKEN_THEN_RETRY = "REFRESH_TOKEN_THEN_RETRY"
     REDUCE_PAGE_SIZE = "REDUCE_PAGE_SIZE"
@@ -637,6 +638,7 @@ class HttpResponseFilter(BaseModel):
             "RETRY",
             "IGNORE",
             "RESET_PAGINATION",
+            "RESET_SYNC_TOKEN",
             "RATE_LIMITED",
             "REFRESH_TOKEN_THEN_RETRY",
             "REDUCE_PAGE_SIZE",

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -4181,6 +4181,9 @@ def _get_url(req: Requester) -> str:
             page_observer=(
                 cursor.on_page_completed if isinstance(cursor, SyncTokenCursor) else None
             ),
+            token_reset_handler=(
+                cursor.restart_without_token if isinstance(cursor, SyncTokenCursor) else None
+            ),
             parameters=model.parameters or {},
         )
 

airbyte_cdk/sources/declarative/requesters/error_handlers/composite_error_handler.py

@@ -71,6 +71,7 @@ def interpret_response(
                 ResponseAction.RETRY,
                 ResponseAction.IGNORE,
                 ResponseAction.RESET_PAGINATION,
+                ResponseAction.RESET_SYNC_TOKEN,
                 ResponseAction.REDUCE_PAGE_SIZE,
                 ResponseAction.SPLIT_REQUEST_WINDOW,
             ]:

airbyte_cdk/sources/declarative/retrievers/simple_retriever.py

@@ -68,6 +68,10 @@
     RequestWindowSplitNotSupportedException,
     RequestWindowSplitRequiredException,
 )
+from airbyte_cdk.sources.streams.http.sync_token_reset_exception import (
+    SyncTokenResetNotSupportedException,
+    SyncTokenResetRequiredException,
+)
 from airbyte_cdk.sources.types import Config, Record, StreamSlice
 from airbyte_cdk.utils.mapping_helpers import combine_mappings
 from airbyte_cdk.utils.traced_exception import AirbyteTracedException
@@ -154,6 +158,7 @@ class SimpleRetriever(Retriever):
         Callable[[StreamSlice, Optional[datetime.timedelta]], Optional[List[StreamSlice]]]
     ] = None
     page_observer: Optional[Callable[["PageCompleted"], None]] = None
+    token_reset_handler: Optional[Callable[[StreamSlice, Optional[str]], StreamSlice]] = None
 
     def __post_init__(self, parameters: Mapping[str, Any]) -> None:
         self._paginator = self.paginator or NoPagination(parameters=parameters)
@@ -703,6 +708,19 @@ def _read_records_or_split_request_window(
                 yield from self._read_records_or_split_request_window(
                     records_schema, child, original_slice, depth + 1
                 )
+        except SyncTokenResetRequiredException as exception:
+            if self.token_reset_handler is None:
+                raise SyncTokenResetNotSupportedException(stream_name=self.name) from exception
+
+            reset_slice = self.token_reset_handler(stream_slice, exception.error_message)
+            reason = exception.error_message or "the API rejected the stored change token"
+            LOGGER.warning(
+                f"Stream {self.name}: {reason}; reading this stream in full. "
+                f"{emitted_count} records read earlier in this attempt may be emitted again."
+            )
+            yield from self._read_records_or_split_request_window(
+                records_schema, reset_slice, original_slice, depth + 1
+            )
 
     def _with_request_window_failure_message(self, message: str) -> str:
         """

airbyte_cdk/sources/streams/http/error_handlers/response_models.py

@@ -17,6 +17,7 @@ class ResponseAction(Enum):
     FAIL = "FAIL"
     IGNORE = "IGNORE"
     RESET_PAGINATION = "RESET_PAGINATION"
+    RESET_SYNC_TOKEN = "RESET_SYNC_TOKEN"
     RATE_LIMITED = "RATE_LIMITED"
     REFRESH_TOKEN_THEN_RETRY = "REFRESH_TOKEN_THEN_RETRY"
     REDUCE_PAGE_SIZE = "REDUCE_PAGE_SIZE"

airbyte_cdk/sources/streams/http/http_client.py

@@ -65,6 +65,9 @@
     ResponseAwareAuthenticator,
     TokenRotatingAuthenticator,
 )
+from airbyte_cdk.sources.streams.http.sync_token_reset_exception import (
+    SyncTokenResetRequiredException,
+)
 from airbyte_cdk.sources.streams.http.tcp_keepalive import TcpKeepaliveHTTPAdapter
 from airbyte_cdk.sources.utils.types import JsonType
 from airbyte_cdk.utils.airbyte_secrets_utils import filter_secrets
@@ -620,6 +623,9 @@ def _handle_error_resolution(
         if error_resolution.response_action == ResponseAction.RESET_PAGINATION:
             raise PaginationResetRequiredException()
 
+        if error_resolution.response_action == ResponseAction.RESET_SYNC_TOKEN:
+            raise SyncTokenResetRequiredException(error_resolution.error_message)
+
         if error_resolution.response_action == ResponseAction.REDUCE_PAGE_SIZE:
             raise PageSizeReductionRequiredException(
                 stream_name=self._name, error_message=error_resolution.error_message

airbyte_cdk/sources/streams/http/sync_token_reset_exception.py

@@ -0,0 +1,28 @@
+#
+# Copyright (c) 2026 Airbyte, Inc., all rights reserved.
+#
+
+from typing import Optional
+
+from airbyte_cdk.models import FailureType
+from airbyte_cdk.utils.traced_exception import AirbyteTracedException
+
+
+class SyncTokenResetRequiredException(Exception):
+    def __init__(self, error_message: Optional[str] = None) -> None:
+        self.error_message = error_message
+        super().__init__(error_message)
+
+
+class SyncTokenResetNotSupportedException(AirbyteTracedException):
+    def __init__(self, stream_name: Optional[str] = None) -> None:
+        stream = f"Stream {stream_name}" if stream_name else "The stream"
+        message = (
+            f"{stream} resolves to a token reset but does not use a SyncTokenCursor. "
+            "Remove the RESET_SYNC_TOKEN action from its error handler."
+        )
+        super().__init__(
+            internal_message=message,
+            message=message,
+            failure_type=FailureType.config_error,
+        )