feat: add SPLIT_REQUEST_WINDOW action for adaptive request-window splitting

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -2803,6 +2803,7 @@ definitions:
           - RATE_LIMITED
           - REFRESH_TOKEN_THEN_RETRY
           - REDUCE_PAGE_SIZE
+          - SPLIT_REQUEST_WINDOW
         examples:
           - SUCCESS
           - FAIL
@@ -2812,6 +2813,7 @@ definitions:
           - RATE_LIMITED
           - REFRESH_TOKEN_THEN_RETRY
           - REDUCE_PAGE_SIZE
+          - SPLIT_REQUEST_WINDOW
       failure_type:
         title: Failure Type
         description: Failure type of traced exception if a response matches the filter.
@@ -4379,6 +4381,13 @@ definitions:
           re-issuing the page would emit them twice. A page_token_option of type RequestPath is rejected as
           well, because the next page is then a URL built by the API which already carries the page size.
         "$ref": "#/definitions/PageSizeReduction"
+      request_window_splitting:
+        description: >-
+          Describes how the request window is split when an error handler resolves to the
+          SPLIT_REQUEST_WINDOW action. Requires an incremental_sync cursor that defines cursor_granularity.
+          Distinct from page_size_reduction: this splits the cursor's window rather than a paginator's page
+          size, and applies even when the API rejects the request before returning a single page.
+        "$ref": "#/definitions/RequestWindowSplitting"
       ignore_stream_slicer_parameters_on_paginated_requests:
         description: If true, the partition router and incremental request options will be ignored when paginating requests. Request options set directly on the requester will not be ignored.
         type: boolean
@@ -4559,6 +4568,42 @@ definitions:
           - NEVER
           - AFTER_SUCCESSFUL_PAGE
         default: NEVER
+  RequestWindowSplitting:
+    title: Request Window Splitting
+    description: >-
+      Describes how the request window is split when an error handler resolves to the SPLIT_REQUEST_WINDOW
+      action. The connector replaces the failing window with two smaller, ordered windows covering the same
+      range and reads each one fully before checkpointing. If the failing window had already emitted records,
+      those may be re-emitted; downstream primary-key deduplication is expected to handle that. Only supported
+      on a stream whose incremental_sync cursor defines cursor_granularity. Distinct from page_size_reduction,
+      which shrinks a paginator's page size rather than the cursor's window.
+    type: object
+    required:
+      - type
+    properties:
+      type:
+        type: string
+        enum: [RequestWindowSplitting]
+      failure_message:
+        title: Failure Message
+        description: >-
+          Sentence appended to the error message when the connector can no longer split the window further.
+          Use it to tell the user what to do for this API, for instance which filter narrows the query down.
+        type: string
+        examples:
+          - Narrow the sync down by selecting fewer fields on this stream.
+          - Lower time_window so that each request covers less data.
+      min_split_window:
+        title: Minimum Split Window
+        description: >-
+          Approximate smallest window (ISO 8601 duration) the connector will split down to - a window at or
+          below this size is not split further, though the result can end up a bit smaller than this, not
+          larger. Independent of, and usually looser than, cursor_granularity - useful when a cursor could
+          split further but many small requests are worse for this API than a few large ones.
+        type: string
+        examples:
+          - "P1D"
+          - "PT1H"
   GzipDecoder:
     title: gzip
     description: Select 'gzip' for response data that is compressed with gzip. Requires specifying an inner data type/decoder to parse the decompressed data.

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

@@ -72,6 +72,7 @@ def interpret_response(
                 ResponseAction.IGNORE,
                 ResponseAction.RESET_PAGINATION,
                 ResponseAction.REDUCE_PAGE_SIZE,
+                ResponseAction.SPLIT_REQUEST_WINDOW,
             ]:
                 return matched_error_resolution
 

airbyte_cdk/sources/declarative/requesters/paginators/default_paginator.py

@@ -268,6 +268,10 @@ class PaginatorTestReadDecorator(Paginator):
     class (`connector_builder/test_reader/reader.py::_has_reached_limit`). The two counts agree because
     `HttpClient` logs a response resolving to `REDUCE_PAGE_SIZE` as an auxiliary request, which the Builder
     does not turn into a page.
