diff --git a/awx/api/serializers.py b/awx/api/serializers.py
index 5ee7d1b3b7..eb6264c5e2 100644
--- a/awx/api/serializers.py
+++ b/awx/api/serializers.py
@@ -3199,6 +3199,13 @@ class JobSerializer(UnifiedJobSerializer, JobOptionsSerializer):
 
     def get_summary_fields(self, obj):
         summary_fields = super(JobSerializer, self).get_summary_fields(obj)
+        if obj.internal_limit:
+            summary_fields['internal_limit'] = {}
+            if obj.internal_limit.startswith('shard'):
+                offset, step = Inventory.parse_shard_params(obj.internal_limit)
+                summary_fields['internal_limit']['shard'] = {'offset': offset, 'step': step}
+            else:
+                summary_fields['internal_limit']['unknown'] = self.internal_limit
         all_creds = []
         # Organize credential data into multitude of deprecated fields
         # TODO: remove most of this as v1 is removed
diff --git a/awx/api/views/__init__.py b/awx/api/views/__init__.py
index cd96f3fcbc..e99532cbca 100644
--- a/awx/api/views/__init__.py
+++ b/awx/api/views/__init__.py
@@ -2914,11 +2914,13 @@ class JobTemplateLaunch(RetrieveAPIView):
             return Response(data, status=status.HTTP_400_BAD_REQUEST)
         else:
             data = OrderedDict()
-            data['job'] = new_job.id
-            data['ignored_fields'] = self.sanitize_for_response(ignored_fields)
             if isinstance(new_job, WorkflowJob):
+                data['workflow_job'] = new_job.id
+                data['ignored_fields'] = self.sanitize_for_response(ignored_fields)
                 data.update(WorkflowJobSerializer(new_job, context=self.get_serializer_context()).to_representation(new_job))
             else:
+                data['job'] = new_job.id
+                data['ignored_fields'] = self.sanitize_for_response(ignored_fields)
                 data.update(JobSerializer(new_job, context=self.get_serializer_context()).to_representation(new_job))
             headers = {'Location': new_job.get_absolute_url(request)}
             return Response(data, status=status.HTTP_201_CREATED, headers=headers)
