anchor.diff:              end
       end.
   
  +-spec send_term_signal_to_task(task_id(), state()) -> state().
   send_term_signal_to_task(TaskId, S) ->
       #state{tasks = Tasks} = S1 = mark_task_inputs_done(S, TaskId),
       TaskInfo = jc_utils:task_info(TaskId, Tasks),
  
positive.code:         end),
      add_inputs_to_data_map(S1, NewInputs).
  
  mark_task_inputs_done(S, TaskId) ->
      update_taskspec(S, TaskId,
          fun(TaskSpec) ->
  
negative.code:                                [Stage, Since], Event)
      end.
  
  -spec mark_stage_finished(stage_name(), state()) -> state().
  mark_stage_finished(Stage, #state{stage_info = SI} = S) ->
      StageInfo0 = jc_utils:stage_info(Stage, SI),
      StageInfo = StageInfo0#stage_info{finished = true},
      SI1 = jc_utils:update_stage(Stage, StageInfo, SI),
      S#state{stage_info = SI1}.
  
  -spec do_next_stage(stage_name(), state()) -> state().
  do_next_stage(Stage, #state{pipeline = P, stage_info = SI} = S) ->
      case pipeline_utils:next_stage(P, Stage) of
          done ->
  