upstream-corteza/server/automation/service/workflow_converter.go

package service

import (
	"context"
	"fmt"
	"strings"

	"github.com/cortezaproject/corteza/server/automation/types"
	"github.com/cortezaproject/corteza/server/pkg/auth"
	"github.com/cortezaproject/corteza/server/pkg/errors"
	"github.com/cortezaproject/corteza/server/pkg/expr"
	"github.com/cortezaproject/corteza/server/pkg/wfexec"
	"github.com/cortezaproject/corteza/server/system/automation"
	"go.uber.org/zap"
)

type (
	wfExecutor interface {
		handleToID(string) uint64
		Exec(ctx context.Context, workflowID uint64, p types.WorkflowExecParams) (*expr.Vars, uint64, types.Stacktrace, error)
	}

	workflowConverter struct {
		// workflow function registry
		reg    *registry
		parser expr.Parsable
		log    *zap.Logger

		graphs wfExecutor
	}
)

func Convert(wfService *workflow, wf *types.Workflow) (*wfexec.Graph, types.WorkflowIssueSet) {
	conv := &workflowConverter{
		reg:    wfService.reg,
		parser: wfService.parser,
		log:    wfService.log,
		graphs: wfService,
	}

	return conv.makeGraph(wf)
}

// Converts workflow definition to wf execution graph
func (svc workflowConverter) makeGraph(def *types.Workflow) (*wfexec.Graph, types.WorkflowIssueSet) {
	var (
		g    = wfexec.NewGraph()
		wfii = types.WorkflowIssueSet{}
		IDs  = make(map[uint64]int)
	)

	// Basic step verification
	for i, s := range def.Steps {
		if _, has := IDs[s.ID]; has {
			wfii = wfii.Append(
				fmt.Errorf("step ID not unique"),
				map[string]int{"step": i, "duplicate": IDs[s.ID]},
			)
		} else {
			IDs[s.ID] = i
		}
	}

	// if we have one or more duplicated-id error we need to abort right away
	// because all further step positions in issue culprit will be invalid
	if len(wfii) > 0 {
		return nil, wfii
	}
	// remove all steps that should not go through
	// resolving procedure and should not be added to the graph
	ss, _ := def.Steps.Filter(func(step *types.WorkflowStep) (bool, error) {
		switch {
		case step.Kind == types.WorkflowStepKindVisual:
			return false, nil
		}

		return true, nil
	})

	for g.Len() < len(ss) {
		progress := false

		for _, step := range ss {
			if g.StepByID(step.ID) != nil {
				// resolved
				continue
			}

			// Collect all incoming and outgoing paths
			inPaths := make([]*types.WorkflowPath, 0, 8)
			outPaths := make([]*types.WorkflowPath, 0, 8)
			for _, path := range def.Paths {
				if path.ChildID == step.ID {
					inPaths = append(inPaths, path)
				} else if path.ParentID == step.ID {
					outPaths = append(outPaths, path)
				}
			}

			stepIssues := verifyStep(step, inPaths, outPaths)

			if resolved, err := svc.workflowStepDefConv(g, def, step, inPaths, outPaths); err != nil {
				switch aux := err.(type) {
				case types.WorkflowIssueSet:
					stepIssues = append(stepIssues, aux...)
					continue
				case error:
					stepIssues = stepIssues.Append(err, nil)
				}
			} else if resolved {
				progress = true
			}

			wfii = append(wfii, stepIssues.SetCulprit("step", IDs[step.ID])...)
		}

		if !progress {
			// nothing resolved for 1 cycle;
			// report every step that is still waiting on a neighbour
			//
			// steps that are unresolved for their own reasons (unknown
			// function, missing argument, ...) already have their issues
			// collected above
			for _, step := range ss {
				if g.StepByID(step.ID) != nil {
					continue
				}

				var (
					children []uint64
					parents  []uint64
				)

				for _, path := range def.Paths {
					if path.ParentID == step.ID && g.StepByID(path.ChildID) == nil {
						children = append(children, path.ChildID)
					}

					if path.ChildID == step.ID && g.StepByID(path.ParentID) == nil {
						parents = append(parents, path.ParentID)
					}
				}

				if len(children) == 0 && len(parents) == 0 {
					continue
				}

				wfii = wfii.Append(fmt.Errorf(
					"failed to resolve dependencies for %s step %d, waiting for unresolved parent steps %v and child steps %v",
					step.Kind,
					step.ID,
					parents,
					children,
				), map[string]int{"step": IDs[step.ID]})
			}

			if len(wfii) == 0 {
				// resolving stalled but nothing explains why
				wfii = wfii.Append(fmt.Errorf("failed to resolve workflow step dependencies"), map[string]int{})
			}

			break
		}
	}

	// add parents that were still unresolved when a join gateway was created
	for _, step := range ss {
		if step.Kind != types.WorkflowStepKindGateway || step.Ref != "join" {
			continue
		}

		resolved, is := g.StepByID(step.ID).(interface{ AddPath(wfexec.Step) })
		if !is {
			continue
		}

		for _, path := range def.Paths {
			if path.ChildID != step.ID {
				continue
			}

			if parent := g.StepByID(path.ParentID); parent != nil {
				resolved.AddPath(parent)
			}
		}
	}

	for pos, path := range def.Paths {
		if g.StepByID(path.ChildID) == nil {
			wfii = wfii.Append(fmt.Errorf("failed to resolve step with ID %d", path.ChildID), map[string]int{"path": pos})
			continue
		}

		if g.StepByID(path.ParentID) == nil {
			wfii = wfii.Append(fmt.Errorf("failed to resolve step with ID %d", path.ParentID), map[string]int{"path": pos})
			continue
		}

		if len(wfii) > 0 {
			// pointless to fill the if there are errors
			continue
		}

		g.AddParent(
			g.StepByID(path.ChildID),
			g.StepByID(path.ParentID),
		)
	}

	if len(wfii) > 0 {
		return nil, wfii.Distinct()
	}

	return g, nil
}

