diff --git a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml
index 6a814328d4..97dda3b98b 100644
--- a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml
+++ b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml
@@ -1295,11 +1295,11 @@ definitions:
           - "P1D"
       is_data_feed:
         title: Data Feed API
-        description: A data feed API is an API that does not allow filtering and paginates the content from the most recent to the least recent. Given this, the CDK needs to know when to stop paginating and this field will generate a stop condition for pagination.
+        description: A data feed API is an API that does not allow filtering and paginates the content from the most recent to the least recent. Given this, the CDK needs to know when to stop paginating and this field will generate a stop condition for pagination. The last page fetched still holds records that fall outside the cursor window, and those are filtered out as well, so Client-side Incremental Filtering does not need to be enabled alongside this field. Records are kept when their cursor value is within the window that starts at the previous sync's cursor value (or the start date) and ends at the end date, defaulting to the current time, so records dated in the future are filtered out too.
         type: boolean
       is_client_side_incremental:
         title: Client-side Incremental Filtering
-        description: Set to True if the target API endpoint does not take cursor values to filter records and returns all records anyway. This will cause the connector to filter out records locally, and only emit new records from the last sync, hence incremental. This means that all records would be read from the API, but only new records will be emitted to the destination.
+        description: Set to True if the target API endpoint does not take cursor values to filter records and returns all records anyway. This will cause the connector to filter out records locally, keeping only the ones whose cursor value falls within the window that starts at the previous sync's cursor value (or the start date) and ends at the end date, defaulting to the current time. This means that all records would be read from the API, but only the records within that window will be emitted to the destination. This is not needed when Data Feed API is enabled, as a data feed already filters on the same window.
         type: boolean
       is_compare_strictly:
         title: Strict Start-End Time Comparison
diff --git a/airbyte_cdk/sources/declarative/extractors/record_filter.py b/airbyte_cdk/sources/declarative/extractors/record_filter.py
index 943068f875..cacee83727 100644
--- a/airbyte_cdk/sources/declarative/extractors/record_filter.py
+++ b/airbyte_cdk/sources/declarative/extractors/record_filter.py
@@ -72,9 +72,12 @@ def filter_records(
             record
             for record in records
             if self._cursor.should_be_synced(
-                # Record is created on the fly to align with cursors interface; stream name is ignored as we don't need it here
-                # Record stream name is empty because it is not used during the filtering
-                Record(data=record, associated_slice=stream_slice, stream_name="")
+                record
+                if isinstance(record, Record)
+                # Record is created on the fly to align with cursors interface; stream name is empty because it is not
+                # used during the filtering. Callers that already hold records pass them through untouched, so that the
+                # cursor keeps seeing the real stream name and slice.
+                else Record(data=record, associated_slice=stream_slice, stream_name="")
             )
         )
         if self.condition:
