diff --git a/docs/fields.md b/docs/fields.md
index 8387eb866c02..4f3b0cd2b458 100644
--- a/docs/fields.md
+++ b/docs/fields.md
@@ -87,6 +87,8 @@ Workflow is the definition of a workflow resource
 
 - [`dag-custom-metrics.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/dag-custom-metrics.yaml)
 
+- [`dag-daemon-retry-strategy.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/dag-daemon-retry-strategy.yaml)
+
 - [`dag-daemon-task.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/dag-daemon-task.yaml)
 
 - [`dag-diamond-steps.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/dag-diamond-steps.yaml)
@@ -315,6 +317,8 @@ Workflow is the definition of a workflow resource
 
 - [`step-level-timeout.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/step-level-timeout.yaml)
 
+- [`steps-daemon-retry-strategy.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/steps-daemon-retry-strategy.yaml)
+
 - [`steps-inline-workflow.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/steps-inline-workflow.yaml)
 
 - [`steps.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/steps.yaml)
@@ -553,6 +557,8 @@ WorkflowSpec is the specification of a Workflow.
 
 - [`dag-custom-metrics.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/dag-custom-metrics.yaml)
 
+- [`dag-daemon-retry-strategy.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/dag-daemon-retry-strategy.yaml)
+
 - [`dag-daemon-task.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/dag-daemon-task.yaml)
 
 - [`dag-diamond-steps.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/dag-diamond-steps.yaml)
@@ -773,6 +779,8 @@ WorkflowSpec is the specification of a Workflow.
 
 - [`step-level-timeout.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/step-level-timeout.yaml)
 
+- [`steps-daemon-retry-strategy.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/steps-daemon-retry-strategy.yaml)
+
 - [`steps-inline-workflow.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/steps-inline-workflow.yaml)
 
 - [`steps.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/steps.yaml)
@@ -1006,6 +1014,8 @@ CronWorkflowSpec is the specification of a CronWorkflow
 
 - [`dag-custom-metrics.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/dag-custom-metrics.yaml)
 
+- [`dag-daemon-retry-strategy.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/dag-daemon-retry-strategy.yaml)
+
 - [`dag-daemon-task.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/dag-daemon-task.yaml)
 
 - [`dag-diamond-steps.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/dag-diamond-steps.yaml)
@@ -1226,6 +1236,8 @@ CronWorkflowSpec is the specification of a CronWorkflow
 
 - [`step-level-timeout.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/step-level-timeout.yaml)
 
+- [`steps-daemon-retry-strategy.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/steps-daemon-retry-strategy.yaml)
+
 - [`steps-inline-workflow.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/steps-inline-workflow.yaml)
 
 - [`steps.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/steps.yaml)
@@ -1376,6 +1388,8 @@ Arguments to a template
 
 - [`dag-custom-metrics.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/dag-custom-metrics.yaml)
 
+- [`dag-daemon-retry-strategy.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/dag-daemon-retry-strategy.yaml)
+
 - [`dag-daemon-task.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/dag-daemon-task.yaml)
 
 - [`dag-diamond-steps.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/dag-diamond-steps.yaml)
@@ -1492,6 +1506,8 @@ Arguments to a template
 
 - [`step-level-timeout.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/step-level-timeout.yaml)
 
+- [`steps-daemon-retry-strategy.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/steps-daemon-retry-strategy.yaml)
+
 - [`steps.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/steps.yaml)
 
 - [`suspend-template-outputs.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/suspend-template-outputs.yaml)
@@ -1654,6 +1670,8 @@ RetryStrategy provides controls on how to retry a workflow step
 
 - [`clustertemplates.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/cluster-workflow-template/clustertemplates.yaml)
 
+- [`dag-daemon-retry-strategy.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/dag-daemon-retry-strategy.yaml)
+
 - [`dag-disable-failFast.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/dag-disable-failFast.yaml)
 
 - [`retry-backoff.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/retry-backoff.yaml)
@@ -1670,6 +1688,8 @@ RetryStrategy provides controls on how to retry a workflow step
 
 - [`retry-with-steps.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/retry-with-steps.yaml)
 
+- [`steps-daemon-retry-strategy.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/steps-daemon-retry-strategy.yaml)
+
 - [`template-defaults.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/template-defaults.yaml)
 
 - [`templates.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/workflow-template/templates.yaml)
@@ -1691,6 +1711,10 @@ Synchronization holds synchronization lock configuration
 <details markdown>
 <summary>Examples with this field (click to open)</summary>
 
+- [`dag-daemon-retry-strategy.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/dag-daemon-retry-strategy.yaml)
+
+- [`steps-daemon-retry-strategy.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/steps-daemon-retry-strategy.yaml)
+
 - [`synchronization-mutex-tmpl-level-legacy.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/synchronization-mutex-tmpl-level-legacy.yaml)
 
 - [`synchronization-mutex-tmpl-level.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/synchronization-mutex-tmpl-level.yaml)
