diff --git a/client/src/components/Workflow/Run/WorkflowRunDefaultStep.vue b/client/src/components/Workflow/Run/WorkflowRunDefaultStep.vue
index ba49c2c49ce..beae0371d3e 100644
--- a/client/src/components/Workflow/Run/WorkflowRunDefaultStep.vue
+++ b/client/src/components/Workflow/Run/WorkflowRunDefaultStep.vue
@@ -2,13 +2,19 @@
     <div :step-label="model.step_label">
         <FormCard :title="model.fixed_title" :icon="icon" :collapsible="true" :expanded.sync="expanded">
             <template v-slot:body>
+                <FormMessage :message="errorText" variant="danger" :persistent="true" />
                 <FormDisplay
-                    v-if="hasInputs"
-                    :inputs="inputs"
+                    :inputs="modelInputs"
+                    :sustain-repeats="true"
+                    :sustain-conditionals="true"
+                    :replace-params="replaceParams"
                     :validation-scroll-to="validationScrollTo"
+                    collapsed-enable-text="Edit"
+                    collapsed-enable-icon="fa fa-edit"
+                    collapsed-disable-text="Undo"
+                    collapsed-disable-icon="fa fa-undo"
                     @onChange="onChange"
                     @onValidation="onValidation" />
-                <div v-else class="py-2">No options available.</div>
             </template>
         </FormCard>
     </div>
@@ -17,52 +23,43 @@
 <script>
 import WorkflowIcons from "components/Workflow/icons";
 import FormDisplay from "components/Form/FormDisplay";
+import FormMessage from "components/Form/FormMessage";
 import FormCard from "components/Form/FormCard";
