diff --git a/lib/galaxy/dependencies/pipfiles/default/pinned-requirements.txt b/lib/galaxy/dependencies/pipfiles/default/pinned-requirements.txt
index d396c0c0e51..63e18ec1de7 100644
--- a/lib/galaxy/dependencies/pipfiles/default/pinned-requirements.txt
+++ b/lib/galaxy/dependencies/pipfiles/default/pinned-requirements.txt
@@ -56,7 +56,7 @@ functools32==3.2.3.post2 ; python_version == '2.7'
 future==0.16.0
 futures==3.2.0 ; python_version == '2.6' or python_version == '2.7'
 galaxy-sequence-utils==1.1.3
-gxformat2==0.7.1
+gxformat2==0.8.0
 h5py==2.8.0
 idna==2.7
 ipaddress==1.0.22 ; python_version < '3.3'
diff --git a/lib/galaxy/managers/workflows.py b/lib/galaxy/managers/workflows.py
index c447a0b4eb6..c81da03c9f7 100644
--- a/lib/galaxy/managers/workflows.py
+++ b/lib/galaxy/managers/workflows.py
@@ -771,6 +771,14 @@ class WorkflowContentsManager(UsesAnnotations):
                 for output in module.get_data_outputs():
                     step_dict['outputs'].append({'name': output['name'], 'type': output['extensions'][0]})
 
+            step_in = {}
+            for step_input in step.inputs:
+                if step_input.default_value_set:
+                    step_in[step_input.name] = {"default": step_input.default_value}
+
+            if step_in:
+                step_dict["in"] = step_in
+
             # Connections
             input_connections = step.input_connections
             if step.type is None or step.type == 'tool':
@@ -998,6 +1006,16 @@ class WorkflowContentsManager(UsesAnnotations):
                     label=label,
                 )
                 trans.sa_session.add(m)
+
+        if "in" in step_dict:
+            for input_name, input_dict in step_dict["in"].items():
+                step_input = step.get_or_add_input(input_name)
+                NO_DEFAULT_DEFINED = object()
+                default = input_dict.get("default", NO_DEFAULT_DEFINED)
+                if default is not NO_DEFAULT_DEFINED:
+                    step_input.default_value = default
+                    step_input.default_value_set = True
+
         return module, step
 
     def __load_subworkflow_from_step_dict(self, trans, step_dict, subworkflow_id_map, **kwds):
@@ -1041,23 +1059,21 @@ class WorkflowContentsManager(UsesAnnotations):
                     continue
                 if not isinstance(conn_list, list):  # Older style singleton connection
                     conn_list = [conn_list]
+
                 for conn_dict in conn_list:
                     if 'output_name' not in conn_dict or 'id' not in conn_dict:
                         template = "Invalid connection [%s] - must be dict with output_name and id fields."
                         message = template % conn_dict
                         raise exceptions.MessageException(message)
-                    conn = model.WorkflowStepConnection()
-                    conn.input_step = step
-                    conn.input_name = input_name
-                    conn.output_name = conn_dict['output_name']
                     external_id = conn_dict['id']
                     if external_id not in steps_by_external_id:
                         raise KeyError("Failed to find external id %s in %s" % (external_id, steps_by_external_id.keys()))
-                    conn.output_step = steps_by_external_id[external_id]
+                    output_step = steps_by_external_id[external_id]
 
+                    output_name = conn_dict["output_name"]
                     input_subworkflow_step_index = conn_dict.get('input_subworkflow_step_id', None)
-                    if input_subworkflow_step_index is not None:
-                        conn.input_subworkflow_step = step.subworkflow.step_by_index(input_subworkflow_step_index)
+
+                    step.add_connection(input_name, output_name, output_step, input_subworkflow_step_index)
 
             del step.temp_input_connections
 
diff --git a/lib/galaxy/model/__init__.py b/lib/galaxy/model/__init__.py
index 887f49c4818..10828dbfd6a 100644
--- a/lib/galaxy/model/__init__.py
+++ b/lib/galaxy/model/__init__.py
@@ -4071,13 +4071,45 @@ class WorkflowStep(RepresentById):
         self.tool_inputs = None
         self.tool_errors = None
         self.position = None
-        self.input_connections = []
+        self.inputs = []
         self.config = None
         self.label = None
         self.uuid = uuid4()
         self.workflow_outputs = []
         self._input_connections_by_name = None
 