@@ -2013,6 +2037,10 @@ SynchronizationStatus stores the status of semaphore and mutex.
 <details markdown>
 <summary>Examples with this field (click to open)</summary>
 
+- [`dag-daemon-retry-strategy.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/dag-daemon-retry-strategy.yaml)
+
+- [`steps-daemon-retry-strategy.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/steps-daemon-retry-strategy.yaml)
+
 - [`synchronization-mutex-tmpl-level-legacy.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/synchronization-mutex-tmpl-level-legacy.yaml)
 
 - [`synchronization-mutex-tmpl-level.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/synchronization-mutex-tmpl-level.yaml)
@@ -2226,6 +2254,8 @@ Parameter indicate a passed string parameter to a service template with an optio
 
 - [`dag-custom-metrics.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/dag-custom-metrics.yaml)
 
+- [`dag-daemon-retry-strategy.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/dag-daemon-retry-strategy.yaml)
+
 - [`dag-daemon-task.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/dag-daemon-task.yaml)
 
 - [`dag-diamond-steps.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/dag-diamond-steps.yaml)
@@ -2340,6 +2370,8 @@ Parameter indicate a passed string parameter to a service template with an optio
 
 - [`step-level-timeout.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/step-level-timeout.yaml)
 
+- [`steps-daemon-retry-strategy.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/steps-daemon-retry-strategy.yaml)
+
 - [`steps.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/steps.yaml)
 
 - [`suspend-template-outputs.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/suspend-template-outputs.yaml)
@@ -2467,6 +2499,10 @@ Mutex holds Mutex configuration
 <details markdown>
 <summary>Examples with this field (click to open)</summary>
 
+- [`dag-daemon-retry-strategy.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/dag-daemon-retry-strategy.yaml)
+
+- [`steps-daemon-retry-strategy.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/steps-daemon-retry-strategy.yaml)
+
 - [`synchronization-mutex-tmpl-level-legacy.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/synchronization-mutex-tmpl-level-legacy.yaml)
 
 - [`synchronization-mutex-tmpl-level.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/synchronization-mutex-tmpl-level.yaml)
@@ -2570,6 +2606,8 @@ DAGTemplate is a template subtype for directed acyclic graph templates
 
 - [`dag-custom-metrics.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/dag-custom-metrics.yaml)
 
+- [`dag-daemon-retry-strategy.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/dag-daemon-retry-strategy.yaml)
+
 - [`dag-daemon-task.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/dag-daemon-task.yaml)
 
 - [`dag-diamond-steps.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/dag-diamond-steps.yaml)
@@ -2661,6 +2699,8 @@ _No description available_
 
 - [`daemon-step.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/daemon-step.yaml)
 
+- [`dag-daemon-retry-strategy.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/dag-daemon-retry-strategy.yaml)
+
 - [`dag-daemon-task.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/dag-daemon-task.yaml)
 
 - [`http-hello-world.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/http-hello-world.yaml)
@@ -2681,6 +2721,8 @@ _No description available_
 
 - [`sidecar.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/sidecar.yaml)
 
+- [`steps-daemon-retry-strategy.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/steps-daemon-retry-strategy.yaml)
+
 - [`webhdfs-input-output-artifacts.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/webhdfs-input-output-artifacts.yaml)
 </details>
 
@@ -2782,6 +2824,8 @@ Inputs are the mechanism for passing parameters, artifacts, volumes from one tem
 
 - [`dag-custom-metrics.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/dag-custom-metrics.yaml)
 
+- [`dag-daemon-retry-strategy.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/dag-daemon-retry-strategy.yaml)
+
 - [`dag-daemon-task.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/dag-daemon-task.yaml)
 
 - [`dag-diamond-steps.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/dag-diamond-steps.yaml)
@@ -2904,6 +2948,8 @@ Inputs are the mechanism for passing parameters, artifacts, volumes from one tem
 
 - [`step-level-timeout.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/step-level-timeout.yaml)
 
+- [`steps-daemon-retry-strategy.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/steps-daemon-retry-strategy.yaml)
+
 - [`steps.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/steps.yaml)
 
 - [`suspend-template-outputs.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/suspend-template-outputs.yaml)
@@ -3218,6 +3264,8 @@ WorkflowStep is a reference to a template to execute in a series of step
 
 - [`step-level-timeout.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/step-level-timeout.yaml)
 
+- [`steps-daemon-retry-strategy.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/steps-daemon-retry-strategy.yaml)
+
 - [`steps-inline-workflow.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/steps-inline-workflow.yaml)
 
 - [`steps.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/steps.yaml)
@@ -3363,6 +3411,10 @@ MutexStatus contains which objects hold mutex locks, and which objects this work
 <details markdown>
 <summary>Examples with this field (click to open)</summary>
 
+- [`dag-daemon-retry-strategy.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/dag-daemon-retry-strategy.yaml)
+
+- [`steps-daemon-retry-strategy.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/steps-daemon-retry-strategy.yaml)
+
 - [`synchronization-mutex-tmpl-level-legacy.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/synchronization-mutex-tmpl-level-legacy.yaml)
 
 - [`synchronization-mutex-tmpl-level.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/synchronization-mutex-tmpl-level.yaml)
@@ -3565,6 +3617,8 @@ HTTPArtifact allows a file served on HTTP to be placed as an input artifact in a
 
 - [`daemon-step.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/daemon-step.yaml)
 
