diff --git a/awx/api/urls/workflow_job_node.py b/awx/api/urls/workflow_job_node.py
index e0942a4790..da029b34c2 100644
--- a/awx/api/urls/workflow_job_node.py
+++ b/awx/api/urls/workflow_job_node.py
@@ -23,7 +23,7 @@ urls = [
     re_path(r'^(?P<pk>[0-9]+)/always_nodes/$', WorkflowJobNodeAlwaysNodesList.as_view(), name='workflow_job_node_always_nodes_list'),
     re_path(r'^(?P<pk>[0-9]+)/credentials/$', WorkflowJobNodeCredentialsList.as_view(), name='workflow_job_node_credentials_list'),
     re_path(r'^(?P<pk>[0-9]+)/labels/$', WorkflowJobNodeLabelsList.as_view(), name='workflow_job_node_labels_list'),
-    re_path(r'^(?P<pk>[0-9]+)/instance_groups/$', WorkflowJobNodeInstanceGroupsList.as_view(), name='workflow_job_node_instance_group_list'),
+    re_path(r'^(?P<pk>[0-9]+)/instance_groups/$', WorkflowJobNodeInstanceGroupsList.as_view(), name='workflow_job_node_instance_groups_list'),
 ]
 
 __all__ = ['urls']
diff --git a/awx/api/views/__init__.py b/awx/api/views/__init__.py
index 016075e53f..54f195ec45 100644
--- a/awx/api/views/__init__.py
+++ b/awx/api/views/__init__.py
@@ -209,6 +209,26 @@ def api_exception_handler(exc, context):
     return exception_handler(exc, context)
 
 
+class LabelList(DeleteLastUnattachLabelMixin, SubListCreateAttachDetachAPIView):
+    def post(self, request, *args, **kwargs):
+        # If a label already exists in the database, attach it instead of erroring out
+        # that it already exists
+        if not getattr(self, 'label_filter', None):
+            return Response(dict(msg=_('Class {} missing label filter.'.format(self.__class__.__name__))), status=status.HTTP_400_BAD_REQUEST)
+        if 'id' not in request.data and 'name' in request.data and 'organization' in request.data:
+            existing = models.Label.objects.filter(name=request.data['name'], organization_id=request.data['organization'])
+            if existing.exists():
+                existing = existing[0]
+                request.data['id'] = existing.id
+                del request.data['name']
+                del request.data['organization']
+        if models.Label.objects.filter(**{self.label_filter: self.kwargs['pk']}).count() > 100:
+            return Response(
+                dict(msg=_('Maximum number of labels for {} reached.'.format(self.parent_model._meta.verbose_name_raw))), status=status.HTTP_400_BAD_REQUEST
+            )
+        return super().post(request, *args, **kwargs)
+
+
 class DashboardView(APIView):
 
     deprecated = True
@@ -618,28 +638,13 @@ class ScheduleCredentialsList(LaunchConfigCredentialsBase):
     parent_model = models.Schedule
 
 
-class ScheduleLabelsList(DeleteLastUnattachLabelMixin, SubListCreateAttachDetachAPIView):
+class ScheduleLabelsList(LabelList):
 
     model = models.Label
     serializer_class = serializers.LabelSerializer
     parent_model = models.Schedule
     relationship = 'labels'
-
-    def post(self, request, *args, **kwargs):
-        # If a label already exists in the database, attach it instead of erroring out
-        # that it already exists
-        if 'id' not in request.data and 'name' in request.data and 'organization' in request.data:
-            existing = models.Label.objects.filter(name=request.data['name'], organization_id=request.data['organization'])
-            if existing.exists():
-                existing = existing[0]
-                request.data['id'] = existing.id
-                del request.data['name']
-                del request.data['organization']
-        if models.Label.objects.filter(schedule_labels=self.kwargs['pk']).count() > 100:
-            return Response(
-                dict(msg=_('Maximum number of labels for {} reached.'.format(self.parent_model._meta.verbose_name_raw))), status=status.HTTP_400_BAD_REQUEST
-            )
-        return super(ScheduleLabelsList, self).post(request, *args, **kwargs)
+    label_filter = 'schedule_labels'
 
 
 class ScheduleInstanceGroupList(SubListAttachDetachAPIView):
@@ -2752,28 +2757,13 @@ class JobTemplateCredentialsList(SubListCreateAttachDetachAPIView):
         return super(JobTemplateCredentialsList, self).is_valid_relation(parent, sub, created)
 
 
