fix(concurrent): use last slice for client-side incremental filtering

The _get_concurrent_state() method was using slices[0] (the earliest slice)
instead of slices[-1] (the most recent slice) when extracting the cursor
value from partitioned state. Since merge_intervals() sorts slices in
ascending order by (START_KEY, END_KEY), using the first slice caused
should_be_synced() to use the earliest timestamp as the lower bound,
admitting ALL records instead of only new records.

This bug affected all low-code connectors using DatetimeBasedCursor with
is_client_side_incremental: true, including Notion's pages, databases,
comments, and blocks streams.

Fixes: airbytehq/oncall#8854
Co-Authored-By: unknown <>

Author: devin-ai-integration[bot]

airbyte_cdk/sources/streams/concurrent/cursor.py

@@ -233,12 +233,14 @@ def _get_concurrent_state(
 
             value_from_partitioned_state = None
             if slices_from_partitioned_state:
-                # We assume here that the slices have been already merged
-                first_slice = slices_from_partitioned_state[0]
+                # We assume here that the slices have been already merged.
+                # After merging, slices are sorted in ascending order by (START_KEY, END_KEY),
+                # so the last slice contains the most recent cursor value for client-side filtering.
+                last_slice = slices_from_partitioned_state[-1]
                 value_from_partitioned_state = (
-                    first_slice[self._connector_state_converter.MOST_RECENT_RECORD_KEY]
-                    if self._connector_state_converter.MOST_RECENT_RECORD_KEY in first_slice
-                    else first_slice[self._connector_state_converter.END_KEY]
+                    last_slice[self._connector_state_converter.MOST_RECENT_RECORD_KEY]
+                    if self._connector_state_converter.MOST_RECENT_RECORD_KEY in last_slice
+                    else last_slice[self._connector_state_converter.END_KEY]
                 )
             return (
                 value_from_partitioned_state