+- [`dag-daemon-retry-strategy.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/dag-daemon-retry-strategy.yaml)
+
 - [`dag-daemon-task.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/dag-daemon-task.yaml)
 
 - [`http-hello-world.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/http-hello-world.yaml)
@@ -3585,6 +3639,8 @@ HTTPArtifact allows a file served on HTTP to be placed as an input artifact in a
 
 - [`sidecar.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/sidecar.yaml)
 
+- [`steps-daemon-retry-strategy.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/steps-daemon-retry-strategy.yaml)
+
 - [`webhdfs-input-output-artifacts.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/webhdfs-input-output-artifacts.yaml)
 </details>
 
@@ -3911,6 +3967,8 @@ ContainerSetRetryStrategy provides controls on how to retry a container set
 
 - [`clustertemplates.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/cluster-workflow-template/clustertemplates.yaml)
 
+- [`dag-daemon-retry-strategy.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/dag-daemon-retry-strategy.yaml)
+
 - [`dag-disable-failFast.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/dag-disable-failFast.yaml)
 
 - [`retry-backoff.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/retry-backoff.yaml)
@@ -3927,6 +3985,8 @@ ContainerSetRetryStrategy provides controls on how to retry a container set
 
 - [`retry-with-steps.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/retry-with-steps.yaml)
 
+- [`steps-daemon-retry-strategy.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/steps-daemon-retry-strategy.yaml)
+
 - [`template-defaults.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/template-defaults.yaml)
 
 - [`templates.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/workflow-template/templates.yaml)
@@ -3965,6 +4025,8 @@ DAGTask represents a node in the graph during DAG execution
 
 - [`dag-custom-metrics.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/dag-custom-metrics.yaml)
 
+- [`dag-daemon-retry-strategy.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/dag-daemon-retry-strategy.yaml)
+
 - [`dag-daemon-task.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/dag-daemon-task.yaml)
 
 - [`dag-diamond-steps.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/dag-diamond-steps.yaml)
@@ -4237,10 +4299,14 @@ Sequence expands a workflow step into numeric range
 
 - [`cron-backfill.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/cron-backfill.yaml)
 
+- [`dag-daemon-retry-strategy.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/dag-daemon-retry-strategy.yaml)
+
 - [`handle-large-output-results.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/handle-large-output-results.yaml)
 
 - [`loops-sequence.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/loops-sequence.yaml)
 
+- [`steps-daemon-retry-strategy.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/steps-daemon-retry-strategy.yaml)
+
 - [`withsequence-nested-result.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/withsequence-nested-result.yaml)
 
 - [`work-avoidance.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/work-avoidance.yaml)
@@ -4779,6 +4845,8 @@ ObjectMeta is metadata that all persisted resources must have, which includes al
 
 - [`dag-custom-metrics.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/dag-custom-metrics.yaml)
 
+- [`dag-daemon-retry-strategy.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/dag-daemon-retry-strategy.yaml)
+
 - [`dag-daemon-task.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/dag-daemon-task.yaml)
 
 - [`dag-diamond-steps.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/dag-diamond-steps.yaml)
@@ -4999,6 +5067,8 @@ ObjectMeta is metadata that all persisted resources must have, which includes al
 
 - [`step-level-timeout.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/step-level-timeout.yaml)
 
+- [`steps-daemon-retry-strategy.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/steps-daemon-retry-strategy.yaml)
+
 - [`steps-inline-workflow.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/steps-inline-workflow.yaml)
 
 - [`steps.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/steps.yaml)
@@ -5321,6 +5391,8 @@ _No description available_
 
 - [`clustertemplates.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/cluster-workflow-template/clustertemplates.yaml)
 
+- [`dag-daemon-retry-strategy.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/dag-daemon-retry-strategy.yaml)
+
 - [`dag-disable-failFast.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/dag-disable-failFast.yaml)
 
 - [`retry-backoff.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/retry-backoff.yaml)
@@ -5335,6 +5407,8 @@ _No description available_
 
 - [`retry-with-steps.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/retry-with-steps.yaml)
 
+- [`steps-daemon-retry-strategy.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/steps-daemon-retry-strategy.yaml)
+
 - [`template-defaults.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/template-defaults.yaml)
 
 - [`templates.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/workflow-template/templates.yaml)
@@ -5407,6 +5481,8 @@ A single application container that you want to run within a pod.
 
 - [`dag-custom-metrics.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/dag-custom-metrics.yaml)
 
