// SPDX-License-Identifier: EUPL-1.2 package agentic import ( "io" "os" core "dappco.re/go/core" ) type FlowRunStepOutput struct { Name string `json:"name,omitempty"` Command string `json:"command,omitempty"` Args []string `json:"args,omitempty"` Success bool `json:"success"` ContinueOnError bool `json:"continue_on_error,omitempty"` Stdout string `json:"stdout,omitempty"` Stderr string `json:"stderr,omitempty"` Error string `json:"error,omitempty"` } type flowExecutionSummary struct { Success bool Executed int Passed int Failed int StepResults []FlowRunStepOutput } func (s *PrepSubsystem) runFlowExecutionCommand(options core.Options, commandLabel string) core.Result { if optionBoolValue(options, "dry_run", "dry-run") { return s.runFlowCommand(options, commandLabel) } flowPath := optionStringValue(options, "_arg", "path", "slug") if flowPath == "" { core.Print(nil, "usage: core-agent %s [--dry-run] [--var=key=value] [--vars='{\"key\":\"value\"}'] [--variables='{\"key\":\"value\"}']", commandLabel) return core.Result{Value: core.E("agentic.cmdRunFlow", "flow path or slug is required", nil), OK: false} } variables := optionStringMapValue(options, "var", "vars", "variables") flowResult := readFlowDocument(flowPath, variables) if !flowResult.OK { core.Print(nil, "error: %v", flowResult.Value) return core.Result{Value: flowResult.Value, OK: false} } document, ok := flowResult.Value.(flowRunDocument) if !ok || !document.Parsed { err := core.E("agentic.cmdRunFlow", "invalid flow definition", nil) core.Print(nil, "error: %v", err) return core.Result{Value: err, OK: false} } validation := s.validateExecutableFlowDefinition(document) if !validation.OK { err, ok := validation.Value.(error) if !ok { err = core.E("agentic.cmdRunFlow", "invalid flow definition", nil) } core.Print(nil, "error: %v", err) return core.Result{Value: err, OK: false} } output := FlowRunOutput{ Success: true, Source: document.Source, Name: document.Definition.Name, Description: document.Definition.Description, Steps: len(document.Definition.Steps), Parsed: document.Parsed, } core.Print(nil, "flow: %s", document.Source) if len(variables) > 0 { core.Print(nil, "vars: %d", len(variables)) } if output.Name != "" { core.Print(nil, "name: %s", output.Name) } if output.Description != "" { core.Print(nil, "desc: %s", output.Description) } if len(document.Definition.Steps) == 0 { core.Print(nil, "steps: 0") printFlowExecutionSummary(output) return core.Result{Value: output, OK: true} } core.Print(nil, "steps: %d", len(document.Definition.Steps)) execution := s.executeFlowDefinition(document) output.Success = execution.Success output.Executed = execution.Executed output.Passed = execution.Passed output.Failed = execution.Failed output.StepResults = execution.StepResults printFlowExecutionSummary(output) return core.Result{Value: output, OK: output.Success} } func (s *PrepSubsystem) validateExecutableFlowDefinition(document flowRunDocument) core.Result { for index, step := range document.Definition.Steps { if err := s.validateExecutableFlowStep(index+1, step); err != nil { return core.Result{Value: err, OK: false} } } return core.Result{OK: true} } func (s *PrepSubsystem) validateExecutableFlowStep(index int, step flowDefinitionStep) error { stepName := flowStepDisplayName(index, step) if core.Trim(step.Cmd) == "" { switch { case core.Trim(step.Flow) != "": return core.E("agentic.validateExecutableFlowStep", core.Concat("step \"", stepName, "\" cannot execute nested flow references; use flow/preview or convert to cmd"), nil) case core.Trim(step.Run) != "": return core.E("agentic.validateExecutableFlowStep", core.Concat("step \"", stepName, "\" uses legacy run syntax; use cmd and args"), nil) default: return core.E("agentic.validateExecutableFlowStep", core.Concat("step \"", stepName, "\" must define cmd"), nil) } } commandResult := s.Core().Command(step.Cmd) if !commandResult.OK { return core.E("agentic.validateExecutableFlowStep", core.Concat("step \"", stepName, "\" references unknown command: ", step.Cmd), nil) } command, ok := commandResult.Value.(*core.Command) if !ok || command == nil || command.Action == nil { return core.E("agentic.validateExecutableFlowStep", core.Concat("step \"", stepName, "\" references a non-executable command: ", step.Cmd), nil) } return nil } func (s *PrepSubsystem) executeFlowDefinition(document flowRunDocument) flowExecutionSummary { summary := flowExecutionSummary{Success: true} for index, step := range document.Definition.Steps { stepOutput := s.executeFlowStep(index+1, step) summary.Executed++ summary.StepResults = append(summary.StepResults, stepOutput) if stepOutput.Success { summary.Passed++ continue } summary.Failed++ if stepOutput.ContinueOnError { continue } summary.Success = false break } return summary } func (s *PrepSubsystem) executeFlowStep(index int, step flowDefinitionStep) FlowRunStepOutput { stepOutput := FlowRunStepOutput{ Name: flowStepDisplayName(index, step), Command: step.Cmd, Args: append([]string(nil), step.Args...), ContinueOnError: step.ContinueOnError, } core.Print(nil, "%d. %s", index, flowStepSummary(step)) command := s.Core().Command(step.Cmd).Value.(*core.Command) result, stdout, stderr, err := captureFlowStepOutput(func() core.Result { return command.Run(flowStepOptions(step.Args)) }) if stdout != "" { stepOutput.Stdout = stdout printFlowStepStream("stdout", stdout) } if stderr != "" { stepOutput.Stderr = stderr printFlowStepStream("stderr", stderr) } if err != nil { stepOutput.Error = err.Error() core.Print(nil, " status: failed") core.Print(nil, " error: %s", stepOutput.Error) return stepOutput } stepOutput.Success = result.OK if result.OK { core.Print(nil, " status: passed") return stepOutput } stepOutput.Error = commandResultError("agentic.cmdRunFlow", result).Error() if stepOutput.ContinueOnError { core.Print(nil, " status: failed (continued)") } else { core.Print(nil, " status: failed") } core.Print(nil, " error: %s", stepOutput.Error) return stepOutput } func flowStepDisplayName(index int, step flowDefinitionStep) string { if name := core.Trim(step.Name); name != "" { return name } if name := core.Trim(step.Cmd); name != "" { return name } if name := core.Trim(step.Flow); name != "" { return name } if name := core.Trim(step.Run); name != "" { return name } return core.Concat("step-", core.Itoa(index)) } func flowStepCommandLine(step flowDefinitionStep) string { command := core.Trim(step.Cmd) if len(step.Args) == 0 { return command } return core.Concat(command, " ", core.Join(" ", step.Args...)) } func flowStepOptions(args []string) core.Options { options := core.NewOptions() for _, arg := range args { key, value, ok := core.ParseFlag(arg) if ok { if core.Contains(arg, "=") { options.Set(key, value) } else { options.Set(key, true) } continue } if !core.IsFlag(arg) { options.Set("_arg", arg) } } return options } func captureFlowStepOutput(run func() core.Result) (core.Result, string, string, error) { stdoutReader, stdoutWriter, err := os.Pipe() if err != nil { return core.Result{}, "", "", core.E("agentic.captureFlowStepOutput", "create stdout pipe", err) } stderrReader, stderrWriter, err := os.Pipe() if err != nil { stdoutReader.Close() stdoutWriter.Close() return core.Result{}, "", "", core.E("agentic.captureFlowStepOutput", "create stderr pipe", err) } oldStdout := os.Stdout oldStderr := os.Stderr os.Stdout = stdoutWriter os.Stderr = stderrWriter result := run() os.Stdout = oldStdout os.Stderr = oldStderr if err := stdoutWriter.Close(); err != nil { stdoutReader.Close() stderrReader.Close() stderrWriter.Close() return result, "", "", core.E("agentic.captureFlowStepOutput", "close stdout pipe", err) } if err := stderrWriter.Close(); err != nil { stdoutReader.Close() stderrReader.Close() return result, "", "", core.E("agentic.captureFlowStepOutput", "close stderr pipe", err) } stdoutData, err := io.ReadAll(stdoutReader) if err != nil { stdoutReader.Close() stderrReader.Close() return result, "", "", core.E("agentic.captureFlowStepOutput", "read stdout pipe", err) } stderrData, err := io.ReadAll(stderrReader) if err != nil { stdoutReader.Close() stderrReader.Close() return result, "", "", core.E("agentic.captureFlowStepOutput", "read stderr pipe", err) } stdoutReader.Close() stderrReader.Close() return result, string(stdoutData), string(stderrData), nil } func printFlowStepStream(label, stream string) { trimmed := core.TrimSuffix(core.Replace(stream, "\r\n", "\n"), "\n") if trimmed == "" { return } core.Print(nil, " %s:", label) for _, line := range core.Split(trimmed, "\n") { core.Print(nil, " %s", line) } } func printFlowExecutionSummary(output FlowRunOutput) { core.Print(nil, "") core.Print(nil, "summary:") for _, step := range output.StepResults { status := "passed" if !step.Success { if step.ContinueOnError { status = "failed (continued)" } else { status = "failed" } } core.Print(nil, " %s: %s", step.Name, status) } core.Print(nil, "totals: ran=%d passed=%d failed=%d", output.Executed, output.Passed, output.Failed) }