diff --git a/client/galaxy/scripts/mvc/form/form-data.js b/client/galaxy/scripts/mvc/form/form-data.js
index 7ea27dda0ff..b55e5cc564e 100644
--- a/client/galaxy/scripts/mvc/form/form-data.js
+++ b/client/galaxy/scripts/mvc/form/form-data.js
@@ -84,7 +84,9 @@ export var Manager = Backbone.Model.extend({
                             if (field && field.value) {
                                 value = field.value();
                                 if (input.ignore === undefined || input.ignore != value) {
-                                    if (field.collapsed && input.collapsible_value) {
+                                    if (field.collapsed && field.connected) {
+                                        value = { __class__: "ConnectedValue" };
+                                    } else if (field.collapsed && input.collapsible_value) {
                                         value = input.collapsible_value;
                                     }
                                     add(flat_id, input.id, value);
diff --git a/client/galaxy/scripts/mvc/form/form-input.js b/client/galaxy/scripts/mvc/form/form-input.js
index 56c606a20dd..6ebac7587ef 100644
--- a/client/galaxy/scripts/mvc/form/form-input.js
+++ b/client/galaxy/scripts/mvc/form/form-input.js
@@ -25,6 +25,7 @@ export default Backbone.View.extend({
         this.$collapsible = this.$(".ui-form-collapsible");
         this.$collapsible_text = this.$(".ui-form-collapsible-text");
         this.$collapsible_icon = this.$(".ui-form-collapsible-icon");
+        this.$connected_icon = this.$(".ui-form-connected-icon");
         this.$title = this.$(".ui-form-title");
         this.$title_text = this.$(".ui-form-title-text");
         this.$error_text = this.$(".ui-form-error-text");
@@ -36,16 +37,35 @@ export default Backbone.View.extend({
 
         // decide wether to expand or collapse fields
         var collapsible_value = this.model.get("collapsible_value");
+        var value = JSON.stringify(this.model.get("value"));
+        var connected = value == JSON.stringify({ __class__: "ConnectedValue" });
+        this.field.connected = connected;
         this.field.collapsed =
-            collapsible_value !== undefined &&
-            JSON.stringify(this.model.get("value")) == JSON.stringify(collapsible_value);
+            this.field.connected ||
+            (collapsible_value !== undefined &&
+                JSON.stringify(this.model.get("value")) == JSON.stringify(collapsible_value));
         this.listenTo(this.model, "change", this.render, this);
         this.render();
 
         // add click handler
         var self = this;
-        this.$collapsible.on("click", () => {
+        this.$collapsible_icon.on("click", () => {
+            if (self.field.connected) {
+                return;
+            }
             self.field.collapsed = !self.field.collapsed;
+            if (self.field.collapsed) {
+                self.field.connected = false;
+            }
+            app.trigger && app.trigger("change");
+            self.render();
+        });
+        this.$connected_icon.on("click", () => {
+            self.field.connected = !self.field.connected;
+            self.field.collapsed = self.field.connected;
+            if (!self.field.connected) {
+                this.model.set("value", null);
+            }
             app.trigger && app.trigger("change");
             self.render();
         });
@@ -108,21 +128,41 @@ export default Backbone.View.extend({
                 style: this.model.get("style")
             });
         // render collapsible options
-        if (
+        const connected = this.field.connected;
+        const collapsible =
+            !connected &&
             !this.field.collapsible_disabled &&
             !this.model.get("disabled") &&
-            this.model.get("collapsible_value") !== undefined
-        ) {
+            this.model.get("collapsible_value") !== undefined;
+        const connectable = this.model.get("connectable");
+        if (collapsible || connectable) {
             var collapsible_state = this.field.collapsed ? "enable" : "disable";
             this.$title_text.hide();
             this.$collapsible.show();
             this.$collapsible_text.text(this.model.get("label"));
-            this.$collapsible_icon
-                .removeClass()
-                .addClass("icon")
-                .addClass(this.model.get(`cls_${collapsible_state}`))
-                .attr("data-original-title", this.model.get(`text_${collapsible_state}`))
-                .tooltip({ placement: "bottom" });
+            if (collapsible && !connected) {
+                this.$collapsible_icon
+                    .removeClass()
+                    .addClass("icon ui-form-collapsible-icon")
+                    .addClass(this.model.get(`cls_${collapsible_state}`))
+                    .attr("data-original-title", this.model.get(`text_${collapsible_state}`))
+                    .tooltip({ placement: "bottom" })
+                    .show();
+            } else {
+                this.$collapsible_icon.hide();
+            }
+            if (connectable) {
+                const connectedIconStyle = this.field.connected ? "fa fa-times" : "fa fa-arrows-h";
+                this.$connected_icon
+                    .removeClass()
+                    .addClass("icon ui-form-connected-icon")
+                    .addClass(connectedIconStyle)
+                    .attr("data-original-title", this.model.get(`text_${collapsible_state}`))
+                    .tooltip({ placement: "bottom" })
+                    .show();
+            } else {
+                this.$connected_icon.hide();
+            }
         } else {
             this.$title_text.show().text(this.model.get("label"));
             this.$collapsible.hide();
@@ -145,6 +185,7 @@ export default Backbone.View.extend({
                         $("<div/>")
                             .addClass("ui-form-collapsible")
                             .append($("<i/>").addClass("ui-form-collapsible-icon"))
+                            .append($("<i/>").addClass("ui-form-connected-icon"))
                             .append($("<span/>").addClass("ui-form-collapsible-text"))
                     )
                     .append($("<span/>").addClass("ui-form-title-text"))
diff --git a/client/galaxy/scripts/mvc/form/form-section.js b/client/galaxy/scripts/mvc/form/form-section.js
index 2b0dbbe6f73..75111933817 100644
--- a/client/galaxy/scripts/mvc/form/form-section.js
+++ b/client/galaxy/scripts/mvc/form/form-section.js
@@ -175,6 +175,7 @@ var View = Backbone.View.extend({
             text_value: input_def.text_value,
             collapsible_value: input_def.collapsible_value,
             collapsible_preview: input_def.collapsible_preview,
+            connectable: input_def.connectable,
             help: input_def.help,
             argument: input_def.argument,
             disabled: input_def.disabled,
diff --git a/client/galaxy/scripts/mvc/workflow/workflow-forms.js b/client/galaxy/scripts/mvc/workflow/workflow-forms.js
index ba95ef2506a..d9602b540f3 100644
--- a/client/galaxy/scripts/mvc/workflow/workflow-forms.js
+++ b/client/galaxy/scripts/mvc/workflow/workflow-forms.js
@@ -90,6 +90,7 @@ var Tool = Backbone.View.extend({
         var options = form.model.attributes;
         Utils.deepeach(options.inputs, input => {
             if (input.type) {
+                input.connectable = true;
                 if (["data", "data_collection"].indexOf(input.type) != -1) {
                     input.type = "hidden";
                     input.info = `Data input '${input.name}' (${Utils.textify(input.extensions)})`;
@@ -105,6 +106,7 @@ var Tool = Backbone.View.extend({
         });
         Utils.deepeach(options.inputs, input => {
             if (input.type === "conditional") {
+                input.connectable = false;
                 input.test_param.collapsible_value = undefined;
             }
         });
diff --git a/client/galaxy/scripts/mvc/workflow/workflow-icons.js b/client/galaxy/scripts/mvc/workflow/workflow-icons.js
index a0239205fa2..a3b84c4e543 100644
--- a/client/galaxy/scripts/mvc/workflow/workflow-icons.js
+++ b/client/galaxy/scripts/mvc/workflow/workflow-icons.js
@@ -3,5 +3,6 @@ export default {
     data_input: "fa-file-o",
     data_collection_input: "fa-folder-o",
     subworkflow: "fa-sitemap fa-rotate-270",
+    parameter_input: "fa-pencil",
     pause: "fa-pause"
 };
diff --git a/client/galaxy/scripts/mvc/workflow/workflow-node.js b/client/galaxy/scripts/mvc/workflow/workflow-node.js
index d4c07360b1b..87b333e5c53 100644
--- a/client/galaxy/scripts/mvc/workflow/workflow-node.js
+++ b/client/galaxy/scripts/mvc/workflow/workflow-node.js
@@ -6,6 +6,14 @@ import NodeView from "mvc/workflow/workflow-view-node";
 /* global $ */
 /* global Galaxy */
 
+var StepParameterTypes = [
+    'text',
+    'integer',
+    'float',
+    'boolean',
+    'color',
+]
+
 var Node = Backbone.Model.extend({
     initialize: function(app, attr) {
         this.app = app;
@@ -192,6 +200,7 @@ var Node = Backbone.Model.extend({
             this.content_id = this.config_form.id;
         }
     },
+
     init_field_data: function(data) {
         //console.debug("init_field_data: ", data);
         if (data.type) {
@@ -214,13 +223,14 @@ var Node = Backbone.Model.extend({
             node: node
         });
         node.nodeView = nodeView;
-        $.each(data.data_inputs, (i, input) => {
+        $.each(data.inputs, (i, input) => {
             nodeView.addDataInput(input);
         });
-        if (data.data_inputs.length > 0 && data.data_outputs.length > 0) {
+
+        if (data.inputs.length > 0 && data.outputs.length > 0) {
             nodeView.addRule();
         }
-        $.each(data.data_outputs, (i, output) => {
+        $.each(data.outputs, (i, output) => {
             nodeView.addDataOutput(output);
         });
         nodeView.render();
@@ -229,7 +239,7 @@ var Node = Backbone.Model.extend({
     update_field_data: function(data) {
         var node = this;
         var nodeView = node.nodeView;
-        // remove unused output views and remove pre-existing output views from data.data_outputs,
+        // remove unused output views and remove pre-existing output views from data.outputs,
         // so that these are not added twice.
         var unused_outputs = [];
         // nodeView.outputViews contains pre-existing outputs,
@@ -237,7 +247,7 @@ var Node = Backbone.Model.extend({
         // Now we gather the unused outputs
         $.each(nodeView.outputViews, (i, output_view) => {
             var cur_name = output_view.output.name;
-            var data_names = data.data_outputs;
+            var data_names = data.outputs;
             var cur_name_in_data_outputs = false;
             _.each(data_names, data_name => {
                 if (data_name.name == cur_name) {
@@ -265,7 +275,7 @@ var Node = Backbone.Model.extend({
                 node.workflow_outputs.splice(i, 1); // removes output from list of workflow outputs
             }
         });
-        $.each(data.data_outputs, (i, output) => {
+        $.each(data.outputs, (i, output) => {
             if (!nodeView.outputViews[output.name]) {
                 nodeView.addDataOutput(output); // add data output if it does not yet exist
             } else {
@@ -291,7 +301,7 @@ var Node = Backbone.Model.extend({
         var old_body = nodeView.$("div.inputs");
         var new_body = nodeView.newInputsDiv();
         var newTerminalViews = {};
-        _.each(data.data_inputs, input => {
+        _.each(data.inputs, input => {
             var terminalView = node.nodeView.addDataInput(input, new_body);
             newTerminalViews[input.name] = terminalView;
         });
@@ -305,8 +315,9 @@ var Node = Backbone.Model.extend({
         // type (not really valid right?) but adding special logic here for
         // data collection input parameters that can have their collection
         // change.
-        if (data.data_outputs.length == 1 && "collection_type" in data.data_outputs[0]) {
-            nodeView.updateDataOutput(data.data_outputs[0]);
+        var data_outputs = data.outputs;
+        if (data_outputs.length == 1 && "collection_type" in data_outputs[0]) {
+            nodeView.updateDataOutput(data_outputs[0]);
         }
         old_body.replaceWith(new_body);
         if ("workflow_outputs" in data) {
diff --git a/client/galaxy/scripts/mvc/workflow/workflow-terminals.js b/client/galaxy/scripts/mvc/workflow/workflow-terminals.js
index dcea5622c1f..dbbae41ad1d 100644
--- a/client/galaxy/scripts/mvc/workflow/workflow-terminals.js
+++ b/client/galaxy/scripts/mvc/workflow/workflow-terminals.js
@@ -431,6 +431,18 @@ var InputTerminal = BaseInputTerminal.extend({
     }
 });
 
+var InputParameterTerminal = BaseInputTerminal.extend({
+    update: function(input) {
+        this.type = input.type;
+    },
+    connect: function(connector) {
+        BaseInputTerminal.prototype.connect.call(this, connector);
+    },
+    attachable: function(other) {
+        return this.type == other.attributes.type;
+    },
+});
+
 var InputCollectionTerminal = BaseInputTerminal.extend({
     update: function(input) {
         this.multiple = false;
@@ -566,9 +578,14 @@ var OutputCollectionTerminal = Terminal.extend({
     }
 });
 
+var OutputParameterTerminal = Terminal.extend({
+});
+
 export default {
     InputTerminal: InputTerminal,
+    InputParameterTerminal: InputParameterTerminal,
     OutputTerminal: OutputTerminal,
+    OutputParameterTerminal: OutputParameterTerminal,
     InputCollectionTerminal: InputCollectionTerminal,
     OutputCollectionTerminal: OutputCollectionTerminal,
     TerminalMapping: TerminalMapping,
diff --git a/client/galaxy/scripts/mvc/workflow/workflow-view-data.js b/client/galaxy/scripts/mvc/workflow/workflow-view-data.js
index 64affe2be68..5aa2f16764c 100644
--- a/client/galaxy/scripts/mvc/workflow/workflow-view-data.js
+++ b/client/galaxy/scripts/mvc/workflow/workflow-view-data.js
@@ -9,7 +9,7 @@ var DataInputView = Backbone.View.extend({
         this.nodeView = options.nodeView;
         this.terminalElement = options.terminalElement;
 
-        this.$el.attr("name", this.input.name).html(this.input.label);
+        this.$el.attr("name", this.input.name).html(this.input.label || this.input.name);
 
         if (!options.skipResize) {
             this.$el.css({
@@ -82,6 +82,61 @@ var DataOutputView = Backbone.View.extend({
     }
 });
 
+var ParameterOutputView = Backbone.View.extend({
+    className: "form-row dataRow",
+
+    initialize: function(options) {
+        this.output = options.output;
+        this.terminalElement = options.terminalElement;
+        this.nodeView = options.nodeView;
+
+        var output = this.output;
+        var label = output.label || output.name;
+        var node = this.nodeView.node;
+
+        this.$el.html(label);
+        this.calloutView = null;
+        if (["tool", "subworkflow"].indexOf(node.type) >= 0) {
+            var calloutView = new OutputCalloutView({
+                label: label,
+                output: output,
+                node: node
+            });
+            this.calloutView = calloutView;
+            this.$el.append(calloutView.el);
+            this.$el.hover(
+                () => {
+                    calloutView.hoverImage();
+                },
+                () => {
+                    calloutView.resetImage();
+                }
+            );
+        }
+        this.$el.css({
+            position: "absolute",
+            left: -1000,
+            top: -1000,
+            display: "none"
+        });
+        $("body").append(this.el);
+        this.nodeView.updateMaxWidth(this.$el.outerWidth() + 17);
+        this.$el
+            .css({
+                position: "",
+                left: "",
+                top: "",
+                display: ""
+            })
+            .detach();
+    },
+    redrawWorkflowOutput: function() {
+        if (this.calloutView) {
+            this.calloutView.resetImage();
+        }
+    }
+});
+
 var OutputCalloutView = Backbone.View.extend({
     tagName: "div",
 
@@ -133,5 +188,6 @@ var OutputCalloutView = Backbone.View.extend({
 
 export default {
     DataInputView: DataInputView,
-    DataOutputView: DataOutputView
+    DataOutputView: DataOutputView,
+    ParameterOutputView: ParameterOutputView,
 };
diff --git a/client/galaxy/scripts/mvc/workflow/workflow-view-node.js b/client/galaxy/scripts/mvc/workflow/workflow-view-node.js
index 82defeb87fe..f07e7a72ee6 100644
--- a/client/galaxy/scripts/mvc/workflow/workflow-view-node.js
+++ b/client/galaxy/scripts/mvc/workflow/workflow-view-node.js
@@ -48,10 +48,12 @@ export default Backbone.View.extend({
             skipResize = false;
         }
         var terminalView = this.terminalViews[input.name];
-        var terminalViewClass =
-            input.input_type == "dataset_collection"
-                ? TerminalViews.InputCollectionTerminalView
-                : TerminalViews.InputTerminalView;
+        var terminalViewClass = TerminalViews.InputTerminalView;
+        if (input.input_type == "dataset_collection") {
+            terminalViewClass = TerminalViews.InputCollectionTerminalView;
+        } else if (input.input_type == "parameter") {
+            terminalViewClass = TerminalViews.InputParameterTerminalView;
+        }
         if (terminalView && !(terminalView instanceof terminalViewClass)) {
             terminalView.el.terminal.destroy();
             terminalView = null;
@@ -80,14 +82,19 @@ export default Backbone.View.extend({
     },
 
     addDataOutput: function(output) {
-        var terminalViewClass = output.collection
-            ? TerminalViews.OutputCollectionTerminalView
-            : TerminalViews.OutputTerminalView;
+        var terminalViewClass = TerminalViews.OutputTerminalView;
+        var outputViewClass = DataViews.DataOutputView;
+        if (output.collection) {
+            terminalViewClass = TerminalViews.OutputCollectionTerminalView;
+        } else if (output.parameter) {
+            terminalViewClass = TerminalViews.OutputParameterTerminalView;
+            outputViewClass = DataViews.ParameterOutputView;
+        }
         var terminalView = new terminalViewClass({
             node: this.node,
             output: output
         });
-        var outputView = new DataViews.DataOutputView({
+        var outputView = new outputViewClass({
             output: output,
             terminalElement: terminalView.el,
             nodeView: this
diff --git a/client/galaxy/scripts/mvc/workflow/workflow-view-terminals.js b/client/galaxy/scripts/mvc/workflow/workflow-view-terminals.js
index dc8ff84d78c..3adf380e003 100644
--- a/client/galaxy/scripts/mvc/workflow/workflow-view-terminals.js
+++ b/client/galaxy/scripts/mvc/workflow/workflow-view-terminals.js
@@ -151,6 +151,17 @@ var InputTerminalView = BaseInputTerminalView.extend({
     }
 });
 
+var InputParameterTerminalView = BaseInputTerminalView.extend({
+    terminalMappingClass: Terminals.TerminalMapping,
+    terminalMappingViewClass: InputTerminalMappingView,
+    terminalForInput: function(input) {
+        return new Terminals.InputParameterTerminal({
+            element: this.el,
+            input: input
+        });
+    }
+});
+
 var InputCollectionTerminalView = BaseInputTerminalView.extend({
     terminalMappingClass: Terminals.TerminalMapping,
     terminalMappingViewClass: InputTerminalMappingView,
@@ -269,9 +280,25 @@ var OutputCollectionTerminalView = BaseOutputTerminalView.extend({
     }
 });
 
+var OutputParameterTerminalView = BaseOutputTerminalView.extend({
+    terminalMappingClass: Terminals.TerminalMapping,
+    terminalMappingViewClass: TerminalMappingView,
+    terminalForOutput: function(output) {
+        var collection_type = output.collection_type;
+        var collection_type_source = output.collection_type_source;
+        var terminal = new Terminals.OutputCollectionTerminal({
+            element: this.el,
+            type: output.type,
+        });
+        return terminal;
+    }
+});
+
 export default {
     InputTerminalView: InputTerminalView,
+    InputParameterTerminalView: InputParameterTerminalView,
     OutputTerminalView: OutputTerminalView,
+    OutputParameterTerminalView: OutputParameterTerminalView,
     InputCollectionTerminalView: InputCollectionTerminalView,
     OutputCollectionTerminalView: OutputCollectionTerminalView
 };
diff --git a/client/galaxy/scripts/qunit/tests/workflow_editor_tests.js b/client/galaxy/scripts/qunit/tests/workflow_editor_tests.js
index 3e4ad0a49cb..a2be483cfba 100644
--- a/client/galaxy/scripts/qunit/tests/workflow_editor_tests.js
+++ b/client/galaxy/scripts/qunit/tests/workflow_editor_tests.js
@@ -357,16 +357,16 @@ QUnit.module("Node unit test", {
     },
     init_field_data_simple: function(option_overrides) {
         var data = Utils.merge(option_overrides, {
-            data_inputs: [{ name: "input1", extensions: ["data"] }],
-            data_outputs: [{ name: "output1", extensions: ["data"] }],
+            inputs: [{ name: "input1", extensions: ["data"] }],
+            outputs: [{ name: "output1", extensions: ["data"] }],
             label: null
         });
         this.node.init_field_data(data);
     },
     update_field_data_with_new_input: function(option_overrides) {
         var new_data = Utils.merge(option_overrides, {
-            data_inputs: [{ name: "input1", extensions: ["data"] }, { name: "extra_0|input1", extensions: ["data"] }],
-            data_outputs: [{ name: "output1", extensions: ["data"] }],
+            inputs: [{ name: "input1", extensions: ["data"] }, { name: "extra_0|input1", extensions: ["data"] }],
+            outputs: [{ name: "output1", extensions: ["data"] }],
             post_job_actions: "{}",
             label: "New Label"
         });
@@ -401,8 +401,8 @@ QUnit.test("init_field_data properties", function(assert) {
     var node = this.node;
     this.expect_workflow_node_changed(assert, function() {
         var data = {
-            data_inputs: [],
-            data_outputs: [],
+            inputs: [],
+            outputs: [],
             type: "tool",
             name: "cat1",
             config_form: "{}",
@@ -507,8 +507,8 @@ QUnit.test("update_field_data destroys old terminals", function(assert) {
     var node = this.node;
     this.expect_workflow_node_changed(assert, function() {
         var data = {
-            data_inputs: [{ name: "input1", extensions: ["data"] }, { name: "willDisappear", extensions: ["data"] }],
-            data_outputs: [{ name: "output1", extensions: ["data"] }]
+            inputs: [{ name: "input1", extensions: ["data"] }, { name: "willDisappear", extensions: ["data"] }],
+            outputs: [{ name: "output1", extensions: ["data"] }]
         };
         node.init_field_data(data);
         var old_input_terminal = node.input_terminals.willDisappear;
diff --git a/lib/galaxy/managers/workflows.py b/lib/galaxy/managers/workflows.py
index 319b01dc872..c447a0b4eb6 100644
--- a/lib/galaxy/managers/workflows.py
+++ b/lib/galaxy/managers/workflows.py
@@ -561,7 +561,7 @@ class WorkflowContentsManager(UsesAnnotations):
                     data['upgrade_messages'][step.order_index] = {module.tool.name: "\n".join(module.version_changes)}
             # Get user annotation.
             annotation_str = self.get_item_annotation_str(trans.sa_session, trans.user, step) or ''
-            config_form = module.get_config_form()
+            config_form = module.get_config_form(step=step)
             # Pack attributes into plain dictionary
             step_dict = {
                 'id': step.order_index,
@@ -571,8 +571,8 @@ class WorkflowContentsManager(UsesAnnotations):
                 'name': module.get_name(),
                 'tool_state': module.get_state(),
                 'errors': module.get_errors(),
-                'data_inputs': module.get_data_inputs(),
-                'data_outputs': module.get_data_outputs(),
+                'inputs': module.get_all_inputs(connectable_only=True),
+                'outputs': module.get_all_outputs(),
                 'config_form': config_form,
                 'annotation': annotation_str,
                 'post_job_actions': {},
@@ -598,9 +598,6 @@ class WorkflowContentsManager(UsesAnnotations):
                         if isinstance(input, DataCollectionToolParameter):
                             input_connections_type[input.name] = "dataset_collection"
                 visit_input_values(module.tool.inputs, module.state.inputs, callback)
-                # Filter
-                # FIXME: this removes connection without displaying a message currently!
-                input_connections = [conn for conn in input_connections if conn.input_name in data_input_names]
                 # post_job_actions
                 pja_dict = {}
                 for pja in step.post_job_actions:
@@ -653,12 +650,12 @@ class WorkflowContentsManager(UsesAnnotations):
         """
         for order_index in sorted(steps):
             step = steps[order_index]
-            for i, step_data_output in enumerate(step['data_outputs']):
+            for i, step_data_output in enumerate(step['outputs']):
                 if step_data_output.get('collection_type_source') and step_data_output['collection_type'] is None:
                     collection_type_source = step_data_output['collection_type_source']
                     for input_connection in step['input_connections'].get(collection_type_source, []):
                         input_step = steps[input_connection['id']]
-                        for input_step_data_output in input_step['data_outputs']:
+                        for input_step_data_output in input_step['outputs']:
                             if input_step_data_output['name'] == input_connection['output_name']:
                                 step_data_output['collection_type'] = input_step_data_output.get('collection_type')
         return steps
@@ -789,8 +786,6 @@ class WorkflowContentsManager(UsesAnnotations):
                     # If the tool is installed we attempt to verify input values
                     # and connections, otherwise the last known state will be dumped without modifications.
                     visit_input_values(module.tool.inputs, module.state.inputs, callback)
-                    # FIXME: this removes connection without displaying a message currently!
-                    input_connections = [conn for conn in input_connections if (conn.input_name in data_input_names or conn.non_data_connection)]
 
             # Encode input connections as dictionary
             input_conn_dict = {}
@@ -846,6 +841,8 @@ class WorkflowContentsManager(UsesAnnotations):
                 label = "Input Dataset"
             elif step_type == "data_collection_input":
                 label = "Input Dataset Collection"
+            elif step_type == 'parameter_input':
+                label = "Input Parameter"
             else:
                 raise ValueError("Invalid step_type %s" % step_type)
             if legacy:
diff --git a/lib/galaxy/model/__init__.py b/lib/galaxy/model/__init__.py
index 03347ff6c27..887f49c4818 100644
--- a/lib/galaxy/model/__init__.py
+++ b/lib/galaxy/model/__init__.py
@@ -4192,14 +4192,9 @@ class WorkflowStepConnection(RepresentById):
         self.input_step_id = None
         self.input_name = None
 
-    def set_non_data_connection(self):
-        self.output_name = WorkflowStepConnection.NON_DATA_CONNECTION
-        self.input_name = WorkflowStepConnection.NON_DATA_CONNECTION
-
     @property
     def non_data_connection(self):
-        return (self.output_name == WorkflowStepConnection.NON_DATA_CONNECTION and
-                self.input_name == WorkflowStepConnection.NON_DATA_CONNECTION)
+        return (self.output_name == self.input_name == WorkflowStepConnection.NON_DATA_CONNECTION)
 
     def copy(self):
         # TODO: handle subworkflow ids...
diff --git a/lib/galaxy/tools/parameters/__init__.py b/lib/galaxy/tools/parameters/__init__.py
index b9f108ee965..006b7c783fb 100644
--- a/lib/galaxy/tools/parameters/__init__.py
+++ b/lib/galaxy/tools/parameters/__init__.py
@@ -9,7 +9,7 @@ from boltons.iterutils import remap
 
 from galaxy.util.expressions import ExpressionContext
 from galaxy.util.json import safe_loads
-from .basic import DataCollectionToolParameter, DataToolParameter, RuntimeValue, SelectToolParameter
+from .basic import DataCollectionToolParameter, DataToolParameter, is_runtime_value, runtime_to_json, SelectToolParameter
 from .grouping import Conditional, Repeat, Section, UploadDataset
 
 REPLACE_ON_TRUTHY = object()
@@ -179,11 +179,8 @@ def check_param(trans, param, incoming_value, param_values):
     error = None
     try:
         if trans.workflow_building_mode:
-            if isinstance(value, RuntimeValue):
-                return [{'__class__' : 'RuntimeValue'}, None]
-            if isinstance(value, dict):
-                if value.get('__class__') == 'RuntimeValue':
-                    return [value, None]
+            if is_runtime_value(value):
+                return [runtime_to_json(value), None]
         value = param.from_json(value, trans, param_values)
         param.validate(value, trans)
     except ValueError as e:
diff --git a/lib/galaxy/tools/parameters/basic.py b/lib/galaxy/tools/parameters/basic.py
index d2cae015407..cb3c5ae00b0 100644
--- a/lib/galaxy/tools/parameters/basic.py
+++ b/lib/galaxy/tools/parameters/basic.py
@@ -56,7 +56,7 @@ def contains_workflow_parameter(value, search=False):
 
 def is_runtime_value(value):
     return isinstance(value, RuntimeValue) or (isinstance(value, dict)
-        and value.get("__class__") == "RuntimeValue")
+        and value.get("__class__") in ["RuntimeValue", "ConnectedValue"])
 
 
 def is_runtime_context(trans, other_values):
@@ -155,13 +155,13 @@ class ToolParameter(Dictifiable):
 
     def value_to_basic(self, value, app, use_security=False):
         if is_runtime_value(value):
-            return {'__class__': 'RuntimeValue'}
+            return runtime_to_json(value)
         return self.to_json(value, app, use_security)
 
     def value_from_basic(self, value, app, ignore_errors=False):
         # Handle Runtime and Unvalidated values
         if is_runtime_value(value):
-            return RuntimeValue()
+            return runtime_to_object(value)
         elif isinstance(value, dict) and value.get('__class__') == 'UnvalidatedValue':
             return value['value']
         # Delegate to the 'to_python' method
@@ -2246,7 +2246,7 @@ class BaseJsonToolParameter(ToolParameter):
 
     def value_to_basic(self, value, app, use_security=False):
         if is_runtime_value(value):
-            return {'__class__': 'RuntimeValue'}
+            return runtime_to_json(value)
         return value
 
     def to_json(self, value, app, use_security):
@@ -2323,8 +2323,29 @@ parameter_types = dict(
 )
 
 
+def runtime_to_json(runtime_value):
+    if isinstance(runtime_value, ConnectedValue) or (isinstance(runtime_value, dict) and runtime_value["__class__"] == "ConnectedValue"):
+        return {"__class__": "ConnectedValue"}
+    else:
+        return {"__class__": "RuntimeValue"}
+
+
+def runtime_to_object(runtime_value):
+    if isinstance(runtime_value, ConnectedValue) or (isinstance(runtime_value, dict) and runtime_value["__class__"] == "ConnectedValue"):
+        return ConnectedValue()
+    else:
+        return RuntimeValue()
+
+
 class RuntimeValue(object):
     """
     Wrapper to note a value that is not yet set, but will be required at runtime.
     """
     pass
+
+
+class ConnectedValue(RuntimeValue):
+    """
+    Wrapper to note a value that is not yet set, but will be inferred from a connection.
+    """
+    pass
diff --git a/lib/galaxy/webapps/galaxy/api/workflows.py b/lib/galaxy/webapps/galaxy/api/workflows.py
index 17ccb03773d..481b553f50b 100644
--- a/lib/galaxy/webapps/galaxy/api/workflows.py
+++ b/lib/galaxy/webapps/galaxy/api/workflows.py
@@ -557,8 +557,8 @@ class WorkflowsAPIController(BaseAPIController, UsesStoredWorkflowMixin, UsesAnn
             'annotation'        : inputs.get('__annotation', ''),
             'name'              : module.get_name(),
             'tool_state'        : module.get_state(),
-            'data_inputs'       : module.get_data_inputs(),
-            'data_outputs'      : module.get_data_outputs(),
+            'inputs'            : module.get_all_inputs(connectable_only=True),
+            'outputs'           : module.get_all_outputs(),
             'config_form'       : module.get_config_form(),
             'post_job_actions'  : module.get_post_job_actions(inputs)
         }
diff --git a/lib/galaxy/workflow/modules.py b/lib/galaxy/workflow/modules.py
index 41e1fbba21c..9743d4f42c9 100644
--- a/lib/galaxy/workflow/modules.py
+++ b/lib/galaxy/workflow/modules.py
@@ -30,11 +30,12 @@ from galaxy.tools.parameters import (
 )
 from galaxy.tools.parameters.basic import (
     BooleanToolParameter,
+    ConnectedValue,
     DataCollectionToolParameter,
     DataToolParameter,
     is_runtime_value,
     parameter_types,
-    RuntimeValue,
+    runtime_to_json,
     SelectToolParameter,
     TextToolParameter,
     workflow_building_modes
@@ -151,12 +152,18 @@ class WorkflowModule(object):
         """ This returns inputs displayed in the workflow editor """
         return {}
 
+    def get_all_inputs(self, data_only=False, connectable_only=False):
+        return []
+
     def get_data_inputs(self):
         """ Get configure time data input descriptions. """
+        return self.get_all_inputs(data_only=True)
+
+    def get_all_outputs(self, data_only=False):
         return []
 
     def get_data_outputs(self):
-        return []
+        return self.get_all_outputs(data_only=True)
 
     def get_post_job_actions(self, incoming):
         return []
@@ -172,7 +179,7 @@ class WorkflowModule(object):
         """ Replace connected inputs with placeholder/dummy values. """
         pass
 
-    def get_config_form(self):
+    def get_config_form(self, step=None):
         """ Serializes input parameters of a module into input dictionaries. """
         return {
             'title' : self.name,
@@ -361,7 +368,7 @@ class SubWorkflowModule(WorkflowModule):
             return self.subworkflow.name
         return self.name
 
-    def get_all_inputs(self, data_only=False):
+    def get_all_inputs(self, data_only=False, connectable_only=False):
         """ Get configure time data input descriptions. """
         # Filter subworkflow steps and get inputs
         step_to_input_type = {
@@ -389,10 +396,7 @@ class SubWorkflowModule(WorkflowModule):
                 inputs.append(input)
         return inputs
 
-    def get_data_inputs(self):
-        return self.get_all_inputs(data_only=True)
-
-    def get_data_outputs(self):
+    def get_all_outputs(self, data_only=False):
         outputs = []
         if hasattr(self.subworkflow, 'workflow_outputs'):
             from galaxy.managers.workflows import WorkflowContentsManager
@@ -403,7 +407,7 @@ class SubWorkflowModule(WorkflowModule):
                                                                                   tooltip=False)
             for order_index in sorted(subworkflow_dict['steps']):
                 step = subworkflow_dict['steps'][order_index]
-                data_outputs = subworkflow_dict['steps'][order_index]['data_outputs']
+                data_outputs = subworkflow_dict['steps'][order_index]['outputs']
                 for workflow_output in step['workflow_outputs']:
                     label = workflow_output['label']
                     if not label:
@@ -498,7 +502,7 @@ class InputModule(WorkflowModule):
         state.inputs = dict(input=None)
         return state
 
-    def get_data_inputs(self):
+    def get_all_inputs(self, data_only=False, connectable_only=False):
         return []
 
     def execute(self, trans, progress, invocation_step, use_cached_job=False):
@@ -538,7 +542,7 @@ class InputDataModule(InputModule):
     type = "data_input"
     name = "Input dataset"
 
-    def get_data_outputs(self):
+    def get_all_outputs(self, data_only=False):
         return [dict(name='output', extensions=['input'])]
 
     def get_filter_set(self, connections=None):
@@ -579,7 +583,7 @@ class InputDataCollectionModule(InputModule):
         input_element = Element("param", name="input", label=self.label, type="data_collection", collection_type=collection_type)
         return dict(input=DataCollectionToolParameter(None, input_element, self.trans))
 
-    def get_data_outputs(self):
+    def get_all_outputs(self, data_only=False):
         return [
             dict(
                 name='output',
@@ -604,14 +608,20 @@ class InputParameterModule(WorkflowModule):
         optional = self.state.inputs.get("optional", self.default_optional)
         input_parameter_type = SelectToolParameter(None, XML(
             '''
-            <param name="parameter_type" label="Parameter type" type="select" value="%s">
+            <param name="parameter_type" label="Parameter type" type="select">
                 <option value="text">Text</option>
                 <option value="integer">Integer</option>
                 <option value="float">Float</option>
                 <option value="boolean">Boolean (True or False)</option>
                 <option value="color">Color</option>
             </param>
-            ''' % parameter_type))
+            '''))
+        for i, option in enumerate(input_parameter_type.static_options):
+            option = list(option)
+            if option[1] == parameter_type:
+                # item 0 is option description, item 1 is value, item 2 is "selected"
+                option[2] = True
+                input_parameter_type.static_options[i] = tuple(option)
         return odict([("parameter_type", input_parameter_type),
                       ("optional", BooleanToolParameter(None, Element("param", name="optional", label="Optional", type="boolean", value=optional)))])
 
@@ -635,8 +645,16 @@ class InputParameterModule(WorkflowModule):
         state.inputs = dict(input=None)
         return state
 
-    def get_data_inputs(self):
-        return []
+    def get_all_outputs(self, data_only=False):
+        if data_only:
+            return []
+
+        return [dict(
+            name='output',
+            label=self.label,
+            type=self.state.inputs.get('parameter_type', self.parameter_type),
+            parameter=True,
+        )]
 
     def execute(self, trans, progress, invocation_step, use_cached_job=False):
         step = invocation_step.workflow_step
@@ -651,7 +669,7 @@ class PauseModule(WorkflowModule):
     type = "pause"
     name = "Pause for dataset review"
 
-    def get_data_inputs(self):
+    def get_all_inputs(self, data_only=False, connectable_only=False):
         input = dict(
             name="input",
             label="Dataset for Review",
@@ -659,9 +677,9 @@ class PauseModule(WorkflowModule):
             extensions='input',
             input_type="dataset",
         )
-        return [input]
+        return [input] if not data_only else []
 
-    def get_data_outputs(self):
+    def get_all_outputs(self, data_only=False):
         return [dict(name="output", label="Reviewed Dataset", extensions=['input'])]
 
     def get_runtime_state(self):
@@ -803,14 +821,24 @@ class ToolModule(WorkflowModule):
     def get_inputs(self):
         return self.tool.inputs if self.tool else {}
 
-    def get_all_inputs(self, data_only=False):
+    def get_all_inputs(self, data_only=False, connectable_only=False):
+        if data_only and connectable_only:
+            raise Exception("Must specify at most one of data_only and connectable_only as True.")
+
         inputs = []
         if self.tool:
 
-            def callback(input, prefixed_name, prefixed_label, **kwargs):
+            def callback(input, prefixed_name, prefixed_label, value=None, **kwargs):
                 visible = not hasattr(input, 'hidden') or not input.hidden
+                input_type = input.type
                 is_data = isinstance(input, DataToolParameter) or isinstance(input, DataCollectionToolParameter)
-                skip = data_only and (not visible or not is_data)
+                is_connectable = is_runtime_value(value) and runtime_to_json(value)["__class__"] == "ConnectedValue"
+                if data_only:
+                    skip = not visible or not is_data
+                elif connectable_only:
+                    skip = not visible or not (is_data or is_connectable)
+                else:
+                    skip = not visible
                 if not skip:
                     if isinstance(input, DataToolParameter):
                         inputs.append(dict(
@@ -835,16 +863,14 @@ class ToolModule(WorkflowModule):
                                 label=prefixed_label,
                                 multiple=False,
                                 input_type="parameter",
+                                type=input_type,
                             )
                         )
 
             visit_input_values(self.tool.inputs, self.state.inputs, callback)
         return inputs
 
-    def get_data_inputs(self):
-        return self.get_all_inputs(data_only=True)
-
-    def get_data_outputs(self):
+    def get_all_outputs(self, data_only=False):
         data_outputs = []
         if self.tool:
             for name, tool_output in self.tool.outputs.items():
@@ -890,9 +916,9 @@ class ToolModule(WorkflowModule):
                 )
         return data_outputs
 
-    def get_config_form(self):
+    def get_config_form(self, step=None):
         if self.tool:
-            self.add_dummy_datasets()
+            self.add_dummy_datasets(connections=step and step.input_connections)
             incoming = {}
             params_to_incoming(incoming, self.tool.inputs, self.state.inputs, self.trans.app)
             return self.tool.to_json(self.trans, incoming, workflow_building_mode=True)
@@ -911,22 +937,23 @@ class ToolModule(WorkflowModule):
 
             # Any input needs to have value RuntimeValue or obtain the value from connected steps
             def callback(input, prefixed_name, context, **kwargs):
-                if isinstance(input, DataToolParameter) or isinstance(input, DataCollectionToolParameter):
-                    if connections is not None and steps is not None and self.trans.workflow_building_mode is workflow_building_modes.USE_HISTORY:
-                        if prefixed_name in input_connections_by_name:
-                            connection = input_connections_by_name[prefixed_name]
-                            output_step = next(output_step for output_step in steps if connection.output_step_id == output_step.id)
-                            if output_step.type.startswith('data'):
-                                output_inputs = output_step.module.get_runtime_inputs(connections=connections)
-                                output_value = output_inputs['input'].get_initial_value(self.trans, context)
-                                if isinstance(input, DataToolParameter) and isinstance(output_value, self.trans.app.model.HistoryDatasetCollectionAssociation):
-                                    return output_value.to_hda_representative()
-                                return output_value
-                            return RuntimeValue()
-                        else:
-                            return input.get_initial_value(self.trans, context)
-                    elif connections is None or prefixed_name in input_connections_by_name:
-                        return RuntimeValue()
+                input_type = input.type
+                is_data = input_type in ['data', 'data_collection']
+                if is_data and connections is not None and steps is not None and self.trans.workflow_building_mode is workflow_building_modes.USE_HISTORY:
+                    if prefixed_name in input_connections_by_name:
+                        connection = input_connections_by_name[prefixed_name]
+                        output_step = next(output_step for output_step in steps if connection.output_step_id == output_step.id)
+                        if output_step.type.startswith('data'):
+                            output_inputs = output_step.module.get_runtime_inputs(connections=connections)
+                            output_value = output_inputs['input'].get_initial_value(self.trans, context)
+                            if input_type == "data" and isinstance(output_value, self.trans.app.model.HistoryDatasetCollectionAssociation):
+                                return output_value.to_hda_representative()
+                            return output_value
+                        return ConnectedValue()
+                    else:
+                        return input.get_initial_value(self.trans, context)
+                elif (is_data and connections is None) or prefixed_name in input_connections_by_name:
+                    return ConnectedValue()
             visit_input_values(self.tool.inputs, self.state.inputs, callback)
         else:
             raise ToolMissingException("Tool %s missing. Cannot add dummy datasets." % self.tool_id)
@@ -1300,6 +1327,11 @@ def load_module_sections(trans):
                 "title": "Input Dataset Collection",
                 "description": "Input dataset collection"
             },
+            {
+                "name": "parameter_input",
+                "title": "Parameter Input",
+                "description": "Simple inputs used for workflow logic"
+            },
         ],
     }
 
@@ -1312,12 +1344,7 @@ def load_module_sections(trans):
                     "name": "pause",
                     "title": "Pause Workflow for Dataset Review",
                     "description": "Pause for Review"
-                },
-                {
-                    "name": "parameter_input",
-                    "title": "Parameter Input",
-                    "description": "Simple inputs used for workflow logic"
-                },
+                }
             ],
         }
 
diff --git a/lib/galaxy/workflow/run.py b/lib/galaxy/workflow/run.py
index e696c3d86a7..b75bee7ecac 100644
--- a/lib/galaxy/workflow/run.py
+++ b/lib/galaxy/workflow/run.py
@@ -352,13 +352,10 @@ class WorkflowProgress(object):
         try:
             replacement = step_outputs[output_name]
         except KeyError:
-            if is_data:
-                # Must resolve.
-                template = "Workflow evaluation problem - failed to find output_name %s in step_outputs %s"
-                message = template % (output_name, step_outputs)
-                raise Exception(message)
-            else:
-                replacement = modules.NO_REPLACEMENT
+            # Must resolve.
+            template = "Workflow evaluation problem - failed to find output_name %s in step_outputs %s"
+            message = template % (output_name, step_outputs)
+            raise Exception(message)
         if isinstance(replacement, model.HistoryDatasetCollectionAssociation):
             if not replacement.collection.populated:
                 if not replacement.collection.waiting_for_elements:
@@ -390,11 +387,12 @@ class WorkflowProgress(object):
 
         if self.inputs_by_step_id:
             step_id = step.id
-            if step_id not in self.inputs_by_step_id:
+            if step_id not in self.inputs_by_step_id and 'output' not in outputs:
                 template = "Step with id %s not found in inputs_step_id (%s)"
                 message = template % (step_id, self.inputs_by_step_id)
                 raise ValueError(message)
-            outputs['output'] = self.inputs_by_step_id[step_id]
+            elif step_id in self.inputs_by_step_id:
+                outputs['output'] = self.inputs_by_step_id[step_id]
 
         self.set_step_outputs(invocation_step, outputs)
 
diff --git a/lib/galaxy/workflow/run_request.py b/lib/galaxy/workflow/run_request.py
index 26320d03ae1..ff495019438 100644
--- a/lib/galaxy/workflow/run_request.py
+++ b/lib/galaxy/workflow/run_request.py
@@ -265,6 +265,17 @@ def build_workflow_run_configs(trans, workflow, payload):
             # inputs with referential integrity if parameters are already normalized (coming from tool form).
             normalized_inputs = {}
 
+        if param_map:
+            # disentangle raw parameter dictionaries into formal request structures if we can
+            # to setup proper WorkflowRequestToInputDatasetAssociation, WorkflowRequestToInputDatasetCollectionAssociation
+            # and WorkflowRequestInputStepParameter objects.
+            for step in workflow.steps:
+                normalized_key = step.id
+                if step.type == "parameter_input":
+                    if normalized_key in param_map:
+                        value = param_map.pop(normalized_key)
+                        normalized_inputs[normalized_key] = value["input"]
+
         steps_by_id = workflow.steps_by_id
         # Set workflow inputs.
         for key, input_dict in normalized_inputs.items():
diff --git a/test/api/test_workflows.py b/test/api/test_workflows.py
index 30f45e385f0..f4ac72775a2 100644
--- a/test/api/test_workflows.py
+++ b/test/api/test_workflows.py
@@ -468,8 +468,8 @@ class WorkflowsApiTestCase(BaseWorkflowsApiTestCase):
                 'name',
                 'tool_state',
                 'tooltip',
-                'data_inputs',
-                'data_outputs',
+                'inputs',
+                'outputs',
                 'config_form',
                 'annotation',
                 'post_job_actions',
@@ -494,7 +494,7 @@ steps:
         downloaded_workflow = self._download_workflow(workflow_id, style="editor")
         steps = downloaded_workflow['steps']
         assert len(steps) == 2
-        assert steps['1']['data_outputs'][0]['collection_type'] == 'list:paired'
+        assert steps['1']['outputs'][0]['collection_type'] == 'list:paired'
 
     @skip_without_tool('collection_type_source')
     def test_export_editor_subworkflow_collection_type_source(self):
@@ -525,7 +525,7 @@ steps:
         steps = downloaded_workflow['steps']
         assert len(steps) == 2
         assert steps['1']['type'] == 'subworkflow'
-        assert steps['1']['data_outputs'][0]['collection_type'] == 'list:paired'
+        assert steps['1']['outputs'][0]['collection_type'] == 'list:paired'
 
     def test_import_missing_tool(self):
         workflow = self.workflow_populator.load_workflow_from_resource(name="test_workflow_missing_tool")
diff --git a/test/galaxy_selenium/navigation.yml b/test/galaxy_selenium/navigation.yml
index 23714adfbb7..35159d4a9e9 100644
--- a/test/galaxy_selenium/navigation.yml
+++ b/test/galaxy_selenium/navigation.yml
@@ -236,7 +236,8 @@ workflows:
 workflow_run:
 
   selectors: 
-    input_div: "[step-label='${label}'] .select2-container"
+    input_div: "[step-label='${label}']"
+    input_data_div: "[step-label='${label}'] .select2-container"
     # TODO: put step labels in the DOM ideally
     subworkflow_step_icon: ".portlet-title-icon.fa-sitemap + span"
 
@@ -264,6 +265,9 @@ workflow_editor:
     tool_menu_section_link: '#title___workflow__${section_name}__ a span'
     tool_menu_item_link: 'a#tool-menu-${section_name}-${item_name}'
 
+    connect_icon: 'div.ui-form-element[tour_id="${name}"] .ui-form-connected-icon'
+    collapse_icon: 'div.ui-form-element[tour_id="${name}"] .ui-form-collapsible-icon'
+
     label_input: "[tour_id='__label'] input"
     annotation_input: "[tour_id='__annotation'] textarea"
 
diff --git a/test/selenium_tests/test_workflow_editor.py b/test/selenium_tests/test_workflow_editor.py
index 61ffd23d03e..a014ca9ebf7 100644
--- a/test/selenium_tests/test_workflow_editor.py
+++ b/test/selenium_tests/test_workflow_editor.py
@@ -54,10 +54,7 @@ class WorkflowEditorTestCase(SeleniumTestCase):
         editor = self.components.workflow_editor
 
         name = self.workflow_create_new()
-        editor.canvas_body.wait_for_visible()
-        editor.tool_menu.wait_for_visible()
-        editor.tool_menu_section_link(section_name="inputs").wait_for_and_click()
-        editor.tool_menu_item_link(section_name="inputs", item_name="data_input").wait_for_and_click()
+        self.workflow_editor_add_input(item_name="data_input")
         self.screenshot("workflow_editor_data_input_new")
         editor.label_input.wait_for_and_send_keys("input1")
         editor.annotation_input.wait_for_and_send_keys("my cool annotation")
@@ -85,16 +82,13 @@ class WorkflowEditorTestCase(SeleniumTestCase):
         editor = self.components.workflow_editor
 
         name = self.workflow_create_new()
-        editor.canvas_body.wait_for_visible()
-        editor.tool_menu.wait_for_visible()
-        editor.tool_menu_section_link(section_name="inputs").wait_for_and_click()
-        editor.tool_menu_item_link(section_name="inputs", item_name="data_collection_input").wait_for_and_click()
-        self.screenshot("workflow_editor_data_input_collection_new")
+        self.workflow_editor_add_input(item_name="data_collection_input")
+        self.screenshot("workflow_editor_data_collection_input_new")
         editor.label_input.wait_for_and_send_keys("input1")
         editor.annotation_input.wait_for_and_send_keys("my cool annotation")
         editor.label_input.wait_for_and_click()  # Seems to help force the save of whole annotation.
         self.sleep_for(self.wait_types.UX_RENDER)
-        self.screenshot("workflow_editor_data_input_collection_filled_in")
+        self.screenshot("workflow_editor_data_collection_input_filled_in")
         self.workflow_editor_save_and_close()
         self.workflow_index_open_with_name(name)
         data_input_node = editor.node._(label="input1")
@@ -107,7 +101,82 @@ class WorkflowEditorTestCase(SeleniumTestCase):
 
         data_input_node.destroy.wait_for_and_click()
         data_input_node.wait_for_absent()
-        self.screenshot("workflow_editor_data_input_deleted")
+        self.screenshot("workflow_editor_data_collection_input_deleted")
+
+    @selenium_test
+    def test_integer_input(self):
+        editor = self.components.workflow_editor
+
+        name = self.workflow_create_new()
+        self.workflow_editor_add_input(item_name="parameter_input")
+        self.screenshot("workflow_editor_parameter_input_new")
+
+        editor.label_input.wait_for_and_send_keys("input1")
+        editor.annotation_input.wait_for_and_send_keys("my cool annotation")
+        editor.label_input.wait_for_and_click()  # Seems to help force the save of whole annotation.
+        self.sleep_for(self.wait_types.UX_RENDER)
+        self.screenshot("workflow_editor_parameter_input_filled_in")
+        self.workflow_editor_save_and_close()
+        self.workflow_index_open_with_name(name)
+        data_input_node = editor.node._(label="input1")
+        data_input_node.title.wait_for_and_click()
+
+        label = editor.label_input.wait_for_value()
+        assert label == "input1", label
+        # should work but Galaxy is broken.
+        # assert editor.annotation_input.wait_for_value() == "my cool annotation"
+
+        data_input_node.destroy.wait_for_and_click()
+        data_input_node.wait_for_absent()
+        self.screenshot("workflow_editor_parameter_input_deleted")
+
+    @selenium_test
+    def test_non_data_connections(self):
+        self.open_in_workflow_editor("""
+class: GalaxyWorkflow
+inputs:
+  input_int: integer
+steps:
+  simple_constructs:
+    tool_id: simple_constructs
+    label: tool_exec
+    in:
+      inttest: input_int
+""")
+        self.screenshot("workflow_editor_parameter_connection_simple")
+        self.assert_connected("input_int#output", "simple_constructs#inttest")
+
+        editor = self.components.workflow_editor
+
+        tool_node = editor.node._(label="simple_constructs")
+        tool_input = tool_node.input_terminal(name="inttest")
+        tool_input.wait_for_and_click()
+
+        editor.connector_destroy_callout.wait_for_and_click()
+        self.assert_not_connected("input_int#output", "simple_constructs#inttest")
+        self.screenshot("workflow_editor_parameter_connection_destroyed")
+
+        # When connected, cannot turn it into a RuntimeValue..
+        collapse_input = editor.collapse_icon(name="inttest")
+        collapse_input.wait_for_absent_or_hidden()
+
+        # If it is disconnected, then can specify as RuntimeValue
+        connect_icon = editor.connect_icon(name="inttest")
+        connect_icon.wait_for_visible()
+        connect_icon.wait_for_and_click()
+        collapse_input.wait_for_visible()
+
+        # Also the connector should disappear
+        tool_input.wait_for_absent_or_hidden()
+
+        # Now make it connected again and watch the requestss
+        connect_icon.wait_for_and_click()
+
+        tool_input.wait_for_visible()
+        collapse_input.wait_for_absent_or_hidden()
+
+        self.workflow_editor_connect("input_int#output", "simple_constructs#inttest", screenshot_partial="workflow_editor_parameter_connection_dragging")
+        self.assert_connected("input_int#output", "simple_constructs#inttest")
 
     @selenium_test
     def test_existing_connections(self):
@@ -329,6 +398,16 @@ steps:
 
         return source_id, sink_id
 
+    def workflow_editor_add_input(self, item_name="data_input"):
+        editor = self.components.workflow_editor
+
+        # Make sure we're on the the workflow editor and not clicking the main tool panel.
+        editor.canvas_body.wait_for_visible()
+
+        editor.tool_menu.wait_for_visible()
+        editor.tool_menu_section_link(section_name="inputs").wait_for_and_click()
+        editor.tool_menu_item_link(section_name="inputs", item_name=item_name).wait_for_and_click()
+
     def workflow_editor_destroy_connection(self, sink):
         editor = self.components.workflow_editor
 
diff --git a/test/selenium_tests/test_workflow_run.py b/test/selenium_tests/test_workflow_run.py
index 60f9913fdc3..cc9d1026147 100644
--- a/test/selenium_tests/test_workflow_run.py
+++ b/test/selenium_tests/test_workflow_run.py
@@ -96,6 +96,39 @@ class WorkflowRunTestCase(SeleniumTestCase, UsesHistoryItemAssertions):
         details = self.dataset_populator.get_history_dataset_details(history_id, hid=output_hid)
         assert details["name"] == "moocow suffix", details
 
+    @selenium_test
+    @managed_history
+    def test_step_parameter_inputs(self):
+        self.perform_upload(self.get_filename("1.txt"))
+        self.wait_for_history()
+        self.open_in_workflow_run("""
+class: GalaxyWorkflow
+inputs:
+  input_int: integer
+  input_data: data
+steps:
+  simple_constructs:
+    tool_id: simple_constructs
+    label: tool_exec
+    in:
+      inttest: input_int
+      files_0|file: input_data
+""")
+        workflow_run = self.components.workflow_run
+        input_div_element = workflow_run.input_div(label="input_int").wait_for_visible()
+        input_element = input_div_element.find_element_by_css_selector("input")
+        input_element.clear()
+        input_element.send_keys("12345")
+
+        self.screenshot("workflow_run_step_parameter_input")
+        self.workflow_run_submit()
+        output_hid = 2
+        self.history_panel_wait_for_hid_ok(output_hid, allowed_force_refreshes=1)
+        history_id = self.current_history_id()
+        content = self.dataset_populator.get_history_dataset_content(history_id, hid=output_hid)
+        assert "12345" in content, content
+        assert "chr6_hla_hap2" in content
+
     @selenium_test
     @managed_history
     def test_replacement_parameters_on_subworkflows(self):
@@ -160,7 +193,7 @@ class WorkflowRunTestCase(SeleniumTestCase, UsesHistoryItemAssertions):
     def workflow_run_specify_inputs(self, inputs):
         workflow_run = self.components.workflow_run
         for label, value in inputs.items():
-            input_div_element = workflow_run.input_div(label=label).wait_for_visible()
+            input_div_element = workflow_run.input_data_div(label=label).wait_for_visible()
             self.select2_set_value(input_div_element, "%d: " % value["hid"])
 
     def workflow_run_with_name(self, name):