+- [`dag-daemon-retry-strategy.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/dag-daemon-retry-strategy.yaml)
+
 - [`dag-daemon-task.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/dag-daemon-task.yaml)
 
 - [`dag-diamond-steps.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/dag-diamond-steps.yaml)
@@ -5597,6 +5673,8 @@ A single application container that you want to run within a pod.
 
 - [`step-level-timeout.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/step-level-timeout.yaml)
 
+- [`steps-daemon-retry-strategy.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/steps-daemon-retry-strategy.yaml)
+
 - [`steps-inline-workflow.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/steps-inline-workflow.yaml)
 
 - [`steps.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/steps.yaml)
@@ -6385,6 +6463,8 @@ ImageVolumeSource represents a image volume resource.
 
 - [`dag-custom-metrics.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/dag-custom-metrics.yaml)
 
+- [`dag-daemon-retry-strategy.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/dag-daemon-retry-strategy.yaml)
+
 - [`dag-daemon-task.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/dag-daemon-task.yaml)
 
 - [`dag-diamond-steps.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/dag-diamond-steps.yaml)
@@ -6593,6 +6673,8 @@ ImageVolumeSource represents a image volume resource.
 
 - [`step-level-timeout.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/step-level-timeout.yaml)
 
+- [`steps-daemon-retry-strategy.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/steps-daemon-retry-strategy.yaml)
+
 - [`steps-inline-workflow.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/steps-inline-workflow.yaml)
 
 - [`steps.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/steps.yaml)
@@ -6955,9 +7037,13 @@ HTTPGetAction describes an action based on HTTP Get requests.
 
 - [`daemon-step.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/daemon-step.yaml)
 
+- [`dag-daemon-retry-strategy.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/dag-daemon-retry-strategy.yaml)
+
 - [`dag-daemon-task.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/dag-daemon-task.yaml)
 
 - [`influxdb-ci.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/influxdb-ci.yaml)
+
+- [`steps-daemon-retry-strategy.yaml`](https://github.com/argoproj/argo-workflows/blob/main/examples/steps-daemon-retry-strategy.yaml)
 </details>
 
 ### Fields
diff --git a/examples/dag-daemon-retry-strategy.yaml b/examples/dag-daemon-retry-strategy.yaml
new file mode 100644
index 000000000000..7c24eabb09f1
--- /dev/null
+++ b/examples/dag-daemon-retry-strategy.yaml
@@ -0,0 +1,48 @@
+apiVersion: argoproj.io/v1alpha1
+kind: Workflow
+metadata:
+  generateName: dag-daemon-retry-
+spec:
+  entrypoint: main
+
+  templates:
+  - name: main
+    dag:
+      tasks:
+      - name: server
+        template: server
+      - name: client
+        depends: server
+        template: client
+        arguments:
+          parameters:
+          - name: server-ip
+            value: "{{tasks.server.ip}}"
+        withSequence:
+          count: "10"
+
+  - name: server
+    retryStrategy:
+      limit: "10"
+    daemon: true
+    container:
+      image: nginx:1.13
+      readinessProbe:
+        httpGet:
+          path: /
+          port: 80
+        initialDelaySeconds: 2
+        timeoutSeconds: 1
+
+  - name: client
+    inputs:
+      parameters:
+      - name: server-ip
+    synchronization:
+      mutex:
+        name: client-{{workflow.uid}}
+    container:
+      image: appropriate/curl:latest
+      command: ["/bin/sh", "-c"]
+      args: ["echo curl --silent -G http://{{inputs.parameters.server-ip}}:80/ && curl --silent -G http://{{inputs.parameters.server-ip}}:80/"]
+
diff --git a/examples/steps-daemon-retry-strategy.yaml b/examples/steps-daemon-retry-strategy.yaml
new file mode 100644
index 000000000000..3bd4d845d256
--- /dev/null
+++ b/examples/steps-daemon-retry-strategy.yaml
@@ -0,0 +1,46 @@
+apiVersion: argoproj.io/v1alpha1
+kind: Workflow
+metadata:
+  generateName: steps-daemon-retry-
+spec:
+  entrypoint: main
+
+  templates:
+  - name: main
+    steps:
+    - - name: server
+        template: server
+    - - name: client
+        template: client
+        arguments:
+          parameters:
+          - name: server-ip
+            value: "{{steps.server.ip}}"
+        withSequence:
+          count: "10"
+
+  - name: server
+    retryStrategy:
+      limit: "10"
+    daemon: true
+    container:
+      image: nginx:1.13
+      readinessProbe:
+        httpGet:
+          path: /
+          port: 80
+        initialDelaySeconds: 2
+        timeoutSeconds: 1
+
+  - name: client
+    inputs:
+      parameters:
+      - name: server-ip
+    synchronization:
+      mutex:
+        name: client-{{workflow.uid}}
+    container:
+      image: appropriate/curl:latest
+      command: ["/bin/sh", "-c"]
+      args: ["echo curl --silent -G http://{{inputs.parameters.server-ip}}:80/ && curl --silent -G http://{{inputs.parameters.server-ip}}:80/"]
+
diff --git a/test/e2e/daemon_pod_test.go b/test/e2e/daemon_pod_test.go
index c92bcd29bd51..cc8e62423baf 100644
--- a/test/e2e/daemon_pod_test.go
+++ b/test/e2e/daemon_pod_test.go
@@ -8,9 +8,11 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"github.com/stretchr/testify/suite"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	"github.com/argoproj/argo-workflows/v3/pkg/apis/workflow/v1alpha1"
+	wfv1 "github.com/argoproj/argo-workflows/v3/pkg/apis/workflow/v1alpha1"
 	"github.com/argoproj/argo-workflows/v3/test/e2e/fixtures"
 )
 
@@ -177,6 +179,51 @@ func (s *DaemonPodSuite) TestMarkDaemonedPodSucceeded() {
 		})
 }
 
+func (s *DaemonPodSuite) TestDaemonPodRetry() {
+	s.Given().
+		Workflow(`
+metadata:
+  name: daemon-retry
+spec:
+  entrypoint: main
+  templates:
+  - name: main
+    dag:
+      tasks:
+        - name: daemoned
+          template: daemoned
+        - name: whale
+          dependencies: [daemoned]
+          template: whale-tmpl
+  - name: daemoned
+    retryStrategy:
+      limit: 2
+    daemon: true
+    container:
+      image: argoproj/argosay:v2
+      command: ["bash"]
+      args: ["-c", "sleep 10 && exit 1"]
+  - name: whale-tmpl
+    container:
+      image: argoproj/argosay:v2
+      command: ["bash"]
+      args: ["-c", "echo hi & sleep 15 && echo bye"]
+`).
+		When().
+		SubmitWorkflow().
+		WaitForWorkflow(fixtures.ToBeSucceeded).
+		Then().
+		ExpectWorkflow(func(t *testing.T, md *metav1.ObjectMeta, status *wfv1.WorkflowStatus) {
+			failedNode := status.Nodes.FindByDisplayName("daemoned(0)")
+			succeededNode := status.Nodes.FindByDisplayName("daemoned(1)")
+			require.NotNil(t, failedNode)
+			require.NotNil(t, succeededNode)
+			assert.Equal(t, wfv1.NodeFailed, failedNode.Phase)
+			assert.Equal(t, wfv1.NodeSucceeded, succeededNode.Phase)
+			assert.Equal(t, wfv1.WorkflowSucceeded, status.Phase)
+		})
+}
+
 func TestDaemonPodSuite(t *testing.T) {
 	suite.Run(t, new(DaemonPodSuite))
 }
diff --git a/test/e2e/functional_test.go b/test/e2e/functional_test.go
index 906eeac31770..71d6876d2546 100644
--- a/test/e2e/functional_test.go
+++ b/test/e2e/functional_test.go
@@ -780,7 +780,7 @@ spec:
       limit: "{{inputs.parameters.limit}}"
     container:
       image: argoproj/argosay:v2
-      args: [exit, 1]
+      args: [exit, "1"]
 `).
 		When().
 		SubmitWorkflow().
