feat(declarative): add RESET_OPAQUE_TOKEN recovery for OpaqueTokenCursor (2/2)

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

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

airbyte_cdk/sources/declarative/incremental/opaque_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_OPAQUE_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_OPAQUE_TOKEN = "RESET_OPAQUE_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_OPAQUE_TOKEN",
             "RATE_LIMITED",
             "REFRESH_TOKEN_THEN_RETRY",
             "REDUCE_PAGE_SIZE",

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -4182,6 +4182,9 @@ def _get_url(req: Requester) -> str:
             page_observer=(
                 cursor.on_page_completed if isinstance(cursor, OpaqueTokenCursor) else None
             ),
+            token_reset_handler=(
+                cursor.restart_without_token if isinstance(cursor, OpaqueTokenCursor) 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_OPAQUE_TOKEN,
                 ResponseAction.REDUCE_PAGE_SIZE,
                 ResponseAction.SPLIT_REQUEST_WINDOW,
             ]:

airbyte_cdk/sources/declarative/retrievers/simple_retriever.py

@@ -57,6 +57,10 @@
 from airbyte_cdk.sources.declarative.stream_slicers.stream_slicer import StreamSlicer
 from airbyte_cdk.sources.source import ExperimentalClassWarning
 from airbyte_cdk.sources.streams.core import StreamData
+from airbyte_cdk.sources.streams.http.opaque_token_reset_exception import (
+    OpaqueTokenResetNotSupportedException,
+    OpaqueTokenResetRequiredException,
+)
 from airbyte_cdk.sources.streams.http.page_size_reduction_exception import (
     PageSizeReductionNotSupportedException,
     PageSizeReductionRequiredException,
@@ -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 OpaqueTokenResetRequiredException as exception:
+            if self.token_reset_handler is None:
+                raise OpaqueTokenResetNotSupportedException(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_OPAQUE_TOKEN = "RESET_OPAQUE_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

@@ -43,6 +43,9 @@
     RequestBodyException,
     UserDefinedBackoffException,
 )
+from airbyte_cdk.sources.streams.http.opaque_token_reset_exception import (
+    OpaqueTokenResetRequiredException,
+)
 from airbyte_cdk.sources.streams.http.page_size_reduction_exception import (
     PageSizeReductionRequiredException,
 )
@@ -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_OPAQUE_TOKEN:
+            raise OpaqueTokenResetRequiredException(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/opaque_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 OpaqueTokenResetRequiredException(Exception):
+    def __init__(self, error_message: Optional[str] = None) -> None:
+        self.error_message = error_message
+        super().__init__(error_message)
+
+
+class OpaqueTokenResetNotSupportedException(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 an OpaqueTokenCursor. "
+            "Remove the RESET_OPAQUE_TOKEN action from its error handler."
+        )
+        super().__init__(
+            internal_message=message,
+            message=message,
+            failure_type=FailureType.config_error,
+        )