diff --git a/.hgignore b/.hgignore
index b3ea5101292..b1cb3ca9553 100644
--- a/.hgignore
+++ b/.hgignore
@@ -67,6 +67,7 @@ data_manager_conf.xml
 shed_data_manager_conf.xml
 object_store_conf.xml
 job_metrics_conf.xml
+workflow_schedulers_conf.xml
 config/*
 static/welcome.html.*
 static/welcome.html
diff --git a/config/workflow_schedulers_conf.xml.sample b/config/workflow_schedulers_conf.xml.sample
new file mode 100644
index 00000000000..c7572d9da6e
--- /dev/null
+++ b/config/workflow_schedulers_conf.xml.sample
@@ -0,0 +1,14 @@
+<?xml version="1.0"?>
+<!-- If workflow_schedulers_conf.xml exists it defines the workflow scheduling
+     plugins to load and how to configure them. Currently only the core workflow
+     scheduling is available.
+-->
+<workflow_schedulers default="core">
+  <!-- Each element in this file corresponds to a workflow scheduling plugin
+       in lib/galaxy/workflow/schedulers. -->
+
+  <!-- Core plugin schedules whole workflow at outset inside Galaxy and doesn't
+       require any external dependencies. -->
+  <core id="core" />
+
+</workflow_schedulers>
diff --git a/lib/galaxy/app.py b/lib/galaxy/app.py
index 9cda4879f70..69f27d3d233 100644
--- a/lib/galaxy/app.py
+++ b/lib/galaxy/app.py
@@ -142,6 +142,11 @@ class UniverseApplication( object, config.ConfiguresGalaxyMixin ):
         self.proxy_manager = ProxyManager( self.config )
         # Initialize the external service types
         self.external_service_types = external_service_types.ExternalServiceTypesCollection( self.config.external_service_type_config_file, self.config.external_service_type_path, self )
+
+        from galaxy.workflow import scheduling_manager
+        # Must be initialized after job_config.
+        self.workflow_scheduling_manager = scheduling_manager.WorkflowSchedulingManager( self )
+
         self.model.engine.dispose()
         self.control_worker = GalaxyQueueWorker(self,
                                                 galaxy.queues.control_queue_from_config(self.config),
@@ -150,6 +155,7 @@ class UniverseApplication( object, config.ConfiguresGalaxyMixin ):
         self.control_worker.start()
 
     def shutdown( self ):
+        self.workflow_scheduling_manager.shutdown()
         self.job_manager.shutdown()
         self.object_store.shutdown()
         if self.heartbeat:
@@ -171,3 +177,6 @@ class UniverseApplication( object, config.ConfiguresGalaxyMixin ):
             self.trace_logger = FluentTraceLogger( 'galaxy', self.config.fluent_host, self.config.fluent_port )
         else:
             self.trace_logger = None
+
+    def is_job_handler( self ):
+        return (self.config.track_jobs_in_database and self.job_config.is_handler(self.config.server_name)) or not self.config.track_jobs_in_database
diff --git a/lib/galaxy/config.py b/lib/galaxy/config.py
index 6f1d3194357..7415afc9347 100644
--- a/lib/galaxy/config.py
+++ b/lib/galaxy/config.py
@@ -139,6 +139,7 @@ class Configuration( object ):
         self.collect_outputs_from = [ x.strip() for x in kwargs.get( 'collect_outputs_from', 'new_file_path,job_working_directory' ).lower().split(',') ]
         self.template_path = resolve_path( kwargs.get( "template_path", "templates" ), self.root )
         self.template_cache = resolve_path( kwargs.get( "template_cache_path", "database/compiled_templates" ), self.root )
+        self.workflow_schedulers_config_file = resolve_path( kwargs.get( 'workflow_schedulers_config_file', 'config/workflow_schedulers_conf.xml' ), self.root )
         self.local_job_queue_workers = int( kwargs.get( "local_job_queue_workers", "5" ) )
         self.cluster_job_queue_workers = int( kwargs.get( "cluster_job_queue_workers", "3" ) )
         self.job_queue_cleanup_interval = int( kwargs.get("job_queue_cleanup_interval", "5") )
diff --git a/lib/galaxy/jobs/manager.py b/lib/galaxy/jobs/manager.py
index 20eeb8502c2..4cf56040f57 100644
--- a/lib/galaxy/jobs/manager.py
+++ b/lib/galaxy/jobs/manager.py
@@ -18,7 +18,7 @@ class JobManager( object ):
     """
     def __init__( self, app ):
         self.app = app
-        if (self.app.config.track_jobs_in_database and self.app.job_config.is_handler(self.app.config.server_name)) or not self.app.config.track_jobs_in_database:
+        if self.app.is_job_handler():
             log.debug("Starting job handler")
             self.job_handler = handler.JobHandler( app )
             self.job_queue = self.job_handler.job_queue
diff --git a/lib/galaxy/managers/workflows.py b/lib/galaxy/managers/workflows.py
index af762780253..5d0246a3ba9 100644
--- a/lib/galaxy/managers/workflows.py
+++ b/lib/galaxy/managers/workflows.py
@@ -1,5 +1,6 @@
 from galaxy import model
 from galaxy import exceptions
+from galaxy.workflow import modules
 
 
 class WorkflowsManager( object ):
@@ -49,6 +50,46 @@ class WorkflowsManager( object ):
         self.check_security( trans, workflow_invocation, check_ownership=True, check_accessible=False )
         return workflow_invocation
 
+    def cancel_invocation( self, trans, decoded_invocation_id ):
+        workflow_invocation = self.get_invocation( trans, decoded_invocation_id )
+        cancelled = workflow_invocation.cancel()
+
+        if cancelled:
+            trans.sa_session.add( workflow_invocation )
+            trans.sa_session.flush()
+        else:
+            # TODO: More specific exception?
+            raise exceptions.MessageException( "Cannot cancel an inactive workflow invocation." )
+
+        return workflow_invocation
+
+    def get_invocation_step( self, trans, decoded_workflow_invocation_step_id ):
+        try:
+            workflow_invocation_step = trans.sa_session.query(
+                model.WorkflowInvocationStep
+            ).get( decoded_workflow_invocation_step_id )
+        except Exception:
+            raise exceptions.ObjectNotFound()
+        self.check_security( trans, workflow_invocation_step.workflow_invocation, check_ownership=True, check_accessible=False )
+        return workflow_invocation_step
+
+    def update_invocation_step( self, trans, decoded_workflow_invocation_step_id, action ):
+        if action is None:
+            raise exceptions.RequestParameterMissingException( "Updating workflow invocation step requires an action parameter. " )
+
+        workflow_invocation_step = self.get_invocation_step( trans, decoded_workflow_invocation_step_id )
+        workflow_invocation = workflow_invocation_step.workflow_invocation
+        if not workflow_invocation.active:
+            raise exceptions.RequestParameterInvalidException( "Attempting to modify the state of an completed workflow invocation." )
+
+        step = workflow_invocation_step.workflow_step
+        module = modules.module_factory.from_workflow_step( trans, step )
+        performed_action = module.do_invocation_step_action( step, action )
+        workflow_invocation_step.action = performed_action
+        trans.sa_session.add( workflow_invocation_step )
+        trans.sa_session.flush()
+        return workflow_invocation_step
+
     def build_invocations_query( self, trans, decoded_stored_workflow_id ):
         try:
             stored_workflow = trans.sa_session.query(
diff --git a/lib/galaxy/model/__init__.py b/lib/galaxy/model/__init__.py
index 97328fa77c1..f124ccff1a8 100644
--- a/lib/galaxy/model/__init__.py
+++ b/lib/galaxy/model/__init__.py
@@ -39,6 +39,7 @@ from galaxy.web.framework.helpers import to_unicode
 from galaxy.web.form_builder import (AddressField, CheckboxField, HistoryField,
         PasswordField, SelectField, TextArea, TextField, WorkflowField,
         WorkflowMappingField)
+from galaxy.model.orm import and_, or_
 from sqlalchemy.orm import object_session
 from sqlalchemy.orm import joinedload
 from sqlalchemy.sql.expression import func
@@ -3096,8 +3097,74 @@ class StoredWorkflowMenuEntry( object ):
 
 
 class WorkflowInvocation( object, Dictifiable ):
-    dict_collection_visible_keys = ( 'id', 'update_time', 'workflow_id' )
-    dict_element_visible_keys = ( 'id', 'update_time', 'workflow_id' )
+    dict_collection_visible_keys = ( 'id', 'update_time', 'workflow_id', 'history_id', 'uuid', 'state' )
+    dict_element_visible_keys = ( 'id', 'update_time', 'workflow_id', 'history_id', 'uuid', 'state' )
+    states = Bunch(
+        NEW='new',  # Brand new workflow invocation... maybe this should be same as READY
+        READY='ready',  # Workflow ready for another iteration of scheduling.
+        SCHEDULED='scheduled',  # Workflow has been scheduled.
+        CANCELLED='cancelled',
+        FAILED='failed',
+    )
+
+    @property
+    def active( self ):
+        """ Indicates the workflow invocation is somehow active - and in
+        particular valid actions may be performed on its
+        ``WorkflowInvocationStep``s.
+        """
+        states = WorkflowInvocation.states
+        return self.state in [ states.NEW, states.READY ]
+
+    def cancel( self ):
+        if not self.active:
+            return False
+        else:
+            self.state = WorkflowInvocation.states.CANCELLED
+            return True
+
+    def fail( self ):
+        self.state = WorkflowInvocation.states.FAILED
+
+    def step_states_by_step_id( self ):
+        step_states = {}
+        for step_state in self.step_states:
+            step_id = step_state.workflow_step_id
+            step_states[ step_id ] = step_state
+        return step_states
+
+    def step_invocations_by_step_id( self ):
+        step_invocations = {}
+        for invocation_step in self.steps:
+            step_id = invocation_step.workflow_step_id
+            if step_id not in step_invocations:
+                step_invocations[ step_id ] = []
+            step_invocations[ step_id ].append( invocation_step )
+        return step_invocations
+
+    @staticmethod
+    def poll_active_workflow_ids(
+        sa_session,
+        scheduler=None,
+        handler=None
+    ):
+        and_conditions = [
+            or_(
+                WorkflowInvocation.state == WorkflowInvocation.states.NEW,
+                WorkflowInvocation.state == WorkflowInvocation.states.READY
+            ),
+        ]
+        if scheduler is not None:
+            and_conditions.append( WorkflowInvocation.scheduler == scheduler )
+        if handler is not None:
+            and_conditions.append( WorkflowInvocation.handler == handler )
+
+        query = sa_session.query(
+            WorkflowInvocation
+        ).filter( and_( *and_conditions ) )
+        # Immediately just load all ids into memory so time slicing logic
+        # is relatively intutitive.
+        return map( lambda wi: wi.id, query.all() )
 
     def to_dict( self, view='collection', value_mapper=None ):
         rval = super( WorkflowInvocation, self ).to_dict( view=view, value_mapper=value_mapper )
@@ -3123,14 +3190,63 @@ class WorkflowInvocation( object, Dictifiable ):
 
 
 class WorkflowInvocationStep( object, Dictifiable ):
-    dict_collection_visible_keys = ( 'id', 'update_time', 'job_id', 'workflow_step_id' )
-    dict_element_visible_keys = ( 'id', 'update_time', 'job_id', 'workflow_step_id' )
+    dict_collection_visible_keys = ( 'id', 'update_time', 'job_id', 'workflow_step_id', 'action' )
+    dict_element_visible_keys = ( 'id', 'update_time', 'job_id', 'workflow_step_id', 'action' )
 
     def to_dict( self, view='collection', value_mapper=None ):
         rval = super( WorkflowInvocationStep, self ).to_dict( view=view, value_mapper=value_mapper )
         rval['order_index'] = self.workflow_step.order_index
         return rval
 
+
+class WorkflowRequest( object, Dictifiable ):
+    dict_collection_visible_keys = [ 'id', 'name', 'type', 'state', 'history_id', 'workflow_id' ]
+    dict_element_visible_keys = [ 'id', 'name', 'type', 'state', 'history_id', 'workflow_id' ]
+
+    def to_dict( self, view='collection', value_mapper=None ):
+        rval = super( WorkflowRequest, self ).to_dict( view=view, value_mapper=value_mapper )
+        return rval
+
+
+class WorkflowRequestInputParameter(object, Dictifiable):
+    """ Workflow-related parameters not tied to steps or inputs.
+    """
+    dict_collection_visible_keys = ['id', 'name', 'value', 'type']
+    types = Bunch(
+        REPLACEMENT_PARAMETERS='replacements',
+        META_PARAMETERS='meta',  #
+    )
+
+    def __init__( self, name=None, value=None, type=None ):
+        self.name = name
+        self.value = value
+        self.type = type
+
+
+class WorkflowRequestStepState(object, Dictifiable):
+    """ Workflow step value parameters.
+    """
+    dict_collection_visible_keys = ['id', 'name', 'value', 'workflow_step_id']
+
+    def __init__( self, workflow_step=None, name=None, value=None ):
+        self.workflow_step = workflow_step
+        self.name = name
+        self.value = value
+        self.type = type
+
+
+class WorkflowRequestToInputDatasetAssociation(object, Dictifiable):
+    """ Workflow step input dataset parameters.
+    """
+    dict_collection_visible_keys = ['id', 'workflow_invocation_id', 'workflow_step_id', 'dataset_id', 'name' ]
+
+
+class WorkflowRequestToInputDatasetCollectionAssociation(object, Dictifiable):
+    """ Workflow step input dataset collection parameters.
+    """
+    dict_collection_visible_keys = ['id', 'workflow_invocation_id', 'workflow_step_id', 'dataset_collection_id', 'name' ]
+
+
 class MetadataFile( object ):
 
     def __init__( self, dataset=None, name=None ):
@@ -4038,6 +4154,18 @@ class ToolTagAssociation( ItemTagAssociation ):
         self.value = None
         self.user_value = None
 
+
+class WorkRequestTagAssociation( ItemTagAssociation ):
+    def __init__( self, id=None, user=None, workflow_request_id=None, tag_id=None, user_tname=None, value=None ):
+        self.id = id
+        self.user = user
+        self.workflow_request_id = workflow_request_id
+        self.tag_id = tag_id
+        self.user_tname = user_tname
+        self.value = None
+        self.user_value = None
+
+
 # Item annotation classes.
 
 class HistoryAnnotationAssociation( object ):
diff --git a/lib/galaxy/model/mapping.py b/lib/galaxy/model/mapping.py
index 8f7ca819444..d96f5b41f30 100644
--- a/lib/galaxy/model/mapping.py
+++ b/lib/galaxy/model/mapping.py
@@ -710,6 +710,46 @@ model.WorkflowStep.table = Table( "workflow_step", metadata,
     ## Column( "input_connections", JSONType )
     )
 
+
+model.WorkflowRequestStepState.table = Table(
+    "workflow_request_step_states", metadata,
+    Column( "id", Integer, primary_key=True ),
+    Column( "workflow_invocation_id", Integer, ForeignKey("workflow_invocation.id", onupdate="CASCADE", ondelete="CASCADE" )),
+    Column( "workflow_step_id", Integer, ForeignKey("workflow_step.id" )),
+    Column( "value", JSONType ),
+)
+
+
+model.WorkflowRequestInputParameter.table = Table(
+    "workflow_request_input_parameters", metadata,
+    Column( "id", Integer, primary_key=True ),
+    Column( "workflow_invocation_id", Integer, ForeignKey("workflow_invocation.id", onupdate="CASCADE", ondelete="CASCADE" )),
+    Column( "name", Unicode(255) ),
+    Column( "value", TEXT ),
+    Column( "type", Unicode(255) ),
+)
+
+
+model.WorkflowRequestToInputDatasetAssociation.table = Table(
+    "workflow_request_to_input_dataset", metadata,
+    Column( "id", Integer, primary_key=True ),
+    Column( "name", String(255) ),
+    Column( "workflow_invocation_id", Integer, ForeignKey( "workflow_invocation.id" ), index=True ),
+    Column( "workflow_step_id", Integer, ForeignKey("workflow_step.id") ),
+    Column( "dataset_id", Integer, ForeignKey( "history_dataset_association.id" ), index=True ),
+)
+
+
+model.WorkflowRequestToInputDatasetCollectionAssociation.table = Table(
+    "workflow_request_to_input_collection_dataset", metadata,
+    Column( "id", Integer, primary_key=True ),
+    Column( "name", String(255) ),
+    Column( "workflow_invocation_id", Integer, ForeignKey( "workflow_invocation.id" ), index=True ),
+    Column( "workflow_step_id", Integer, ForeignKey("workflow_step.id") ),
+    Column( "dataset_collection_id", Integer, ForeignKey( "history_dataset_collection_association.id" ), index=True ),
+)
+
+
 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 ),
@@ -728,8 +768,13 @@ model.WorkflowInvocation.table = Table( "workflow_invocation", metadata,
     Column( "id", Integer, primary_key=True ),
     Column( "create_time", DateTime, default=now ),
     Column( "update_time", DateTime, default=now, onupdate=now ),
-    Column( "workflow_id", Integer, ForeignKey( "workflow.id" ), index=True, nullable=False )
-    )
+    Column( "workflow_id", Integer, ForeignKey( "workflow.id" ), index=True, nullable=False ),
+    Column( "state", TrimmedString( 64 ), index=True ),
+    Column( "scheduler", TrimmedString( 255 ), index=True ),
+    Column( "handler", TrimmedString( 255 ), index=True ),
+    Column( 'uuid', UUIDType() ),
+    Column( "history_id", Integer, ForeignKey( "history.id" ), index=True ),
+)
 
 model.WorkflowInvocationStep.table = Table( "workflow_invocation_step", metadata,
     Column( "id", Integer, primary_key=True ),
@@ -737,8 +782,9 @@ model.WorkflowInvocationStep.table = Table( "workflow_invocation_step", metadata
     Column( "update_time", DateTime, default=now, onupdate=now ),
     Column( "workflow_invocation_id", Integer, ForeignKey( "workflow_invocation.id" ), index=True, nullable=False ),
     Column( "workflow_step_id",  Integer, ForeignKey( "workflow_step.id" ), index=True, nullable=False ),
-    Column( "job_id",  Integer, ForeignKey( "job.id" ), index=True, nullable=True )
-    )
+    Column( "job_id",  Integer, ForeignKey( "job.id" ), index=True, nullable=True ),
+    Column( "action", JSONType, nullable=True ),
+)
 
 model.StoredWorkflowUserShareAssociation.table = Table( "stored_workflow_user_share_connection", metadata,
     Column( "id", Integer, primary_key=True ),
@@ -1968,6 +2014,11 @@ mapper( model.StoredWorkflowMenuEntry, model.StoredWorkflowMenuEntry.table,
 
 mapper( model.WorkflowInvocation, model.WorkflowInvocation.table,
     properties=dict(
+        history=relation( model.History ),
+        input_parameters=relation( model.WorkflowRequestInputParameter ),
+        step_states=relation( model.WorkflowRequestStepState ),
+        input_datasets=relation( model.WorkflowRequestToInputDatasetAssociation ),
+        input_dataset_collections=relation( model.WorkflowRequestToInputDatasetCollectionAssociation ),
         steps=relation( model.WorkflowInvocationStep, backref='workflow_invocation', lazy=False ),
         workflow=relation( model.Workflow ) ) )
 
@@ -1976,6 +2027,33 @@ mapper( model.WorkflowInvocationStep, model.WorkflowInvocationStep.table,
         workflow_step = relation( model.WorkflowStep ),
         job = relation( model.Job, backref=backref( 'workflow_invocation_step', uselist=False ) ) ) )
 
+simple_mapping(
+    model.WorkflowRequestInputParameter,
+    workflow_invocation=relation( model.WorkflowInvocation ),
+)
+
+simple_mapping(
+    model.WorkflowRequestStepState,
+    workflow_invocation=relation( model.WorkflowInvocation ),
+    workflow_step=relation( model.WorkflowStep ),
+)
+
+simple_mapping(
+    model.WorkflowRequestToInputDatasetAssociation,
+    workflow_invocation=relation( model.WorkflowInvocation ),
+    workflow_step=relation( model.WorkflowStep ),
+    dataset=relation( model.HistoryDatasetAssociation ),
+)
+
+
+simple_mapping(
+    model.WorkflowRequestToInputDatasetCollectionAssociation,
+    workflow_invocation=relation( model.WorkflowInvocation ),
+    workflow_step=relation( model.WorkflowStep ),
+    dataset_collection=relation( model.HistoryDatasetCollectionAssociation ),
+)
+
+
 mapper( model.MetadataFile, model.MetadataFile.table,
     properties=dict( history_dataset=relation( model.HistoryDatasetAssociation ), library_dataset=relation( model.LibraryDatasetDatasetAssociation ) ) )
 
diff --git a/lib/galaxy/model/migrate/versions/0123_add_workflow_request_tables.py b/lib/galaxy/model/migrate/versions/0123_add_workflow_request_tables.py
new file mode 100644
index 00000000000..653220fad22
--- /dev/null
+++ b/lib/galaxy/model/migrate/versions/0123_add_workflow_request_tables.py
@@ -0,0 +1,144 @@
+"""
+Migration script for workflow request tables.
+"""
+from sqlalchemy import *
+from sqlalchemy.orm import *
+from migrate import *
+from migrate.changeset import *
+from galaxy.model.custom_types import *
+
+import datetime
+now = datetime.datetime.utcnow
+
+import logging
+log = logging.getLogger( __name__ )
+
+metadata = MetaData()
+
+
+WorkflowRequestInputParameter_table = Table(
+    "workflow_request_input_parameters", metadata,
+    Column( "id", Integer, primary_key=True ),
+    Column( "workflow_invocation_id", Integer, ForeignKey("workflow_invocation.id", onupdate="CASCADE", ondelete="CASCADE" )),
+    Column( "name", Unicode(255) ),
+    Column( "type", Unicode(255) ),
+    Column( "value", TEXT ),
+)
+
+
+WorkflowRequestStepState_table = Table(
+    "workflow_request_step_states", metadata,
+    Column( "id", Integer, primary_key=True ),
+    Column( "workflow_invocation_id", Integer, ForeignKey("workflow_invocation.id", onupdate="CASCADE", ondelete="CASCADE" )),
+    Column( "workflow_step_id", Integer, ForeignKey("workflow_step.id" )),
+    Column( "value", JSONType ),
+)
+
+
+WorkflowRequestToInputDatasetAssociation_table = Table(
+    "workflow_request_to_input_dataset", metadata,
+    Column( "id", Integer, primary_key=True ),
+    Column( "name", String(255) ),
+    Column( "workflow_invocation_id", Integer, ForeignKey( "workflow_invocation.id" ), index=True ),
+    Column( "workflow_step_id", Integer, ForeignKey("workflow_step.id") ),
+    Column( "dataset_id", Integer, ForeignKey( "history_dataset_association.id" ), index=True ),
+)
+
+
+WorkflowRequestToInputDatasetCollectionAssociation_table = Table(
+    "workflow_request_to_input_collection_dataset", metadata,
+    Column( "id", Integer, primary_key=True ),
+    Column( "name", String(255) ),
+    Column( "workflow_invocation_id", Integer, ForeignKey( "workflow_invocation.id" ), index=True ),
+    Column( "workflow_step_id", Integer, ForeignKey("workflow_step.id") ),
+    Column( "dataset_collection_id", Integer, ForeignKey( "history_dataset_collection_association.id" ), index=True ),
+)
+
+
+TABLES = [
+    WorkflowRequestInputParameter_table,
+    WorkflowRequestStepState_table,
+    WorkflowRequestToInputDatasetAssociation_table,
+    WorkflowRequestToInputDatasetCollectionAssociation_table,
+]
+
+
+def upgrade(migrate_engine):
+    metadata.bind = migrate_engine
+    print __doc__
+    metadata.reflect()
+
+    for table in TABLES:
+        __create(table)
+
+    History_column = Column( "history_id", Integer, ForeignKey( "history.id" ), nullable=True )
+    State_column = Column( "state", TrimmedString( 64 ) )
+
+    # TODO: Handle indexes correctly
+    SchedulerId_column = Column( "scheduler", TrimmedString(255) )
+    HandlerId_column = Column( "handler", TrimmedString(255) )
+    WorkflowUUID_column = Column( "uuid", UUIDType, nullable=True )
+    __add_column( History_column, "workflow_invocation", metadata )
+    __add_column( State_column, "workflow_invocation", metadata )
+    __add_column( SchedulerId_column, "workflow_invocation", metadata, index_nane="id_workflow_invocation_scheduler" )
+    __add_column( HandlerId_column, "workflow_invocation", metadata, index_name="id_workflow_invocation_handler" )
+    __add_column( WorkflowUUID_column, "workflow_invocation", metadata )
+
+    # All previous invocations have been scheduled...
+    cmd = "UPDATE workflow_invocation SET state = 'scheduled'"
+    try:
+        migrate_engine.execute( cmd )
+    except Exception, e:
+        log.debug( "failed to update past workflow invocation states: %s" % ( str( e ) ) )
+
+    WorkflowInvocationStepAction_column = Column( "action", JSONType, nullable=True )
+    __add_column( WorkflowInvocationStepAction_column, "workflow_invocation_step", metadata )
+ 
+
+def downgrade(migrate_engine):
+    metadata.bind = migrate_engine
+    metadata.reflect()
+
+    for table in TABLES:
+        __drop(table)
+
+    __drop_column( "state", "workflow_invocation", metadata )
+    __drop_column( "scheduler_id", "workflow_invocation", metadata )
+    __drop_column( "uuid", "workflow_invocation", metadata )
+    __drop_column( "history_id", "workflow_invocation", metadata )
+    __drop_column( "handler_id", "workflow_invocation", metadata )
+    __drop_column( "action", "workflow_invocation_step", metadata )
+
+
+def __add_column(column, table_name, metadata, **kwds):
+    try:
+        table = Table( table_name, metadata, autoload=True )
+        column.create( table, **kwds )
+    except Exception as e:
+        print str(e)
+        log.exception( "Adding column %s column failed." % column)
+
+
+def __drop_column( column_name, table_name, metadata ):
+    try:
+        table = Table( table_name, metadata, autoload=True )
+        getattr( table.c, column_name ).drop()
+    except Exception as e:
+        print str(e)
+        log.exception( "Dropping column %s failed." % column_name )
+
+
+def __create(table):
+    try:
+        table.create()
+    except Exception as e:
+        print str(e)
+        log.exception("Creating %s table failed: %s" % (table.name, str( e ) ) )
+
+
+def __drop(table):
+    try:
+        table.drop()
+    except Exception as e:
+        print str(e)
+        log.exception("Dropping %s table failed: %s" % (table.name, str( e ) ) )
diff --git a/lib/galaxy/webapps/galaxy/api/workflows.py b/lib/galaxy/webapps/galaxy/api/workflows.py
index 8d2e27cfe97..588107d9e00 100644
--- a/lib/galaxy/webapps/galaxy/api/workflows.py
+++ b/lib/galaxy/webapps/galaxy/api/workflows.py
@@ -16,7 +16,7 @@ from galaxy.web.base.controller import BaseAPIController, url_for, UsesStoredWor
 from galaxy.web.base.controller import UsesHistoryMixin
 from galaxy.web.base.controller import SharableMixin
 from galaxy.workflow.extract import extract_workflow
-from galaxy.workflow.run import invoke
+from galaxy.workflow.run import invoke, queue_invoke
 from galaxy.workflow.run_request import build_workflow_run_config
 
 log = logging.getLogger(__name__)
@@ -224,7 +224,7 @@ class WorkflowsAPIController(BaseAPIController, UsesStoredWorkflowMixin, UsesHis
 
         # invoke may throw MessageExceptions on tool erors, failure
         # to match up inputs, etc...
-        outputs = invoke(
+        outputs, invocation = invoke(
             trans=trans,
             workflow=workflow,
             workflow_run_config=run_config,
@@ -235,14 +235,19 @@ class WorkflowsAPIController(BaseAPIController, UsesStoredWorkflowMixin, UsesHis
         # Build legacy output - should probably include more information from
         # outputs.
         rval = {}
-        rval['history'] = trans.security.encode_id(history.id)
+        rval['history'] = trans.security.encode_id( history.id )
         rval['outputs'] = []
         for step in workflow.steps:
             if step.type == 'tool' or step.type is None:
                 for v in outputs[ step.id ].itervalues():
                     rval[ 'outputs' ].append( trans.security.encode_id( v.id ) )
 
-        return rval
+        # Newer version of this API just returns the invocation as a dict, to
+        # facilitate migration - produce the newer style response and blend in
+        # the older information.
+        invocation_response = self.__encode_invocation( trans, invocation )
+        invocation_response.update( rval )
+        return invocation_response
 
     @expose_api
     def workflow_dict( self, trans, workflow_id, **kwd ):
@@ -369,6 +374,32 @@ class WorkflowsAPIController(BaseAPIController, UsesStoredWorkflowMixin, UsesHis
         item['url'] = url_for('workflow', id=encoded_id)
         return item
 
+    @expose_api
+    def workflow_request( self, trans, workflow_id, payload, **kwd ):
+        """
+        POST /api/workflows/{encoded_workflow_id}/usage
+
+        Schedule the workflow specified by `workflow_id` to run.
+        """
+        # /usage is awkward in this context but is consistent with the rest of
+        # this module. Would prefer to redo it all to use /invocation(s).
+        # Get workflow + accessibility check.
+        stored_workflow = self.__get_stored_accessible_workflow( trans, workflow_id )
+        workflow = stored_workflow.latest_workflow
+
+        run_config = build_workflow_run_config( trans, workflow, payload )
+        workflow_scheduler_id = payload.get( "scheduler", None )
+        # TODO: workflow scheduler hints
+        work_request_params = dict( scheduler=workflow_scheduler_id )
+
+        workflow_invocation = queue_invoke(
+            trans=trans,
+            workflow=workflow,
+            workflow_run_config=run_config,
+            request_params=work_request_params
+        )
+        return self.encode_all_ids( trans, workflow_invocation.to_dict(), recursive=True )
+
     @expose_api
     def workflow_usage(self, trans, workflow_id, **kwd):
         """
@@ -407,6 +438,87 @@ class WorkflowsAPIController(BaseAPIController, UsesStoredWorkflowMixin, UsesHis
             return self.__encode_invocation( trans, workflow_invocation )
         return None
 
+    @expose_api
+    def cancel_workflow_invocation(self, trans, workflow_id, usage_id, **kwd):
+        """
+        DELETE /api/workflows/{workflow_id}/usage/{usage_id}
+        Cancel the specified workflow invocation.
+
+        :param  workflow_id:      the workflow id (required)
+        :type   workflow_id:      str
+
+        :param  usage_id:      the usage id (required)
+        :type   usage_id:      str
+
+        :raises: exceptions.MessageException, exceptions.ObjectNotFound
+        """
+        decoded_workflow_invocation_id = self.__decode_id( trans, usage_id )
+        workflow_invocation = self.workflow_manager.cancel_invocation( trans, decoded_workflow_invocation_id )
+        return self.__encode_invocation( trans, workflow_invocation )
+
+    @expose_api
+    def workflow_invocation_step(self, trans, workflow_id, usage_id, step_id, **kwd):
+        """
+        GET /api/workflows/{workflow_id}/usage/{usage_id}/steps/{step_id}
+
+        :param  workflow_id:      the workflow id (required)
+        :type   workflow_id:      str
+
+        :param  usage_id:      the usage id (required)
+        :type   usage_id:      str
+
+        :param  step_id:      encoded id of the WorkflowInvocationStep (required)
+        :type   step_id:      str
+
+        :param  payload:       payload containing update action information
+                               for running workflow.
+
+        :raises: exceptions.MessageException, exceptions.ObjectNotFound
+        """
+        decoded_invocation_step_id = self.__decode_id( trans, step_id )
+        invocation_step = self.workflow_manager.get_invocation_step(
+            trans,
+            decoded_invocation_step_id
+        )
+        return self.__encode_invocation_step( trans, invocation_step )
+
+    @expose_api
+    def workflow_invocation_step_update(self, trans, workflow_id, usage_id, step_id, payload, **kwd):
+        """
+        PUT /api/workflows/{workflow_id}/usage/{usage_id}/steps/{step_id}
+        Update state of running workflow step invocation - still very nebulous
+        but this would be for stuff like confirming paused steps can proceed
+        etc....
+
+
+        :param  workflow_id:      the workflow id (required)
+        :type   workflow_id:      str
+
+        :param  usage_id:      the usage id (required)
+        :type   usage_id:      str
+
+        :param  step_id:      encoded id of the WorkflowInvocationStep (required)
+        :type   step_id:      str
+
+        :raises: exceptions.MessageException, exceptions.ObjectNotFound
+        """
+        decoded_invocation_step_id = self.__decode_id( trans, step_id )
+        action = payload.get( "action", None )
+
+        invocation_step = self.workflow_manager.update_invocation_step(
+            trans,
+            decoded_invocation_step_id,
+            action=action,
+        )
+        return self.__encode_invocation_step( trans, invocation_step )
+
+    def __encode_invocation_step( self, trans, invocation_step ):
+        return self.encode_all_ids(
+            trans,
+            invocation_step.to_dict( 'element' ),
+            True
+        )
+
     def __get_stored_accessible_workflow( self, trans, workflow_id ):
         stored_workflow = self.__get_stored_workflow( trans, workflow_id )
 
diff --git a/lib/galaxy/webapps/galaxy/buildapp.py b/lib/galaxy/webapps/galaxy/buildapp.py
index 5de6f2557d1..661c338be42 100644
--- a/lib/galaxy/webapps/galaxy/buildapp.py
+++ b/lib/galaxy/webapps/galaxy/buildapp.py
@@ -227,7 +227,12 @@ def populate_api_routes( webapp, app ):
     webapp.mapper.connect( 'import_shared_workflow_deprecated', '/api/workflows/import', controller='workflows', action='import_shared_workflow_deprecated', conditions=dict( method=['POST'] ) )
     webapp.mapper.connect( 'workflow_usage', '/api/workflows/{workflow_id}/usage', controller='workflows', action='workflow_usage', conditions=dict(method=['GET']))
     webapp.mapper.connect( 'workflow_usage_contents', '/api/workflows/{workflow_id}/usage/{usage_id}', controller='workflows', action='workflow_usage_contents', conditions=dict(method=['GET']))
+    webapp.mapper.connect( 'cancel_workflow_invocation', '/api/workflows/{workflow_id}/usage/{usage_id}', controller='workflows', action='cancel_workflow_invocation', conditions=dict(method=['DELETE']))
 
+    webapp.mapper.connect( 'workflow_invocation_step', '/api/workflows/{workflow_id}/usage/{usage_id}/steps/{step_id}', controller='workflows', action='workflow_invocation_step', conditions=dict(method=['GET']))
+    webapp.mapper.connect( 'workflow_invocation_step_update', '/api/workflows/{workflow_id}/usage/{usage_id}/steps/{step_id}', controller='workflows', action='workflow_invocation_step_update', conditions=dict(method=['PUT']))
+
+    webapp.mapper.connect( 'workflow_request', '/api/workflows/{workflow_id}/usage', controller='workflows', action='workflow_request', conditions=dict( method=['POST'] ) )
     # ============================
     # ===== AUTHENTICATE API =====
     # ============================
diff --git a/lib/galaxy/webapps/galaxy/controllers/workflow.py b/lib/galaxy/webapps/galaxy/controllers/workflow.py
index cb9586c9a41..1ceb9ce915f 100644
--- a/lib/galaxy/webapps/galaxy/controllers/workflow.py
+++ b/lib/galaxy/webapps/galaxy/controllers/workflow.py
@@ -1295,7 +1295,7 @@ class WorkflowController( BaseUIController, SharableMixin, UsesStoredWorkflowMix
                             copy_inputs_to_history=new_history is not None
                         )
 
-                        outputs = invoke(
+                        outputs, invocation = invoke(
                             trans=trans,
                             workflow=workflow,
                             workflow_run_config=run_config
diff --git a/lib/galaxy/work/__init__.py b/lib/galaxy/work/__init__.py
new file mode 100644
index 00000000000..e69de29bb2d
diff --git a/lib/galaxy/work/context.py b/lib/galaxy/work/context.py
new file mode 100644
index 00000000000..47782440b00
--- /dev/null
+++ b/lib/galaxy/work/context.py
@@ -0,0 +1,46 @@
+from galaxy.managers.context import (
+    ProvidesAppContext,
+    ProvidesUserContext,
+    ProvidesHistoryContext
+)
+
+
+class WorkRequestContext( ProvidesAppContext, ProvidesUserContext, ProvidesHistoryContext ):
+    """ Stripped down implementation of Galaxy web transaction god object for
+    work request handling outside of web threads - uses mix-ins shared with
+    GalaxyWebTransaction to provide app, user, and history context convience
+    methods - but nothing related to HTTP handling, mako views, etc....
+
+    Things that only need app shouldn't be consuming trans - but there is a
+    need for actions potentially tied to users and histories and  hopefully
+    this can define that stripped down interface providing access to user and
+    history information - but not dealing with web request and response
+    objects.
+    """
+
+    def __init__( self, app, user=None, history=None ):
+        self.app = app
+        self.security = app.security
+        self.__user = user
+        self.__history = history
+        self.api_inherit_admin = False
+
+    def get_history( self, create=False ):
+        if create:
+            raise NotImplementedError( "Cannot create histories from a work request context." )
+        return self.__history
+
+    def set_history( self ):
+        raise NotImplementedError( "Cannot change histories from a work request context." )
+
+    history = property( get_history, set_history )
+
+    def get_user( self ):
+        """Return the current user if logged in or None."""
+        return self.__user
+
+    def set_user( self, user ):
+        """Set the current user."""
+        raise NotImplementedError( "Cannot change users from a work request context." )
+
+    user = property( get_user, set_user )
diff --git a/lib/galaxy/workflow/modules.py b/lib/galaxy/workflow/modules.py
index df87c85285e..77943060387 100644
--- a/lib/galaxy/workflow/modules.py
+++ b/lib/galaxy/workflow/modules.py
@@ -171,6 +171,21 @@ class WorkflowModule( object ):
         """
         raise TypeError( "Abstract method" )
 
+    def do_invocation_step_action( self, step, action ):
+        """ Update or set the workflow invocation state action - generic
+        extension point meant to allows users to interact with interactive
+        workflow modules. The action object returned from this method will
+        be attached to the WorkflowInvocationStep and be available the next
+        time the workflow scheduler visits the workflow.
+        """
+        raise exceptions.RequestParameterInvalidException( "Attempting to perform invocation step action on module that does not support actions." )
+
+    def recover_mapping( self, step, step_invocations, progress ):
+        """ Re-populate progress object with information about connections
+        from previously executed steps recorded via step_invocations.
+        """
+        raise TypeError( "Abstract method" )
+
 
 class InputModule( WorkflowModule ):
 
@@ -240,6 +255,19 @@ class InputModule( WorkflowModule ):
         state.inputs = dict( input=None )
         return state
 
+    def recover_runtime_state( self, runtime_state ):
+        """ Take secure runtime state from persisted invocation and convert it
+        into a DefaultToolState object for use during workflow invocation.
+        """
+        fake_tool = Bunch( inputs=self.get_runtime_inputs() )
+        state = galaxy.tools.DefaultToolState()
+        state.decode( runtime_state, fake_tool, self.trans.app, secure=False )
+        return state
+
+    def normalize_runtime_state( self, runtime_state ):
+        fake_tool = Bunch( inputs=self.get_runtime_inputs() )
+        return runtime_state.encode( fake_tool, self.trans.app, secure=False )
+
     def encode_runtime_state( self, trans, state ):
         fake_tool = Bunch( inputs=self.get_runtime_inputs() )
         return state.encode( fake_tool, trans.app )
@@ -291,6 +319,9 @@ class InputModule( WorkflowModule ):
         progress.set_outputs_for_input( step, step_outputs )
         return job
 
+    def recover_mapping( self, step, step_invocations, progress ):
+        progress.set_outputs_for_input( step )
+
 
 class InputDataModule( InputModule ):
     type = "data_input"
@@ -439,6 +470,18 @@ class ToolModule( WorkflowModule ):
         )
         self.state.inputs = self.tool.params_from_strings( state, app, **params_from_kwds )
 
+    def recover_runtime_state( self, runtime_state ):
+        """ Take secure runtime state from persisted invocation and convert it
+        into a DefaultToolState object for use during workflow invocation.
+        """
+        state = galaxy.tools.DefaultToolState()
+        app = self.trans.app
+        state.decode( runtime_state, self.tool, app, secure=False )
+        return state
+
+    def normalize_runtime_state( self, runtime_state ):
+        return runtime_state.encode( self.tool, self.trans.app, secure=False )
+
     @classmethod
     def __get_tool_version( cls, trans, tool_id ):
         # Return a ToolVersion if one exists for tool_id.
@@ -664,7 +707,7 @@ class ToolModule( WorkflowModule ):
             param_combinations=param_combinations,
             history=invocation.history,
             collection_info=collection_info,
-            workflow_invocation_uuid=invocation.uuid
+            workflow_invocation_uuid=invocation.uuid.hex
         )
         if collection_info:
             step_outputs = dict( execution_tracker.created_collections )
@@ -731,6 +774,23 @@ class ToolModule( WorkflowModule ):
 
         visit_input_values( self.tool.inputs, self.state.inputs, callback )
 
+    def recover_mapping( self, step, step_invocations, progress ):
+        # Grab a job representing this invocation - for normal workflows
+        # there will be just one job but if this step was mapped over there
+        # may be many.
+        job_0 = step_invocations[ 0 ].job
+
+        outputs = {}
+        for job_output in job_0.output_datasets:
+            replacement_name = job_output.name
+            replacement_value = job_output.dataset
+            # If was a mapping step, grab the output mapped collection for
+            # replacement instead.
+            if replacement_value.hidden_beneath_collection_instance:
+                replacement_value = replacement_value.hidden_beneath_collection_instance
+            outputs[ replacement_name ] = replacement_value
+        progress.set_step_outputs( step, outputs )
+
 
 class WorkflowModuleFactory( object ):
 
@@ -796,6 +856,14 @@ class MissingToolException( Exception ):
     module is missing. """
 
 
+class DelayedWorkflowEvaluation(Exception):
+    pass
+
+
+class CancelWorkflowEvaluation(Exception):
+    pass
+
+
 class WorkflowModuleInjector(object):
     """ Injects workflow step objects from the ORM with appropriate module and
     module generated/influenced state. """
diff --git a/lib/galaxy/workflow/run.py b/lib/galaxy/workflow/run.py
index 679edcf9923..98fcc627659 100644
--- a/lib/galaxy/workflow/run.py
+++ b/lib/galaxy/workflow/run.py
@@ -6,54 +6,110 @@ from galaxy import util
 from galaxy.util.odict import odict
 from galaxy.workflow import modules
 from galaxy.workflow.run_request import WorkflowRunConfig
+from galaxy.workflow.run_request import workflow_run_config_to_request
 
 import logging
 log = logging.getLogger( __name__ )
 
 
-def invoke( trans, workflow, workflow_run_config, populate_state=False ):
+def invoke( trans, workflow, workflow_run_config, workflow_invocation=None, populate_state=False ):
     """ Run the supplied workflow in the supplied target_history.
     """
     if populate_state:
         modules.populate_module_and_state( trans, workflow, workflow_run_config.param_map )
 
-    return WorkflowInvoker(
+    invoker = WorkflowInvoker(
         trans,
         workflow,
         workflow_run_config,
-    ).invoke()
+        workflow_invocation=workflow_invocation,
+    )
+    try:
+        outputs = invoker.invoke()
+    except modules.CancelWorkflowEvaluation:
+        if workflow_invocation:
+            if workflow_invocation.cancel():
+                trans.sa_session.add( workflow_invocation )
+        outputs = []
+    except Exception:
+        log.exception("Failed to execute scheduled workflow.")
+        if workflow_invocation:
+            # Running workflow invocation in background, just mark
+            # persistent workflow invocation as failed.
+            workflow_invocation.fail()
+            trans.sa_session.add( workflow_invocation )
+        else:
+            # Running new transient workflow invocation in legacy
+            # controller action - propage the exception up.
+            raise
+        outputs = []
+
+    if workflow_invocation:
+        # Be sure to update state of workflow_invocation.
+        trans.sa_session.flush()
+
+    return outputs, invoker.workflow_invocation
+
+
+def queue_invoke( trans, workflow, workflow_run_config, request_params ):
+    modules.populate_module_and_state( trans, workflow, workflow_run_config.param_map )
+    workflow_invocation = workflow_run_config_to_request( trans, workflow_run_config, workflow )
+    workflow_invocation.workflow = workflow
+    return trans.app.workflow_scheduling_manager.queue(
+        workflow_invocation,
+        request_params
+    )
 
 
 class WorkflowInvoker( object ):
 
-    def __init__( self, trans, workflow, workflow_run_config ):
+    def __init__( self, trans, workflow, workflow_run_config, workflow_invocation=None ):
         self.trans = trans
         self.workflow = workflow
-        workflow_invocation = model.WorkflowInvocation()
-        workflow_invocation.workflow = self.workflow
-        self.workflow_invocation = workflow_invocation
-        self.progress = WorkflowProgress( self.workflow_invocation, workflow_run_config.inputs )
+        if workflow_invocation is None:
+            invocation_uuid = uuid.uuid1()
 
-        invocation_uuid = uuid.uuid1().hex
+            workflow_invocation = model.WorkflowInvocation()
+            workflow_invocation.workflow = self.workflow
+
+            # In one way or another, following attributes will become persistent
+            # so they are available during delayed/revisited workflow scheduling.
+            workflow_invocation.uuid = invocation_uuid
+            workflow_invocation.history = workflow_run_config.target_history
+
+            self.workflow_invocation = workflow_invocation
+        else:
+            self.workflow_invocation = workflow_invocation
 
-        # In one way or another, following attributes will become persistent
-        # so they are available during delayed/revisited workflow scheduling.
-        self.workflow_invocation.uuid = invocation_uuid
-        self.workflow_invocation.history = workflow_run_config.target_history
         self.workflow_invocation.copy_inputs_to_history = workflow_run_config.copy_inputs_to_history
         self.workflow_invocation.replacement_dict = workflow_run_config.replacement_dict
 
+        module_injector = modules.WorkflowModuleInjector( trans )
+        self.progress = WorkflowProgress( self.workflow_invocation, workflow_run_config.inputs, module_injector )
+
     def invoke( self ):
         workflow_invocation = self.workflow_invocation
         remaining_steps = self.progress.remaining_steps()
+        delayed_steps = False
         for step in remaining_steps:
-            jobs = self._invoke_step( step )
-            for job in util.listify( jobs ):
-                # Record invocation
-                workflow_invocation_step = model.WorkflowInvocationStep()
-                workflow_invocation_step.workflow_invocation = workflow_invocation
-                workflow_invocation_step.workflow_step = step
-                workflow_invocation_step.job = job
+            jobs = None
+            try:
+                jobs = self._invoke_step( step )
+                for job in (util.listify( jobs ) or [None]):
+                    # Record invocation
+                    workflow_invocation_step = model.WorkflowInvocationStep()
+                    workflow_invocation_step.workflow_invocation = workflow_invocation
+                    workflow_invocation_step.workflow_step = step
+                    workflow_invocation_step.job = job
+            except modules.DelayedWorkflowEvaluation:
+                delayed_steps = True
+                self.progress.mark_step_outputs_delayed( step )
+
+        if delayed_steps:
+            state = model.WorkflowInvocation.states.READY
+        else:
+            state = model.WorkflowInvocation.states.SCHEDULED
+        workflow_invocation.state = state
 
         # All jobs ran successfully, so we can save now
         self.trans.sa_session.add( workflow_invocation )
@@ -66,18 +122,35 @@ class WorkflowInvoker( object ):
         jobs = step.module.execute( self.trans, self.progress, self.workflow_invocation, step )
         return jobs
 
+STEP_OUTPUT_DELAYED = object()
+
 
 class WorkflowProgress( object ):
 
-    def __init__( self, workflow_invocation, inputs_by_step_id ):
+    def __init__( self, workflow_invocation, inputs_by_step_id, module_injector ):
         self.outputs = odict()
+        self.module_injector = module_injector
         self.workflow_invocation = workflow_invocation
         self.inputs_by_step_id = inputs_by_step_id
 
     def remaining_steps(self):
+        # Previously computed and persisted step states.
+        step_states = self.workflow_invocation.step_states_by_step_id()
         steps = self.workflow_invocation.workflow.steps
+        remaining_steps = []
+        step_invocations_by_id = self.workflow_invocation.step_invocations_by_step_id()
+        for step in steps:
+            if not hasattr( step, 'module' ):
+                self.module_injector.inject( step )
+                runtime_state = step_states[ step.id ].value
+                step.state = step.module.recover_runtime_state( runtime_state )
 
-        return steps
+            invocation_steps = step_invocations_by_id.get( step.id, None )
+            if invocation_steps:
+                self._recover_mapping( step, invocation_steps )
+            else:
+                remaining_steps.append( step )
+        return remaining_steps
 
     def replacement_for_tool_input( self, step, input, prefixed_name ):
         """ For given workflow 'step' that has had input_connections_by_name
@@ -100,6 +173,8 @@ class WorkflowProgress( object ):
 
     def replacement_for_connection( self, connection ):
         step_outputs = self.outputs[ connection.output_step.id ]
+        if step_outputs is STEP_OUTPUT_DELAYED:
+            raise modules.DelayedWorkflowEvaluation()
         return step_outputs[ connection.output_name ]
 
     def set_outputs_for_input( self, step, outputs={} ):
@@ -111,5 +186,13 @@ class WorkflowProgress( object ):
     def set_step_outputs(self, step, outputs):
         self.outputs[ step.id ] = outputs
 
+    def mark_step_outputs_delayed(self, step):
+        self.outputs[ step.id ] = STEP_OUTPUT_DELAYED
+
+    def _recover_mapping( self, step, step_invocations ):
+        try:
+            step.module.recover_mapping( step, step_invocations, self )
+        except modules.DelayedWorkflowEvaluation:
+            self.mark_step_outputs_delayed( step )
 
 __all__ = [ invoke, WorkflowRunConfig ]
diff --git a/lib/galaxy/workflow/run_request.py b/lib/galaxy/workflow/run_request.py
index 3a9a475eba2..88b9bd01e57 100644
--- a/lib/galaxy/workflow/run_request.py
+++ b/lib/galaxy/workflow/run_request.py
@@ -1,9 +1,15 @@
+import uuid
+
 from galaxy import exceptions
+from galaxy import model
 
 from galaxy.managers import histories
 
 INPUT_STEP_TYPES = [ 'data_input', 'data_collection_input' ]
 
+import logging
+log = logging.getLogger( __name__ )
+
 
 class WorkflowRunConfig( object ):
     """ Wrapper around all the ways a workflow execution can be parameterized.
@@ -231,6 +237,96 @@ def build_workflow_run_config( trans, workflow, payload ):
     return run_config
 
 
+def workflow_run_config_to_request( trans, run_config, workflow ):
+    param_types = model.WorkflowRequestInputParameter.types
+
+    workflow_invocation = model.WorkflowInvocation()
+    workflow_invocation.uuid = uuid.uuid1()
+    workflow_invocation.history = run_config.target_history
+
+    def add_parameter( name, value, type ):
+        parameter = model.WorkflowRequestInputParameter(
+            name=name,
+            value=value,
+            type=type,
+        )
+        workflow_invocation.input_parameters.append( parameter )
+
+    replacement_dict = run_config.replacement_dict
+    for name, value in replacement_dict.iteritems():
+        add_parameter(
+            name=name,
+            value=value,
+            type=param_types.REPLACEMENT_PARAMETERS,
+        )
+
+    for step_id, content in run_config.inputs.iteritems():
+        if content.history_content_type == "dataset":
+            request_to_content = model.WorkflowRequestToInputDatasetAssociation()
+            request_to_content.dataset = content
+            request_to_content.workflow_step_id = step_id
+            workflow_invocation.input_datasets.append( request_to_content )
+        else:
+            request_to_content = model.WorkflowRequestToInputDatasetCollectionAssociation()
+            request_to_content.dataset_collection = content
+            request_to_content.workflow_step_id = step_id
+            workflow_invocation.input_dataset_collections.append( request_to_content )
+
+    for step in workflow.steps:
+        state = step.state
+        serializable_runtime_state = step.module.normalize_runtime_state( state )
+        step_state = model.WorkflowRequestStepState()
+        step_state.workflow_step_id = step.id
+        step_state.value = serializable_runtime_state
+        workflow_invocation.step_states.append( step_state )
+
+    add_parameter( "copy_inputs_to_history", "true" if run_config.copy_inputs_to_history else "false", param_types.META_PARAMETERS )
+    return workflow_invocation
+
+
+def workflow_request_to_run_config( work_request_context, workflow_invocation ):
+    param_types = model.WorkflowRequestInputParameter.types
+
+    history = workflow_invocation.history
+    replacement_dict = {}
+    inputs = {}
+    param_map = {}
+    copy_inputs_to_history = None
+
+    for parameter in workflow_invocation.input_parameters:
+        parameter_type = parameter.type
+
+        if parameter_type == param_types.REPLACEMENT_PARAMETERS:
+            replacement_dict[ parameter.name ] = parameter.value
+        elif parameter_type == param_types.META_PARAMETERS:
+            if parameter.name == "copy_inputs_to_history":
+                copy_inputs_to_history = (parameter.value == "true")
+
+    #for parameter in workflow_invocation.step_parameters:
+    #    step_id = parameter.workflow_step_id
+    #    if step_id not in param_map:
+    #        param_map[ step_id ] = {}
+    #    param_map[ step_id ][ parameter.name ] = parameter.value
+
+    for input_association in workflow_invocation.input_datasets:
+        inputs[ input_association.workflow_step_id ] = input_association.dataset
+
+    for input_association in workflow_invocation.input_dataset_collections:
+        inputs[ input_association.workflow_step_id ] = input_association.dataset_collection
+
+    if copy_inputs_to_history is None:
+        raise exceptions.InconsistentDatabase("Failed to find copy_inputs_to_history parameter loading workflow_invocation from database.")
+
+    workflow_run_config = WorkflowRunConfig(
+        target_history=history,
+        replacement_dict=replacement_dict,
+        inputs=inputs,
+        param_map=param_map,
+        copy_inputs_to_history=copy_inputs_to_history,
+    )
+    return workflow_run_config
+
+
 def __decode_id( trans, workflow_id, model_type="workflow" ):
     try:
         return trans.security.decode_id( workflow_id )
diff --git a/lib/galaxy/workflow/schedulers/__init__.py b/lib/galaxy/workflow/schedulers/__init__.py
new file mode 100644
index 00000000000..a7836e16198
--- /dev/null
+++ b/lib/galaxy/workflow/schedulers/__init__.py
@@ -0,0 +1,41 @@
+""" Module containing Galaxy workflow scheduling plugins. Galaxy's interface
+for workflow scheduling is highly experimental and the interface required for
+scheduling plugins will almost certainly change.
+"""
+from abc import ABCMeta
+from abc import abstractmethod
+
+
+class WorkflowSchedulingPlugin( object ):
+    """ A plugin defining how Galaxy should schedule plugins. By default
+    plugins are passive and should monitor Galaxy's work queue for
+    WorkflowRequests. Inherit from ActiveWorkflowSchedulingPlugin instead if
+    the scheduling plugin should be forced (i.e. if scheduling happen all at
+    once or the request will be stored and monitored outside of Galaxy.)
+    """
+    __metaclass__ = ABCMeta
+
+    @property
+    @abstractmethod
+    def plugin_type( self ):
+        """ Short string providing labelling this plugin """
+
+    def startup( self, app ):
+        """ Called when Galaxy starts up if the plugin is enabled.
+        """
+
+    def shutdown( self ):
+        """ Called when Galaxy is shutting down, workflow scheduling should
+        end.
+        """
+
+
+class ActiveWorkflowSchedulingPlugin( WorkflowSchedulingPlugin ):
+    __metaclass__ = ABCMeta
+
+    @abstractmethod
+    def schedule( self, workflow_invocation ):
+        """ Optionally return one or more commands to instrument job. These
+        commands will be executed on the compute server prior to the job
+        running.
+        """
diff --git a/lib/galaxy/workflow/schedulers/core.py b/lib/galaxy/workflow/schedulers/core.py
new file mode 100644
index 00000000000..4a5fbd24e6a
--- /dev/null
+++ b/lib/galaxy/workflow/schedulers/core.py
@@ -0,0 +1,46 @@
+""" The class defines the stock Galaxy workflow scheduling plugin - currently
+it simply schedules the whole workflow up front when offered.
+"""
+from ..schedulers import ActiveWorkflowSchedulingPlugin
+
+from galaxy.work import context
+
+from galaxy.workflow import run
+from galaxy.workflow import run_request
+
+import logging
+log = logging.getLogger( __name__ )
+
+
+class CoreWorkflowSchedulingPlugin( ActiveWorkflowSchedulingPlugin ):
+    plugin_type = "core"
+
+    def __init__( self, **kwds ):
+        pass
+
+    def startup( self, app ):
+        self.app = app
+
+    def shutdown( self ):
+        pass
+
+    def schedule( self, workflow_invocation ):
+        workflow = workflow_invocation.workflow
+        history = workflow_invocation.history
+        request_context = context.WorkRequestContext(
+            app=self.app,
+            history=history,
+            user=history.user
+        )  # trans-like object not tied to a web-thread.
+        workflow_run_config = run_request.workflow_request_to_run_config(
+            request_context,
+            workflow_invocation
+        )
+        run.invoke(
+            trans=request_context,
+            workflow=workflow,
+            workflow_run_config=workflow_run_config,
+            workflow_invocation=workflow_invocation,
+        )
+
+__all__ = [ CoreWorkflowSchedulingPlugin ]
diff --git a/lib/galaxy/workflow/scheduling_manager.py b/lib/galaxy/workflow/scheduling_manager.py
new file mode 100644
index 00000000000..28bfaaf08f6
--- /dev/null
+++ b/lib/galaxy/workflow/scheduling_manager.py
@@ -0,0 +1,197 @@
+import os
+import time
+import logging
+import threading
+
+from xml.etree import ElementTree
+
+from galaxy import model
+from galaxy.util import plugin_config
+
+import galaxy.workflow.schedulers
+
+log = logging.getLogger( __name__ )
+
+DEFAULT_SCHEDULER_ID = "default"  # well actually this should be called DEFAULT_DEFAULT_SCHEDULER_ID...
+DEFAULT_SCHEDULER_PLUGIN_TYPE = "core"
+
+EXCEPTION_MESSAGE_SHUTDOWN = "Exception raised while attempting to shutdown workflow scheduler."
+EXCEPTION_MESSAGE_NO_SCHEDULERS = "Failed to defined workflow schedulers - no workflow schedulers defined."
+EXCEPTION_MESSAGE_NO_DEFAULT_SCHEDULER = "Failed to defined workflow schedulers - no workflow scheduler found for default id '%s'."
+EXCEPTION_MESSAGE_DUPLICATE_SCHEDULERS = "Failed to defined workflow schedulers - workflow scheduling plugin id '%s' duplicated."
+
+
+class WorkflowSchedulingManager( object ):
+    """ A workflow scheduling manager based loosely on pattern established by
+    ``galaxy.manager.JobManager``. Only schedules workflows on handler
+    processes.
+    """
+
+    def __init__( self, app ):
+        self.app = app
+        self.__job_config = app.job_config
+        self.workflow_schedulers = {}
+        self.active_workflow_schedulers = {}  # Passive workflow schedulers
+                                              # won't need to be monitored I
+                                              # guess.
+        self.request_monitor = None
+        
+        self.__plugin_classes = self.__plugins_dict()
+        self.__init_schedulers()
+
+        if self._is_workflow_handler():
+            log.debug("Starting workflow schedulers")
+            self.__start_schedulers()
+            if self.active_workflow_schedulers:
+                self.__start_request_monitor()
+        else:
+            # Process should not schedule workflows - do nothing.
+            pass
+
+    # Provide a handler config-like interface by delegating to job handler
+    # config. Perhaps it makes sense to let there be explicit workflow
+    # handlers?
+    def _is_workflow_handler( self ):
+        return self.app.is_job_handler()
+
+    def _get_handler( self ):
+        return self.__job_config.get_handler( None )
+
+    def shutdown( self ):
+        for workflow_scheduler in self.workflow_schedulers.itervalues():
+            try:
+                workflow_scheduler.shutdown()
+            except Exception:
+                log.exception( EXCEPTION_MESSAGE_SHUTDOWN )
+        if self.request_monitor:
+            try:
+                self.request_monitor.shutdown()
+            except Exception:
+                log.exception( "Failed to shutdown worklfow request monitor." )
+
+    def queue( self, workflow_invocation, request_params ):
+        workflow_invocation.state = model.WorkflowInvocation.states.NEW
+        scheduler = request_params.get( "scheduler", None ) or self.default_scheduler_id
+        handler = self._get_handler()
+
+        workflow_invocation.scheduler = scheduler
+        workflow_invocation.handler = handler
+
+        sa_session = self.app.model.context
+        sa_session.add( workflow_invocation )
+        sa_session.flush()
+        return workflow_invocation
+
+    def __start_schedulers( self ):
+        for workflow_scheduler in self.workflow_schedulers.itervalues():
+            workflow_scheduler.startup( self.app )
+
+    def __plugins_dict( self ):
+        return plugin_config.plugins_dict( galaxy.workflow.schedulers, 'plugin_type' )
+
+    def __init_schedulers( self ):
+        config_file = self.app.config.workflow_schedulers_config_file
+        use_default_scheduler = False
+        if not config_file:
+            log.info( "Not workflow schedulers plugin config file defined, using default scheduler." )
+            use_default_scheduler = True
+        elif not os.path.exists( config_file ):
+            log.info( "Cannot find workflow schedulers plugin config file '%s', using default scheduler." % config_file )
+            use_default_scheduler = True
+
+        if use_default_scheduler:
+            self.__init_default_scheduler()
+        else:
+            plugins_element = ElementTree.parse( config_file ).getroot()
+            self.__init_schedulers_for_element( plugins_element )
+
+    def __init_default_scheduler( self ):
+        self.default_scheduler_id = DEFAULT_SCHEDULER_ID
+        self.__init_plugin( DEFAULT_SCHEDULER_PLUGIN_TYPE )
+
+    def __init_schedulers_for_element( self, plugins_element ):
+        plugins_kwds = dict( plugins_element.items() )
+        self.default_scheduler_id = plugins_kwds.get( 'default', DEFAULT_SCHEDULER_ID )
+
+        for plugin_element in plugins_element.getchildren():
+            plugin_type = plugin_element.tag
+            plugin_kwds = dict( plugin_element.items() )
+            plugin_kwds.update( self.extra_kwargs )
+            workflow_scheduler_id = plugin_kwds.get( 'id', None )
+            self.__init_plugin( plugin_type, workflow_scheduler_id, **plugin_kwds )
+
+        if not self.workflow_schedulers:
+            raise Exception( EXCEPTION_MESSAGE_NO_SCHEDULERS )
+        if self.default_scheduler_id not in self.workflow_schedulers:
+            raise Exception( EXCEPTION_MESSAGE_NO_DEFAULT_SCHEDULER % self.default_scheduler_id )
+
+    def __init_plugin( self, plugin_type, workflow_scheduler_id=None, **kwds ):
+        workflow_scheduler_id = workflow_scheduler_id or self.default_scheduler_id
+
+        if workflow_scheduler_id in self.workflow_schedulers:
+            raise Exception( EXCEPTION_MESSAGE_DUPLICATE_SCHEDULERS % workflow_scheduler_id )
+
+        workflow_scheduler = self.__plugin_classes[ plugin_type ]( **kwds )
+        self.workflow_schedulers[ workflow_scheduler_id ] = workflow_scheduler
+        if isinstance( workflow_scheduler, galaxy.workflow.schedulers.ActiveWorkflowSchedulingPlugin ):
+            self.active_workflow_schedulers[ workflow_scheduler_id ] = workflow_scheduler
+            
+    def __start_request_monitor( self ):
+        self.request_monitor = WorkflowRequestMonitor( self.app, self )
+
+
+class WorkflowRequestMonitor( object ):
+
+    def __init__( self, app, workflow_scheduling_manager ):
+        self.app = app
+        self.active = True
+        self.workflow_scheduling_manager = workflow_scheduling_manager
+        self.monitor_thread = threading.Thread( name="WorkflowRequestMonitor.monitor_thread", target=self.__monitor )
+        self.monitor_thread.setDaemon( True )
+        self.monitor_thread.start()
+
+    def __monitor( self ):
+        to_monitor = self.workflow_scheduling_manager.active_workflow_schedulers
+        while self.active:
+            for workflow_scheduler_id, workflow_scheduler in to_monitor.iteritems():
+                if not self.active:
+                    return
+
+                self.__schedule( workflow_scheduler_id, workflow_scheduler )
+                time.sleep(1)  # TODO: wake if stopped
+
+    def __schedule( self, workflow_scheduler_id, workflow_scheduler ):
+        invocation_ids = self.__active_invocation_ids( workflow_scheduler_id )
+        for invocation_id in invocation_ids:
+            self.__attempt_schedule( invocation_id, workflow_scheduler )
+            if not self.active:
+                return
+
+    def __attempt_schedule( self, invocation_id, workflow_scheduler ):
+        sa_session = self.app.model.context
+        workflow_invocation = sa_session.query( model.WorkflowInvocation ).get( invocation_id )
+
+        if not workflow_invocation or not workflow_invocation.active:
+            return False
+
+        try:
+            workflow_scheduler.schedule( workflow_invocation )
+        except Exception:
+            # TODO: eventually fail this - or fail it right away?
+            log.exception( "Exception raised while attempting to schedule workflow request." )
+            return False
+
+        # A workflow was obtained and scheduled...
+        return True
+
+    def __active_invocation_ids( self, scheduler_id ):
+        sa_session = self.app.model.context
+        handler = self.app.config.server_name
+        return model.WorkflowInvocation.poll_active_workflow_ids(
+            sa_session,
+            scheduler=scheduler_id,
+            handler=handler,
+        )
+
+    def shutdown( self ):
+        self.active = False
diff --git a/test/api/test_workflows.py b/test/api/test_workflows.py
index dab0146425d..c064377cb8a 100644
--- a/test/api/test_workflows.py
+++ b/test/api/test_workflows.py
@@ -135,9 +135,28 @@ class WorkflowsApiTestCase( api.ApiTestCase ):
         # TODO: This should really be a post to workflows/<workflow_id>/run or
         # something like that.
         run_workflow_response = self._post( "workflows", data=workflow_request )
+
+        invocation_id = run_workflow_response.json()[ "id" ]
+        invocation = self._invocation_details( workflow_request[ "workflow_id" ], invocation_id )
+        assert invocation[ "state" ] == "scheduled", invocation
+
         self._assert_status_code_is( run_workflow_response, 200 )
         self.dataset_populator.wait_for_history( history_id, assert_ok=True )
 
+    def test_workflow_request( self ):
+        workflow = self.workflow_populator.load_workflow( name="test_for_queue" )
+        workflow_request, history_id = self._setup_workflow_run( workflow )
+        # TODO: This should really be a post to workflows/<workflow_id>/run or
+        # something like that.
+        url = "workflows/%s/request" % ( workflow_request[ "workflow_id" ] )
+        del workflow_request[ "workflow_id" ]
+        run_workflow_response = self._post( url, data=workflow_request )
+
+        self._assert_status_code_is( run_workflow_response, 200 )
+        # Give some time for workflow to get scheduled before scanning the history.
+        time.sleep( 5 )
+        self.dataset_populator.wait_for_history( history_id, assert_ok=True )
+
     def test_cannot_run_inaccessible_workflow( self ):
         workflow = self.workflow_populator.load_workflow( name="test_for_run_cannot_access" )
         workflow_request, history_id = self._setup_workflow_run( workflow )
@@ -563,11 +582,18 @@ class WorkflowsApiTestCase( api.ApiTestCase ):
     @skip_without_tool( "cat1" )
     def test_invocation_usage( self ):
         workflow_id, usage = self._run_workflow_once_get_invocation( "test_usage")
-        usage_details = self._invocation_details( workflow_id, usage[ "id" ] )
+        invocation_id = usage[ "id" ]
+        usage_details = self._invocation_details( workflow_id, invocation_id )
         # Assert some high-level things about the structure of data returned.
         self._assert_has_keys( usage_details, "inputs", "steps" )
-        for step in usage_details[ "steps" ]:
+        invocation_steps = usage_details[ "steps" ]
+        for step in invocation_steps:
             self._assert_has_keys( step, "workflow_step_id", "order_index", "id" )
+        an_invocation_step = invocation_steps[ 0 ]
+        step_id = an_invocation_step[ "id" ]
+        step_response = self._get( "workflows/%s/usage/%s/steps/%s" % ( workflow_id, invocation_id, step_id ) )
+        self._assert_status_code_is( step_response, 200 )
+        self._assert_has_keys( step_response.json(), "id", "order_index" )
 
     def _invocation_details( self, workflow_id, invocation_id ):
         invocation_details_response = self._get( "workflows/%s/usage/%s" % ( workflow_id, invocation_id ) )
diff --git a/test/unit/test_galaxy_mapping.py b/test/unit/test_galaxy_mapping.py
index 4aa1672bbce..468b0a3d91b 100644
--- a/test/unit/test_galaxy_mapping.py
+++ b/test/unit/test_galaxy_mapping.py
@@ -1,6 +1,7 @@
 # -*- coding: utf-8 -*-
 import unittest
 import galaxy.model.mapping as mapping
+import uuid
 
 
 class MappingTests( unittest.TestCase ):
@@ -358,6 +359,54 @@ class MappingTests( unittest.TestCase ):
 
         assert contents_iter_names( ids=[ d1.id, d3.id ] ) == [ "1", "3" ]
 
+    def test_workflows( self ):
+        model = self.model
+        user = model.User(
+            email="testworkflows@bx.psu.edu",
+            password="password"
+        )
+        stored_workflow = model.StoredWorkflow()
+        stored_workflow.user = user
+        workflow = model.Workflow()
+        workflow_step = model.WorkflowStep()
+        workflow.steps = [ workflow_step ]
+        workflow.stored_workflow = stored_workflow
+
+        self.persist( workflow )
+        assert workflow_step.id is not None
+
+        invocation_uuid = uuid.uuid1()
+
+        workflow_invocation = model.WorkflowInvocation()
+        workflow_invocation.uuid = invocation_uuid
+
+        workflow_invocation_step1 = model.WorkflowInvocationStep()
+        workflow_invocation_step1.workflow_invocation = workflow_invocation
+        workflow_invocation_step1.workflow_step = workflow_step
+
+        workflow_invocation_step2 = model.WorkflowInvocationStep()
+        workflow_invocation_step2.workflow_invocation = workflow_invocation
+        workflow_invocation_step2.workflow_step = workflow_step
+
+        workflow_invocation.workflow = workflow
+
+        h1 = model.History( name="WorkflowHistory1", user=user)
+        d1 = self.new_hda( h1, name="1" )
+        workflow_request_dataset = model.WorkflowRequestToInputDatasetAssociation()
+        workflow_request_dataset.workflow_invocation = workflow_invocation
+        workflow_request_dataset.workflow_step = workflow_step
+        workflow_request_dataset.dataset = d1
+        self.persist( workflow_invocation )
+        assert workflow_request_dataset is not None
+        assert workflow_invocation.id is not None
+
+        self.expunge()
+
+        loaded_invocation = self.query( model.WorkflowInvocation ).get( workflow_invocation.id )
+        assert loaded_invocation.uuid == invocation_uuid, "%s != %s" % (loaded_invocation.uuid, invocation_uuid)
+        assert loaded_invocation
+        assert len( loaded_invocation.steps ) == 2
+
     def new_hda( self, history, **kwds ):
         return history.add_dataset( self.model.HistoryDatasetAssociation( create_dataset=True, sa_session=self.model.session, **kwds ) )
 
diff --git a/test/unit/workflows/__init__.py b/test/unit/workflows/__init__.py
new file mode 100644
index 00000000000..e69de29bb2d