anchor.diff:                   ({kill_job, term()}, state()) -> gs_noreply().
   handle_cast({update_nodes, Hosts}, S) ->
       {noreply, do_update_nodes(Hosts, S)};
  +handle_cast({inputs, Inputs}, S) ->
  +    {noreply, do_use_inputs(Inputs, S)};
  +handle_cast({stage_done, Stage}, S) ->
  +    {noreply, do_stage_done(Stage, S)};
   handle_cast({submit_tasks, Mode, Tasks}, S) ->
       {noreply, do_submit_tasks(Mode, Tasks, S)};
  -handle_cast({stage_done, Stage}, S) ->
  -    {noreply, do_stage_done(Stage, S)};
   handle_cast({task_done, TaskId, Host, Results}, S) ->
       {noreply, do_task_done(TaskId, Host, Results, S)};
   handle_cast(pipeline_done, S) ->
  
positive.code:                 schedule         :: task_schedule(),
                  next_taskid = 0  :: task_id(),
                  next_runid  = 0  :: task_run_id(),
                  % cluster membership: [host()]
                  hosts      = gb_sets:empty()  :: gb_set(),
                  % input | task_id() -> task_info().
  
negative.code:         {error, E} ->
              {stop, E};
          K:V ->
              lager:error("job submission failed: ~p", [erlang:get_stacktrace()]),
              {stop, disco:format("job init error: ~p:~p", [K, V])}
      end.
  
  