i’ve a flow that does this:

flow.terminal_state_handler = cycle_state_handler(...)

where

cycle_state_handler

looks like this:

def cycle_state_handler(project, **kwargs):
   """Re-runs `flow` once it reaches a finished state.

   Must be used as a `terminal_state_handler`."""
   def _handler(flow, final_state, _reference_task_states):
       if not isinstance(final_state, state.Cancelled):
           # set this to None to force the backend to make a new name for us
           prefect.context["flow_run_name"] = None

           create_flow_run.run(
               project_name=project,
               flow_name=flow.name,
               parameters=prefect.context.parameters,
               **kwargs
           )
       return final_state
   return _handler

occasionally this handler spawns 2 flows instead of one. i track this by setting a parameter on the new flow to be the current flow run’s name and log said name when the child flow starts

there’s only one of those

🤔

i can be more explicit and put one of my own at the start of the handler just in case, but i suspect it’ll just be verification of what that’s showing