fix: do not crash on unparseable cursor state value during concurrent cursor construction

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -666,6 +666,7 @@
 from airbyte_cdk.sources.streams.concurrent.state_converters.incrementing_count_stream_state_converter import (
     IncrementingCountStreamStateConverter,
 )
+from airbyte_cdk.sources.streams.core import NO_CURSOR_STATE_KEY
 from airbyte_cdk.sources.streams.http.error_handlers.response_models import ResponseAction
 from airbyte_cdk.sources.types import Config
 from airbyte_cdk.sources.utils.transform import TransformConfig, TypeTransformer
@@ -4164,6 +4165,14 @@ def _instantiate_parent_stream_state_manager(
                 )
 
                 if not extracted_parent_state and not isinstance(extracted_parent_state, dict):
+                    # The no-cursor sentinel (`{NO_CURSOR_STATE_KEY: True}`) is written by
+                    # streams that completed without a cursor value (e.g. HubSpot's
+                    # `associations_*` streams). It is bookkeeping, not a cursor value, so it
+                    # must not be re-keyed as `{cursor_field: True}` — doing so feeds a boolean
+                    # into the datetime cursor parser and crashes the whole source at startup
+                    # (see airbytehq/oncall#13084). Start the parent cleanly instead.
+                    if set(child_state.keys()) == {NO_CURSOR_STATE_KEY}:
+                        return ConnectorStateManager([])
                     cursor_values = child_state.values()
                     if cursor_values and len(cursor_values) == 1:
                         incremental_sync_model: Union[

airbyte_cdk/sources/streams/concurrent/state_converters/datetime_stream_state_converter.py

@@ -2,6 +2,7 @@
 # Copyright (c) 2023 Airbyte, Inc., all rights reserved.
 #
 
+import logging
 from abc import abstractmethod
 from datetime import datetime, timedelta, timezone
 from typing import Any, Callable, List, MutableMapping, Optional, Tuple
@@ -16,6 +17,8 @@
 )
 from airbyte_cdk.utils.datetime_helpers import AirbyteDateTime, ab_datetime_now, ab_datetime_parse
 
+logger = logging.getLogger("airbyte")
+
 
 class DateTimeStreamStateConverter(AbstractStreamStateConverter):
     def _from_state_message(self, value: Any) -> Any:
@@ -100,11 +103,23 @@ def _get_sync_start(
         start: Optional[datetime],
     ) -> datetime:
         sync_start = start if start is not None else self.zero_value
-        prev_sync_low_water_mark = (
-            self.parse_timestamp(stream_state[cursor_field.cursor_field_key])
-            if cursor_field.cursor_field_key in stream_state
-            else None
-        )
+        prev_sync_low_water_mark = None
+        if cursor_field.cursor_field_key in stream_state:
+            saved_cursor_value = stream_state[cursor_field.cursor_field_key]
+            # Defense-in-depth guard. The primary/root fix for the boolean leaking into
+            # here lives in `ModelToComponentFactory._instantiate_parent_stream_state_manager`,
+            # which no longer re-keys the `__ab_no_cursor_state_message` sentinel as a cursor
+            # value. This guard is kept as a backstop: a boolean is never a valid datetime
+            # cursor value, so ignore it and fall back to the start date instead of crashing
+            # the whole source at startup (see airbytehq/oncall#13084).
+            if isinstance(saved_cursor_value, bool):
+                logger.warning(
+                    "Ignoring saved state for cursor field '%s': value %r is not a valid datetime. Falling back to the start date.",
+                    cursor_field.cursor_field_key,
+                    saved_cursor_value,
+                )
+            else:
+                prev_sync_low_water_mark = self.parse_timestamp(saved_cursor_value)
         if prev_sync_low_water_mark and prev_sync_low_water_mark >= sync_start:
             return prev_sync_low_water_mark
         else:

unit_tests/sources/declarative/parsers/test_model_to_component_factory.py

@@ -82,6 +82,7 @@
 from airbyte_cdk.sources.declarative.models import JwtAuthenticator as JwtAuthenticatorModel
 from airbyte_cdk.sources.declarative.models import ListPartitionRouter as ListPartitionRouterModel
 from airbyte_cdk.sources.declarative.models import OAuthAuthenticator as OAuthAuthenticatorModel
+from airbyte_cdk.sources.declarative.models import ParentStreamConfig as ParentStreamConfigModel
 from airbyte_cdk.sources.declarative.models import PropertyChunking as PropertyChunkingModel
 from airbyte_cdk.sources.declarative.models import RecordSelector as RecordSelectorModel
 from airbyte_cdk.sources.declarative.models import SimpleRetriever as SimpleRetrieverModel
@@ -3923,6 +3924,68 @@ def test_create_concurrent_cursor_from_perpartition_cursor_runs_state_migrations
     )
 
 