+    def get_input(self, input_name):
+        for step_input in self.inputs:
+            if step_input.name == input_name:
+                return step_input
+
+        return None
+
+    def get_or_add_input(self, input_name):
+        step_input = self.get_input(input_name)
+
+        if step_input is None:
+            step_input = WorkflowStepInput(self)
+            step_input.name = input_name
+        return step_input
+
+    def add_connection(self, input_name, output_name, output_step, input_subworkflow_step_index=None):
+        step_input = self.get_or_add_input(input_name)
+
+        conn = WorkflowStepConnection()
+        conn.input_step_input = step_input
+        conn.output_name = output_name
+        conn.output_step = output_step
+        if input_subworkflow_step_index is not None:
+            input_subworkflow_step = self.subworkflow.step_by_index(input_subworkflow_step_index)
+            conn.input_subworkflow_step = input_subworkflow_step
+        return conn
+
+    @property
+    def input_connections(self):
+        connections = [_ for step_input in self.inputs for _ in step_input.connections]
+        return connections
+
     @property
     def unique_workflow_outputs(self):
         # Older Galaxy workflows may have multiple WorkflowOutputs
@@ -4151,7 +4183,7 @@ class WorkflowStep(RepresentById):
         copied_step.position = self.position
         copied_step.config = self.config
         copied_step.label = self.label
-        copied_step.input_connections = copy_list(self.input_connections)
+        copied_step.inputs = copy_list(self.inputs, copied_step)
 
         subworkflow_step_mapping = {}
         subworkflow = self.subworkflow
@@ -4162,8 +4194,7 @@ class WorkflowStep(RepresentById):
                 subworkflow_step_mapping[subworkflow_step.id] = copied_subworkflow_step
 
         for old_conn, new_conn in zip(self.input_connections, copied_step.input_connections):
-            # new_conn.input_step = new_
-            new_conn.input_step = step_mapping[old_conn.input_step_id]
+            new_conn.input_step_input = copied_step.get_or_add_input(old_conn.input_name)
             new_conn.output_step = step_mapping[old_conn.output_step_id]
             if old_conn.input_subworkflow_step_id:
                 new_conn.input_subworkflow_step = subworkflow_step_mapping[old_conn.input_subworkflow_step_id]
@@ -4178,6 +4209,30 @@ class WorkflowStep(RepresentById):
         return "WorkflowStep[index=%d,type=%s]" % (self.order_index, self.type)
 
 
+class WorkflowStepInput(RepresentById):
+    default_merge_type = None
+    default_scatter_type = None
+
+    def __init__(self, workflow_step):
+        self.workflow_step = workflow_step
+        self.name = None
+        self.default_value = None
+        self.default_value_set = False
+        self.merge_type = self.default_merge_type
+        self.scatter_type = self.default_scatter_type
+
+    def copy(self, copied_step):
+        copied_step_input = WorkflowStepInput(copied_step)
+        copied_step_input.name = self.name
+        copied_step_input.default_value = self.default_value
+        copied_step_input.default_value_set = self.default_value_set
+        copied_step_input.merge_type = self.merge_type
+        copied_step_input.scatter_type = self.scatter_type
+
+        copied_step_input.connections = copy_list(self.connections)
+        return copied_step_input
+
+
 class WorkflowStepConnection(RepresentById):
     # Constant used in lieu of output_name and input_name to indicate an
     # implicit connection between two steps that is not dependent on a dataset
@@ -4189,18 +4244,29 @@ class WorkflowStepConnection(RepresentById):
     def __init__(self):
         self.output_step_id = None
         self.output_name = None
-        self.input_step_id = None
-        self.input_name = None
+        self.input_step_input_id = None
 
     @property
     def non_data_connection(self):
         return (self.output_name == self.input_name == WorkflowStepConnection.NON_DATA_CONNECTION)
 
+    @property
+    def input_name(self):
+        return self.input_step_input.name
+
+    @property
+    def input_step(self):
+        return self.input_step_input and self.input_step_input.workflow_step
+
+    @property
+    def input_step_id(self):
+        input_step = self.input_step
+        return input_step and input_step.id
+
     def copy(self):
         # TODO: handle subworkflow ids...
         copied_connection = WorkflowStepConnection()
         copied_connection.output_name = self.output_name
-        copied_connection.input_name = self.input_name
         return copied_connection
 
 
diff --git a/lib/galaxy/model/mapping.py b/lib/galaxy/model/mapping.py
index 0fc1203b841..151ae354397 100644
--- a/lib/galaxy/model/mapping.py
+++ b/lib/galaxy/model/mapping.py
@@ -908,6 +908,23 @@ model.WorkflowStep.table = Table(
     # Column( "input_connections", JSONType ),
     Column("label", Unicode(255)))
 
