Is there a way to get the original task id from the join step inputs? especially if an exception occurred in that task

I think using

current

you can fetch it. You can set it inside the task and then use it in join steps. https://docs.metaflow.org/api/current

Does that work in the join step on the inputs?

something like this?

def called_by_foreach(self):
  try:
    # You can define some dataclass also for task_status
    self.task_status = dict(task_id=current.task_id, status="SUCCESS", err=None)
  except Exception as e:
    # maybe add traceback or error?
    self.task_status = dict(task_id=current.task_id, status="FAILED", err=str(e))
   
   self.next(self.join)

def join(self, inputs):
     tasks = [input.task_status for input in inputs]

     # report failed task?
     for task_status in tasks:
         # report failing stuff?

That would definitely work thank you. I guess I'm just surprised that info isn't automatically available already.

@creamy-stone-99746 yes, such a functionality is on our roadmap!