diff --git a/packages/engine/internal/audit/audit.go b/packages/engine/internal/audit/audit.go
index 40383be..0607997 100644
--- a/packages/engine/internal/audit/audit.go
+++ b/packages/engine/internal/audit/audit.go
@@ -17,8 +17,9 @@ const (
 	ActionStepSkipped           Action = "step.skipped"
 	ActionStepContinuedOnError  Action = "step.continued_on_error"
 	ActionExecutionCancelled    Action = "execution.cancelled"
+	ActionChildWorkflowExecuted Action = "workflow.child_executed"
 	ActionExecutionRetried      Action = "execution.retried"
-	ActionArtifactPersisted  Action = "artifact.persisted"
+	ActionArtifactPersisted     Action = "artifact.persisted"
 
 	// Admin operations.
 	ActionUserCreated       Action = "user.created"
diff --git a/packages/engine/internal/cli/cancel.go b/packages/engine/internal/cli/cancel.go
index 80045d3..b98b00e 100644
--- a/packages/engine/internal/cli/cancel.go
+++ b/packages/engine/internal/cli/cancel.go
@@ -1,13 +1,24 @@
 package cli
 
 import (
+	"database/sql"
 	"fmt"
+	"time"
 
+	"github.com/dvflw/mantle/internal/audit"
 	"github.com/dvflw/mantle/internal/config"
 	"github.com/dvflw/mantle/internal/db"
 	"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 <execution-id>",
@@ -28,44 +39,108 @@ func newCancelCommand() *cobra.Command {
 			}
 			defer database.Close()
 
-			// Only cancel if currently pending, running, or queued.
-			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', 'queued')`,
+			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.
+			// Use RETURNING id to capture which executions were cancelled.
+			rows, err := tx.QueryContext(cmd.Context(),
+				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`,
 				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 {
-				// Check if execution exists at all.
+			if len(cancelledIDs) == 0 {
+				// 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 {
+				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 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')`,
+			// 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')
+				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 {
+				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)
+				}
+			}
+			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)
+			}
 