-class JobTemplateLabelList(DeleteLastUnattachLabelMixin, SubListCreateAttachDetachAPIView):
+class JobTemplateLabelList(LabelList):
 
     model = models.Label
     serializer_class = serializers.LabelSerializer
     parent_model = models.JobTemplate
     relationship = 'labels'
-
-    def post(self, request, *args, **kwargs):
-        # If a label already exists in the database, attach it instead of erroring out
-        # that it already exists
-        if 'id' not in request.data and 'name' in request.data and 'organization' in request.data:
-            existing = models.Label.objects.filter(name=request.data['name'], organization_id=request.data['organization'])
-            if existing.exists():
-                existing = existing[0]
-                request.data['id'] = existing.id
-                del request.data['name']
-                del request.data['organization']
-        if models.Label.objects.filter(unifiedjobtemplate_labels=self.kwargs['pk']).count() > 100:
-            return Response(
-                dict(msg=_('Maximum number of labels for {} reached.'.format(self.parent_model._meta.verbose_name_raw))), status=status.HTTP_400_BAD_REQUEST
-            )
-        return super(JobTemplateLabelList, self).post(request, *args, **kwargs)
+    label_filter = 'unifiedjobtemplate_labels'
 
 
 class JobTemplateCallback(GenericAPIView):
@@ -2999,28 +2989,13 @@ class WorkflowJobNodeCredentialsList(SubListAPIView):
     relationship = 'credentials'
 
 
-class WorkflowJobNodeLabelsList(DeleteLastUnattachLabelMixin, SubListCreateAttachDetachAPIView):
+class WorkflowJobNodeLabelsList(LabelList):
 
     model = models.Label
     serializer_class = serializers.LabelSerializer
     parent_model = models.WorkflowJobNode
     relationship = 'labels'
-
-    def post(self, request, *args, **kwargs):
-        # If a label already exists in the database, attach it instead of erroring out
-        # that it already exists
-        if 'id' not in request.data and 'name' in request.data and 'organization' in request.data:
-            existing = models.Label.objects.filter(name=request.data['name'], organization_id=request.data['organization'])
-            if existing.exists():
-                existing = existing[0]
-                request.data['id'] = existing.id
-                del request.data['name']
-                del request.data['organization']
-        if models.Label.objects.filter(workflowjobnode_labels=self.kwargs['pk']).count() > 100:
-            return Response(
-                dict(msg=_('Maximum number of labels for {} reached.'.format(self.parent_model._meta.verbose_name_raw))), status=status.HTTP_400_BAD_REQUEST
-            )
-        return super(WorkflowJobNodeLabelsList, self).post(request, *args, **kwargs)
+    label_filter = 'workflowjobnode_labels'
 
 
 class WorkflowJobNodeInstanceGroupsList(SubListAttachDetachAPIView):
@@ -3049,28 +3024,13 @@ class WorkflowJobTemplateNodeCredentialsList(LaunchConfigCredentialsBase):
     parent_model = models.WorkflowJobTemplateNode
 
 
-class WorkflowJobTemplateNodeLabelsList(DeleteLastUnattachLabelMixin, SubListCreateAttachDetachAPIView):
+class WorkflowJobTemplateNodeLabelsList(LabelList):
 
     model = models.Label
     serializer_class = serializers.LabelSerializer
     parent_model = models.WorkflowJobTemplateNode
     relationship = 'labels'
-
-    def post(self, request, *args, **kwargs):
-        # If a label already exists in the database, attach it instead of erroring out
-        # that it already exists
-        if 'id' not in request.data and 'name' in request.data and 'organization' in request.data:
-            existing = models.Label.objects.filter(name=request.data['name'], organization_id=request.data['organization'])
-            if existing.exists():
-                existing = existing[0]
-                request.data['id'] = existing.id
-                del request.data['name']
-                del request.data['organization']
-        if models.Label.objects.filter(workflowjobtemplatenode_labels=self.kwargs['pk']).count() > 100:
-            return Response(
-                dict(msg=_('Maximum number of labels for {} reached.'.format(self.parent_model._meta.verbose_name_raw))), status=status.HTTP_400_BAD_REQUEST
-            )
-        return super(WorkflowJobTemplateNodeLabelsList, self).post(request, *args, **kwargs)
+    label_filter = 'workflowjobtemplatenode_labels'
 
 
 class WorkflowJobTemplateNodeInstanceGroupsList(SubListAttachDetachAPIView):
