fix: Raise ValueError for unparseable cursor datetime when api_retention_period is set

Co-Authored-By: alfredo.garcia@airbyte.io <freddy.garcia7.fg@gmail.com>

Author: devin-ai-integration[bot]

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -3609,9 +3609,9 @@ def _is_cursor_older_than_retention_period(
         Delegates cursor datetime extraction to cursor class instances via
         get_cursor_datetime_from_state, which handles format-specific parsing.
 
-        Returns True if the cursor is older than the retention period or if the cursor is
-        invalid/unparseable (should use full refresh).
+        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).
+        Raises ValueError if the cursor datetime could not be parsed from state.
         """
         # Skip retention check for concurrent state format (e.g. {"state_type": "date-range", "slices": [...]}).
         # The DatetimeBasedCursor used for the age check only handles sequential state format.
@@ -3640,7 +3640,11 @@ def _is_cursor_older_than_retention_period(
                     break
 
         if cursor_datetime is None:
-            return True
+            raise ValueError(
+                f"Stream '{stream_name}' has api_retention_period set to '{api_retention_period}' "
+                f"but the cursor datetime could not be parsed from state. Check that cursor_field "
+                f"and datetime_format match the state format."
+            )
 
         retention_duration = parse_duration(api_retention_period)
         retention_cutoff = datetime.datetime.now(datetime.timezone.utc) - retention_duration

unit_tests/sources/declarative/test_state_delegating_stream.py

@@ -8,6 +8,7 @@
 from unittest.mock import MagicMock
 
 import freezegun
+import pytest
 
 from airbyte_cdk.models import (
     AirbyteStateBlob,
@@ -553,3 +554,25 @@ def test_cursor_age_validation_skips_incrementing_count_cursor():
 
         records = get_records(source, _CONFIG, configured_catalog, state)
         assert len(records) >= 0
+
+
+def test_cursor_age_validation_raises_error_for_unparseable_cursor():
+    """Test that unparseable cursor datetime raises ValueError when api_retention_period is set."""
+    manifest = _create_manifest_with_retention_period("P7D")
+
+    state = [
+        AirbyteStateMessage(
+            type=AirbyteStateType.STREAM,
+            stream=AirbyteStreamState(
+                stream_descriptor=StreamDescriptor(name="TestStream", namespace=None),
+                stream_state=AirbyteStateBlob(updated_at="not-a-date"),
+            ),
+        )
+    ]
+
+    source = ConcurrentDeclarativeSource(
+        source_config=manifest, config=_CONFIG, catalog=None, state=state
+    )
+
+    with pytest.raises(ValueError, match="could not be parsed from state"):
+        source.discover(logger=MagicMock(), config=_CONFIG)