docs: def sos_run(workflow=None,
              targets=None,
              shared=None,
              args=None,
              source=None,
              **kwargs):
      '''Execute a workflow from the current SoS script or a specified source
      (in .sos or .ipynb format), with _input as the initial input of workflow.'''
      if '__std_out__' in env.sos_dict and '__std_err__' in env.sos_dict:
          raise RuntimeError(
              'Executing nested workflow (action sos_run) in tasks is not supported.'
          )
  
      if isinstance(workflow, str):
          workflows = [workflow]
      elif isinstance(workflow, Sequence):
          workflows = list(workflow)
      elif workflow is not None:
          raise ValueError(
              'workflow has to be None, a workflow name, or a list of workflow names'
          )
  
      if source is None:
          script = SoS_Script(env.sos_dict['__step_context__'].content,
                              env.sos_dict['__step_context__'].filename)
          wfs = [script.workflow(wf, use_default=not targets) for wf in workflows]
      else:
          # reading workflow from another file
          script = SoS_Script(filename=source)
          wfs = [script.workflow(wf, use_default=not targets) for wf in workflows]
      # if wf contains the current step or one of the previous one, this constitute
      # recusive nested workflow and should not be allowed
      all_parameters = set()
      for wf in wfs:
          all_parameters |= set(wf.parameters())
          if env.sos_dict['step_name'] in [
                  f'{x.name}_{x.index}' for x in wf.sections
          ]:
              raise RuntimeError(
                  f'Nested workflow {workflow} contains the current step {env.sos_dict["step_name"]}'
              )
  
      # args can be specified both as a dictionary or keyword arguments
      if args is None:
          args = kwargs
      else:
          args.update(kwargs)
  
      for key in args.keys():
          if key not in all_parameters and key not in SOS_ACTION_OPTIONS:
              raise ValueError(
                  f'No parameter {key} is defined for workflow {workflow}')
  
      if shared is None:
          shared = []
      elif isinstance(shared, str):
          shared = [shared]
  
      # for nested workflow, _input would becomes the input of workflow.
      env.sos_dict.set('__step_output__',
                       copy.deepcopy(env.sos_dict.get('_input', None)))
      shared.append('__step_output__')
      try:
          my_name = env.sos_dict['step_name']
          args_output = ', '.join(f'{x}={short_repr(y)}' for x, y in args.items()
                                  if not x.startswith('__'))
          if 'ACTION' in env.config['SOS_DEBUG'] or 'ALL' in env.config['SOS_DEBUG']:
              env.log_to_file(
                  'ACTION',
                  'Executing workflow ``{}`` with input ``{}`` and {}'.format(
                      workflow,
                      short_repr(env.sos_dict.get('_input', None), True),
                      'no args' if not args_output else args_output))
  
          if not hasattr(env, '__socket__'):
              raise RuntimeError(
                  'sos_run function cannot be executed in scratch cell.')
          # tell the master process to receive a workflow
          # really send the workflow
          shared = {
              x: (env.sos_dict[x] if x in env.sos_dict else None) for x in shared
          }
  
          wf_ids = [str(uuid.uuid4()) for wf in wfs]
  
          blocking = not env.sos_dict.get('__concurrent_subworkflow__', False)
          env.__socket__.send_pyobj([
              'workflow', wf_ids, wfs, targets, args, shared, env.config, blocking
          ])
  
          if not blocking:
              return {'pending_workflows': wf_ids}
          res = {}
          for wf in wfs:
              wf_res = env.__socket__.recv_pyobj()
              res.update(wf_res)
              if wf_res is None:
                  sys.exit(0)
              elif isinstance(wf_res, Exception):
                  raise wf_res
              else:
                  env.sos_dict.quick_update(wf_res['shared'])
          return res
      finally:
          # restore step_name in case the subworkflow re-defines it
          env.sos_dict.set('step_name', my_name)