diff --git a/awx/api/serializers.py b/awx/api/serializers.py
index d55571d27..6944b663e 100644
--- a/awx/api/serializers.py
+++ b/awx/api/serializers.py
@@ -4238,6 +4238,7 @@ class Meta:
             'condition_nodes',
             'condition_edges',
             'all_parents_must_converge',
+            'max_retries',
             'identifier',
         )
 
@@ -4282,6 +4283,7 @@ class WorkflowJobNodeSerializer(LaunchConfigurationBaseSerializer):
     condition_edges = serializers.SerializerMethodField(
         help_text=_('List of condition data for each conditional edge originating from this node; id is the target node.')
     )
+    retried_jobs = serializers.PrimaryKeyRelatedField(many=True, read_only=True)
 
     class Meta:
         model = WorkflowJobNode
@@ -4304,6 +4306,9 @@ class Meta:
             'do_not_run',
             'prior_run_succeeded',
             'prior_run_elapsed',
+            'max_retries',
+            'retry_attempts',
+            'retried_jobs',
             'identifier',
         )
 
diff --git a/awx/api/views/mixin.py b/awx/api/views/mixin.py
index 7fb00da81..30f71f260 100644
--- a/awx/api/views/mixin.py
+++ b/awx/api/views/mixin.py
@@ -41,6 +41,10 @@ def destroy(self, request, *args, **kwargs):
                 raise PermissionDenied(detail=_('Cannot delete job resource when associated workflow job is running.'))
         except self.model.unified_job_node.RelatedObjectDoesNotExist:
             pass
+        # a superseded attempt of a retried workflow node no longer has a
+        # unified_job_node, but its workflow may still be running
+        if obj.retried_workflow_nodes.filter(workflow_job__status__in=ACTIVE_STATES).exists():
+            raise PermissionDenied(detail=_('Cannot delete job resource when associated workflow job is running.'))
         # Still allow deletion of new status, because these can be manually created
         if obj.status in ACTIVE_STATES and obj.status != 'new':
             raise PermissionDenied(detail=_("Cannot delete running job resource."))
diff --git a/awx/main/migrations/0203_workflow_node_retries.py b/awx/main/migrations/0203_workflow_node_retries.py
new file mode 100644
index 000000000..8fa3aec39
--- /dev/null
+++ b/awx/main/migrations/0203_workflow_node_retries.py
@@ -0,0 +1,55 @@
+import django.core.validators
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ('main', '0202_project_webhooks'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='workflowjobtemplatenode',
+            name='max_retries',
+            field=models.PositiveIntegerField(
+                default=0,
+                validators=[django.core.validators.MaxValueValidator(100)],
+                help_text=(
+                    "Maximum number of times this node's job is automatically retried after "
+                    "failing before its failure paths are followed. Canceled jobs are never retried."
+                ),
+            ),
+        ),
+        migrations.AddField(
+            model_name='workflowjobnode',
+            name='max_retries',
+            field=models.PositiveIntegerField(
+                default=0,
+                validators=[django.core.validators.MaxValueValidator(100)],
+                help_text=(
+                    "Maximum number of times this node's job is automatically retried after "
+                    "failing before its failure paths are followed. Canceled jobs are never retried."
+                ),
+            ),
+        ),
+        migrations.AddField(
+            model_name='workflowjobnode',
+            name='retry_attempts',
+            field=models.PositiveIntegerField(
+                default=0,
+                editable=False,
+                help_text="Number of times this node's job has been automatically retried after failing.",
+            ),
+        ),
+        migrations.AddField(
+            model_name='workflowjobnode',
+            name='retried_jobs',
+            field=models.ManyToManyField(
+                blank=True,
+                editable=False,
+                help_text='Jobs from earlier attempts of this node that were superseded by an automatic retry.',
+                related_name='retried_workflow_nodes',
+                to='main.unifiedjob',
+            ),
+        ),
+    ]
diff --git a/awx/main/models/unified_jobs.py b/awx/main/models/unified_jobs.py
index 3482c04ec..e9f94b27c 100644
--- a/awx/main/models/unified_jobs.py
+++ b/awx/main/models/unified_jobs.py
@@ -1238,7 +1238,11 @@ def get_workflow_job(self):
             try:
                 return self.unified_job_node.workflow_job
             except UnifiedJob.unified_job_node.RelatedObjectDoesNotExist:
-                pass
+                # superseded attempt of a retried node; the node points at the
+                # newest attempt but this job still belongs to its workflow
+                node = self.retried_workflow_nodes.first()
+                if node is not None:
+                    return node.workflow_job
         return None
 
     @property
@@ -1247,7 +1251,9 @@ def workflow_node_id(self):
             try:
                 return self.unified_job_node.pk
             except UnifiedJob.unified_job_node.RelatedObjectDoesNotExist:
-                pass
+                node = self.retried_workflow_nodes.first()
+                if node is not None:
+                    return node.pk
         return None
 
     def get_effective_artifacts(self, **kwargs):
diff --git a/awx/main/models/workflow.py b/awx/main/models/workflow.py
index b624156d7..fce8fdd24 100644
--- a/awx/main/models/workflow.py
+++ b/awx/main/models/workflow.py
@@ -13,6 +13,7 @@
 from django.conf import settings
 from django.utils.translation import gettext_lazy as _
 from django.core.exceptions import ObjectDoesNotExist
+from django.core.validators import MaxValueValidator
 from django.utils.timezone import now, timedelta
 
 # from django import settings as tower_settings
@@ -61,6 +62,12 @@
 
 logger = logging.getLogger('awx.main.models.workflow')
 
