feat: workflow composition — workflow/run connector (#54)

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"

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,
 			)
 

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,17 +178,26 @@ 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 := ""
 		if si.StartedAt != nil && si.CompletedAt != nil {
 			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.

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 != "" {

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 {

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
 		},
 	}

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")

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;

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