+import { visitInputs } from "components/Form/utilities";
+import { getTool } from "./services";
 
 export default {
     components: {
         FormDisplay,
         FormCard,
+        FormMessage,
     },
     props: {
         model: {
             type: Object,
             required: true,
         },
+        replaceParams: {
+            type: Object,
+            default: null,
+        },
         validationScrollTo: {
             type: Array,
             required: true,
         },
+        historyId: {
+            type: String,
+            default: null,
+        },
     },
     data() {
         return {
             expanded: this.model.expanded,
+            errorText: null,
+            modelIndex: {},
+            modelInputs: this.model.inputs,
         };
     },
-    computed: {
-        icon() {
-            return WorkflowIcons[this.model.step_type];
-        },
-        isSimpleInput() {
-            return (
-                this.model.step_type.startsWith("data_input") ||
-                this.model.step_type.startsWith("data_collection_input")
-            );
-        },
-        inputs() {
-            this.model.inputs.forEach((input) => {
-                input.flavor = "module";
-                input.hide_label = this.isSimpleInput;
-            });
-            if (this.model.inputs && this.model.inputs.length > 0) {
-                return this.model.inputs;
-            }
-            return [];
-        },
-        hasInputs() {
-            return this.inputs.length > 0;
-        },
-    },
     watch: {
         validationScrollTo() {
             if (this.validationScrollTo.length > 0) {
@@ -70,8 +67,35 @@ export default {
             }
         },
     },
+    computed: {
+        icon() {
+            return WorkflowIcons[this.model.step_type];
+        },
+    },
     methods: {
-        onChange(data) {
+        onCreateIndex() {
+            this.modelIndex = {};
+            visitInputs(this.modelInputs, (input, name) => {
+                this.modelIndex[name] = input;
+            });
+        },
+        onChange(data, refreshRequest) {
+            if (refreshRequest) {
+                getTool(this.model.id, this.model.version, data, this.historyId).then(
+                    (newModel) => {
+                        this.onCreateIndex();
+                        visitInputs(newModel.inputs, (newInput, name) => {
+                            const input = this.modelIndex[name];
+                            input.options = newInput.options;
+                            input.textable = newInput.textable;
+                        });
+                        this.modelInputs = JSON.parse(JSON.stringify(this.modelInputs));
+                    },
+                    (errorText) => {
+                        this.errorText = errorText;
+                    }
+                );
+            }
             this.$emit("onChange", this.model.index, data);
         },
         onValidation(validation) {
diff --git a/client/src/components/Workflow/Run/WorkflowRunForm.vue b/client/src/components/Workflow/Run/WorkflowRunForm.vue
index 67669e38790..1fea4c76115 100644
--- a/client/src/components/Workflow/Run/WorkflowRunForm.vue
+++ b/client/src/components/Workflow/Run/WorkflowRunForm.vue
@@ -36,15 +36,15 @@
                     </template>
                 </FormCard>
                 <div v-for="step in model.steps" :key="step.index">
-                    <WorkflowRunToolStep
-                        v-if="step.step_type == 'tool'"
+                    <WorkflowRunDefaultStep
+                        v-if="step.step_type == 'tool' || step.step_type == 'subworkflow'"
                         :model="step"
                         :replace-params="getReplaceParams(step.inputs)"
                         :validation-scroll-to="getValidationScrollTo(step.index)"
                         :history-id="currentHistoryId"
                         @onChange="onToolStepInputs"
                         @onValidation="onValidation" />
-                    <WorkflowRunDefaultStep
+                    <WorkflowRunInputStep
                         v-else
                         :model="step"
                         :validation-scroll-to="getValidationScrollTo(step.index)"
@@ -64,7 +64,7 @@ import FormCard from "components/Form/FormCard";
 import FormElement from "components/Form/FormElement";
 import UserHistories from "components/providers/UserHistories";
 import WorkflowRunDefaultStep from "./WorkflowRunDefaultStep";
-import WorkflowRunToolStep from "./WorkflowRunToolStep";
+import WorkflowRunInputStep from "./WorkflowRunInputStep";
 import { allowCachedJobs } from "components/Tool/utilities";
 import { getReplacements } from "./model";
 import { invokeWorkflow } from "./services";
@@ -78,7 +78,7 @@ export default {
         FormElement,
         UserHistories,
         WorkflowRunDefaultStep,
-        WorkflowRunToolStep,
+        WorkflowRunInputStep,
     },
     props: {
         model: {
@@ -93,6 +93,7 @@ export default {
             stepValidations: {},
             stepScrollTo: {},
             wpData: {},
+            inputs: {},
             historyData: {},
             useCachedJobs: false,
             historyInputs: [
@@ -155,8 +156,7 @@ export default {
             return [];
         },
         onDefaultStepInputs(stepId, data) {
-            this.stepData[stepId] = data;
-            this.stepData = Object.assign({}, this.stepData);
+            this.inputs[stepId] = data.input;
         },
         onToolStepInputs(stepId, data) {
             this.stepData[stepId] = data;
@@ -201,6 +201,7 @@ export default {
                 resource_params: this.resourceData,
                 replacement_params: this.wpData,
                 use_cached_job: this.useCachedJobs,
+                inputs: this.inputs,
                 parameters: parameters,
                 // Tool form will submit flat maps for each parameter
                 // (e.g. "repeat_0|cond|param": "foo" instead of nested
diff --git a/client/src/components/Workflow/Run/WorkflowRunInputStep.vue b/client/src/components/Workflow/Run/WorkflowRunInputStep.vue
new file mode 100644
index 00000000000..1c27fe9fac8
--- /dev/null
+++ b/client/src/components/Workflow/Run/WorkflowRunInputStep.vue
@@ -0,0 +1,83 @@
+<template>
+    <div :step-label="model.step_label">
+        <FormCard :title="model.fixed_title" :icon="icon" :collapsible="true" :expanded.sync="expanded">
+            <template v-slot:body>
+                <FormDisplay
+                    v-if="hasInputs"
+                    :inputs="inputs"
+                    :validation-scroll-to="validationScrollTo"
+                    @onChange="onChange"
+                    @onValidation="onValidation" />
+                <div v-else class="py-2">No options available.</div>
+            </template>
+        </FormCard>
+    </div>
+</template>
+
+<script>
+import WorkflowIcons from "components/Workflow/icons";
+import FormDisplay from "components/Form/FormDisplay";
+import FormCard from "components/Form/FormCard";
+
+export default {
+    components: {
+        FormDisplay,
+        FormCard,
+    },
+    props: {
+        model: {
+            type: Object,
+            required: true,
+        },
+        validationScrollTo: {
+            type: Array,
+            required: true,
+        },
+    },
+    data() {
+        return {
+            expanded: this.model.expanded,
+        };
+    },
+    computed: {
+        icon() {
+            return WorkflowIcons[this.model.step_type];
+        },
+        isSimpleInput() {
+            return (
+                this.model.step_type.startsWith("data_input") ||
+                this.model.step_type.startsWith("data_collection_input")
+            );
+        },
+        inputs() {
+            this.model.inputs.forEach((input) => {
+                input.flavor = "module";
+                input.hide_label = this.isSimpleInput;
+            });
+            if (this.model.inputs && this.model.inputs.length > 0) {
+                return this.model.inputs;
+            }
+            return [];
+        },
+        hasInputs() {
+            return this.inputs.length > 0;
+        },
+    },
+    watch: {
+        validationScrollTo() {
+            if (this.validationScrollTo.length > 0) {
+                this.expanded = true;
+            }
+        },
+    },
+    methods: {
+        onChange(data) {
+            console.log("emitting default change", data);
+            this.$emit("onChange", this.model.index, data);
+        },
+        onValidation(validation) {
+            this.$emit("onValidation", this.model.index, validation);
+        },
+    },
+};
+</script>
diff --git a/client/src/components/Workflow/Run/WorkflowRunToolStep.vue b/client/src/components/Workflow/Run/WorkflowRunToolStep.vue
deleted file mode 100644
index a90e575aa0b..00000000000
--- a/client/src/components/Workflow/Run/WorkflowRunToolStep.vue
+++ /dev/null
@@ -1,100 +0,0 @@
-<template>
-    <div :step-label="model.step_label">
-        <FormCard :title="model.fixed_title" icon="fa-wrench" :collapsible="true" :expanded.sync="expanded">
-            <template v-slot:body>
-                <FormMessage :message="errorText" variant="danger" :persistent="true" />
-                <FormDisplay
-                    :inputs="modelInputs"
-                    :sustain-repeats="true"
-                    :sustain-conditionals="true"
-                    :replace-params="replaceParams"
-                    :validation-scroll-to="validationScrollTo"
-                    collapsed-enable-text="Edit"
-                    collapsed-enable-icon="fa fa-edit"
-                    collapsed-disable-text="Undo"
-                    collapsed-disable-icon="fa fa-undo"
-                    @onChange="onChange"
-                    @onValidation="onValidation" />
-            </template>
-        </FormCard>
-    </div>
-</template>
-
-<script>
-import FormDisplay from "components/Form/FormDisplay";
-import FormMessage from "components/Form/FormMessage";
-import FormCard from "components/Form/FormCard";
-import { visitInputs } from "components/Form/utilities";
-import { getTool } from "./services";
-
-export default {
-    components: {
-        FormDisplay,
-        FormCard,
-        FormMessage,
-    },
-    props: {
-        model: {
-            type: Object,
-            required: true,
-        },
-        replaceParams: {
-            type: Object,
-            default: null,
-        },
-        validationScrollTo: {
-            type: Array,
-            required: true,
-        },
-        historyId: {
-            type: String,
-            default: null,
-        },
-    },
-    data() {
-        return {
-            expanded: this.model.expanded,
-            errorText: null,
-            modelIndex: {},
-            modelInputs: this.model.inputs,
-        };
-    },
-    watch: {
-        validationScrollTo() {
-            if (this.validationScrollTo.length > 0) {
-                this.expanded = true;
-            }
-        },
-    },
-    methods: {
-        onCreateIndex() {
-            this.modelIndex = {};
-            visitInputs(this.modelInputs, (input, name) => {
-                this.modelIndex[name] = input;
-            });
-        },
-        onChange(data, refreshRequest) {
-            if (refreshRequest) {
-                getTool(this.model.id, this.model.version, data, this.historyId).then(
-                    (newModel) => {
-                        this.onCreateIndex();
-                        visitInputs(newModel.inputs, (newInput, name) => {
-                            const input = this.modelIndex[name];
-                            input.options = newInput.options;
-                            input.textable = newInput.textable;
-                        });
-                        this.modelInputs = JSON.parse(JSON.stringify(this.modelInputs));
-                    },
-                    (errorText) => {
-                        this.errorText = errorText;
-                    }
-                );
-            }
-            this.$emit("onChange", this.model.index, data);
-        },
-        onValidation(validation) {
-            this.$emit("onValidation", this.model.index, validation);
-        },
-    },
-};
-</script>
diff --git a/client/src/components/WorkflowInvocationState/WorkflowInvocationStep.vue b/client/src/components/WorkflowInvocationState/WorkflowInvocationStep.vue
index 26df5c7301f..d76ddaf1a19 100644
--- a/client/src/components/WorkflowInvocationState/WorkflowInvocationStep.vue
+++ b/client/src/components/WorkflowInvocationState/WorkflowInvocationStep.vue
@@ -86,6 +86,7 @@
 <script>
 import { mapCacheActions } from "vuex-cache";
 import { mapGetters, mapActions } from "vuex";
+import WorkflowIcons from "components/Workflow/icons";
 import JobStep from "./JobStep";
 import ParameterStep from "./ParameterStep";
 import GenericHistoryItem from "components/History/Content/GenericItem";
@@ -134,18 +135,7 @@ export default {
             return ["data_input", "data_collection_input"].includes(this.workflowStepType);
         },
         stepIcon() {
-            switch (this.workflowStepType) {
-                case "data_input":
-                    return "fa-file";
-                case "data_collection_input":
-                    return "fa-folder-o";
-                case "parameter_input":
-                    return "fa-pencil";
-                case "subworkflow":
-                    return "fa-tasks";
-                default:
-                    return "fa-wrench";
-            }
+            return WorkflowIcons[this.workflowStepType];
         },
         stepLabel() {
             return this.labelForWorkflowStep(this.workflowStep.id);
diff --git a/lib/galaxy/model/__init__.py b/lib/galaxy/model/__init__.py
index 3b39e27169e..cf79cc631be 100644
--- a/lib/galaxy/model/__init__.py
+++ b/lib/galaxy/model/__init__.py
@@ -7034,6 +7034,10 @@ class WorkflowStep(Base, RepresentById):
         self.uuid = uuid4()
         self._input_connections_by_name = None
 
+    @reconstructor
+    def init_on_load(self):
+        self._input_connections_by_name = None
+
     @property
     def tool_uuid(self):
         return self.dynamic_tool and self.dynamic_tool.uuid
@@ -7047,13 +7051,17 @@ class WorkflowStep(Base, RepresentById):
 
     @property
     def input_default_value(self):
-        tool_inputs = self.tool_inputs
-        tool_state = tool_inputs
+        tool_state = self.tool_inputs
         default_value = tool_state.get("default")
         if default_value:
             default_value = json.loads(default_value)["value"]
         return default_value
 
+    @property
+    def input_optional(self):
+        tool_state = self.tool_inputs
+        return tool_state.get("optional") or False
+
     def get_input(self, input_name):
         for step_input in self.inputs:
             if step_input.name == input_name:
@@ -7196,7 +7204,9 @@ class WorkflowStep(Base, RepresentById):
         copied_step.workflow_outputs = copy_list(self.workflow_outputs, copied_step)
 
     def log_str(self):
-        return "WorkflowStep[index=%d,type=%s]" % (self.order_index, self.type)
+        return (
+            f"WorkflowStep[index={self.order_index},type={self.type},label={self.label},uuid={self.uuid},id={self.id}]"
+        )
 
     def clear_module_extras(self):
         # the module code adds random dynamic state to the step, this
diff --git a/lib/galaxy/workflow/run.py b/lib/galaxy/workflow/run.py
index 983a2410c90..45312a26ec5 100644
--- a/lib/galaxy/workflow/run.py
+++ b/lib/galaxy/workflow/run.py
@@ -450,12 +450,10 @@ class WorkflowProgress:
             step_id = step.id
             if step_id not in self.inputs_by_step_id and "output" not in outputs:
                 default_value = step.input_default_value
-                if default_value:
+                if default_value or step.input_optional:
                     outputs["output"] = default_value
                 else:
-                    template = "Step with id %s not found in inputs_step_id (%s)"
-                    message = template % (step.log_str(), self.inputs_by_step_id)
-                    raise ValueError(message)
+                    raise ValueError(f"{step.log_str()} not found in inputs_step_id {self.inputs_by_step_id}")
             elif step_id in self.inputs_by_step_id:
                 outputs["output"] = self.inputs_by_step_id[step_id]
 
@@ -536,9 +534,9 @@ class WorkflowProgress:
         subworkflow_inputs = {}
         for input_subworkflow_step in subworkflow.input_steps:
             connection_found = False
+            subworkflow_step_id = input_subworkflow_step.id
             for input_connection in step.input_connections:
-                if input_connection.input_subworkflow_step == input_subworkflow_step:
-                    subworkflow_step_id = input_subworkflow_step.id
+                if input_connection.input_subworkflow_step_id == subworkflow_step_id:
                     is_data = input_connection.output_step.type != "parameter_input"
                     replacement = self.replacement_for_connection(
                         input_connection,
@@ -548,7 +546,7 @@ class WorkflowProgress:
                     connection_found = True
                     break
 
-            if not connection_found:
+            if not connection_found and not input_subworkflow_step.input_optional:
                 raise Exception("Could not find connections for all subworkflow inputs.")
 
         return WorkflowProgress(
diff --git a/lib/galaxy/workflow/run_request.py b/lib/galaxy/workflow/run_request.py
index 6272d6fa393..3df28af70be 100644
--- a/lib/galaxy/workflow/run_request.py
+++ b/lib/galaxy/workflow/run_request.py
@@ -110,7 +110,7 @@ def _normalize_inputs(
             if possible_input_key in inputs:
                 inputs_key = possible_input_key
         default_value = step.tool_inputs.get("default")
-        optional = step.tool_inputs.get("optional") or False
+        optional = step.input_optional
         # Need to be careful here to make sure 'default' has correct type - not sure how to do that
         # but asserting 'optional' is definitely a bool and not a String->Bool or something is a good
         # start to ensure tool state is being preserved and loaded in a type safe way.
diff --git a/lib/galaxy_test/api/test_workflows.py b/lib/galaxy_test/api/test_workflows.py
index 376fc1b8f9e..596b225cb6b 100644
--- a/lib/galaxy_test/api/test_workflows.py
+++ b/lib/galaxy_test/api/test_workflows.py
@@ -16,6 +16,7 @@ from typing import (
 from uuid import uuid4
 
 import pytest
+import yaml
 from requests import (
     delete,
     get,
@@ -41,6 +42,7 @@ from galaxy_test.base.workflow_fixtures import (
     WORKFLOW_ONE_STEP_DEFAULT,
     WORKFLOW_OPTIONAL_FALSE_INPUT_COLLECTION,
     WORKFLOW_OPTIONAL_FALSE_INPUT_DATA,
+    WORKFLOW_OPTIONAL_INPUT_DELAYED_SCHEDULING,
     WORKFLOW_OPTIONAL_TRUE_INPUT_COLLECTION,
     WORKFLOW_OPTIONAL_TRUE_INPUT_DATA,
     WORKFLOW_PARAMETER_INPUT_INTEGER_DEFAULT,
@@ -3168,6 +3170,40 @@ input1:
             content = self.dataset_populator.get_history_dataset_content(history_id)
             assert "No input selected" in content
 
+    def test_run_with_optional_data_unspecified_survives_delayed_step(self):
+        with self.dataset_populator.test_history() as history_id:
+            self._run_workflow(
+                WORKFLOW_OPTIONAL_INPUT_DELAYED_SCHEDULING,
+                history_id=history_id,
+                wait=True,
+                assert_ok=True,
+            )
+
+    def test_run_subworkflow_with_optional_data_unspecified(self):
+        with self.dataset_populator.test_history() as history_id:
+            subworkflow = yaml.safe_load(
+                """
+class: GalaxyWorkflow
+inputs:
+  required: data
+steps:
+  nested_workflow:
+    in:
+      required: required
+test_data:
+  required:
+    value: 1.bed
+    type: File
+"""
+            )
+            subworkflow["steps"]["nested_workflow"]["run"] = yaml.safe_load(WORKFLOW_OPTIONAL_INPUT_DELAYED_SCHEDULING)
+            self._run_workflow(
+                subworkflow,
+                history_id=history_id,
+                wait=True,
+                assert_ok=True,
+            )
+
     def test_run_with_non_optional_data_unspecified_fails_invocation(self):
         with self.dataset_populator.test_history() as history_id:
             error = self._run_jobs(
diff --git a/lib/galaxy_test/base/populators.py b/lib/galaxy_test/base/populators.py
index 2797b516c1b..1a8ff1cc5f7 100644
--- a/lib/galaxy_test/base/populators.py
+++ b/lib/galaxy_test/base/populators.py
@@ -1746,8 +1746,11 @@ class BaseWorkflowPopulator(BasePopulator):
         if test_data is None:
             if jobs_descriptions is None:
                 assert source_type != "path"
-                assert isinstance(has_workflow, str)
-                jobs_descriptions = yaml.safe_load(has_workflow)
+                if isinstance(has_workflow, dict):
+                    jobs_descriptions = has_workflow
+                else:
+                    assert isinstance(has_workflow, str)
+                    jobs_descriptions = yaml.safe_load(has_workflow)
 
             test_data_dict = jobs_descriptions.get("test_data", {})
         elif not isinstance(test_data, dict):
diff --git a/lib/galaxy_test/base/workflow_fixtures.py b/lib/galaxy_test/base/workflow_fixtures.py
index bd0632633ed..f01a3c7a13e 100644
--- a/lib/galaxy_test/base/workflow_fixtures.py
+++ b/lib/galaxy_test/base/workflow_fixtures.py
@@ -614,6 +614,44 @@ steps:
 """
 
 
+WORKFLOW_OPTIONAL_INPUT_DELAYED_SCHEDULING = """
+class: GalaxyWorkflow
+inputs:
+  required:
+    type: data
+  optional:
+    type: data
+    optional: true
+outputs:
+  out1:
+    outputSource: count_multi_file/out_file1
+steps:
+  expression:
+    tool_id: expression_parse_int
+    state:
+      input1: 1
+  head:
+    tool_id: head
+    in:
+      input: required
+    state:
+      lineNum:
+        $link:  expression/out1
+  count_multi_file:
+    tool_id: count_multi_file
+    in:
+      input1:
+      - optional
+      - head/out_file1
+    out:
+      out_file1: out_file1
+test_data:
+  required:
+    value: 1.bed
+    type: File
+"""
+
+
 WORKFLOW_RUNTIME_PARAMETER_SIMPLE = """
 class: GalaxyWorkflow
 inputs:
diff --git a/test/unit/workflows/test_workflow_progress.py b/test/unit/workflows/test_workflow_progress.py
index 9293f5da066..b1e78a7f47a 100644
--- a/test/unit/workflows/test_workflow_progress.py
+++ b/test/unit/workflows/test_workflow_progress.py
@@ -189,6 +189,8 @@ class WorkflowProgressTestCase(unittest.TestCase):
         subworkflow_invocation = self.invocation.create_subworkflow_invocation_for_step(
             self.invocation.workflow.step_by_index(1)
         )
+        self.app.model.session.add(subworkflow_invocation)
+        self.app.model.session.flush()
         progress = self._new_workflow_progress()
         remaining_steps = progress.remaining_steps()
         (subworkflow_step, subworkflow_invocation_step) = remaining_steps[0]