diff --git a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py
index af6d566069..a84412eaf9 100644
--- a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py
+++ b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py
@@ -1857,12 +1857,12 @@ class DatetimeBasedCursor(BaseModel):
     )
     is_data_feed: Optional[bool] = Field(
         None,
-        description="A data feed API is an API that does not allow filtering and paginates the content from the most recent to the least recent. Given this, the CDK needs to know when to stop paginating and this field will generate a stop condition for pagination.",
+        description="A data feed API is an API that does not allow filtering and paginates the content from the most recent to the least recent. Given this, the CDK needs to know when to stop paginating and this field will generate a stop condition for pagination. The last page fetched still holds records that fall outside the cursor window, and those are filtered out as well, so Client-side Incremental Filtering does not need to be enabled alongside this field. Records are kept when their cursor value is within the window that starts at the previous sync's cursor value (or the start date) and ends at the end date, defaulting to the current time, so records dated in the future are filtered out too.",
         title="Data Feed API",
     )
     is_client_side_incremental: Optional[bool] = Field(
         None,
-        description="Set to True if the target API endpoint does not take cursor values to filter records and returns all records anyway. This will cause the connector to filter out records locally, and only emit new records from the last sync, hence incremental. This means that all records would be read from the API, but only new records will be emitted to the destination.",
+        description="Set to True if the target API endpoint does not take cursor values to filter records and returns all records anyway. This will cause the connector to filter out records locally, keeping only the ones whose cursor value falls within the window that starts at the previous sync's cursor value (or the start date) and ends at the end date, defaulting to the current time. This means that all records would be read from the API, but only the records within that window will be emitted to the destination. This is not needed when Data Feed API is enabled, as a data feed already filters on the same window.",
         title="Client-side Incremental Filtering",
     )
     is_compare_strictly: Optional[bool] = Field(
diff --git a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py
index 77116fae6c..b992b4201a 100644
--- a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py
+++ b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py
@@ -3330,6 +3330,7 @@ def create_record_selector(
         transformations: List[RecordTransformation] | None = None,
         decoder: Decoder | None = None,
         client_side_incremental_sync_cursor: Optional[Cursor] = None,
+        is_client_side_incremental_sync: bool = False,
         file_uploader: Optional[DefaultFileUploader] = None,
         **kwargs: Any,
     ) -> RecordSelector:
@@ -3342,8 +3343,16 @@ def create_record_selector(
             else None
         )
 
+        # A client-side incremental stream transforms before filtering by default. That default belongs to the flag,
+        # not to the component that ends up doing the cursor comparison: a data feed does it in the retriever and
+        # receives no cursor here, but its `record_filter` condition must keep running after the transformations.
+        default_transform_before_filtering = bool(
+            client_side_incremental_sync_cursor or is_client_side_incremental_sync
+        )
         transform_before_filtering = (
-            False if model.transform_before_filtering is None else model.transform_before_filtering
+            default_transform_before_filtering
+            if model.transform_before_filtering is None
+            else model.transform_before_filtering
         )
         if client_side_incremental_sync_cursor:
             record_filter = ClientSideIncrementalRecordFilterDecorator(
@@ -3354,11 +3363,6 @@ def create_record_selector(
                 else None,
                 cursor=client_side_incremental_sync_cursor,
             )
-            transform_before_filtering = (
-                True
-                if model.transform_before_filtering is None
-                else model.transform_before_filtering
-            )
 
         if model.schema_normalization is None:
             # default to no schema normalization if not set
@@ -3465,6 +3469,33 @@ def _get_url(req: Requester) -> str:
         if cursor is None:
             cursor = FinalStateCursor(name, None, self._message_repository)
 
+        # A data feed drops the records the cursor considers already synced in the retriever, which sits downstream of
+        # the paginator. Letting the record selector drop them as well would be redundant and would hide them from the
+        # pagination stop condition, so a data feed never delegates that filtering to the record selector, whether
+        # `is_client_side_incremental` is set or not. The `condition` from `record_filter` is intentionally left out of
+        # the post-pagination filter and stays in the record selector, which preserves the existing behaviour: the
+        # selector runs inside the page loop, so the records the condition rejects never reach the paginator's
+        # accounting. Moving it downstream would start counting them.
+        post_pagination_filter = (
+            ClientSideIncrementalRecordFilterDecorator(
+                config=config,
+                parameters=model.parameters or {},
+                condition=None,
+                cursor=cursor,
+            )
+            if has_stop_condition_cursor
+            else None
+        )
+        client_side_incremental_cursor = (
+            cursor if is_client_side_incremental_sync and not post_pagination_filter else None
+        )
+        if post_pagination_filter and is_client_side_incremental_sync:
+            LOGGER.warning(
+                f"Stream {name}: `is_client_side_incremental` adds no record filtering when `is_data_feed` is set, "
+                "as a data feed already filters on the cursor value. It still makes the record selector apply the "
+                "transformations before the `record_filter` condition."
+            )
+
         decoder = (
             self._create_component_from_model(model=model.decoder, config=config)
             if model.decoder
@@ -3476,7 +3507,8 @@ def _get_url(req: Requester) -> str:
             config=config,
             decoder=decoder,
             transformations=transformations,
-            client_side_incremental_sync_cursor=cursor if is_client_side_incremental_sync else None,
+            client_side_incremental_sync_cursor=client_side_incremental_cursor,
+            is_client_side_incremental_sync=is_client_side_incremental_sync,
             file_uploader=file_uploader,
         )
 
@@ -3611,6 +3643,7 @@ def _get_url(req: Requester) -> str:
                 request_option_provider=request_options_provider,
                 config=config,
                 ignore_stream_slicer_parameters_on_paginated_requests=ignore_stream_slicer_parameters_on_paginated_requests,
+                post_pagination_filter=post_pagination_filter,
                 parameters=model.parameters or {},
             )
 
@@ -3636,6 +3669,7 @@ def _get_url(req: Requester) -> str:
             pagination_tracker_factory=self._create_pagination_tracker_factory(
                 model.pagination_reset, cursor
             ),
+            post_pagination_filter=post_pagination_filter,
             parameters=model.parameters or {},
         )
 
diff --git a/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py b/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py
index 1f2eb1c668..6f82b8ebd9 100644
--- a/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py
+++ b/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py
@@ -24,6 +24,9 @@
 from typing_extensions import deprecated
 
 from airbyte_cdk.sources.declarative.extractors.http_selector import HttpSelector
