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>

Author: github-actions[bot]

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 <execution-id>",
@@ -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)

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.