Skip to content

Commit 13540d9

Browse files
michaelmcneesclaudegithub-actions[bot]
authored
feat: workflow composition — workflow/run connector (#54) (#120)
* 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) <noreply@anthropic.com> * 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) <noreply@anthropic.com> * 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) <noreply@anthropic.com> * 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) <noreply@anthropic.com> * 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) <noreply@anthropic.com> * 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) <noreply@anthropic.com> * 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) <noreply@anthropic.com> * 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) <noreply@anthropic.com> * 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) <noreply@anthropic.com> * 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 <michaelmcnees@users.noreply.github.com> * 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 <michaelmcnees@users.noreply.github.com> --------- Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com> Co-authored-by: claude[bot] <41898282+claude[bot]@users.noreply.github.com> Co-authored-by: Michael McNees <michaelmcnees@users.noreply.github.com>
1 parent e73029e commit 13540d9

13 files changed

Lines changed: 1031 additions & 45 deletions

File tree

‎packages/engine/internal/audit/audit.go‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -17,8 +17,9 @@ const (
1717
ActionStepSkipped Action = "step.skipped"
1818
ActionStepContinuedOnError Action = "step.continued_on_error"
1919
ActionExecutionCancelled Action = "execution.cancelled"
20+
ActionChildWorkflowExecuted Action = "workflow.child_executed"
2021
ActionExecutionRetried Action = "execution.retried"
21-
ActionArtifactPersisted Action = "artifact.persisted"
22+
ActionArtifactPersisted Action = "artifact.persisted"
2223

2324
// Admin operations.
2425
ActionUserCreated Action = "user.created"

‎packages/engine/internal/cli/cancel.go‎

Lines changed: 93 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -1,13 +1,24 @@
11
package cli
22

33
import (
4+
"database/sql"
45
"fmt"
6+
"time"
57

8+
"github.com/dvflw/mantle/internal/audit"
69
"github.com/dvflw/mantle/internal/config"
710
"github.com/dvflw/mantle/internal/db"
811
"github.com/spf13/cobra"
912
)
1013

14+
// childrenCTE is a recursive CTE that selects an execution and all its descendants.
15+
const childrenCTE = `WITH RECURSIVE children AS (
16+
SELECT id FROM workflow_executions WHERE id = $1
17+
UNION ALL
18+
SELECT e.id FROM workflow_executions e
19+
JOIN children c ON e.parent_execution_id = c.id
20+
)`
21+
1122
func newCancelCommand() *cobra.Command {
1223
return &cobra.Command{
1324
Use: "cancel <execution-id>",
@@ -28,44 +39,108 @@ func newCancelCommand() *cobra.Command {
2839
}
2940
defer database.Close()
3041

31-
// Only cancel if currently pending, running, or queued.
32-
result, err := database.ExecContext(cmd.Context(),
33-
`UPDATE workflow_executions
34-
SET status = 'cancelled', completed_at = NOW(), updated_at = NOW()
35-
WHERE id = $1 AND status IN ('pending', 'running', 'queued')`,
42+
tx, err := database.BeginTx(cmd.Context(), nil)
43+
if err != nil {
44+
return fmt.Errorf("starting transaction: %w", err)
45+
}
46+
defer tx.Rollback() //nolint:errcheck
47+
48+
// Cancel the execution and all child executions recursively.
49+
// Use RETURNING id to capture which executions were cancelled.
50+
rows, err := tx.QueryContext(cmd.Context(),
51+
childrenCTE+`
52+
UPDATE workflow_executions SET status = 'cancelled', completed_at = NOW(), updated_at = NOW()
53+
WHERE id IN (SELECT id FROM children) AND status IN ('pending', 'running', 'queued')
54+
RETURNING id`,
3655
execID,
3756
)
3857
if err != nil {
3958
return fmt.Errorf("cancelling execution: %w", err)
4059
}
4160

42-
rows, err := result.RowsAffected()
43-
if err != nil {
44-
return fmt.Errorf("checking result: %w", err)
61+
var cancelledIDs []string
62+
for rows.Next() {
63+
var id string
64+
if err := rows.Scan(&id); err != nil {
65+
rows.Close()
66+
return fmt.Errorf("scanning cancelled id: %w", err)
67+
}
68+
cancelledIDs = append(cancelledIDs, id)
69+
}
70+
rows.Close()
71+
if err := rows.Err(); err != nil {
72+
return fmt.Errorf("iterating cancelled ids: %w", err)
4573
}
4674

47-
if rows == 0 {
48-
// Check if execution exists at all.
75+
if len(cancelledIDs) == 0 {
76+
// Check if execution exists at all (use tx for consistent read).
4977
var status string
50-
err := database.QueryRowContext(cmd.Context(),
78+
err := tx.QueryRowContext(cmd.Context(),
5179
`SELECT status FROM workflow_executions WHERE id = $1`, execID,
5280
).Scan(&status)
53-
if err != nil {
81+
if err == sql.ErrNoRows {
5482
return fmt.Errorf("execution %q not found", execID)
5583
}
84+
if err != nil {
85+
return fmt.Errorf("checking execution status: %w", err)
86+
}
5687
fmt.Fprintf(cmd.OutOrStdout(), "Execution %s is already %s\n", execID, status)
5788
return nil
5889
}
5990

60-
// Also mark any running/pending steps as cancelled.
61-
database.ExecContext(cmd.Context(),
62-
`UPDATE step_executions
63-
SET status = 'cancelled', completed_at = NOW(), updated_at = NOW()
64-
WHERE execution_id = $1 AND status IN ('pending', 'running')`,
91+
// Also mark any running/pending steps in the tree as cancelled,
92+
// collecting their IDs for audit events.
93+
stepRows, err := tx.QueryContext(cmd.Context(),
94+
childrenCTE+`
95+
UPDATE step_executions SET status = 'cancelled', completed_at = NOW(), updated_at = NOW()
96+
WHERE execution_id IN (SELECT id FROM children) AND status IN ('pending', 'running')
97+
RETURNING id`,
6598
execID,
6699
)
100+
if err != nil {
101+
return fmt.Errorf("cancelling step executions: %w", err)
102+
}
103+
var cancelledStepIDs []string
104+
for stepRows.Next() {
105+
var id string
106+
if err := stepRows.Scan(&id); err != nil {
107+
stepRows.Close()
108+
return fmt.Errorf("scanning cancelled step id: %w", err)
109+
}
110+
cancelledStepIDs = append(cancelledStepIDs, id)
111+
}
112+
stepRows.Close()
113+
if err := stepRows.Err(); err != nil {
114+
return fmt.Errorf("iterating cancelled step ids: %w", err)
115+
}
116+
117+
// Emit audit events inside the transaction so they commit atomically.
118+
for _, id := range cancelledIDs {
119+
if err := audit.EmitTx(cmd.Context(), tx, audit.Event{
120+
Timestamp: time.Now(),
121+
Actor: "cli",
122+
Action: audit.ActionExecutionCancelled,
123+
Resource: audit.Resource{Type: "workflow_execution", ID: id},
124+
}); err != nil {
125+
return fmt.Errorf("emitting audit event for %s: %w", id, err)
126+
}
127+
}
128+
for _, id := range cancelledStepIDs {
129+
if err := audit.EmitTx(cmd.Context(), tx, audit.Event{
130+
Timestamp: time.Now(),
131+
Actor: "cli",
132+
Action: audit.ActionExecutionCancelled,
133+
Resource: audit.Resource{Type: "step_execution", ID: id},
134+
}); err != nil {
135+
return fmt.Errorf("emitting audit event for step %s: %w", id, err)
136+
}
137+
}
138+
139+
if err := tx.Commit(); err != nil {
140+
return fmt.Errorf("committing cancellation: %w", err)
141+
}
67142

68-
fmt.Fprintf(cmd.OutOrStdout(), "Cancelled execution %s\n", execID)
143+
fmt.Fprintf(cmd.OutOrStdout(), "Cancelled execution %s (%d executions affected)\n", execID, len(cancelledIDs))
69144
return nil
70145
},
71146
}

‎packages/engine/internal/cli/logs.go‎

Lines changed: 124 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
package cli
22

33
import (
4+
"database/sql"
45
"encoding/json"
56
"fmt"
67
"strconv"
@@ -13,6 +14,25 @@ import (
1314
"github.com/spf13/cobra"
1415
)
1516

17+
// subStep represents a sub-step (tool call) within a step execution.
18+
type subStep struct {
19+
StepName string `json:"step_name"`
20+
Status string `json:"status"`
21+
Started *time.Time `json:"started_at,omitempty"`
22+
Completed *time.Time `json:"completed_at,omitempty"`
23+
}
24+
25+
// stepInfo represents a top-level step execution.
26+
type stepInfo struct {
27+
ID string `json:"id"`
28+
Name string `json:"name"`
29+
Status string `json:"status"`
30+
Error string `json:"error,omitempty"`
31+
StartedAt *time.Time `json:"started_at,omitempty"`
32+
CompletedAt *time.Time `json:"completed_at,omitempty"`
33+
SubSteps []subStep `json:"sub_steps,omitempty"`
34+
}
35+
1636
func newLogsCommand() *cobra.Command {
1737
cmd := &cobra.Command{
1838
Use: "logs [execution-id]",
@@ -38,6 +58,7 @@ When called without arguments, lists recent executions with optional filters.`,
3858
cmd.Flags().String("status", "", "filter by status (pending, running, completed, failed, cancelled)")
3959
cmd.Flags().String("since", "", "filter by time (e.g., 1h, 24h, 7d)")
4060
cmd.Flags().Int("limit", 20, "max results to return")
61+
cmd.Flags().Bool("shallow", false, "suppress child workflow details")
4162

4263
return cmd
4364
}
@@ -81,12 +102,6 @@ func showExecutionDetail(cmd *cobra.Command, execID string) error {
81102
defer rows.Close()
82103

83104
// Fetch sub-steps grouped by parent.
84-
type subStep struct {
85-
StepName string `json:"step_name"`
86-
Status string `json:"status"`
87-
Started *time.Time `json:"started_at,omitempty"`
88-
Completed *time.Time `json:"completed_at,omitempty"`
89-
}
90105
subStepsByParent := make(map[string][]subStep)
91106
subRows, err := database.QueryContext(cmd.Context(),
92107
`SELECT parent_step_id, step_name, status, started_at, completed_at
@@ -105,16 +120,6 @@ func showExecutionDetail(cmd *cobra.Command, execID string) error {
105120
}
106121

107122
// Collect step data.
108-
type stepInfo struct {
109-
ID string `json:"id"`
110-
Name string `json:"name"`
111-
Status string `json:"status"`
112-
Error string `json:"error,omitempty"`
113-
StartedAt *time.Time `json:"started_at,omitempty"`
114-
CompletedAt *time.Time `json:"completed_at,omitempty"`
115-
SubSteps []subStep `json:"sub_steps,omitempty"`
116-
}
117-
118123
var stepsData []stepInfo
119124
for rows.Next() {
120125
var stepID, stepName, stepStatus string
@@ -173,17 +178,26 @@ func showExecutionDetail(cmd *cobra.Command, execID string) error {
173178
}
174179
fmt.Fprintln(cmd.OutOrStdout())
175180

181+
shallow, _ := cmd.Flags().GetBool("shallow")
182+
176183
fmt.Fprintln(cmd.OutOrStdout(), "Steps:")
184+
printSteps(cmd, database, execID, teamID, stepsData, shallow, " ")
185+
186+
return nil
187+
}
188+
189+
// printSteps renders step details, optionally recursing into child workflow executions.
190+
func printSteps(cmd *cobra.Command, database *sql.DB, execID, teamID string, stepsData []stepInfo, shallow bool, indent string) {
177191
for _, si := range stepsData {
178192
icon := statusIcon(si.Status)
179193
duration := ""
180194
if si.StartedAt != nil && si.CompletedAt != nil {
181195
duration = fmt.Sprintf(" %s", si.CompletedAt.Sub(*si.StartedAt).Round(time.Millisecond))
182196
}
183197

184-
fmt.Fprintf(cmd.OutOrStdout(), " %s %-20s %s%s\n", icon, si.Name, si.Status, duration)
198+
fmt.Fprintf(cmd.OutOrStdout(), "%s%s %-20s %s%s\n", indent, icon, si.Name, si.Status, duration)
185199
if si.Error != "" {
186-
fmt.Fprintf(cmd.OutOrStdout(), " error: %s\n", si.Error)
200+
fmt.Fprintf(cmd.OutOrStdout(), "%s error: %s\n", indent, si.Error)
187201
}
188202

189203
// Render sub-steps (tool calls) as indented tree.
@@ -208,7 +222,7 @@ func showExecutionDetail(cmd *cobra.Command, execID string) error {
208222
if isLastRound {
209223
connector = "└─"
210224
}
211-
fmt.Fprintf(cmd.OutOrStdout(), " %s round %s\n", connector, round)
225+
fmt.Fprintf(cmd.OutOrStdout(), "%s %s round %s\n", indent, connector, round)
212226

213227
for si, s := range rounds[round] {
214228
isLast := si == len(rounds[round])-1
@@ -229,13 +243,102 @@ func showExecutionDetail(cmd *cobra.Command, execID string) error {
229243
if s.Started != nil && s.Completed != nil {
230244
subDur = fmt.Sprintf(" %s", s.Completed.Sub(*s.Started).Round(time.Millisecond))
231245
}
232-
fmt.Fprintf(cmd.OutOrStdout(), " %s %s %-16s %s%s\n", prefix, subConn, toolName, s.Status, subDur)
246+
fmt.Fprintf(cmd.OutOrStdout(), "%s %s %s %-16s %s%s\n", indent, prefix, subConn, toolName, s.Status, subDur)
233247
}
234248
}
235249
}
250+
251+
// Show child workflow execution details if not in shallow mode.
252+
if !shallow {
253+
printChildExecution(cmd, database, execID, si.Name, teamID, indent+" ")
254+
}
236255
}
256+
}
237257

238-
return nil
258+
// printChildExecution queries for a child workflow execution spawned by a given step
259+
// and recursively renders its steps.
260+
func printChildExecution(cmd *cobra.Command, database *sql.DB, parentExecID, parentStepName, teamID, indent string) {
261+
rows, err := database.QueryContext(cmd.Context(),
262+
`SELECT id, workflow_name, status FROM workflow_executions
263+
WHERE parent_execution_id = $1 AND parent_step_name = $2 AND team_id = $3`,
264+
parentExecID, parentStepName, teamID,
265+
)
266+
if err != nil {
267+
return
268+
}
269+
defer rows.Close()
270+
271+
for rows.Next() {
272+
var childID, childWorkflow, childStatus string
273+
if err := rows.Scan(&childID, &childWorkflow, &childStatus); err != nil {
274+
continue
275+
}
276+
277+
fmt.Fprintf(cmd.OutOrStdout(), "%s↳ child: %s (%s) [%s]\n", indent, childWorkflow, childID, childStatus)
278+
279+
// Fetch child's steps.
280+
childSteps := fetchStepsForExecution(cmd, database, childID)
281+
if len(childSteps) > 0 {
282+
printSteps(cmd, database, childID, teamID, childSteps, false, indent+" ")
283+
}
284+
}
285+
}
286+
287+
// fetchStepsForExecution retrieves step data for a given execution ID.
288+
func fetchStepsForExecution(cmd *cobra.Command, database *sql.DB, execID string) []stepInfo {
289+
stepRows, err := database.QueryContext(cmd.Context(),
290+
`SELECT id, step_name, status, error, started_at, completed_at
291+
FROM step_executions WHERE execution_id = $1 AND parent_step_id IS NULL
292+
ORDER BY created_at ASC`, execID,
293+
)
294+
if err != nil {
295+
return nil
296+
}
297+
defer stepRows.Close()
298+
299+
// Fetch sub-steps for the child execution.
300+
subStepsByParent := make(map[string][]subStep)
301+
subRows, err := database.QueryContext(cmd.Context(),
302+
`SELECT parent_step_id, step_name, status, started_at, completed_at
303+
FROM step_executions WHERE execution_id = $1 AND parent_step_id IS NOT NULL
304+
ORDER BY created_at ASC`, execID,
305+
)
306+
if err == nil {
307+
defer subRows.Close()
308+
for subRows.Next() {
309+
var parentID, sName, sStatus string
310+
var sStarted, sCompleted *time.Time
311+
if err := subRows.Scan(&parentID, &sName, &sStatus, &sStarted, &sCompleted); err == nil {
312+
subStepsByParent[parentID] = append(subStepsByParent[parentID], subStep{sName, sStatus, sStarted, sCompleted})
313+
}
314+
}
315+
}
316+
317+
var steps []stepInfo
318+
for stepRows.Next() {
319+
var stepID, stepName, stepStatus string
320+
var stepError *string
321+
var stepStarted, stepCompleted *time.Time
322+
if err := stepRows.Scan(&stepID, &stepName, &stepStatus, &stepError, &stepStarted, &stepCompleted); err != nil {
323+
continue
324+
}
325+
326+
si := stepInfo{
327+
ID: stepID,
328+
Name: stepName,
329+
Status: stepStatus,
330+
StartedAt: stepStarted,
331+
CompletedAt: stepCompleted,
332+
}
333+
if stepError != nil && *stepError != "" {
334+
si.Error = *stepError
335+
}
336+
if subs, ok := subStepsByParent[stepID]; ok {
337+
si.SubSteps = subs
338+
}
339+
steps = append(steps, si)
340+
}
341+
return steps
239342
}
240343

241344
// listExecutions lists recent executions with optional filters.

‎packages/engine/internal/cli/root.go‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,7 @@ Full documentation: https://mantle.dvflw.co/docs`,
3333
cmd.PersistentFlags().String("log-level", "", "log level (debug, info, warn, error)")
3434
cmd.PersistentFlags().String("api-key", "", "API key for authentication (overrides cached credentials)")
3535
cmd.PersistentFlags().StringP("output", "o", "text", "Output format: text or json")
36+
cmd.PersistentFlags().Int("max-workflow-depth", 0, "Maximum workflow nesting depth (default: 10)")
3637

3738
// Command groups for organized help output.
3839
cmd.AddGroup(

‎packages/engine/internal/cli/run.go‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -64,8 +64,10 @@ func newRunCommand() *cobra.Command {
6464
if err != nil {
6565
return fmt.Errorf("creating engine: %w", err)
6666
}
67+
eng.MaxWorkflowDepth = cfg.Engine.MaxWorkflowDepth
6768
eng.MaxConcurrentExecutionsPerTeam = cfg.Engine.MaxConcurrentExecutionsPerTeam
6869
eng.CEL.SetConfigEnv(cfg.Env)
70+
eng.RegisterWorkflowConnector()
6971

7072
// Configure credential resolver with Postgres-backed store when encryption key is set.
7173
if cfg.Encryption.Key != "" {

0 commit comments

Comments
 (0)