+
+model.WorkflowStepInput.table = Table(
+    "workflow_step_input", metadata,
+    Column("id", Integer, primary_key=True),
+    Column("workflow_step_id", Integer, ForeignKey("workflow_step.id"), index=True),
+    Column("name", TEXT),
+    Column("merge_type", TEXT),
+    Column("scatter_type", TEXT),
+    Column("value_from", JSONType),
+    Column("value_from_type", TEXT),
+    Column("default_value", JSONType),
+    Column("default_value_set", Boolean, default=False),
+    Column("runtime_value", Boolean, default=False),
+    UniqueConstraint("workflow_step_id", "name"),
+)
+
+
 model.WorkflowRequestStepState.table = Table(
     "workflow_request_step_states", metadata,
     Column("id", Integer, primary_key=True),
@@ -953,9 +970,8 @@ model.WorkflowStepConnection.table = Table(
     "workflow_step_connection", metadata,
     Column("id", Integer, primary_key=True),
     Column("output_step_id", Integer, ForeignKey("workflow_step.id"), index=True),
-    Column("input_step_id", Integer, ForeignKey("workflow_step.id"), index=True),
+    Column("input_step_input_id", Integer, ForeignKey("workflow_step_input.id"), index=True),
     Column("output_name", TEXT),
-    Column("input_name", TEXT),
     Column("input_subworkflow_step_id", Integer, ForeignKey("workflow_step.id"), index=True),
 )
 
@@ -2218,6 +2234,13 @@ mapper(model.WorkflowStep, model.WorkflowStep.table, properties=dict(
         backref="workflow_steps")
 ))
 