+
+    `get_initial_token()` also resets this count, and `SimpleRetriever._read_pages` calls it again from scratch
+    for every child window a `SPLIT_REQUEST_WINDOW` response produces. A split slice can therefore fetch up to
+    `maximum_number_of_pages` per child rather than sharing one budget across the whole original partition.
     """
 
     _PAGE_COUNT_BEFORE_FIRST_NEXT_CALL = 1

airbyte_cdk/sources/declarative/retrievers/request_window_splitting.py

@@ -0,0 +1,21 @@
+#
+# Copyright (c) 2026 Airbyte, Inc., all rights reserved.
+#
+
+import datetime
+from dataclasses import dataclass
+from typing import Optional
+
+
+@dataclass(frozen=True)
+class RequestWindowSplitting:
+    """
+    Configuration for the `request_window_splitting` retriever field - created once per stream and shared,
+    mirroring `PageSizeReduction`. Unlike `PageSizeReducer`, there is no per-partition mutable counterpart: a
+    split window is either split again (recursion) or the sync fails, so a plain config object is enough.
+    """
+
+    failure_message: Optional[str] = None
+    # Independent of, and typically looser than, the cursor's own granularity floor - useful when a cursor
+    # could split further but the API's rate limit makes many small requests worse than a few large ones.
+    min_split_window: Optional[datetime.timedelta] = None

airbyte_cdk/sources/declarative/retrievers/simple_retriever.py

@@ -2,6 +2,7 @@
 # Copyright (c) 2025 Airbyte, Inc., all rights reserved.
 #
 
+import datetime
 import json
 import logging
 from collections import defaultdict
@@ -49,6 +50,9 @@
     PageSizeReduction,
 )
 from airbyte_cdk.sources.declarative.retrievers.pagination_tracker import PaginationTracker
+from airbyte_cdk.sources.declarative.retrievers.request_window_splitting import (
+    RequestWindowSplitting,
+)
 from airbyte_cdk.sources.declarative.retrievers.retriever import Retriever
 from airbyte_cdk.sources.declarative.stream_slicers.stream_slicer import StreamSlicer
 from airbyte_cdk.sources.source import ExperimentalClassWarning
@@ -60,13 +64,21 @@
 from airbyte_cdk.sources.streams.http.pagination_reset_exception import (
     PaginationResetRequiredException,
 )
+from airbyte_cdk.sources.streams.http.request_window_split_exception import (
+    RequestWindowSplitNotSupportedException,
+    RequestWindowSplitRequiredException,
+)
 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
 
 FULL_REFRESH_SYNC_COMPLETE_KEY = "__ab_full_refresh_sync_complete"
 LOGGER = logging.getLogger("airbyte")
 
+# A defense-in-depth bound independent of any `request_window_splitter`'s own no-progress guard: protects
+# against a misbehaving custom cursor whose children don't actually shrink the window. Not user-configurable.
+_MAX_REQUEST_WINDOW_SPLIT_DEPTH = 10
+
 
 @dataclass
 class SimpleRetriever(Retriever):
@@ -97,6 +109,14 @@ class SimpleRetriever(Retriever):
             `_read_pages` creates per call, so the retriever and its paginator - both shared by every
             partition of the stream, read concurrently - stay stateless. When `page_size_reduction` is
             `None` no reducer is created and `_read_pages` keeps its previous behaviour
+        request_window_splitting (Optional[RequestWindowSplitting]): Policy applied when an error handler
+            resolves to `ResponseAction.SPLIT_REQUEST_WINDOW`, or when `RequestWindowSplitRequiredException`
+            is raised directly by custom code. `None` disables request-window splitting entirely: `read_records`
+            converts the exception into `RequestWindowSplitNotSupportedException` instead of attempting a
+            split it cannot perform.
+        request_window_splitter: Bound method - normally the stream's own cursor's `split_request_window` -
+            asked to replace a failing `StreamSlice` with smaller children. `None` has the same effect as
+            `request_window_splitting` being `None`.
     """
 
     requester: Requester
@@ -122,6 +142,10 @@ class SimpleRetriever(Retriever):
     )
     post_pagination_filter: Optional[ClientSideIncrementalRecordFilterDecorator] = None
     page_size_reduction: Optional[PageSizeReduction] = None
+    request_window_splitting: Optional[RequestWindowSplitting] = None
+    request_window_splitter: Optional[
+        Callable[[StreamSlice, Optional[datetime.timedelta]], Optional[List[StreamSlice]]]
+    ] = None
 
     def __post_init__(self, parameters: Mapping[str, Any]) -> None:
         self._paginator = self.paginator or NoPagination(parameters=parameters)
