feat(exec): support interpreter on serial and parallel steps

cmd/internal/exec.go

@@ -252,18 +252,13 @@ func execAdHoc(ctx *context.Context, cmd *cobra.Command, verb executable.Verb, c
 	}
 
 	interpreter := executable.ExecInterpreter(flags.ValueFor[string](cmd, *flags.InterpreterFlag, false))
+	var interpreterPtr *executable.ExecInterpreter
 	if interpreter != "" {
-		// Inline serial/parallel steps carry no interpreter of their own, so a
-		// multi-command batch could only run the first one as requested.
-		if len(commands) > 1 {
-			errhandler.HandleUsage(ctx, cmd, "--interpreter cannot be combined with multiple --cmd values")
-			return
-		}
-		probe := &executable.ExecExecutableType{Interpreter: &interpreter}
-		if err := probe.Validate(); err != nil {
+		if err := (&executable.ExecExecutableType{Interpreter: &interpreter}).Validate(); err != nil {
 			errhandler.HandleUsage(ctx, cmd, "%v", err)
 			return
 		}
+		interpreterPtr = &interpreter
 	}
 
 	joined := strings.Join(commands, "\n")
@@ -278,19 +273,17 @@ func execAdHoc(ctx *context.Context, cmd *cobra.Command, verb executable.Verb, c
 			Dir:     executable.Directory(dir),
 			LogMode: logMode,
 		}
-		if interpreter != "" {
-			e.Exec.Interpreter = &interpreter
-		}
+		e.Exec.Interpreter = interpreterPtr
 	} else {
 		steps := make(executable.SerialRefConfigList, len(commands))
 		for i, c := range commands {
-			steps[i] = executable.SerialRefConfig{Cmd: c}
+			steps[i] = executable.SerialRefConfig{Cmd: c, Interpreter: interpreterPtr}
 		}
 		mode := flags.ValueFor[string](cmd, *flags.CmdModeFlag, false)
 		if mode == "parallel" {
 			pSteps := make(executable.ParallelRefConfigList, len(commands))
 			for i, c := range commands {
-				pSteps[i] = executable.ParallelRefConfig{Cmd: c}
+				pSteps[i] = executable.ParallelRefConfig{Cmd: c, Interpreter: interpreterPtr}
 			}
 			e.Parallel = &executable.ParallelExecutableType{Execs: pSteps}
 		} else {

cmd/internal/flags/types.go

@@ -346,8 +346,8 @@ var CmdFlag = &Metadata{
 
 var InterpreterFlag = &Metadata{
 	Name: "interpreter",
-	Usage: "The interpreter to run an ad-hoc --cmd with: 'sh' (default) or 'python'. " +
-		"Only valid with a single --cmd.",
+	Usage: "The interpreter to run ad-hoc --cmd commands with: 'sh' (default) or 'python'. " +
+		"Applies to every --cmd in the invocation.",
 	Default:  "",
 	Required: false,
 }

docs/cli/flow_exec.md

@@ -50,7 +50,7 @@ flow exec EXECUTABLE_ID [-- args...] [flags]
       --cmd flow logs        Run an ad-hoc shell command through flow instead of a named executable. The command runs with the current workspace's environment and is recorded in flow logs. Repeat --cmd to run multiple commands in one invocation (see --mode).
       --dir string           Working directory for an ad-hoc command (defaults to the current directory). Only valid with --cmd.
   -h, --help                 help for exec
-      --interpreter string   The interpreter to run an ad-hoc --cmd with: 'sh' (default) or 'python'. Only valid with a single --cmd.
+      --interpreter string   The interpreter to run ad-hoc --cmd commands with: 'sh' (default) or 'python'. Applies to every --cmd in the invocation.
       --label string         A short, human-readable label for an ad-hoc command (used in history). Only valid with --cmd.
   -m, --log-mode string      Log mode (text, logfmt, json, hidden)
       --mode string          How to run multiple --cmd commands: 'serial' (default) or 'parallel'. (default "serial")

docs/guides/executables.md

@@ -350,12 +350,34 @@ Notes and limitations:
   `user: root` to opt out.
 - `.bat`, `.cmd`, and `.ps1` files are not supported with `container`.
 - `container` applies to `exec` executables only; inline `cmd` steps inside `serial`/`parallel` do
-  not inherit it — reference a container-backed executable instead.
+  not inherit it — reference a container-backed executable instead. (`interpreter` *is* available on
+  those steps — see below.)
 - `outputFile` destinations for params/args should resolve under the workspace root so the container
   can see them (use `//`-prefixed or flow-file-relative paths).
 - On macOS, Docker Desktop does not share `/var/folders` by default, so `dir: f:tmp` may fail to
   mount; use a workspace-relative directory instead.
 
+#### Per-step interpreters
+
+Inline `cmd` steps inside `serial` and `parallel` take their own `interpreter`, so one workflow can
+mix shell and Python without splitting into separate executables:
+
+```yaml
+executables:
+  - verb: run
+    name: pipeline
+    serial:
+      execs:
+        - cmd: ./fetch-data.sh
+        - cmd: |
+            import json
+            print(json.load(open("data.json"))["total"])
+          interpreter: python
+```
+
+A step that omits `interpreter` runs under the shell as before. A step using `ref` ignores the field
+— the referenced executable brings its own.
+
 ### serial - Sequential Execution
 
 Run multiple steps in order:

docs/public/schemas/flowfile_schema.json

@@ -346,6 +346,10 @@
           "type": "string",
           "default": ""
         },
+        "interpreter": {
+          "$ref": "#/definitions/ExecutableExecInterpreter",
+          "description": "The interpreter used to run `cmd` for this step. Defaults to `sh`.\nOnly applies to `cmd`; a `ref` uses the referenced executable's own interpreter.\n"
+        },
         "name": {
           "description": "A human-readable label for this step, used for display purposes.",
           "type": "string",
@@ -595,6 +599,10 @@
           "type": "string",
           "default": ""
         },
+        "interpreter": {
+          "$ref": "#/definitions/ExecutableExecInterpreter",
+          "description": "The interpreter used to run `cmd` for this step. Defaults to `sh`.\nOnly applies to `cmd`; a `ref` uses the referenced executable's own interpreter.\n"
+        },
         "name": {
           "description": "A human-readable label for this step, used for display purposes.",
           "type": "string",

docs/types/flowfile.md

@@ -280,6 +280,7 @@ Configuration for a parallel executable.
 | `args` | Arguments to pass to the executable. | `array` (`string`) | [] |  |
 | `cmd` | The command to execute. One of `cmd` or `ref` must be set.  | `string` |  |  |
 | `if` | An expression that determines whether the executable should run, using the Expr language syntax. The expression is evaluated at runtime and must resolve to a boolean value.  The expression has access to OS/architecture information (os, arch), environment variables (env), stored data (store), and context information (ctx) like workspace and paths.  For example, `os == "darwin"` will only run on macOS, `len(store["feature"]) > 0` will run if a value exists in the store, and `env["CI"] == "true"` will run in CI environments. See the [Expr documentation](https://expr-lang.org/docs/language-definition) for more information.  | `string` |  |  |
+| `interpreter` | The interpreter used to run `cmd` for this step. Defaults to `sh`. Only applies to `cmd`; a `ref` uses the referenced executable's own interpreter.  | [ExecutableExecInterpreter](#executableexecinterpreter) |  |  |
 | `name` | A human-readable label for this step, used for display purposes. | `string` |  |  |
 | `ref` | A reference to another executable to run in serial. One of `cmd` or `ref` must be set.  | [ExecutableRef](#executableref) |  |  |
 | `retries` | The number of times to retry the executable if it fails. | `integer` | 0 |  |
@@ -431,6 +432,7 @@ Configuration for a serial executable.
 | `args` | Arguments to pass to the executable. | `array` (`string`) | [] |  |
 | `cmd` | The command to execute. One of `cmd` or `ref` must be set.  | `string` |  |  |
 | `if` | An expression that determines whether the executable should run, using the Expr language syntax. The expression is evaluated at runtime and must resolve to a boolean value.  The expression has access to OS/architecture information (os, arch), environment variables (env), stored data (store), and context information (ctx) like workspace and paths.  For example, `os == "darwin"` will only run on macOS, `len(store["feature"]) > 0` will run if a value exists in the store, and `env["CI"] == "true"` will run in CI environments. See the [Expr documentation](https://expr-lang.org/docs/language-definition) for more information.  | `string` |  |  |
+| `interpreter` | The interpreter used to run `cmd` for this step. Defaults to `sh`. Only applies to `cmd`; a `ref` uses the referenced executable's own interpreter.  | [ExecutableExecInterpreter](#executableexecinterpreter) |  |  |
 | `name` | A human-readable label for this step, used for display purposes. | `string` |  |  |
 | `ref` | A reference to another executable to run in serial. One of `cmd` or `ref` must be set.  | [ExecutableRef](#executableref) |  |  |
 | `retries` | The number of times to retry the executable if it fails. | `integer` | 0 |  |

internal/runner/parallel/parallel.go

@@ -127,7 +127,7 @@ func handleExec(
 				return err
 			}
 		case refConfig.Cmd != "":
-			exec = execUtils.ExecutableForCmd(parent, refConfig.Cmd, i)
+			exec = execUtils.ExecutableForCmd(parent, refConfig.Cmd, refConfig.Interpreter, i)
 		default:
 			return errors.New("parallel executable must have a ref or cmd")
 		}

internal/runner/serial/serial.go

@@ -117,7 +117,7 @@ func handleExec(
 				return err
 			}
 		case refConfig.Cmd != "":
-			exec = execUtils.ExecutableForCmd(parent, refConfig.Cmd, i)
+			exec = execUtils.ExecutableForCmd(parent, refConfig.Cmd, refConfig.Interpreter, i)
 		default:
 			return errors.New("serial executable must have a ref or cmd")
 		}

internal/templates/templates.go

@@ -158,7 +158,8 @@ func runExecutables(
 			if err != nil {
 				return errors.Wrap(err, fmt.Sprintf("unable to process %s executable %d", stage, i))
 			}
-			exec = execUtils.ExecutableForCmd(templateParent(ws.AssignedName(), ws.Location(), flowfileDir), cmd.String(), i)
+			exec = execUtils.ExecutableForCmd(
+				templateParent(ws.AssignedName(), ws.Location(), flowfileDir), cmd.String(), nil, i)
 		default:
 			return errors.New("post-run executable must have a ref or cmd")
 		}

internal/utils/executables/executables.go

@@ -36,14 +36,20 @@ func ExecutableForRef(
 	return exec, nil
 }
 
-func ExecutableForCmd(parent *executable.Executable, cmd string, _ int) *executable.Executable {
+// ExecutableForCmd wraps an inline `cmd` step from a serial/parallel executable
+// in a transient executable. interpreter may be empty, in which case the step
+// runs under flow's built-in shell as it always has.
+func ExecutableForCmd(
+	parent *executable.Executable, cmd string, interpreter *executable.ExecInterpreter, _ int,
+) *executable.Executable {
 	vis := executable.ExecutableVisibility(common.VisibilityInternal)
 	exec := &executable.Executable{
 		Verb:       parent.Verb,
 		Name:       parent.Name,
 		Visibility: &vis,
 		Exec: &executable.ExecExecutableType{
-			Cmd: cmd,
+			Cmd:         cmd,
+			Interpreter: interpreter,
 		},
 	}
 	fields := map[string]interface{}{"executable": exec.Ref().String()}

internal/validation/flowfile_schema.json

@@ -346,6 +346,10 @@
           "type": "string",
           "default": ""
         },
+        "interpreter": {
+          "$ref": "#/definitions/ExecutableExecInterpreter",
+          "description": "The interpreter used to run `cmd` for this step. Defaults to `sh`.\nOnly applies to `cmd`; a `ref` uses the referenced executable's own interpreter.\n"
+        },
         "name": {
           "description": "A human-readable label for this step, used for display purposes.",
           "type": "string",
@@ -595,6 +599,10 @@
           "type": "string",
           "default": ""
         },
+        "interpreter": {
+          "$ref": "#/definitions/ExecutableExecInterpreter",
+          "description": "The interpreter used to run `cmd` for this step. Defaults to `sh`.\nOnly applies to `cmd`; a `ref` uses the referenced executable's own interpreter.\n"
+        },
         "name": {
           "description": "A human-readable label for this step, used for display purposes.",
           "type": "string",

tests/python_exec_e2e_test.go

@@ -117,14 +117,40 @@ var _ = Describe("python exec e2e", func() {
 			Expect(err).To(HaveOccurred())
 		})
 
-		It("rejects --interpreter with multiple commands", func() {
-			// Serial/parallel steps carry no interpreter, so only the single-command
-			// form can honour the flag.
+		It("applies the interpreter to every command in a batch", func() {
 			runner := utils.NewE2ECommandRunner()
-			ctx.ExpectFailure()
-			err := runner.Run(ctx.Context, "exec", "--interpreter", "python",
-				"--cmd", "print(1)", "--cmd", "print(2)")
-			Expect(err).To(HaveOccurred())
+			stdOut := ctx.StdOut()
+			Expect(runner.Run(ctx.Context, "exec", "--interpreter", "python",
+				"--cmd", "print('batch one')", "--cmd", "print('batch two')")).To(Succeed())
+			out, _ := readFileContent(stdOut)
+			Expect(out).To(ContainSubstring("batch one"))
+			Expect(out).To(ContainSubstring("batch two"))
+		})
+	})
+
+	When("a serial executable mixes interpreters across steps", func() {
+		It("runs each step under its own interpreter", func() {
+			runner := utils.NewE2ECommandRunner()
+			stdOut := ctx.StdOut()
+			spec := `{"verb":"run","name":"mixed-steps","serial":{"execs":[` +
+				`{"cmd":"echo from-shell"},` +
+				`{"cmd":"import sys; print('from-python', sys.version_info[0])","interpreter":"python"}` +
+				`]}}`
+			Expect(runner.Run(ctx.Context, "exec", "--spec", spec)).To(Succeed())
+			out, _ := readFileContent(stdOut)
+			Expect(out).To(ContainSubstring("from-shell"))
+			Expect(out).To(ContainSubstring("from-python 3"))
+		})
+
+		It("leaves steps without an interpreter on the shell", func() {
+			runner := utils.NewE2ECommandRunner()
+			stdOut := ctx.StdOut()
+			// `echo` is a shell builtin, so this only succeeds if the step really
+			// stayed on flow's POSIX interpreter.
+			spec := `{"verb":"run","name":"default-steps","serial":{"execs":[{"cmd":"echo still-shell"}]}}`
+			Expect(runner.Run(ctx.Context, "exec", "--spec", spec)).To(Succeed())
+			out, _ := readFileContent(stdOut)
+			Expect(out).To(ContainSubstring("still-shell"))
 		})
 	})
 

tests/utils/sub_execs.go

@@ -22,7 +22,7 @@ func findSerialSubExecs(root *executable.Executable, flowFiles executable.FlowFi
 	var subExecs []*executable.Executable
 	for i, refCfg := range serial.Execs {
 		if refCfg.Cmd != "" {
-			subExecs = append(subExecs, execUtils.ExecutableForCmd(root, refCfg.Cmd, i))
+			subExecs = append(subExecs, execUtils.ExecutableForCmd(root, refCfg.Cmd, refCfg.Interpreter, i))
 		}
 
 		for _, flowFile := range flowFiles {
@@ -40,7 +40,7 @@ func findParallelSubExecs(root *executable.Executable, flowFiles executable.Flow
 	var subExecs []*executable.Executable
 	for i, refCfg := range parallel.Execs {
 		if refCfg.Cmd != "" {
-			subExecs = append(subExecs, execUtils.ExecutableForCmd(root, refCfg.Cmd, i))
+			subExecs = append(subExecs, execUtils.ExecutableForCmd(root, refCfg.Cmd, refCfg.Interpreter, i))
 		}
 
 		for _, flowFile := range flowFiles {

types/executable/executable_schema.yaml

@@ -454,6 +454,11 @@ definitions:
           The command to execute.
           One of `cmd` or `ref` must be set.
         default: ""
+      interpreter:
+        $ref: '#/definitions/ExecInterpreter'
+        description: |
+          The interpreter used to run `cmd` for this step. Defaults to `sh`.
+          Only applies to `cmd`; a `ref` uses the referenced executable's own interpreter.
       ref:
         $ref: '#/definitions/Ref'
         description: |
@@ -636,6 +641,11 @@ definitions:
           The command to execute.
           One of `cmd` or `ref` must be set.
         default: ""
+      interpreter:
+        $ref: '#/definitions/ExecInterpreter'
+        description: |
+          The interpreter used to run `cmd` for this step. Defaults to `sh`.
+          Only applies to `cmd`; a `ref` uses the referenced executable's own interpreter.
       ref:
         $ref: '#/definitions/Ref'
         description: |