+def test_instantiate_parent_stream_state_manager_ignores_no_cursor_sentinel():
+    """Regression test for airbytehq/oncall#13084.
+
+    A child stream that completed without a cursor value saves the sentinel state
+    `{"__ab_no_cursor_state_message": True}`. When the parent has
+    `incremental_dependency: true`, the factory used to re-key that boolean sentinel as
+    `{cursor_field: True}`, which fed a boolean into the datetime cursor parser and crashed
+    the whole source at startup with `ValueError: No format ... matching True`. The parent
+    state must instead be empty so the parent starts cleanly from its configured start date.
+    """
+    parent_stream_config_model = ParentStreamConfigModel.parse_obj(
+        {
+            "type": "ParentStreamConfig",
+            "parent_key": "id",
+            "partition_field": "id",
+            "incremental_dependency": True,
+            "stream": {
+                "type": "DeclarativeStream",
+                "name": "parent_stream",
+                "primary_key": "id",
+                "schema_loader": {
+                    "type": "InlineSchemaLoader",
+                    "schema": {
+                        "$schema": "http://json-schema.org/draft-07/schema",
+                        "type": "object",
+                        "properties": {"id": {"type": "string"}},
+                    },
+                },
+                "incremental_sync": {
+                    "type": "DatetimeBasedCursor",
+                    "cursor_field": "updated_at",
+                    "datetime_format": "%Y-%m-%dT%H:%M:%S.%f%z",
+                    "start_datetime": "{{ config['start_time'] }}",
+                },
+                "retriever": {
+                    "type": "SimpleRetriever",
+                    "requester": {
+                        "type": "HttpRequester",
+                        "url_base": "https://api.test.com/v3/parent",
+                        "http_method": "GET",
+                    },
+                    "record_selector": {
+                        "type": "RecordSelector",
+                        "extractor": {"type": "DpathExtractor", "field_path": []},
+                    },
+                },
+            },
+        }
+    )
+
+    factory = ModelToComponentFactory()
+
+    # Previously raised `ValueError: No format ... matching True`.
+    state_manager = factory._instantiate_parent_stream_state_manager(
+        child_state={"__ab_no_cursor_state_message": True},
+        config=input_config,
+        model=parent_stream_config_model,
+    )
+
+    assert state_manager.get_stream_state("parent_stream", None) == {}
+
+
 def test_incrementing_count_cursor_with_partition_router_raises_error():
     content = """
     type: DeclarativeStream

unit_tests/sources/streams/concurrent/test_datetime_state_converter.py

@@ -165,12 +165,56 @@ def test_concurrent_stream_state_converter_is_state_message_compatible(
             datetime(2023, 8, 22, 5, 3, 27, tzinfo=timezone.utc),
             id="isomillis-converter-state-after-start-start-is-from-state",
         ),
+        pytest.param(
+            CustomFormatConcurrentStreamStateConverter(datetime_format="%Y-%m-%dT%H:%M:%S.%fZ"),
+            datetime(2022, 8, 22, 5, 3, 27, tzinfo=timezone.utc),
+            {"created_at": True},
+            datetime(2022, 8, 22, 5, 3, 27, tzinfo=timezone.utc),
+            id="customformat-converter-boolean-state-falls-back-to-start",
+        ),
+        pytest.param(
+            CustomFormatConcurrentStreamStateConverter(datetime_format="%Y-%m-%dT%H:%M:%S.%fZ"),
+            None,
+            {"created_at": True},
+            CustomFormatConcurrentStreamStateConverter(
+                datetime_format="%Y-%m-%dT%H:%M:%S.%fZ"
+            ).zero_value,
+            id="customformat-converter-boolean-state-no-start-falls-back-to-zero-value",
+        ),
     ],
 )
 def test_get_sync_start(converter, start, state, expected_start):
     assert converter._get_sync_start(CursorField("created_at"), state, start) == expected_start
 
 
+def test_get_sync_start_with_boolean_state_value_does_not_raise():
+    """Regression test for a boolean state bookkeeping marker reaching the cursor parser.
+
+    A malformed connection state (e.g. a boolean `True` from internal markers such as
+    `use_global_cursor` reaching the datetime cursor parser during cursor construction)
+    must not crash the whole source. The converter should ignore the bad value and fall
+    back to the configured start date.
+    """
+    converter = CustomFormatConcurrentStreamStateConverter(datetime_format="%Y-%m-%dT%H:%M:%S.%fZ")
+    start = datetime(2022, 8, 22, 5, 3, 27, tzinfo=timezone.utc)
+
+    sync_start = converter._get_sync_start(CursorField("updatedAt"), {"updatedAt": True}, start)
+
+    assert sync_start == start
+
+
+def test_convert_from_sequential_state_with_boolean_state_value_does_not_raise():
+    """Regression test: `convert_from_sequential_state` must not raise on a boolean cursor value."""
+    converter = CustomFormatConcurrentStreamStateConverter(datetime_format="%Y-%m-%dT%H:%M:%S.%fZ")
+    start = datetime(2022, 8, 22, 5, 3, 27, tzinfo=timezone.utc)
+
+    sync_start, _ = converter.convert_from_sequential_state(
+        CursorField("updatedAt"), {"updatedAt": True}, start
+    )
+
+    assert sync_start == start
+
+
 @pytest.mark.parametrize(
     "converter, start, sequential_state, expected_output_state",
     [