@@ -547,27 +571,158 @@ def read_records(
         :return: The records read from the API source
         """
         _slice = stream_slice or StreamSlice(partition={}, cursor_slice={})  # None-check
+        yield from self._read_records_or_split_request_window(
+            records_schema, _slice, _slice, depth=0
+        )
 
+    def _read_records_or_split_request_window(
+        self,
+        records_schema: Mapping[str, Any],
+        stream_slice: StreamSlice,
+        original_slice: StreamSlice,
+        depth: int,
+    ) -> Iterable[StreamData]:
+        """
+        Read `stream_slice` to completion, replacing it with smaller children and recursing into each of them in
+        turn when a `RequestWindowSplitRequiredException` is raised while reading it.
+
+        Each recursive call re-enters this method - and, through it, `_read_pages` - from scratch for the child
+        slice it is given, so every child gets its own paginator token, `PaginationTracker`, and
+        `PageSizeReducer` for free: nothing here needs to reset that state explicitly. Because the whole
+        recursion lives inside the single generator `DeclarativePartition.read()` consumes,
+        `PartitionReader.process_partition()` only calls `cursor.close_partition()` once, after this generator is
+        fully exhausted - so a failure anywhere in the recursion (a child that cannot be read, or a window that
+        cannot be split any further) propagates out without ever checkpointing the original partition.
+
+        `original_slice` is threaded through the recursion unchanged so every record yielded, however deep the
+        recursion went to produce it, is re-stamped with the partition's own slice rather than the child slice it
+        was actually read against - see `_reassociate_with_original_slice`. `depth` bounds the recursion
+        independently of `request_window_splitter`'s own no-progress guard: it is enforced here so a
+        misbehaving custom cursor cannot recurse indefinitely regardless of what that guard does or does not
+        catch.
+        """
         record_generator = partial(
             self._parse_records,
             stream_slice=stream_slice,
             records_schema=records_schema,
         )
-        records: Iterable[Mapping[str, Any]] = self._read_pages(record_generator, _slice)
-        if self.post_pagination_filter:
-            # A data feed paginates until it reaches a record older than the cursor, so the page that triggers the stop
-            # condition still holds already-synced records. Those are filtered here rather than in the record selector
-            # so that the paginator keeps seeing the whole page: the stop condition is evaluated on the last record of
-            # the page, which is precisely one of the records being dropped. Two consequences of filtering this late:
-            # the pagination tracker observes the dropped records, and a `file_uploader` on the record selector has
-            # already uploaded their files by the time they are dropped.
-            records = self.post_pagination_filter.filter_records(
-                records,
-                # the filter is only used for its cursor comparison, which does not read the stream state
-                stream_state={},
-                stream_slice=_slice,
+
+        emitted_count = 0
+        try:
+            records: Iterable[Mapping[str, Any]] = self._read_pages(record_generator, stream_slice)
+            if self.post_pagination_filter:
+                # A data feed paginates until it reaches a record older than the cursor, so the page that triggers the stop
+                # condition still holds already-synced records. Those are filtered here rather than in the record selector
+                # so that the paginator keeps seeing the whole page: the stop condition is evaluated on the last record of
+                # the page, which is precisely one of the records being dropped. Two consequences of filtering this late:
+                # the pagination tracker observes the dropped records, and a `file_uploader` on the record selector has
+                # already uploaded their files by the time they are dropped.
+                records = self.post_pagination_filter.filter_records(
+                    records,
+                    # the filter is only used for its cursor comparison, which does not read the stream state
+                    stream_state={},
+                    stream_slice=stream_slice,
+                )
+            for record in records:
+                emitted_count += 1
+                yield self._reassociate_with_original_slice(record, original_slice)
+        except RequestWindowSplitRequiredException as exception:
+            if self.request_window_splitting is None or self.request_window_splitter is None:
+                raise RequestWindowSplitNotSupportedException(stream_name=self.name) from exception
+
+            if emitted_count:
+                # Splitting and re-reading the window re-emits these records: a failed partition is never
+                # checkpointed, so the next attempt would re-emit them anyway. Always replaying makes forward
+                # progress instead of retrying the same oversized window forever; primary-key dedup downstream
+                # handles the duplicates either way.
+                LOGGER.warning(
+                    f"Stream {self.name} already emitted {emitted_count} record(s) from {stream_slice} before "
+                    f"it was rejected; splitting and re-reading the window may re-emit them."
+                )
+
+            if depth >= _MAX_REQUEST_WINDOW_SPLIT_DEPTH:
+                # Logged separately from the exception below so it stays visible even if the trace message's
+                # internal_message isn't surfaced by whatever catches it.
+                LOGGER.warning(
+                    f"Stream {self.name} hit the maximum request window split depth "
+                    f"({_MAX_REQUEST_WINDOW_SPLIT_DEPTH}) while splitting {original_slice}."
+                )
+                raise AirbyteTracedException(
+                    internal_message=f"Stream {self.name} exceeded the maximum request window split depth of {_MAX_REQUEST_WINDOW_SPLIT_DEPTH} while splitting {original_slice}",
+                    # `transient_error`, so the only remediation is the connector's own, if it defined one.
+                    message=self._with_request_window_failure_message(
+                        f"Stream {self.name} could not split its request window to a size the API accepts "
+                        f"within {_MAX_REQUEST_WINDOW_SPLIT_DEPTH} splits. This usually means the stream's "
+                        f"cursor is not actually shrinking the window on each split; if it uses a custom "
+                        f"cursor, check its `split_request_window` implementation."
+                    ),
+                    failure_type=FailureType.transient_error,
+                ) from exception
+
+            children = self.request_window_splitter(
+                stream_slice, self.request_window_splitting.min_split_window
+            )
+            if children is None:
+                min_split_window_note = (
+                    f", its configured `min_split_window` ({self.request_window_splitting.min_split_window})"
+                    if self.request_window_splitting.min_split_window
+                    else ""
+                )
+                raise AirbyteTracedException(
+                    internal_message=f"Stream {self.name} could not split its request window {stream_slice} any further",
+                    # `transient_error` regardless of how the triggering response was classified: exhaustion is
+                    # never the user's fault, matching `PageSizeReducer`'s equivalent exhaustion branch.
+                    message=self._with_request_window_failure_message(
+                        f"The API kept rejecting stream {self.name}'s request window even at the smallest window "
+                        f"its cursor allows{min_split_window_note}, or splitting the window further would not "
+                        f"make progress."
+                    ),
+                    failure_type=FailureType.transient_error,
+                ) from exception
+
+            LOGGER.info(
+                f"Stream {self.name}: the API rejected request window {stream_slice} (split depth {depth}); "
+                f"reducing it to {children} and reading each in turn."
+            )
+            for child in children:
+                yield from self._read_records_or_split_request_window(
+                    records_schema, child, original_slice, depth + 1
+                )
+
+    def _with_request_window_failure_message(self, message: str) -> str:
+        """
+        :return: the message followed by `request_window_splitting`'s configured `failure_message`, when it
+            defined one - mirroring `PageSizeReducer._with_failure_message`
+        """
+        assert self.request_window_splitting is not None
+        failure_message = (self.request_window_splitting.failure_message or "").strip()
+        if not failure_message:
+            return message
+        return f"{message} {failure_message}"
+
+    @staticmethod
+    def _reassociate_with_original_slice(
+        record: StreamData, original_slice: StreamSlice
+    ) -> StreamData:
+        """
+        Records read while recursing into a split child window are parsed against that child's `StreamSlice`,
+        so `RecordSelector` stamps them with the child as `associated_slice`. `DeclarativePartition.read()`
+        passes an already-built `Record` through unchanged rather than re-wrapping it, so left uncorrected the
+        child slice - not the partition's own slice - is what `ConcurrentCursor.observe()` would key its
+        per-partition bookkeeping by. `close_partition()` always looks that bookkeeping up by the partition's own
+        slice, so it would never find it, silently losing the precise most-recently-observed cursor value for a
+        split partition (recoverable in practice today only because callers of that lookup fall back to the
+        slice's end boundary when it is missing). Re-stamping every record with `original_slice` here restores
+        the same tracking a non-split read already gets, rather than relying on that fallback.
+        """
+        if isinstance(record, Record) and record.associated_slice is not original_slice:
+            return Record(
+                data=record.data,
+                stream_name=record.stream_name,
+                associated_slice=original_slice,
+                file_reference=record.file_reference,
             )
-        yield from records
+        return record
 
     def _parse_records(
         self,