// converts all step definitions into workflow.Step instances
//
// if this func returns nil for step and error, assume unresolved dependencies
func (svc workflowConverter) workflowStepDefConv(g *wfexec.Graph, def *types.Workflow, s *types.WorkflowStep, in, out []*types.WorkflowPath) (bool, error) {
	if err := svc.parseExpressions(s.Arguments...); err != nil {
		return false, errors.Internal("failed to parse step arguments expressions for %s: %s", s.Kind, err).Wrap(err)
	}

	if err := svc.parseExpressions(s.Results...); err != nil {
		return false, errors.Internal("failed to parse step results expressions for %s: %s", s.Kind, err).Wrap(err)
	}

	conv, err := func() (wfexec.Step, error) {
		switch s.Kind {
		case types.WorkflowStepKindVisual:
			return nil, nil

		case types.WorkflowStepKindDebug:
			return svc.convDebugStep(s)

		case types.WorkflowStepKindExpressions:
			return svc.convExpressionStep(s)

		case types.WorkflowStepKindGateway:
			return svc.convGateway(g, s, in, out)

		case types.WorkflowStepKindFunction, types.WorkflowStepKindIterator:
			return svc.convFunctionStep(g, s, out)

		case types.WorkflowStepKindError:
			return svc.convErrorStep(s)

		case types.WorkflowStepKindTermination:
			return svc.convTerminationStep()

		case types.WorkflowStepKindPrompt:
			return svc.convPromptStep(s)

		case types.WorkflowStepKindDelay:
			return svc.convDelayStep(s)

		case types.WorkflowStepKindErrHandler:
			return svc.convErrorHandlerStep(g, s, out)

		case types.WorkflowStepKindBreak:
			return svc.convBreakStep()

		case types.WorkflowStepKindContinue:
			return svc.convContinueStep()

		case types.WorkflowStepKindExecWorkflow:
			return svc.convExecWorkflowStep(def, s)

		default:
			return nil, errors.Internal("unsupported step kind %q", s.Kind)
		}
	}()

	if err != nil {
		return false, err
	} else if conv != nil {
		conv.SetID(s.ID)
		g.AddStep(conv)
		return true, err
	} else {
		// signal caller that we were unable to
		// resolve definition at the moment
		return false, nil
	}
}

