fix: child execs should inherit parent argument values (#284)

Author: jahvon

.golangci.yaml

@@ -95,6 +95,14 @@ linters:
         text: (SA5011|SA5001)
       - path: (.+)\.go$
         text: declaration of "(err|ctx)" shadows declaration at
+      # The serial and parallel runners have inherently complex logic.
+      - path: internal/runner/(parallel|serial)
+        linters:
+          - cyclop
+          - nestif
+          - funlen
+          - gocognit
+          - gocyclo
     paths:
       - third_party$
       - builtin$

cmd/internal/exec.go

@@ -95,11 +95,6 @@ func execFunc(ctx *context.Context, cmd *cobra.Command, verb executable.Verb, ar
 		logger.Log().FatalErr(err)
 	}
 
-	// populate args for environment handling ignoring arg 0 (the ref)
-	if len(args) >= 2 {
-		ctx.Args = args[1:]
-	}
-
 	var ref executable.Ref
 	if len(args) == 0 {
 		ref = context.ExpandRef(ctx, executable.NewRef("", verb))
@@ -166,7 +161,13 @@ func execFunc(ctx *context.Context, cmd *cobra.Command, verb executable.Verb, ar
 	}
 	startTime := time.Now()
 	eng := engine.NewExecEngine()
-	if err := runner.Exec(ctx, e, eng, envMap); err != nil {
+
+	var execArgs []string
+	if len(args) >= 2 {
+		execArgs = args[1:]
+	}
+
+	if err := runner.Exec(ctx, e, eng, envMap, execArgs); err != nil {
 		logger.Log().FatalErr(err)
 	}
 	dur := time.Since(startTime)

internal/context/context.go

@@ -35,8 +35,9 @@ type Context struct {
 	WorkspacesCache  cache.WorkspaceCache
 	ExecutableCache  cache.ExecutableCache
 
-	// Args includes the command line arguments passed to the exec command. It is only populated when that command is used.
-	Args []string
+	// RootExecutable is the executable that is being run in the current context.
+	// This will be nil if the context is not associated with an executable run.
+	RootExecutable *executable.Executable
 
 	// ProcessTmpDir is the temporary directory for the current process. If set, it will be
 	// used to store temporary files all executable runs when the tmpDir value is specified.

internal/runner/exec/exec.go

@@ -34,10 +34,11 @@ func (r *execRunner) Exec(
 	e *executable.Executable,
 	_ engine.Engine,
 	inputEnv map[string]string,
+	inputArgs []string,
 ) error {
 	execSpec := e.Exec
 	defaultEnv := env.DefaultEnv(ctx, e)
-	envMap, err := env.BuildEnvMap(ctx.Config.CurrentVaultName(), e.Env(), ctx.Args, inputEnv, defaultEnv)
+	envMap, err := env.BuildEnvMap(ctx.Config.CurrentVaultName(), e.Env(), inputArgs, inputEnv, defaultEnv)
 	if err != nil {
 		return errors.Wrap(err, "unable to set parameters to env")
 	}
@@ -48,7 +49,7 @@ func (r *execRunner) Exec(
 		e.FlowFilePath(),
 		e.WorkspacePath(),
 		e.Env(),
-		ctx.Args,
+		inputArgs,
 		envMap,
 	); err != nil {
 		ctx.AddCallback(cb)

internal/runner/launch/launch.go

@@ -37,19 +37,20 @@ func (r *launchRunner) Exec(
 	e *executable.Executable,
 	_ engine.Engine,
 	inputEnv map[string]string,
+	inputArgs []string,
 ) error {
 	launchSpec := e.Launch
 	envMap, err := env.BuildEnvMap(
 		ctx.Config.CurrentVaultName(),
 		e.Env(),
-		ctx.Args,
+		inputArgs,
 		inputEnv,
 		env.DefaultEnv(ctx, e),
 	)
 	if err != nil {
 		return errors.Wrap(err, "unable to set parameters to env")
 	}
-	if err := env.SetEnv(ctx.Config.CurrentVaultName(), e.Env(), ctx.Args, envMap); err != nil {
+	if err := env.SetEnv(ctx.Config.CurrentVaultName(), e.Env(), inputArgs, envMap); err != nil {
 		return errors.Wrap(err, "unable to set parameters to env")
 	}
 
@@ -58,7 +59,7 @@ func (r *launchRunner) Exec(
 		e.FlowFilePath(),
 		e.WorkspacePath(),
 		e.Env(),
-		ctx.Args,
+		inputArgs,
 		envMap,
 	); err != nil {
 		ctx.AddCallback(cb)

internal/runner/mocks/mock_runner.go

@@ -4,6 +4,9 @@ import (
 	stdCtx "context"
 	"fmt"
 	"maps"
+	"os"
+	"path/filepath"
+	"strings"
 
 	"github.com/pkg/errors"
 	"golang.org/x/sync/errgroup"
@@ -41,9 +44,10 @@ func (r *parallelRunner) Exec(
 	e *executable.Executable,
 	eng engine.Engine,
 	inputEnv map[string]string,
+	inputArgs []string,
 ) error {
 	parallelSpec := e.Parallel
-	if err := envUtils.SetEnv(ctx.Config.CurrentVaultName(), e.Env(), ctx.Args, inputEnv); err != nil {
+	if err := envUtils.SetEnv(ctx.Config.CurrentVaultName(), e.Env(), inputArgs, inputEnv); err != nil {
 		return errors.Wrap(err, "unable to set parameters to env")
 	}
 
@@ -52,7 +56,7 @@ func (r *parallelRunner) Exec(
 		e.FlowFilePath(),
 		e.WorkspacePath(),
 		e.Env(),
-		ctx.Args,
+		inputArgs,
 		inputEnv,
 	); err != nil {
 		ctx.AddCallback(cb)
@@ -83,12 +87,11 @@ func (r *parallelRunner) Exec(
 	return fmt.Errorf("no parallel executables to run")
 }
 
-//nolint:funlen,gocognit
 func handleExec(
 	ctx *context.Context, parent *executable.Executable,
 	eng engine.Engine,
 	parallelSpec *executable.ParallelExecutableType,
-	promptedEnv map[string]string,
+	inputEnv map[string]string,
 	cacheData map[string]string,
 ) error {
 	groupCtx, cancel := stdCtx.WithCancel(ctx.Ctx)
@@ -100,18 +103,44 @@ func handleExec(
 	}
 	group.SetLimit(limit)
 
-	dataMap := expr.ExpressionEnv(ctx, parent, cacheData, promptedEnv)
+	// Expand the directory of the parallel execution. The root / parent's directory is used if one is not specified.
+	var root *executable.Executable
+	if ctx.RootExecutable != nil {
+		root = ctx.RootExecutable
+	} else {
+		root = parent
+	}
+	if parallelSpec.Dir == "" {
+		parallelSpec.Dir = executable.Directory(filepath.Dir(root.FlowFilePath()))
+	}
+	targetDir, isTmp, err := parallelSpec.Dir.ExpandDirectory(
+		root.WorkspacePath(),
+		root.FlowFilePath(),
+		ctx.ProcessTmpDir,
+		inputEnv,
+	)
+	if err != nil {
+		return errors.Wrap(err, "unable to expand directory")
+	} else if isTmp && ctx.ProcessTmpDir == "" {
+		ctx.ProcessTmpDir = targetDir
+	}
 
+	// Build the list of steps to execute
 	var execs []engine.Exec
+	conditionalData := expr.ExpressionEnv(ctx, parent, cacheData, inputEnv)
+
 	for i, refConfig := range parallelSpec.Execs {
+		// Skip over steps that do not match the condition
 		if refConfig.If != "" {
-			if truthy, err := expr.IsTruthy(refConfig.If, &dataMap); err != nil {
+			if truthy, err := expr.IsTruthy(refConfig.If, &conditionalData); err != nil {
 				return err
 			} else if !truthy {
 				logger.Log().Debugf("skipping execution %d/%d", i+1, len(parallelSpec.Execs))
 				continue
 			}
 		}
+
+		// Get the executable for the step
 		var exec *executable.Executable
 		switch {
 		case len(refConfig.Ref) > 0:
@@ -126,8 +155,10 @@ func handleExec(
 			return errors.New("parallel executable must have a ref or cmd")
 		}
 
-		execPromptedEnv := make(map[string]string)
-		maps.Copy(promptedEnv, execPromptedEnv)
+		// Prepare the environment and arguments for the child executable
+		childEnv := make(map[string]string)
+		childArgs := make([]string, 0)
+		maps.Copy(childEnv, inputEnv)
 		if len(refConfig.Args) > 0 {
 			execEnv := exec.Env()
 			if execEnv == nil || execEnv.Args == nil {
@@ -136,14 +167,31 @@ func handleExec(
 					exec.Ref().String(),
 				)
 			} else {
-				a, err := envUtils.BuildArgsEnvMap(execEnv.Args, refConfig.Args, execPromptedEnv)
+				for _, arg := range os.Environ() {
+					kv := strings.SplitN(arg, "=", 2)
+					if len(kv) == 2 {
+						childEnv[kv[0]] = kv[1]
+					}
+				}
+
+				if parallelSpec.Args == nil {
+					childArgs = refConfig.Args
+				} else {
+					childArgs = envUtils.BuildArgsFromEnv(execEnv.Args, childEnv)
+					if len(childArgs) == 0 {
+						childArgs = refConfig.Args // If no resolved args, fallback to original args
+					}
+				}
+
+				a, err := envUtils.BuildArgsEnvMap(execEnv.Args, childArgs, childEnv)
 				if err != nil {
 					logger.Log().Error(err, "unable to process arguments")
 				}
-				maps.Copy(execPromptedEnv, a)
+				maps.Copy(childEnv, a)
 			}
 		}
 
+		// Set log fields and directory for the executable
 		switch {
 		case exec.Exec != nil:
 			fields := map[string]interface{}{"step": exec.Ref().String()}
@@ -159,10 +207,18 @@ func handleExec(
 			if parallelSpec.Dir != "" && exec.Serial.Dir == "" {
 				exec.Serial.Dir = parallelSpec.Dir
 			}
+		case exec.Request != nil:
+			if exec.Request.ResponseFile != nil && parallelSpec.Dir != "" && exec.Request.ResponseFile.Dir == "" {
+				exec.Request.ResponseFile.Dir = parallelSpec.Dir
+			}
+		case exec.Render != nil:
+			if parallelSpec.Dir != "" && exec.Render.Dir == "" {
+				exec.Render.Dir = parallelSpec.Dir
+			}
 		}
 
 		runExec := func() error {
-			err := runner.Exec(ctx, exec, eng, execPromptedEnv)
+			err := runner.Exec(ctx, exec, eng, childEnv, childArgs)
 			if err != nil {
 				return err
 			}
@@ -171,6 +227,7 @@ func handleExec(
 
 		execs = append(execs, engine.Exec{ID: exec.Ref().String(), Function: runExec, MaxRetries: refConfig.Retries})
 	}
+
 	results := eng.Execute(
 		ctx.Ctx, execs,
 		engine.WithMode(engine.Parallel),

internal/runner/parallel/parallel.go

@@ -9,6 +9,7 @@ import (
 	. "github.com/onsi/gomega"
 	"go.uber.org/mock/gomock"
 
+	"github.com/flowexec/flow/internal/context"
 	"github.com/flowexec/flow/internal/runner"
 	"github.com/flowexec/flow/internal/runner/engine"
 	"github.com/flowexec/flow/internal/runner/engine/mocks"
@@ -108,7 +109,7 @@ var _ = Describe("ParallelRunner", func() {
 			mockEngine.EXPECT().
 				Execute(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
 				Return(results).Times(1)
-			Expect(parallelRnr.Exec(ctx.Ctx, rootExec, mockEngine, promptedEnv)).To(Succeed())
+			Expect(parallelRnr.Exec(ctx.Ctx, rootExec, mockEngine, promptedEnv, nil)).To(Succeed())
 		})
 
 		It("fail when there is an engine error", func() {
@@ -128,7 +129,7 @@ var _ = Describe("ParallelRunner", func() {
 			mockEngine.EXPECT().
 				Execute(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
 				Return(results).Times(1)
-			Expect(parallelRnr.Exec(ctx.Ctx, rootExec, mockEngine, promptedEnv)).ToNot(Succeed())
+			Expect(parallelRnr.Exec(ctx.Ctx, rootExec, mockEngine, promptedEnv, nil)).ToNot(Succeed())
 		})
 
 		It("should skip execution when condition is false", func() {
@@ -145,7 +146,58 @@ var _ = Describe("ParallelRunner", func() {
 			mockEngine.EXPECT().
 				Execute(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
 				Return(results).Times(1)
-			Expect(parallelRnr.Exec(ctx.Ctx, rootExec, mockEngine, make(map[string]string))).To(Succeed())
+			Expect(parallelRnr.Exec(ctx.Ctx, rootExec, mockEngine, make(map[string]string), nil)).To(Succeed())
+		})
+
+		It("should pass environment args from parent to child executables", func() {
+			pos1 := 1
+			parentExec := &executable.Executable{
+				Parallel: &executable.ParallelExecutableType{
+					Args:  executable.ArgumentList{{EnvKey: "TEST_VAR", Pos: &pos1}},
+					Execs: []executable.ParallelRefConfig{{Ref: "test:child", Args: []string{"var=$TEST_VAR"}}},
+				},
+			}
+			parentExec.SetContext("test", "/test", "test", "/test/parent.flow")
+
+			childExec := &executable.Executable{
+				Exec: &executable.ExecExecutableType{
+					Cmd:  "echo $TEST_VAR",
+					Args: executable.ArgumentList{{EnvKey: "TEST_VAR", Flag: "var"}},
+				},
+			}
+			childExec.SetContext("test", "/test", "test", "/test/child.flow")
+
+			mockCache := ctx.ExecutableCache
+			mockCache.EXPECT().GetExecutableByRef(gomock.Any()).Return(childExec, nil).Times(1)
+
+			ctx.RunnerMock.EXPECT().IsCompatible(gomock.Any()).Return(true).Times(1)
+			ctx.RunnerMock.EXPECT().
+				Exec(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), []string{"var=test_value"}).
+				DoAndReturn(func(
+					_ *context.Context,
+					exec *executable.Executable,
+					_ engine.Engine,
+					inputEnv map[string]string,
+					inputArgs []string,
+				) error {
+					Expect(inputEnv).To(HaveKeyWithValue("TEST_VAR", "test_value"))
+					Expect(inputArgs).To(ContainElement("var=test_value"))
+					return nil
+				}).Times(1)
+
+			results := engine.ResultSummary{Results: []engine.Result{{}}}
+			mockEngine.EXPECT().
+				Execute(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
+				DoAndReturn(func(
+					_ stdCtx.Context, execs []engine.Exec, _ ...engine.OptionFunc) engine.ResultSummary {
+					for _, exec := range execs {
+						Expect(exec.Function()).To(Succeed())
+					}
+					return results
+				})
+
+			Expect(parallelRnr.Exec(ctx.Ctx, parentExec, mockEngine, make(map[string]string), []string{"test_value"})).
+				To(Succeed())
 		})
 	})
 })