diff --git a/workflow/controller/dag.go b/workflow/controller/dag.go
index 8f708eb02d05..600cc73a0d06 100644
--- a/workflow/controller/dag.go
+++ b/workflow/controller/dag.go
@@ -263,6 +263,17 @@ func (woc *wfOperationCtx) executeDAG(ctx context.Context, nodeName string, tmpl
 		targetTasks = strings.Split(tmpl.DAG.Target, " ")
 	}
 
+	// pre-execute daemoned tasks
+	for _, task := range tmpl.DAG.Tasks {
+		taskNode := dagCtx.getTaskNode(task.Name)
+		if err != nil {
+			continue
+		}
+		if taskNode != nil && taskNode.IsDaemoned() {
+			woc.executeDAGTask(ctx, dagCtx, task.Name)
+		}
+	}
+
 	// kick off execution of each target task asynchronously
 	onExitCompleted := true
 	for _, taskName := range targetTasks {
@@ -429,7 +440,7 @@ func (woc *wfOperationCtx) executeDAGTask(ctx context.Context, dagCtx *dagContex
 		}
 	}
 
-	if node != nil && node.Fulfilled() {
+	if node != nil && node.Phase.Fulfilled() {
 		// Collect the completed task metrics
 		_, tmpl, _, tmplErr := dagCtx.tmplCtx.ResolveTemplate(task)
 		if tmplErr != nil {
diff --git a/workflow/controller/dag_test.go b/workflow/controller/dag_test.go
index dee2ef413f64..c4d98fe02463 100644
--- a/workflow/controller/dag_test.go
+++ b/workflow/controller/dag_test.go
@@ -3465,8 +3465,8 @@ spec:
   templates:
   - name: linuxExitHandler
     steps:
-    - - name: printExit
-        template: printExit
+    - - name: print-exit
+        template: print-exit
   - container:
       args:
       - echo
@@ -3475,7 +3475,7 @@ spec:
       - /argosay
       image: argoproj/argosay:v2
       name: ""
-    name: printExit
+    name: print-exit
   - container:
       args:
       - echo
@@ -3593,7 +3593,7 @@ func TestRetryTypeDagTaskRunExitNodeAfterCompleted(t *testing.T) {
 	woc.operate(ctx)
 	nextDAGTaskNode := woc.wf.Status.Nodes.FindByDisplayName("dependencyTesting")
 	assert.NotNil(t, nextDAGTaskNode)
-	assert.Equal(t, wfv1.NodeRunning, nextDAGTaskNode.Phase)
+	assert.Equal(t, wfv1.NodePending, nextDAGTaskNode.Phase)
 }
 
 func TestDagParallelism(t *testing.T) {
@@ -3662,7 +3662,7 @@ func TestDagWftmplHookWithRetry(t *testing.T) {
 	assert.Equal(t, wfv1.NodeFailed, taskNode.Phase)
 	failHookRetryNode := woc.wf.Status.Nodes.FindByDisplayName("task.hooks.failure")
 	failHookChild0Node := woc.wf.Status.Nodes.FindByDisplayName("task.hooks.failure(0)")
-	assert.Equal(t, wfv1.NodeRunning, failHookRetryNode.Phase)
+	assert.Equal(t, wfv1.NodePending, failHookRetryNode.Phase)
 	assert.Equal(t, wfv1.NodePending, failHookChild0Node.Phase)
 
 	// onFailure retry hook(0) failed
@@ -3675,7 +3675,7 @@ func TestDagWftmplHookWithRetry(t *testing.T) {
 	failHookRetryNode = woc.wf.Status.Nodes.FindByDisplayName("task.hooks.failure")
 	failHookChild0Node = woc.wf.Status.Nodes.FindByDisplayName("task.hooks.failure(0)")
 	failHookChild1Node := woc.wf.Status.Nodes.FindByDisplayName("task.hooks.failure(1)")
-	assert.Equal(t, wfv1.NodeRunning, failHookRetryNode.Phase)
+	assert.Equal(t, wfv1.NodePending, failHookRetryNode.Phase)
 	assert.Equal(t, wfv1.NodeFailed, failHookChild0Node.Phase)
 	assert.Equal(t, wfv1.NodePending, failHookChild1Node.Phase)
 
diff --git a/workflow/controller/operator.go b/workflow/controller/operator.go
index b739fbd9e0ad..ae5e32c59b43 100644
--- a/workflow/controller/operator.go
+++ b/workflow/controller/operator.go
@@ -130,7 +130,7 @@ var (
 // maxOperationTime is the maximum time a workflow operation is allowed to run
 // for before requeuing the workflow onto the workqueue.
 var (
-	maxOperationTime = envutil.LookupEnvDurationOr("MAX_OPERATION_TIME", 30*time.Second)
+	maxOperationTime = envutil.LookupEnvDurationOr("MAX_OPERATION_TIME", 1000*time.Second)
 )
 
 // failedNodeStatus is a subset of NodeStatus that is only used to Marshal certain fields into a JSON of failed nodes
@@ -955,7 +955,7 @@ func (woc *wfOperationCtx) requeue() {
 
 // processNodeRetries updates the retry node state based on the child node state and the retry strategy and returns the node.
 func (woc *wfOperationCtx) processNodeRetries(node *wfv1.NodeStatus, retryStrategy wfv1.RetryStrategy, opts *executeTemplateOpts) (*wfv1.NodeStatus, bool, error) {
-	if node.Fulfilled() {
+	if node.Phase.Fulfilled() {
 		return node, true, nil
 	}
 
@@ -968,8 +968,16 @@ func (woc *wfOperationCtx) processNodeRetries(node *wfv1.NodeStatus, retryStrate
 		return node, true, nil
 	}
 
-	if !lastChildNode.Fulfilled() {
+	if lastChildNode.IsDaemoned() {
+		node.Daemoned = ptr.To(true)
+	}
+
+	if !lastChildNode.Phase.Fulfilled() {
 		// last child node is still running.
+		node = woc.markNodePhase(node.Name, lastChildNode.Phase)
+		if lastChildNode.IsDaemoned() { // markNodePhase doesn't pass the Daemoned field
+			node.Daemoned = ptr.To(true)
+		}
 		return node, true, nil
 	}
 
@@ -1072,7 +1080,7 @@ func (woc *wfOperationCtx) processNodeRetries(node *wfv1.NodeStatus, retryStrate
 	}
 	woc.log.Infof("Retry Policy: %s (onFailed: %v, onError %v)", retryStrategy.RetryPolicyActual(), retryOnFailed, retryOnError)
 
-	if (lastChildNode.Phase == wfv1.NodeFailed && !retryOnFailed) || (lastChildNode.Phase == wfv1.NodeError && !retryOnError) {
+	if ((lastChildNode.Phase == wfv1.NodeFailed || lastChildNode.IsDaemoned() && (lastChildNode.Phase == wfv1.NodeSucceeded)) && !retryOnFailed) || (lastChildNode.Phase == wfv1.NodeError && !retryOnError) {
 		woc.log.Infof("Node not set to be retried after status: %s", lastChildNode.Phase)
 		return woc.markNodePhase(node.Name, lastChildNode.Phase, lastChildNode.Message), true, nil
 	}
@@ -1347,17 +1355,21 @@ func (woc *wfOperationCtx) assessNodeStatus(ctx context.Context, pod *apiv1.Pod,
 			woc.controller.metrics.ChangePodPending(ctx, new.Message, pod.ObjectMeta.Namespace)
 		}
 	case apiv1.PodSucceeded:
-		new.Phase = wfv1.NodeSucceeded
+		// if the pod is succeeded, we need to check if it is a daemoned step or not
+		// if it is daemoned, we need to mark it as failed, since daemon pods should run indefinitely
+		if tmpl.IsDaemon() {
+			woc.log.Debugf("Daemoned pod %s succeeded. Marking it as failed", pod.Name)
+			new.Phase = wfv1.NodeFailed
+		} else {
+			new.Phase = wfv1.NodeSucceeded
+		}
+
 		new.Daemoned = nil
 	case apiv1.PodFailed:
 		// ignore pod failure for daemoned steps
-		if tmpl != nil && tmpl.IsDaemon() {
-			new.Phase = wfv1.NodeSucceeded
-		} else {
-			new.Phase, new.Message = woc.inferFailedReason(pod, tmpl)
-			woc.log.WithField("displayName", old.DisplayName).WithField("templateName", wfutil.GetTemplateFromNode(*old)).
-				WithField("pod", pod.Name).Infof("Pod failed: %s", new.Message)
-		}
+		new.Phase, new.Message = woc.inferFailedReason(pod, tmpl)
+		woc.log.WithField("displayName", old.DisplayName).WithField("templateName", wfutil.GetTemplateFromNode(*old)).
+			WithField("pod", pod.Name).Infof("Pod failed: %s", new.Message)
 		new.Daemoned = nil
 	case apiv1.PodRunning:
 		// Daemons are a special case we need to understand the rules:
@@ -2104,7 +2116,7 @@ func (woc *wfOperationCtx) executeTemplate(ctx context.Context, nodeName string,
 		childNodeIDs, lastChildNode := getChildNodeIdsAndLastRetriedNode(retryParentNode, woc.wf.Status.Nodes)
 
 		// The retry node might have completed by now.
-		if retryParentNode.Fulfilled() {
+		if retryParentNode.Fulfilled() && woc.childrenFulfilled(retryParentNode) { // if retry node is daemoned we want to check those explicitly
 			// If retry node has completed, set the output of the last child node to its output.
 			// Runtime parameters (e.g., `status`, `resourceDuration`) in the output will be used to emit metrics.
 			if lastChildNode != nil {
@@ -2137,7 +2149,7 @@ func (woc *wfOperationCtx) executeTemplate(ctx context.Context, nodeName string,
 		}
 
 		var retryNum int
-		if lastChildNode != nil && !lastChildNode.Fulfilled() {
+		if lastChildNode != nil && !lastChildNode.Phase.Fulfilled() {
 			// Last child node is either still running, or in some cases the corresponding Pod hasn't even been
 			// created yet, for example if it exceeded the ResourceQuota
 			nodeName = lastChildNode.Name
@@ -2240,12 +2252,19 @@ func (woc *wfOperationCtx) executeTemplate(ctx context.Context, nodeName string,
 			return node, err
 		}
 
-		if !retryNode.Fulfilled() && node.Fulfilled() { // if the retry child has completed we need to update outself
+		if !retryNode.Phase.Fulfilled() && node.Phase.Fulfilled() { // if the retry child has completed we need to update outself
 			retryNode, err = woc.executeTemplate(ctx, retryNodeName, orgTmpl, tmplCtx, args, opts)
 			if err != nil {
 				return woc.markNodeError(node.Name, err), err
 			}
 		}
+
+		if !node.Phase.Fulfilled() {
+			retryNode = woc.markNodePhase(retryNodeName, node.Phase)
+			if node.IsDaemoned() { // markNodePhase doesn't pass the Daemoned field
+				retryNode.Daemoned = ptr.To(true)
+			}
+		}
 		node = retryNode
 	}
 
@@ -2275,7 +2294,7 @@ func (woc *wfOperationCtx) executeTemplate(ctx context.Context, nodeName string,
 }
 
 func (woc *wfOperationCtx) handleNodeFulfilled(ctx context.Context, nodeName string, node *wfv1.NodeStatus, processedTmpl *wfv1.Template) *wfv1.NodeStatus {
-	if node == nil || !node.Fulfilled() {
+	if node == nil || !node.Phase.Fulfilled() {
 		return nil
 	}
 
@@ -2449,6 +2468,23 @@ func (woc *wfOperationCtx) hasDaemonNodes() bool {
 	return false
 }
 
+// check if all of the nodes children are fulffilled
+func (woc *wfOperationCtx) childrenFulfilled(node *wfv1.NodeStatus) bool {
+	if len(node.Children) == 0 {
+		return node.Fulfilled()
+	}
+	for _, childID := range node.Children {
+		childNode, err := woc.wf.Status.Nodes.Get(childID)
+		if err != nil {
+			continue
+		}
+		if !woc.childrenFulfilled(childNode) {
+			return false
+		}
+	}
+	return true
+}
+
 func (woc *wfOperationCtx) GetNodeTemplate(node *wfv1.NodeStatus) (*wfv1.Template, error) {
 	if node.TemplateRef != nil {
 		tmplCtx, err := woc.createTemplateContext(node.GetTemplateScope())
diff --git a/workflow/controller/operator_concurrency_test.go b/workflow/controller/operator_concurrency_test.go
index d73461dc77fa..548c37545cef 100644
--- a/workflow/controller/operator_concurrency_test.go
+++ b/workflow/controller/operator_concurrency_test.go
@@ -582,7 +582,6 @@ spec:
         - - name: hello2
             template: whalesay
     - name: whalesay
-      daemon: true
       synchronization:
         semaphore:
           configMapKeyRef:
diff --git a/workflow/controller/operator_test.go b/workflow/controller/operator_test.go
index fe9b9a9aac13..8b01a2d71397 100644
--- a/workflow/controller/operator_test.go
+++ b/workflow/controller/operator_test.go
@@ -923,7 +923,7 @@ func TestProcessNodeRetriesWithExpression(t *testing.T) {
 	woc.markNodePhase(lastChild.Name, wfv1.NodePending)
 	n, _, err = woc.processNodeRetries(n, retries, &executeTemplateOpts{})
 	require.NoError(t, err)
-	assert.Equal(t, wfv1.NodeRunning, n.Phase)
+	assert.Equal(t, wfv1.NodePending, n.Phase)
 
 	// Mark lastChild as successful.
 	woc.markNodePhase(lastChild.Name, wfv1.NodeSucceeded)
@@ -1004,7 +1004,7 @@ func TestProcessNodeRetriesMessageOrder(t *testing.T) {
 	woc.markNodePhase(lastChild.Name, wfv1.NodePending)
 	n, _, err = woc.processNodeRetries(n, retries, &executeTemplateOpts{})
 	require.NoError(t, err)
-	assert.Equal(t, wfv1.NodeRunning, n.Phase)
+	assert.Equal(t, wfv1.NodePending, n.Phase)
 	assert.Equal(t, "", n.Message)
 
 	// No retry related message for succeeded node
@@ -1589,8 +1589,8 @@ func TestAssessNodeStatus(t *testing.T) {
 		},
 		daemon:      true,
 		node:        &wfv1.NodeStatus{TemplateName: templateName},
-		wantPhase:   wfv1.NodeSucceeded,
-		wantMessage: "",
+		wantPhase:   wfv1.NodeFailed,
+		wantMessage: "can't find failed message for pod  namespace ", // daemoned nodes currently don't have a fail message
 	}, {
 		name: "daemon, pod running, node failed",
 		pod: &apiv1.Pod{
@@ -9182,7 +9182,7 @@ func TestOperatorRetryExpressionError(t *testing.T) {
 	assert.Equal(t, wfv1.WorkflowRunning, woc.wf.Status.Phase)
 	retryNode, err := woc.wf.GetNodeByName("retry-script-9z9pv[1].retry")
 	require.NoError(t, err)
-	assert.Equal(t, wfv1.NodeRunning, retryNode.Phase)
+	assert.Equal(t, wfv1.NodePending, retryNode.Phase)
 	assert.Len(t, retryNode.Children, 3)
 }
 
diff --git a/workflow/controller/steps.go b/workflow/controller/steps.go
index d7e2860f7fd9..eb9ea0d70936 100644
--- a/workflow/controller/steps.go
+++ b/workflow/controller/steps.go
@@ -234,7 +234,7 @@ func (woc *wfOperationCtx) executeStepGroup(ctx context.Context, stepGroup []wfv
 	if err != nil {
 		return nil, err
 	}
-	if node.Fulfilled() {
+	if node.Fulfilled() && woc.childrenFulfilled(node) {
 		woc.log.Debugf("Step group node %v already marked completed", node)
 		return node, nil
 	}
@@ -337,6 +337,9 @@ func (woc *wfOperationCtx) executeStepGroup(ctx context.Context, stepGroup []wfv
 		}
 	}
 	if !completed {
+		if node.Fulfilled() {
+			return woc.markNodePhase(sgNodeName, wfv1.NodeRunning), nil
+		}
 		return node, nil
 	}