Ensure first parent uses cache

airbyte_cdk/sources/declarative/concurrent_declarative_source.py

@@ -201,7 +201,7 @@ def _group_streams(
         state_manager = ConnectorStateManager(state=self._state)  # type: ignore  # state is always in the form of List[AirbyteStateMessage]. The ConnectorStateManager should use generics, but this can be done later
 
         name_to_stream_mapping = {
-            stream["name"]: stream for stream in self.resolved_manifest["streams"]
+            stream["name"]: stream for stream in self._initialize_cache_for_parent_streams(self.resolved_manifest["streams"])
         }
 
         for declarative_stream in self.streams(config=config):