From 5a492a17c47096696d47cf504132f2a6d89c9f84 Mon Sep 17 00:00:00 2001 From: Michael McNees Date: Fri, 27 Mar 2026 00:41:46 -0400 Subject: [PATCH 01/11] feat(db): add migration and config for workflow composition (#54) Add parent_execution_id, parent_step_name, and depth columns to workflow_executions for workflow composition support. Add MaxWorkflowDepth config with env var binding (default: 10). Co-Authored-By: Claude Opus 4.6 (1M context) --- packages/engine/internal/config/config.go | 3 +++ .../db/migrations/016_workflow_composition.sql | 11 +++++++++++ 2 files changed, 14 insertions(+) create mode 100644 packages/engine/internal/db/migrations/016_workflow_composition.sql diff --git a/packages/engine/internal/config/config.go b/packages/engine/internal/config/config.go index 3252fe4..7766dbb 100644 --- a/packages/engine/internal/config/config.go +++ b/packages/engine/internal/config/config.go @@ -131,6 +131,7 @@ type EngineConfig struct { AllowedBaseURLs []string `mapstructure:"allowed_base_urls"` AllowedModels []string `mapstructure:"allowed_models"` // empty = all allowed MaxToolRoundsLimit int `mapstructure:"max_tool_rounds_limit"` // 0 = no limit + MaxWorkflowDepth int `mapstructure:"max_workflow_depth"` Budget BudgetConfig `mapstructure:"budget"` } @@ -171,6 +172,7 @@ func Load(cmd *cobra.Command) (*Config, error) { v.SetDefault("engine.step_output_max_bytes", 1048576) v.SetDefault("engine.default_max_tool_rounds", 10) v.SetDefault("engine.default_max_tool_calls_per_round", 10) + v.SetDefault("engine.max_workflow_depth", 10) // Budget defaults v.SetDefault("engine.budget.reset_mode", budget.ResetModeCalendar) @@ -251,6 +253,7 @@ func Load(cmd *cobra.Command) (*Config, error) { _ = v.BindEnv("engine.allowed_base_urls", "MANTLE_ENGINE_ALLOWED_BASE_URLS") _ = v.BindEnv("engine.allowed_models", "MANTLE_ENGINE_ALLOWED_MODELS") _ = v.BindEnv("engine.max_tool_rounds_limit", "MANTLE_ENGINE_MAX_TOOL_ROUNDS_LIMIT") + _ = v.BindEnv("engine.max_workflow_depth", "MANTLE_ENGINE_MAX_WORKFLOW_DEPTH") // Budget env var bindings _ = v.BindEnv("engine.budget.reset_mode", "MANTLE_ENGINE_BUDGET_RESET_MODE") diff --git a/packages/engine/internal/db/migrations/016_workflow_composition.sql b/packages/engine/internal/db/migrations/016_workflow_composition.sql new file mode 100644 index 0000000..20e2641 --- /dev/null +++ b/packages/engine/internal/db/migrations/016_workflow_composition.sql @@ -0,0 +1,11 @@ +-- +goose Up +ALTER TABLE workflow_executions ADD COLUMN parent_execution_id UUID REFERENCES workflow_executions(id); +ALTER TABLE workflow_executions ADD COLUMN parent_step_name TEXT; +ALTER TABLE workflow_executions ADD COLUMN depth INTEGER NOT NULL DEFAULT 0; +CREATE INDEX idx_workflow_executions_parent ON workflow_executions(parent_execution_id) WHERE parent_execution_id IS NOT NULL; + +-- +goose Down +DROP INDEX IF EXISTS idx_workflow_executions_parent; +ALTER TABLE workflow_executions DROP COLUMN IF EXISTS depth; +ALTER TABLE workflow_executions DROP COLUMN IF EXISTS parent_step_name; +ALTER TABLE workflow_executions DROP COLUMN IF EXISTS parent_execution_id; From f706760a5521219ded2f00282db38b4f87c43556 Mon Sep 17 00:00:00 2001 From: Michael McNees Date: Fri, 27 Mar 2026 00:41:53 -0400 Subject: [PATCH 02/11] feat(engine): workflow/run connector for workflow composition (#54) Add WorkflowConnector that implements connector.Connector to execute child workflows with depth limiting, checkpoint recovery, and parent linkage. Wire into engine via RegisterWorkflowConnector() in both CLI run and serve. Co-Authored-By: Claude Opus 4.6 (1M context) --- packages/engine/internal/audit/audit.go | 3 +- packages/engine/internal/cli/run.go | 2 + packages/engine/internal/cli/serve.go | 2 + packages/engine/internal/engine/engine.go | 1 + .../internal/engine/workflow_connector.go | 188 ++++++++++++++++++ 5 files changed, 195 insertions(+), 1 deletion(-) create mode 100644 packages/engine/internal/engine/workflow_connector.go diff --git a/packages/engine/internal/audit/audit.go b/packages/engine/internal/audit/audit.go index 4dfac91..a733c23 100644 --- a/packages/engine/internal/audit/audit.go +++ b/packages/engine/internal/audit/audit.go @@ -17,7 +17,8 @@ const ( ActionStepSkipped Action = "step.skipped" ActionStepContinuedOnError Action = "step.continued_on_error" ActionExecutionCancelled Action = "execution.cancelled" - ActionArtifactPersisted Action = "artifact.persisted" + ActionChildWorkflowExecuted Action = "workflow.child_executed" + ActionArtifactPersisted Action = "artifact.persisted" // Admin operations. ActionUserCreated Action = "user.created" diff --git a/packages/engine/internal/cli/run.go b/packages/engine/internal/cli/run.go index 01a9015..2f76e12 100644 --- a/packages/engine/internal/cli/run.go +++ b/packages/engine/internal/cli/run.go @@ -64,6 +64,8 @@ func newRunCommand() *cobra.Command { if err != nil { return fmt.Errorf("creating engine: %w", err) } + eng.MaxWorkflowDepth = cfg.Engine.MaxWorkflowDepth + eng.RegisterWorkflowConnector() // Configure credential resolver with Postgres-backed store when encryption key is set. if cfg.Encryption.Key != "" { diff --git a/packages/engine/internal/cli/serve.go b/packages/engine/internal/cli/serve.go index 9c45479..35b0a84 100644 --- a/packages/engine/internal/cli/serve.go +++ b/packages/engine/internal/cli/serve.go @@ -50,6 +50,8 @@ func newServeCommand() *cobra.Command { return fmt.Errorf("creating engine: %w", err) } eng.MaxToolRoundsLimit = cfg.Engine.MaxToolRoundsLimit + eng.MaxWorkflowDepth = cfg.Engine.MaxWorkflowDepth + eng.RegisterWorkflowConnector() // Configure AWS defaults for AI (Bedrock) and S3 connectors. if aiConn, err := eng.Registry.Get("ai/completion"); err == nil { diff --git a/packages/engine/internal/engine/engine.go b/packages/engine/internal/engine/engine.go index d3bafd9..9843983 100644 --- a/packages/engine/internal/engine/engine.go +++ b/packages/engine/internal/engine/engine.go @@ -30,6 +30,7 @@ type Engine struct { CEL *mantleCEL.Evaluator Resolver *secret.Resolver MaxToolRoundsLimit int // admin ceiling for max_rounds; 0 = no limit + MaxWorkflowDepth int // max nesting depth for workflow/run; 0 = use default (10) BudgetChecker *budget.Checker // nil = budget enforcement disabled BudgetStore *budget.Store // nil = token usage recording disabled ArtifactStore *artifact.Store // nil = artifact system disabled diff --git a/packages/engine/internal/engine/workflow_connector.go b/packages/engine/internal/engine/workflow_connector.go new file mode 100644 index 0000000..b4cde6e --- /dev/null +++ b/packages/engine/internal/engine/workflow_connector.go @@ -0,0 +1,188 @@ +package engine + +import ( + "context" + "database/sql" + "encoding/json" + "fmt" + "time" + + "github.com/dvflw/mantle/internal/audit" + "github.com/dvflw/mantle/internal/auth" + "github.com/dvflw/mantle/internal/workflow" +) + +// WorkflowConnector implements connector.Connector for the "workflow/run" action. +// It lives in the engine package (not connector) to avoid circular imports, +// since it needs a reference to *Engine. +type WorkflowConnector struct { + engine *Engine +} + +// Execute runs a child workflow as part of workflow composition. +// +// Params: +// - workflow (string, required): name of the child workflow +// - version (int, optional): version to run; 0 or omitted = latest +// - inputs (map[string]any, optional): input parameters for the child +// - token_budget (int64, optional): token budget override for the child +func (wc *WorkflowConnector) Execute(ctx context.Context, params map[string]any) (map[string]any, error) { + // Extract required workflow name. + workflowName, ok := params["workflow"].(string) + if !ok || workflowName == "" { + return nil, fmt.Errorf("workflow/run: 'workflow' parameter is required and must be a string") + } + + // Extract optional version (default: latest). + version := 0 + if v, ok := params["version"]; ok { + switch tv := v.(type) { + case float64: + version = int(tv) + case int: + version = tv + case json.Number: + i, err := tv.Int64() + if err == nil { + version = int(i) + } + } + } + + // Extract optional inputs. + var inputs map[string]any + if inp, ok := params["inputs"].(map[string]any); ok { + inputs = inp + } + + teamID := auth.TeamIDFromContext(ctx) + + // Resolve latest version if not specified. + if version == 0 { + v, err := workflow.GetLatestVersion(ctx, wc.engine.DB, workflowName) + if err != nil { + return nil, fmt.Errorf("workflow/run: resolving latest version: %w", err) + } + if v == 0 { + return nil, fmt.Errorf("workflow/run: workflow %q not found", workflowName) + } + version = v + } + + // Determine parent execution context. + parentExecID := ExecutionIDFromContext(ctx) + if parentExecID == "" { + return nil, fmt.Errorf("workflow/run: no parent execution ID in context") + } + + // Look up parent step name from params (injected by engine step execution as "_step"). + parentStepName, _ := params["_step"].(string) + + // Check depth limit. + maxDepth := wc.engine.MaxWorkflowDepth + if maxDepth <= 0 { + maxDepth = 10 + } + + var parentDepth int + err := wc.engine.DB.QueryRowContext(ctx, + `SELECT depth FROM workflow_executions WHERE id = $1 AND team_id = $2`, + parentExecID, teamID, + ).Scan(&parentDepth) + if err != nil { + return nil, fmt.Errorf("workflow/run: querying parent depth: %w", err) + } + + childDepth := parentDepth + 1 + if childDepth > maxDepth { + return nil, fmt.Errorf("workflow/run: max workflow depth %d exceeded (current depth: %d)", maxDepth, childDepth) + } + + // Checkpoint recovery: check for existing child execution. + var existingChildID string + err = wc.engine.DB.QueryRowContext(ctx, + `SELECT id FROM workflow_executions + WHERE parent_execution_id = $1 AND parent_step_name = $2 AND team_id = $3 + LIMIT 1`, + parentExecID, parentStepName, teamID, + ).Scan(&existingChildID) + + var childExecID string + + if err == nil { + // Existing child found — resume it. + childExecID = existingChildID + } else if err == sql.ErrNoRows { + // Create new child execution with parent linkage. + inputsJSON, jsonErr := json.Marshal(inputs) + if jsonErr != nil { + return nil, fmt.Errorf("workflow/run: marshaling inputs: %w", jsonErr) + } + + err = wc.engine.DB.QueryRowContext(ctx, + `INSERT INTO workflow_executions + (workflow_name, workflow_version, status, inputs, started_at, team_id, + parent_execution_id, parent_step_name, depth) + VALUES ($1, $2, 'pending', $3, NOW(), $4, $5, $6, $7) + RETURNING id`, + workflowName, version, inputsJSON, teamID, + parentExecID, parentStepName, childDepth, + ).Scan(&childExecID) + if err != nil { + return nil, fmt.Errorf("workflow/run: creating child execution: %w", err) + } + } else { + return nil, fmt.Errorf("workflow/run: checking for existing child: %w", err) + } + + // Run the child workflow. + result, err := wc.engine.resumeExecution(ctx, childExecID, workflowName, version, inputs) + if err != nil { + return nil, fmt.Errorf("workflow/run: child execution failed: %w", err) + } + + // Emit audit event. + wc.engine.Auditor.Emit(ctx, audit.Event{ + Timestamp: time.Now(), + Actor: "engine", + Action: audit.ActionChildWorkflowExecuted, + Resource: audit.Resource{Type: "workflow_execution", ID: childExecID}, + Metadata: map[string]string{ + "parent_execution_id": parentExecID, + "parent_step_name": parentStepName, + "child_workflow": workflowName, + "child_version": fmt.Sprintf("%d", version), + "child_status": result.Status, + "depth": fmt.Sprintf("%d", childDepth), + }, + }) + + // Build output: {execution_id, status, steps: {step_name: {output: ...}}} + stepsOutput := make(map[string]any, len(result.Steps)) + for name, sr := range result.Steps { + stepMap := map[string]any{ + "output": sr.Output, + } + if sr.Error != "" { + stepMap["error"] = sr.Error + } + stepsOutput[name] = stepMap + } + + output := map[string]any{ + "execution_id": childExecID, + "status": result.Status, + "steps": stepsOutput, + } + + if result.Status == "failed" { + return output, fmt.Errorf("child workflow %q failed: %s", workflowName, result.Error) + } + + return output, nil +} + +// RegisterWorkflowConnector registers the workflow/run connector with the engine's registry. +func (e *Engine) RegisterWorkflowConnector() { + e.Registry.Register("workflow/run", &WorkflowConnector{engine: e}) +} From 591a99d59710e145bd21cf87d34e1b22bd0c8484 Mon Sep 17 00:00:00 2001 From: Michael McNees Date: Fri, 27 Mar 2026 00:41:58 -0400 Subject: [PATCH 03/11] feat(cli): recursive cancellation cascade for workflow composition (#54) Replace single-execution cancellation with recursive CTE that cancels the entire parent-child execution tree, including all nested step executions. Co-Authored-By: Claude Opus 4.6 (1M context) --- packages/engine/internal/cli/cancel.go | 26 ++++++++++++++++++-------- 1 file changed, 18 insertions(+), 8 deletions(-) diff --git a/packages/engine/internal/cli/cancel.go b/packages/engine/internal/cli/cancel.go index d085405..71bc1c2 100644 --- a/packages/engine/internal/cli/cancel.go +++ b/packages/engine/internal/cli/cancel.go @@ -28,11 +28,16 @@ func newCancelCommand() *cobra.Command { } defer database.Close() - // Only cancel if currently pending or running. + // Cancel the execution and all child executions recursively. result, err := database.ExecContext(cmd.Context(), - `UPDATE workflow_executions - SET status = 'cancelled', completed_at = NOW(), updated_at = NOW() - WHERE id = $1 AND status IN ('pending', 'running')`, + `WITH RECURSIVE children AS ( + SELECT id FROM workflow_executions WHERE id = $1 + UNION ALL + SELECT e.id FROM workflow_executions e + JOIN children c ON e.parent_execution_id = c.id + ) + UPDATE workflow_executions SET status = 'cancelled', completed_at = NOW(), updated_at = NOW() + WHERE id IN (SELECT id FROM children) AND status IN ('pending', 'running', 'queued')`, execID, ) if err != nil { @@ -57,11 +62,16 @@ func newCancelCommand() *cobra.Command { return nil } - // Also mark any running/pending steps as cancelled. + // Also mark any running/pending steps in the tree as cancelled. database.ExecContext(cmd.Context(), - `UPDATE step_executions - SET status = 'cancelled', completed_at = NOW(), updated_at = NOW() - WHERE execution_id = $1 AND status IN ('pending', 'running')`, + `WITH RECURSIVE children AS ( + SELECT id FROM workflow_executions WHERE id = $1 + UNION ALL + SELECT e.id FROM workflow_executions e + JOIN children c ON e.parent_execution_id = c.id + ) + UPDATE step_executions SET status = 'cancelled', completed_at = NOW(), updated_at = NOW() + WHERE execution_id IN (SELECT id FROM children) AND status IN ('pending', 'running')`, execID, ) From 1761279af4c32a3d4e9582b46a67f9eef6cd3767 Mon Sep 17 00:00:00 2001 From: Michael McNees Date: Fri, 27 Mar 2026 00:50:18 -0400 Subject: [PATCH 04/11] feat(cli): show child workflow details in logs and status (#54) Add --shallow flag to logs command and recursive child execution display. Status command now shows child execution tree under "Children:" heading. Co-Authored-By: Claude Opus 4.6 (1M context) --- packages/engine/internal/cli/logs.go | 145 +++++++++++++++++++++---- packages/engine/internal/cli/status.go | 35 ++++++ 2 files changed, 159 insertions(+), 21 deletions(-) diff --git a/packages/engine/internal/cli/logs.go b/packages/engine/internal/cli/logs.go index d9f78b3..d378a1c 100644 --- a/packages/engine/internal/cli/logs.go +++ b/packages/engine/internal/cli/logs.go @@ -1,6 +1,7 @@ package cli import ( + "database/sql" "encoding/json" "fmt" "strconv" @@ -13,6 +14,25 @@ import ( "github.com/spf13/cobra" ) +// subStep represents a sub-step (tool call) within a step execution. +type subStep struct { + StepName string `json:"step_name"` + Status string `json:"status"` + Started *time.Time `json:"started_at,omitempty"` + Completed *time.Time `json:"completed_at,omitempty"` +} + +// stepInfo represents a top-level step execution. +type stepInfo struct { + ID string `json:"id"` + Name string `json:"name"` + Status string `json:"status"` + Error string `json:"error,omitempty"` + StartedAt *time.Time `json:"started_at,omitempty"` + CompletedAt *time.Time `json:"completed_at,omitempty"` + SubSteps []subStep `json:"sub_steps,omitempty"` +} + func newLogsCommand() *cobra.Command { cmd := &cobra.Command{ Use: "logs [execution-id]", @@ -38,6 +58,7 @@ When called without arguments, lists recent executions with optional filters.`, cmd.Flags().String("status", "", "filter by status (pending, running, completed, failed, cancelled)") cmd.Flags().String("since", "", "filter by time (e.g., 1h, 24h, 7d)") cmd.Flags().Int("limit", 20, "max results to return") + cmd.Flags().Bool("shallow", false, "suppress child workflow details") return cmd } @@ -81,12 +102,6 @@ func showExecutionDetail(cmd *cobra.Command, execID string) error { defer rows.Close() // Fetch sub-steps grouped by parent. - type subStep struct { - StepName string `json:"step_name"` - Status string `json:"status"` - Started *time.Time `json:"started_at,omitempty"` - Completed *time.Time `json:"completed_at,omitempty"` - } subStepsByParent := make(map[string][]subStep) subRows, err := database.QueryContext(cmd.Context(), `SELECT parent_step_id, step_name, status, started_at, completed_at @@ -105,16 +120,6 @@ func showExecutionDetail(cmd *cobra.Command, execID string) error { } // Collect step data. - type stepInfo struct { - ID string `json:"id"` - Name string `json:"name"` - Status string `json:"status"` - Error string `json:"error,omitempty"` - StartedAt *time.Time `json:"started_at,omitempty"` - CompletedAt *time.Time `json:"completed_at,omitempty"` - SubSteps []subStep `json:"sub_steps,omitempty"` - } - var stepsData []stepInfo for rows.Next() { var stepID, stepName, stepStatus string @@ -173,7 +178,16 @@ func showExecutionDetail(cmd *cobra.Command, execID string) error { } fmt.Fprintln(cmd.OutOrStdout()) + shallow, _ := cmd.Flags().GetBool("shallow") + fmt.Fprintln(cmd.OutOrStdout(), "Steps:") + printSteps(cmd, database, execID, teamID, stepsData, shallow, " ") + + return nil +} + +// printSteps renders step details, optionally recursing into child workflow executions. +func printSteps(cmd *cobra.Command, database *sql.DB, execID, teamID string, stepsData []stepInfo, shallow bool, indent string) { for _, si := range stepsData { icon := statusIcon(si.Status) duration := "" @@ -181,9 +195,9 @@ func showExecutionDetail(cmd *cobra.Command, execID string) error { duration = fmt.Sprintf(" %s", si.CompletedAt.Sub(*si.StartedAt).Round(time.Millisecond)) } - fmt.Fprintf(cmd.OutOrStdout(), " %s %-20s %s%s\n", icon, si.Name, si.Status, duration) + fmt.Fprintf(cmd.OutOrStdout(), "%s%s %-20s %s%s\n", indent, icon, si.Name, si.Status, duration) if si.Error != "" { - fmt.Fprintf(cmd.OutOrStdout(), " error: %s\n", si.Error) + fmt.Fprintf(cmd.OutOrStdout(), "%s error: %s\n", indent, si.Error) } // Render sub-steps (tool calls) as indented tree. @@ -208,7 +222,7 @@ func showExecutionDetail(cmd *cobra.Command, execID string) error { if isLastRound { connector = "└─" } - fmt.Fprintf(cmd.OutOrStdout(), " %s round %s\n", connector, round) + fmt.Fprintf(cmd.OutOrStdout(), "%s %s round %s\n", indent, connector, round) for si, s := range rounds[round] { isLast := si == len(rounds[round])-1 @@ -229,13 +243,102 @@ func showExecutionDetail(cmd *cobra.Command, execID string) error { if s.Started != nil && s.Completed != nil { subDur = fmt.Sprintf(" %s", s.Completed.Sub(*s.Started).Round(time.Millisecond)) } - fmt.Fprintf(cmd.OutOrStdout(), " %s %s %-16s %s%s\n", prefix, subConn, toolName, s.Status, subDur) + fmt.Fprintf(cmd.OutOrStdout(), "%s %s %s %-16s %s%s\n", indent, prefix, subConn, toolName, s.Status, subDur) } } } + + // Show child workflow execution details if not in shallow mode. + if !shallow { + printChildExecution(cmd, database, execID, si.Name, teamID, indent+" ") + } } +} - return nil +// printChildExecution queries for a child workflow execution spawned by a given step +// and recursively renders its steps. +func printChildExecution(cmd *cobra.Command, database *sql.DB, parentExecID, parentStepName, teamID, indent string) { + rows, err := database.QueryContext(cmd.Context(), + `SELECT id, workflow_name, status FROM workflow_executions + WHERE parent_execution_id = $1 AND parent_step_name = $2 AND team_id = $3`, + parentExecID, parentStepName, teamID, + ) + if err != nil { + return + } + defer rows.Close() + + for rows.Next() { + var childID, childWorkflow, childStatus string + if err := rows.Scan(&childID, &childWorkflow, &childStatus); err != nil { + continue + } + + fmt.Fprintf(cmd.OutOrStdout(), "%s↳ child: %s (%s) [%s]\n", indent, childWorkflow, childID, childStatus) + + // Fetch child's steps. + childSteps := fetchStepsForExecution(cmd, database, childID) + if len(childSteps) > 0 { + printSteps(cmd, database, childID, teamID, childSteps, false, indent+" ") + } + } +} + +// fetchStepsForExecution retrieves step data for a given execution ID. +func fetchStepsForExecution(cmd *cobra.Command, database *sql.DB, execID string) []stepInfo { + stepRows, err := database.QueryContext(cmd.Context(), + `SELECT id, step_name, status, error, started_at, completed_at + FROM step_executions WHERE execution_id = $1 AND parent_step_id IS NULL + ORDER BY created_at ASC`, execID, + ) + if err != nil { + return nil + } + defer stepRows.Close() + + // Fetch sub-steps for the child execution. + subStepsByParent := make(map[string][]subStep) + subRows, err := database.QueryContext(cmd.Context(), + `SELECT parent_step_id, step_name, status, started_at, completed_at + FROM step_executions WHERE execution_id = $1 AND parent_step_id IS NOT NULL + ORDER BY created_at ASC`, execID, + ) + if err == nil { + defer subRows.Close() + for subRows.Next() { + var parentID, sName, sStatus string + var sStarted, sCompleted *time.Time + if err := subRows.Scan(&parentID, &sName, &sStatus, &sStarted, &sCompleted); err == nil { + subStepsByParent[parentID] = append(subStepsByParent[parentID], subStep{sName, sStatus, sStarted, sCompleted}) + } + } + } + + var steps []stepInfo + for stepRows.Next() { + var stepID, stepName, stepStatus string + var stepError *string + var stepStarted, stepCompleted *time.Time + if err := stepRows.Scan(&stepID, &stepName, &stepStatus, &stepError, &stepStarted, &stepCompleted); err != nil { + continue + } + + si := stepInfo{ + ID: stepID, + Name: stepName, + Status: stepStatus, + StartedAt: stepStarted, + CompletedAt: stepCompleted, + } + if stepError != nil && *stepError != "" { + si.Error = *stepError + } + if subs, ok := subStepsByParent[stepID]; ok { + si.SubSteps = subs + } + steps = append(steps, si) + } + return steps } // listExecutions lists recent executions with optional filters. diff --git a/packages/engine/internal/cli/status.go b/packages/engine/internal/cli/status.go index ea90783..ddcf964 100644 --- a/packages/engine/internal/cli/status.go +++ b/packages/engine/internal/cli/status.go @@ -99,6 +99,41 @@ func newStatusCommand() *cobra.Command { fmt.Fprintf(cmd.OutOrStdout(), " %s: %d\n", stepStatus, count) } + // Query for child executions. + childRows, err := database.QueryContext(cmd.Context(), + `SELECT id, workflow_name, workflow_version, status + FROM workflow_executions + WHERE parent_execution_id = $1 + ORDER BY started_at`, execID, + ) + if err == nil { + defer childRows.Close() + + type childExec struct { + ID string + Workflow string + Version int + Status string + } + var children []childExec + for childRows.Next() { + var c childExec + if err := childRows.Scan(&c.ID, &c.Workflow, &c.Version, &c.Status); err == nil { + children = append(children, c) + } + } + + if len(children) > 0 { + fmt.Fprintln(cmd.OutOrStdout()) + fmt.Fprintln(cmd.OutOrStdout(), "Children:") + for _, c := range children { + icon := statusIcon(c.Status) + fmt.Fprintf(cmd.OutOrStdout(), " %s %-20s v%-4d %s %s\n", + icon, c.Workflow, c.Version, c.Status, c.ID) + } + } + } + return nil }, } From 7d5ad80b064f6273bb25a5d7c6dc812bddeeaab1 Mon Sep 17 00:00:00 2001 From: Michael McNees Date: Fri, 27 Mar 2026 00:50:24 -0400 Subject: [PATCH 05/11] docs: document workflow/run connector for workflow composition (#54) Add workflow/run section to connector reference with params, output format, CEL access patterns, depth limiting, checkpoint recovery, and examples. Co-Authored-By: Claude Opus 4.6 (1M context) --- .../docs/workflow-reference/connectors.md | 78 +++++++++++++++++++ 1 file changed, 78 insertions(+) diff --git a/packages/site/src/content/docs/workflow-reference/connectors.md b/packages/site/src/content/docs/workflow-reference/connectors.md index 99f8ac5..6cb45ac 100644 --- a/packages/site/src/content/docs/workflow-reference/connectors.md +++ b/packages/site/src/content/docs/workflow-reference/connectors.md @@ -595,6 +595,84 @@ mantle secrets create --name minio --type generic \ --field endpoint=http://localhost:9000 ``` +## workflow/run + +Invokes another workflow as a child execution. The child workflow runs synchronously within the parent step, with full checkpoint-and-resume support. If the parent crashes and recovers, the child execution is reused rather than re-executed. + +**Params:** + +| Param | Type | Required | Description | +|---|---|---|---| +| `workflow` | string | Yes | Name of the child workflow to execute. | +| `version` | integer | No | Specific version to run. Omit to use the latest applied version. | +| `inputs` | map | No | Input parameters to pass to the child workflow. | +| `token_budget` | integer | No | Token budget override for the child execution. | + +**Output:** + +| Field | Type | Description | +|---|---|---| +| `execution_id` | string | The child workflow's execution ID. | +| `status` | string | Final status of the child execution (`completed` or `failed`). | +| `steps` | map | Map of child step names to their results. Each entry has an `output` field (and optionally `error`). | + +**Accessing child results in CEL:** + +Child step outputs are nested under the parent step's output: + +``` +steps['my-step'].output.steps['child-step'].output.field +``` + +**Depth limiting:** Workflow nesting depth is configurable via `engine.max_workflow_depth` (default: `10`). Exceeding this limit returns an error. + +**Checkpoint recovery:** If the parent workflow crashes mid-execution, the child execution record is preserved in the database. On resume, the engine detects the existing child and reuses its result instead of creating a duplicate. + +**Cancellation:** Running `mantle cancel` on a parent execution cascades cancellation to all child executions. + +**Example -- parent workflow invoking a reusable child:** + +```yaml +name: order-pipeline +description: Process an order using a reusable validation workflow +steps: + - name: validate + action: workflow/run + params: + workflow: validate-order + inputs: + order_id: "{{ inputs.order_id }}" + + - name: notify + action: slack/send + credential: slack-bot + params: + channel: "#orders" + text: "Order validated: {{ steps.validate.output.steps['check-inventory'].output.available }}" +``` + +**Example -- child workflow (validate-order):** + +```yaml +name: validate-order +description: Validate an order's inventory and pricing +inputs: + order_id: + type: string +steps: + - name: check-inventory + action: http/request + params: + method: GET + url: "https://api.example.com/inventory/{{ inputs.order_id }}" + + - name: check-pricing + action: http/request + params: + method: GET + url: "https://api.example.com/pricing/{{ inputs.order_id }}" +``` + ## docker/run Runs a Docker container to completion and captures its output. The container is created, started, waited on, and optionally removed. Non-zero exit codes do not constitute a step failure — use `if` conditions to branch on exit code. From 712c8f7911c692bf9793f64fd8012d8ece311ec6 Mon Sep 17 00:00:00 2001 From: Michael McNees Date: Fri, 27 Mar 2026 00:50:29 -0400 Subject: [PATCH 06/11] test: integration tests for workflow composition (#54) Add tests for workflow/run connector: basic invocation with child output wrapping, depth limit enforcement, checkpoint recovery (reuse existing child), and unit tests for depth checking and output wrapping logic. Co-Authored-By: Claude Opus 4.6 (1M context) --- .../engine/workflow_connector_test.go | 466 ++++++++++++++++++ 1 file changed, 466 insertions(+) create mode 100644 packages/engine/internal/engine/workflow_connector_test.go diff --git a/packages/engine/internal/engine/workflow_connector_test.go b/packages/engine/internal/engine/workflow_connector_test.go new file mode 100644 index 0000000..ed1fd64 --- /dev/null +++ b/packages/engine/internal/engine/workflow_connector_test.go @@ -0,0 +1,466 @@ +package engine + +import ( + "context" + "database/sql" + "encoding/json" + "fmt" + "testing" + + "github.com/dvflw/mantle/internal/workflow" +) + +// TestWorkflowConnector_BasicInvocation creates a parent and child workflow, +// executes the parent with a workflow/run step, and verifies the child ran +// and output is wrapped correctly. +func TestWorkflowConnector_BasicInvocation(t *testing.T) { + database := setupTestDB(t) + + // Apply the child workflow first. + childYAML := []byte(`name: child-wf +description: A simple child workflow +steps: + - name: greet + action: test/echo + params: + message: "hello from child" +`) + applyWorkflow(t, database, childYAML) + + // Apply the parent workflow that invokes the child via workflow/run. + parentYAML := []byte(`name: parent-wf +description: Parent that calls child +steps: + - name: call-child + action: workflow/run + params: + workflow: child-wf +`) + parentVersion := applyWorkflow(t, database, parentYAML) + + // Create engine and register a mock "test/echo" connector. + eng, err := New(database) + if err != nil { + t.Fatalf("New() error: %v", err) + } + eng.Registry.Register("test/echo", &mockConnector{ + fn: func(ctx context.Context, params map[string]any) (map[string]any, error) { + msg, _ := params["message"].(string) + return map[string]any{"echoed": msg}, nil + }, + }) + eng.RegisterWorkflowConnector() + + // Execute the parent. We use WithExecutionID so the workflow connector + // can find the parent execution ID when called during resumeExecution. + // First, create the execution record manually (same as Engine.Execute does). + ctx := context.Background() + parentExecID, err := eng.createExecution(ctx, "parent-wf", parentVersion, nil) + if err != nil { + t.Fatalf("createExecution error: %v", err) + } + + ctx = WithExecutionID(ctx, parentExecID) + result, err := eng.resumeExecution(ctx, parentExecID, "parent-wf", parentVersion, nil) + if err != nil { + t.Fatalf("resumeExecution() error: %v", err) + } + + if result.Status != "completed" { + t.Fatalf("parent status = %q, want %q (error: %s)", result.Status, "completed", result.Error) + } + + // Verify the call-child step output contains child execution details. + callChild := result.Steps["call-child"] + if callChild.Status != "completed" { + t.Fatalf("call-child status = %q, want %q", callChild.Status, "completed") + } + + // Output should have execution_id, status, and steps. + if callChild.Output["status"] != "completed" { + t.Errorf("child output.status = %v, want %q", callChild.Output["status"], "completed") + } + if callChild.Output["execution_id"] == nil || callChild.Output["execution_id"] == "" { + t.Error("child output.execution_id should be non-empty") + } + + // Verify nested steps output. + stepsMap, ok := callChild.Output["steps"].(map[string]any) + if !ok { + t.Fatalf("child output.steps is not a map, got %T", callChild.Output["steps"]) + } + greetStep, ok := stepsMap["greet"].(map[string]any) + if !ok { + t.Fatalf("child output.steps['greet'] is not a map, got %T", stepsMap["greet"]) + } + greetOutput, ok := greetStep["output"].(map[string]any) + if !ok { + t.Fatalf("child output.steps['greet'].output is not a map, got %T", greetStep["output"]) + } + if greetOutput["echoed"] != "hello from child" { + t.Errorf("child greet output.echoed = %v, want %q", greetOutput["echoed"], "hello from child") + } + + // Verify child execution exists in the database with correct linkage. + childExecID := callChild.Output["execution_id"].(string) + var parentID, parentStepName string + var depth int + err = database.QueryRowContext(ctx, + `SELECT parent_execution_id, parent_step_name, depth + FROM workflow_executions WHERE id = $1`, + childExecID, + ).Scan(&parentID, &parentStepName, &depth) + if err != nil { + t.Fatalf("querying child execution: %v", err) + } + if parentID != parentExecID { + t.Errorf("child parent_execution_id = %q, want %q", parentID, parentExecID) + } + if parentStepName != "call-child" { + t.Errorf("child parent_step_name = %q, want %q", parentStepName, "call-child") + } + if depth != 1 { + t.Errorf("child depth = %d, want 1", depth) + } +} + +// TestWorkflowConnector_DepthLimit verifies that exceeding the max workflow +// nesting depth returns an error. +func TestWorkflowConnector_DepthLimit(t *testing.T) { + database := setupTestDB(t) + + // Apply a child workflow. + childYAML := []byte(`name: deep-child +description: A simple child +steps: + - name: noop + action: test/noop + params: + msg: "noop" +`) + applyWorkflow(t, database, childYAML) + + // Apply a parent workflow that invokes the child. + parentYAML := []byte(`name: deep-parent +description: Parent hitting depth limit +steps: + - name: call-deep + action: workflow/run + params: + workflow: deep-child +`) + parentVersion := applyWorkflow(t, database, parentYAML) + + eng, err := New(database) + if err != nil { + t.Fatalf("New() error: %v", err) + } + eng.Registry.Register("test/noop", &mockConnector{ + fn: func(_ context.Context, _ map[string]any) (map[string]any, error) { + return map[string]any{"done": true}, nil + }, + }) + eng.RegisterWorkflowConnector() + eng.MaxWorkflowDepth = 1 // Set max depth to 1. + + // Create parent execution at depth 0. + ctx := context.Background() + parentExecID, err := eng.createExecution(ctx, "deep-parent", parentVersion, nil) + if err != nil { + t.Fatalf("createExecution error: %v", err) + } + + ctx = WithExecutionID(ctx, parentExecID) + result, err := eng.resumeExecution(ctx, parentExecID, "deep-parent", parentVersion, nil) + if err != nil { + t.Fatalf("resumeExecution() error: %v", err) + } + + // The parent should fail because the child would be at depth 1, + // and the parent itself is at depth 0, so child depth = 1 which exceeds max depth 1. + // Actually: childDepth = parentDepth(0) + 1 = 1, maxDepth = 1, + // and the check is childDepth > maxDepth, so 1 > 1 is false. + // We need maxDepth = 0 for it to fail, but 0 means "use default 10". + // Let's set it differently. + // Re-check: the connector sets maxDepth=10 when MaxWorkflowDepth <= 0. + // When MaxWorkflowDepth = 1, childDepth = 1, and 1 > 1 is false. + // So this should succeed. Let me set it to trigger failure properly. + + // Actually, we should verify what happens. The depth check in the connector is: + // childDepth := parentDepth + 1 + // if childDepth > maxDepth { error } + // With MaxWorkflowDepth=1, parentDepth=0, childDepth=1, 1>1=false, so it passes. + // To trigger the error, we need to simulate the parent being at depth >= maxDepth. + // Let's just verify it works at depth 1, then test the error case with a + // parent already at depth 1 and max=1. + + // This test should succeed since parent is at depth 0 and max is 1. + if result.Status != "completed" { + // If it failed for depth reasons, that's fine for the test intent. + // But let's be precise. + t.Logf("result status=%s, error=%s", result.Status, result.Error) + } + + // Now test the actual depth limit: create an execution at depth 1 and max=1. + var deepExecID string + err = database.QueryRowContext(ctx, + `INSERT INTO workflow_executions + (workflow_name, workflow_version, status, started_at, team_id, depth) + VALUES ($1, $2, 'pending', NOW(), $3, $4) + RETURNING id`, + "deep-parent", parentVersion, "00000000-0000-0000-0000-000000000001", 1, + ).Scan(&deepExecID) + if err != nil { + t.Fatalf("inserting deep execution: %v", err) + } + + ctx2 := WithExecutionID(context.Background(), deepExecID) + result2, err := eng.resumeExecution(ctx2, deepExecID, "deep-parent", parentVersion, nil) + if err != nil { + t.Fatalf("resumeExecution() error: %v", err) + } + + // The call-deep step should fail with a depth limit error. + if result2.Status != "failed" { + t.Errorf("expected failed status for depth limit, got %q", result2.Status) + } + callDeep := result2.Steps["call-deep"] + if callDeep.Status != "failed" { + t.Errorf("call-deep status = %q, want %q", callDeep.Status, "failed") + } + if callDeep.Error == "" { + t.Error("call-deep error should be non-empty") + } + t.Logf("depth limit error: %s", callDeep.Error) +} + +// TestWorkflowConnector_CheckpointRecovery verifies that if a child execution +// already exists (from a previous attempt), the connector reuses it rather +// than creating a new one. +func TestWorkflowConnector_CheckpointRecovery(t *testing.T) { + database := setupTestDB(t) + + // Apply the child workflow. + childYAML := []byte(`name: recoverable-child +description: Child for checkpoint test +steps: + - name: work + action: test/echo + params: + message: "working" +`) + childVersion := applyWorkflow(t, database, childYAML) + + // Apply the parent workflow. + parentYAML := []byte(`name: recoverable-parent +description: Parent for checkpoint test +steps: + - name: run-child + action: workflow/run + params: + workflow: recoverable-child +`) + parentVersion := applyWorkflow(t, database, parentYAML) + + eng, err := New(database) + if err != nil { + t.Fatalf("New() error: %v", err) + } + + callCount := 0 + eng.Registry.Register("test/echo", &mockConnector{ + fn: func(ctx context.Context, params map[string]any) (map[string]any, error) { + callCount++ + msg, _ := params["message"].(string) + return map[string]any{"echoed": msg, "call": callCount}, nil + }, + }) + eng.RegisterWorkflowConnector() + + ctx := context.Background() + + // Create the parent execution. + parentExecID, err := eng.createExecution(ctx, "recoverable-parent", parentVersion, nil) + if err != nil { + t.Fatalf("createExecution error: %v", err) + } + + // Pre-create a completed child execution (simulating a previous run that completed + // before the parent crashed). + childOutput := map[string]any{"echoed": "working", "call": 99} + childOutputJSON, _ := json.Marshal(childOutput) + var childExecID string + err = database.QueryRowContext(ctx, + `INSERT INTO workflow_executions + (workflow_name, workflow_version, status, started_at, completed_at, team_id, + parent_execution_id, parent_step_name, depth) + VALUES ($1, $2, 'completed', NOW(), NOW(), $3, $4, $5, $6) + RETURNING id`, + "recoverable-child", childVersion, "00000000-0000-0000-0000-000000000001", + parentExecID, "run-child", 1, + ).Scan(&childExecID) + if err != nil { + t.Fatalf("inserting child execution: %v", err) + } + + // Insert the child's completed step. + _, err = database.ExecContext(ctx, + `INSERT INTO step_executions (execution_id, step_name, status, output, completed_at) + VALUES ($1, 'work', 'completed', $2, NOW())`, + childExecID, childOutputJSON, + ) + if err != nil { + t.Fatalf("inserting child step: %v", err) + } + + // Execute the parent. The workflow connector should find the existing child + // and reuse it, not create a new one. + ctx = WithExecutionID(ctx, parentExecID) + result, err := eng.resumeExecution(ctx, parentExecID, "recoverable-parent", parentVersion, nil) + if err != nil { + t.Fatalf("resumeExecution() error: %v", err) + } + + if result.Status != "completed" { + t.Fatalf("parent status = %q, want %q (error: %s)", result.Status, "completed", result.Error) + } + + // The test/echo connector should NOT have been called because the child + // was already completed (checkpoint recovery). + if callCount != 0 { + t.Errorf("test/echo call count = %d, want 0 (child should be reused)", callCount) + } + + // Verify the output references the pre-existing child execution. + runChild := result.Steps["run-child"] + if runChild.Output["execution_id"] != childExecID { + t.Errorf("output.execution_id = %v, want %q", runChild.Output["execution_id"], childExecID) + } + + // Verify only one child execution exists (no duplicate created). + var childCount int + err = database.QueryRowContext(ctx, + `SELECT COUNT(*) FROM workflow_executions + WHERE parent_execution_id = $1 AND parent_step_name = 'run-child'`, + parentExecID, + ).Scan(&childCount) + if err != nil { + t.Fatalf("counting child executions: %v", err) + } + if childCount != 1 { + t.Errorf("child execution count = %d, want 1", childCount) + } +} + +// TestWorkflowConnector_DepthCheck_Unit tests the depth checking logic directly +// without a full workflow execution. +func TestWorkflowConnector_DepthCheck_Unit(t *testing.T) { + tests := []struct { + name string + parentDepth int + maxDepth int + wantErr bool + }{ + {"depth 0, max 10", 0, 10, false}, + {"depth 9, max 10", 9, 10, false}, + {"depth 10, max 10", 10, 10, true}, + {"depth 5, max 5", 5, 5, true}, + {"depth 0, max 1", 0, 1, false}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + childDepth := tt.parentDepth + 1 + exceeded := childDepth > tt.maxDepth + if exceeded != tt.wantErr { + t.Errorf("childDepth(%d) > maxDepth(%d) = %v, want %v", + childDepth, tt.maxDepth, exceeded, tt.wantErr) + } + }) + } +} + +// TestWorkflowConnector_OutputWrapping_Unit tests that child execution results +// are wrapped in the expected format. +func TestWorkflowConnector_OutputWrapping_Unit(t *testing.T) { + // Simulate the output wrapping logic from WorkflowConnector.Execute. + childResult := &ExecutionResult{ + ExecutionID: "child-123", + Status: "completed", + Steps: map[string]StepResult{ + "step-a": {Status: "completed", Output: map[string]any{"value": "alpha"}}, + "step-b": {Status: "completed", Output: map[string]any{"value": "beta"}}, + }, + } + + stepsOutput := make(map[string]any, len(childResult.Steps)) + for name, sr := range childResult.Steps { + stepMap := map[string]any{ + "output": sr.Output, + } + if sr.Error != "" { + stepMap["error"] = sr.Error + } + stepsOutput[name] = stepMap + } + + output := map[string]any{ + "execution_id": childResult.ExecutionID, + "status": childResult.Status, + "steps": stepsOutput, + } + + if output["execution_id"] != "child-123" { + t.Errorf("execution_id = %v, want %q", output["execution_id"], "child-123") + } + if output["status"] != "completed" { + t.Errorf("status = %v, want %q", output["status"], "completed") + } + + steps := output["steps"].(map[string]any) + stepA := steps["step-a"].(map[string]any) + stepAOutput := stepA["output"].(map[string]any) + if stepAOutput["value"] != "alpha" { + t.Errorf("steps['step-a'].output.value = %v, want %q", stepAOutput["value"], "alpha") + } + + // Verify no error key when error is empty. + if _, hasErr := stepA["error"]; hasErr { + t.Error("step-a should not have 'error' key when error is empty") + } + + // Test with error. + failedResult := &ExecutionResult{ + Status: "failed", + Steps: map[string]StepResult{ + "fail-step": {Status: "failed", Error: "something broke", Output: map[string]any{}}, + }, + } + failStepsOutput := make(map[string]any) + for name, sr := range failedResult.Steps { + stepMap := map[string]any{"output": sr.Output} + if sr.Error != "" { + stepMap["error"] = sr.Error + } + failStepsOutput[name] = stepMap + } + failStep := failStepsOutput["fail-step"].(map[string]any) + if failStep["error"] != "something broke" { + t.Errorf("fail-step error = %v, want %q", failStep["error"], "something broke") + } +} + +// applyWorkflowVersioned is a helper that stores a workflow and returns both the version and any error. +// Used in tests that need to control version behavior. +func applyWorkflowVersioned(t *testing.T, database *sql.DB, yamlContent []byte) (int, error) { + t.Helper() + result, err := workflow.ParseBytes(yamlContent) + if err != nil { + return 0, fmt.Errorf("ParseBytes: %w", err) + } + version, err := workflow.Save(context.Background(), database, result, yamlContent) + if err != nil { + return 0, fmt.Errorf("Save: %w", err) + } + return version, nil +} From 50623c819d297b9459a12a5baa83d22390d15b46 Mon Sep 17 00:00:00 2001 From: Michael McNees Date: Fri, 27 Mar 2026 10:07:43 -0400 Subject: [PATCH 07/11] fix: lint, transactional cancel, execution tree, cancellation check (PR review) - Remove unused applyWorkflowVersioned function and dead imports - Wrap cancel command in transaction, check both ExecContext errors, emit audit events - Replace direct-children query with recursive CTE for full execution tree with depth - Add per-step cancellation check in resumeExecution to respect external cancels - Guard updateExecutionStatus against overwriting cancelled status - Add --max-workflow-depth CLI flag for consistency Co-Authored-By: Claude Opus 4.6 (1M context) --- packages/engine/internal/cli/cancel.go | 53 ++++++++++++++++--- packages/engine/internal/cli/root.go | 1 + packages/engine/internal/cli/status.go | 26 +++++---- packages/engine/internal/config/config.go | 3 ++ packages/engine/internal/engine/engine.go | 18 +++++++ .../engine/workflow_connector_test.go | 19 ------- 6 files changed, 84 insertions(+), 36 deletions(-) diff --git a/packages/engine/internal/cli/cancel.go b/packages/engine/internal/cli/cancel.go index 71bc1c2..b5654f3 100644 --- a/packages/engine/internal/cli/cancel.go +++ b/packages/engine/internal/cli/cancel.go @@ -2,7 +2,9 @@ package cli import ( "fmt" + "time" + "github.com/dvflw/mantle/internal/audit" "github.com/dvflw/mantle/internal/config" "github.com/dvflw/mantle/internal/db" "github.com/spf13/cobra" @@ -28,8 +30,15 @@ func newCancelCommand() *cobra.Command { } defer database.Close() + tx, err := database.BeginTx(cmd.Context(), nil) + if err != nil { + return fmt.Errorf("starting transaction: %w", err) + } + defer tx.Rollback() //nolint:errcheck + // Cancel the execution and all child executions recursively. - result, err := database.ExecContext(cmd.Context(), + // Use RETURNING id to capture which executions were cancelled. + rows, err := tx.QueryContext(cmd.Context(), `WITH RECURSIVE children AS ( SELECT id FROM workflow_executions WHERE id = $1 UNION ALL @@ -37,19 +46,29 @@ func newCancelCommand() *cobra.Command { JOIN children c ON e.parent_execution_id = c.id ) UPDATE workflow_executions SET status = 'cancelled', completed_at = NOW(), updated_at = NOW() - WHERE id IN (SELECT id FROM children) AND status IN ('pending', 'running', 'queued')`, + WHERE id IN (SELECT id FROM children) AND status IN ('pending', 'running', 'queued') + RETURNING id`, execID, ) if err != nil { return fmt.Errorf("cancelling execution: %w", err) } - rows, err := result.RowsAffected() - if err != nil { - return fmt.Errorf("checking result: %w", err) + var cancelledIDs []string + for rows.Next() { + var id string + if err := rows.Scan(&id); err != nil { + rows.Close() + return fmt.Errorf("scanning cancelled id: %w", err) + } + cancelledIDs = append(cancelledIDs, id) + } + rows.Close() + if err := rows.Err(); err != nil { + return fmt.Errorf("iterating cancelled ids: %w", err) } - if rows == 0 { + if len(cancelledIDs) == 0 { // Check if execution exists at all. var status string err := database.QueryRowContext(cmd.Context(), @@ -63,7 +82,7 @@ func newCancelCommand() *cobra.Command { } // Also mark any running/pending steps in the tree as cancelled. - database.ExecContext(cmd.Context(), + _, err = tx.ExecContext(cmd.Context(), `WITH RECURSIVE children AS ( SELECT id FROM workflow_executions WHERE id = $1 UNION ALL @@ -74,8 +93,26 @@ func newCancelCommand() *cobra.Command { WHERE execution_id IN (SELECT id FROM children) AND status IN ('pending', 'running')`, execID, ) + if err != nil { + return fmt.Errorf("cancelling step executions: %w", err) + } + + if err := tx.Commit(); err != nil { + return fmt.Errorf("committing cancellation: %w", err) + } + + // Emit audit events for each cancelled execution. + auditor := &audit.PostgresEmitter{DB: database} + for _, id := range cancelledIDs { + _ = auditor.Emit(cmd.Context(), audit.Event{ + Timestamp: time.Now(), + Actor: "cli", + Action: audit.ActionExecutionCancelled, + Resource: audit.Resource{Type: "workflow_execution", ID: id}, + }) + } - fmt.Fprintf(cmd.OutOrStdout(), "Cancelled execution %s\n", execID) + fmt.Fprintf(cmd.OutOrStdout(), "Cancelled execution %s (%d executions affected)\n", execID, len(cancelledIDs)) return nil }, } diff --git a/packages/engine/internal/cli/root.go b/packages/engine/internal/cli/root.go index 125ecd5..f363686 100644 --- a/packages/engine/internal/cli/root.go +++ b/packages/engine/internal/cli/root.go @@ -33,6 +33,7 @@ Full documentation: https://mantle.dvflw.co/docs`, cmd.PersistentFlags().String("log-level", "", "log level (debug, info, warn, error)") cmd.PersistentFlags().String("api-key", "", "API key for authentication (overrides cached credentials)") cmd.PersistentFlags().StringP("output", "o", "text", "Output format: text or json") + cmd.PersistentFlags().Int("max-workflow-depth", 0, "Maximum workflow nesting depth (default: 10)") // Command groups for organized help output. cmd.AddGroup( diff --git a/packages/engine/internal/cli/status.go b/packages/engine/internal/cli/status.go index ddcf964..6c9d76d 100644 --- a/packages/engine/internal/cli/status.go +++ b/packages/engine/internal/cli/status.go @@ -3,6 +3,7 @@ package cli import ( "encoding/json" "fmt" + "strings" "time" "github.com/dvflw/mantle/internal/config" @@ -99,12 +100,17 @@ func newStatusCommand() *cobra.Command { fmt.Fprintf(cmd.OutOrStdout(), " %s: %d\n", stepStatus, count) } - // Query for child executions. + // Query for full descendant execution tree using recursive CTE. childRows, err := database.QueryContext(cmd.Context(), - `SELECT id, workflow_name, workflow_version, status - FROM workflow_executions - WHERE parent_execution_id = $1 - ORDER BY started_at`, execID, + `WITH RECURSIVE tree AS ( + SELECT id, workflow_name, workflow_version, status, 1 as depth + FROM workflow_executions WHERE parent_execution_id = $1 + UNION ALL + SELECT e.id, e.workflow_name, e.workflow_version, e.status, t.depth + 1 + FROM workflow_executions e + JOIN tree t ON e.parent_execution_id = t.id + ) + SELECT id, workflow_name, workflow_version, status, depth FROM tree ORDER BY depth, id`, execID, ) if err == nil { defer childRows.Close() @@ -114,22 +120,24 @@ func newStatusCommand() *cobra.Command { Workflow string Version int Status string + Depth int } var children []childExec for childRows.Next() { var c childExec - if err := childRows.Scan(&c.ID, &c.Workflow, &c.Version, &c.Status); err == nil { + if err := childRows.Scan(&c.ID, &c.Workflow, &c.Version, &c.Status, &c.Depth); err == nil { children = append(children, c) } } if len(children) > 0 { fmt.Fprintln(cmd.OutOrStdout()) - fmt.Fprintln(cmd.OutOrStdout(), "Children:") + fmt.Fprintln(cmd.OutOrStdout(), "Execution Tree:") for _, c := range children { icon := statusIcon(c.Status) - fmt.Fprintf(cmd.OutOrStdout(), " %s %-20s v%-4d %s %s\n", - icon, c.Workflow, c.Version, c.Status, c.ID) + indent := strings.Repeat(" ", c.Depth) + fmt.Fprintf(cmd.OutOrStdout(), "%s%s %-20s v%-4d %s %s\n", + indent, icon, c.Workflow, c.Version, c.Status, c.ID) } } } diff --git a/packages/engine/internal/config/config.go b/packages/engine/internal/config/config.go index 7766dbb..a0af94f 100644 --- a/packages/engine/internal/config/config.go +++ b/packages/engine/internal/config/config.go @@ -271,6 +271,9 @@ func Load(cmd *cobra.Command) (*Config, error) { if f := cmd.Flags().Lookup("log-level"); f != nil { _ = v.BindPFlag("log.level", f) } + if f := cmd.Flags().Lookup("max-workflow-depth"); f != nil { + _ = v.BindPFlag("engine.max_workflow_depth", f) + } var cfg Config if err := v.Unmarshal(&cfg); err != nil { diff --git a/packages/engine/internal/engine/engine.go b/packages/engine/internal/engine/engine.go index 9843983..67f55e6 100644 --- a/packages/engine/internal/engine/engine.go +++ b/packages/engine/internal/engine/engine.go @@ -178,6 +178,16 @@ func (e *Engine) resumeExecution(ctx context.Context, execID string, workflowNam // Execute steps sequentially. for _, step := range wf.Steps { + // Check if execution was cancelled externally (e.g., mantle cancel). + var currentStatus string + if err := e.DB.QueryRowContext(ctx, "SELECT status FROM workflow_executions WHERE id = $1", execID).Scan(¤tStatus); err == nil { + if currentStatus == "cancelled" { + result.Status = "cancelled" + result.Error = "execution cancelled" + return result, nil + } + } + // Skip already-completed steps (checkpoint recovery). if _, done := completedSteps[step.Name]; done { result.Steps[step.Name] = StepResult{ @@ -825,6 +835,14 @@ func (e *Engine) createExecution(ctx context.Context, workflowName string, versi // updateExecutionStatus updates the status of a workflow execution. func (e *Engine) updateExecutionStatus(ctx context.Context, execID, status, errMsg string) error { + // Don't overwrite a cancelled status. + var existing string + if err := e.DB.QueryRowContext(ctx, "SELECT status FROM workflow_executions WHERE id = $1", execID).Scan(&existing); err == nil { + if existing == "cancelled" { + return nil // Already cancelled, don't overwrite. + } + } + var completedAt any if status == "completed" || status == "failed" || status == "cancelled" { completedAt = time.Now() diff --git a/packages/engine/internal/engine/workflow_connector_test.go b/packages/engine/internal/engine/workflow_connector_test.go index ed1fd64..1810c7e 100644 --- a/packages/engine/internal/engine/workflow_connector_test.go +++ b/packages/engine/internal/engine/workflow_connector_test.go @@ -2,12 +2,8 @@ package engine import ( "context" - "database/sql" "encoding/json" - "fmt" "testing" - - "github.com/dvflw/mantle/internal/workflow" ) // TestWorkflowConnector_BasicInvocation creates a parent and child workflow, @@ -449,18 +445,3 @@ func TestWorkflowConnector_OutputWrapping_Unit(t *testing.T) { t.Errorf("fail-step error = %v, want %q", failStep["error"], "something broke") } } - -// applyWorkflowVersioned is a helper that stores a workflow and returns both the version and any error. -// Used in tests that need to control version behavior. -func applyWorkflowVersioned(t *testing.T, database *sql.DB, yamlContent []byte) (int, error) { - t.Helper() - result, err := workflow.ParseBytes(yamlContent) - if err != nil { - return 0, fmt.Errorf("ParseBytes: %w", err) - } - version, err := workflow.Save(context.Background(), database, result, yamlContent) - if err != nil { - return 0, fmt.Errorf("Save: %w", err) - } - return version, nil -} From 04bded295f03824bd2a228fd5c9c3bb005e4c1e9 Mon Sep 17 00:00:00 2001 From: Michael McNees Date: Fri, 27 Mar 2026 11:29:52 -0400 Subject: [PATCH 08/11] fix: preorder tree display, audit before commit, deeper cancellation checks (PR review) - status.go: Use text path in recursive CTE for correct preorder traversal - cancel.go: Emit audit events inside transaction via EmitTx before commit - engine.go: Check cancellation status before each retry attempt - engine.go: Guard updateStep against overwriting cancelled status - workflow_connector_test.go: Assert specific depth error substring Co-Authored-By: Claude Opus 4.6 (1M context) --- packages/engine/internal/cli/cancel.go | 17 +++++++++-------- packages/engine/internal/cli/status.go | 12 ++++++++---- packages/engine/internal/engine/engine.go | 9 ++++++++- .../internal/engine/workflow_connector_test.go | 4 ++++ 4 files changed, 29 insertions(+), 13 deletions(-) diff --git a/packages/engine/internal/cli/cancel.go b/packages/engine/internal/cli/cancel.go index b5654f3..f6e2ab1 100644 --- a/packages/engine/internal/cli/cancel.go +++ b/packages/engine/internal/cli/cancel.go @@ -97,19 +97,20 @@ func newCancelCommand() *cobra.Command { return fmt.Errorf("cancelling step executions: %w", err) } - if err := tx.Commit(); err != nil { - return fmt.Errorf("committing cancellation: %w", err) - } - - // Emit audit events for each cancelled execution. - auditor := &audit.PostgresEmitter{DB: database} + // Emit audit events inside the transaction so they commit atomically. for _, id := range cancelledIDs { - _ = auditor.Emit(cmd.Context(), audit.Event{ + if err := audit.EmitTx(cmd.Context(), tx, audit.Event{ Timestamp: time.Now(), Actor: "cli", Action: audit.ActionExecutionCancelled, Resource: audit.Resource{Type: "workflow_execution", ID: id}, - }) + }); err != nil { + return fmt.Errorf("emitting audit event for %s: %w", id, err) + } + } + + if err := tx.Commit(); err != nil { + return fmt.Errorf("committing cancellation: %w", err) } fmt.Fprintf(cmd.OutOrStdout(), "Cancelled execution %s (%d executions affected)\n", execID, len(cancelledIDs)) diff --git a/packages/engine/internal/cli/status.go b/packages/engine/internal/cli/status.go index 6c9d76d..92c1d2f 100644 --- a/packages/engine/internal/cli/status.go +++ b/packages/engine/internal/cli/status.go @@ -101,16 +101,19 @@ func newStatusCommand() *cobra.Command { } // Query for full descendant execution tree using recursive CTE. + // Build a text path for correct preorder traversal ordering of nested trees. childRows, err := database.QueryContext(cmd.Context(), `WITH RECURSIVE tree AS ( - SELECT id, workflow_name, workflow_version, status, 1 as depth + SELECT id, workflow_name, workflow_version, status, 1 as depth, + id::text as path FROM workflow_executions WHERE parent_execution_id = $1 UNION ALL - SELECT e.id, e.workflow_name, e.workflow_version, e.status, t.depth + 1 + SELECT e.id, e.workflow_name, e.workflow_version, e.status, t.depth + 1, + t.path || '/' || e.id::text FROM workflow_executions e JOIN tree t ON e.parent_execution_id = t.id ) - SELECT id, workflow_name, workflow_version, status, depth FROM tree ORDER BY depth, id`, execID, + SELECT id, workflow_name, workflow_version, status, depth, path FROM tree ORDER BY path`, execID, ) if err == nil { defer childRows.Close() @@ -121,11 +124,12 @@ func newStatusCommand() *cobra.Command { Version int Status string Depth int + Path string } var children []childExec for childRows.Next() { var c childExec - if err := childRows.Scan(&c.ID, &c.Workflow, &c.Version, &c.Status, &c.Depth); err == nil { + if err := childRows.Scan(&c.ID, &c.Workflow, &c.Version, &c.Status, &c.Depth, &c.Path); err == nil { children = append(children, c) } } diff --git a/packages/engine/internal/engine/engine.go b/packages/engine/internal/engine/engine.go index 59c5d5a..5438045 100644 --- a/packages/engine/internal/engine/engine.go +++ b/packages/engine/internal/engine/engine.go @@ -622,6 +622,12 @@ func (e *Engine) executeStepLogic(ctx context.Context, execID string, step workf var lastErr error for attempt := 1; attempt <= maxAttempts; attempt++ { + // Check if the execution has been cancelled before each retry attempt. + var execStatus string + if err := e.DB.QueryRowContext(ctx, "SELECT status FROM workflow_executions WHERE id = $1", execID).Scan(&execStatus); err == nil && execStatus == "cancelled" { + return nil, fmt.Errorf("execution cancelled") + } + // Create a fresh artifacts scratch dir per attempt. if len(step.Artifacts) > 0 && e.Storage != nil { if artifactsDir != "" { @@ -1055,9 +1061,10 @@ func (e *Engine) updateStep(ctx context.Context, execID, stepName, status string completedAt = time.Now() } + // Don't overwrite a cancelled step. _, err = e.DB.ExecContext(ctx, `UPDATE step_executions SET status = $1, output = $2, error = $3, completed_at = $4, continue_on_error = $5, updated_at = NOW() - WHERE execution_id = $6 AND step_name = $7 AND attempt = 1`, + WHERE execution_id = $6 AND step_name = $7 AND attempt = 1 AND status != 'cancelled'`, status, outputJSON, errorVal, completedAt, continueOnError, execID, stepName, ) if err != nil { diff --git a/packages/engine/internal/engine/workflow_connector_test.go b/packages/engine/internal/engine/workflow_connector_test.go index 8b04225..5e5d518 100644 --- a/packages/engine/internal/engine/workflow_connector_test.go +++ b/packages/engine/internal/engine/workflow_connector_test.go @@ -3,6 +3,7 @@ package engine import ( "context" "encoding/json" + "strings" "testing" ) @@ -227,6 +228,9 @@ steps: if callDeep.Error == "" { t.Error("call-deep error should be non-empty") } + if !strings.Contains(callDeep.Error, "depth") { + t.Errorf("expected depth limit error, got: %s", callDeep.Error) + } t.Logf("depth limit error: %s", callDeep.Error) } From 8655bcd407ba4f09d80d9119897d2419bcc03036 Mon Sep 17 00:00:00 2001 From: Michael McNees Date: Fri, 27 Mar 2026 11:43:07 -0400 Subject: [PATCH 09/11] fix: safe assertions, tx-consistent reads, atomic status update, team scoping (PR review) - Safe type assertion on execution_id in workflow connector test - Use tx instead of database for existence check in cancel command - Add team_id filter to cancellation check in retry loop - Replace SELECT-then-UPDATE with atomic UPDATE in updateExecutionStatus - Set team context for depth limit test to match DB insert - Clean up verbose depth-check comments Co-Authored-By: Claude Opus 4.6 (1M context) --- packages/engine/internal/cli/cancel.go | 4 +- packages/engine/internal/engine/engine.go | 26 ++++++------- .../engine/workflow_connector_test.go | 39 +++++++------------ 3 files changed, 28 insertions(+), 41 deletions(-) diff --git a/packages/engine/internal/cli/cancel.go b/packages/engine/internal/cli/cancel.go index f6e2ab1..d87815b 100644 --- a/packages/engine/internal/cli/cancel.go +++ b/packages/engine/internal/cli/cancel.go @@ -69,9 +69,9 @@ func newCancelCommand() *cobra.Command { } if len(cancelledIDs) == 0 { - // Check if execution exists at all. + // Check if execution exists at all (use tx for consistent read). var status string - err := database.QueryRowContext(cmd.Context(), + err := tx.QueryRowContext(cmd.Context(), `SELECT status FROM workflow_executions WHERE id = $1`, execID, ).Scan(&status) if err != nil { diff --git a/packages/engine/internal/engine/engine.go b/packages/engine/internal/engine/engine.go index 5438045..ca1a1f8 100644 --- a/packages/engine/internal/engine/engine.go +++ b/packages/engine/internal/engine/engine.go @@ -241,7 +241,7 @@ func (e *Engine) resumeExecution(ctx context.Context, execID string, workflowNam for _, step := range wf.Steps { // Check if execution was cancelled externally (e.g., mantle cancel). var currentStatus string - if err := e.DB.QueryRowContext(ctx, "SELECT status FROM workflow_executions WHERE id = $1", execID).Scan(¤tStatus); err == nil { + if err := e.DB.QueryRowContext(ctx, "SELECT status FROM workflow_executions WHERE id = $1 AND team_id = $2", execID, sc.TeamID).Scan(¤tStatus); err == nil { if currentStatus == "cancelled" { result.Status = "cancelled" result.Error = "execution cancelled" @@ -624,7 +624,7 @@ func (e *Engine) executeStepLogic(ctx context.Context, execID string, step workf for attempt := 1; attempt <= maxAttempts; attempt++ { // Check if the execution has been cancelled before each retry attempt. var execStatus string - if err := e.DB.QueryRowContext(ctx, "SELECT status FROM workflow_executions WHERE id = $1", execID).Scan(&execStatus); err == nil && execStatus == "cancelled" { + if err := e.DB.QueryRowContext(ctx, "SELECT status FROM workflow_executions WHERE id = $1 AND team_id = $2", execID, sc.TeamID).Scan(&execStatus); err == nil && execStatus == "cancelled" { return nil, fmt.Errorf("execution cancelled") } @@ -994,29 +994,29 @@ func (e *Engine) createExecutionTx(ctx context.Context, tx *sql.Tx, workflowName return id, nil } -// updateExecutionStatus updates the status of a workflow execution. +// updateExecutionStatus atomically updates the status of a workflow execution. +// If the execution is already cancelled, the update is a no-op (returns nil). func (e *Engine) updateExecutionStatus(ctx context.Context, execID, status, errMsg string) error { - // Don't overwrite a cancelled status. - var existing string - if err := e.DB.QueryRowContext(ctx, "SELECT status FROM workflow_executions WHERE id = $1", execID).Scan(&existing); err == nil { - if existing == "cancelled" { - return nil // Already cancelled, don't overwrite. - } - } - var completedAt any if status == "completed" || status == "failed" || status == "cancelled" || status == "timed_out" { completedAt = time.Now() } teamID := auth.TeamIDFromContext(ctx) - _, err := e.DB.ExecContext(ctx, - `UPDATE workflow_executions SET status = $1, completed_at = $2, updated_at = NOW() WHERE id = $3 AND team_id = $4`, + result, err := e.DB.ExecContext(ctx, + `UPDATE workflow_executions SET status = $1, completed_at = $2, updated_at = NOW() + WHERE id = $3 AND team_id = $4 AND status != 'cancelled'`, status, completedAt, execID, teamID, ) if err != nil { return fmt.Errorf("updating execution %s status to %s: %w", execID, status, err) } + + // If no rows were affected, the execution was already cancelled (or doesn't exist). + // This is not an error — the cancellation takes precedence. + if rows, _ := result.RowsAffected(); rows == 0 { + return nil + } return nil } diff --git a/packages/engine/internal/engine/workflow_connector_test.go b/packages/engine/internal/engine/workflow_connector_test.go index 5e5d518..f63a398 100644 --- a/packages/engine/internal/engine/workflow_connector_test.go +++ b/packages/engine/internal/engine/workflow_connector_test.go @@ -5,6 +5,8 @@ import ( "encoding/json" "strings" "testing" + + "github.com/dvflw/mantle/internal/auth" ) // TestWorkflowConnector_BasicInvocation creates a parent and child workflow, @@ -99,7 +101,10 @@ steps: } // Verify child execution exists in the database with correct linkage. - childExecID := callChild.Output["execution_id"].(string) + childExecID, ok := callChild.Output["execution_id"].(string) + if !ok { + t.Fatalf("expected execution_id string in child output, got %T", callChild.Output["execution_id"]) + } var parentID, parentStepName string var depth int err = database.QueryRowContext(ctx, @@ -173,45 +178,27 @@ steps: t.Fatalf("resumeExecution() error: %v", err) } - // The parent should fail because the child would be at depth 1, - // and the parent itself is at depth 0, so child depth = 1 which exceeds max depth 1. - // Actually: childDepth = parentDepth(0) + 1 = 1, maxDepth = 1, - // and the check is childDepth > maxDepth, so 1 > 1 is false. - // We need maxDepth = 0 for it to fail, but 0 means "use default 10". - // Let's set it differently. - // Re-check: the connector sets maxDepth=10 when MaxWorkflowDepth <= 0. - // When MaxWorkflowDepth = 1, childDepth = 1, and 1 > 1 is false. - // So this should succeed. Let me set it to trigger failure properly. - - // Actually, we should verify what happens. The depth check in the connector is: - // childDepth := parentDepth + 1 - // if childDepth > maxDepth { error } - // With MaxWorkflowDepth=1, parentDepth=0, childDepth=1, 1>1=false, so it passes. - // To trigger the error, we need to simulate the parent being at depth >= maxDepth. - // Let's just verify it works at depth 1, then test the error case with a - // parent already at depth 1 and max=1. - - // This test should succeed since parent is at depth 0 and max is 1. + // With maxDepth=1 and parentDepth=0, childDepth=1 which does not exceed max, so this succeeds. if result.Status != "completed" { - // If it failed for depth reasons, that's fine for the test intent. - // But let's be precise. t.Logf("result status=%s, error=%s", result.Status, result.Error) } // Now test the actual depth limit: create an execution at depth 1 and max=1. + teamID := "00000000-0000-0000-0000-000000000001" var deepExecID string err = database.QueryRowContext(ctx, `INSERT INTO workflow_executions (workflow_name, workflow_version, status, started_at, team_id, depth) VALUES ($1, $2, 'pending', NOW(), $3, $4) RETURNING id`, - "deep-parent", parentVersion, "00000000-0000-0000-0000-000000000001", 1, + "deep-parent", parentVersion, teamID, 1, ).Scan(&deepExecID) if err != nil { t.Fatalf("inserting deep execution: %v", err) } - ctx2 := WithExecutionID(context.Background(), deepExecID) + ctx2 := auth.WithUser(context.Background(), &auth.User{TeamID: teamID}) + ctx2 = WithExecutionID(ctx2, deepExecID) result2, err := eng.resumeExecution(ctx2, deepExecID, "deep-parent", parentVersion, nil) if err != nil { t.Fatalf("resumeExecution() error: %v", err) @@ -352,8 +339,8 @@ steps: } } -// TestWorkflowConnector_DepthCheck_Unit tests the depth checking logic directly -// without a full workflow execution. +// TestWorkflowConnector_DepthCheck_Unit tests the expected depth-limit algorithm +// (childDepth > maxDepth) in isolation, not the actual connector method. func TestWorkflowConnector_DepthCheck_Unit(t *testing.T) { tests := []struct { name string From 165acd55cd0ebd8c360b26e14207f6c9a1e3d27e Mon Sep 17 00:00:00 2001 From: "claude[bot]" <41898282+claude[bot]@users.noreply.github.com> Date: Fri, 27 Mar 2026 19:06:06 +0000 Subject: [PATCH 10/11] fix: cancel.go CTE dedup, ErrNoRows distinction, step audit events; test assertion upgrade - Extract repeated WITH RECURSIVE children CTE into childrenCTE constant - Distinguish sql.ErrNoRows from other DB errors in zero-update path - Collect cancelled step IDs via RETURNING and emit audit events for each - Replace t.Logf with t.Fatalf in depth-limit test so regressions fail the test Co-authored-by: Michael McNees --- packages/engine/internal/cli/cancel.go | 59 ++++++++++++++----- .../engine/workflow_connector_test.go | 2 +- 2 files changed, 44 insertions(+), 17 deletions(-) diff --git a/packages/engine/internal/cli/cancel.go b/packages/engine/internal/cli/cancel.go index d87815b..b98b00e 100644 --- a/packages/engine/internal/cli/cancel.go +++ b/packages/engine/internal/cli/cancel.go @@ -1,6 +1,7 @@ package cli import ( + "database/sql" "fmt" "time" @@ -10,6 +11,14 @@ import ( "github.com/spf13/cobra" ) +// childrenCTE is a recursive CTE that selects an execution and all its descendants. +const childrenCTE = `WITH RECURSIVE children AS ( + SELECT id FROM workflow_executions WHERE id = $1 + UNION ALL + SELECT e.id FROM workflow_executions e + JOIN children c ON e.parent_execution_id = c.id + )` + func newCancelCommand() *cobra.Command { return &cobra.Command{ Use: "cancel ", @@ -39,12 +48,7 @@ func newCancelCommand() *cobra.Command { // Cancel the execution and all child executions recursively. // Use RETURNING id to capture which executions were cancelled. rows, err := tx.QueryContext(cmd.Context(), - `WITH RECURSIVE children AS ( - SELECT id FROM workflow_executions WHERE id = $1 - UNION ALL - SELECT e.id FROM workflow_executions e - JOIN children c ON e.parent_execution_id = c.id - ) + childrenCTE+` UPDATE workflow_executions SET status = 'cancelled', completed_at = NOW(), updated_at = NOW() WHERE id IN (SELECT id FROM children) AND status IN ('pending', 'running', 'queued') RETURNING id`, @@ -74,28 +78,41 @@ func newCancelCommand() *cobra.Command { err := tx.QueryRowContext(cmd.Context(), `SELECT status FROM workflow_executions WHERE id = $1`, execID, ).Scan(&status) - if err != nil { + if err == sql.ErrNoRows { return fmt.Errorf("execution %q not found", execID) } + if err != nil { + return fmt.Errorf("checking execution status: %w", err) + } fmt.Fprintf(cmd.OutOrStdout(), "Execution %s is already %s\n", execID, status) return nil } - // Also mark any running/pending steps in the tree as cancelled. - _, err = tx.ExecContext(cmd.Context(), - `WITH RECURSIVE children AS ( - SELECT id FROM workflow_executions WHERE id = $1 - UNION ALL - SELECT e.id FROM workflow_executions e - JOIN children c ON e.parent_execution_id = c.id - ) + // Also mark any running/pending steps in the tree as cancelled, + // collecting their IDs for audit events. + stepRows, err := tx.QueryContext(cmd.Context(), + childrenCTE+` UPDATE step_executions SET status = 'cancelled', completed_at = NOW(), updated_at = NOW() - WHERE execution_id IN (SELECT id FROM children) AND status IN ('pending', 'running')`, + WHERE execution_id IN (SELECT id FROM children) AND status IN ('pending', 'running') + RETURNING id`, execID, ) if err != nil { return fmt.Errorf("cancelling step executions: %w", err) } + var cancelledStepIDs []string + for stepRows.Next() { + var id string + if err := stepRows.Scan(&id); err != nil { + stepRows.Close() + return fmt.Errorf("scanning cancelled step id: %w", err) + } + cancelledStepIDs = append(cancelledStepIDs, id) + } + stepRows.Close() + if err := stepRows.Err(); err != nil { + return fmt.Errorf("iterating cancelled step ids: %w", err) + } // Emit audit events inside the transaction so they commit atomically. for _, id := range cancelledIDs { @@ -108,6 +125,16 @@ func newCancelCommand() *cobra.Command { return fmt.Errorf("emitting audit event for %s: %w", id, err) } } + for _, id := range cancelledStepIDs { + if err := audit.EmitTx(cmd.Context(), tx, audit.Event{ + Timestamp: time.Now(), + Actor: "cli", + Action: audit.ActionExecutionCancelled, + Resource: audit.Resource{Type: "step_execution", ID: id}, + }); err != nil { + return fmt.Errorf("emitting audit event for step %s: %w", id, err) + } + } if err := tx.Commit(); err != nil { return fmt.Errorf("committing cancellation: %w", err) diff --git a/packages/engine/internal/engine/workflow_connector_test.go b/packages/engine/internal/engine/workflow_connector_test.go index f63a398..ef2d3d2 100644 --- a/packages/engine/internal/engine/workflow_connector_test.go +++ b/packages/engine/internal/engine/workflow_connector_test.go @@ -180,7 +180,7 @@ steps: // With maxDepth=1 and parentDepth=0, childDepth=1 which does not exceed max, so this succeeds. if result.Status != "completed" { - t.Logf("result status=%s, error=%s", result.Status, result.Error) + t.Fatalf("expected completed status for non-exceeding depth, got status=%q error=%q", result.Status, result.Error) } // Now test the actual depth limit: create an execution at depth 1 and max=1. From 1c08927066991925e65838e5a28c48b25124d7f1 Mon Sep 17 00:00:00 2001 From: "claude[bot]" <41898282+claude[bot]@users.noreply.github.com> Date: Sat, 28 Mar 2026 01:38:03 +0000 Subject: [PATCH 11/11] fix: propagate DB errors in cancellation checks, remove unimplemented token_budget - engine.go: don't silently ignore DB errors in cancellation status checks; distinguish sql.ErrNoRows from other errors and return them instead of silently proceeding with unknown cancellation state - workflow_connector.go: remove token_budget from docstring (not implemented) - connectors.md: remove undocumented token_budget param row; add 'cel' language specifier to CEL path code fence Co-authored-by: Michael McNees --- packages/engine/internal/engine/engine.go | 22 ++++++++++++++----- .../internal/engine/workflow_connector.go | 1 - .../docs/workflow-reference/connectors.md | 3 +-- 3 files changed, 17 insertions(+), 9 deletions(-) diff --git a/packages/engine/internal/engine/engine.go b/packages/engine/internal/engine/engine.go index ca1a1f8..6bdea3e 100644 --- a/packages/engine/internal/engine/engine.go +++ b/packages/engine/internal/engine/engine.go @@ -241,12 +241,16 @@ func (e *Engine) resumeExecution(ctx context.Context, execID string, workflowNam for _, step := range wf.Steps { // Check if execution was cancelled externally (e.g., mantle cancel). var currentStatus string - if err := e.DB.QueryRowContext(ctx, "SELECT status FROM workflow_executions WHERE id = $1 AND team_id = $2", execID, sc.TeamID).Scan(¤tStatus); err == nil { - if currentStatus == "cancelled" { - result.Status = "cancelled" - result.Error = "execution cancelled" - return result, nil + if err := e.DB.QueryRowContext(ctx, "SELECT status FROM workflow_executions WHERE id = $1 AND team_id = $2", execID, sc.TeamID).Scan(¤tStatus); err != nil { + if errors.Is(err, sql.ErrNoRows) { + return nil, fmt.Errorf("execution %s not found", execID) } + return nil, fmt.Errorf("checking cancellation status for %s: %w", execID, err) + } + if currentStatus == "cancelled" { + result.Status = "cancelled" + result.Error = "execution cancelled" + return result, nil } // Skip already-completed steps (checkpoint recovery). @@ -624,7 +628,13 @@ func (e *Engine) executeStepLogic(ctx context.Context, execID string, step workf for attempt := 1; attempt <= maxAttempts; attempt++ { // Check if the execution has been cancelled before each retry attempt. var execStatus string - if err := e.DB.QueryRowContext(ctx, "SELECT status FROM workflow_executions WHERE id = $1 AND team_id = $2", execID, sc.TeamID).Scan(&execStatus); err == nil && execStatus == "cancelled" { + if err := e.DB.QueryRowContext(ctx, "SELECT status FROM workflow_executions WHERE id = $1 AND team_id = $2", execID, sc.TeamID).Scan(&execStatus); err != nil { + if errors.Is(err, sql.ErrNoRows) { + return nil, fmt.Errorf("execution %s not found", execID) + } + return nil, fmt.Errorf("checking cancellation status for %s: %w", execID, err) + } + if execStatus == "cancelled" { return nil, fmt.Errorf("execution cancelled") } diff --git a/packages/engine/internal/engine/workflow_connector.go b/packages/engine/internal/engine/workflow_connector.go index b4cde6e..8186bf1 100644 --- a/packages/engine/internal/engine/workflow_connector.go +++ b/packages/engine/internal/engine/workflow_connector.go @@ -25,7 +25,6 @@ type WorkflowConnector struct { // - workflow (string, required): name of the child workflow // - version (int, optional): version to run; 0 or omitted = latest // - inputs (map[string]any, optional): input parameters for the child -// - token_budget (int64, optional): token budget override for the child func (wc *WorkflowConnector) Execute(ctx context.Context, params map[string]any) (map[string]any, error) { // Extract required workflow name. workflowName, ok := params["workflow"].(string) diff --git a/packages/site/src/content/docs/workflow-reference/connectors.md b/packages/site/src/content/docs/workflow-reference/connectors.md index 6cb45ac..d08393e 100644 --- a/packages/site/src/content/docs/workflow-reference/connectors.md +++ b/packages/site/src/content/docs/workflow-reference/connectors.md @@ -606,7 +606,6 @@ Invokes another workflow as a child execution. The child workflow runs synchrono | `workflow` | string | Yes | Name of the child workflow to execute. | | `version` | integer | No | Specific version to run. Omit to use the latest applied version. | | `inputs` | map | No | Input parameters to pass to the child workflow. | -| `token_budget` | integer | No | Token budget override for the child execution. | **Output:** @@ -620,7 +619,7 @@ Invokes another workflow as a child execution. The child workflow runs synchrono Child step outputs are nested under the parent step's output: -``` +```cel steps['my-step'].output.steps['child-step'].output.field ```