diff --git a/awx/main/migrations/0048_v330_split_jobs.py b/awx/main/migrations/0048_v330_split_jobs.py
new file mode 100644
index 0000000000..17c6ab65ca
--- /dev/null
+++ b/awx/main/migrations/0048_v330_split_jobs.py
@@ -0,0 +1,37 @@
+# -*- coding: utf-8 -*-
+# Generated by Django 1.11.11 on 2018-09-13 15:55
+from __future__ import unicode_literals
+
+import awx.main.utils.polymorphic
+from django.db import migrations, models
+import django.db.models.deletion
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('main', '0047_v330_activitystream_instance'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='jobtemplate',
+            name='job_shard_count',
+            field=models.IntegerField(blank=True, default=0, help_text='The number of jobs to split into at runtime. Will cause the Job Template to launch a workflow if value is non-zero.'),
+        ),
+        migrations.AddField(
+            model_name='workflowjob',
+            name='job_template',
+            field=models.ForeignKey(blank=True, default=None, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='sharded_jobs', to='main.JobTemplate'),
+        ),
+        migrations.AlterField(
+            model_name='unifiedjob',
+            name='unified_job_template',
+            field=models.ForeignKey(default=None, editable=False, null=True, on_delete=awx.main.utils.polymorphic.SET_NULL, related_name='unifiedjob_unified_jobs', to='main.UnifiedJobTemplate'),
+        ),
+        migrations.AddField(
+            model_name='job',
+            name='internal_limit',
+            field=models.CharField(default=b'', editable=False, max_length=1024),
+        ),
+    ]
diff --git a/awx/main/migrations/0048_v340_split_jobs.py b/awx/main/migrations/0048_v340_split_jobs.py
deleted file mode 100644
index de1242760a..0000000000
--- a/awx/main/migrations/0048_v340_split_jobs.py
+++ /dev/null
@@ -1,20 +0,0 @@
-# -*- coding: utf-8 -*-
-# Generated by Django 1.11.11 on 2018-08-14 13:43
-from django.db import migrations, models
-
-
-class Migration(migrations.Migration):
-
-    dependencies = [
-        ('main', '0047_v330_activitystream_instance'),
-    ]
-
-    operations = [
-        migrations.AddField(
-            model_name='jobtemplate',
-            name='job_shard_count',
-            field=models.IntegerField(blank=True,
-                                      default=0,
-                                      help_text='The number of jobs to split into at runtime. Will cause the Job Template to launch a workflow.'),
-        ),
-    ]
diff --git a/awx/main/migrations/0049_v340_add_job_template.py b/awx/main/migrations/0049_v340_add_job_template.py
deleted file mode 100644
index 3174ca9532..0000000000
--- a/awx/main/migrations/0049_v340_add_job_template.py
+++ /dev/null
@@ -1,26 +0,0 @@
-# -*- coding: utf-8 -*-
-# Generated by Django 1.11.11 on 2018-08-14 16:04
-from __future__ import unicode_literals
-
-from django.conf import settings
-from django.db import migrations, models
-import django.db.models.deletion
-
-
-class Migration(migrations.Migration):
-
-    dependencies = [
-        ('main', '0048_v340_split_jobs'),
-    ]
-
-    operations = [
-        migrations.AddField(
-            model_name='workflowjob',
-            name='job_template',
-            field=models.ForeignKey(blank=True,
-                                    default=None,
-                                    null=True,
-                                    on_delete=django.db.models.deletion.SET_NULL,
-                                    related_name='sharded_jobs', to='main.JobTemplate'),
-        ),
-    ]
diff --git a/awx/main/migrations/0050_v340_unified_jt_set_null.py b/awx/main/migrations/0050_v340_unified_jt_set_null.py
deleted file mode 100644
index 7ad65087f4..0000000000
--- a/awx/main/migrations/0050_v340_unified_jt_set_null.py
+++ /dev/null
@@ -1,21 +0,0 @@
-# -*- coding: utf-8 -*-
-# Generated by Django 1.11.11 on 2018-09-10 17:41
-from __future__ import unicode_literals
-
-import awx.main.utils.polymorphic
-from django.db import migrations, models
-
-
-class Migration(migrations.Migration):
-
-    dependencies = [
-        ('main', '0049_v340_add_job_template'),
-    ]
-
-    operations = [
-        migrations.AlterField(
-            model_name='unifiedjob',
-            name='unified_job_template',
-            field=models.ForeignKey(default=None, editable=False, null=True, on_delete=awx.main.utils.polymorphic.SET_NULL, related_name='unifiedjob_unified_jobs', to='main.UnifiedJobTemplate'),
-        ),
-    ]
diff --git a/awx/main/models/jobs.py b/awx/main/models/jobs.py
index bb1b229a43..d698ad3726 100644
--- a/awx/main/models/jobs.py
+++ b/awx/main/models/jobs.py
@@ -280,7 +280,8 @@ class JobTemplate(UnifiedJobTemplate, JobOptions, SurveyJobTemplateMixin, Resour
     job_shard_count = models.IntegerField(
         blank=True,
         default=0,
-        help_text=_("The number of jobs to split into at runtime. Will cause the Job Template to launch a workflow."),
+        help_text=_("The number of jobs to split into at runtime. "
+                    "Will cause the Job Template to launch a workflow if value is non-zero."),
     )
 
     admin_role = ImplicitRoleField(
@@ -301,7 +302,7 @@ class JobTemplate(UnifiedJobTemplate, JobOptions, SurveyJobTemplateMixin, Resour
     @classmethod
     def _get_unified_job_field_names(cls):
         return set(f.name for f in JobOptions._meta.fields) | set(
-            ['name', 'description', 'schedule', 'survey_passwords', 'labels', 'credentials']
+            ['name', 'description', 'schedule', 'survey_passwords', 'labels', 'credentials', 'internal_limit']
         )
 
     @property
@@ -327,10 +328,8 @@ class JobTemplate(UnifiedJobTemplate, JobOptions, SurveyJobTemplateMixin, Resour
         return self.create_unified_job(**kwargs)
 
     def create_unified_job(self, **kwargs):
-        split_event = bool(
-            self.job_shard_count > 1 and
-            not kwargs.pop('_prevent_sharding', False)
-        )
+        prevent_sharding = kwargs.pop('_prevent_sharding', False)
+        split_event = bool(self.job_shard_count > 1 and (not prevent_sharding))
         if split_event:
             # A sharded Job Template will generate a WorkflowJob rather than a Job
             from awx.main.models.workflow import WorkflowJobTemplate, WorkflowJobNode
@@ -532,6 +531,11 @@ class Job(UnifiedJob, JobOptions, SurveyJobMixin, JobNotificationMixin, TaskMana
         on_delete=models.SET_NULL,
         help_text=_('The SCM Refresh task used to make sure the playbooks were available for the job run'),
     )
+    internal_limit = models.CharField(
+        max_length=1024,
+        default='',
+        editable=False,
+    )
 
 
     def _get_parent_field_name(self):
@@ -575,6 +579,13 @@ class Job(UnifiedJob, JobOptions, SurveyJobMixin, JobNotificationMixin, TaskMana
     def event_class(self):
         return JobEvent
 
+    def copy_unified_job(self, **new_prompts):
+        new_prompts['_prevent_sharding'] = True
+        if self.internal_limit:
+            new_prompts.setdefault('_eager_fields', {})
+            new_prompts['_eager_fields']['internal_limit'] = self.internal_limit  # oddball, not from JT or prompts
+        return super(Job, self).copy_unified_job(**new_prompts)
+
     @property
     def ask_diff_mode_on_launch(self):
         if self.job_template is not None:
diff --git a/awx/main/models/workflow.py b/awx/main/models/workflow.py
index 8994c1aa1a..f15e66b226 100644
--- a/awx/main/models/workflow.py
+++ b/awx/main/models/workflow.py
@@ -259,6 +259,10 @@ class WorkflowJobNode(WorkflowNodeBase):
                 shard_str
             )
             data['_eager_fields']['allow_simultaneous'] = True
+            data['_eager_fields']['internal_limit'] = 'shard{0}of{1}'.format(
+                self.ancestor_artifacts['job_shard'],
+                self.workflow_job.workflow_job_nodes.count()
+            )
             data['_prevent_sharding'] = True
         return data
 
@@ -314,7 +318,7 @@ class WorkflowJobOptions(BaseModel):
 
     def create_relaunch_workflow_job(self):
         new_workflow_job = self.copy_unified_job()
-        if self.workflow_job_template is None:
+        if self.unified_job_template_id is None:
             new_workflow_job.copy_nodes_from_original(original=self)
         return new_workflow_job
 
diff --git a/awx/main/scheduler/task_manager.py b/awx/main/scheduler/task_manager.py
index 08cb6cd247..540b94181e 100644
--- a/awx/main/scheduler/task_manager.py
+++ b/awx/main/scheduler/task_manager.py
@@ -419,7 +419,7 @@ class TaskManager():
                 logger.debug(six.text_type("Dependent {} couldn't be scheduled on graph, waiting for next cycle").format(task.log_format))
 
     def process_pending_tasks(self, pending_tasks):
-        running_workflow_templates = set([wf.workflow_job_template_id for wf in self.get_running_workflow_jobs()])
+        running_workflow_templates = set([wf.unified_job_template_id for wf in self.get_running_workflow_jobs()])
         for task in pending_tasks:
             self.process_dependencies(task, self.generate_dependencies(task))
             if self.is_job_blocked(task):
@@ -429,12 +429,12 @@ class TaskManager():
             found_acceptable_queue = False
             idle_instance_that_fits = None
             if isinstance(task, WorkflowJob):
-                if task.workflow_job_template_id in running_workflow_templates:
+                if task.unified_job_template_id in running_workflow_templates:
                     if not task.allow_simultaneous:
                         logger.debug(six.text_type("{} is blocked from running, workflow already running").format(task.log_format))
                         continue
                 else:
-                    running_workflow_templates.add(task.workflow_job_template_id)
+                    running_workflow_templates.add(task.unified_job_template_id)
                 self.start_task(task, None, task.get_jobs_fail_chain(), None)
                 continue
             for rampart_group in preferred_instance_groups:
diff --git a/awx/main/tasks.py b/awx/main/tasks.py
index db573b0b68..2c543d88d3 100644
--- a/awx/main/tasks.py
+++ b/awx/main/tasks.py
@@ -825,15 +825,9 @@ class BaseTask(object):
         return False
 
     def build_inventory(self, instance, **kwargs):
-        workflow_job = instance.get_workflow_job()
-        if workflow_job and workflow_job.job_template_id:
-            shard_address = 'shard{0}of{1}'.format(
-                instance.unified_job_node.ancestor_artifacts['job_shard'],
-                workflow_job.workflow_job_nodes.count()
-            )
-            script_data = instance.inventory.get_script_data(hostvars=True, subset=shard_address)
-        else:
-            script_data = instance.inventory.get_script_data(hostvars=True)
+        script_data = instance.inventory.get_script_data(
+            hostvars=True, subset=getattr(instance, 'internal_limit', '')
+        )
         json_data = json.dumps(script_data)
         handle, path = tempfile.mkstemp(dir=kwargs.get('private_data_dir', None))
         f = os.fdopen(handle, 'w')
diff --git a/awx/ui/client/features/output/details.component.js b/awx/ui/client/features/output/details.component.js
index b1cc4734cd..bc39abeeaf 100644
--- a/awx/ui/client/features/output/details.component.js
+++ b/awx/ui/client/features/output/details.component.js
@@ -126,6 +126,26 @@ function getSourceWorkflowJobDetails () {
     return { link, tooltip };
 }
 
+function getShardDetails () {
+    const internalLimitDetails = resource.model.get('summary_fields.internal_limit');
+
+    if (!internalLimitDetails) {
+        return null;
+    }
+
+    const shardDetails = resource.model.get('summary_fields.internal_limit.shard');
+
+    if (!shardDetails) {
+        return null;
+    }
+
+    const label = strings.get('labels.SHARD_DETAILS');
+    const offset = `${shardDetails.offset} of ${shardDetails.step} shards`;
+    const tooltip = strings.get('tooltips.SHARD_DETAILS');
+
+    return { label, offset, tooltip };
+}
+
 function getJobTemplateDetails () {
     const jobTemplate = resource.model.get('summary_fields.job_template');
 
@@ -671,6 +691,7 @@ function JobDetailsController (
         vm.jobType = getJobTypeDetails();
         vm.jobTemplate = getJobTemplateDetails();
         vm.sourceWorkflowJob = getSourceWorkflowJobDetails();
+        vm.shardDetails = getShardDetails();
         vm.inventory = getInventoryDetails();
         vm.project = getProjectDetails();
         vm.projectUpdate = getProjectUpdateDetails();
diff --git a/awx/ui/client/features/output/details.partial.html b/awx/ui/client/features/output/details.partial.html
index f681059f90..2f3670f4ae 100644
--- a/awx/ui/client/features/output/details.partial.html
+++ b/awx/ui/client/features/output/details.partial.html
@@ -151,6 +151,12 @@
     <div class="JobResults-resultRowText">{{ vm.jobType.value }}</div>
 </div>
 
+<!-- SHAAAAAARD -->
+<div class="JobResults-resultRow" ng-if="vm.shardDetails">
+    <label class="JobResults-resultRowLabel">{{ vm.shardDetails.label }}</label>
+    <div class="JobResults-resultRowText">{{ vm.shardDetails.offset }}</div>
+</div>
+
 <!-- LAUNCHED BY DETAIL -->
 <div class="JobResults-resultRow" ng-if="vm.launchedBy">
     <label class="JobResults-resultRowLabel">{{ vm.launchedBy.label }}</label>
diff --git a/awx/ui/client/features/output/output.strings.js b/awx/ui/client/features/output/output.strings.js
index 538b533cb0..4fa6fe4335 100644
--- a/awx/ui/client/features/output/output.strings.js
+++ b/awx/ui/client/features/output/output.strings.js
@@ -23,6 +23,7 @@ function OutputStrings (BaseString) {
         EXTRA_VARS: t.s('Read-only view of extra variables added to the job template'),
         INVENTORY: t.s('View the Inventory'),
         JOB_TEMPLATE: t.s('View the Job Template'),
+        SHARD_DETAILS: t.s('Job is one of several shards from a JT that splits on inventory'),
         PROJECT: t.s('View the Project'),
         PROJECT_UPDATE: t.s('View Project checkout results'),
         SCHEDULE: t.s('View the Schedule'),
@@ -55,6 +56,7 @@ function OutputStrings (BaseString) {
         JOB_EXPLANATION: t.s('Explanation'),
         JOB_TAGS: t.s('Job Tags'),
         JOB_TEMPLATE: t.s('Job Template'),
+        SHARD_DETAILS: t.s('Shard Details'),
         JOB_TYPE: t.s('Job Type'),
         LABELS: t.s('Labels'),
         LAUNCHED_BY: t.s('Launched By'),
diff --git a/awx/ui/client/src/templates/job_templates/job-template.form.js b/awx/ui/client/src/templates/job_templates/job-template.form.js
index 54bbe2c70d..3773d2b865 100644
--- a/awx/ui/client/src/templates/job_templates/job-template.form.js
+++ b/awx/ui/client/src/templates/job_templates/job-template.form.js
@@ -271,6 +271,20 @@ function(NotificationsList, i18n) {
                     },
                     ngDisabled: '!(job_template_obj.summary_fields.user_capabilities.edit || canAddJobTemplate)'
                 },
+                job_shard_count: {
+                    label: i18n._('Number of job shards to use'),
+                    type: 'number',
+                    integer: true,
+                    min: 0,
+                    spinner: true,
+                    // 'class': "input-small",
+                    // toggleSource: 'diff_mode',
+                    dataTitle: i18n._('Job Shard Count'),
+                    dataPlacement: 'right',
+                    dataContainer: 'body',
+                    awPopOver: "<p>" + i18n._("If non-zero, split into multiple jobs that run on mutually exclusive slices of the inventory.") + "</p>",
+                    ngDisabled: '!(job_template_obj.summary_fields.user_capabilities.edit || canAddJobTemplate)'
+                },
                 checkbox_group: {
                     label: i18n._('Options'),
                     type: 'checkbox_group',
diff --git a/awx/ui/client/src/workflow-results/workflow-results.controller.js b/awx/ui/client/src/workflow-results/workflow-results.controller.js
index 31ad3191f8..f66e2a09e9 100644
--- a/awx/ui/client/src/workflow-results/workflow-results.controller.js
+++ b/awx/ui/client/src/workflow-results/workflow-results.controller.js
@@ -39,6 +39,7 @@ export default ['workflowData', 'workflowResultsService', 'workflowDataOptions',
                     DELETE: i18n._('Delete'),
                     EDIT_USER: i18n._('Edit the user'),
                     EDIT_WORKFLOW: i18n._('Edit the workflow job template'),
+                    EDIT_SHARD_TEMPLATE: i18n._('Edit the shard job template'),
                     EDIT_SCHEDULE: i18n._('Edit the schedule'),
                     TOGGLE_STDOUT_FULLSCREEN: i18n._('Expand Output'),
                     STATUS: '' // re-assigned elsewhere
@@ -49,6 +50,7 @@ export default ['workflowData', 'workflowResultsService', 'workflowDataOptions',
                     STARTED: i18n._('Started'),
                     FINISHED: i18n._('Finished'),
                     LABELS: i18n._('Labels'),
+                    SHARD_TEMPLATE: i18n._('Shard Template'),
                     STATUS: i18n._('Status')
                 },
                 details: {
@@ -109,6 +111,11 @@ export default ['workflowData', 'workflowResultsService', 'workflowDataOptions',
                     $scope.workflow_job_template_link = `/#/templates/workflow_job_template/${$scope.workflow.summary_fields.workflow_job_template.id}`;
             }
 
+            if(workflowData.summary_fields && workflowData.summary_fields.job_template &&
+                workflowData.summary_fields.job_template.id){
+                    $scope.shard_job_template_link = `/#/templates/job_template/${$scope.workflow.summary_fields.job_template.id}`;
+            }
+
             // turn related api browser routes into front end routes
             getLinks();
 
diff --git a/awx/ui/client/src/workflow-results/workflow-results.partial.html b/awx/ui/client/src/workflow-results/workflow-results.partial.html
index 97ef30a914..f3f8943dfc 100644
--- a/awx/ui/client/src/workflow-results/workflow-results.partial.html
+++ b/awx/ui/client/src/workflow-results/workflow-results.partial.html
@@ -144,6 +144,22 @@
                         </div>
                     </div>
 
+                    <!-- SHAAAAAARD -->
+                    <div class="WorkflowResults-resultRow"
+                        ng-show="workflow.summary_fields.job_template.name">
+                        <label
+                            class="WorkflowResults-resultRowLabel">
+                            {{ strings.labels.SHARD_TEMPLATE }}
+                        </label>
+                        <div class="WorkflowResults-resultRowText">
+                            <a href="{{ shard_job_template_link }}"
+                                aw-tool-tip="{{ strings.tooltips.EDIT_SHARD_TEMPLATE }}"
+                                data-placement="top">
+                                {{ workflow.summary_fields.job_template.name }}
+                            </a>
+                        </div>
+                    </div>
+
                     <!-- EXTRA VARIABLES DETAIL -->
                     <at-code-mirror
                         ng-if="variables"