diff --git a/awx/main/access.py b/awx/main/access.py
index b34d24e023..81a1d4372f 100644
--- a/awx/main/access.py
+++ b/awx/main/access.py
@@ -1923,24 +1923,29 @@ class JobLaunchConfigAccess(BaseAccess):
         return self.check_related('inventory', Inventory, data, obj=obj, role_field='use_role')
 
     def can_attach(self, obj, sub_obj, relationship, data, skip_sub_obj_read_check=False):
+        try:
+            obj_name = obj.name
+        except AttributeError:
+            obj_name = obj.identifier
+
         if isinstance(sub_obj, Credential) and relationship == 'credentials':
             if not self.user in sub_obj.use_role:
                 logger.debug(
-                    "User {} not allowed access to credential {} for {} {} ({})".format(self.user.username, sub_obj.name, obj.__class__, obj.name, obj.id)
+                    "User {} not allowed access to credential {} for {} {} ({})".format(self.user.username, sub_obj.name, obj.__class__, obj_name, obj.id)
                 )
                 return False
             return True
 
         if isinstance(sub_obj, Label) and relationship == 'labels':
             if not self.user.can_access(Label, 'read', sub_obj):
-                logger.debug("User {} not allowed access to label {} for {} {} ({})".format(self.user.username, sub_obj.name, obj.__class__, obj.name, obj.id))
+                logger.debug("User {} not allowed access to label {} for {} {} ({})".format(self.user.username, sub_obj.name, obj.__class__, obj_name, obj.id))
                 return False
             return True
 
         if isinstance(sub_obj, InstanceGroup) and relationship == 'instance_groups':
             if not sub_obj in self.user.get_queryset(InstanceGroup):
                 logger.debug(
-                    "User {} not allowed access to instance_group {} for {} {} ({})".format(self.user.username, sub_obj.name, obj.__class__, obj.name, obj.id)
+                    "User {} not allowed access to instance_group {} for {} {} ({})".format(self.user.username, sub_obj.name, obj.__class__, obj_name, obj.id)
                 )
                 return False
             return True
@@ -1948,18 +1953,23 @@ class JobLaunchConfigAccess(BaseAccess):
         raise NotImplementedError('Only credentials, labels and instance groups can be attached to launch configurations.')
 
     def can_unattach(self, obj, sub_obj, relationship, data, skip_sub_obj_read_check=False):
+        try:
+            obj_name = obj.name
+        except AttributeError:
+            obj_name = obj.identifier
+
         if isinstance(sub_obj, Credential) and relationship == 'credentials':
             if not skip_sub_obj_read_check:
                 logger.debug(
                     "Skipping check if user {} can access credential {} ({}) for removal from {} {} ({})".format(
-                        self.user.username, sub_obj.name, sub_obj.id, obj.__class__, obj.name, obj.id
+                        self.user.username, sub_obj.name, sub_obj.id, obj.__class__, obj_name, obj.id
                     )
                 )
                 return True
             if not self.user in sub_obj.read_role:
                 logger.debug(
                     "User {} can not read credential {} ({}) for removal from {} {} ({})".format(
-                        self.user.username, sub_obj.name, sub_obj.id, obj.__class__, obj.name, obj.id
+                        self.user.username, sub_obj.name, sub_obj.id, obj.__class__, obj_name, obj.id
                     )
                 )
                 return False
@@ -1968,7 +1978,7 @@ class JobLaunchConfigAccess(BaseAccess):
             if skip_sub_obj_read_check:
                 logger.debug(
                     "Skipping check if user {} can access label {} ({}) for removal from {} {} ({})".format(
-                        self.user.username, sub_obj.name, sub_obj.id, obj.__class__, obj.name, obj.id
+                        self.user.username, sub_obj.name, sub_obj.id, obj.__class__, obj_name, obj.id
                     )
                 )
                 return True
@@ -1976,7 +1986,7 @@ class JobLaunchConfigAccess(BaseAccess):
                 return True
             logger.debug(
                 "User {} can not read label {} ({}) for removal from {} {} ({})".format(
-                    self.user.username, sub_obj.name, sub_obj.id, obj.__class__, obj.name, obj.id
+                    self.user.username, sub_obj.name, sub_obj.id, obj.__class__, obj_name, obj.id
                 )
             )
             return False
