anchor.diff:          }
       }
   ).
  --define(ATM_WORKFLOW_SCHEMA_DRAFT, ?ATM_WORKFLOW_SCHEMA_DRAFT(
  -    return_value, gen_correct_time_series_measurements(), ?CORRECT_ATM_TS_DISPATCH_RULES
  -)).
  +
  +-define(TASK1_SELECTOR(__ATM_LANE_RUN_SELECTOR), {__ATM_LANE_RUN_SELECTOR, <<"pb1">>, <<"task1">>}).
  +-define(TASK2_SELECTOR(__ATM_LANE_RUN_SELECTOR), {__ATM_LANE_RUN_SELECTOR, <<"pb2">>, <<"task2">>}).
  +
  +-define(PB1_SELECTOR(__ATM_LANE_RUN_SELECTOR), {__ATM_LANE_RUN_SELECTOR, <<"pb1">>}).
  +-define(PB2_SELECTOR(__ATM_LANE_RUN_SELECTOR), {__ATM_LANE_RUN_SELECTOR, <<"pb2">>}).
  +
  +-define(PB1_TASK1_STATUS_EXPECTATIONS(Status), [
  +    {task, ?TASK1_SELECTOR({2, 1}), Status},
  +    {parallel_box, ?PB1_SELECTOR({2, 1}), Status}
  +]).
  +-define(PB2_TASK2_STATUS_EXPECTATIONS(Status), [
  +    {task, ?TASK2_SELECTOR({2, 1}), Status},
  +    {parallel_box, ?PB2_SELECTOR({2, 1}), Status}
  +]).
  +
  +-define(CRASHING_TASK_EXECUTION_STOPPED_STEP_MOCK, #atm_step_mock_spec{
  +    strategy = {yield, {error, crashed}},
  +    after_step_exp_state_diff = no_diff
  +}).
   
   -define(INVALID_DISPATCH_RULE, #atm_time_series_dispatch_rule{
       measurement_ts_name_matcher_type = exact,
  
positive.code:                     selector = {2, 1},
                      process_task_result_for_item = #atm_step_mock_spec{
                          before_step_hook = fun atm_workflow_execution_test_utils:pause_workflow_execution/1,
  
                          before_step_exp_state_diff = fun(#atm_mock_call_ctx{workflow_execution_exp_state = ExpState0}) ->
                              {true, expect_execution_stopping_while_processing_lane2(ExpState0, pause)}
                          end
                      },
                      handle_task_execution_stopped = #atm_step_mock_spec{
                          % Ensure cancel prevails against pause
                          before_step_hook = fun atm_workflow_execution_test_utils:cancel_workflow_execution/1,
                          %% already stopped task2 changes status from paused to cancelled
                          before_step_exp_state_diff = build_expect_task2_stopped_exp_state_diff(<<"cancelled">>),
                          after_step_hook = fun atm_workflow_execution_test_utils:pause_workflow_execution/1,
                          after_step_exp_state_diff = build_lane2_task_execution_stopped_after_step_diff(<<"cancelled">>)
                      },
                      handle_lane_execution_stopped = #atm_step_mock_spec{
                          after_step_exp_state_diff = fun(#atm_mock_call_ctx{workflow_execution_exp_state = ExpState0}) ->
                              {true, atm_workflow_execution_exp_state_builder:expect_lane_run_cancelled({2, 1}, ExpState0)}
                          end
                      }
                  }
              ],
              handle_workflow_execution_stopped = #atm_step_mock_spec{
                  after_step_exp_state_diff = fun(#atm_mock_call_ctx{workflow_execution_exp_state = ExpState0}) ->
                      ExpState1 = expect_lane_runs_rerunable([{1, 1}, {2, 1}], ExpState0),
                      {true, atm_workflow_execution_exp_state_builder:expect_workflow_execution_cancelled(ExpState1)}
                  end
              }
          }]
      }).
  
negative.code:                     selector = {2, 1},
                      process_task_result_for_item = #atm_step_mock_spec{
                          before_step_hook = fun atm_workflow_execution_test_utils:pause_workflow_execution/1,
                          before_step_exp_state_diff = lists:flatten([
                              ?PB1_TASK1_STATUS_EXPECTATIONS(stopping),
                              ?PB2_TASK2_STATUS_EXPECTATIONS(paused),
                              {lane_run, {2, 1}, stopping},
                              workflow_stopping
                          ]),
                          % Delay execution of chosen batch to ensure it happens after execution is interrupted
                          strategy = fun(#atm_mock_call_ctx{call_args = [_, _, _, ItemBatch, _]}) ->
                              case ContainsDelayedMeasurementPred(ItemBatch) of
                                  true -> {passthrough_with_delay, timer:seconds(1)};
                                  false -> passthrough
                              end
                          end,
                          after_step_hook = fun atm_workflow_execution_test_utils:delete_offline_session/1
                      },
                      % This is called as part of `handle_workflow_abruptly_stopped`
                      handle_lane_execution_stopped = #atm_step_mock_spec{
                          after_step_exp_state_diff = [
                              {all_tasks, {2, 1}, abruptly, interrupted},
                              {lane_run, {2, 1}, interrupted}
                          ]
                      }
                  }
              ],
              handle_exception = #atm_step_mock_spec{
                  % Ensure interrupt prevails against pause
                  after_step_hook = fun atm_workflow_execution_test_utils:pause_workflow_execution/1,
                  % Already paused task2 changes status to interrupted
                  after_step_exp_state_diff = ?PB2_TASK2_STATUS_EXPECTATIONS(interrupted)
              },
              handle_workflow_abruptly_stopped = #atm_step_mock_spec{
                  % Ensure interrupt prevails against pause
                  before_step_hook = fun atm_workflow_execution_test_utils:pause_workflow_execution/1,
  
                  after_step_exp_state_diff = [workflow_interrupted]
              }
          }]
      }).
  