code: def task_pre_step(
      self,
      step_name,
      task_datastore,
      metadata,
      run_id,
      task_id,
      flow,
      graph,
      retry_count,
      max_user_code_retries,
      ubf_context,
      inputs,
  ):
      self.task_id = task_id
      self.run_id = run_id
  
      triggers = []
      # Expose event triggering metadata through current singleton
      for key, payload in os.environ.items():
          if key.startswith("METAFLOW_ARGO_EVENT_PAYLOAD_"):
              if payload != "null":  # Argo-Workflow's None
                  try:
                      payload = json.loads(payload)
                  except (TypeError, ValueError) as e:
                      # There could be arbitrary events that Metaflow doesn't know of
                      payload = {}
                  triggers.append(
                      {
                          "timestamp": payload.get("timestamp"),
                          "id": payload.get("id"),
                          "name": payload.get(
                              "name"
                          ),  # will exist since filter
                          "type": key[
                              len("METAFLOW_ARGO_EVENT_PAYLOAD_") :
                          ].split("_", 1)[
                              0
                          ]  # infer type from env var key
                          # Add more event metadata here in the future
                      }
                  )
  
      meta = {}
      if triggers:
          # Enable current.trigger
          current._update_env({"trigger": Trigger(triggers)})
          # Luckily there aren't many events for us to be concerned about the
          # size of the metadata field yet! However we don't really need this
          # metadata outside of the start step so we can save a few bytes in the
          # db.
          if step_name == "start":
              meta["execution-triggers"] = json.dumps(triggers)
  
      meta["argo-workflow-template"] = os.environ[
          "ARGO_WORKFLOW_TEMPLATE"
      ]
      meta["argo-workflow-name"] = os.environ["ARGO_WORKFLOW_NAME"]
      meta["argo-workflow-namespace"] = os.environ[
          "ARGO_WORKFLOW_NAMESPACE"
      ]
      meta["auto-emit-argo-events"] = self.attributes[
          "auto-emit-argo-events"
      ]
      entries = [
          MetaDatum(
              field=k,
              value=v,
              type=k,
              tags=["attempt_id:{0}".format(retry_count)],
          )
          for k, v in meta.items()
      ]
      # Register book-keeping metadata for debugging.
      metadata.register_metadata(run_id, step_name, task_id, entries)
  
file_path: metaflow/metaflow/plugins/argo/argo_workflows_decorator.py
imports: import json
  import os
  import time
  from metaflow import current
  from metaflow.decorators import StepDecorator
  from metaflow.events import Trigger
  from metaflow.metadata import MetaDatum
  from metaflow.metaflow_config import ARGO_EVENTS_WEBHOOK_URL
  from .argo_events import ArgoEvent
label_desc: Use image node_id 1 for calling the ArgoWorkflowsInternalDecorator obj's underlying member method code with example usage: obj.task_pre_step(step_name, task_datastore, metadata, run_id, task_id, flow, graph, retry_count, max_user_code_retries, ubf_context, inputs) without return types