+# Sanity ceiling for the retry budget; retries have no delay between attempts,
+# so an unbounded user value could flood the jobs table before failure paths run
+MAX_RETRIES_CEILING = 100
+# Statuses a spawned unified job cannot leave without outside action
+TERMINAL_JOB_STATUSES = ('successful', 'failed', 'error', 'canceled')
+
 
 def evaluate_artifact_condition(artifacts, artifact_key, operator, expected_value):
     """
@@ -208,6 +215,14 @@ class Meta:
     all_parents_must_converge = models.BooleanField(
         default=False, help_text=_("If enabled then the node will only run if all of the parent nodes have met the criteria to reach this node")
     )
+    max_retries = models.PositiveIntegerField(
+        default=0,
+        validators=[MaxValueValidator(MAX_RETRIES_CEILING)],
+        help_text=_(
+            "Maximum number of times this node's job is automatically retried after "
+            "failing before its failure paths are followed. Canceled jobs are never retried."
+        ),
+    )
     unified_job_template = models.ForeignKey(
         'UnifiedJobTemplate',
         related_name='%(class)ss',
@@ -239,6 +254,7 @@ def _get_workflow_job_field_names(cls):
             'credentials',
             'char_prompts',
             'all_parents_must_converge',
+            'max_retries',
             'labels',
             'instance_groups',
             'execution_environment',
@@ -291,6 +307,7 @@ class WorkflowJobTemplateNode(WorkflowNodeBase):
         'survey_passwords',
         'char_prompts',
         'all_parents_must_converge',
+        'max_retries',
         'identifier',
         'labels',
         'execution_environment',
@@ -435,6 +452,18 @@ class WorkflowJobNode(WorkflowNodeBase):
         editable=False,
         help_text=_("Elapsed time (seconds) of this node's job in the run it was carried forward from."),
     )
+    retry_attempts = models.PositiveIntegerField(
+        default=0,
+        editable=False,
+        help_text=_("Number of times this node's job has been automatically retried after failing."),
+    )
+    retried_jobs = models.ManyToManyField(
+        'UnifiedJob',
+        related_name='retried_workflow_nodes',
+        blank=True,
+        editable=False,
+        help_text=_('Jobs from earlier attempts of this node that were superseded by an automatic retry.'),
+    )
     identifier = models.CharField(
         max_length=512,
         blank=True,  # blank denotes pre-migration job nodes
@@ -463,6 +492,35 @@ class Meta:
     def event_processing_finished(self):
         return True
 
+    def retry_pending(self):
+        """True when this node's job failed but automatic retries remain, in which
+        case the scheduler treats the node as if its job were still running and
+        respawns it instead of following its failure paths. Canceled jobs are
+        explicit user actions and are never retried."""
+        if not self.job or self.job.status not in ('failed', 'error'):
+            return False
+        # a retry cannot spawn once the template is gone (deleted mid run);
+        # without this the workflow would wait on the retry forever
+        if self.unified_job_template_id is None:
+            return False
+        if self.retry_attempts >= self.max_retries:
+            return False
+        # approvals are human decisions, never retried automatically; resolve
+        # the class through the polymorphic ctype because a base UnifiedJob
+        # instance (e.g. loaded via select_related('job')) defeats isinstance
+        job_cls = self.job.get_real_instance_class() if self.job.polymorphic_ctype_id else type(self.job)
+        if issubclass(job_cls, WorkflowApproval):
+            return False
+        return True
+
+    def job_finished(self):
+        """The node's job reached a terminal status with no automatic retry pending."""
+        return bool(self.job and self.job.status in TERMINAL_JOB_STATUSES and not self.retry_pending())
+
+    def finally_failed(self):
+        """The node's job failed for good: a failure status and no retry pending."""
+        return bool(self.job and self.job.status in ('failed', 'error', 'canceled') and not self.retry_pending())
+
     def get_absolute_url(self, request=None):
         return reverse('api:workflow_job_node_detail', kwargs={'pk': self.pk}, request=request)
 
diff --git a/awx/main/scheduler/dag_workflow.py b/awx/main/scheduler/dag_workflow.py
index 56b51f06b..91e16fa5d 100644
--- a/awx/main/scheduler/dag_workflow.py
+++ b/awx/main/scheduler/dag_workflow.py
@@ -103,8 +103,9 @@ def _are_relevant_parents_finished(self, node):
             # do_not_run is False, node might still run a job and thus blocks children
             elif not p.job:
                 return False
-            # Node decidedly got a job; check if job is done
-            elif p.job and p.job.status not in ['successful', 'failed', 'error', 'canceled']:
+            # Node decidedly got a job; check if job is done. A failed job with
+            # automatic retries left is not done, a new attempt will be spawned.
+            elif p.job and not p.job_finished():
                 return False
         return True
 
@@ -116,8 +117,9 @@ def _all_parents_met_convergence_criteria(self, node):
         parent_nodes = [p['node_object'] for p in self.get_parents(obj)]
         for p in parent_nodes:
             status = None
-            # node has a status
-            if p.job and p.job.status in ["successful", "failed"]:
+            # node has a status; a failed job with automatic retries left has
+            # no final status yet
+            if p.job_finished() and p.job.status in ["successful", "failed"]:
                 if p.job and p.job.status == "successful":
                     status = "success_nodes"
                 elif p.job and p.job.status == "failed":
@@ -157,7 +159,11 @@ def bfs_nodes_to_run(self):
                     self.get_children(obj, 'success_nodes') + self.get_children(obj, 'always_nodes') + self._passing_condition_children(obj, 'success')
                 )
             elif obj.job:
-                if obj.job.status in ['failed', 'error', 'canceled']:
+                # a failed job with automatic retries left is respawned instead
+                # of following its failure paths
+                if obj.retry_pending():
+                    nodes_found.append(n)
+                elif obj.job.status in ['failed', 'error', 'canceled']:
                     nodes.extend(
                         self.get_children(obj, 'failure_nodes') + self.get_children(obj, 'always_nodes') + self._passing_condition_children(obj, 'failure')
                     )
@@ -198,7 +204,7 @@ def is_workflow_done(self):
             obj = node['node_object']
             if obj.do_not_run is False and not obj.prior_run_succeeded and not obj.job and obj.unified_job_template:
                 return False
-            elif obj.job and obj.job.status not in ['successful', 'failed', 'canceled', 'error']:
+            elif obj.job and not obj.job_finished():
                 return False
         return True
 
@@ -212,7 +218,7 @@ def has_workflow_failed(self):
             obj = node['node_object']
             if obj.do_not_run is False and not obj.prior_run_succeeded and obj.unified_job_template is None:
                 failed_nodes.append(node)
-            elif obj.job and obj.job.status in ['failed', 'canceled', 'error']:
+            elif obj.finally_failed():
                 failed_nodes.append(node)
 
         for node in failed_nodes:
@@ -262,7 +268,7 @@ def has_workflow_failed(self):
 
     def _are_all_nodes_dnr_decided(self, workflow_nodes):
         for n in workflow_nodes:
-            if n.do_not_run is False and not n.prior_run_succeeded and not n.job and n.unified_job_template:
+            if n.do_not_run is False and not n.prior_run_succeeded and (not n.job or n.retry_pending()) and n.unified_job_template:
                 return False
         return True
 
@@ -287,7 +293,10 @@ def _should_mark_node_dnr(self, node, parent_nodes):
                 ):
                     return False
             elif p.job:
-                if p.job.status == 'successful':
+                # a failed job with automatic retries left is not decided yet
+                if p.retry_pending():
+                    return False
+                elif p.job.status == 'successful':
                     if node in (self.get_children(p, 'success_nodes') + self.get_children(p, 'always_nodes')) or self._edge_condition_passes(
                         p, node['node_object'], 'success'
                     ):
diff --git a/awx/main/scheduler/task_manager.py b/awx/main/scheduler/task_manager.py
index 7936d0b57..918596727 100644
--- a/awx/main/scheduler/task_manager.py
+++ b/awx/main/scheduler/task_manager.py
@@ -225,11 +225,29 @@ def spawn_workflow_graph_jobs(self):
                 for spawn_node in spawn_nodes:
                     if spawn_node.unified_job_template is None:
                         continue
+                    # a node that already has a (failed) job is being automatically retried
+                    superseded_job_id = spawn_node.job_id
+                    is_retry = superseded_job_id is not None
                     kv = spawn_node.get_job_kwargs()
                     job = spawn_node.unified_job_template.create_unified_job(**kv)
                     spawn_node.job = job