@@ -1984,7 +1994,7 @@ class JobLaunchConfigAccess(BaseAccess):
             if skip_sub_obj_read_check:
                 logger.debug(
                     "Skipping check if user {} can access instance_group {} ({}) for removal from {} {} ({})".format(
-                        self.user.username, sub_obj.name, sub_obj.id, obj.__class__, obj.name, obj.id
+                        self.user.username, sub_obj.name, sub_obj.id, obj.__class__, obj_name, obj.id
                     )
                 )
                 return True
@@ -1992,7 +2002,7 @@ class JobLaunchConfigAccess(BaseAccess):
                 return True
             logger.debug(
                 "User {} can not read instance_group {} ({}) for removal from {} {} ({})".format(
-                    self.user.username, sub_obj.name, sub_obj.id, obj.__class__, obj.name, obj.id
+                    self.user.username, sub_obj.name, sub_obj.id, obj.__class__, obj_name, obj.id
                 )
             )
             return False
@@ -2069,61 +2079,25 @@ class WorkflowJobTemplateNodeAccess(BaseAccess):
     def can_attach(self, obj, sub_obj, relationship, data, skip_sub_obj_read_check=False):
         if not self.wfjt_admin(obj):
             return False
-        if relationship == 'credentials':
+        if relationship in ['credentials', 'labels', 'instance_groups']:
             # Need permission to related template to attach a credential
             if not self.ujt_execute(obj):
                 return False
             return JobLaunchConfigAccess(self.user).can_attach(obj, sub_obj, relationship, data, skip_sub_obj_read_check=skip_sub_obj_read_check)
         elif relationship in ('success_nodes', 'failure_nodes', 'always_nodes'):
             return self.check_same_WFJT(obj, sub_obj)
-        elif relationship == 'labels':
-            if self.user.can_access(Label, 'read', sub_obj):
-                return True
-            logger.debug(
-                "User {} can not read label {} ({}) for removal from {} {} ({})".format(
-                    self.user.username, sub_obj.name, sub_obj.id, obj.__class__, obj.name, obj.id
-                )
-            )
-            return False
-        elif relationship == 'instance_groups':
-            if sub_obj in self.user.get_queryset(InstanceGroup):
-                return True
-            logger.debug(
-                "User {} can not read instance_group {} ({}) for removal from {} {} ({})".format(
-                    self.user.username, sub_obj.name, sub_obj.id, obj.__class__, obj.name, obj.id
-                )
-            )
-            return False
         else:
             raise NotImplementedError('Relationship {} not understood for WFJT nodes.'.format(relationship))
 
     def can_unattach(self, obj, sub_obj, relationship, data, skip_sub_obj_read_check=False):
         if not self.wfjt_admin(obj):
             return False
-        if relationship == 'credentials':
+        if relationship in ['credentials', 'labels', 'instance_groups']:
             if not self.ujt_execute(obj):
                 return False
             return JobLaunchConfigAccess(self.user).can_unattach(obj, sub_obj, relationship, data, skip_sub_obj_read_check=skip_sub_obj_read_check)
         elif relationship in ('success_nodes', 'failure_nodes', 'always_nodes'):
             return self.check_same_WFJT(obj, sub_obj)
-        elif relationship == 'labels':
-            if self.user.can_access(Label, 'read', sub_obj):
-                return True
-            logger.debug(
-                "User {} can not read label {} ({}) for removal from {} {} ({})".format(
-                    self.user.username, sub_obj.name, sub_obj.id, obj.__class__, obj.name, obj.id
-                )
-            )
-            return False
-        elif relationship == 'instance_groups':
-            if sub_obj in self.user.get_queryset(InstanceGroup):
-                return True
-            logger.debug(
-                "User {} can not read instance_group {} ({}) for removal from {} {} ({})".format(
-                    self.user.username, sub_obj.name, sub_obj.id, obj.__class__, obj.name, obj.id
-                )
-            )
-            return False
         else:
             raise NotImplementedError('Relationship {} not understood for WFJT nodes.'.format(relationship))
 
diff --git a/awx/main/migrations/0167_jt_prompt_everything_on_launch.py b/awx/main/migrations/0167_jt_prompt_everything_on_launch.py
index e2c87b3fa0..9bd3736c70 100644
--- a/awx/main/migrations/0167_jt_prompt_everything_on_launch.py
+++ b/awx/main/migrations/0167_jt_prompt_everything_on_launch.py
@@ -1,4 +1,4 @@
-# Generated by Django 3.2.13 on 2022-08-31 19:15
+# Generated by Django 3.2.13 on 2022-09-06 19:50
 
 import awx.main.fields
 import awx.main.utils.polymorphic