func (svc workflowConverter) convGateway(g *wfexec.Graph, s *types.WorkflowStep, in, out []*types.WorkflowPath) (wfexec.Step, error) {
	switch s.Ref {
	case "fork":
		return wfexec.ForkGateway(), nil

	case "join":
		var (
			ss []wfexec.Step
		)
		// Collect parents that are already in the graph; the rest are added
		// once the whole graph is resolved.
		//
		// Waiting for all of them here deadlocks on an iterator with its exit
		// path leading to this gateway: the iterator can not be resolved
		// before the gateway is in the graph.
		for _, p := range in {
			if parent := g.StepByID(p.ParentID); parent != nil {
				ss = append(ss, parent)
			}
		}

		return wfexec.JoinGateway(ss...), nil

	case "incl", "excl":
		var (
			pp []*wfexec.GatewayPath
		)

		for _, c := range out {
			child := g.StepByID(c.ChildID)
			if child == nil {
				return nil, nil
			}

			if len(c.Expr) > 0 {
				if err := svc.parser.ParseEvaluators(c); err != nil {
					return nil, err
				}
			}

			// wrapping with fn to make sure that we're dealing with the right wf path inside gw-path tester fn
			err := func(c types.WorkflowPath) error {
				p, err := wfexec.NewGatewayPath(child, func(ctx context.Context, scope *expr.Vars) (bool, error) {
					if len(c.Expr) == 0 {
						return true, nil
					}

					return c.Test(ctx, scope)
				})

				if err != nil {
					return err
				}

				pp = append(pp, p)
				return nil
			}(*c)

			if err != nil {
				return nil, err
			}
		}

		if s.Ref == "excl" {
			return wfexec.ExclGateway(pp...)
		} else {
			return wfexec.InclGateway(pp...)
		}
	}

	return nil, errors.Internal("unexpected workflow configuration")
}

func (svc workflowConverter) convErrorHandlerStep(g *wfexec.Graph, s *types.WorkflowStep, out []*types.WorkflowPath) (wfexec.Step, error) {
	var (
		results = &expr.Vars{}
	)

	// evaluate results and set target path for each expression
	for _, r := range s.Results {
		err := results.Set(r.Expr, r.Target)
		if err != nil {
			return nil, err
		}
	}

	switch len(out) {
	case 0:
		return nil, fmt.Errorf("expecting at least one path out of error handling step")
	case 1:
		// remove error handler
		return types.ErrorHandlerStep(nil, results), nil
	case 2:
		errorHandler := g.StepByID(out[1].ChildID)
		if errorHandler == nil {
			// wait for it to be resolved
			return nil, nil
		}

		return types.ErrorHandlerStep(errorHandler, results), nil

	default:
		// this might be extended in the future to allow different paths using expression
		// but then again, this can be solved by gateway path following the error handling step
		return nil, fmt.Errorf("max 2 paths out of error handling step")
	}
}

func (svc workflowConverter) convExpressionStep(s *types.WorkflowStep) (wfexec.Step, error) {
	return types.ExpressionsStep(s.Arguments...), nil
}

// internal debug step that can log entire
func (svc workflowConverter) convDebugStep(_ *types.WorkflowStep) (wfexec.Step, error) {
	return types.DebugStep(svc.log), nil
}

func (svc workflowConverter) convFunctionStep(g *wfexec.Graph, s *types.WorkflowStep, out []*types.WorkflowPath) (wfexec.Step, error) {
	reg := Registry()

	if def := reg.Function(s.Ref); def == nil {
		return nil, errors.Internal("unknown function %q", s.Ref)
	} else {
		if def.Kind != string(s.Kind) {
			return nil, fmt.Errorf("unexpected %s on %s step", def.Kind, s.Kind)
		}

		var (
			err        error
			isIterator = def.Kind == types.FunctionKindIterator
		)

		if isIterator {
			if def.Iterator == nil {
				return nil, errors.Internal("iterator handler for %q not set", s.Ref)
			}
		} else {
			if def.Handler == nil {
				return nil, errors.Internal("function handler for %q not set", s.Ref)
			}
		}

		if err = def.Parameters.VerifyArguments(s.Arguments); err != nil {
			return nil, errors.Internal("failed to verify argument expressions for %s %s: %s", s.Kind, s.Ref, err).Wrap(err)
		}

		if err = def.Results.VerifyResults(s.Results); err != nil {
			return nil, errors.Internal("failed to verify result expressions for %s %s: %s", s.Kind, s.Ref, err).Wrap(err)
		}

		if isIterator {
			if len(out) != 2 {
				return nil, fmt.Errorf("expecting exactly 2 outbound paths for iterator")
			}

			var (
				next = g.StepByID(out[0].ChildID)
				exit = g.StepByID(out[1].ChildID)
			)

			if next == nil || exit == nil {
				// wait for steps to be resolved
				return nil, nil
			}

			return types.IteratorStep(def, s.Arguments, s.Results, next, exit)

		} else {
			return types.FunctionStep(def, s.Arguments, s.Results)
		}
	}
}