+                    if is_retry:
+                        spawn_node.retry_attempts += 1
                     spawn_node.save()
-                    logger.debug('Spawned %s in %s for node %s', job.log_format, workflow_job.log_format, spawn_node.pk)
+                    if is_retry:
+                        # keep the superseded attempt linked to the node so it stays
+                        # protected from deletion and traceable to this workflow
+                        spawn_node.retried_jobs.add(superseded_job_id)
+                        logger.info(
+                            'Spawned %s in %s for node %s, retry %s of %s after failed attempt',
+                            job.log_format,
+                            workflow_job.log_format,
+                            spawn_node.pk,
+                            spawn_node.retry_attempts,
+                            spawn_node.max_retries,
+                        )
+                    else:
+                        logger.debug('Spawned %s in %s for node %s', job.log_format, workflow_job.log_format, spawn_node.pk)
                     can_start = True
                     if isinstance(spawn_node.unified_job_template, WorkflowJobTemplate):
                         workflow_ancestors = job.get_ancestor_workflows()
@@ -267,6 +285,12 @@ def spawn_workflow_graph_jobs(self):
                                 "Job spawned from workflow could not start because it was not in the right state or required manual credentials"
                             )
                     if not can_start:
+                        # the job never started for a structural reason (missing
+                        # resources, recursion, credentials); retrying cannot help,
+                        # so exhaust the node's automatic retry budget
+                        if spawn_node.retry_attempts < spawn_node.max_retries:
+                            spawn_node.retry_attempts = spawn_node.max_retries
+                            spawn_node.save(update_fields=['retry_attempts'])
                         job.status = 'failed'
                         job.save(update_fields=['status', 'job_explanation'])
                         job.websocket_emit_status('failed')