@@ -94,6 +94,21 @@ class Migration(migrations.Migration):
             name='labels',
             field=models.ManyToManyField(related_name='workflowjobnode_labels', to='main.Label'),
         ),
+        migrations.AddField(
+            model_name='workflowjobtemplate',
+            name='ask_labels_on_launch',
+            field=awx.main.fields.AskForField(blank=True, default=False),
+        ),
+        migrations.AddField(
+            model_name='workflowjobtemplate',
+            name='ask_skip_tags_on_launch',
+            field=awx.main.fields.AskForField(blank=True, default=False),
+        ),
+        migrations.AddField(
+            model_name='workflowjobtemplate',
+            name='ask_tags_on_launch',
+            field=awx.main.fields.AskForField(blank=True, default=False),
+        ),
         migrations.AddField(
             model_name='workflowjobtemplatenode',
             name='execution_environment',
@@ -117,7 +132,7 @@ class Migration(migrations.Migration):
                 ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
                 ('position', models.PositiveIntegerField(db_index=True, default=None, null=True)),
                 ('instancegroup', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='main.instancegroup')),
-                ('schedule', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='main.workflowjobtemplatenode')),
+                ('workflowjobtemplatenode', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='main.workflowjobtemplatenode')),
             ],
         ),
         migrations.CreateModel(
@@ -126,7 +141,7 @@ class Migration(migrations.Migration):
                 ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
                 ('position', models.PositiveIntegerField(db_index=True, default=None, null=True)),
                 ('instancegroup', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='main.instancegroup')),