// creates error step
//
// Expects ZERO outgoing paths and
func (svc workflowConverter) convErrorStep(s *types.WorkflowStep) (wfexec.Step, error) {
	const (
		argName = "message"
	)

	var (
		args = types.ExprSet(s.Arguments)
	)

	return wfexec.NewGenericStep(func(ctx context.Context, r *wfexec.ExecRequest) (wfexec.ExecResponse, error) {
		var (
			msg         string
			result, err = args.Eval(ctx, r.Scope)
		)
		if err != nil {
			return nil, err
		}

		if result.Has(argName) {
			str, _ := expr.NewString(expr.Must(result.Select(argName)))
			msg = str.GetValue()
		} else {
			if aux, is := args.GetByTarget(argName).Value.(string); is {
				msg = aux
			} else {
				msg = "ERROR"
			}
		}

		return nil, errors.Automation("%s", msg)
	}), nil
}

// converts termination definition to wfexec.Step
func (svc workflowConverter) convTerminationStep() (wfexec.Step, error) {
	return wfexec.NewGenericStep(func(ctx context.Context, r *wfexec.ExecRequest) (wfexec.ExecResponse, error) {
		return wfexec.Termination(), nil
	}), nil
}

// converts prompt definition to wfexec.Step
//
// If input is not passed (session was not resumed from suspension after prompt), we send wait-for-input
// result that signals session it needs to suspend and wait for the input.
// At this point, we take the current scope run it through step arguments and store the output with the suspended state
//
// After session is resumed, input should be set (not nil) and evaluated through results.
func (svc workflowConverter) convPromptStep(s *types.WorkflowStep) (wfexec.Step, error) {
	// Use expression step as base for prompt step
	var (
		args = types.ExprSet(s.Arguments)
		res  = types.ExprSet(s.Results)
	)
	const (
		ownerArgName = "owner"
	)

	return wfexec.NewGenericStep(func(ctx context.Context, r *wfexec.ExecRequest) (wfexec.ExecResponse, error) {
		if r.Input == nil {
			// input is only set (not nil) when session is resumed on prompt step
			// suspend the session and wait for input

			// @todo take scope, eval with arguments and add it to suspended state
			payload, err := args.Eval(ctx, r.Scope)
			if err != nil {
				return nil, err
			}

			var ownerId uint64 = 0
			if payload.Has(ownerArgName) {
				// @note for now only types.User and uint64 are supported.
				// At some point we should extend this to handles, emails, usernames, ...
				// but for now this should be fine

				// @todo should we validate existence here?
				// I'm not sure how to bring in a thing to validate it so I omitted it for now.
				param := expr.Must(payload.Select(ownerArgName))
				switch casted := param.(type) {
				case *automation.User:
					ownerId = casted.GetValue().ID
				case *expr.ID:
					ownerId = casted.GetValue()
				default:
					return nil, fmt.Errorf("cannot select owner: unsupported type provided %T", casted)
				}
			} else {
				if i := auth.GetIdentityFromContextWithKey(ctx, workflowInvokerCtxKey{}); i != nil {
					ownerId = i.Identity()
				}
			}

			return wfexec.Prompt(ownerId, s.Ref, payload), nil
		}

		results, err := res.Eval(ctx, r.Scope.MustMerge(r.Input))
		if err != nil {
			return nil, err
		}

		return results, nil
	}), nil
}

// converts delay definition to wfexec.Step
func (svc workflowConverter) convDelayStep(s *types.WorkflowStep) (wfexec.Step, error) {
	return types.DelayStep(s.Arguments), nil
}

func (svc workflowConverter) convBreakStep() (wfexec.Step, error) {
	return wfexec.NewGenericStep(func(ctx context.Context, r *wfexec.ExecRequest) (wfexec.ExecResponse, error) {
		return wfexec.LoopBreak(), nil
	}), nil

}

func (svc workflowConverter) convContinueStep() (wfexec.Step, error) {
	return wfexec.NewGenericStep(func(ctx context.Context, r *wfexec.ExecRequest) (wfexec.ExecResponse, error) {
		return wfexec.LoopContinue(), nil
	}), nil

}

