anchor.diff:          end),
       add_inputs_to_data_map(S1, NewInputs).
   
  +-spec mark_task_inputs_done(state(), task_id()) -> state().
   mark_task_inputs_done(S, TaskId) ->
       update_taskspec(S, TaskId,
           fun(TaskSpec) ->
  
positive.code:             S1
      end.
  
  
  send_termination_signal(Stage, #state{stage_info = SI, pending = Pending} = S) ->
      S2 = lists:foldl(fun(TaskId, S1) ->
                      send_term_signal_to_task(TaskId, S1)
  
negative.code:     end,
      S.
  
  -spec save_ddfs(jobname(), [[url()]]) -> ok.
  save_ddfs(JobName, Results) ->
      Tag = list_to_binary(disco:format("disco:results:~s", [JobName])),
      T = <<"tag://", Tag/binary>>,
  