fix: assume the flowfile context when expanding refs for serial/parallel (#322)

Currently the current context is being used when running a serial or
parallel executable with incomplete references. This causes issues when
trying to run an executable from another workspace. This change makes it
so that the parent executable's workspace and namespace is used for the
ref instead.

Author: jahvon

internal/context/context.go

@@ -214,6 +214,18 @@ func ExpandRef(ctx *Context, ref executable.Ref) executable.Ref {
 	return executable.NewRef(executable.NewExecutableID(ws, ns, name), ref.Verb())
 }
 
+func ExpandRefFromParent(parent *executable.Executable, ref executable.Ref) executable.Ref {
+	id := ref.ID()
+	ws, ns, name := executable.MustParseExecutableID(id)
+	if ws == "" || ws == executable.WildcardWorkspace {
+		ws = parent.Workspace()
+	}
+	if ns == "" {
+		ns = parent.Namespace()
+	}
+	return executable.NewRef(executable.NewExecutableID(ws, ns, name), ref.Verb())
+}
+
 func currentWorkspace(cfg *config.Config) (*workspace.Workspace, error) {
 	ws, err := cfg.CurrentWorkspaceName()
 	if err != nil {

internal/runner/parallel/parallel.go

@@ -118,7 +118,7 @@ func handleExec(
 		switch {
 		case len(refConfig.Ref) > 0:
 			var err error
-			exec, err = execUtils.ExecutableForRef(ctx, refConfig.Ref)
+			exec, err = execUtils.ExecutableForRef(ctx, parent, refConfig.Ref)
 			if err != nil {
 				return err
 			}

internal/runner/serial/serial.go

@@ -107,7 +107,7 @@ func handleExec(
 		switch {
 		case refConfig.Ref != "":
 			var err error
-			exec, err = execUtils.ExecutableForRef(ctx, refConfig.Ref)
+			exec, err = execUtils.ExecutableForRef(ctx, parent, refConfig.Ref)
 			if err != nil {
 				return err
 			}

internal/templates/templates.go

@@ -131,7 +131,7 @@ func runExecutables(
 			if err != nil {
 				return errors.Wrap(err, fmt.Sprintf("unable to process %s executable %d", stage, i))
 			}
-			exec, err = execUtils.ExecutableForRef(ctx, executable.Ref(ref.String()))
+			exec, err = execUtils.ExecutableForRef(ctx, nil, executable.Ref(ref.String()))
 			if err != nil {
 				return err
 			}

internal/utils/executables/executables.go

@@ -8,8 +8,17 @@ import (
 	"github.com/flowexec/flow/types/executable"
 )
 
-func ExecutableForRef(ctx *context.Context, ref executable.Ref) (*executable.Executable, error) {
-	executableRef := context.ExpandRef(ctx, ref)
+func ExecutableForRef(
+	ctx *context.Context,
+	parent *executable.Executable,
+	ref executable.Ref,
+) (*executable.Executable, error) {
+	var executableRef executable.Ref
+	if parent != nil {
+		executableRef = context.ExpandRefFromParent(parent, ref)
+	} else {
+		executableRef = context.ExpandRef(ctx, ref)
+	}
 	exec, err := ctx.ExecutableCache.GetExecutableByRef(executableRef)
 	if err != nil {
 		return nil, err