+mapper(model.WorkflowStepInput, model.WorkflowStepInput.table, properties=dict(
+    workflow_step=relation(model.WorkflowStep,
+        backref=backref("inputs", uselist=True),
+        cascade="all",
+        primaryjoin=(model.WorkflowStepInput.table.c.workflow_step_id == model.WorkflowStep.table.c.id))
+))
+
 mapper(model.WorkflowOutput, model.WorkflowOutput.table, properties=dict(
     workflow_step=relation(model.WorkflowStep,
         backref='workflow_outputs',
@@ -2225,10 +2248,10 @@ mapper(model.WorkflowOutput, model.WorkflowOutput.table, properties=dict(
 ))
 
 mapper(model.WorkflowStepConnection, model.WorkflowStepConnection.table, properties=dict(
-    input_step=relation(model.WorkflowStep,
-        backref="input_connections",
+    input_step_input=relation(model.WorkflowStepInput,
+        backref="connections",
         cascade="all",
-        primaryjoin=(model.WorkflowStepConnection.table.c.input_step_id == model.WorkflowStep.table.c.id)),
+        primaryjoin=(model.WorkflowStepConnection.table.c.input_step_input_id == model.WorkflowStepInput.table.c.id)),
     input_subworkflow_step=relation(model.WorkflowStep,
         backref=backref("parent_workflow_input_connections", uselist=True),
         primaryjoin=(model.WorkflowStepConnection.table.c.input_subworkflow_step_id == model.WorkflowStep.table.c.id),
diff --git a/lib/galaxy/model/migrate/versions/0136_collection_and_workflow_state.py b/lib/galaxy/model/migrate/versions/0136_collection_and_workflow_state.py
index 0a787a5f7bb..d2597359041 100644
--- a/lib/galaxy/model/migrate/versions/0136_collection_and_workflow_state.py
+++ b/lib/galaxy/model/migrate/versions/0136_collection_and_workflow_state.py
@@ -93,14 +93,6 @@ def upgrade(migrate_engine):
     for table in tables.values():
         __create(table)
 
-    def nextval(table, col='id'):
-        if migrate_engine.name in ['postgres', 'postgresql']:
-            return "nextval('%s_%s_seq')" % (table, col)
-        elif migrate_engine.name in ['mysql', 'sqlite']:
-            return "null"
-        else:
-            raise Exception("Unhandled database type")
-
     # Set default for creation to scheduled, actual mapping has new as default.
     workflow_invocation_step_state_column = Column("state", TrimmedString(64), default="scheduled")
     if migrate_engine.name in ['postgres', 'postgresql']:
diff --git a/lib/galaxy/model/migrate/versions/0145_add_workflow_step_input.py b/lib/galaxy/model/migrate/versions/0145_add_workflow_step_input.py
new file mode 100644
index 00000000000..d648d4b204c
--- /dev/null
+++ b/lib/galaxy/model/migrate/versions/0145_add_workflow_step_input.py
@@ -0,0 +1,123 @@
+"""
+Migration script for workflow step input table.
+"""
+from __future__ import print_function
+
+import logging
+
+from sqlalchemy import (
+    Boolean,
+    Column,
+    ForeignKey,
+    Integer,
+    MetaData,
+    Table,
+    TEXT,
+    UniqueConstraint
+)
+
+from galaxy.model.custom_types import JSONType
+
+log = logging.getLogger(__name__)
+metadata = MetaData()
+
+
+WorkflowStepInput_table = Table(
+    "workflow_step_input", metadata,
+    Column("id", Integer, primary_key=True),
+    Column("workflow_step_id", Integer, ForeignKey("workflow_step.id"), index=True),
+    Column("name", TEXT),
+    Column("merge_type", TEXT),
+    Column("scatter_type", TEXT),
+    Column("value_from", JSONType),
+    Column("value_from_type", TEXT),
+    Column("default_value", JSONType),
+    Column("default_value_set", Boolean, default=False),
+    Column("runtime_value", Boolean, default=False),
+    UniqueConstraint("workflow_step_id", "name"),
+)
+
+
+def upgrade(migrate_engine):
+    metadata.bind = migrate_engine
+    print(__doc__)
+    metadata.reflect()
+
+    OldWorkflowStepConnection_table = Table("workflow_step_connection", metadata, autoload=True)
+    for index in OldWorkflowStepConnection_table.indexes:
+        index.drop()
+    OldWorkflowStepConnection_table.rename("workflow_step_connection_preupgrade145")
+    # Try to deregister that table to work around some caching problems it seems.
+    OldWorkflowStepConnection_table.deregister()
+    metadata._remove_table("workflow_step_connection", metadata.schema)
+    metadata.reflect()
+
+    NewWorkflowStepConnection_table = Table(
+        "workflow_step_connection", metadata,
+        Column("id", Integer, primary_key=True),
+        Column("output_step_id", Integer, ForeignKey("workflow_step.id"), index=True),
+        Column("input_step_input_id", Integer, ForeignKey("workflow_step_input.id"), index=True),
+        Column("output_name", TEXT),
+        Column("input_subworkflow_step_id", Integer, ForeignKey("workflow_step.id"), index=True),
+    )
+    for table in (WorkflowStepInput_table, NewWorkflowStepConnection_table):
+        _create(table)
+
+    insert_step_inputs_cmd = \
+        "INSERT INTO workflow_step_input (workflow_step_id, name) " + \
+        "SELECT input_step_id, input_name FROM workflow_step_connection_preupgrade145"
+    migrate_engine.execute(insert_step_inputs_cmd)
+
+    insert_step_connections_cmd = \
+        "INSERT INTO workflow_step_connection (output_step_id, input_step_input_id, output_name, input_subworkflow_step_id) " + \
+        "SELECT wsc.output_step_id, wsi.id, wsc.output_name, wsc.input_subworkflow_step_id " + \
+        "FROM workflow_step_connection_preupgrade145 AS wsc JOIN workflow_step_input AS wsi ON wsc.input_step_id = wsi.workflow_step_id AND wsc.input_name = wsi.name ORDER BY wsc.id"
+    migrate_engine.execute(insert_step_connections_cmd)
+    _drop(OldWorkflowStepConnection_table)
+
+
+def downgrade(migrate_engine):
+    metadata.bind = migrate_engine
+
+    NewWorkflowStepConnection_table = Table("workflow_step_connection", metadata, autoload=True)
+    for index in NewWorkflowStepConnection_table.indexes:
+        index.drop()
+    NewWorkflowStepConnection_table.rename("workflow_step_connection_predowngrade145")
+    # Try to deregister that table to work around some caching problems it seems.
+    NewWorkflowStepConnection_table.deregister()
+    metadata._remove_table("workflow_step_connection", metadata.schema)
+    metadata.reflect()
+
+    OldWorkflowStepConnection_table = Table(
+        "workflow_step_connection", metadata,
+        Column("id", Integer, primary_key=True),
+        Column("output_step_id", Integer, ForeignKey("workflow_step.id"), index=True),
+        Column("input_step_id", Integer, ForeignKey("workflow_step.id"), index=True),
+        Column("output_name", TEXT),
+        Column("input_name", TEXT),
+        Column("input_subworkflow_step_id", Integer, ForeignKey("workflow_step.id"), index=True),
+    )
+    _create(OldWorkflowStepConnection_table)
+
+    insert_step_connections_cmd = \
+        "INSERT INTO workflow_step_connection (output_step_id, input_step_id, output_name, input_name, input_subworkflow_step_id) " + \
+        "SELECT wsc.output_step_id, wsi.workflow_step_id, wsc.output_name, wsi.name, wsc.input_subworkflow_step_id " + \
+        "FROM workflow_step_connection_predowngrade145 AS wsc JOIN workflow_step_input AS wsi ON wsc.input_step_input_id = wsi.id ORDER BY wsc.id"
+    migrate_engine.execute(insert_step_connections_cmd)
+
+    for table in (WorkflowStepInput_table, NewWorkflowStepConnection_table):
+        _drop(table)
+
+
+def _create(table):
+    try:
+        table.create()
+    except Exception:
+        log.exception("Creating %s table failed.", table.name)
+
+
+def _drop(table):
+    try:
+        table.drop()
+    except Exception:
+        log.exception("Dropping %s table failed.", table.name)
diff --git a/lib/galaxy/webapps/tool_shed/model/__init__.py b/lib/galaxy/webapps/tool_shed/model/__init__.py
index e2720dc2518..bf3ec3b30db 100644
--- a/lib/galaxy/webapps/tool_shed/model/__init__.py
+++ b/lib/galaxy/webapps/tool_shed/model/__init__.py
@@ -479,10 +479,34 @@ class WorkflowStep(object):
         self.tool_inputs = None
         self.tool_errors = None
         self.position = None
-        self.input_connections = []
+        self.inputs = []
         self.config = None
         self.label = None
 
+    def get_or_add_input(self, input_name):
+        for step_input in self.inputs:
+            if step_input.name == input_name:
+                return step_input
+
+        step_input = WorkflowStepInput()
+        step_input.workflow_step = self
+        step_input.name = input_name
+        self.inputs.append(step_input)
+        return step_input
+
+    @property
+    def input_connections(self):
+        connections = [_ for step_input in self.inputs for _ in step_input.connections]
+        return connections
+
+
+class WorkflowStepInput(object):
+
+    def __init__(self):
+        self.id = None
+        self.name = None
+        self.connections = []
+
 
 class WorkflowStepConnection(object):
 
diff --git a/lib/galaxy/workflow/extract.py b/lib/galaxy/workflow/extract.py
index 44bdda300a2..bf2f4165787 100644
--- a/lib/galaxy/workflow/extract.py
+++ b/lib/galaxy/workflow/extract.py
@@ -124,10 +124,10 @@ def extract_steps(trans, history=None, job_ids=None, dataset_ids=None, dataset_c
                 else:
                     log.info("Cannot find implicit input collection for %s" % input_name)
             if other_hid in hid_to_output_pair:
+                step_input = step.get_or_add_input(input_name)
                 other_step, other_name = hid_to_output_pair[other_hid]
                 conn = model.WorkflowStepConnection()
-                conn.input_step = step
-                conn.input_name = input_name
+                conn.input_step_input = step_input
                 # Should always be connected to an earlier step
                 conn.output_step = other_step
                 conn.output_name = other_name
diff --git a/lib/galaxy/workflow/modules.py b/lib/galaxy/workflow/modules.py
index 9743d4f42c9..707695844fb 100644
--- a/lib/galaxy/workflow/modules.py
+++ b/lib/galaxy/workflow/modules.py
@@ -197,11 +197,14 @@ class WorkflowModule(object):
         """
         return {}
 
-    def compute_runtime_state(self, trans, step_updates=None):
+    def compute_runtime_state(self, trans, step=None, step_updates=None):
         """ Determine the runtime state (potentially different from self.state
         which describes configuration state). This (again unlike self.state) is
         currently always a `DefaultToolState` object.
 
+        If `step` is not `None`, it will be used to search for default values
+        defined in workflow input steps.
+
         If `step_updates` is `None`, this is likely for rendering the run form
         for instance and no runtime properties are available and state must be
         solely determined by the default runtime state described by the step.
@@ -211,6 +214,21 @@ class WorkflowModule(object):
         """
         state = self.get_runtime_state()
         step_errors = {}
+
+        if step is not None:
+
+            def update_value(input, context, prefixed_name, **kwargs):
+                step_input = step.get_input(prefixed_name)
+                if step_input is None:
+                    return NO_REPLACEMENT
+
+                if step_input.default_value_set:
+                    return step_input.default_value
+
+                return NO_REPLACEMENT
+
+            visit_input_values(self.get_runtime_inputs(), state.inputs, update_value, no_replacement_value=NO_REPLACEMENT)
+
         if step_updates:
 
             def update_value(input, context, prefixed_name, **kwargs):
@@ -489,6 +507,14 @@ class InputProxy(object):
         self.input = input
         self.prefixed_name = prefixed_name
 
+    @property
+    def name(self):
+        return self.prefixed_name
+
+    @property
+    def label(self):
+        return self.prefixed_name
+
     def to_dict(self, *args, **kwds):
         as_dict = self.input.to_dict(*args, **kwds)
         as_dict["name"] = self.prefixed_name
@@ -972,7 +998,7 @@ class ToolModule(WorkflowModule):
         """
         super(ToolModule, self).recover_state(state, **kwds)
         if kwds.get("fill_defaults", False) and self.tool:
-            self.compute_runtime_state(self.trans, step_updates=None)
+            self.compute_runtime_state(self.trans, step=None, step_updates=None)
             self.augment_tool_state_for_input_connections(**kwds)
             self.tool.check_and_update_param_values(self.state.inputs, self.trans, workflow_building_mode=True)
 
@@ -1042,14 +1068,14 @@ class ToolModule(WorkflowModule):
     def get_runtime_inputs(self, **kwds):
         return self.get_inputs()
 
-    def compute_runtime_state(self, trans, step_updates=None):
+    def compute_runtime_state(self, trans, step=None, step_updates=None):
         # Warning: This method destructively modifies existing step state.
         if self.tool:
             step_errors = {}
             state = self.state
             self.runtime_post_job_actions = {}
+            state, step_errors = super(ToolModule, self).compute_runtime_state(trans, step, step_updates)
             if step_updates:
-                state, step_errors = super(ToolModule, self).compute_runtime_state(trans, step_updates)
                 self.runtime_post_job_actions = step_updates.get(RUNTIME_POST_JOB_ACTIONS_KEY, {})
                 step_metadata_runtime_state = self.__step_meta_runtime_state()
                 if step_metadata_runtime_state:
@@ -1404,7 +1430,7 @@ class WorkflowModuleInjector(object):
             subworkflow = step.subworkflow
             populate_module_and_state(self.trans, subworkflow, param_map=unjsonified_subworkflow_param_map)
 
-        state, step_errors = module.compute_runtime_state(self.trans, step_args)
+        state, step_errors = module.compute_runtime_state(self.trans, step, step_args)
         step.state = state
 
         # Fix any missing parameters
diff --git a/lib/tool_shed/util/workflow_util.py b/lib/tool_shed/util/workflow_util.py
index 9de41eb748b..976220cfc3e 100644
--- a/lib/tool_shed/util/workflow_util.py
+++ b/lib/tool_shed/util/workflow_util.py
@@ -306,13 +306,12 @@ def get_workflow_from_dict(trans, workflow_dict, tools_metadata, repository_id,
         # Input connections.
         for input_name, conn_dict in step.temp_input_connections.items():
             if conn_dict:
+                step_input = step.get_or_add_input(input_name)
                 output_step = steps_by_external_id[conn_dict['id']]
                 conn = trans.model.WorkflowStepConnection()
-                conn.input_step = step
-                conn.input_name = input_name
+                conn.input_step_input = step_input
                 conn.output_step = output_step
                 conn.output_name = conn_dict['output_name']
-                step.input_connections.append(conn)
         del step.temp_input_connections
     # Order the steps if possible.
     attach_ordered_steps(workflow, steps)
diff --git a/test/api/test_workflows.py b/test/api/test_workflows.py
index 351627e2fa8..c29ff11cd55 100644
--- a/test/api/test_workflows.py
+++ b/test/api/test_workflows.py
@@ -1410,13 +1410,14 @@ steps:
       steps:
         random_lines:
           tool_id: random_lines1
-          state:
-            num_lines: 2
-            input:
-              $link: inner_input
-            seed_source:
-              seed_source_selector: set_seed
-              seed: asdf
+          in:
+            input: inner_input
+            num_lines:
+              default: 2
+            seed_source|seed_source_selector:
+              default: set_seed
+            seed_source|seed:
+              default: asdf
         split:
           tool_id: split
           in:
@@ -2868,6 +2869,73 @@ steps:
         self.dataset_populator.wait_for_history(history_id, assert_ok=True)
         self.assertEqual("2\n", self.dataset_populator.get_history_dataset_content(history_id))
 
+    @skip_without_tool("random_lines1")
+    def test_run_replace_params_over_default(self):
+        with self.dataset_populator.test_history() as history_id:
+            self._run_jobs("""
+class: GalaxyWorkflow
+inputs:
+  input: data
+steps:
+  randomlines:
+    tool_id: random_lines1
+    in:
+      input: input
+      num_lines:
+        default: 6
+""", test_data="""
+step_parameters:
+  '1':
+    num_lines: 4
+input:
+  value: 1.bed
+  type: File
+""", history_id=history_id, wait=True, assert_ok=True, round_trip_format_conversion=True)
+            result = self.dataset_populator.get_history_dataset_content(history_id)
+            assert result.count("\n") == 4
+
+    @skip_without_tool("random_lines1")
+    def test_run_replace_params_over_default_delayed(self):
+        with self.dataset_populator.test_history() as history_id:
+            run_summary = self._run_jobs("""
+class: GalaxyWorkflow
+inputs:
+  input: data
+steps:
+  first_cat:
+    tool_id: cat1
+    in:
+      input1: input
+  the_pause:
+    type: pause
+    in:
+      input: first_cat/out_file1
+  randomlines:
+    tool_id: random_lines1
+    in:
+      input: the_pause
+      num_lines:
+        default: 6
+""", test_data="""
+step_parameters:
+  '3':
+    num_lines: 4
+input:
+  value: 1.bed
+  type: File
+""", history_id=history_id, wait=False)
+            wait_on(lambda: len(self._history_jobs(history_id)) >= 2 or None, "history jobs")
+            self.dataset_populator.wait_for_history(history_id, assert_ok=True)
+
+            workflow_id = run_summary.workflow_id
+            invocation_id = run_summary.invocation_id
+
+            self.__review_paused_steps(workflow_id, invocation_id, order_index=2, action=True)
+            self.wait_for_invocation_and_jobs(history_id, workflow_id, invocation_id)
+
+            result = self.dataset_populator.get_history_dataset_content(history_id)
+            assert result.count("\n") == 4
+
     def test_pja_import_export(self):
         workflow = self.workflow_populator.load_workflow(name="test_for_pja_import", add_pja=True)
         uploaded_workflow_id = self.workflow_populator.create_workflow(workflow)
diff --git a/test/unit/test_galaxy_mapping.py b/test/unit/test_galaxy_mapping.py
index 9bddc973a22..32448cd314b 100644
--- a/test/unit/test_galaxy_mapping.py
+++ b/test/unit/test_galaxy_mapping.py
@@ -485,6 +485,11 @@ class MappingTests(unittest.TestCase):
         workflow_step_2.type = "subworkflow"
         workflow_step_2.subworkflow = child_workflow
 
+        workflow_step_1.get_or_add_input("moo1")
+        workflow_step_1.get_or_add_input("moo2")
+        workflow_step_2.get_or_add_input("moo")
+        workflow_step_1.add_connection("foo", "cow", workflow_step_2)
+
         workflow = workflow_from_steps([workflow_step_1, workflow_step_2])
         self.persist(workflow)
 
diff --git a/test/unit/workflows/test_modules.py b/test/unit/workflows/test_modules.py
index 1394a39645e..64d479b50be 100644
--- a/test/unit/workflows/test_modules.py
+++ b/test/unit/workflows/test_modules.py
@@ -40,7 +40,7 @@ def test_data_input_step_modified_state():
 
 def test_data_input_compute_runtime_state_default():
     module = __from_step(type="data_input")
-    state, errors = module.compute_runtime_state(module.trans)
+    state, errors = module.compute_runtime_state(module.trans, module.test_step)
     assert not errors
     assert "input" in state.inputs
     assert state.inputs["input"] is None
@@ -52,7 +52,7 @@ def test_data_input_compute_runtime_state_args():
     hda = model.HistoryDatasetAssociation()
     with mock.patch("galaxy.workflow.modules.check_param") as check_method:
         check_method.return_value = (hda, None)
-        state, errors = module.compute_runtime_state(module.trans, {"input": 4, "tool_state": tool_state})
+        state, errors = module.compute_runtime_state(module.trans, module.test_step, {"input": 4, "tool_state": tool_state})
     assert not errors
     assert "input" in state.inputs
     assert state.inputs["input"] is hda
@@ -166,25 +166,28 @@ steps:
     label: "input2"
   - type: "tool"
     tool_id: "cat1"
-    input_connections:
-    -  input_name: "input1"
-       "@output_step": 0
-       output_name: "output"
+    inputs:
+      input1:
+        connections:
+        - "@output_step": 0
+          output_name: "output"
   - type: "tool"
     tool_id: "cat1"
-    input_connections:
-    -  input_name: "input1"
-       "@output_step": 0
-       output_name: "output"
+    inputs:
+      input1:
+        connections:
+        - "@output_step": 0
+          output_name: "output"
     workflow_outputs:
     -   output_name: "out_file1"
         label: "out1"
   - type: "tool"
     tool_id: "cat1"
-    input_connections:
-    -  input_name: "input1"
-       "@output_step": 2
-       output_name: "out_file1"
+    inputs:
+      input1:
+        connections:
+        - "@output_step": 2
+          output_name: "out_file1"
     workflow_outputs:
     -   output_name: "out_file1"
 """
diff --git a/test/unit/workflows/test_render.py b/test/unit/workflows/test_render.py
index b7afcd28cd1..04ee10c791c 100644
--- a/test/unit/workflows/test_render.py
+++ b/test/unit/workflows/test_render.py
@@ -6,28 +6,28 @@ steps:
   - type: "data_input"
     order_index: 0
     tool_inputs: {"name": "input1"}
-    input_connections: []
     position: {"top": 3, "left": 3}
   - type: "data_input"
     order_index: 1
     tool_inputs: {"name": "input2"}
-    input_connections: []
     position: {"top": 6, "left": 4}
   - type: "tool"
     tool_id: "cat1"
     order_index: 2
-    input_connections:
-    -  input_name: "input1"
-       "@output_step": 0
-       output_name: "di1"
+    inputs:
+      input1:
+        connection:
+        - "@output_step": 0
+          output_name: "di1"
     position: {"top": 13, "left": 10}
   - type: "tool"
     tool_id: "cat1"
     order_index: 3
-    input_connections:
-    -  input_name: "input1"
-       "@output_step": 0
-       output_name: "di1"
+    inputs:
+      input1:
+        connection:
+        - "@output_step": 0
+          output_name: "di1"
     position: {"top": 33, "left": 103}
 """
 
diff --git a/test/unit/workflows/test_workflow_progress.py b/test/unit/workflows/test_workflow_progress.py
index 1cede6d09aa..028483f1640 100644
--- a/test/unit/workflows/test_workflow_progress.py
+++ b/test/unit/workflows/test_workflow_progress.py
@@ -12,22 +12,25 @@ steps:
     tool_inputs: {"name": "input2"}
   - type: "tool"
     tool_id: "cat1"
-    input_connections:
-    -  input_name: "input1"
-       "@output_step": 0
-       output_name: "output"
+    inputs:
+      "input1":
+        connections:
+        - "@output_step": 0
+          output_name: "output"
   - type: "tool"
     tool_id: "cat1"
-    input_connections:
-    -  input_name: "input1"
-       "@output_step": 0
-       output_name: "output"
+    inputs:
+      input1:
+        connections:
+        - "@output_step": 0
+          output_name: "output"
   - type: "tool"
     tool_id: "cat1"
-    input_connections:
-    -  input_name: "input1"
-       "@output_step": 2
-       output_name: "out_file1"
+    inputs:
+      "input1":
+        connections:
+        - "@output_step": 2
+          output_name: "out_file1"
 """
 
 TEST_SUBWORKFLOW_YAML = """
@@ -41,15 +44,17 @@ steps:
             tool_inputs: {"name": "inner_input"}
           - type: "tool"
             tool_id: "cat1"
-            input_connections:
-            -  input_name: "input1"
-               "@output_step": 0
-               output_name: "output"
-    input_connections:
-    -  input_name: "inner_input"
-       "@output_step": 0
-       output_name: "output"
-       "@input_subworkflow_step": 0
+            inputs:
+              "input1":
+                  connections:
+                  - "@output_step": 0
+                    output_name: "output"
+    inputs:
+      inner_input:
+        connections:
+        - "@output_step": 0
+          output_name: "output"
+          "@input_subworkflow_step": 0
 """
 
 UNSCHEDULED_STEP = object()
diff --git a/test/unit/workflows/workflow_support.py b/test/unit/workflows/workflow_support.py
index 6f4f567cb66..b460bfe1061 100644
--- a/test/unit/workflows/workflow_support.py
+++ b/test/unit/workflows/workflow_support.py
@@ -99,20 +99,28 @@ def yaml_to_model(has_dict, id_offset=100):
 
         for key, value in step.items():
             if key == "input_connections":
-                connections = []
-                for conn_dict in value:
-                    conn = model.WorkflowStepConnection()
-                    for conn_key, conn_value in conn_dict.items():
-                        if conn_key == "@output_step":
-                            target_step = workflow.steps[conn_value]
-                            conn_value = target_step
-                            conn_key = "output_step"
-                        if conn_key == "@input_subworkflow_step":
-                            conn_value = step["subworkflow"].step_by_index(conn_value)
-                            conn_key = "input_subworkflow_step"
-                        setattr(conn, conn_key, conn_value)
-                    connections.append(conn)
-                value = connections
+                raise NotImplementedError()
+            if key == "inputs":
+                inputs = []
+                for input_name, input_def in value.items():
+                    step_input = model.WorkflowStepInput(workflow_step)
+                    step_input.name = input_name
+                    connections = []
+                    for conn_dict in input_def.get("connections", []):
+                        conn = model.WorkflowStepConnection()
+                        for conn_key, conn_value in conn_dict.items():
+                            if conn_key == "@output_step":
+                                target_step = workflow.steps[conn_value]
+                                conn_value = target_step
+                                conn_key = "output_step"
+                            if conn_key == "@input_subworkflow_step":
+                                conn_value = step["subworkflow"].step_by_index(conn_value)
+                                conn_key = "input_subworkflow_step"
+                            setattr(conn, conn_key, conn_value)
+                        connections.append(conn)
+                    step_input.connections = connections
+                    inputs.append(step_input)
+                value = inputs
             if key == "workflow_outputs":
                 value = [partial(_dict_to_workflow_output, workflow_step)(_) for _ in value]
             setattr(workflow_step, key, value)