-			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/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/root.go b/packages/engine/internal/cli/root.go
index d435528..be773ce 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/run.go b/packages/engine/internal/cli/run.go
index 6e3f32e..21f82de 100644
--- a/packages/engine/internal/cli/run.go
+++ b/packages/engine/internal/cli/run.go
@@ -64,8 +64,10 @@ func newRunCommand() *cobra.Command {
 			if err != nil {
 				return fmt.Errorf("creating engine: %w", err)
 			}
+			eng.MaxWorkflowDepth = cfg.Engine.MaxWorkflowDepth
 			eng.MaxConcurrentExecutionsPerTeam = cfg.Engine.MaxConcurrentExecutionsPerTeam
 			eng.CEL.SetConfigEnv(cfg.Env)
+			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 8face8e..7fdb2ff 100644
--- a/packages/engine/internal/cli/serve.go
+++ b/packages/engine/internal/cli/serve.go
@@ -51,7 +51,9 @@ func newServeCommand() *cobra.Command {
 			}
 			eng.CEL.SetConfigEnv(cfg.Env)
 			eng.MaxToolRoundsLimit = cfg.Engine.MaxToolRoundsLimit
+			eng.MaxWorkflowDepth = cfg.Engine.MaxWorkflowDepth
 			eng.MaxConcurrentExecutionsPerTeam = cfg.Engine.MaxConcurrentExecutionsPerTeam
+			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/cli/status.go b/packages/engine/internal/cli/status.go
index ea90783..92c1d2f 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,6 +100,52 @@ func newStatusCommand() *cobra.Command {
 				fmt.Fprintf(cmd.OutOrStdout(), "  %s: %d\n", stepStatus, count)
 			}
 
+			// 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,
+					       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,
+					       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, path FROM tree ORDER BY path`, execID,
+			)
+			if err == nil {
+				defer childRows.Close()
+
+				type childExec struct {
+					ID       string
+					Workflow string
+					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, &c.Path); err == nil {
+						children = append(children, c)
+					}
+				}
+
+				if len(children) > 0 {
+					fmt.Fprintln(cmd.OutOrStdout())
+					fmt.Fprintln(cmd.OutOrStdout(), "Execution Tree:")
+					for _, c := range children {
+						icon := statusIcon(c.Status)
+						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)
+					}
+				}
+			}
+
 			return nil
 		},
 	}
diff --git a/packages/engine/internal/config/config.go b/packages/engine/internal/config/config.go
index 3b03a57..f566176 100644
--- a/packages/engine/internal/config/config.go
+++ b/packages/engine/internal/config/config.go
@@ -135,7 +135,8 @@ 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
-	MaxConcurrentExecutionsPerTeam int        `mapstructure:"max_concurrent_executions_per_team"`
+	MaxWorkflowDepth               int           `mapstructure:"max_workflow_depth"`
+	MaxConcurrentExecutionsPerTeam int           `mapstructure:"max_concurrent_executions_per_team"`
 	Budget                      BudgetConfig  `mapstructure:"budget"`
 }
 
@@ -176,6 +177,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)
@@ -256,6 +258,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")
 	_ = v.BindEnv("engine.max_concurrent_executions_per_team", "MANTLE_ENGINE_MAX_CONCURRENT_EXECUTIONS_PER_TEAM")
 
 	// Budget env var bindings
@@ -274,6 +277,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/db/migrations/017_workflow_composition.sql b/packages/engine/internal/db/migrations/017_workflow_composition.sql
new file mode 100644
index 0000000..20e2641
--- /dev/null
+++ b/packages/engine/internal/db/migrations/017_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;
diff --git a/packages/engine/internal/engine/engine.go b/packages/engine/internal/engine/engine.go
index b85f1c9..6bdea3e 100644
--- a/packages/engine/internal/engine/engine.go
+++ b/packages/engine/internal/engine/engine.go
@@ -31,6 +31,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)
 	MaxConcurrentExecutionsPerTeam int              // 0 = unlimited; per-team concurrency cap
 	BudgetChecker                  *budget.Checker  // nil = budget enforcement disabled
 	BudgetStore                    *budget.Store    // nil = token usage recording disabled
@@ -238,6 +239,20 @@ func (e *Engine) resumeExecution(ctx context.Context, execID string, workflowNam
 	// Execute steps sequentially, tracking failure state for hooks.
 	var failedStepName string
 	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(&currentStatus); 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).
 		if _, done := completedSteps[step.Name]; done {
 			result.Steps[step.Name] = StepResult{
@@ -611,6 +626,18 @@ 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 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")
+		}
+
 		// Create a fresh artifacts scratch dir per attempt.
 		if len(step.Artifacts) > 0 && e.Storage != nil {
 			if artifactsDir != "" {
@@ -977,7 +1004,8 @@ 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 {
 	var completedAt any
 	if status == "completed" || status == "failed" || status == "cancelled" || status == "timed_out" {
@@ -985,13 +1013,20 @@ func (e *Engine) updateExecutionStatus(ctx context.Context, execID, status, errM
 	}
 
 	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
 }
 
@@ -1036,9 +1071,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.go b/packages/engine/internal/engine/workflow_connector.go
new file mode 100644
index 0000000..8186bf1
--- /dev/null
+++ b/packages/engine/internal/engine/workflow_connector.go
@@ -0,0 +1,187 @@
+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
+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})
+}
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..ef2d3d2
--- /dev/null
+++ b/packages/engine/internal/engine/workflow_connector_test.go
@@ -0,0 +1,438 @@
+package engine
+
+import (
+	"context"
+	"encoding/json"
+	"strings"
+	"testing"
+
+	"github.com/dvflw/mantle/internal/auth"
+)
+
+// 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, "pending")
+	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, 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,
+		`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, "pending")
+	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)
+	}
+
+	// With maxDepth=1 and parentDepth=0, childDepth=1 which does not exceed max, so this succeeds.
+	if result.Status != "completed" {
+		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.
+	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, teamID, 1,
+	).Scan(&deepExecID)
+	if err != nil {
+		t.Fatalf("inserting deep execution: %v", err)
+	}
+
+	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)
+	}
+
+	// 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")
+	}
+	if !strings.Contains(callDeep.Error, "depth") {
+		t.Errorf("expected depth limit error, got: %s", callDeep.Error)
+	}
+	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, "pending")
+	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 expected depth-limit algorithm
+// (childDepth > maxDepth) in isolation, not the actual connector method.
+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")
+	}
+}
diff --git a/packages/site/src/content/docs/workflow-reference/connectors.md b/packages/site/src/content/docs/workflow-reference/connectors.md
index 99f8ac5..d08393e 100644
--- a/packages/site/src/content/docs/workflow-reference/connectors.md
+++ b/packages/site/src/content/docs/workflow-reference/connectors.md
@@ -595,6 +595,83 @@ 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. |
+
+**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:
+
+```cel
+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.