anchor.diff:      {noreply, S}.
   
   -spec handle_cast({update_nodes, [host()]}, state()) -> gs_noreply();
  +                 ({inputs, [task_output()]}, state()) -> gs_noreply();
                    ({stage_done, stage_name()}, state()) -> gs_noreply();
                    ({submit_tasks, submit_mode(), [task_id()]}, state()) ->
                            gs_noreply();
  
positive.code:         {error, E} ->
              {stop, E};
          K:V ->
              {stop, disco:format("job init error: ~p:~p", [K, V])}
      end.
  
  
negative.code:     {JobInfo#jobinfo{jobname = JobName, jobfile = JobFile}, Hosts}.
  
  -spec init_state(jobinfo(), [host()]) -> state().
  init_state(#jobinfo{jobname  = _JobName,
                      schedule = Schedule,
                      inputs   = Inputs,
                      pipeline = Pipeline} = JobInfo, Hosts) ->
      % Create a dummy completed 'input' task.
      Tasks = gb_trees:from_orddict([{input,
                                      #task_info{spec = input,
                                                 outputs = Inputs}}]),
      lager:info("initialized job ~p with pipeline ~p and inputs ~p",
                 [_JobName, Pipeline, Inputs]),
      % Convert inputs into pipeline form.
      Coord = self(),
      InputPid = spawn_link(fun() -> do_preprocess_inputs(Coord, Inputs) end),
      #state{jobinfo    = JobInfo,
             pipeline   = Pipeline,
             schedule   = Schedule,
             input_pid  = InputPid,
             hosts      = gb_sets:from_list(Hosts),
             tasks      = Tasks}.
  
  %% We need to convert any dir:// urls into pipeline form, which
  %% involves reading and parsing the dir files.  We hence do this in a
  %% separate process.
  
  do_preprocess_inputs(Coord, Inputs) ->
      % TODO.
      use_inputs(Coord, Inputs).
  
  %% ===================================================================
  %% state access and update utils
  
  -spec stage_outputs(stage_name(), state()) -> [{task_id(), [task_output()]}].
  stage_outputs(Stage, #state{stage_info = SI, tasks = Tasks}) ->
      Done = (jc_utils:stage_info(Stage, SI))#stage_info.done,
      [{Id, jc_utils:task_outputs(Id, Tasks)} || Id <- Done].
  
  %% ===================================================================
  %% Callback implementations.
  
  -spec do_use_inputs([task_output()], state()) -> state().
  do_use_inputs(Inputs, #state{jobinfo = JobInfo} = S) ->
      % Mark the 'input' stage as done, and send notification.
      InputStage = #stage_info{all = 1, done = [input]},
      SI = gb_trees:from_orddict([{?INPUT, InputStage}]),
      stage_done(?INPUT),
      S#state{jobinfo    = JobInfo#jobinfo{inputs = Inputs},
              input_pid  = none,
              stage_info = SI}.
  
  -spec do_update_nodes([host()], state()) -> state().
  do_update_nodes(Hosts, S) ->
  