Revert "fix: Clear state BEFORE constructing full_refresh_stream in stale cursor fallback"

This reverts commit 39a2aee.

Author: agarctfi

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -618,7 +618,6 @@
     NoopMessageRepository,
 )
 from airbyte_cdk.sources.message.repository import StateFilteringMessageRepository
-from airbyte_cdk.sources.streams import NO_CURSOR_STATE_KEY
 from airbyte_cdk.sources.streams.call_rate import (
     APIBudget,
     FixedWindowCallRatePolicy,
@@ -3595,50 +3594,53 @@ def create_state_delegating_stream(
             or model.name in self._stream_name_to_configured_stream
         )
         if model.api_retention_period and stream_is_in_catalog:
+            full_refresh_stream: DefaultStream = self._create_component_from_model(
+                model.full_refresh_stream, config=config, **kwargs
+            )  # type: ignore[assignment]
             if self._is_cursor_older_than_retention_period(
                 stream_state,
+                full_refresh_stream.cursor,
                 incremental_stream.cursor,
                 model.api_retention_period,
                 model.name,
             ):
-                # Clear state BEFORE constructing the full_refresh_stream so that
-                # its cursor starts from start_date instead of the stale cursor.
                 self._connector_state_manager.update_state_for_stream(model.name, None, {})
                 state_message = self._connector_state_manager.create_state_message(model.name, None)
                 self._message_repository.emit_message(state_message)
-                return self._create_component_from_model(  # type: ignore[no-any-return]
-                    model.full_refresh_stream, config=config, **kwargs
-                )
+                return full_refresh_stream
 
         return incremental_stream
 
     @staticmethod
     def _is_cursor_older_than_retention_period(
         stream_state: Mapping[str, Any],
+        full_refresh_cursor: Cursor,
         incremental_cursor: Cursor,
         api_retention_period: str,
         stream_name: str,
     ) -> bool:
         """Check if the cursor value in the state is older than the API's retention period.
 
-        If the state contains NO_CURSOR_STATE_KEY, the previous sync was a completed
-        full refresh and the cursor is considered current — returns False.
-
-        Otherwise, uses the incremental cursor to parse the datetime from state.
+        Checks cursors in sequence: full refresh cursor first, then incremental cursor.
+        FinalStateCursor returns now() for completed full refresh state (NO_CURSOR_STATE_KEY),
+        which is always within retention, so we use incremental. For other states, it returns
+        None and we fall back to checking the incremental cursor.
 
         Returns True if the cursor is older than the retention period (should use full refresh).
         Returns False if the cursor is within the retention period (safe to use incremental).
         """
-        if stream_state.get(NO_CURSOR_STATE_KEY):
-            return False
-
         retention_duration = parse_duration(api_retention_period)
         retention_cutoff = datetime.datetime.now(datetime.timezone.utc) - retention_duration
 
-        cursor_datetime = incremental_cursor.get_cursor_datetime_from_state(stream_state)
+        # Check full refresh cursor first
+        cursor_datetime = full_refresh_cursor.get_cursor_datetime_from_state(stream_state)
+
+        # If full refresh cursor returns None, check incremental cursor
+        if cursor_datetime is None:
+            cursor_datetime = incremental_cursor.get_cursor_datetime_from_state(stream_state)
 
         if cursor_datetime is None:
-            # Cursor could not parse the state - fall back to full refresh to be safe
+            # Neither cursor could parse the state - fall back to full refresh to be safe
             return True
 
         if cursor_datetime < retention_cutoff:

unit_tests/sources/declarative/test_state_delegating_stream.py

@@ -613,20 +613,14 @@ def test_cursor_age_validation_raises_error_for_unparseable_cursor():
 
 
 @freezegun.freeze_time("2024-07-15")
-def test_no_cursor_state_key_uses_incremental_not_full_refresh():
-    """When state has NO_CURSOR_STATE_KEY, the previous sync was a completed full refresh.
-
-    The cursor is considered current so the stream should use incremental sync,
-    not fall back to full refresh again.
-    """
+def test_final_state_cursor_falls_back_to_full_refresh_when_state_unparseable():
+    """When state is a final state (NO_CURSOR_STATE_KEY), ConcurrentCursor cannot parse it,
+    so both cursors return None and the implementation falls back to full refresh as the safe default."""
     manifest = _create_manifest_with_retention_period("P7D")
 
     with HttpMocker() as http_mocker:
-        # Mock the incremental endpoint with query params (start_date=2024-07-01, step=P15D, frozen at 2024-07-15)
         http_mocker.get(
-            HttpRequest(
-                url="https://api.test.com/items_with_filtration?start=2024-07-01&end=2024-07-15"
-            ),
+            HttpRequest(url="https://api.test.com/items"),
             HttpResponse(
                 body=json.dumps(
                     [