diff --git a/awx/main/tests/functional/api/test_workflow_node.py b/awx/main/tests/functional/api/test_workflow_node.py
index 5f30f6afd..96d3f9a8a 100644
--- a/awx/main/tests/functional/api/test_workflow_node.py
+++ b/awx/main/tests/functional/api/test_workflow_node.py
@@ -52,6 +52,44 @@ def test_node_accepts_prompted_fields(inventory, project, workflow_job_template,
     post(url, {'unified_job_template': job_template.pk, 'limit': 'webservers'}, user=admin_user, expect=201)
 
 
+@pytest.mark.django_db
+def test_node_max_retries_settable(inventory, project, workflow_job_template, post, admin_user):
+    job_template = JobTemplate.objects.create(inventory=inventory, project=project, playbook='helloworld.yml')
+    url = reverse('api:workflow_job_template_workflow_nodes_list', kwargs={'pk': workflow_job_template.pk})
+    r = post(url, {'unified_job_template': job_template.pk, 'max_retries': 2}, user=admin_user, expect=201)
+    node = WorkflowJobTemplateNode.objects.get(pk=r.data['id'])
+    assert node.max_retries == 2
+
+
+@pytest.mark.django_db
+def test_node_max_retries_rejects_out_of_range_values(inventory, project, workflow_job_template, post, admin_user):
+    # retries have no delay between attempts, so the budget is capped to keep a
+    # misconfigured node from flooding the jobs table before failure paths run
+    job_template = JobTemplate.objects.create(inventory=inventory, project=project, playbook='helloworld.yml')
+    url = reverse('api:workflow_job_template_workflow_nodes_list', kwargs={'pk': workflow_job_template.pk})
+    r = post(url, {'unified_job_template': job_template.pk, 'max_retries': 101}, user=admin_user, expect=400)
+    assert 'max_retries' in r.data
+    r = post(url, {'unified_job_template': job_template.pk, 'max_retries': -1}, user=admin_user, expect=400)
+    assert 'max_retries' in r.data
+
+
+@pytest.mark.django_db
+def test_superseded_retry_attempt_protected_while_workflow_runs(delete, admin_user, inventory, project, workflow_job_template):
+    # a job superseded by an automatic retry loses its unified_job_node link,
+    # but must stay undeletable while its workflow is still running
+    job_template = JobTemplate.objects.create(name='retried-jt', inventory=inventory, project=project, playbook='helloworld.yml')
+    wfj = WorkflowJob.objects.create(workflow_job_template=workflow_job_template, status='running')
+    node = wfj.workflow_nodes.create(unified_job_template=job_template)
+    old_job = job_template.create_job()
+    old_job.status = 'failed'
+    old_job.launch_type = 'workflow'
+    old_job.save()
+    node.retried_jobs.add(old_job)
+
+    url = reverse('api:job_detail', kwargs={'pk': old_job.pk})
+    delete(url, user=admin_user, expect=403)
+
+
 @pytest.mark.django_db
 @pytest.mark.parametrize(
     "field_name, field_value",
diff --git a/awx/main/tests/functional/models/test_workflow.py b/awx/main/tests/functional/models/test_workflow.py
index 8a05f66a9..19046a671 100644
--- a/awx/main/tests/functional/models/test_workflow.py
+++ b/awx/main/tests/functional/models/test_workflow.py
@@ -418,6 +418,59 @@ def test_relaunch_from_failed_reruns_canceled_node(self):
         dag.mark_dnr_nodes()
         assert canceled_node in dag.bfs_nodes_to_run()
 
+    def test_failed_node_with_retries_is_respawned_not_failed(self):
+        wfj = self.workflow_job(states=['failed', None, None, None, None])
+        nodes = list(wfj.workflow_nodes.order_by('id'))
+        nodes[0].max_retries = 1
+        nodes[0].save()
+
+        dag = WorkflowDAG(workflow_job=wfj)
+        assert dag.mark_dnr_nodes() == [], "nothing is decided while a retry is pending"
+        assert dag.is_workflow_done() is False
+        spawn_nodes = dag.bfs_nodes_to_run()
+        assert [n.id for n in spawn_nodes] == [nodes[0].id], "the failed node itself must be respawned"
+
+        # simulate the task manager retrying the node and the retry failing too
+        node0 = spawn_nodes[0]
+        node0.job = node0.unified_job_template.create_job()
+        node0.retry_attempts += 1
+        node0.save()
+        node0.job.status = 'failed'
+        node0.job.save()
+
+        # retries exhausted: the failure path finally runs
+        dag = WorkflowDAG(workflow_job=wfj)
+        dnr_ids = {n.id for n in dag.mark_dnr_nodes()}
+        assert dnr_ids == {nodes[1].id, nodes[2].id}
+        assert [n.id for n in dag.bfs_nodes_to_run()] == [nodes[3].id]
+
+    def test_deleted_template_finishes_workflow_despite_pending_retry(self):
+        # deleting the JT while a retry is pending must not hang the workflow:
+        # the retry can never spawn, so the failure becomes final
+        wfj = self.workflow_job(states=['failed', None, None, None, None])
+        wfj.workflow_nodes.update(max_retries=5)
+        for jt in JobTemplate.objects.all():
+            jt.delete()
+
+        dag = WorkflowDAG(workflow_job=wfj)
+        dag.mark_dnr_nodes()
+        assert dag.bfs_nodes_to_run() == [], "no retry can spawn without a template"
+        assert dag.is_workflow_done() is True
+        has_failed, reason = dag.has_workflow_failed()
+        assert has_failed is True
+
+    def test_node_without_retries_fails_right_away(self):
+        # the retry budget is per node: other nodes having one does not delay
+        # the failure paths of a node with max_retries 0
+        wfj = self.workflow_job(states=['failed', None, None, None, None])
+        nodes = list(wfj.workflow_nodes.order_by('id'))
+        wfj.workflow_nodes.exclude(id=nodes[0].id).update(max_retries=3)
+
+        dag = WorkflowDAG(workflow_job=wfj)
+        dnr_ids = {n.id for n in dag.mark_dnr_nodes()}
+        assert dnr_ids == {nodes[1].id, nodes[2].id}, "no retry: the failure is final right away"
+        assert [n.id for n in dag.bfs_nodes_to_run()] == [nodes[3].id]
+
 
 @pytest.mark.django_db
 class TestWorkflowDNR:
diff --git a/awx/main/tests/functional/task_management/test_scheduler.py b/awx/main/tests/functional/task_management/test_scheduler.py
index 8d8ace682..5af505fd7 100644
--- a/awx/main/tests/functional/task_management/test_scheduler.py
+++ b/awx/main/tests/functional/task_management/test_scheduler.py
@@ -76,6 +76,94 @@ def test_task_manager_workflow_rescheduling(self, job_template_factory, inventor
         # no further action is necessary, so rescheduling should not happen
         self.run_tm(WorkflowManager(), [mock.call('successful')])
 
+    def test_workflow_node_automatic_retry(self, inventory, project, controlplane_instance_group):
+        jt = JobTemplate.objects.create(inventory=inventory, project=project, playbook='helloworld.yml', name='flaky-jt')
+        wfjt = WorkflowJobTemplate.objects.create(name='retry-wf')
+        wfjt.workflow_nodes.create(unified_job_template=jt, max_retries=1)
+        wj = wfjt.create_unified_job()
+        wj.signal_start()
+
+        self.run_tm(TaskManager(), [mock.call('running')])
+        self.run_tm(WorkflowManager(), [mock.call('pending')])
+
+        node = wj.workflow_nodes.first()
+        first_job = node.job
+        assert node.max_retries == 1, "max_retries must be copied from the template node"
+        assert node.retry_attempts == 0
+        first_job.status = 'failed'
+        first_job.save()
+
+        # the workflow does not fail; the node is respawned with a new job
+        self.run_tm(WorkflowManager(), [mock.call('pending')])
+        wj.refresh_from_db()
+        assert wj.status == 'running'
+        node.refresh_from_db()
+        assert node.retry_attempts == 1
+        assert node.job_id != first_job.id
+        assert jt.jobs.count() == 2
+
+        # the superseded attempt stays linked to its workflow through the node
+        assert list(node.retried_jobs.all()) == [first_job]
+        assert first_job.get_workflow_job() == wj
+        assert first_job.workflow_node_id == node.pk
+
+        # the retry succeeds and the workflow finishes successfully
+        node.job.status = 'successful'
+        node.job.save()
+        self.run_tm(WorkflowManager(), [mock.call('successful')])
+        wj.refresh_from_db()
+        assert wj.status == 'successful'
+
+    def test_workflow_node_retries_exhausted(self, inventory, project, controlplane_instance_group):
+        jt = JobTemplate.objects.create(inventory=inventory, project=project, playbook='helloworld.yml', name='flaky-jt')
+        wfjt = WorkflowJobTemplate.objects.create(name='retry-wf')
+        wfjt.workflow_nodes.create(unified_job_template=jt, max_retries=1)
+        wj = wfjt.create_unified_job()
+        wj.signal_start()
+
+        self.run_tm(TaskManager(), [mock.call('running')])
+        self.run_tm(WorkflowManager(), [mock.call('pending')])
+
+        node = wj.workflow_nodes.first()
+        node.job.status = 'failed'
+        node.job.save()
+
+        # first failure: retried
+        self.run_tm(WorkflowManager(), [mock.call('pending')])
+        node.refresh_from_db()
+        assert node.retry_attempts == 1
+
+        # second failure: retries exhausted, workflow fails
+        node.job.status = 'failed'
+        node.job.save()
+        self.run_tm(WorkflowManager(), [mock.call('failed')])
+        wj.refresh_from_db()
+        assert wj.status == 'failed'
+        node.refresh_from_db()
+        assert node.retry_attempts == 1
+        assert jt.jobs.count() == 2
+
+    def test_workflow_node_never_started_job_is_not_retried(self, inventory, project, controlplane_instance_group):
+        # a job that cannot start for a structural reason (here: no project or
+        # inventory) must exhaust the retry budget instead of hot-looping
+        jt = JobTemplate.objects.create(name='resourceless-jt')
+        wfjt = WorkflowJobTemplate.objects.create(name='retry-wf')
+        wfjt.workflow_nodes.create(unified_job_template=jt, max_retries=5)
+        wj = wfjt.create_unified_job()
+        wj.signal_start()
+
+        self.run_tm(TaskManager(), [mock.call('running')])
+        # first pass: the job is spawned and fails to start
+        self.run_tm(WorkflowManager(), [mock.call('failed')])
+        # second pass: no retry is spawned, the workflow fails
+        self.run_tm(WorkflowManager(), [mock.call('failed')])
+
+        wj.refresh_from_db()
+        assert wj.status == 'failed'
+        node = wj.workflow_nodes.first()
+        assert node.retry_attempts == node.max_retries
+        assert jt.jobs.count() == 1
+
     def test_task_manager_workflow_workflow_rescheduling(self, controlplane_instance_group):
         wfjts = [WorkflowJobTemplate.objects.create(name='foo')]
         for i in range(5):
diff --git a/awx/main/tests/unit/models/test_workflow_unit.py b/awx/main/tests/unit/models/test_workflow_unit.py
index 21233d27d..26230982a 100644
--- a/awx/main/tests/unit/models/test_workflow_unit.py
+++ b/awx/main/tests/unit/models/test_workflow_unit.py
@@ -1,8 +1,8 @@
 import pytest
 
-from awx.main.models.jobs import JobTemplate
+from awx.main.models.jobs import JobTemplate, Job
 from awx.main.models import Inventory, CredentialType, Credential, Project
-from awx.main.models.workflow import WorkflowJobTemplate, WorkflowJobTemplateNode, WorkflowJob, WorkflowJobNode, merge_stats_artifacts
+from awx.main.models.workflow import WorkflowJobTemplate, WorkflowJobTemplateNode, WorkflowJob, WorkflowJobNode, WorkflowApproval, merge_stats_artifacts
 from unittest import mock
 
 
@@ -152,6 +152,63 @@ def test_node_getter_and_setters():
     assert node.job_type == 'check'
 
 
+class TestNodeMaxRetries:
+    def node(self, max_retries=0):
+        wfj = WorkflowJob(name='workflow', status='running')
+        # a template id must be present for a retry to be spawnable
+        return WorkflowJobNode(workflow_job=wfj, max_retries=max_retries, unified_job_template_id=1)
+
+    def test_default_is_no_retries(self):
+        assert self.node().max_retries == 0
+
+    def test_retry_pending_for_failed_job_with_retries_left(self):
+        node = self.node(max_retries=1)
+        node.job = Job(status='failed')
+        assert node.retry_pending() is True
+
+    def test_retry_pending_false_when_exhausted(self):
+        node = self.node(max_retries=1)
+        node.job = Job(status='failed')
+        node.retry_attempts = 1
+        assert node.retry_pending() is False
+
+    def test_retry_pending_false_without_job_or_on_success(self):
+        node = self.node(max_retries=1)
+        assert node.retry_pending() is False
+        node.job = Job(status='successful')
+        assert node.retry_pending() is False
+
+    def test_canceled_job_is_never_retried(self):
+        node = self.node(max_retries=1)
+        node.job = Job(status='canceled')
+        assert node.retry_pending() is False
+
+    def test_approvals_are_never_retried(self):
+        node = self.node(max_retries=1)
+        node.job = WorkflowApproval(status='failed')
+        assert node.retry_pending() is False
+
+    def test_deleted_template_disqualifies_retry(self):
+        # a retry can never spawn without a template; treating it as pending
+        # would leave the workflow running forever
+        node = self.node(max_retries=1)
+        node.unified_job_template_id = None
+        node.job = Job(status='failed')
+        assert node.retry_pending() is False
+
+    def test_job_finished_and_finally_failed_track_retries(self):
+        node = self.node(max_retries=1)
+        node.job = Job(status='failed')
+        assert node.job_finished() is False
+        assert node.finally_failed() is False
+        node.retry_attempts = 1
+        assert node.job_finished() is True
+        assert node.finally_failed() is True
+        node.job = Job(status='successful')
+        assert node.job_finished() is True
+        assert node.finally_failed() is False
+
+
 @pytest.mark.django_db
 class TestWorkflowJobCreate:
     def test_create_no_prompts(self, wfjt_node_no_prompts, workflow_job_unit, mocker):
@@ -160,6 +217,7 @@ def test_create_no_prompts(self, wfjt_node_no_prompts, workflow_job_unit, mocker
             wfjt_node_no_prompts.create_workflow_job_node(workflow_job=workflow_job_unit)
             mock_create.assert_called_once_with(
                 all_parents_must_converge=False,
+                max_retries=0,
                 extra_data={},
                 survey_passwords={},
                 char_prompts=wfjt_node_no_prompts.char_prompts,
@@ -176,6 +234,7 @@ def test_create_with_prompts(self, wfjt_node_with_prompts, workflow_job_unit, cr
             wfjt_node_with_prompts.create_workflow_job_node(workflow_job=workflow_job_unit)
             mock_create.assert_called_once_with(
                 all_parents_must_converge=False,
+                max_retries=0,
                 extra_data={},
                 survey_passwords={},
                 char_prompts=wfjt_node_with_prompts.char_prompts,
diff --git a/awx/main/tests/unit/scheduler/test_dag_workflow.py b/awx/main/tests/unit/scheduler/test_dag_workflow.py
index e4cf48aec..84a76440d 100644
--- a/awx/main/tests/unit/scheduler/test_dag_workflow.py
+++ b/awx/main/tests/unit/scheduler/test_dag_workflow.py
@@ -24,13 +24,25 @@ def get_effective_artifacts(self, **kwargs):
 
 
 class WorkflowNode(object):
-    def __init__(self, id=None, job=None, do_not_run=False, unified_job_template=None, prior_run_succeeded=False):
+    def __init__(self, id=None, job=None, do_not_run=False, unified_job_template=None, prior_run_succeeded=False, max_retries=0, retry_attempts=0):
         self.id = id if id is not None else uuid.uuid4()
         self.job = job
         self.do_not_run = do_not_run
         self.prior_run_succeeded = prior_run_succeeded
         self.unified_job_template = unified_job_template
         self.all_parents_must_converge = False
+        self.max_retries = max_retries
+        self.retry_attempts = retry_attempts
+
+    # the three methods below mirror WorkflowJobNode
+    def retry_pending(self):
+        return bool(self.job and self.job.status in ('failed', 'error') and self.unified_job_template is not None and self.retry_attempts < self.max_retries)
+
+    def job_finished(self):
+        return bool(self.job and self.job.status in ('successful', 'failed', 'error', 'canceled') and not self.retry_pending())
+
+    def finally_failed(self):
+        return bool(self.job and self.job.status in ('failed', 'error', 'canceled') and not self.retry_pending())
 
 
 @pytest.fixture
@@ -794,6 +806,116 @@ def test_deleted_ujt_parent_traverses_passing_failure_condition(self, condition_
         assert g.has_workflow_failed() == (False, None)
 
 
+class TestNodeRetry:
+    @pytest.fixture
+    def workflow_dag_retry(self, wf_node_generator):
+        g = WorkflowDAG()
+        nodes = [wf_node_generator() for i in range(3)]
+        for n in nodes:
+            g.add_node(n)
+        r'''
+               0
+               |\ F
+               | \
+              S|  1
+               |
+               2
+        '''
+        g.add_edge(nodes[0], nodes[1], "failure_nodes")
+        g.add_edge(nodes[0], nodes[2], "success_nodes")
+        return (g, nodes)
+
+    def test_failed_node_with_retries_left_is_respawned(self, workflow_dag_retry):
+        g, nodes = workflow_dag_retry
+        nodes[0].max_retries = 2
+        nodes[0].job = Job(status='failed')
+
+        assert g.mark_dnr_nodes() == [], "children must stay undecided while a retry is pending"
+        assert g.bfs_nodes_to_run() == [nodes[0]], "the failed node itself should be respawned"
+        assert g.is_workflow_done() is False
+        has_failed, reason = g.has_workflow_failed()
+        assert has_failed is False
+        assert reason is None
+
+    def test_exhausted_retries_follow_failure_path(self, workflow_dag_retry):
+        g, nodes = workflow_dag_retry
+        nodes[0].max_retries = 2
+        nodes[0].retry_attempts = 2
+        nodes[0].job = Job(status='failed')
+
+        dnr_nodes = g.mark_dnr_nodes()
+        assert dnr_nodes == [nodes[2]], "success child should be marked DNR once retries are exhausted"
+        assert g.bfs_nodes_to_run() == [nodes[1]], "failure child should run once retries are exhausted"
+
+    def test_retry_after_success_never_pending(self, workflow_dag_retry):
+        g, nodes = workflow_dag_retry
+        nodes[0].max_retries = 2
+        nodes[0].job = Job(status='successful')
+
+        dnr_nodes = g.mark_dnr_nodes()
+        assert dnr_nodes == [nodes[1]]
+        assert g.bfs_nodes_to_run() == [nodes[2]]
+
+    def test_canceled_job_is_not_retried(self, workflow_dag_retry):
+        g, nodes = workflow_dag_retry
+        nodes[0].max_retries = 2
+        nodes[0].job = Job(status='canceled')
+
+        dnr_nodes = g.mark_dnr_nodes()
+        assert dnr_nodes == [nodes[2]]
+        assert g.bfs_nodes_to_run() == [nodes[1]], "canceled node should follow its failure path, not retry"
+
+    def test_error_job_is_retried(self, workflow_dag_retry):
+        g, nodes = workflow_dag_retry
+        nodes[0].max_retries = 1
+        nodes[0].job = Job(status='error')
+
+        assert g.bfs_nodes_to_run() == [nodes[0]]
+        assert g.is_workflow_done() is False
+
+    def test_retry_pending_leaf_does_not_fail_workflow(self, wf_node_generator):
+        g = WorkflowDAG()
+        node = wf_node_generator(max_retries=1)
+        g.add_node(node)
+        node.job = Job(status='failed')
+
+        assert g.is_workflow_done() is False
+        has_failed, reason = g.has_workflow_failed()
+        assert has_failed is False
+
+        node.retry_attempts = 1
+        assert g.is_workflow_done() is True
+        has_failed, reason = g.has_workflow_failed()
+        assert has_failed is True
+
+    def test_retry_pending_parent_blocks_convergence(self, wf_node_generator):
+        g = WorkflowDAG()
+        nodes = [wf_node_generator() for i in range(3)]
+        for n in nodes:
+            g.add_node(n)
+        r'''
+               0    1
+              S \  / S
+                 \/
+                 2
+        '''
+        g.add_edge(nodes[0], nodes[2], "success_nodes")
+        g.add_edge(nodes[1], nodes[2], "success_nodes")
+        nodes[2].all_parents_must_converge = True
+        nodes[0].job = Job(status='successful')
+        nodes[1].max_retries = 1
+        nodes[1].job = Job(status='failed')
+
+        assert g.mark_dnr_nodes() == [], "convergence child undecided while parent retry is pending"
+        assert g.bfs_nodes_to_run() == [nodes[1]], "only the retried parent should spawn"
+
+        # retry succeeded: convergence node runs
+        nodes[1].retry_attempts = 1
+        nodes[1].job = Job(status='successful')
+        assert g.mark_dnr_nodes() == []
+        assert g.bfs_nodes_to_run() == [nodes[2]]
+
+
 @pytest.mark.skip(reason="Run manually to re-generate doc images")
 class TestDocsExample:
     @pytest.fixture
diff --git a/awx/ui/src/components/PromptDetail/PromptDetail.js b/awx/ui/src/components/PromptDetail/PromptDetail.js
index 543eb0844..4f189a78f 100644
--- a/awx/ui/src/components/PromptDetail/PromptDetail.js
+++ b/awx/ui/src/components/PromptDetail/PromptDetail.js
@@ -140,6 +140,15 @@ function PromptDetail({
             }
           />
         )}
+        {workflowNode &&
+          details.unified_job_type !== 'workflow_approval' &&
+          details.type !== 'workflow_approval_template' && (
+            <Detail
+              label={t`Max Retries`}
+              dataCy="prompt-detail-max-retries"
+              value={workflowNode?.max_retries || 0}
+            />
+          )}
         <Detail
           label={t`Timeout`}
           dataCy="prompt-detail-timeout"
diff --git a/awx/ui/src/components/Workflow/workflowReducer.js b/awx/ui/src/components/Workflow/workflowReducer.js
index def66544e..0e0720197 100644
--- a/awx/ui/src/components/Workflow/workflowReducer.js
+++ b/awx/ui/src/components/Workflow/workflowReducer.js
@@ -189,6 +189,7 @@ function createNode(state, node) {
     isInvalidLinkTarget: false,
     promptValues: node.promptValues,
     all_parents_must_converge: node.all_parents_must_converge,
+    max_retries: node.max_retries,
     identifier: node.identifier,
   });
 
@@ -699,12 +700,14 @@ function updateNode(state, editedNode) {
     launchConfig,
     promptValues,
     all_parents_must_converge,
+    max_retries,
     identifier,
   } = editedNode;
   const newNodes = [...nodes];
 
   const matchingNode = newNodes.find((node) => node.id === nodeToEdit.id);
   matchingNode.all_parents_must_converge = all_parents_must_converge;
+  matchingNode.max_retries = max_retries;
   matchingNode.fullUnifiedJobTemplate = nodeResource;
   matchingNode.isEdited = true;
   matchingNode.launchConfig = launchConfig;
diff --git a/awx/ui/src/screens/Template/WorkflowJobTemplateVisualizer/Modals/NodeModals/NodeAddModal.js b/awx/ui/src/screens/Template/WorkflowJobTemplateVisualizer/Modals/NodeModals/NodeAddModal.js
index 0e8f566e4..9ab283257 100644
--- a/awx/ui/src/screens/Template/WorkflowJobTemplateVisualizer/Modals/NodeModals/NodeAddModal.js
+++ b/awx/ui/src/screens/Template/WorkflowJobTemplateVisualizer/Modals/NodeModals/NodeAddModal.js
@@ -26,6 +26,7 @@ function NodeAddModal() {
       linkConditionOperator,
       linkConditionExpectedValue,
       convergence,
+      maxRetries,
       identifier,
     } = values;
 
@@ -42,6 +43,7 @@ function NodeAddModal() {
     const node = {
       linkType,
       all_parents_must_converge: convergence === 'all',
+      max_retries: Number(maxRetries) || 0,
       identifier,
     };
 
@@ -55,6 +57,7 @@ function NodeAddModal() {
     }
 
     delete values.convergence;
+    delete values.maxRetries;
 
     delete values.linkType;
     delete values.linkConditionTrigger;
diff --git a/awx/ui/src/screens/Template/WorkflowJobTemplateVisualizer/Modals/NodeModals/NodeAddModal.test.js b/awx/ui/src/screens/Template/WorkflowJobTemplateVisualizer/Modals/NodeModals/NodeAddModal.test.js
index 7b9dc8d07..cf0d71ba4 100644
--- a/awx/ui/src/screens/Template/WorkflowJobTemplateVisualizer/Modals/NodeModals/NodeAddModal.test.js
+++ b/awx/ui/src/screens/Template/WorkflowJobTemplateVisualizer/Modals/NodeModals/NodeAddModal.test.js
@@ -44,6 +44,7 @@ describe('NodeAddModal', () => {
     expect(dispatch).toHaveBeenCalledWith({
       node: {
         all_parents_must_converge: false,
+        max_retries: 0,
         linkType: 'success',
         nodeResource: {
           id: 448,
diff --git a/awx/ui/src/screens/Template/WorkflowJobTemplateVisualizer/Modals/NodeModals/NodeEditModal.js b/awx/ui/src/screens/Template/WorkflowJobTemplateVisualizer/Modals/NodeModals/NodeEditModal.js
index 0c9a587c8..e5da21a77 100644
--- a/awx/ui/src/screens/Template/WorkflowJobTemplateVisualizer/Modals/NodeModals/NodeEditModal.js
+++ b/awx/ui/src/screens/Template/WorkflowJobTemplateVisualizer/Modals/NodeModals/NodeEditModal.js
@@ -24,6 +24,7 @@ function NodeEditModal() {
       timeoutMinutes,
       timeoutSeconds,
       convergence,
+      maxRetries,
       identifier,
       ...rest
     } = values;
@@ -31,6 +32,7 @@ function NodeEditModal() {
     if (values.nodeType === 'workflow_approval_template') {
       node = {
         all_parents_must_converge: convergence === 'all',
+        max_retries: 0,
         nodeResource: {
           description: approvalDescription,
           name: approvalName,
@@ -43,6 +45,7 @@ function NodeEditModal() {
       node = {
         nodeResource,
         all_parents_must_converge: convergence === 'all',
+        max_retries: Number(maxRetries) || 0,
         identifier,
       };
       if (nodeType === 'job_template' || nodeType === 'workflow_job_template') {
diff --git a/awx/ui/src/screens/Template/WorkflowJobTemplateVisualizer/Modals/NodeModals/NodeEditModal.test.js b/awx/ui/src/screens/Template/WorkflowJobTemplateVisualizer/Modals/NodeModals/NodeEditModal.test.js
index 95d490f8d..d1a5e315b 100644
--- a/awx/ui/src/screens/Template/WorkflowJobTemplateVisualizer/Modals/NodeModals/NodeEditModal.test.js
+++ b/awx/ui/src/screens/Template/WorkflowJobTemplateVisualizer/Modals/NodeModals/NodeEditModal.test.js
@@ -75,6 +75,7 @@ describe('NodeEditModal', () => {
     expect(dispatch).toHaveBeenCalledWith({
       node: {
         all_parents_must_converge: false,
+        max_retries: 0,
         nodeResource: { id: 448, name: 'Test JT', type: 'job_template' },
       },
       type: 'UPDATE_NODE',
diff --git a/awx/ui/src/screens/Template/WorkflowJobTemplateVisualizer/Modals/NodeModals/NodeModal.js b/awx/ui/src/screens/Template/WorkflowJobTemplateVisualizer/Modals/NodeModals/NodeModal.js
index 506f7d473..355c5de84 100644
--- a/awx/ui/src/screens/Template/WorkflowJobTemplateVisualizer/Modals/NodeModals/NodeModal.js
+++ b/awx/ui/src/screens/Template/WorkflowJobTemplateVisualizer/Modals/NodeModals/NodeModal.js
@@ -415,6 +415,7 @@ const NodeModal = ({ onSave, askLinkType, title }) => {
         timeoutMinutes: 0,
         timeoutSeconds: 0,
         convergence: 'any',
+        maxRetries: nodeToEdit?.max_retries || 0,
         linkType: 'success',
         linkConditionTrigger: 'success',
         linkConditionArtifactKey: '',
diff --git a/awx/ui/src/screens/Template/WorkflowJobTemplateVisualizer/Modals/NodeModals/NodeModal.test.js b/awx/ui/src/screens/Template/WorkflowJobTemplateVisualizer/Modals/NodeModals/NodeModal.test.js
index 9c7378eb0..1661520d1 100644
--- a/awx/ui/src/screens/Template/WorkflowJobTemplateVisualizer/Modals/NodeModals/NodeModal.test.js
+++ b/awx/ui/src/screens/Template/WorkflowJobTemplateVisualizer/Modals/NodeModals/NodeModal.test.js
@@ -540,6 +540,7 @@ describe('NodeModal', () => {
       expect(onSave).toHaveBeenCalledWith(
         {
           convergence: 'any',
+          maxRetries: 0,
           identifier: '',
           linkType: 'always',
           linkConditionTrigger: 'success',
@@ -585,6 +586,7 @@ describe('NodeModal', () => {
       expect(onSave).toHaveBeenCalledWith(
         {
           convergence: 'any',
+          maxRetries: 0,
           identifier: '',
           linkType: 'failure',
           linkConditionTrigger: 'success',
@@ -627,6 +629,7 @@ describe('NodeModal', () => {
       expect(onSave).toHaveBeenCalledWith(
         {
           convergence: 'any',
+          maxRetries: 0,
           identifier: '',
           linkType: 'failure',
           linkConditionTrigger: 'success',
@@ -672,6 +675,7 @@ describe('NodeModal', () => {
       expect(onSave).toHaveBeenCalledWith(
         {
           convergence: 'any',
+          maxRetries: 0,
           identifier: '',
           linkType: 'success',
           linkConditionTrigger: 'success',
@@ -751,6 +755,7 @@ describe('NodeModal', () => {
       expect(onSave).toHaveBeenCalledWith(
         {
           convergence: 'any',
+          maxRetries: 0,
           approvalDescription: 'Test Approval Description',
           approvalName: 'Test Approval',
           identifier: '',
@@ -860,6 +865,7 @@ describe('Edit existing node', () => {
       expect(onSave).toHaveBeenCalledWith(
         {
           convergence: 'any',
+          maxRetries: 0,
           identifier: 'Foo',
           approvalDescription: 'Test Approval Description',
           approvalName: 'Test Approval',
@@ -936,6 +942,7 @@ describe('Edit existing node', () => {
       expect(onSave).toHaveBeenCalledWith(
         {
           convergence: 'any',
+          maxRetries: 0,
           identifier: 'Foo',
           linkType: 'success',
           linkConditionTrigger: 'success',
diff --git a/awx/ui/src/screens/Template/WorkflowJobTemplateVisualizer/Modals/NodeModals/NodeTypeStep/NodeTypeStep.js b/awx/ui/src/screens/Template/WorkflowJobTemplateVisualizer/Modals/NodeModals/NodeTypeStep/NodeTypeStep.js
index 60777e971..b09ca67ab 100644
--- a/awx/ui/src/screens/Template/WorkflowJobTemplateVisualizer/Modals/NodeModals/NodeTypeStep/NodeTypeStep.js
+++ b/awx/ui/src/screens/Template/WorkflowJobTemplateVisualizer/Modals/NodeModals/NodeTypeStep/NodeTypeStep.js
@@ -285,6 +285,18 @@ function NodeTypeStep({ isIdentifierRequired }) {
               validate={isIdentifierRequired ? required(null) : null}
               validated={isValid ? 'default' : 'error'}
             />
+            {nodeTypeField.value !== 'workflow_approval_template' && (
+              <FormField
+                id="node-max-retries"
+                name="maxRetries"
+                type="number"
+                min="0"
+                max="100"
+                aria-label={t`Max Retries`}
+                label={t`Max Retries`}
+                tooltip={t`Maximum number of times this node's job is automatically retried after failing before its failure paths are followed. Canceled jobs are never retried.`}
+              />
+            )}
           </FormFullWidthLayout>
         </FormColumnLayout>
       </Form>
diff --git a/awx/ui/src/screens/Template/WorkflowJobTemplateVisualizer/Modals/NodeModals/NodeViewModal.js b/awx/ui/src/screens/Template/WorkflowJobTemplateVisualizer/Modals/NodeModals/NodeViewModal.js
index ee62c130c..61dcdbc5e 100644
--- a/awx/ui/src/screens/Template/WorkflowJobTemplateVisualizer/Modals/NodeModals/NodeViewModal.js
+++ b/awx/ui/src/screens/Template/WorkflowJobTemplateVisualizer/Modals/NodeModals/NodeViewModal.js
@@ -239,6 +239,10 @@ function NodeViewModal({ readOnly }) {
       };
     }
 
+    if (nodeToView?.max_retries !== undefined) {
+      nodeUpdatedConvergence.max_retries = nodeToView.max_retries;
+    }
+
     Content = (
       <PromptDetail
         launchConfig={launchConfig}
diff --git a/awx/ui/src/screens/Template/WorkflowJobTemplateVisualizer/Modals/NodeModals/useWorkflowNodeSteps.js b/awx/ui/src/screens/Template/WorkflowJobTemplateVisualizer/Modals/NodeModals/useWorkflowNodeSteps.js
index ef4900af2..8cdd5c150 100644
--- a/awx/ui/src/screens/Template/WorkflowJobTemplateVisualizer/Modals/NodeModals/useWorkflowNodeSteps.js
+++ b/awx/ui/src/screens/Template/WorkflowJobTemplateVisualizer/Modals/NodeModals/useWorkflowNodeSteps.js
@@ -79,10 +79,21 @@ const getNodeToEditDefaultValues = (
       ? 'all'
       : 'any';
   };
+  const getMaxRetries = () => {
+    if (
+      nodeToEdit &&
+      Object.prototype.hasOwnProperty.call(nodeToEdit, 'max_retries')
+    ) {
+      return nodeToEdit.max_retries;
+    }
+    return nodeToEdit?.originalNodeObject?.max_retries || 0;
+  };
+
   const initialValues = {
     nodeResource: nodeToEdit?.fullUnifiedJobTemplate || null,
     nodeType: nodeToEdit?.fullUnifiedJobTemplate?.type || 'job_template',
     convergence: getConvergence(),
+    maxRetries: getMaxRetries(),
     identifier,
   };
 
@@ -361,6 +372,7 @@ export default function useWorkflowNodeSteps(
         );
         initialValues.identifier = formikValues.identifier;
         initialValues.convergence = formikValues.convergence;
+        initialValues.maxRetries = formikValues.maxRetries;
       }
 
       const errors = formikErrors.nodeResource
diff --git a/awx/ui/src/screens/Template/WorkflowJobTemplateVisualizer/Visualizer.js b/awx/ui/src/screens/Template/WorkflowJobTemplateVisualizer/Visualizer.js
index 788500d80..8ed85573d 100644
--- a/awx/ui/src/screens/Template/WorkflowJobTemplateVisualizer/Visualizer.js
+++ b/awx/ui/src/screens/Template/WorkflowJobTemplateVisualizer/Visualizer.js
@@ -485,6 +485,7 @@ function Visualizer({ template }) {
                 inventory: node.promptValues?.inventory?.id || null,
                 unified_job_template: node.fullUnifiedJobTemplate.id,
                 all_parents_must_converge: node.all_parents_must_converge,
+                max_retries: node.max_retries || 0,
                 identifier: node.identifier || undefined,
               }).then(({ data }) => {
                 node.originalNodeObject = data;
@@ -593,6 +594,7 @@ function Visualizer({ template }) {
                 inventory: node.promptValues?.inventory?.id || null,
                 unified_job_template: node.fullUnifiedJobTemplate.id,
                 all_parents_must_converge: node.all_parents_must_converge,
+                max_retries: node.max_retries || 0,
                 ...(replaceIdentifier(node) && {
                   identifier: node.identifier,
                 }),
diff --git a/awxkit/awxkit/api/pages/workflow_job_template_nodes.py b/awxkit/awxkit/api/pages/workflow_job_template_nodes.py
index 1fe84d7f9..5522bed6e 100644
--- a/awxkit/awxkit/api/pages/workflow_job_template_nodes.py
+++ b/awxkit/awxkit/api/pages/workflow_job_template_nodes.py
@@ -31,6 +31,7 @@ def payload(self, workflow_job_template, unified_job_template, **kwargs):
             'extra_data',
             'identifier',
             'all_parents_must_converge',
+            'max_retries',
             # prompt fields for JTs
             'job_slice_count',
             'forks',
diff --git a/docs/workflow.md b/docs/workflow.md
index 07db8a30d..c67e9aecb 100644
--- a/docs/workflow.md
+++ b/docs/workflow.md
@@ -133,6 +133,17 @@ Workflow jobs support simultaneous job runs just like that of ordinary jobs. It
 A workflow job is marked as failed if a job spawned by a workflow job fails, without a failure handler. A failure handler is a `failure` or `always` link in the workflow job template. A job that is canceled is, effectively, considered a failure for the purposes of determining if a job nodes is failed.
 
 
+### Automatic Node Retries
+
+Workflow nodes can retry their job automatically when it fails, which is useful when jobs run against flaky infrastructure. The retry count is the `max_retries` field of the workflow job template node, settable through the API or the node form in the workflow visualizer. It defaults to `0` (no retries, the previous behavior) and is capped at 100 since there is no delay between attempts.
+
+Each node keeps its own independent retry budget. `max_retries: 2` means the node may run up to 3 times in total: the original attempt plus 2 retries. When a workflow job is launched the value is copied from each template node to its workflow job node, so changing the template mid run does not affect running workflows. Since the budget belongs to the node, a nested workflow node retries the inner workflow as a whole when it fails; nodes inside the nested workflow retry only if its own template says so, and slices of a sliced job template are not retried individually.
+
+While a node has retries left, a failed or errored job does not send the workflow down the node's failure paths. Instead the scheduler treats the node as if its job were still running: children stay undecided, convergence nodes keep waiting, and the workflow does not complete. On the next scheduler cycle a new job is spawned for the node with the same launch configuration, the node's read only `retry_attempts` counter is incremented, and the node's `job` field points at the newest attempt. Superseded attempts are recorded in the node's read only `retried_jobs` list, keep their link to the workflow, and cannot be deleted while it is still running. Only when the retries are exhausted does the node count as failed and its failure and always paths run, evaluated against the last attempt.
+
+Some failures are never retried: canceled jobs (cancellation is an explicit user action), workflow approval nodes (approvals are human decisions), jobs whose template was deleted mid run, and jobs that could not start at all for structural reasons such as a missing project or inventory, where retrying cannot change the outcome and the budget is exhausted immediately. Automatic retries happen within a single workflow run and complement relaunching a workflow from its failed nodes, which is a manual action taken afterwards; relaunched workflows start their nodes with fresh retry counters. There is no delay between attempts; a backoff interval between retries may be considered in the future.
+
+
 ### Workflow Copy and Relaunch
 
 Other than the normal way of creating workflow job templates, it is also possible to copy existing workflow job templates. The resulting new workflow job template will be mostly identical to the original, except for the `name` field which will be appended in a way to indicate that it's a copy.