anchor.diff:                           gs_noreply();
                    (pipeline_done, state()) -> gs_noreply();
                    ({kill_job, term()}, state()) -> gs_noreply().
  +handle_cast({start, Inputs}, S) ->
  +    {noreply, do_start(Inputs, S)};
   handle_cast({update_nodes, Hosts}, S) ->
       {noreply, do_update_nodes(Hosts, S)};
  -handle_cast({inputs, Inputs}, S) ->
  +handle_cast({inputs, {error, E}}, S) ->
  +    % TODO: log an appropriate job event.
  +    {stop, {"fetching inputs failed", E}, S#state{input_pid = none}};
  +handle_cast({inputs, {ok, Inputs}}, S) ->
       {noreply, do_use_inputs(Inputs, S)};
   handle_cast({stage_done, Stage}, S) ->
       {noreply, do_stage_done(Stage, S)};
  
positive.code: kill_job(Reason) ->
      gen_server:cast(self(), {kill_job, Reason}).
  
  -spec use_inputs(pid(), [task_output()]) -> ok.
  use_inputs(Coord, Inputs) ->
      gen_server:cast(Coord, {inputs, Inputs}).
  
  
negative.code: handle_call(_M, _F, S) ->
      {noreply, S}.
  
  -spec handle_cast({start, [task_output()]}, state()) -> gs_noreply();
                   ({update_nodes, [host()]}, state()) -> gs_noreply();
                   ({inputs, {ok, [task_output()]} | {error, term()}}, state()) ->
                           gs_noreply();
                   ({stage_done, stage_name()}, state()) -> gs_noreply();
                   ({submit_tasks, submit_mode(), [task_id()]}, state()) ->
                           gs_noreply();
  