chore(do not merge): CDK #1079 on current main for OC-13084 prerelease SDM

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -634,6 +634,7 @@
     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,
@@ -4138,6 +4139,14 @@ def create_parent_stream_config_with_substream_wrapper(
         self, model: ParentStreamConfigModel, config: Config, *, stream_name: str, **kwargs: Any
     ) -> Any:
         child_state = self._connector_state_manager.get_stream_state(stream_name, None)
+        if NO_CURSOR_STATE_KEY in child_state:
+            # Full refresh streams checkpoint a `{NO_CURSOR_STATE_KEY: true}` sentinel. When such a
+            # stream is later converted to incremental with an incremental_dependency parent,
+            # `_instantiate_parent_stream_state_manager` would treat the sentinel's boolean as a legacy
+            # cursor value and re-key it under the parent's cursor field, crashing cursor initialization.
+            child_state = {
+                key: value for key, value in child_state.items() if key != NO_CURSOR_STATE_KEY
+            }
 
         parent_state: Optional[Mapping[str, Any]] = (
             child_state if model.incremental_dependency and child_state else None

unit_tests/sources/declarative/parsers/test_model_to_component_factory.py

@@ -4020,6 +4020,108 @@ def test_create_concurrent_cursor_from_perpartition_cursor_runs_state_migrations
     )
 
 
+def test_create_concurrent_cursor_from_perpartition_cursor_ignores_full_refresh_sentinel_state():
+    """
+    A stream that synced as full refresh checkpoints `{"__ab_no_cursor_state_message": true}`. If a later
+    connector version converts that stream to incremental with an `incremental_dependency` parent, the
+    sentinel must not be re-keyed under the parent's cursor field as if it were a legacy cursor value —
+    doing so crashed cursor initialization with `ValueError: No format in [...] matching True`.
+    """
+    content = """
+    type: DeclarativeStream
+    primary_key: "id"
+    name: test
+    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
+      name: test
+      requester:
+        type: HttpRequester
+        name: "test"
+        url_base: "https://api.test.com/v3/"
+        http_method: "GET"
+        authenticator:
+          type: NoAuth
+      record_selector:
+        type: RecordSelector
+        extractor:
+          type: DpathExtractor
+          field_path: []
+      partition_router:
+        type: SubstreamPartitionRouter
+        parent_stream_configs:
+          - type: ParentStreamConfig
+            parent_key: id
+            partition_field: id
+            incremental_dependency: true
+            stream:
+              type: DeclarativeStream
+              primary_key: id
+              name: parent_stream
+              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: []
+      """
+
+    connector_state_manager = ConnectorStateManager(
+        state=[
+            AirbyteStateMessage(
+                type=AirbyteStateType.STREAM,
+                stream=AirbyteStreamState(
+                    stream_descriptor=StreamDescriptor(name="test"),
+                    stream_state=AirbyteStateBlob({"__ab_no_cursor_state_message": True}),
+                ),
+            )
+        ]
+    )
+    factory = ModelToComponentFactory(
+        emit_connector_builder_messages=True, connector_state_manager=connector_state_manager
+    )
+    stream = factory.create_component(
+        model_type=DeclarativeStreamModel,
+        component_definition=YamlDeclarativeSource._parse(content),
+        config=input_config,
+    )
+
+    parent_cursor_state = stream.cursor._partition_router.parent_stream_configs[
+        0
+    ].stream.cursor.state
+    assert all(value is not True for value in parent_cursor_state.values())
+
+
 def test_incrementing_count_cursor_with_partition_router_raises_error():
     content = """
     type: DeclarativeStream