From 366fea8b33cb55a371abb55dc776ff0c6ad2f37b Mon Sep 17 00:00:00 2001 From: Jahvon Dockery Date: Wed, 26 Aug 2026 23:27:31 -0400 Subject: [PATCH] feat(exec): support interpreter on serial and parallel steps MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Inline `cmd` steps in serial/parallel executables now take their own `interpreter`, so one workflow can mix shell and Python without splitting into separate executables. A step that omits it runs under the shell as before, and a `ref` step ignores it — the referenced executable brings its own. Both step configs $ref the same ExecInterpreter definition the exec type uses, so there is one enum rather than three parallel ones. This also lifts the restriction added with the --interpreter flag: the flag now applies to every --cmd in an invocation, in serial and parallel mode alike. ExecutableForCmd takes the interpreter as a new parameter; its unused int parameter is left alone to keep this diff to the one concern. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01R328pa3FUUfga4gYah1iQi --- cmd/internal/exec.go | 19 ++++------- cmd/internal/flags/types.go | 4 +-- docs/cli/flow_exec.md | 2 +- docs/guides/executables.md | 24 +++++++++++++- docs/public/schemas/flowfile_schema.json | 8 +++++ docs/types/flowfile.md | 2 ++ internal/runner/parallel/parallel.go | 2 +- internal/runner/serial/serial.go | 2 +- internal/templates/templates.go | 3 +- internal/utils/executables/executables.go | 10 ++++-- internal/validation/flowfile_schema.json | 8 +++++ tests/python_exec_e2e_test.go | 40 +++++++++++++++++++---- tests/utils/sub_execs.go | 4 +-- types/executable/executable.gen.go | 12 +++++++ types/executable/executable_schema.yaml | 10 ++++++ 15 files changed, 119 insertions(+), 31 deletions(-) diff --git a/cmd/internal/exec.go b/cmd/internal/exec.go index 06bd1c0a..1d240e6b 100644 --- a/cmd/internal/exec.go +++ b/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 { diff --git a/cmd/internal/flags/types.go b/cmd/internal/flags/types.go index 48fd2412..bdbdf350 100644 --- a/cmd/internal/flags/types.go +++ b/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, } diff --git a/docs/cli/flow_exec.md b/docs/cli/flow_exec.md index 833ffe91..0b7a7812 100644 --- a/docs/cli/flow_exec.md +++ b/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") diff --git a/docs/guides/executables.md b/docs/guides/executables.md index 37a6d97f..f8e91602 100644 --- a/docs/guides/executables.md +++ b/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: diff --git a/docs/public/schemas/flowfile_schema.json b/docs/public/schemas/flowfile_schema.json index c06720e1..c5a2e505 100644 --- a/docs/public/schemas/flowfile_schema.json +++ b/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", diff --git a/docs/types/flowfile.md b/docs/types/flowfile.md index c78d0c55..7d282faa 100644 --- a/docs/types/flowfile.md +++ b/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 | | diff --git a/internal/runner/parallel/parallel.go b/internal/runner/parallel/parallel.go index a1c0d87a..4c619e0c 100644 --- a/internal/runner/parallel/parallel.go +++ b/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") } diff --git a/internal/runner/serial/serial.go b/internal/runner/serial/serial.go index 06c01f19..ab1f24f6 100644 --- a/internal/runner/serial/serial.go +++ b/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") } diff --git a/internal/templates/templates.go b/internal/templates/templates.go index f427f3bd..1846cbaa 100644 --- a/internal/templates/templates.go +++ b/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") } diff --git a/internal/utils/executables/executables.go b/internal/utils/executables/executables.go index 6d06384f..f9e94d72 100644 --- a/internal/utils/executables/executables.go +++ b/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()} diff --git a/internal/validation/flowfile_schema.json b/internal/validation/flowfile_schema.json index c06720e1..c5a2e505 100644 --- a/internal/validation/flowfile_schema.json +++ b/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", diff --git a/tests/python_exec_e2e_test.go b/tests/python_exec_e2e_test.go index 80df21ee..e34f257d 100644 --- a/tests/python_exec_e2e_test.go +++ b/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")) }) }) diff --git a/tests/utils/sub_execs.go b/tests/utils/sub_execs.go index 81545b89..89649f0d 100644 --- a/tests/utils/sub_execs.go +++ b/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 { diff --git a/types/executable/executable.gen.go b/types/executable/executable.gen.go index a879684a..6e546b7a 100644 --- a/types/executable/executable.gen.go +++ b/types/executable/executable.gen.go @@ -327,6 +327,12 @@ type ParallelRefConfig struct { // If string `json:"if,omitempty" yaml:"if,omitempty" mapstructure:"if,omitempty"` + // 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. + // + Interpreter *ExecInterpreter `json:"interpreter,omitempty" yaml:"interpreter,omitempty" mapstructure:"interpreter,omitempty"` + // A human-readable label for this step, used for display purposes. Name string `json:"name,omitempty" yaml:"name,omitempty" mapstructure:"name,omitempty"` @@ -541,6 +547,12 @@ type SerialRefConfig struct { // If string `json:"if,omitempty" yaml:"if,omitempty" mapstructure:"if,omitempty"` + // 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. + // + Interpreter *ExecInterpreter `json:"interpreter,omitempty" yaml:"interpreter,omitempty" mapstructure:"interpreter,omitempty"` + // A human-readable label for this step, used for display purposes. Name string `json:"name,omitempty" yaml:"name,omitempty" mapstructure:"name,omitempty"` diff --git a/types/executable/executable_schema.yaml b/types/executable/executable_schema.yaml index 0f0d5f08..468b9111 100644 --- a/types/executable/executable_schema.yaml +++ b/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: |