// creates workflow-executor step
//
// Expects max ONE outgoing paths and ONE incoming path
func (svc workflowConverter) convExecWorkflowStep(wf *types.Workflow, s *types.WorkflowStep) (_ wfexec.Step, err error) {
	const (
		// sub-workflow's scope
		// expecting expr.Vars
		argNameScope = "scope"

		// workflow id or handle
		argNameWorkflow = "workflow"
	)

	return wfexec.NewGenericStep(func(ctx context.Context, r *wfexec.ExecRequest) (resp wfexec.ExecResponse, err error) {
		var (
			p = types.WorkflowExecParams{
				// sub-workflows should not be executed asynchronously
				// might lead to uncontrolled resource consumption
				Async: false,

				// @todo figure out how check if trace is enabled
				Trace: false,

				// always wait for the sub-workflow even when deferred
				Wait: true,

				Input: expr.EmptyVars(),

				// who's calling sub-workflow?
				CallerSessionID:  r.SessionID,
				CallerStepID:     s.ID,
				CallerWorkflowID: wf.ID,
			}

			result *expr.Vars

			workflowID     uint64
			workflowHandle string
		)

		if ap, err := processArguments(ctx, s.Arguments, r.Scope); err != nil {
			return nil, err
		} else {
			if ap.string(argNameWorkflow, &workflowHandle) {
				if workflowID = svc.graphs.handleToID(workflowHandle); workflowID == 0 {
					return nil, errors.NotFound("workflow with handle %s not found", workflowHandle)
				}
			} else if !ap.uint64(argNameWorkflow, &workflowID) {
				return nil, errors.InvalidData("workflow ID must be provided")
			}

			// let's make sure we're not calling ourselves
			if p.CallerWorkflowID == workflowID {
				return nil, errors.InvalidData("recursive workflow call is not allowed")
			}

			// scope for the sub-workflow
			ap.vars(argNameScope, p.Input)
		}

		result, _, _, err = svc.graphs.Exec(ctx, workflowID, p)
		if err != nil {
			return
		}

		result, err = types.ExprSet(s.Results).Eval(ctx, result)
		if err != nil {
			return
		}

		return result, nil
	}), nil
}

func (svc workflowConverter) parseExpressions(ee ...*types.Expr) (err error) {
	for _, e := range ee {

		if len(strings.TrimSpace(e.Expr)) > 0 {
			if err = svc.parser.ParseEvaluators(e); err != nil {
				return
			}
		}

		if err = e.SetType(exprTypeSetter(svc.reg, e)); err != nil {
			return err
		}

		for _, t := range e.Tests {
			if err = svc.parser.ParseEvaluators(t); err != nil {
				return
			}
		}
	}

	return nil
}

