<commit_before>module XYZ 
  module WorkflowAdapter
    module RuoteParticipant
      class Top
        include ::Ruote::LocalParticipant
        def initialize(opts=nil)
          @opts = opts
        end

        def get_params(workitem)
          task_info = get_and_delete_task_info(workitem)
          params = {"task_id" => workitem.params["task_id"]}
          workflow = task_info["workflow"]
          params.merge!("workflow" => workflow)
          params.merge!("task" => task_info["task"])
          params.merge!("action" => task_info["action"])
          params.merge!("top_task_idh" => task_info["top_task_idh"])
          params.merge!("top_task" => workflow.top_task)
          params
        end

        def set_result_succeeded(workitem,new_result,task,action)
          update_hash = {
            :status => "succeeded",
            :result => TaskAction::Result::Succeeded.new()
          }             
          task.update(update_hash)
          action.update_state_change_status(task.model_handle,:completed)  #this updates pending state
          set_result_succeeded__stack(workitem,new_result,task,action)
        end

        def set_result_failed(workitem,new_result,task,action)
          error = 
            if not new_result[:statuscode] == 0
              CommandAndControl::Error::Communication.new
            else
              data = new_result[:data]
              if data and data[:status] == :failed and (data[:error]||{})[:formatted_exception]
                CommandAndControl::Error::FailedResponse.new(data[:error][:formatted_exception])
              else
                CommandAndControl::Error.new 
              end
          end
          update_hash = {
            :status => "failed",
            :result => TaskAction::Result::Failed.new(error)
          }             
          task.update(update_hash)
        end

        def set_result_timeout(workitem,new_result,task)
          update_hash = {
            :status => "failed",
            :result => TaskAction::Result::Failed.new(CommandAndControl::Error::Timeout.new)
          }             
          task.update(update_hash)
        end

         private
        #if use must coordinate with concurrence merge type
        def set_result_succeeded__stack(workitem,new_result,task,action)
          workitem.fields["result"] = {:action_completed => action.type}
        end
        def get_and_delete_task_info(workitem)
          params = workitem.params
          Ruote::TaskInfo.get_and_delete(params["task_id"],params["task_type"])
        end
      end

      class DetectCreatedNodeIsReady < Top
        def consume(workitem)
          params = get_params(workitem) 
          task_id,action,workflow,task,task_end,top_task = %w{task_id action workflow task task_end top_task}.map{|k|params[k]}
          callbacks = {
            :on_msg_received => proc do |msg|
              result = {:type => :completed_create_node, :task_id => task_id} 
              event = task.add_event(:complete_succeeded, top_task,result)
              pp [:found,msg[:senderid]]
              task[:executable_action][:node].update_operational_status!(:running)
              set_result_succeeded(workitem,result,task,action)
              action.get_and_propagate_dynamic_attributes(result)
              self.reply_to_engine(workitem)
            end,
            :on_timeout => proc do 
              self.reply_to_engine(workitem)
            end
          }
          num_poll_cycles = 10
          poll_cycle = 6
          context = {:callbacks => callbacks, :expected_count => 1,:count => num_poll_cycles,:poll_cycle => poll_cycle} 
          workflow.poll_to_detect_node_ready(action[:node],context)
        end
      end

      #TODO: currently; just using above; either use this or get rid of it
      class DetectIfNodeIsResponding < Top
        def consume(workitem)
          params = get_params(workitem) 
          action,task,workflow = %w{action task workflow}.map{|k|params[k]}
          callbacks = {
            :on_msg_received => proc do |msg|
              pp [:found,msg[:senderid]]
              task[:executable_action][:node].update_operational_status!(:running)
              self.reply_to_engine(workitem)
            end,
            :on_timeout => proc do 
              self.reply_to_engine(workitem)
            end
          }
          poll_cycle = 2
          context = {:callbacks => callbacks, :expected_count => 1,:count => 1,:poll_cycle => poll_cycle} 
          workflow.poll_to_detect_node_ready(action[:node],context)
        end
      end
      class ExecuteOnNode < Top
        #LockforDebug = Mutex.new
        def consume(workitem)
          #LockforDebug.synchronize{pp [:in_consume, Thread.current, Thread.list];STDOUT.flush}
          params = get_params(workitem) 
          task_id,action,workflow,task,top_task = %w{task_id action workflow task top_task}.map{|k|params[k]}
          task.update_input_attributes!()
          task.add_internal_guards!(workflow.guards[:internal])
          event = task.add_event(:start,top_task)
          pp ["executing #{action.class.to_s}", task_id,event] if event
          workitem.fields["guard_id"] = task_id # ${guard_id} is referenced if guard for execution of this
          execution_context(task) do
            if action.long_running?
              callbacks = {
                :on_msg_received => proc do |msg|
                  result = msg[:body].merge("task_id" => task_id)
                  #result[:statuscode] is for transport errors and data is for errors for agent
                  succeeded = (result[:statuscode] == 0 and [:succeeded,:ok].include?((result[:data]||{})[:status]))
                  if succeeded
                    event = top_task.add_event(:complete_succeeded,task,result)
                    pp ["task_complete_succeeded #{action.class.to_s}", task_id,event] if event
                    set_result_succeeded(workitem,result,task,action) 
                    action.get_and_propagate_dynamic_attributes(result)
                  else
                    event,errors = task.add_event_and_errors(:complete_failed,top_task,result)
                    pp ["task_complete_failed #{action.class.to_s}", task_id,event,{:errors => errors}] if event
                    set_result_failed(workitem,result,task,action)
                  end
                  self.reply_to_engine(workitem)
                end,
                :on_timeout => proc do 
                  result = {
                    "status" => "timeout" 
                  }
                  event,errors = task.add_event_and_errors(:complete_timeout,top_task,result)
                  pp ["task_complete_timeout #{action.class.to_s}", task_id,event,{:errors => errors}] if event
                  set_result_timeout(workitem,result,task)
                  self.reply_to_engine(workitem)
                end
              }<commit_msg>working on task error processing<commit_after>module XYZ 
  module WorkflowAdapter
    module RuoteParticipant
      class Top
        include ::Ruote::LocalParticipant
        def initialize(opts=nil)
          @opts = opts
        end

        def get_params(workitem)
          task_info = get_and_delete_task_info(workitem)
          params = {"task_id" => workitem.params["task_id"]}
          workflow = task_info["workflow"]
          params.merge!("workflow" => workflow)
          params.merge!("task" => task_info["task"])
          params.merge!("action" => task_info["action"])
          params.merge!("top_task_idh" => task_info["top_task_idh"])
          params.merge!("top_task" => workflow.top_task)
          params
        end

        def set_result_succeeded(workitem,new_result,task,action)
          update_hash = {
            :status => "succeeded",
            :result => TaskAction::Result::Succeeded.new()
          }             
          task.update(update_hash)
          action.update_state_change_status(task.model_handle,:completed)  #this updates pending state
          set_result_succeeded__stack(workitem,new_result,task,action)
        end

        def set_result_failed(workitem,new_result,task,action)
          error = 
            if not new_result[:statuscode] == 0
              CommandAndControl::Error::Communication.new
            else
              data = new_result[:data]
              if data and data[:status] == :failed and (data[:error]||{})[:formatted_exception]
                CommandAndControl::Error::FailedResponse.new(data[:error][:formatted_exception])
              else
                CommandAndControl::Error.new 
              end
          end
          update_hash = {
            :status => "failed",
            :result => TaskAction::Result::Failed.new(error)
          }             
          task.update(update_hash)
        end

        def set_result_timeout(workitem,new_result,task)
          update_hash = {
            :status => "failed",
            :result => TaskAction::Result::Failed.new(CommandAndControl::Error::Timeout.new)
          }             
          task.update(update_hash)
        end

         private
        #if use must coordinate with concurrence merge type
        def set_result_succeeded__stack(workitem,new_result,task,action)
          workitem.fields["result"] = {:action_completed => action.type}
        end
        def get_and_delete_task_info(workitem)
          params = workitem.params
          Ruote::TaskInfo.get_and_delete(params["task_id"],params["task_type"])
        end
      end

      class DetectCreatedNodeIsReady < Top
        def consume(workitem)
          params = get_params(workitem) 
          task_id,action,workflow,task,task_end,top_task = %w{task_id action workflow task task_end top_task}.map{|k|params[k]}
          callbacks = {
            :on_msg_received => proc do |msg|
              result = {:type => :completed_create_node, :task_id => task_id} 
              event = task.add_event(:complete_succeeded, top_task,result)
              pp [:found,msg[:senderid]]
              task[:executable_action][:node].update_operational_status!(:running)
              set_result_succeeded(workitem,result,task,action)
              action.get_and_propagate_dynamic_attributes(result)
              self.reply_to_engine(workitem)
            end,
            :on_timeout => proc do 
              self.reply_to_engine(workitem)
            end
          }
          num_poll_cycles = 10
          poll_cycle = 6
          context = {:callbacks => callbacks, :expected_count => 1,:count => num_poll_cycles,:poll_cycle => poll_cycle} 
          workflow.poll_to_detect_node_ready(action[:node],context)
        end
      end

      #TODO: currently; just using above; either use this or get rid of it
      class DetectIfNodeIsResponding < Top
        def consume(workitem)
          params = get_params(workitem) 
          action,task,workflow = %w{action task workflow}.map{|k|params[k]}
          callbacks = {
            :on_msg_received => proc do |msg|
              pp [:found,msg[:senderid]]
              task[:executable_action][:node].update_operational_status!(:running)
              self.reply_to_engine(workitem)
            end,
            :on_timeout => proc do 
              self.reply_to_engine(workitem)
            end
          }
          poll_cycle = 2
          context = {:callbacks => callbacks, :expected_count => 1,:count => 1,:poll_cycle => poll_cycle} 
          workflow.poll_to_detect_node_ready(action[:node],context)
        end
      end
      class ExecuteOnNode < Top
        #LockforDebug = Mutex.new
        def consume(workitem)
          #LockforDebug.synchronize{pp [:in_consume, Thread.current, Thread.list];STDOUT.flush}
          params = get_params(workitem) 
          task_id,action,workflow,task,top_task = %w{task_id action workflow task top_task}.map{|k|params[k]}
          task.update_input_attributes!()
          task.add_internal_guards!(workflow.guards[:internal])
          event = task.add_event(:start,top_task)
          pp ["executing #{action.class.to_s}", task_id,event] if event
          workitem.fields["guard_id"] = task_id # ${guard_id} is referenced if guard for execution of this
          execution_context(task) do
            if action.long_running?
              callbacks = {
                :on_msg_received => proc do |msg|
                  result = msg[:body].merge("task_id" => task_id)
                  #result[:statuscode] is for transport errors and data is for errors for agent
                  succeeded = (result[:statuscode] == 0 and [:succeeded,:ok].include?((result[:data]||{})[:status]))
                  if succeeded
                    event = task.add_event(:complete_succeeded,top_task,result)
                    pp ["task_complete_succeeded #{action.class.to_s}", task_id,event] if event
                    set_result_succeeded(workitem,result,task,action) 
                    action.get_and_propagate_dynamic_attributes(result)
                  else
                    event,errors = task.add_event_and_errors(:complete_failed,top_task,result)
                    pp ["task_complete_failed #{action.class.to_s}", task_id,event,{:errors => errors}] if event
                    set_result_failed(workitem,result,task,action)
                  end
                  self.reply_to_engine(workitem)
                end,
                :on_timeout => proc do 
                  result = {
                    "status" => "timeout" 
                  }
                  event,errors = task.add_event_and_errors(:complete_timeout,top_task,result)
                  pp ["task_complete_timeout #{action.class.to_s}", task_id,event,{:errors => errors}] if event
                  set_result_timeout(workitem,result,task)
                  self.reply_to_engine(workitem)
                end
              }
{'fraction_non_alphanumeric': 0.10101592333741624, 'fraction_numerical': 0.001296923409467541, 'mean_word_length': 2.0933808780922667, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 69, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2647847', 'n_tokens_mistral': 3987, 'n_tokens_neox': 3771, 'n_words': 662}