-                ('schedule', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='main.workflowjobnode')),
+                ('workflowjobnode', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='main.workflowjobnode')),
             ],
         ),
         migrations.CreateModel(
diff --git a/awx/main/models/ha.py b/awx/main/models/ha.py
index 3d4467403a..5e51284299 100644
--- a/awx/main/models/ha.py
+++ b/awx/main/models/ha.py
@@ -471,7 +471,7 @@ class ScheduleInstanceGroupMembership(models.Model):
 
 class WorkflowJobTemplateNodeBaseInstanceGroupMembership(models.Model):
 
-    schedule = models.ForeignKey('WorkflowJobTemplateNode', on_delete=models.CASCADE)
+    workflowjobtemplatenode = models.ForeignKey('WorkflowJobTemplateNode', on_delete=models.CASCADE)
     instancegroup = models.ForeignKey('InstanceGroup', on_delete=models.CASCADE)
     position = models.PositiveIntegerField(
         null=True,
@@ -482,7 +482,7 @@ class WorkflowJobTemplateNodeBaseInstanceGroupMembership(models.Model):
 
 class WorkflowJobNodeBaseInstanceGroupMembership(models.Model):
 
-    schedule = models.ForeignKey('WorkflowJobNode', on_delete=models.CASCADE)
+    workflowjobnode = models.ForeignKey('WorkflowJobNode', on_delete=models.CASCADE)
     instancegroup = models.ForeignKey('InstanceGroup', on_delete=models.CASCADE)
     position = models.PositiveIntegerField(
         null=True,
diff --git a/awx/main/models/jobs.py b/awx/main/models/jobs.py
index cd2add728b..bfefcb8f83 100644
--- a/awx/main/models/jobs.py
+++ b/awx/main/models/jobs.py
@@ -332,8 +332,11 @@ class JobTemplate(UnifiedJobTemplate, JobOptions, SurveyJobTemplateMixin, Resour
         if self.ask_inventory_on_launch and 'inventory' in kwargs:
             actual_inventory = kwargs['inventory']
         actual_slice_count = self.job_slice_count
-        if self.ask_job_slice_count_on_launch and 'slice_count' in kwargs:
-            actual_slice_count = kwargs['slice_count']
+        if self.ask_job_slice_count_on_launch and 'job_slice_count' in kwargs:
+            actual_slice_count = kwargs['job_slice_count']
+            # Set the eager fields if its there as well
+            if '_eager_fields' in kwargs and 'job_slice_count' in kwargs['_eager_fields']:
+                kwargs['_eager_fields']['job_slice_count'] = actual_slice_count
         if actual_inventory:
             return min(actual_slice_count, actual_inventory.hosts.count())
         else:
diff --git a/awx/main/models/workflow.py b/awx/main/models/workflow.py
index 723e6faa6a..7330128873 100644
--- a/awx/main/models/workflow.py
+++ b/awx/main/models/workflow.py
@@ -29,7 +29,7 @@ from awx.main.models import prevent_search, accepts_json, UnifiedJobTemplate, Un
 from awx.main.models.notifications import NotificationTemplate, JobNotificationMixin
 from awx.main.models.base import CreatedModifiedModel, VarsDictProperty
 from awx.main.models.rbac import ROLE_SINGLETON_SYSTEM_ADMINISTRATOR, ROLE_SINGLETON_SYSTEM_AUDITOR
-from awx.main.fields import ImplicitRoleField, AskForField, JSONBlob, OrderedManyToManyField
+from awx.main.fields import ImplicitRoleField, JSONBlob, OrderedManyToManyField
 from awx.main.models.mixins import (
     ResourceMixin,
     SurveyJobTemplateMixin,
diff --git a/awx/main/tests/functional/models/test_job.py b/awx/main/tests/functional/models/test_job.py
index 06bc2661dd..ae4f053aca 100644
--- a/awx/main/tests/functional/models/test_job.py
+++ b/awx/main/tests/functional/models/test_job.py
@@ -78,4 +78,4 @@ class TestSlicingModels:
         # The inventory slice count will be the min of the number of nodes (4) or the job slice (2)
         assert job_template.get_effective_slice_ct({'inventory': inventory2}) == 2
         # Now we are going to pass in an override (like the prompt would) and as long as that is < host count we expect that back
-        assert job_template.get_effective_slice_ct({'inventory': inventory2, 'slice_count': 3}) == 3
+        assert job_template.get_effective_slice_ct({'inventory': inventory2, 'job_slice_count': 3}) == 3
diff --git a/awx/main/tests/functional/test_rbac_workflow.py b/awx/main/tests/functional/test_rbac_workflow.py
index d48eb3f80b..15577b65a4 100644
--- a/awx/main/tests/functional/test_rbac_workflow.py
+++ b/awx/main/tests/functional/test_rbac_workflow.py
@@ -101,6 +101,80 @@ class TestWorkflowJobTemplateNodeAccess:
         access = WorkflowJobTemplateNodeAccess(rando)
         assert access.can_delete(wfjt_node)
 
+    @pytest.mark.parametrize(
+        "add_wfjt_admin, add_jt_admin, permission_type, expected_result, method_type",
+        [
+            (False, False, None, False, 'can_attach'),
+            (True, False, 'credentials', False, 'can_attach'),
+            (True, True, 'credentials', True, 'can_attach'),
+            (True, False, 'labels', False, 'can_attach'),
+            (True, True, 'labels', True, 'can_attach'),
+            (True, False, 'instance_groups', False, 'can_attach'),
+            (True, True, 'instance_groups', True, 'can_attach'),
+            (False, False, None, False, 'can_unattach'),
+            (True, False, 'credentials', False, 'can_unattach'),
+            (True, True, 'credentials', True, 'can_unattach'),
+            (True, False, 'labels', False, 'can_unattach'),
+            (True, True, 'labels', True, 'can_unattach'),
+            (True, False, 'instance_groups', False, 'can_unattach'),
+            (True, True, 'instance_groups', True, 'can_unattach'),
+        ],
+    )
+    def test_attacher_permissions(self, wfjt_node, job_template, rando, add_wfjt_admin, permission_type, add_jt_admin, expected_result, mocker, method_type):
+        wfjt = wfjt_node.workflow_job_template
+        if add_wfjt_admin:
+            wfjt.admin_role.members.add(rando)
+            wfjt.unified_job_template = job_template
+        if add_jt_admin:
+            job_template.execute_role.members.add(rando)
+
+        # We have to mock the JobLaunchConfigAccess because the attachment methods will look at the object type and the relation
+        # Since we pass None as the second param this will trigger an NotImplementedError from that object
+        with mocker.patch('awx.main.access.JobLaunchConfigAccess.{}'.format(method_type), return_value=True):
+            access = WorkflowJobTemplateNodeAccess(rando)
+            assert getattr(access, method_type)(wfjt_node, None, permission_type, None) == expected_result
+
+    # The actual attachment of labels, credentials and instance groups are tested from JobLaunchConfigAccess
+
+    @pytest.mark.parametrize(
+        "attachment_type, expect_exception, method_type",
+        [
+            ("credentials", False, 'can_attach'),
+            ("labels", False, 'can_attach'),
+            ("instance_groups", False, 'can_attach'),
+            ("success_nodes", False, 'can_attach'),
+            ("failure_nodes", False, 'can_attach'),
+            ("always_nodes", False, 'can_attach'),
+            ("junk", True, 'can_attach'),
+            ("credentials", False, 'can_unattach'),
+            ("labels", False, 'can_unattach'),
+            ("instance_groups", False, 'can_unattach'),
+            ("success_nodes", False, 'can_unattach'),
+            ("failure_nodes", False, 'can_unattach'),
+            ("always_nodes", False, 'can_unattach'),
+            ("junk", True, 'can_unattach'),
+        ],
+    )
+    def test_attacher_raise_not_implemented(self, wfjt_node, rando, attachment_type, expect_exception, method_type):
+        wfjt = wfjt_node.workflow_job_template
+        wfjt.admin_role.members.add(rando)
+        access = WorkflowJobTemplateNodeAccess(rando)
+        if expect_exception:
+            with pytest.raises(NotImplementedError):
+                access.can_attach(wfjt_node, None, attachment_type, None)
+        else:
+            try:
+                getattr(access, method_type)(wfjt_node, None, attachment_type, None)
+            except NotImplementedError:
+                # We explicitly catch NotImplemented because the _nodes type will raise a different exception
+                assert False, "Exception was raised when it should not have been"
+            except Exception:
+                #  File "/awx_devel/awx/main/access.py", line 2074, in check_same_WFJT
+                #    raise Exception('Attaching workflow nodes only allowed for other nodes')
+                pass
+
+    # TODO: Implement additional tests for _nodes attachments here
+
 
 @pytest.mark.django_db
 class TestWorkflowJobAccess:
diff --git a/awx/main/tests/unit/api/serializers/test_workflow_serializers.py b/awx/main/tests/unit/api/serializers/test_workflow_serializers.py
index 0cbf6b7af0..9e7fe51344 100644
--- a/awx/main/tests/unit/api/serializers/test_workflow_serializers.py
+++ b/awx/main/tests/unit/api/serializers/test_workflow_serializers.py
@@ -27,6 +27,7 @@ class TestWorkflowJobTemplateSerializerGetRelated:
             'launch',
             'workflow_nodes',
             'webhook_key',
+            'labels',
         ],
     )
     def test_get_related(self, mocker, test_get_related, workflow_job_template, related_resource_name):
@@ -89,6 +90,8 @@ class TestWorkflowJobTemplateNodeSerializerGetRelated:
             'success_nodes',
             'failure_nodes',
             'always_nodes',
+            'labels',
+            'instance_groups',
         ],
     )
     def test_get_related(self, test_get_related, workflow_job_template_node, related_resource_name):
@@ -233,6 +236,8 @@ class TestWorkflowJobNodeSerializerGetRelated:
             'success_nodes',
             'failure_nodes',
             'always_nodes',
+            'labels',
+            'instance_groups',
         ],
     )
     def test_get_related(self, test_get_related, workflow_job_node, related_resource_name):
diff --git a/awx_collection/plugins/modules/schedule.py b/awx_collection/plugins/modules/schedule.py
index c4c2b5d711..d0fac2384e 100644
--- a/awx_collection/plugins/modules/schedule.py
+++ b/awx_collection/plugins/modules/schedule.py
@@ -73,6 +73,8 @@ options:
     labels:
       description:
         - List of labels applied as a prompt, assuming job template prompts for labels
+      type: list
+      elements: str
     credentials:
       description:
         - List of credentials applied as a prompt, assuming job template prompts for credentials
diff --git a/awx_collection/plugins/modules/workflow_job_template_node.py b/awx_collection/plugins/modules/workflow_job_template_node.py
index 6fbfb2bcc0..f91d308282 100644
--- a/awx_collection/plugins/modules/workflow_job_template_node.py
+++ b/awx_collection/plugins/modules/workflow_job_template_node.py
@@ -172,6 +172,8 @@ options:
     labels:
       description:
         - List of labels applied as a prompt, assuming job template prompts for labels
+      type: list
+      elements: str
     timeout:
       description:
         - Timeout applied as a prompt, assuming job template prompts for timeout