anchor.diff:          TaskInfo#task_info{spec=Fun(TaskSpec)}, Tasks),
       S#state{tasks = Tasks1}.
   
  +-spec add_inputs_to_spec(state(), task_id(), [{input_id(), data_input()}]) -> state().
   add_inputs_to_spec(S, TaskId, NewInputs) ->
  +    InputIds = [Id|| {Id,_} <- NewInputs],
       S1 = update_taskspec(S, TaskId,
           fun(TaskSpec) ->
               CurrentInputs = TaskSpec#task_spec.input,
  -            TaskSpec#task_spec{input = CurrentInputs ++ NewInputs}
  +            TaskSpec#task_spec{input = CurrentInputs ++ InputIds}
           end),
       add_inputs_to_data_map(S1, NewInputs).
   
  
positive.code:     make_stage_tasks(Stage, Grouping, Rest, S2,
                       {TaskNum + 1, [NextTaskId | Acc]}).
  
  add_inputs_to_data_map(#state{data_map = OldDataMap} = S, Inputs) ->
      DataMap = lists:foldl(
                  fun({InputId, DataInput}, DM) ->
  
negative.code:             TaskSpec#task_spec{all_inputs = true}
          end).
  
  -spec send_outputs_to_consumer(state(), task_id(), grouped_output()) -> state().
  send_outputs_to_consumer(S, TaskId, {_, Inputs}) ->
      #state{tasks = Tasks} = S1 = add_inputs_to_spec(S, TaskId, Inputs),
      TaskInfo = jc_utils:task_info(TaskId, Tasks),
  