func verifyStep(s *types.WorkflowStep, in, out types.WorkflowPathSet) types.WorkflowIssueSet {
	const (
		arguments = "argument"
		results   = "result"
		outbound  = "outbound path"
		inbound   = "inbound path"
	)

	var (
		ii = types.WorkflowIssueSet{}

		// checks if count of arguments, results, outbound or inbound paths is between (incl) min and max
		count = func(min, max int, typ string) func() error {
			return func() error {
				var (
					l int
				)

				switch typ {
				case arguments:
					l = len(s.Arguments)
				case results:
					l = len(s.Results)
				case outbound:
					l = len(out)
				case inbound:
					l = len(in)
				}

				switch {
				case max == 0 && min == max && l != min:
					return errors.Internal("%s step does not expect any %ss", s.Kind, typ)

				case max > 0 && min == max && l != min:
					return errors.Internal("%s step expects exactly %d %s(s)", s.Kind, min, typ)

				case l < min:
					return errors.Internal("%s step expects at least %d %s(s)", s.Kind, min, typ)

				case max > 0 && l > max:
					return errors.Internal("%s step expects no more than %d %s(s)", s.Kind, max, typ)

				}

				return nil
			}
		}

		// check if reference is set on the step
		requiredRef = func() error {
			if s.Ref == "" {
				return errors.Internal("%s step expects reference", s.Kind)
			}

			return nil
		}

		// check if reference is set on the step
		numericRef = func() error {
			// @todo SUBWF enable this back
			// if 0 == cast.ToUint64(s.Ref) {
			//	return errors.Internal("%s step expects workflow reference ID", s.Kind)
			// }

			return nil
		}

		// reference should not be set on the step
		noRef = func() error {
			if s.Ref != "" {
				return errors.Internal("%s step does not expects reference", s.Kind)
			}

			return nil
		}

		// check for corredor function step(s) are allowed or not
		checkDisabledFunc = func() error {
			if !DefaultWorkflow.corredorOpt.Enabled && s.Ref == "corredorExec" {
				return WorkflowErrNotAllowedToExecuteCorredorStep()
			}
			return nil
		}

		// checks if argument is present
		checkArg = func(argName string, tt ...expr.Type) func() error {
			return func() error {
				msgArg := types.ExprSet(s.Arguments).GetByTarget(argName)
				if msgArg == nil {
					return nil
				}

				for _, typ := range tt {
					if msgArg.Type == typ.Type() {
						return nil
					}
				}

				return errors.Internal("unexpected type %q for argument %q on step type %q", msgArg.Type, argName, s.Kind)
			}
		}

		// checks if argument is present
		requiredArg = func(argName string, tt ...expr.Type) func() error {
			return func() error {
				if msgArg := types.ExprSet(s.Arguments).GetByTarget(argName); msgArg == nil {
					return errors.Internal("%s step expects to have '%s' argument", s.Kind, argName)
				}

				return checkArg(argName, tt...)()
			}
		}

		// should not have any arguments, results, outbound or inbound paths
		zero = func(typ string) func() error { return count(0, 0, typ) }

		// step should not have any outbound paths
		last = func() error { return count(0, 0, outbound)() }

		// gw sanity check
		gatewayCheck = func(checks ...func() error) []func() error {
			switch s.Ref {
			case "join":
				return append(checks, count(1, -1, inbound))

			case "fork", "incl", "excl":
				return append(checks, count(1, -1, outbound))
			}

			return append(checks, func() error { return fmt.Errorf("unknown gateway type") })
		}

		checks = make([]func() error, 0)
	)

	switch s.Kind {
	case types.WorkflowStepKindErrHandler:
		checks = append(checks,
			noRef,
			zero(arguments),
			count(1, 2, outbound),
		)

	case types.WorkflowStepKindDebug:
		checks = append(checks,
			noRef,
			zero(results),
			count(0, 1, outbound),
		)

	case types.WorkflowStepKindVisual:
		checks = append(checks,
			zero(results),
			zero(arguments),
		)

	case types.WorkflowStepKindExpressions:
		checks = append(checks,
			noRef,
			zero(results),
			count(1, -1, arguments),
			count(0, 1, outbound),
		)

	case types.WorkflowStepKindGateway:
		checks = append(checks, gatewayCheck(zero(arguments), zero(results))...)

	case types.WorkflowStepKindError:
		checks = append(checks,
			requiredArg("message", expr.String{}),
			count(0, 1, arguments),
			zero(results),
			last,
		)

	case types.WorkflowStepKindTermination:
		checks = append(checks,
			noRef,
			zero(arguments),
			zero(results),
			last,
		)

	case types.WorkflowStepKindFunction:
		checks = append(checks,
			requiredRef,
			numericRef,
			checkDisabledFunc,
			count(0, 1, outbound),
		)

	case types.WorkflowStepKindIterator:
		checks = append(checks,
			requiredRef,
			count(2, 2, outbound),
		)

	case types.WorkflowStepKindPrompt:
		checks = append(checks,
			requiredRef,
			count(0, 1, outbound),
		)

	case types.WorkflowStepKindDelay:
		checks = append(checks,
			checkArg("timestamp", expr.DateTime{}),
			checkArg("offset", expr.Duration{}),
			count(1, 1, arguments),
			zero(results),
			count(0, 1, outbound),
		)

	case types.WorkflowStepKindBreak:
		checks = append(checks,
			noRef,
			zero(arguments),
			zero(results),
			count(0, 1, outbound),
			last,
		)

	case types.WorkflowStepKindContinue:
		checks = append(checks,
			noRef,
			zero(arguments),
			zero(results),
			count(0, 1, outbound),
			last,
		)

	case types.WorkflowStepKindExecWorkflow:
		checks = append(checks,
			noRef,
			// required "workflow" with handle or ID
			requiredArg("workflow", expr.Handle{}, expr.ID{}),

			// optional "scope" as expr.Vars
			checkArg("scope", expr.Vars{}),

			// expecting at least workflow + optional scope, nothing more
			count(1, 2, arguments),

			count(0, 1, outbound),
		)

	case "":
		return ii.Append(fmt.Errorf("missing step kind"), nil)

	default:
		return ii.Append(fmt.Errorf("unknown step kind '%s'", s.Kind), nil)

	}

	for _, check := range checks {
		if err := check(); err != nil {
			ii = ii.Append(err, nil)
		}
	}

	return ii
}