docs: def add_forward_workflow(self, dag, sections, satisfies=None):
          '''Add a forward-workflow, return number of nodes added
          '''
          dag.new_forward_workflow()
  
          if 'DAG' in env.config['SOS_DEBUG'] or 'ALL' in env.config['SOS_DEBUG']:
              env.log_to_file(
                  'DAG', f'Adding mini-workflow with {len(sections)} sections')
          default_input: sos_targets = sos_targets([])
          for idx, section in enumerate(sections):
              #
              res = analyze_section(section, default_input=default_input)
  
              environ_vars = res['environ_vars']
              signature_vars = res['signature_vars']
              changed_vars = res['changed_vars']
              # parameters, if used in the step, should be considered environmental
              environ_vars |= env.parameter_vars & signature_vars
  
              # add shared to targets
              if res['changed_vars']:
                  if 'provides' in section.options:
                      if isinstance(section.options['provides'], str):
                          section.options.set('provides',
                                              [section.options['provides']])
                  else:
                      section.options.set('provides', [])
                  #
                  section.options.set(
                      'provides', section.options['provides'] +
                      [sos_variable(var) for var in changed_vars])
  
              context = {
                  '__signature_vars__': signature_vars,
                  '__environ_vars__': environ_vars,
                  '__changed_vars__': changed_vars,
                  '__dynamic_depends__': res['dynamic_depends'],
                  '__dynamic_input__': res['dynamic_input']
              }
  
              # for nested workflow, the input is specified by sos_run, not None.
              if idx == 0:
                  context['__step_output__'] = env.sos_dict['__step_output__']
              # can be the only step
              if idx == len(sections) - 1 and satisfies is not None:
                  res['step_output'].extend(satisfies)
  
              dag.add_step(
                  section.uuid,
                  section.step_name(),
                  idx,
                  res['step_input'],
                  res['step_depends'],
                  res['step_output'],
                  context=context)
              default_input = res['step_output']
          return len(sections)