+from airbyte_cdk.sources.declarative.extractors.record_filter import (
+    ClientSideIncrementalRecordFilterDecorator,
+)
 from airbyte_cdk.sources.declarative.interpolation import InterpolatedString
 from airbyte_cdk.sources.declarative.partition_routers.single_partition_router import (
     SinglePartitionRouter,
@@ -72,6 +75,8 @@ class SimpleRetriever(Retriever):
         paginator (Optional[Paginator]): The paginator
         stream_slicer (Optional[StreamSlicer]): The stream slicer
         parameters (Mapping[str, Any]): Additional runtime parameters to be used for string interpolation
+        post_pagination_filter (Optional[ClientSideIncrementalRecordFilterDecorator]): Set for data feed streams only.
+            Records the cursor considers already synced are dropped once pagination has observed them
     """
 
     requester: Requester
@@ -95,6 +100,7 @@ class SimpleRetriever(Retriever):
     pagination_tracker_factory: Callable[[], PaginationTracker] = field(
         default_factory=lambda: lambda: PaginationTracker()
     )
+    post_pagination_filter: Optional[ClientSideIncrementalRecordFilterDecorator] = None
 
     def __post_init__(self, parameters: Mapping[str, Any]) -> None:
         self._paginator = self.paginator or NoPagination(parameters=parameters)
@@ -457,7 +463,21 @@ def read_records(
             stream_slice=stream_slice,
             records_schema=records_schema,
         )
-        yield from self._read_pages(record_generator, _slice)
+        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,
+            )
+        yield from records
 
     def _parse_records(
         self,
diff --git a/unit_tests/sources/declarative/parsers/test_model_to_component_factory.py b/unit_tests/sources/declarative/parsers/test_model_to_component_factory.py
index eb1d0492c1..95cb630bd4 100644
--- a/unit_tests/sources/declarative/parsers/test_model_to_component_factory.py
+++ b/unit_tests/sources/declarative/parsers/test_model_to_component_factory.py
@@ -177,7 +177,11 @@
 )
 from airbyte_cdk.sources.declarative.requesters.request_path import RequestPath
 from airbyte_cdk.sources.declarative.requesters.requester import HttpMethod
-from airbyte_cdk.sources.declarative.retrievers import AsyncRetriever, SimpleRetriever
+from airbyte_cdk.sources.declarative.retrievers import (
+    AsyncRetriever,
+    LazySimpleRetriever,
+    SimpleRetriever,
+)
 from airbyte_cdk.sources.declarative.schema import InlineSchemaLoader, JsonFileSchemaLoader
 from airbyte_cdk.sources.declarative.schema.caching_schema_loader_decorator import (
     CachingSchemaLoaderDecorator,
@@ -1437,10 +1441,303 @@ def test_incremental_data_feed():
         model_type=DeclarativeStreamModel, component_definition=stream_manifest, config=input_config
     )
 
+    retriever = get_retriever(stream)
     assert isinstance(
-        get_retriever(stream).paginator.pagination_strategy,
+        retriever.paginator.pagination_strategy,
         StopConditionPaginationStrategyDecorator,
     )
+    # the stop condition only prevents the next page from being requested; the already-synced records
+    # of the last page are dropped by the retriever
+    assert isinstance(retriever.post_pagination_filter, ClientSideIncrementalRecordFilterDecorator)
+    assert retriever.post_pagination_filter._cursor is stream.cursor
+
+
+@pytest.mark.parametrize(
+    "is_client_side_incremental",
+    [
+        pytest.param(False, id="test_data_feed_only"),
+        pytest.param(True, id="test_data_feed_with_client_side_incremental"),
+    ],
+)
+def test_incremental_data_feed_filters_already_synced_records_in_the_retriever(
+    is_client_side_incremental,
+):
+    content = f"""
+selector:
+  type: RecordSelector
+  extractor:
+      type: DpathExtractor
+      field_path: ["extractor_path"]
+requester:
+  type: HttpRequester
+  name: "{{{{ parameters['name'] }}}}"
+  url_base: "https://api.sendgrid.com/v3/"
+  http_method: "GET"
+list_stream:
+  type: DeclarativeStream
+  incremental_sync:
+    type: DatetimeBasedCursor
+    $parameters:
+      datetime_format: "%Y-%m-%dT%H:%M:%S.%f%z"
+    start_datetime: "{{{{ config['start_time'] }}}}"
+    cursor_field: "created"
+    is_data_feed: true
+    is_client_side_incremental: {str(is_client_side_incremental).lower()}
+  retriever:
+    type: SimpleRetriever
+    name: "{{{{ parameters['name'] }}}}"
+    paginator:
+      type: DefaultPaginator
+      pagination_strategy:
+        type: "CursorPagination"
+        cursor_value: "{{{{ response._metadata.next }}}}"
+        page_size: 10
+    requester:
+      $ref: "#/requester"
+      path: "/"
+    record_selector:
+      $ref: "#/selector"
+  $parameters:
+    name: "lists"
+    """
+
+    parsed_manifest = YamlDeclarativeSource._parse(content)
+    resolved_manifest = resolver.preprocess_manifest(parsed_manifest)
+    stream_manifest = transformer.propagate_types_and_parameters(
+        "", resolved_manifest["list_stream"], {}
+    )
+
+    stream = factory.create_component(
+        model_type=DeclarativeStreamModel, component_definition=stream_manifest, config=input_config
+    )
+
+    retriever = get_retriever(stream)
+    assert isinstance(retriever.post_pagination_filter, ClientSideIncrementalRecordFilterDecorator)
+    assert retriever.post_pagination_filter._cursor is stream.cursor
+    # the `record_filter` condition stays in the record selector, so the post-pagination filter must not evaluate it
+    assert retriever.post_pagination_filter.condition is None
+    # filtering in the record selector would hide the already-synced records from the paginator and
+    # therefore silently disable the stop condition
+    assert not isinstance(
+        retriever.record_selector.record_filter, ClientSideIncrementalRecordFilterDecorator
+    )
+
+
+def test_given_data_feed_and_record_filter_then_condition_stays_in_the_record_selector():
+    content = """
+selector:
+  type: RecordSelector
+  record_filter:
+    type: RecordFilter
+    condition: "{{ record['id'] > 1 }}"
+  extractor:
+      type: DpathExtractor
+      field_path: ["extractor_path"]
+requester:
+  type: HttpRequester
+  name: "{{ parameters['name'] }}"
+  url_base: "https://api.sendgrid.com/v3/"
+  http_method: "GET"
+list_stream:
+  type: DeclarativeStream
+  incremental_sync:
+    type: DatetimeBasedCursor
+    $parameters:
+      datetime_format: "%Y-%m-%dT%H:%M:%S.%f%z"
+    start_datetime: "{{ config['start_time'] }}"
+    cursor_field: "created"
+    is_data_feed: true
+  retriever:
+    type: SimpleRetriever
+    name: "{{ parameters['name'] }}"
+    paginator:
+      type: DefaultPaginator
+      pagination_strategy:
+        type: "CursorPagination"
+        cursor_value: "{{ response._metadata.next }}"
+        page_size: 10
+    requester:
+      $ref: "#/requester"
+      path: "/"
+    record_selector:
+      $ref: "#/selector"
+  $parameters:
+    name: "lists"
+    """
+
+    parsed_manifest = YamlDeclarativeSource._parse(content)
+    resolved_manifest = resolver.preprocess_manifest(parsed_manifest)
+    stream_manifest = transformer.propagate_types_and_parameters(
+        "", resolved_manifest["list_stream"], {}
+    )
+
+    stream = factory.create_component(
+        model_type=DeclarativeStreamModel, component_definition=stream_manifest, config=input_config
+    )
+
+    retriever = get_retriever(stream)
+    # the condition must keep running upstream of the paginator, otherwise the records it rejects would start counting
+    # towards the page size and could become the record the stop condition is evaluated on
+    assert retriever.record_selector.record_filter.condition == "{{ record['id'] > 1 }}"
+    assert retriever.post_pagination_filter.condition is None
+    # a data feed that is not client-side incremental keeps the `transform_before_filtering` default it had before the
+    # cursor filtering moved to the retriever
+    assert retriever.record_selector.transform_before_filtering is False
+
+
+def test_given_data_feed_and_client_side_incremental_then_transform_before_filtering():
+    """
+    Moving the cursor filtering to the retriever must not change when the `record_filter` condition runs: a
+    client-side incremental stream evaluates it after the transformations, otherwise a condition reading a
+    transformation-produced field silently rejects every record.
+    """
+    content = """
+selector:
+  type: RecordSelector
+  record_filter:
+    type: RecordFilter
+    condition: "{{ record['keep'] == 'yes' }}"
+  extractor:
+      type: DpathExtractor
+      field_path: ["extractor_path"]
+requester:
+  type: HttpRequester
+  name: "{{ parameters['name'] }}"
+  url_base: "https://api.sendgrid.com/v3/"
+  http_method: "GET"
+list_stream:
+  type: DeclarativeStream
+  transformations:
+    - type: AddFields
+      fields:
+        - path: ["keep"]
+          value: "yes"
+  incremental_sync:
+    type: DatetimeBasedCursor
+    $parameters:
+      datetime_format: "%Y-%m-%dT%H:%M:%S.%f%z"
+    start_datetime: "{{ config['start_time'] }}"
+    cursor_field: "created"
+    is_data_feed: true
+    is_client_side_incremental: true
+  retriever:
+    type: SimpleRetriever
+    name: "{{ parameters['name'] }}"
+    paginator:
+      type: DefaultPaginator
+      pagination_strategy:
+        type: "CursorPagination"
+        cursor_value: "{{ response._metadata.next }}"
+        page_size: 10
+    requester:
+      $ref: "#/requester"
+      path: "/"
+    record_selector:
+      $ref: "#/selector"
+  $parameters:
+    name: "lists"
+    """
+
+    parsed_manifest = YamlDeclarativeSource._parse(content)
+    resolved_manifest = resolver.preprocess_manifest(parsed_manifest)
+    stream_manifest = transformer.propagate_types_and_parameters(
+        "", resolved_manifest["list_stream"], {}
+    )
+
+    stream = factory.create_component(
+        model_type=DeclarativeStreamModel, component_definition=stream_manifest, config=input_config
+    )
+
+    retriever = get_retriever(stream)
+    assert retriever.record_selector.transform_before_filtering is True
+
+
+def test_given_data_feed_and_lazy_read_then_lazy_retriever_filters_already_synced_records():
+    """
+    `LazySimpleRetriever` inherits `read_records`, so it only drops the already-synced records of the boundary page
+    if the factory passes it the filter too.
+    """
+    stream_definition = {
+        "type": "DeclarativeStream",
+        "name": "items",
+        "primary_key": [],
+        "schema_loader": {
+            "type": "InlineSchemaLoader",
+            "schema": {"type": "object", "properties": {}},
+        },
+        "incremental_sync": {
+            "type": "DatetimeBasedCursor",
+            "datetime_format": "%Y-%m-%dT%H:%M:%S.%f%z",
+            "start_datetime": "{{ config['start_time'] }}",
+            "cursor_field": "created",
+            "is_data_feed": True,
+        },
+        "retriever": {
+            "type": "SimpleRetriever",
+            "requester": {
+                "type": "HttpRequester",
+                "url_base": "https://api.test.com",
+                "path": "parent/{{ stream_partition.parent_id }}/items",
+                "http_method": "GET",
+            },
+            "record_selector": {
+                "type": "RecordSelector",
+                "extractor": {"type": "DpathExtractor", "field_path": ["data"]},
+            },
+            "paginator": {
+                "type": "DefaultPaginator",
+                "pagination_strategy": {
+                    "type": "CursorPagination",
+                    "cursor_value": '{{ response["data"][-1]["id"] }}',
+                },
+            },
+            "partition_router": {
+                "type": "SubstreamPartitionRouter",
+                "parent_stream_configs": [
+                    {
+                        "type": "ParentStreamConfig",
+                        "parent_key": "id",
+                        "partition_field": "parent_id",
+                        "lazy_read_pointer": ["items"],
+                        "stream": {
+                            "type": "DeclarativeStream",
+                            "name": "parent",
+                            "schema_loader": {
+                                "type": "InlineSchemaLoader",
+                                "schema": {"type": "object", "properties": {}},
+                            },
+                            "retriever": {
+                                "type": "SimpleRetriever",
+                                "requester": {
+                                    "type": "HttpRequester",
+                                    "url_base": "https://api.test.com",
+                                    "path": "/parents",
+                                    "http_method": "GET",
+                                },
+                                "record_selector": {
+                                    "type": "RecordSelector",
+                                    "extractor": {
+                                        "type": "DpathExtractor",
+                                        "field_path": ["data"],
+                                    },
+                                },
+                            },
+                        },
+                    }
+                ],
+            },
+        },
+    }
+
+    stream = factory.create_component(
+        model_type=DeclarativeStreamModel,
+        component_definition=stream_definition,
+        config=input_config,
+    )
+
+    retriever = get_retriever(stream)
+    assert isinstance(retriever, LazySimpleRetriever)
+    assert isinstance(retriever.post_pagination_filter, ClientSideIncrementalRecordFilterDecorator)
 
 
 def test_given_data_feed_and_incremental_then_raise_error():
diff --git a/unit_tests/sources/declarative/retrievers/test_data_feed_integration.py b/unit_tests/sources/declarative/retrievers/test_data_feed_integration.py
new file mode 100644
index 0000000000..d4b71b3ee1
--- /dev/null
+++ b/unit_tests/sources/declarative/retrievers/test_data_feed_integration.py
@@ -0,0 +1,274 @@
+#
+# Copyright (c) 2026 Airbyte, Inc., all rights reserved.
+#
+
+"""
+End-to-end coverage for data feed streams (`is_data_feed: true`).
+
+A data feed returns records in descending cursor order, so the sync must stop paginating on the
+first page that contains a record the cursor considers already synced, and must not emit the
+already-synced records sitting at the tail of that page.
+"""
+
+import json
+import logging
+from typing import Any, List, Mapping, Optional, Tuple
+
+import pytest
+import requests_mock
+
+from airbyte_cdk.models import (
+    AirbyteStateBlob,
+    AirbyteStateMessage,
+    AirbyteStateType,
+    AirbyteStreamState,
+    ConfiguredAirbyteCatalogSerializer,
+    StreamDescriptor,
+    Type,
+)
+from airbyte_cdk.sources.declarative.concurrent_declarative_source import (
+    ConcurrentDeclarativeSource,
+)
+
+_STREAM_NAME = "items"
+
+
+def _manifest(
+    is_client_side_incremental: bool = False, partition_router: Optional[Mapping[str, Any]] = None
+) -> Mapping[str, Any]:
+    incremental_sync: dict[str, Any] = {
+        "type": "DatetimeBasedCursor",
+        "cursor_field": "updated_at",
+        "datetime_format": "%Y-%m-%dT%H:%M:%SZ",
+        "start_datetime": {
+            "type": "MinMaxDatetime",
+            "datetime": "2019-01-01T00:00:00Z",
+            "datetime_format": "%Y-%m-%dT%H:%M:%SZ",
+        },
+        "is_data_feed": True,
+    }
+    if is_client_side_incremental:
+        incremental_sync["is_client_side_incremental"] = True
+
+    retriever: dict[str, Any] = {
+        "type": "SimpleRetriever",
+        "requester": {
+            "type": "HttpRequester",
+            "url_base": "https://api.example.com",
+            "path": "/items" if partition_router is None else "/{{ stream_partition.owner }}/items",
+            "http_method": "GET",
+        },
+        "record_selector": {
+            "type": "RecordSelector",
+            "extractor": {"type": "DpathExtractor", "field_path": []},
+        },
+        "paginator": {
+            "type": "DefaultPaginator",
+            "pagination_strategy": {
+                "type": "PageIncrement",
+                "page_size": 4,
+                "start_from_page": 1,
+                "inject_on_first_request": True,
+            },
+            "page_token_option": {
+                "type": "RequestOption",
+                "inject_into": "request_parameter",
+                "field_name": "page",
+            },
+        },
+    }
+    if partition_router:
+        retriever["partition_router"] = partition_router
+
+    return {
+        "version": "7.0.0",
+        "type": "DeclarativeSource",
+        "check": {"type": "CheckStream", "stream_names": [_STREAM_NAME]},
+        "spec": {
+            "type": "Spec",
+            "connection_specification": {"type": "object", "properties": {}},
+        },
+        "streams": [
+            {
+                "type": "DeclarativeStream",
+                "name": _STREAM_NAME,
+                "primary_key": ["id"],
+                "schema_loader": {
+                    "type": "InlineSchemaLoader",
+                    "schema": {
+                        "type": "object",
+                        "properties": {
+                            "id": {"type": "string"},
+                            "updated_at": {"type": "string"},
+                        },
+                    },
+                },
+                "retriever": retriever,
+                "incremental_sync": incremental_sync,
+            }
+        ],
+    }
+
+
+# Records are sorted in descending order of updated_at, as expected from a data feed. With a cursor
+# of 2021-01-01, page 1 holds two fresh records followed by two already-synced ones.
+_PAGE_1 = [
+    {"id": "4", "updated_at": "2022-06-01T00:00:00Z"},
+    {"id": "3", "updated_at": "2022-05-01T00:00:00Z"},
+    {"id": "2", "updated_at": "2020-06-01T00:00:00Z"},
+    {"id": "1", "updated_at": "2020-05-01T00:00:00Z"},
+]
+_PAGE_2 = [
+    {"id": "0", "updated_at": "2020-04-01T00:00:00Z"},
+]
+
+_CATALOG = ConfiguredAirbyteCatalogSerializer.load(
+    {
+        "streams": [
+            {
+                "stream": {
+                    "name": _STREAM_NAME,
+                    "json_schema": {},
+                    "supported_sync_modes": ["full_refresh", "incremental"],
+                },
+                "sync_mode": "incremental",
+                "destination_sync_mode": "append",
+            }
+        ]
+    }
+)
+
+
+def _state(stream_state: Mapping[str, Any]) -> List[AirbyteStateMessage]:
+    return [
+        AirbyteStateMessage(
+            type=AirbyteStateType.STREAM,
+            stream=AirbyteStreamState(
+                stream_descriptor=StreamDescriptor(name=_STREAM_NAME),
+                stream_state=AirbyteStateBlob(stream_state),
+            ),
+        )
+    ]
+
+
+def _read(
+    manifest: Mapping[str, Any],
+    state: Optional[List[AirbyteStateMessage]],
+    pages_per_partition: Optional[Mapping[Tuple[str, str], List[Mapping[str, Any]]]] = None,
+    pages: Optional[Mapping[str, List[Mapping[str, Any]]]] = None,
+) -> Tuple[List[str], List[str]]:
+    pages_fetched = []
+
+    def paged_response(request: Any, context: Any) -> str:
+        page = request.qs.get("page", ["1"])[0]
+        if pages_per_partition is None:
+            pages_fetched.append(page)
+            if pages is not None:
+                return json.dumps(pages.get(page, []))
+            return json.dumps(_PAGE_1 if page == "1" else _PAGE_2)
+        owner = request.path.strip("/").split("/")[0]
+        pages_fetched.append(f"{owner}:{page}")
+        return json.dumps(pages_per_partition.get((owner, page), []))
+
+    source = ConcurrentDeclarativeSource(
+        source_config=manifest, config={}, catalog=_CATALOG, state=state
+    )
+    with requests_mock.Mocker() as http_mocker:
+        http_mocker.get(requests_mock.ANY, text=paged_response)
+        records = [
+            message.record.data
+            for message in source.read(logging.getLogger("test"), {}, _CATALOG, state)
+            if message.type == Type.RECORD
+        ]
+    return sorted(pages_fetched), sorted(record["id"] for record in records)
+
+
+@pytest.mark.parametrize("is_client_side_incremental", [False, True])
+def test_given_already_synced_records_on_page_then_stop_paginating_and_filter_them_out(
+    is_client_side_incremental: bool,
+) -> None:
+    pages_fetched, record_ids = _read(
+        _manifest(is_client_side_incremental=is_client_side_incremental),
+        _state({"updated_at": "2021-01-01T00:00:00Z"}),
+    )
+
+    assert pages_fetched == ["1"]
+    assert record_ids == ["3", "4"]
+
+
+@pytest.mark.parametrize("is_client_side_incremental", [False, True])
+def test_given_no_already_synced_records_then_paginate_until_the_end(
+    is_client_side_incremental: bool,
+) -> None:
+    pages_fetched, record_ids = _read(
+        _manifest(is_client_side_incremental=is_client_side_incremental), None
+    )
+
+    assert pages_fetched == ["1", "2"]
+    assert record_ids == ["0", "1", "2", "3", "4"]
+
+
+def test_given_record_dated_in_the_future_then_filter_it_out() -> None:
+    """
+    The retriever drops the records the cursor would not sync, and `should_be_synced` is bounded on both ends: with no
+    `end_datetime` the upper bound is `now()`, so records dated ahead of the connector's clock are dropped too. This is
+    the behaviour `is_client_side_incremental` has always had, and a data feed now matches it.
+    """
+    pages_fetched, record_ids = _read(
+        _manifest(),
+        _state({"updated_at": "2021-01-01T00:00:00Z"}),
+        pages={
+            # a full page, so that pagination is only stopped by the already-synced record on page 2 and not by a
+            # short page. Page 1 holds nothing already synced, so reaching page 2 proves the forward-dated record did
+            # not stop the pagination.
+            "1": [
+                {"id": "future", "updated_at": "2099-01-01T00:00:00Z"},
+                {"id": "fresh_1", "updated_at": "2022-06-01T00:00:00Z"},
+                {"id": "fresh_2", "updated_at": "2022-05-01T00:00:00Z"},
+                {"id": "fresh_3", "updated_at": "2022-04-01T00:00:00Z"},
+            ],
+            "2": [
+                {"id": "already_synced", "updated_at": "2020-06-01T00:00:00Z"},
+            ],
+        },
+    )
+
+    # the forward-dated record does not stop the pagination, it is only left out of the emitted records
+    assert pages_fetched == ["1", "2"]
+    assert record_ids == ["fresh_1", "fresh_2", "fresh_3"]
+
+
+def test_given_multiple_partitions_then_each_partition_stops_on_its_own_cursor() -> None:
+    """
+    A single retriever instance is shared by every partition and partitions are read concurrently,
+    so the boundary of one partition must not influence another.
+    """
+    pages_per_partition = {
+        ("a", "1"): [{**record, "id": f"a{record['id']}"} for record in _PAGE_1],
+        ("a", "2"): [{**record, "id": f"a{record['id']}"} for record in _PAGE_2],
+        ("b", "1"): [{**record, "id": f"b{record['id']}"} for record in _PAGE_1],
+        ("b", "2"): [{**record, "id": f"b{record['id']}"} for record in _PAGE_2],
+    }
+    manifest = _manifest(
+        partition_router={
+            "type": "ListPartitionRouter",
+            "values": ["a", "b"],
+            "cursor_field": "owner",
+        }
+    )
+    state = _state(
+        {
+            "use_global_cursor": False,
+            "states": [
+                # partition "a" has already synced everything before 2021 hence it stops on page 1
+                {"partition": {"owner": "a"}, "cursor": {"updated_at": "2021-01-01T00:00:00Z"}},
+                # partition "b" has nothing already synced hence it reads both pages
+                {"partition": {"owner": "b"}, "cursor": {"updated_at": "2019-01-01T00:00:00Z"}},
+            ],
+        }
+    )
+
+    pages_fetched, record_ids = _read(manifest, state, pages_per_partition)
+
+    assert pages_fetched == ["a:1", "b:1", "b:2"]
+    assert record_ids == ["a3", "a4", "b0", "b1", "b2", "b3", "b4"]
diff --git a/unit_tests/sources/declarative/retrievers/test_simple_retriever.py b/unit_tests/sources/declarative/retrievers/test_simple_retriever.py
index d9585dbd44..e05c427acb 100644
--- a/unit_tests/sources/declarative/retrievers/test_simple_retriever.py
+++ b/unit_tests/sources/declarative/retrievers/test_simple_retriever.py
@@ -20,6 +20,9 @@
 from airbyte_cdk.sources.declarative.auth.declarative_authenticator import NoAuth
 from airbyte_cdk.sources.declarative.decoders import JsonDecoder
 from airbyte_cdk.sources.declarative.extractors import DpathExtractor, HttpSelector, RecordSelector
+from airbyte_cdk.sources.declarative.extractors.record_filter import (
+    ClientSideIncrementalRecordFilterDecorator,
+)
 from airbyte_cdk.sources.declarative.partition_routers import SinglePartitionRouter
 from airbyte_cdk.sources.declarative.requesters.paginators import DefaultPaginator, Paginator
 from airbyte_cdk.sources.declarative.requesters.paginators.strategies import (
@@ -1430,3 +1433,87 @@ def _mock_paginator():
     paginator.get_request_body_data.__name__ = "get_request_body_data"
     paginator.get_request_body_json.__name__ = "get_request_body_json"
     return paginator
+
+
+def _data_feed_retriever(cursor: Optional[Mock], paginator: Paginator) -> SimpleRetriever:
+    requester = MagicMock()
+    requester.send_request.return_value = MagicMock()
+    record_selector = MagicMock()
+    return SimpleRetriever(
+        name=A_STREAM_NAME,
+        primary_key=primary_key,
+        requester=requester,
+        paginator=paginator,
+        record_selector=record_selector,
+        stream_slicer=SinglePartitionRouter(parameters={}),
+        post_pagination_filter=ClientSideIncrementalRecordFilterDecorator(
+            config={}, parameters={}, condition=None, cursor=cursor
+        )
+        if cursor
+        else None,
+        parameters={},
+        config={},
+    )
+
+
+def test_given_data_feed_when_read_records_then_filter_out_already_synced_records():
+    page = [
+        Record(data={"id": "1"}, stream_name=A_STREAM_NAME),
+        Record(data={"id": "2"}, stream_name=A_STREAM_NAME),
+        Record(data={"id": "3"}, stream_name=A_STREAM_NAME),
+    ]
+    cursor = Mock()
+    cursor.should_be_synced.side_effect = lambda record: record.data["id"] != "3"
+    paginator = _mock_paginator()
+    paginator.get_initial_token.return_value = None
+    paginator.next_page_token.return_value = None
+    retriever = _data_feed_retriever(cursor, paginator)
+
+    with patch.object(SimpleRetriever, "_parse_records", return_value=iter(page)):
+        actual_records = list(
+            retriever.read_records(records_schema={}, stream_slice=A_STREAM_SLICE)
+        )
+
+    assert actual_records == page[:2]
+
+
+def test_given_data_feed_when_read_records_then_paginator_still_sees_the_whole_page():
+    """
+    The record that stops the pagination is the very one being filtered out, so the paginator must
+    be given the page as returned by the API rather than the filtered one.
+    """
+    page = [
+        Record(data={"id": "1"}, stream_name=A_STREAM_NAME),
+        Record(data={"id": "2"}, stream_name=A_STREAM_NAME),
+        Record(data={"id": "3"}, stream_name=A_STREAM_NAME),
+    ]
+    cursor = Mock()
+    cursor.should_be_synced.side_effect = lambda record: record.data["id"] != "3"
+    paginator = _mock_paginator()
+    paginator.get_initial_token.return_value = None
+    paginator.next_page_token.return_value = None
+    retriever = _data_feed_retriever(cursor, paginator)
+
+    with patch.object(SimpleRetriever, "_parse_records", return_value=iter(page)):
+        list(retriever.read_records(records_schema={}, stream_slice=A_STREAM_SLICE))
+
+    assert paginator.next_page_token.call_args.kwargs["last_page_size"] == 3
+    assert paginator.next_page_token.call_args.kwargs["last_record"] == page[-1]
+
+
+def test_given_no_data_feed_when_read_records_then_emit_every_record():
+    page = [
+        Record(data={"id": "1"}, stream_name=A_STREAM_NAME),
+        Record(data={"id": "2"}, stream_name=A_STREAM_NAME),
+    ]
+    paginator = _mock_paginator()
+    paginator.get_initial_token.return_value = None
+    paginator.next_page_token.return_value = None
+    retriever = _data_feed_retriever(cursor=None, paginator=paginator)
+
+    with patch.object(SimpleRetriever, "_parse_records", return_value=iter(page)):
+        actual_records = list(
+            retriever.read_records(records_schema={}, stream_slice=A_STREAM_SLICE)
+        )
+
+    assert actual_records == page