diff --git a/CLAUDE.md b/CLAUDE.md index c424e92..0eb2f54 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -68,10 +68,13 @@ mantle validate workflow.yaml # Offline schema validation mantle plan workflow.yaml # Diff against applied version mantle apply workflow.yaml # Apply versioned definition mantle run # Manual trigger -mantle cancel # Cancel running workflow +mantle cancel # Cancel running workflow +mantle retry # Retry from failed step +mantle rollback # Rollback to previous version mantle logs # View execution logs mantle status # View execution state mantle secrets create # Create typed credential +mantle secrets rotate-key # Re-encrypt credentials with new key mantle serve # Start persistent server ``` diff --git a/packages/engine/docker-compose.yml b/packages/engine/docker-compose.yml index d04752f..fb6b008 100644 --- a/packages/engine/docker-compose.yml +++ b/packages/engine/docker-compose.yml @@ -57,9 +57,9 @@ services: MANTLE_DATABASE_URL: postgres://mantle:mantle@postgres:5432/mantle?sslmode=disable MANTLE_API_ADDRESS: ":8080" MANTLE_ENCRYPTION_KEY: "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef" - MANTLE_TMP_TYPE: s3 - MANTLE_TMP_BUCKET: mantle-artifacts - MANTLE_TMP_PREFIX: artifacts/ + MANTLE_STORAGE_TYPE: s3 + MANTLE_STORAGE_BUCKET: mantle-artifacts + MANTLE_STORAGE_PREFIX: artifacts/ # Point S3 client at MinIO AWS_ACCESS_KEY_ID: minioadmin AWS_SECRET_ACCESS_KEY: minioadmin diff --git a/packages/engine/internal/artifact/reaper.go b/packages/engine/internal/artifact/reaper.go index 851f8a3..51f71ee 100644 --- a/packages/engine/internal/artifact/reaper.go +++ b/packages/engine/internal/artifact/reaper.go @@ -9,10 +9,10 @@ import ( "time" ) -// Reaper cleans up expired artifacts from both tmp storage and the database. +// Reaper cleans up expired artifacts from both storage and the database. type Reaper struct { Store *Store - TmpStorage TmpStorage + Storage Storage Retention time.Duration Logger *slog.Logger } @@ -40,8 +40,8 @@ func (r *Reaper) Sweep(ctx context.Context) (int, error) { cleaned := 0 for _, a := range expired { - // Delete file from tmp storage. - if delErr := r.TmpStorage.Delete(ctx, a.URL); delErr != nil { + // Delete file from storage. + if delErr := r.Storage.Delete(ctx, a.URL); delErr != nil { // If the blob is already gone, still clean up the metadata. if !errors.Is(delErr, os.ErrNotExist) { logger.Error("failed to delete artifact file", diff --git a/packages/engine/internal/artifact/reaper_test.go b/packages/engine/internal/artifact/reaper_test.go index 18bd94c..f85dd30 100644 --- a/packages/engine/internal/artifact/reaper_test.go +++ b/packages/engine/internal/artifact/reaper_test.go @@ -10,7 +10,7 @@ import ( func TestReaper_CleansExpiredArtifacts(t *testing.T) { dir := t.TempDir() - fs := &FilesystemTmpStorage{BasePath: dir} + fs := &FilesystemStorage{BasePath: dir} db := setupTestDB(t) store := &Store{DB: db} ctx := context.Background() @@ -48,7 +48,7 @@ func TestReaper_CleansExpiredArtifacts(t *testing.T) { reaper := &Reaper{ Store: store, - TmpStorage: fs, + Storage: fs, Retention: 24 * time.Hour, } @@ -72,11 +72,11 @@ func TestReaper_CleansExpiredArtifacts(t *testing.T) { func TestReaper_SkipsWhenRetentionZero(t *testing.T) { dir := t.TempDir() - fs := &FilesystemTmpStorage{BasePath: dir} + fs := &FilesystemStorage{BasePath: dir} reaper := &Reaper{ Store: &Store{}, // safe: Sweep returns early when Retention <= 0, before accessing Store.DB - TmpStorage: fs, + Storage: fs, Retention: 0, } diff --git a/packages/engine/internal/artifact/tmp.go b/packages/engine/internal/artifact/storage.go similarity index 51% rename from packages/engine/internal/artifact/tmp.go rename to packages/engine/internal/artifact/storage.go index 76ffa11..fb13635 100644 --- a/packages/engine/internal/artifact/tmp.go +++ b/packages/engine/internal/artifact/storage.go @@ -9,9 +9,9 @@ import ( "strings" ) -// TmpStorage persists and retrieves artifact files. -type TmpStorage interface { - // Put copies a local file to tmp storage at the given key. +// Storage persists and retrieves artifact files. +type Storage interface { + // Put copies a local file to storage at the given key. // Returns the URL or path where the file can be accessed. Put(ctx context.Context, key string, localPath string) (string, error) @@ -22,32 +22,39 @@ type TmpStorage interface { DeleteByPrefix(ctx context.Context, prefix string) error } -// FilesystemTmpStorage stores artifacts on the local filesystem. -type FilesystemTmpStorage struct { +// FilesystemStorage stores artifacts on the local filesystem. +type FilesystemStorage struct { BasePath string } // Put copies the file at localPath to the storage location identified by key and returns its path. -func (fs *FilesystemTmpStorage) Put(ctx context.Context, key string, localPath string) (string, error) { +func (fs *FilesystemStorage) Put(ctx context.Context, key string, localPath string) (string, error) { + // Reject obviously malicious keys before any filesystem operations. + if filepath.IsAbs(key) || key == ".." || strings.HasPrefix(key, ".."+string(filepath.Separator)) { + return "", fmt.Errorf("invalid artifact key: must be a relative path within the storage directory") + } + destPath := filepath.Join(fs.BasePath, key) // Validate destination is within BasePath to prevent path traversal. - absBase, err := filepath.Abs(fs.BasePath) + // Use EvalSymlinks for canonicalization to prevent symlink-based escapes. + resolvedBase, err := filepath.EvalSymlinks(fs.BasePath) if err != nil { return "", fmt.Errorf("resolving base path: %w", err) } - absDest, err := filepath.Abs(destPath) + // Ensure dest directory exists before resolving symlinks on the full path. + if err := os.MkdirAll(filepath.Dir(destPath), 0755); err != nil { + return "", fmt.Errorf("creating artifact dir: %w", err) + } + resolvedDest, err := filepath.EvalSymlinks(filepath.Dir(destPath)) if err != nil { return "", fmt.Errorf("resolving dest path: %w", err) } - rel, err := filepath.Rel(absBase, absDest) + resolvedDest = filepath.Join(resolvedDest, filepath.Base(destPath)) + rel, err := filepath.Rel(resolvedBase, resolvedDest) if err != nil || strings.HasPrefix(rel, "..") { return "", fmt.Errorf("key escapes base path") } - if err := os.MkdirAll(filepath.Dir(destPath), 0755); err != nil { - return "", fmt.Errorf("creating artifact dir: %w", err) - } - // Reject symlinks and non-regular files to prevent exfiltration. fi, err := os.Lstat(localPath) if err != nil { @@ -79,38 +86,57 @@ func (fs *FilesystemTmpStorage) Put(ctx context.Context, key string, localPath s } // DeleteByPrefix removes all files stored under the given key prefix. -func (fs *FilesystemTmpStorage) DeleteByPrefix(ctx context.Context, prefix string) error { +func (fs *FilesystemStorage) DeleteByPrefix(ctx context.Context, prefix string) error { target := filepath.Join(fs.BasePath, prefix) - absBase, err := filepath.Abs(fs.BasePath) + resolvedBase, err := filepath.EvalSymlinks(fs.BasePath) if err != nil { return fmt.Errorf("resolving base path: %w", err) } - absTarget, err := filepath.Abs(target) + resolvedTarget, err := filepath.EvalSymlinks(target) if err != nil { + // Target doesn't exist — compute candidate relative to resolvedBase for traversal check. + candidate := filepath.Join(resolvedBase, prefix) + rel, relErr := filepath.Rel(resolvedBase, candidate) + if relErr != nil || strings.HasPrefix(rel, "..") { + return fmt.Errorf("prefix escapes base path") + } + // Target doesn't exist and is within base — nothing to delete. + if os.IsNotExist(err) { + return nil + } return fmt.Errorf("resolving target path: %w", err) } - rel, err := filepath.Rel(absBase, absTarget) + rel, err := filepath.Rel(resolvedBase, resolvedTarget) if err != nil || strings.HasPrefix(rel, "..") { return fmt.Errorf("prefix escapes base path") } - return os.RemoveAll(target) + return os.RemoveAll(resolvedTarget) } // Delete removes a single artifact file by its path (as returned by Put). -func (fs *FilesystemTmpStorage) Delete(ctx context.Context, url string) error { - absBase, err := filepath.Abs(fs.BasePath) +func (fs *FilesystemStorage) Delete(ctx context.Context, url string) error { + resolvedBase, err := filepath.EvalSymlinks(fs.BasePath) if err != nil { return fmt.Errorf("resolving base path: %w", err) } - absURL, err := filepath.Abs(url) + resolvedURL, err := filepath.EvalSymlinks(url) if err != nil { + // File doesn't exist — compute path relative to resolvedBase for traversal check. + // url is an absolute path returned by Put, so use it directly. + rel, relErr := filepath.Rel(resolvedBase, url) + if relErr != nil || strings.HasPrefix(rel, "..") { + return fmt.Errorf("artifact path escapes base path") + } + if os.IsNotExist(err) { + return nil // already gone + } return fmt.Errorf("resolving artifact path: %w", err) } - rel, err := filepath.Rel(absBase, absURL) + rel, err := filepath.Rel(resolvedBase, resolvedURL) if err != nil || strings.HasPrefix(rel, "..") { return fmt.Errorf("artifact path escapes base path") } - if err := os.Remove(absURL); err != nil && !os.IsNotExist(err) { + if err := os.Remove(resolvedURL); err != nil && !os.IsNotExist(err) { return err } return nil diff --git a/packages/engine/internal/artifact/tmp_test.go b/packages/engine/internal/artifact/storage_test.go similarity index 86% rename from packages/engine/internal/artifact/tmp_test.go rename to packages/engine/internal/artifact/storage_test.go index 7162a4c..6a1201f 100644 --- a/packages/engine/internal/artifact/tmp_test.go +++ b/packages/engine/internal/artifact/storage_test.go @@ -7,9 +7,9 @@ import ( "testing" ) -func TestFilesystemTmpStorage_PutAndGet(t *testing.T) { +func TestFilesystemStorage_PutAndGet(t *testing.T) { dir := t.TempDir() - fs := &FilesystemTmpStorage{BasePath: dir} + fs := &FilesystemStorage{BasePath: dir} ctx := context.Background() key := "test-workflow/exec-123/my-artifact/data.tar.gz" @@ -40,9 +40,9 @@ func TestFilesystemTmpStorage_PutAndGet(t *testing.T) { } } -func TestFilesystemTmpStorage_Delete(t *testing.T) { +func TestFilesystemStorage_Delete(t *testing.T) { dir := t.TempDir() - fs := &FilesystemTmpStorage{BasePath: dir} + fs := &FilesystemStorage{BasePath: dir} ctx := context.Background() // Write a file @@ -68,9 +68,9 @@ func TestFilesystemTmpStorage_Delete(t *testing.T) { } } -func TestFilesystemTmpStorage_DeleteEscapeProtection(t *testing.T) { +func TestFilesystemStorage_DeleteEscapeProtection(t *testing.T) { dir := t.TempDir() - fs := &FilesystemTmpStorage{BasePath: dir} + fs := &FilesystemStorage{BasePath: dir} ctx := context.Background() err := fs.DeleteByPrefix(ctx, "../../etc") diff --git a/packages/engine/internal/audit/audit.go b/packages/engine/internal/audit/audit.go index 4dfac91..40383be 100644 --- a/packages/engine/internal/audit/audit.go +++ b/packages/engine/internal/audit/audit.go @@ -17,6 +17,7 @@ const ( ActionStepSkipped Action = "step.skipped" ActionStepContinuedOnError Action = "step.continued_on_error" ActionExecutionCancelled Action = "execution.cancelled" + ActionExecutionRetried Action = "execution.retried" ActionArtifactPersisted Action = "artifact.persisted" // Admin operations. @@ -29,7 +30,8 @@ const ( ActionAPIKeyRevoked Action = "apikey.revoked" ActionCredentialCreated Action = "credential.created" ActionCredentialDeleted Action = "credential.deleted" - ActionCredentialRotated Action = "credential.rotated" + ActionCredentialRotated Action = "credential.rotated" + ActionSecretKeyRotated Action = "secret.key_rotated" ActionAuthFailed Action = "auth.failed" // Email trigger operations. @@ -37,10 +39,21 @@ const ( ActionEmailConnectionEstablished Action = "email.connection.established" ActionEmailConnectionFailed Action = "email.connection.failed" + // Hook operations. + ActionHookStepStarted Action = "hook.step.started" + ActionHookStepCompleted Action = "hook.step.completed" + ActionHookStepFailed Action = "hook.step.failed" + // Budget operations. ActionBudgetExceeded Action = "budget.exceeded" ActionBudgetWarning Action = "budget.warning" ActionBudgetUpdated Action = "budget.updated" + + // Promotion operations. + ActionExecutionPromoted Action = "execution.promoted" + + // Rollback operations. + ActionWorkflowRolledBack Action = "workflow.rolled_back" ) // Resource identifies the target of an audit event. diff --git a/packages/engine/internal/audit/postgres.go b/packages/engine/internal/audit/postgres.go index 9e0ebfa..4ce2129 100644 --- a/packages/engine/internal/audit/postgres.go +++ b/packages/engine/internal/audit/postgres.go @@ -33,6 +33,25 @@ func (p *PostgresEmitter) enrichFromContext(ctx context.Context, event Event) Ev func (p *PostgresEmitter) Emit(ctx context.Context, event Event) error { // Enrich event metadata with auth context if available. event = p.enrichFromContext(ctx, event) + return emitEvent(ctx, p.DB, event) +} + +// execer abstracts ExecContext so both *sql.DB and *sql.Tx can emit audit events. +type execer interface { + ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error) +} + +// EmitTx emits an audit event using an existing transaction. +// This ensures the audit record is committed atomically with the operation it describes. +// If emitter is non-nil, the event is enriched with auth context before inserting. +func EmitTx(ctx context.Context, tx *sql.Tx, event Event, emitter ...*PostgresEmitter) error { + if len(emitter) > 0 && emitter[0] != nil { + event = emitter[0].enrichFromContext(ctx, event) + } + return emitEvent(ctx, tx, event) +} + +func emitEvent(ctx context.Context, db execer, event Event) error { beforeJSON, err := marshalNullableJSON(event.Before) if err != nil { return fmt.Errorf("marshaling before state: %w", err) @@ -61,7 +80,7 @@ func (p *PostgresEmitter) Emit(ctx context.Context, event Event) error { teamID = event.TeamID } - _, err = p.DB.ExecContext(ctx, + _, err = db.ExecContext(ctx, `INSERT INTO audit_events (timestamp, actor, action, resource_type, resource_id, before_state, after_state, metadata, team_id) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)`, ts, event.Actor, string(event.Action), event.Resource.Type, event.Resource.ID, diff --git a/packages/engine/internal/cel/cel.go b/packages/engine/internal/cel/cel.go index f93f3ee..826b88f 100644 --- a/packages/engine/internal/cel/cel.go +++ b/packages/engine/internal/cel/cel.go @@ -18,6 +18,8 @@ type Context struct { Inputs map[string]any // inputs. → workflow inputs Trigger map[string]any // trigger.payload → webhook trigger data Artifacts map[string]map[string]any // artifacts. → {name, url, size} + Hooks map[string]map[string]any // hooks..output → hook step outputs + Execution map[string]any // execution.status, execution.error, etc. } const maxProgramCacheSize = 10000 @@ -38,6 +40,8 @@ func NewEvaluator() (*Evaluator, error) { cel.Variable("env", cel.MapType(cel.StringType, cel.StringType)), cel.Variable("trigger", cel.MapType(cel.StringType, cel.DynType)), cel.Variable("artifacts", cel.MapType(cel.StringType, cel.DynType)), + cel.Variable("hooks", cel.MapType(cel.StringType, cel.DynType)), + cel.Variable("execution", cel.MapType(cel.StringType, cel.DynType)), } opts = append(opts, customFunctions()...) @@ -66,12 +70,24 @@ func (e *Evaluator) Eval(expression string, ctx *Context) (any, error) { artifacts = map[string]map[string]any{} } + hooks := ctx.Hooks + if hooks == nil { + hooks = map[string]map[string]any{} + } + + execution := ctx.Execution + if execution == nil { + execution = map[string]any{} + } + vars := map[string]any{ "steps": ctx.Steps, "inputs": ctx.Inputs, "env": e.envCache, "trigger": trigger, "artifacts": artifacts, + "hooks": hooks, + "execution": execution, } out, _, err := prog.Eval(vars) diff --git a/packages/engine/internal/cel/cel_test.go b/packages/engine/internal/cel/cel_test.go index b89882e..99c1948 100644 --- a/packages/engine/internal/cel/cel_test.go +++ b/packages/engine/internal/cel/cel_test.go @@ -318,6 +318,56 @@ func TestEnvVars_PrefixStripping(t *testing.T) { } } +func TestEval_HookAndExecutionVariables(t *testing.T) { + eval, err := NewEvaluator() + if err != nil { + t.Fatalf("NewEvaluator() error: %v", err) + } + + ctx := &Context{ + Steps: map[string]map[string]any{}, + Inputs: map[string]any{}, + Hooks: map[string]map[string]any{ + "notify": { + "output": map[string]any{ + "sent": true, + }, + }, + }, + Execution: map[string]any{ + "status": "failed", + "failed_step": "fetch", + }, + } + + // Test: hooks['notify'].output.sent == true + result, err := eval.EvalBool(`hooks['notify'].output.sent == true`, ctx) + if err != nil { + t.Fatalf("Eval hooks sent: %v", err) + } + if !result { + t.Error("hooks['notify'].output.sent should be true") + } + + // Test: execution.status == "failed" + result, err = eval.EvalBool(`execution.status == "failed"`, ctx) + if err != nil { + t.Fatalf("Eval execution.status: %v", err) + } + if !result { + t.Error("execution.status should be 'failed'") + } + + // Test: execution.failed_step == "fetch" + result, err = eval.EvalBool(`execution.failed_step == "fetch"`, ctx) + if err != nil { + t.Fatalf("Eval execution.failed_step: %v", err) + } + if !result { + t.Error("execution.failed_step should be 'fetch'") + } +} + func TestEval_ArtifactsAccess(t *testing.T) { eval, err := NewEvaluator() if err != nil { diff --git a/packages/engine/internal/cli/cancel.go b/packages/engine/internal/cli/cancel.go index d085405..80045d3 100644 --- a/packages/engine/internal/cli/cancel.go +++ b/packages/engine/internal/cli/cancel.go @@ -28,11 +28,11 @@ func newCancelCommand() *cobra.Command { } defer database.Close() - // Only cancel if currently pending or running. + // 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')`, + WHERE id = $1 AND status IN ('pending', 'running', 'queued')`, execID, ) if err != nil { diff --git a/packages/engine/internal/cli/retry.go b/packages/engine/internal/cli/retry.go new file mode 100644 index 0000000..11eb0dd --- /dev/null +++ b/packages/engine/internal/cli/retry.go @@ -0,0 +1,135 @@ +package cli + +import ( + "encoding/json" + "fmt" + + "github.com/dvflw/mantle/internal/config" + "github.com/dvflw/mantle/internal/db" + "github.com/dvflw/mantle/internal/engine" + "github.com/dvflw/mantle/internal/secret" + "github.com/spf13/cobra" +) + +func newRetryCommand() *cobra.Command { + cmd := &cobra.Command{ + Use: "retry ", + Short: "Retry a failed workflow execution", + Long: `Creates a new execution that resumes from the failure point of a +previous execution. Completed upstream steps are copied; the failed step +and everything downstream re-execute.`, + Example: ` mantle retry 01234567-89ab-cdef-0123-456789abcdef + mantle retry 01234567-89ab-cdef-0123-456789abcdef --from-step process-data + mantle retry 01234567-89ab-cdef-0123-456789abcdef --force`, + Args: cobra.ExactArgs(1), + RunE: func(cmd *cobra.Command, args []string) error { + execID := args[0] + + cfg := config.FromContext(cmd.Context()) + if cfg == nil { + return fmt.Errorf("config not loaded") + } + + database, err := db.Open(cfg.Database) + if err != nil { + return fmt.Errorf("failed to connect to database: %w", err) + } + defer database.Close() + + eng, err := engine.New(database) + if err != nil { + return fmt.Errorf("creating engine: %w", err) + } + eng.MaxConcurrentExecutionsPerTeam = cfg.Engine.MaxConcurrentExecutionsPerTeam + + // Configure credential resolver with Postgres-backed store when encryption key is set. + if cfg.Encryption.Key != "" { + encryptor, encErr := secret.NewEncryptor(cfg.Encryption.Key) + if encErr != nil { + return fmt.Errorf("configuring encryption: %w", encErr) + } + eng.Resolver = &secret.Resolver{ + Store: &secret.Store{DB: database, Encryptor: encryptor}, + } + } + + fromStep, _ := cmd.Flags().GetString("from-step") + force, _ := cmd.Flags().GetBool("force") + outputFormat, _ := cmd.Flags().GetString("output") + verbose, _ := cmd.Flags().GetBool("verbose") + + if outputFormat != "json" { + fmt.Fprintf(cmd.OutOrStdout(), "Retrying execution %s", execID) + if fromStep != "" { + fmt.Fprintf(cmd.OutOrStdout(), " from step %q", fromStep) + } + fmt.Fprintln(cmd.OutOrStdout(), "...") + } + + result, err := eng.RetryExecution(cmd.Context(), execID, fromStep, force) + if err != nil { + return fmt.Errorf("retry failed: %w", err) + } + + // Compute the exit error before writing output so that JSON + // mode exits non-zero on failure/timeout/cancellation. + var exitErr error + switch result.Status { + case "failed", "timed_out", "cancelled": + failedStep := "" + for _, s := range orderedSteps(result) { + if s.status == "failed" { + failedStep = s.name + break + } + } + if failedStep != "" { + exitErr = fmt.Errorf("workflow %s at step %q: %s", result.Status, failedStep, result.Error) + } else { + exitErr = fmt.Errorf("workflow %s: %s", result.Status, result.Error) + } + } + + // JSON output mode. + if outputFormat == "json" { + if encErr := json.NewEncoder(cmd.OutOrStdout()).Encode(result); encErr != nil { + return fmt.Errorf("encoding JSON output: %w", encErr) + } + return exitErr + } + + // Text output mode. + fmt.Fprintf(cmd.OutOrStdout(), "Execution %s: %s (%s)\n", + result.ExecutionID, result.Status, formatDuration(result.Duration)) + + steps := orderedSteps(result) + if verbose { + maxLen := 0 + for _, s := range steps { + if len(s.name) > maxLen { + maxLen = len(s.name) + } + } + for _, step := range steps { + line := fmt.Sprintf(" %s %-*s %s (%s)", + statusIcon(step.status), maxLen, step.name+":", step.status, formatDuration(step.duration)) + if step.output != "" { + line += fmt.Sprintf(" -> %s", truncate(step.output, 500)) + } + fmt.Fprintln(cmd.OutOrStdout(), line) + } + } else { + for _, step := range steps { + fmt.Fprintf(cmd.OutOrStdout(), " %s %s: %s\n", statusIcon(step.status), step.name, step.status) + } + } + + return exitErr + }, + } + + cmd.Flags().String("from-step", "", "Step to retry from (default: first failed step)") + cmd.Flags().Bool("force", false, "Bypass concurrency limits") + cmd.Flags().BoolP("verbose", "v", false, "Show step outputs and durations") + return cmd +} diff --git a/packages/engine/internal/cli/rollback.go b/packages/engine/internal/cli/rollback.go new file mode 100644 index 0000000..1f101c3 --- /dev/null +++ b/packages/engine/internal/cli/rollback.go @@ -0,0 +1,153 @@ +package cli + +import ( + "database/sql" + "fmt" + "log" + "strconv" + + "github.com/dvflw/mantle/internal/audit" + "github.com/dvflw/mantle/internal/auth" + "github.com/dvflw/mantle/internal/config" + "github.com/dvflw/mantle/internal/db" + "github.com/spf13/cobra" +) + +func newRollbackCommand() *cobra.Command { + cmd := &cobra.Command{ + Use: "rollback ", + Short: "Revert a workflow to a previous version", + Long: `Creates a new version of a workflow with the content of a previous version. +In-flight executions are unaffected. The rollback is recorded in the version +history via the rollback_of column.`, + Example: ` mantle rollback my-workflow + mantle rollback my-workflow --to-version 3`, + Args: cobra.ExactArgs(1), + RunE: func(cmd *cobra.Command, args []string) error { + workflowName := args[0] + + cfg := config.FromContext(cmd.Context()) + if cfg == nil { + return fmt.Errorf("config not loaded") + } + + database, err := db.Open(cfg.Database) + if err != nil { + return fmt.Errorf("failed to connect to database: %w", err) + } + defer database.Close() + + teamID := auth.TeamIDFromContext(cmd.Context()) + + // Wrap the entire read+insert in a transaction to prevent race conditions. + tx, err := database.BeginTx(cmd.Context(), nil) + if err != nil { + return fmt.Errorf("starting transaction: %w", err) + } + defer tx.Rollback() + + // 1. Get current (latest) version and its content hash. + var currentVersion int + var currentHash string + err = tx.QueryRowContext(cmd.Context(), + `SELECT version, content_hash FROM workflow_definitions WHERE name = $1 AND team_id = $2 ORDER BY version DESC LIMIT 1`, + workflowName, teamID, + ).Scan(¤tVersion, ¤tHash) + if err == sql.ErrNoRows { + return fmt.Errorf("workflow %q not found", workflowName) + } + if err != nil { + return fmt.Errorf("querying current version: %w", err) + } + + // 2. Determine target version. + toVersionStr, _ := cmd.Flags().GetString("to-version") + var targetVersion int + if toVersionStr != "" { + targetVersion, err = strconv.Atoi(toVersionStr) + if err != nil { + return fmt.Errorf("invalid --to-version value: %w", err) + } + } else { + // Get the second most recent version. + err = tx.QueryRowContext(cmd.Context(), + `SELECT version FROM workflow_definitions WHERE name = $1 AND team_id = $2 AND version < $3 ORDER BY version DESC LIMIT 1`, + workflowName, teamID, currentVersion, + ).Scan(&targetVersion) + if err == sql.ErrNoRows { + return fmt.Errorf("workflow %q has only one version — nothing to roll back to", workflowName) + } + if err != nil { + return fmt.Errorf("querying previous version: %w", err) + } + } + + // 3. Reject version <= 0. + if targetVersion <= 0 { + return fmt.Errorf("target version must be greater than 0, got %d", targetVersion) + } + + // 4. Load target version content and hash. + var targetContent []byte + var targetHash string + err = tx.QueryRowContext(cmd.Context(), + `SELECT content, content_hash FROM workflow_definitions WHERE name = $1 AND version = $2 AND team_id = $3`, + workflowName, targetVersion, teamID, + ).Scan(&targetContent, &targetHash) + if err == sql.ErrNoRows { + return fmt.Errorf("workflow %q version %d not found", workflowName, targetVersion) + } + if err != nil { + return fmt.Errorf("querying target version: %w", err) + } + + // 5. Reject no-op if content_hash matches current. + if targetHash == currentHash { + return fmt.Errorf("version %d has the same content as the current version %d — rollback is a no-op", + targetVersion, currentVersion) + } + + // 6. Insert new version with rollback_of. + newVersion := currentVersion + 1 + _, err = tx.ExecContext(cmd.Context(), + `INSERT INTO workflow_definitions (name, version, content, content_hash, rollback_of, team_id) VALUES ($1, $2, $3, $4, $5, $6)`, + workflowName, newVersion, targetContent, targetHash, targetVersion, teamID, + ) + if err != nil { + return fmt.Errorf("inserting rolled-back version: %w", err) + } + + if err := tx.Commit(); err != nil { + return fmt.Errorf("committing rollback: %w", err) + } + + // 7. Emit audit event. + emitter := &audit.PostgresEmitter{DB: database} + if err := emitter.Emit(cmd.Context(), audit.Event{ + Actor: "cli", + Action: audit.ActionWorkflowRolledBack, + Resource: audit.Resource{ + Type: "workflow_definition", + ID: workflowName, + }, + Metadata: map[string]string{ + "from_version": strconv.Itoa(currentVersion), + "to_version": strconv.Itoa(targetVersion), + "new_version": strconv.Itoa(newVersion), + "rollback_of": strconv.Itoa(targetVersion), + }, + }); err != nil { + log.Printf("warning: failed to emit audit event: %v", err) + } + + // 8. Print result. + fmt.Fprintf(cmd.OutOrStdout(), "Rolled back %s from version %d to content of version %d (now version %d)\n", + workflowName, currentVersion, targetVersion, newVersion) + + return nil + }, + } + + cmd.Flags().String("to-version", "", "Target version to roll back to (default: previous version)") + return cmd +} diff --git a/packages/engine/internal/cli/root.go b/packages/engine/internal/cli/root.go index 125ecd5..d435528 100644 --- a/packages/engine/internal/cli/root.go +++ b/packages/engine/internal/cli/root.go @@ -49,6 +49,8 @@ Full documentation: https://mantle.dvflw.co/docs`, newPlanCommand(), newApplyCommand(), newRunCommand(), + newRetryCommand(), + newRollbackCommand(), newCancelCommand(), newLogsCommand(), newStatusCommand(), diff --git a/packages/engine/internal/cli/rotate_key.go b/packages/engine/internal/cli/rotate_key.go index d743f17..5979120 100644 --- a/packages/engine/internal/cli/rotate_key.go +++ b/packages/engine/internal/cli/rotate_key.go @@ -1,28 +1,53 @@ package cli import ( + "context" "fmt" + "os" + "strings" + "time" + "github.com/dvflw/mantle/internal/audit" + "github.com/dvflw/mantle/internal/config" + "github.com/dvflw/mantle/internal/db" "github.com/dvflw/mantle/internal/secret" "github.com/spf13/cobra" ) func newSecretsRotateKeyCommand() *cobra.Command { - var newKey string + var newKeyFile string + var outputKey string cmd := &cobra.Command{ Use: "rotate-key", Short: "Re-encrypt all credentials with a new master key", Long: "Decrypts all stored credentials with the current master key and re-encrypts them with a new key. Used for key rotation after a security incident.", RunE: func(cmd *cobra.Command, args []string) error { - store, cleanup, err := newSecretStore(cmd) + cfg := config.FromContext(cmd.Context()) + if cfg == nil { + return fmt.Errorf("config not loaded") + } + + if cfg.Encryption.Key == "" { + return fmt.Errorf("encryption key not configured (set MANTLE_ENCRYPTION_KEY or encryption.key in mantle.yaml)") + } + + oldEncryptor, err := secret.NewEncryptor(cfg.Encryption.Key) if err != nil { - return err + return fmt.Errorf("invalid current key: %w", err) } - defer cleanup() - // Generate a new key if none was provided. - if newKey == "" { + // Resolve the new key: from file, or auto-generate. + var newKey string + userProvided := false + if newKeyFile != "" { + raw, err := os.ReadFile(newKeyFile) + if err != nil { + return fmt.Errorf("reading new key file: %w", err) + } + newKey = strings.TrimSpace(string(raw)) + userProvided = true + } else { generated, err := secret.GenerateKey() if err != nil { return fmt.Errorf("generating new key: %w", err) @@ -35,20 +60,62 @@ func newSecretsRotateKeyCommand() *cobra.Command { return fmt.Errorf("invalid new key: %w", err) } - count, err := store.ReEncryptAll(cmd.Context(), newEncryptor) + database, err := db.Open(cfg.Database) + if err != nil { + return fmt.Errorf("failed to connect to database: %w", err) + } + defer database.Close() + + tx, err := database.BeginTx(cmd.Context(), nil) + if err != nil { + return fmt.Errorf("starting transaction: %w", err) + } + defer tx.Rollback() + + count, err := secret.RotateAll(cmd.Context(), tx, oldEncryptor, newEncryptor) if err != nil { return fmt.Errorf("key rotation failed: %w", err) } + // Emit audit event inside the transaction so it commits atomically. + // Use context.Background() to prevent cancellation from dropping the audit record. + if err := audit.EmitTx(context.Background(), tx, audit.Event{ + Timestamp: time.Now(), + Actor: "cli", + Action: audit.ActionSecretKeyRotated, + Resource: audit.Resource{Type: "credentials", ID: "all"}, + Metadata: map[string]string{"count": fmt.Sprintf("%d", count)}, + }); err != nil { + return fmt.Errorf("emitting audit event: %w", err) + } + + if err := tx.Commit(); err != nil { + return fmt.Errorf("committing transaction: %w", err) + } + fmt.Fprintf(cmd.OutOrStdout(), "Re-encrypted %d credential(s).\n", count) - fmt.Fprintf(cmd.OutOrStdout(), "New key: %s\n", newKey) + + // Output the new key securely. + if !userProvided { + if outputKey != "" { + if err := os.WriteFile(outputKey, []byte(newKey+"\n"), 0600); err != nil { + return fmt.Errorf("writing key to file: %w", err) + } + fmt.Fprintf(cmd.OutOrStdout(), "New key written to: %s\n", outputKey) + } else { + fmt.Fprintln(cmd.ErrOrStderr(), "WARNING: The following key is sensitive. Store it securely and clear your terminal.") + fmt.Fprintf(cmd.OutOrStdout(), "New key: %s\n", newKey) + } + } + fmt.Fprintln(cmd.OutOrStdout(), "Update MANTLE_ENCRYPTION_KEY to the new key and restart.") return nil }, } - cmd.Flags().StringVar(&newKey, "new-key", "", "hex-encoded 32-byte new encryption key (auto-generated if omitted)") + cmd.Flags().StringVar(&newKeyFile, "new-key-file", "", "path to file containing hex-encoded 32-byte new encryption key (auto-generated if omitted)") + cmd.Flags().StringVar(&outputKey, "output-key", "", "path to write the auto-generated key (file permissions: 0600)") return cmd } diff --git a/packages/engine/internal/cli/run.go b/packages/engine/internal/cli/run.go index 01a9015..42be4df 100644 --- a/packages/engine/internal/cli/run.go +++ b/packages/engine/internal/cli/run.go @@ -64,6 +64,7 @@ func newRunCommand() *cobra.Command { if err != nil { return fmt.Errorf("creating engine: %w", err) } + eng.MaxConcurrentExecutionsPerTeam = cfg.Engine.MaxConcurrentExecutionsPerTeam // Configure credential resolver with Postgres-backed store when encryption key is set. if cfg.Encryption.Key != "" { @@ -78,19 +79,42 @@ func newRunCommand() *cobra.Command { outputFormat, _ := cmd.Flags().GetString("output") verbose, _ := cmd.Flags().GetBool("verbose") + force, _ := cmd.Flags().GetBool("force") if outputFormat != "json" { fmt.Fprintf(cmd.OutOrStdout(), "Running %s (version %d)...\n", workflowName, version) } - result, err := eng.Execute(cmd.Context(), workflowName, version, inputs) + result, err := eng.ExecuteWithOptions(cmd.Context(), workflowName, version, inputs, engine.ExecuteOptions{Force: force}) if err != nil { return fmt.Errorf("execution failed: %w", err) } + // Compute the exit error before writing output so that JSON + // mode exits non-zero on failure/timeout/cancellation. + var exitErr error + switch result.Status { + case "failed", "timed_out", "cancelled": + failedStep := "" + for _, s := range orderedSteps(result) { + if s.status == "failed" { + failedStep = s.name + break + } + } + if failedStep != "" { + exitErr = fmt.Errorf("workflow %s at step %q: %s", result.Status, failedStep, result.Error) + } else { + exitErr = fmt.Errorf("workflow %s: %s", result.Status, result.Error) + } + } + // JSON output mode. if outputFormat == "json" { - return json.NewEncoder(cmd.OutOrStdout()).Encode(result) + if encErr := json.NewEncoder(cmd.OutOrStdout()).Encode(result); encErr != nil { + return fmt.Errorf("encoding JSON output: %w", encErr) + } + return exitErr } // Text output mode. @@ -120,27 +144,13 @@ func newRunCommand() *cobra.Command { } } - if result.Status == "failed" { - // Find the failed step name for a more actionable error message. - failedStep := "" - for name, sr := range result.Steps { - if sr.Status == "failed" { - failedStep = name - break - } - } - if failedStep != "" { - return fmt.Errorf("workflow failed at step %q: %s", failedStep, result.Error) - } - return fmt.Errorf("workflow failed: %s", result.Error) - } - - return nil + return exitErr }, } cmd.Flags().StringArrayVar(&inputFlags, "input", nil, "Input parameter (key=value), can be specified multiple times") cmd.Flags().BoolP("verbose", "v", false, "Show step outputs and durations") + cmd.Flags().Bool("force", false, "Bypass per-workflow and per-team concurrency limits — executions will not be queued and may exceed configured limits") return cmd } diff --git a/packages/engine/internal/cli/serve.go b/packages/engine/internal/cli/serve.go index 9c45479..7b6abc6 100644 --- a/packages/engine/internal/cli/serve.go +++ b/packages/engine/internal/cli/serve.go @@ -50,6 +50,7 @@ func newServeCommand() *cobra.Command { return fmt.Errorf("creating engine: %w", err) } eng.MaxToolRoundsLimit = cfg.Engine.MaxToolRoundsLimit + eng.MaxConcurrentExecutionsPerTeam = cfg.Engine.MaxConcurrentExecutionsPerTeam // 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/config/config.go b/packages/engine/internal/config/config.go index 3252fe4..8ad2857 100644 --- a/packages/engine/internal/config/config.go +++ b/packages/engine/internal/config/config.go @@ -6,6 +6,7 @@ import ( "encoding/hex" "fmt" "log" + "log/slog" "net/url" "os" "time" @@ -19,6 +20,7 @@ import ( // Config holds all engine configuration. type Config struct { + Version int `mapstructure:"version"` Database DatabaseConfig `mapstructure:"database"` API APIConfig `mapstructure:"api"` Log LogConfig `mapstructure:"log"` @@ -29,7 +31,7 @@ type Config struct { AWS AWSConfig `mapstructure:"aws"` GCP GCPConfig `mapstructure:"gcp"` Azure AzureConfig `mapstructure:"azure"` - Tmp TmpConfig `mapstructure:"tmp"` + Storage StorageConfig `mapstructure:"storage"` } // RetentionConfig holds data retention settings. @@ -54,8 +56,8 @@ type AzureConfig struct { Region string `mapstructure:"region"` } -// TmpConfig configures ephemeral storage for workflow artifacts. -type TmpConfig struct { +// StorageConfig configures ephemeral storage for workflow artifacts. +type StorageConfig struct { Type string `mapstructure:"type"` // "s3" or "filesystem" Bucket string `mapstructure:"bucket"` // S3 bucket name (for type: s3) Prefix string `mapstructure:"prefix"` // Key prefix (for type: s3) @@ -131,6 +133,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 + MaxConcurrentExecutionsPerTeam int `mapstructure:"max_concurrent_executions_per_team"` Budget BudgetConfig `mapstructure:"budget"` } @@ -229,12 +232,12 @@ func Load(cmd *cobra.Command) (*Config, error) { _ = v.BindEnv("retention.execution_days", "MANTLE_RETENTION_EXECUTION_DAYS") _ = v.BindEnv("retention.audit_days", "MANTLE_RETENTION_AUDIT_DAYS") - // Tmp env var bindings - _ = v.BindEnv("tmp.type", "MANTLE_TMP_TYPE") - _ = v.BindEnv("tmp.bucket", "MANTLE_TMP_BUCKET") - _ = v.BindEnv("tmp.prefix", "MANTLE_TMP_PREFIX") - _ = v.BindEnv("tmp.path", "MANTLE_TMP_PATH") - _ = v.BindEnv("tmp.retention", "MANTLE_TMP_RETENTION") + // Storage env var bindings + _ = v.BindEnv("storage.type", "MANTLE_STORAGE_TYPE") + _ = v.BindEnv("storage.bucket", "MANTLE_STORAGE_BUCKET") + _ = v.BindEnv("storage.prefix", "MANTLE_STORAGE_PREFIX") + _ = v.BindEnv("storage.path", "MANTLE_STORAGE_PATH") + _ = v.BindEnv("storage.retention", "MANTLE_STORAGE_RETENTION") // Engine env var bindings _ = v.BindEnv("engine.node_id", "MANTLE_ENGINE_NODE_ID") @@ -251,6 +254,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_concurrent_executions_per_team", "MANTLE_ENGINE_MAX_CONCURRENT_EXECUTIONS_PER_TEAM") // Budget env var bindings _ = v.BindEnv("engine.budget.reset_mode", "MANTLE_ENGINE_BUDGET_RESET_MODE") @@ -274,6 +278,42 @@ func Load(cmd *cobra.Command) (*Config, error) { return nil, err } + // Validate config version. + // If the user did not set "version" at all, default to 1. + // If they explicitly set it (including to 0), it must be 1. + if !v.IsSet("version") { + cfg.Version = 1 + } + if cfg.Version != 1 { + return nil, fmt.Errorf("unsupported config version %d; this version of mantle supports config version 1 — upgrade mantle or check your mantle.yaml", cfg.Version) + } + + // Deprecated: fall back from "tmp" section to "storage" for backward compatibility. + // Field-by-field merge so that env vars / flags on "storage.*" are not clobbered. + if v.IsSet("tmp") { + if sub := v.Sub("tmp"); sub != nil { + var legacy StorageConfig + if err := sub.Unmarshal(&legacy); err == nil { + if cfg.Storage.Type == "" { + cfg.Storage.Type = legacy.Type + } + if cfg.Storage.Bucket == "" { + cfg.Storage.Bucket = legacy.Bucket + } + if cfg.Storage.Prefix == "" { + cfg.Storage.Prefix = legacy.Prefix + } + if cfg.Storage.Path == "" { + cfg.Storage.Path = legacy.Path + } + if cfg.Storage.Retention == "" { + cfg.Storage.Retention = legacy.Retention + } + slog.Warn("config section 'tmp' is deprecated and will be removed in a future release; rename it to 'storage'") + } + } + } + // Validate budget reset_day range. if cfg.Engine.Budget.ResetDay < 1 || cfg.Engine.Budget.ResetDay > 28 { if cfg.Engine.Budget.ResetMode == budget.ResetModeRolling { diff --git a/packages/engine/internal/config/config_test.go b/packages/engine/internal/config/config_test.go index 310ad7a..e78e213 100644 --- a/packages/engine/internal/config/config_test.go +++ b/packages/engine/internal/config/config_test.go @@ -3,6 +3,7 @@ package config import ( "os" "path/filepath" + "strings" "testing" "time" @@ -248,27 +249,27 @@ func TestLoad_BudgetResetDay_CalendarClampsUpperBound(t *testing.T) { assert.Equal(t, 1, cfg.Engine.Budget.ResetDay) } -func TestLoad_TmpConfigFromEnvVars(t *testing.T) { - t.Setenv("MANTLE_TMP_TYPE", "s3") - t.Setenv("MANTLE_TMP_BUCKET", "my-artifacts") - t.Setenv("MANTLE_TMP_PREFIX", "workflows/") - t.Setenv("MANTLE_TMP_RETENTION", "24h") +func TestLoad_StorageConfigFromEnvVars(t *testing.T) { + t.Setenv("MANTLE_STORAGE_TYPE", "s3") + t.Setenv("MANTLE_STORAGE_BUCKET", "my-artifacts") + t.Setenv("MANTLE_STORAGE_PREFIX", "workflows/") + t.Setenv("MANTLE_STORAGE_RETENTION", "24h") cmd := newTestCommand() cfg, err := Load(cmd) require.NoError(t, err) - assert.Equal(t, "s3", cfg.Tmp.Type) - assert.Equal(t, "my-artifacts", cfg.Tmp.Bucket) - assert.Equal(t, "workflows/", cfg.Tmp.Prefix) - assert.Equal(t, "24h", cfg.Tmp.Retention) + assert.Equal(t, "s3", cfg.Storage.Type) + assert.Equal(t, "my-artifacts", cfg.Storage.Bucket) + assert.Equal(t, "workflows/", cfg.Storage.Prefix) + assert.Equal(t, "24h", cfg.Storage.Retention) } -func TestLoad_TmpConfigFromConfigFile(t *testing.T) { +func TestLoad_StorageConfigFromConfigFile(t *testing.T) { dir := t.TempDir() configFile := filepath.Join(dir, "mantle.yaml") err := os.WriteFile(configFile, []byte(` -tmp: +storage: type: "filesystem" path: "/tmp/mantle-artifacts" retention: "48h" @@ -281,22 +282,79 @@ tmp: cfg, err := Load(cmd) require.NoError(t, err) - assert.Equal(t, "filesystem", cfg.Tmp.Type) - assert.Equal(t, "/tmp/mantle-artifacts", cfg.Tmp.Path) - assert.Equal(t, "48h", cfg.Tmp.Retention) + assert.Equal(t, "filesystem", cfg.Storage.Type) + assert.Equal(t, "/tmp/mantle-artifacts", cfg.Storage.Path) + assert.Equal(t, "48h", cfg.Storage.Retention) } -func TestLoad_TmpConfigEnvVarOverridesFile(t *testing.T) { +func TestLoad_VersionValidation(t *testing.T) { + tests := []struct { + name string + yaml string + wantErr string // empty means no error expected + }{ + { + name: "version 1 is valid", + yaml: "version: 1\n", + wantErr: "", + }, + { + name: "missing version defaults to 1", + yaml: "log:\n level: info\n", + wantErr: "", + }, + { + name: "version 0 is rejected", + yaml: "version: 0\n", + wantErr: "unsupported config version 0; this version of mantle supports config version 1", + }, + { + name: "version 2 is rejected", + yaml: "version: 2\n", + wantErr: "unsupported config version 2; this version of mantle supports config version 1", + }, + { + name: "version 99 is rejected", + yaml: "version: 99\n", + wantErr: "unsupported config version 99; this version of mantle supports config version 1", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + dir := t.TempDir() + configFile := filepath.Join(dir, "mantle.yaml") + err := os.WriteFile(configFile, []byte(tt.yaml), 0644) + require.NoError(t, err) + + cmd := newTestCommand() + _ = cmd.Flags().Set("config", configFile) + + cfg, err := Load(cmd) + if tt.wantErr != "" { + require.Error(t, err) + if !strings.Contains(err.Error(), tt.wantErr) { + t.Errorf("error %q does not contain %q", err.Error(), tt.wantErr) + } + } else { + require.NoError(t, err) + assert.Equal(t, 1, cfg.Version) + } + }) + } +} + +func TestLoad_StorageConfigEnvVarOverridesFile(t *testing.T) { dir := t.TempDir() configFile := filepath.Join(dir, "mantle.yaml") _ = os.WriteFile(configFile, []byte(` -tmp: +storage: type: "filesystem" path: "/tmp/file-path" `), 0644) - t.Setenv("MANTLE_TMP_TYPE", "s3") - t.Setenv("MANTLE_TMP_BUCKET", "env-bucket") + t.Setenv("MANTLE_STORAGE_TYPE", "s3") + t.Setenv("MANTLE_STORAGE_BUCKET", "env-bucket") cmd := newTestCommand() _ = cmd.Flags().Set("config", configFile) @@ -304,8 +362,68 @@ tmp: cfg, err := Load(cmd) require.NoError(t, err) - assert.Equal(t, "s3", cfg.Tmp.Type) - assert.Equal(t, "env-bucket", cfg.Tmp.Bucket) + assert.Equal(t, "s3", cfg.Storage.Type) + assert.Equal(t, "env-bucket", cfg.Storage.Bucket) // path from file should still be present since env var didn't override it - assert.Equal(t, "/tmp/file-path", cfg.Tmp.Path) + assert.Equal(t, "/tmp/file-path", cfg.Storage.Path) +} + +func TestLoad_StorageSectionUsedDirectly(t *testing.T) { + dir := t.TempDir() + configFile := filepath.Join(dir, "mantle.yaml") + _ = os.WriteFile(configFile, []byte(` +storage: + type: "s3" + bucket: "my-bucket" +`), 0644) + + cmd := newTestCommand() + _ = cmd.Flags().Set("config", configFile) + + cfg, err := Load(cmd) + require.NoError(t, err) + + assert.Equal(t, "s3", cfg.Storage.Type) + assert.Equal(t, "my-bucket", cfg.Storage.Bucket) +} + +func TestLoad_TmpFallbackToStorageWithWarning(t *testing.T) { + dir := t.TempDir() + configFile := filepath.Join(dir, "mantle.yaml") + _ = os.WriteFile(configFile, []byte(` +tmp: + type: "filesystem" + path: "/tmp/legacy-artifacts" +`), 0644) + + cmd := newTestCommand() + _ = cmd.Flags().Set("config", configFile) + + cfg, err := Load(cmd) + require.NoError(t, err) + + assert.Equal(t, "filesystem", cfg.Storage.Type) + assert.Equal(t, "/tmp/legacy-artifacts", cfg.Storage.Path) +} + +func TestLoad_StorageTakesPrecedenceOverTmp(t *testing.T) { + dir := t.TempDir() + configFile := filepath.Join(dir, "mantle.yaml") + _ = os.WriteFile(configFile, []byte(` +storage: + type: "s3" + bucket: "new-bucket" +tmp: + type: "filesystem" + path: "/tmp/old-path" +`), 0644) + + cmd := newTestCommand() + _ = cmd.Flags().Set("config", configFile) + + cfg, err := Load(cmd) + require.NoError(t, err) + + assert.Equal(t, "s3", cfg.Storage.Type) + assert.Equal(t, "new-bucket", cfg.Storage.Bucket) } diff --git a/packages/engine/internal/db/migrations/016_safety_net.sql b/packages/engine/internal/db/migrations/016_safety_net.sql new file mode 100644 index 0000000..e1d927d --- /dev/null +++ b/packages/engine/internal/db/migrations/016_safety_net.sql @@ -0,0 +1,70 @@ +-- +goose Up + +-- Concurrency controls (#49): add 'queued' and 'timed_out' to execution status check. +ALTER TABLE workflow_executions DROP CONSTRAINT IF EXISTS chk_execution_status; +ALTER TABLE workflow_executions + ADD CONSTRAINT chk_execution_status + CHECK (status IN ('pending', 'running', 'completed', 'failed', 'cancelled', 'queued', 'timed_out')); + +-- Concurrency controls (#49): per-team execution limit override. +ALTER TABLE teams ADD COLUMN max_concurrent_executions INT; + +-- Retry from failed step (#48): link to original execution for traceability. +ALTER TABLE workflow_executions ADD COLUMN retried_from_execution_id UUID + REFERENCES workflow_executions(id); + +-- Step-level timeout detection: add 'timed_out' to step status check. +ALTER TABLE step_executions DROP CONSTRAINT IF EXISTS chk_step_status; +ALTER TABLE step_executions + ADD CONSTRAINT chk_step_status + CHECK (status IN ('pending', 'running', 'completed', 'failed', 'cancelled', 'skipped', 'timed_out')); + +-- Lifecycle hooks (#30): distinguish hook steps from main steps. +ALTER TABLE step_executions ADD COLUMN hook_block TEXT; + +-- Replace the monolithic unique constraint with two partial indexes so that +-- hook steps can reuse the same step_name as main steps (disambiguated by hook_block). +ALTER TABLE step_executions DROP CONSTRAINT IF EXISTS step_executions_execution_id_step_name_attempt_key; +CREATE UNIQUE INDEX idx_step_executions_main_unique + ON step_executions (execution_id, step_name, attempt) WHERE hook_block IS NULL; +CREATE UNIQUE INDEX idx_step_executions_hook_unique + ON step_executions (execution_id, step_name, attempt, hook_block) WHERE hook_block IS NOT NULL; + +-- Workflow rollback (#50): track which version was restored. +ALTER TABLE workflow_definitions ADD COLUMN rollback_of INT; + +-- Index for retry lookups. +CREATE INDEX idx_workflow_executions_retried_from ON workflow_executions(retried_from_execution_id) WHERE retried_from_execution_id IS NOT NULL; + +-- Partial index for hook vs main step filtering. +CREATE INDEX idx_step_executions_hook_block ON step_executions(execution_id, hook_block) WHERE hook_block IS NOT NULL; + +-- +goose Down +DROP INDEX IF EXISTS idx_step_executions_hook_block; +DROP INDEX IF EXISTS idx_step_executions_hook_unique; +DROP INDEX IF EXISTS idx_step_executions_main_unique; +DROP INDEX IF EXISTS idx_workflow_executions_retried_from; +ALTER TABLE workflow_definitions DROP COLUMN IF EXISTS rollback_of; +-- Delete hook step rows before dropping the column so the restored unique +-- constraint does not conflict with rows that were hook-scoped. +DELETE FROM step_executions WHERE hook_block IS NOT NULL; +ALTER TABLE step_executions DROP COLUMN IF EXISTS hook_block; +-- Restore the original unique constraint (no partial index needed since hook rows are gone). +ALTER TABLE step_executions + ADD CONSTRAINT step_executions_execution_id_step_name_attempt_key + UNIQUE (execution_id, step_name, attempt); +ALTER TABLE workflow_executions DROP COLUMN IF EXISTS retried_from_execution_id; +ALTER TABLE teams DROP COLUMN IF EXISTS max_concurrent_executions; +-- Convert new statuses back to old equivalents so the CHECK constraints pass. +UPDATE workflow_executions SET status = 'pending' WHERE status = 'queued'; +UPDATE workflow_executions SET status = 'failed' WHERE status = 'timed_out'; +UPDATE step_executions SET status = 'failed' WHERE status = 'timed_out'; +-- Restore original status constraints without queued/timed_out. +ALTER TABLE workflow_executions DROP CONSTRAINT IF EXISTS chk_execution_status; +ALTER TABLE workflow_executions + ADD CONSTRAINT chk_execution_status + CHECK (status IN ('pending', 'running', 'completed', 'failed', 'cancelled')); +ALTER TABLE step_executions DROP CONSTRAINT IF EXISTS chk_step_status; +ALTER TABLE step_executions + ADD CONSTRAINT chk_step_status + CHECK (status IN ('pending', 'running', 'completed', 'failed', 'cancelled', 'skipped')); diff --git a/packages/engine/internal/engine/concurrency.go b/packages/engine/internal/engine/concurrency.go new file mode 100644 index 0000000..0f61a8c --- /dev/null +++ b/packages/engine/internal/engine/concurrency.go @@ -0,0 +1,179 @@ +package engine + +import ( + "context" + "database/sql" + "fmt" + "hash/fnv" + "time" + + "github.com/dvflw/mantle/internal/audit" + "github.com/dvflw/mantle/internal/auth" + "github.com/dvflw/mantle/internal/metrics" +) + +// ConcurrencyResult describes whether an execution is allowed to start, +// should be queued, or was rejected. +type ConcurrencyResult struct { + Allowed bool + Queued bool + Err error +} + +// CheckConcurrencyLimits evaluates per-team and per-workflow concurrency +// limits within the given transaction. Advisory locks ensure serialisation +// so that two concurrent callers cannot both "win" the last slot. +// +// Evaluation order: +// 1. Per-team limit (teamMaxConcurrent, from engine config / teams table) +// 2. Per-workflow limit (maxParallelExecutions, from workflow YAML) +// +// onLimit controls what happens when a limit is hit: "queue" (default) or +// "reject". +func CheckConcurrencyLimits(ctx context.Context, tx *sql.Tx, workflowName string, maxParallelExecutions int, onLimit string, teamMaxConcurrent int) ConcurrencyResult { + if onLimit == "" { + onLimit = "queue" + } + + teamID := auth.TeamIDFromContext(ctx) + + // --- Per-team check --- + if teamMaxConcurrent > 0 && teamID != "" { + // Acquire transaction-scoped advisory lock keyed on team. + lockKey := hashString("team:" + teamID) + if _, err := tx.ExecContext(ctx, "SELECT pg_advisory_xact_lock($1)", lockKey); err != nil { + return ConcurrencyResult{Err: fmt.Errorf("acquiring team advisory lock: %w", err)} + } + + var count int + err := tx.QueryRowContext(ctx, + `SELECT COUNT(*) FROM workflow_executions WHERE team_id = $1 AND status IN ('pending', 'running')`, + teamID, + ).Scan(&count) + if err != nil { + return ConcurrencyResult{Err: fmt.Errorf("counting team executions: %w", err)} + } + + if count >= teamMaxConcurrent { + if onLimit == "reject" { + metrics.ExecutionsRejectedTotal.WithLabelValues(workflowName).Inc() + return ConcurrencyResult{Err: fmt.Errorf("team %q has reached the maximum of %d concurrent executions", teamID, teamMaxConcurrent)} + } + metrics.ExecutionsQueued.WithLabelValues(workflowName).Inc() + return ConcurrencyResult{Queued: true} + } + } + + // --- Per-workflow check --- + if maxParallelExecutions > 0 { + lockKey := hashString("workflow:" + workflowName) + if _, err := tx.ExecContext(ctx, "SELECT pg_advisory_xact_lock($1)", lockKey); err != nil { + return ConcurrencyResult{Err: fmt.Errorf("acquiring workflow advisory lock: %w", err)} + } + + var count int + err := tx.QueryRowContext(ctx, + `SELECT COUNT(*) FROM workflow_executions WHERE workflow_name = $1 AND status IN ('pending', 'running')`, + workflowName, + ).Scan(&count) + if err != nil { + return ConcurrencyResult{Err: fmt.Errorf("counting workflow executions: %w", err)} + } + + if count >= maxParallelExecutions { + if onLimit == "reject" { + metrics.ExecutionsRejectedTotal.WithLabelValues(workflowName).Inc() + return ConcurrencyResult{Err: fmt.Errorf("workflow %q has reached the maximum of %d concurrent executions", workflowName, maxParallelExecutions)} + } + metrics.ExecutionsQueued.WithLabelValues(workflowName).Inc() + return ConcurrencyResult{Queued: true} + } + } + + return ConcurrencyResult{Allowed: true} +} + +// PromoteQueued picks the oldest queued execution for a workflow and +// promotes it to pending so it will be picked up by the engine. +// Uses FOR UPDATE SKIP LOCKED to avoid contention with other promoters. +func PromoteQueued(ctx context.Context, db *sql.DB, workflowName string, auditor audit.Emitter) error { + var promotedID sql.NullString + err := db.QueryRowContext(ctx, + `UPDATE workflow_executions SET status = 'pending', updated_at = NOW() + WHERE id = ( + SELECT id FROM workflow_executions + WHERE workflow_name = $1 AND status = 'queued' + ORDER BY started_at ASC + LIMIT 1 + FOR UPDATE SKIP LOCKED + ) + RETURNING id`, + workflowName, + ).Scan(&promotedID) + if err == sql.ErrNoRows { + return nil // nothing to promote + } + if err != nil { + return fmt.Errorf("promoting queued execution: %w", err) + } + + if promotedID.Valid { + metrics.ExecutionsQueued.WithLabelValues(workflowName).Dec() + auditor.Emit(ctx, audit.Event{ + Timestamp: time.Now(), + Actor: "engine", + Action: audit.ActionExecutionPromoted, + Resource: audit.Resource{Type: "workflow_execution", ID: promotedID.String}, + Metadata: map[string]string{"workflow": workflowName, "scope": "workflow"}, + }) + } + return nil +} + +// PromoteQueuedByTeam promotes the oldest queued execution across all workflows for a team. +func PromoteQueuedByTeam(ctx context.Context, db *sql.DB, teamID string, auditor audit.Emitter) error { + if teamID == "" { + return nil + } + var promotedID sql.NullString + var promotedWorkflow sql.NullString + err := db.QueryRowContext(ctx, + `UPDATE workflow_executions + SET status = 'pending', updated_at = NOW() + WHERE id = ( + SELECT id FROM workflow_executions + WHERE team_id = $1 AND status = 'queued' + ORDER BY started_at ASC + LIMIT 1 + FOR UPDATE SKIP LOCKED + ) + RETURNING id, workflow_name`, + teamID, + ).Scan(&promotedID, &promotedWorkflow) + if err == sql.ErrNoRows { + return nil // nothing to promote + } + if err != nil { + return fmt.Errorf("promoting queued execution for team: %w", err) + } + + if promotedID.Valid { + wfName := promotedWorkflow.String + metrics.ExecutionsQueued.WithLabelValues(wfName).Dec() + auditor.Emit(ctx, audit.Event{ + Timestamp: time.Now(), + Actor: "engine", + Action: audit.ActionExecutionPromoted, + Resource: audit.Resource{Type: "workflow_execution", ID: promotedID.String}, + Metadata: map[string]string{"workflow": wfName, "team_id": teamID, "scope": "team"}, + }) + } + return nil +} + +// hashString returns a deterministic int64 hash for use as a Postgres advisory lock key. +func hashString(s string) int64 { + h := fnv.New64a() + h.Write([]byte(s)) + return int64(h.Sum64()) +} diff --git a/packages/engine/internal/engine/concurrency_test.go b/packages/engine/internal/engine/concurrency_test.go new file mode 100644 index 0000000..fdce6b3 --- /dev/null +++ b/packages/engine/internal/engine/concurrency_test.go @@ -0,0 +1,319 @@ +package engine + +import ( + "context" + "fmt" + "testing" + + "github.com/dvflw/mantle/internal/audit" + "github.com/dvflw/mantle/internal/auth" +) + +func TestCheckConcurrencyLimits_AllowedWhenUnderLimit(t *testing.T) { + database := setupTestDB(t) + ctx := context.Background() + + // Insert 2 running executions for workflow "wf-a". + for i := 0; i < 2; i++ { + _, err := database.ExecContext(ctx, + `INSERT INTO workflow_executions (workflow_name, workflow_version, status) + VALUES ('wf-a', 1, 'running')`) + if err != nil { + t.Fatalf("inserting running execution: %v", err) + } + } + + tx, err := database.BeginTx(ctx, nil) + if err != nil { + t.Fatalf("BeginTx: %v", err) + } + defer tx.Rollback() + + result := CheckConcurrencyLimits(ctx, tx, "wf-a", 5, "queue", 0) + if err := tx.Commit(); err != nil { + t.Fatalf("Commit: %v", err) + } + + if !result.Allowed { + t.Errorf("expected Allowed=true, got false (Queued=%v, Err=%v)", result.Queued, result.Err) + } + if result.Queued { + t.Error("expected Queued=false, got true") + } + if result.Err != nil { + t.Errorf("expected nil error, got: %v", result.Err) + } +} + +func TestCheckConcurrencyLimits_QueuedWhenOverLimit(t *testing.T) { + database := setupTestDB(t) + ctx := context.Background() + + // Insert 2 running executions for workflow "wf-b". + for i := 0; i < 2; i++ { + _, err := database.ExecContext(ctx, + `INSERT INTO workflow_executions (workflow_name, workflow_version, status) + VALUES ('wf-b', 1, 'running')`) + if err != nil { + t.Fatalf("inserting running execution: %v", err) + } + } + + tx, err := database.BeginTx(ctx, nil) + if err != nil { + t.Fatalf("BeginTx: %v", err) + } + defer tx.Rollback() + + result := CheckConcurrencyLimits(ctx, tx, "wf-b", 2, "queue", 0) + if err := tx.Commit(); err != nil { + t.Fatalf("Commit: %v", err) + } + + if result.Allowed { + t.Error("expected Allowed=false, got true") + } + if !result.Queued { + t.Errorf("expected Queued=true, got false (Err=%v)", result.Err) + } + if result.Err != nil { + t.Errorf("expected nil error, got: %v", result.Err) + } +} + +func TestCheckConcurrencyLimits_RejectedWhenOverLimitReject(t *testing.T) { + database := setupTestDB(t) + ctx := context.Background() + + // Insert 2 running executions for workflow "wf-c". + for i := 0; i < 2; i++ { + _, err := database.ExecContext(ctx, + `INSERT INTO workflow_executions (workflow_name, workflow_version, status) + VALUES ('wf-c', 1, 'running')`) + if err != nil { + t.Fatalf("inserting running execution: %v", err) + } + } + + tx, err := database.BeginTx(ctx, nil) + if err != nil { + t.Fatalf("BeginTx: %v", err) + } + defer tx.Rollback() + + result := CheckConcurrencyLimits(ctx, tx, "wf-c", 2, "reject", 0) + if err := tx.Commit(); err != nil { + t.Fatalf("Commit: %v", err) + } + + if result.Err == nil { + t.Fatal("expected non-nil error for reject policy, got nil") + } + if result.Allowed { + t.Error("expected Allowed=false when rejected") + } +} + +func TestCheckConcurrencyLimits_UnlimitedWhenZero(t *testing.T) { + database := setupTestDB(t) + ctx := context.Background() + + // Insert many running executions. + for i := 0; i < 10; i++ { + _, err := database.ExecContext(ctx, + `INSERT INTO workflow_executions (workflow_name, workflow_version, status) + VALUES ('wf-d', 1, 'running')`) + if err != nil { + t.Fatalf("inserting running execution: %v", err) + } + } + + tx, err := database.BeginTx(ctx, nil) + if err != nil { + t.Fatalf("BeginTx: %v", err) + } + defer tx.Rollback() + + // maxParallelExecutions=0 means unlimited. + result := CheckConcurrencyLimits(ctx, tx, "wf-d", 0, "queue", 0) + if err := tx.Commit(); err != nil { + t.Fatalf("Commit: %v", err) + } + + if !result.Allowed { + t.Errorf("expected Allowed=true when max=0 (unlimited), got false (Queued=%v, Err=%v)", result.Queued, result.Err) + } +} + +func TestPromoteQueued_PromotesOldest(t *testing.T) { + database := setupTestDB(t) + ctx := context.Background() + + // Insert 2 queued executions with different started_at times. + var oldestID, newestID string + err := database.QueryRowContext(ctx, + `INSERT INTO workflow_executions (workflow_name, workflow_version, status, started_at) + VALUES ('wf-promote', 1, 'queued', NOW() - INTERVAL '10 minutes') + RETURNING id`).Scan(&oldestID) + if err != nil { + t.Fatalf("inserting oldest queued: %v", err) + } + + err = database.QueryRowContext(ctx, + `INSERT INTO workflow_executions (workflow_name, workflow_version, status, started_at) + VALUES ('wf-promote', 1, 'queued', NOW() - INTERVAL '1 minute') + RETURNING id`).Scan(&newestID) + if err != nil { + t.Fatalf("inserting newest queued: %v", err) + } + + // Promote oldest. + if err := PromoteQueued(ctx, database, "wf-promote", &audit.NoopEmitter{}); err != nil { + t.Fatalf("PromoteQueued() error: %v", err) + } + + // Verify oldest was promoted to pending. + var oldestStatus, newestStatus string + if err := database.QueryRowContext(ctx, + `SELECT status FROM workflow_executions WHERE id = $1`, oldestID).Scan(&oldestStatus); err != nil { + t.Fatalf("querying oldest status: %v", err) + } + if err := database.QueryRowContext(ctx, + `SELECT status FROM workflow_executions WHERE id = $1`, newestID).Scan(&newestStatus); err != nil { + t.Fatalf("querying newest status: %v", err) + } + + if oldestStatus != "pending" { + t.Errorf("oldest execution status = %q, want %q", oldestStatus, "pending") + } + if newestStatus != "queued" { + t.Errorf("newest execution status = %q, want %q (should remain queued)", newestStatus, "queued") + } +} + +func TestPromoteQueuedByTeam_PromotesOldest(t *testing.T) { + database := setupTestDB(t) + ctx := context.Background() + + teamID := auth.DefaultTeamID + + // Insert 2 queued executions for the default team. + var oldestID, newestID string + err := database.QueryRowContext(ctx, + `INSERT INTO workflow_executions (workflow_name, workflow_version, status, started_at, team_id) + VALUES ('wf-team-a', 1, 'queued', NOW() - INTERVAL '10 minutes', $1) + RETURNING id`, teamID).Scan(&oldestID) + if err != nil { + t.Fatalf("inserting oldest queued: %v", err) + } + err = database.QueryRowContext(ctx, + `INSERT INTO workflow_executions (workflow_name, workflow_version, status, started_at, team_id) + VALUES ('wf-team-b', 1, 'queued', NOW() - INTERVAL '1 minute', $1) + RETURNING id`, teamID).Scan(&newestID) + if err != nil { + t.Fatalf("inserting newest queued: %v", err) + } + + if err := PromoteQueuedByTeam(ctx, database, teamID, &audit.NoopEmitter{}); err != nil { + t.Fatalf("PromoteQueuedByTeam() error: %v", err) + } + + var oldestStatus string + if err := database.QueryRowContext(ctx, + `SELECT status FROM workflow_executions WHERE id = $1`, oldestID).Scan(&oldestStatus); err != nil { + t.Fatalf("querying oldest status: %v", err) + } + if oldestStatus != "pending" { + t.Errorf("oldest execution status = %q, want %q", oldestStatus, "pending") + } +} + +func TestCheckConcurrencyLimits_TeamLevelLimit(t *testing.T) { + database := setupTestDB(t) + ctx := context.Background() + teamID := auth.DefaultTeamID + + // Insert 3 running executions for the team across different workflows. + for i := 0; i < 3; i++ { + _, err := database.ExecContext(ctx, + `INSERT INTO workflow_executions (workflow_name, workflow_version, status, team_id) + VALUES ($1, 1, 'running', $2)`, + fmt.Sprintf("wf-team-%d", i), teamID) + if err != nil { + t.Fatalf("inserting running execution: %v", err) + } + } + + // context.Background() yields DefaultTeamID via TeamIDFromContext. + tx, err := database.BeginTx(ctx, nil) + if err != nil { + t.Fatalf("BeginTx: %v", err) + } + defer tx.Rollback() + + // Team limit of 3 with 3 running — should queue. + result := CheckConcurrencyLimits(ctx, tx, "wf-new", 0, "queue", 3) + if err := tx.Commit(); err != nil { + t.Fatalf("Commit: %v", err) + } + + if result.Allowed { + t.Error("expected Allowed=false when team is at limit, got true") + } + if !result.Queued { + t.Errorf("expected Queued=true, got false (Err=%v)", result.Err) + } + if result.Err != nil { + t.Errorf("expected nil error, got: %v", result.Err) + } +} + +func TestCheckConcurrencyLimits_TeamLevelAllowed(t *testing.T) { + database := setupTestDB(t) + ctx := context.Background() + teamID := auth.DefaultTeamID + + // Insert 1 running execution for the team. + _, err := database.ExecContext(ctx, + `INSERT INTO workflow_executions (workflow_name, workflow_version, status, team_id) + VALUES ('wf-team-ok', 1, 'running', $1)`, teamID) + if err != nil { + t.Fatalf("inserting running execution: %v", err) + } + + tx, err := database.BeginTx(ctx, nil) + if err != nil { + t.Fatalf("BeginTx: %v", err) + } + defer tx.Rollback() + + // Team limit of 5 with 1 running — should be allowed. + result := CheckConcurrencyLimits(ctx, tx, "wf-new", 0, "queue", 5) + if err := tx.Commit(); err != nil { + t.Fatalf("Commit: %v", err) + } + + if !result.Allowed { + t.Errorf("expected Allowed=true when under team limit, got false (Queued=%v, Err=%v)", result.Queued, result.Err) + } +} + +func TestHashString_Returns64Bit(t *testing.T) { + // Verify deterministic output. + a := hashString("team:abc") + b := hashString("team:abc") + if a != b { + t.Errorf("hashString not deterministic: %d != %d", a, b) + } + + // Verify different inputs produce different hashes. + c := hashString("workflow:xyz") + if a == c { + t.Error("hashString returned same value for different inputs") + } + + // Verify non-zero (extremely unlikely for FNV but good sanity check). + if a == 0 { + t.Error("hashString returned 0") + } +} diff --git a/packages/engine/internal/engine/engine.go b/packages/engine/internal/engine/engine.go index d3bafd9..b85f1c9 100644 --- a/packages/engine/internal/engine/engine.go +++ b/packages/engine/internal/engine/engine.go @@ -4,6 +4,7 @@ import ( "context" "database/sql" "encoding/json" + "errors" "fmt" "log" "os" @@ -24,16 +25,17 @@ import ( // Engine executes workflows by running steps sequentially with checkpoint-and-resume. type Engine struct { - DB *sql.DB - Registry *connector.Registry - Auditor audit.Emitter - CEL *mantleCEL.Evaluator - Resolver *secret.Resolver - MaxToolRoundsLimit int // admin ceiling for max_rounds; 0 = no limit - BudgetChecker *budget.Checker // nil = budget enforcement disabled - BudgetStore *budget.Store // nil = token usage recording disabled - ArtifactStore *artifact.Store // nil = artifact system disabled - TmpStorage artifact.TmpStorage // nil = artifact system disabled + DB *sql.DB + Registry *connector.Registry + Auditor audit.Emitter + CEL *mantleCEL.Evaluator + Resolver *secret.Resolver + MaxToolRoundsLimit int // admin ceiling for max_rounds; 0 = no limit + MaxConcurrentExecutionsPerTeam int // 0 = unlimited; per-team concurrency cap + BudgetChecker *budget.Checker // nil = budget enforcement disabled + BudgetStore *budget.Store // nil = token usage recording disabled + ArtifactStore *artifact.Store // nil = artifact system disabled + Storage artifact.Storage // nil = artifact system disabled } // StepContext carries workflow-level metadata needed by step execution. @@ -75,12 +77,70 @@ type StepResult struct { Duration time.Duration `json:"duration"` } +// ExecuteOptions controls execution behavior. +type ExecuteOptions struct { + Force bool // bypass concurrency limits +} + // Execute runs a workflow by name, pinned to the specified version. func (e *Engine) Execute(ctx context.Context, workflowName string, version int, inputs map[string]any) (*ExecutionResult, error) { - // Create execution record. - execID, err := e.createExecution(ctx, workflowName, version, inputs) + return e.ExecuteWithOptions(ctx, workflowName, version, inputs, ExecuteOptions{}) +} + +// ExecuteWithOptions runs a workflow with configurable execution options. +func (e *Engine) ExecuteWithOptions(ctx context.Context, workflowName string, version int, inputs map[string]any, opts ExecuteOptions) (*ExecutionResult, error) { + // Load workflow to check concurrency settings. + wf, err := e.loadWorkflow(ctx, workflowName, version) if err != nil { - return nil, fmt.Errorf("creating execution: %w", err) + return nil, fmt.Errorf("loading workflow: %w", err) + } + + // Check concurrency limits unless force-bypassed. + // When concurrency limits apply, the check and execution insert happen + // in the same transaction so the advisory lock is held until the row + // exists — closing the TOCTOU window. + var execID string + initialStatus := "pending" + if !opts.Force && (wf.MaxParallelExecutions > 0 || e.MaxConcurrentExecutionsPerTeam > 0) { + tx, txErr := e.DB.BeginTx(ctx, nil) + if txErr != nil { + return nil, fmt.Errorf("starting concurrency check transaction: %w", txErr) + } + defer tx.Rollback() + + result := CheckConcurrencyLimits(ctx, tx, workflowName, + wf.MaxParallelExecutions, wf.OnLimit, e.MaxConcurrentExecutionsPerTeam) + + if result.Err != nil { + return nil, result.Err // rejected + } + if result.Queued { + initialStatus = "queued" + } + + // Insert execution row inside the same transaction. + execID, err = e.createExecutionTx(ctx, tx, workflowName, version, inputs, initialStatus) + if err != nil { + return nil, fmt.Errorf("creating execution: %w", err) + } + + if err := tx.Commit(); err != nil { + return nil, fmt.Errorf("committing concurrency check: %w", err) + } + } else { + // No concurrency limits — use implicit transaction. + execID, err = e.createExecution(ctx, workflowName, version, inputs, initialStatus) + if err != nil { + return nil, fmt.Errorf("creating execution: %w", err) + } + } + + // If queued, return immediately with queued status. + if initialStatus == "queued" { + return &ExecutionResult{ + ExecutionID: execID, + Status: "queued", + }, nil } return e.resumeExecution(ctx, execID, workflowName, version, inputs) @@ -175,7 +235,8 @@ func (e *Engine) resumeExecution(ctx context.Context, execID string, workflowNam TeamID: auth.TeamIDFromContext(ctx), } - // Execute steps sequentially. + // Execute steps sequentially, tracking failure state for hooks. + var failedStepName string for _, step := range wf.Steps { // Skip already-completed steps (checkpoint recovery). if _, done := completedSteps[step.Name]; done { @@ -207,7 +268,7 @@ func (e *Engine) resumeExecution(ctx context.Context, execID string, workflowNam e.loadArtifactsIntoCELContext(ctx, execID, celCtx) } else if stepResult.Status == "skipped" { celCtx.Steps[step.Name] = map[string]any{"output": map[string]any{}, "error": nil} - } else if stepResult.Status == "failed" { + } else if stepResult.Status == "failed" || stepResult.Status == "timed_out" { if step.ContinueOnError { // Step failed but workflow continues — expose error and partial output to downstream steps. celCtx.Steps[step.Name] = map[string]any{ @@ -223,26 +284,90 @@ func (e *Engine) resumeExecution(ctx context.Context, execID string, workflowNam }) metrics.StepsContinuedOnErrorTotal.WithLabelValues(workflowName, step.Name).Inc() } else { - // Original behavior — halt execution. - result.Status = "failed" - result.Error = fmt.Sprintf("step %q failed: %s", step.Name, stepResult.Error) + // Halt main execution — record failure or timeout. + failedStepName = step.Name + terminalStatus := stepResult.Status + if terminalStatus == "failed" && ctx.Err() == context.DeadlineExceeded { + terminalStatus = "timed_out" + } + result.Status = terminalStatus + if terminalStatus == "timed_out" { + result.Error = fmt.Sprintf("step %q timed out: %s", step.Name, stepResult.Error) + } else { + result.Error = fmt.Sprintf("step %q failed: %s", step.Name, stepResult.Error) + } result.Duration = time.Since(execStart) - if err := e.updateExecutionStatus(ctx, execID, "failed", result.Error); err != nil { + // Use a background context for the DB update in case the original is expired. + updateCtx := ctx + if ctx.Err() != nil { + updateCtx = context.Background() + if user := auth.UserFromContext(ctx); user != nil { + updateCtx = auth.WithUser(updateCtx, user) + } + } + if err := e.updateExecutionStatus(updateCtx, execID, terminalStatus, result.Error); err != nil { return nil, fmt.Errorf("checkpoint: %w", err) } - metrics.ExecutionsTotal.WithLabelValues(workflowName, "failed").Inc() + metrics.ExecutionsTotal.WithLabelValues(workflowName, terminalStatus).Inc() metrics.ExecutionDuration.WithLabelValues(workflowName).Observe(time.Since(execStart).Seconds()) + break + } + } + } + + // If no step failed, mark execution as completed. + if failedStepName == "" { + if err := e.updateExecutionStatus(ctx, execID, "completed", ""); err != nil { + return nil, fmt.Errorf("checkpoint: %w", err) + } + result.Duration = time.Since(execStart) + metrics.ExecutionsTotal.WithLabelValues(workflowName, "completed").Inc() + metrics.ExecutionDuration.WithLabelValues(workflowName).Observe(time.Since(execStart).Seconds()) + } + + // Run lifecycle hooks if configured. + // Hooks are best-effort — they never alter the workflow execution status. + if wf.Hooks != nil { + mainStatus := result.Status + // Detect timeout: context expired and not a user cancellation. + if ctx.Err() != nil && result.Status != "cancelled" { + mainStatus = "timed_out" + } + + // Determine the hook execution context. + // If the main workflow timed out, hooks need a fresh context so they + // get their own timeout budget. For user cancellations, skip hooks entirely. + hookCtx := ctx + if ctx.Err() != nil { + if result.Status == "cancelled" { + // User cancellation — skip hooks. return result, nil } + // Timeout — create fresh context and propagate team identity. + hookCtx = context.Background() + if user := auth.UserFromContext(ctx); user != nil { + hookCtx = auth.WithUser(hookCtx, user) + } } + + e.executeHooks(hookCtx, execID, workflowName, wf, mainStatus, result.Error, failedStepName, celCtx, sc) } - if err := e.updateExecutionStatus(ctx, execID, "completed", ""); err != nil { - return nil, fmt.Errorf("checkpoint: %w", err) + // Promote next queued execution now that this slot is free (after hooks complete). + // Use a fresh context — the original may be expired if the workflow timed out. + promoteCtx, promoteCancel := context.WithTimeout(context.Background(), 10*time.Second) + defer promoteCancel() + if err := PromoteQueued(promoteCtx, e.DB, workflowName, e.Auditor); err != nil { + log.Printf("failed to promote queued execution for %s: %v", workflowName, err) + metrics.PromotionFailuresTotal.WithLabelValues(workflowName, "workflow").Inc() } - result.Duration = time.Since(execStart) - metrics.ExecutionsTotal.WithLabelValues(workflowName, "completed").Inc() - metrics.ExecutionDuration.WithLabelValues(workflowName).Observe(time.Since(execStart).Seconds()) + + // Also promote by team in case the execution was queued due to a team-level limit. + if err := PromoteQueuedByTeam(promoteCtx, e.DB, sc.TeamID, e.Auditor); err != nil { + log.Printf("failed to promote queued execution for team %s: %v", sc.TeamID, err) + metrics.PromotionFailuresTotal.WithLabelValues(workflowName, "team").Inc() + } + return result, nil } @@ -303,8 +428,12 @@ func (e *Engine) executeStep(ctx context.Context, execID string, workflowName st if lastErr != nil { errMsg := lastErr.Error() - if cpErr := e.updateStep(ctx, execID, step.Name, "failed", output, errMsg, step.ContinueOnError); cpErr != nil { - return StepResult{Status: "failed", Output: output, Error: fmt.Sprintf("checkpoint failure: %v (original: %s)", cpErr, errMsg)} + stepStatus := "failed" + if errors.Is(lastErr, context.DeadlineExceeded) { + stepStatus = "timed_out" + } + if cpErr := e.updateStep(ctx, execID, step.Name, stepStatus, output, errMsg, step.ContinueOnError); cpErr != nil { + return StepResult{Status: stepStatus, Output: output, Error: fmt.Sprintf("checkpoint failure: %v (original: %s)", cpErr, errMsg)} } e.Auditor.Emit(ctx, audit.Event{ Timestamp: time.Now(), @@ -312,8 +441,8 @@ func (e *Engine) executeStep(ctx context.Context, execID string, workflowName st Action: audit.ActionStepFailed, Resource: audit.Resource{Type: "step_execution", ID: step.Name}, }) - metrics.StepsTotal.WithLabelValues(workflowName, step.Name, "failed").Inc() - return StepResult{Status: "failed", Output: output, Error: errMsg} + metrics.StepsTotal.WithLabelValues(workflowName, step.Name, stepStatus).Inc() + return StepResult{Status: stepStatus, Output: output, Error: errMsg} } // Validate output size before persisting. @@ -449,8 +578,8 @@ func (e *Engine) executeStepLogic(ctx context.Context, execID string, step workf // Validate artifact subsystem is configured if step declares artifacts. var artifactsDir string if len(step.Artifacts) > 0 { - if e.TmpStorage == nil || e.ArtifactStore == nil { - return nil, fmt.Errorf("step %q declares artifacts but artifact subsystem is not configured (set tmp storage in mantle.yaml)", step.Name) + if e.Storage == nil || e.ArtifactStore == nil { + return nil, fmt.Errorf("step %q declares artifacts but artifact subsystem is not configured (set storage in mantle.yaml)", step.Name) } } @@ -483,7 +612,7 @@ func (e *Engine) executeStepLogic(ctx context.Context, execID string, step workf for attempt := 1; attempt <= maxAttempts; attempt++ { // Create a fresh artifacts scratch dir per attempt. - if len(step.Artifacts) > 0 && e.TmpStorage != nil { + if len(step.Artifacts) > 0 && e.Storage != nil { if artifactsDir != "" { os.RemoveAll(artifactsDir) // clean previous attempt's scratch } @@ -551,7 +680,7 @@ func (e *Engine) executeStepLogic(ctx context.Context, execID string, step workf } // Persist declared artifacts to tmp storage. - if lastErr == nil && len(step.Artifacts) > 0 && e.TmpStorage != nil && e.ArtifactStore != nil && artifactsDir != "" { + if lastErr == nil && len(step.Artifacts) > 0 && e.Storage != nil && e.ArtifactStore != nil && artifactsDir != "" { type persistedArtifact struct { id string url string @@ -563,7 +692,7 @@ func (e *Engine) executeStepLogic(ctx context.Context, execID string, step workf if delErr := e.ArtifactStore.DeleteByID(ctx, p.id); delErr != nil { log.Printf("warning: rollback: failed to delete artifact metadata %q: %v", p.id, delErr) } - if delErr := e.TmpStorage.Delete(ctx, p.url); delErr != nil { + if delErr := e.Storage.Delete(ctx, p.url); delErr != nil { log.Printf("warning: rollback: failed to delete artifact blob %q: %v", p.url, delErr) } } @@ -579,7 +708,7 @@ func (e *Engine) executeStepLogic(ctx context.Context, execID string, step workf } key := fmt.Sprintf("%s/%s/%s/%s", workflowName, execID, artDecl.Name, relPath) - url, putErr := e.TmpStorage.Put(ctx, key, localPath) + url, putErr := e.Storage.Put(ctx, key, localPath) if putErr != nil { rollback() return nil, fmt.Errorf("persisting artifact %q: %v", artDecl.Name, putErr) @@ -593,7 +722,7 @@ func (e *Engine) executeStepLogic(ctx context.Context, execID string, step workf Size: info.Size(), } if createErr := e.ArtifactStore.Create(ctx, art); createErr != nil { - if delErr := e.TmpStorage.Delete(ctx, url); delErr != nil { + if delErr := e.Storage.Delete(ctx, url); delErr != nil { log.Printf("warning: failed to clean up orphaned artifact blob %q: %v", url, delErr) } rollback() @@ -801,7 +930,9 @@ func (e *Engine) loadWorkflow(ctx context.Context, name string, version int) (*w } // createExecution inserts a new workflow_executions row and returns the ID. -func (e *Engine) createExecution(ctx context.Context, workflowName string, version int, inputs map[string]any) (string, error) { +// status should be "pending" for immediate execution or "queued" when +// concurrency limits require the execution to wait. +func (e *Engine) createExecution(ctx context.Context, workflowName string, version int, inputs map[string]any, status string) (string, error) { inputsJSON, err := json.Marshal(inputs) if err != nil { return "", fmt.Errorf("marshaling inputs: %w", err) @@ -812,9 +943,33 @@ func (e *Engine) createExecution(ctx context.Context, workflowName string, versi var id string err = e.DB.QueryRowContext(ctx, `INSERT INTO workflow_executions (workflow_name, workflow_version, status, inputs, started_at, team_id) - VALUES ($1, $2, 'pending', $3, NOW(), $4) + VALUES ($1, $2, $3, $4, NOW(), $5) + RETURNING id`, + workflowName, version, status, inputsJSON, teamID, + ).Scan(&id) + if err != nil { + return "", err + } + return id, nil +} + +// createExecutionTx inserts a new workflow_executions row within an existing +// transaction and returns the ID. This allows the insert to be atomic with +// other operations (e.g., concurrency checks) in the same transaction. +func (e *Engine) createExecutionTx(ctx context.Context, tx *sql.Tx, workflowName string, version int, inputs map[string]any, status string) (string, error) { + inputsJSON, err := json.Marshal(inputs) + if err != nil { + return "", fmt.Errorf("marshaling inputs: %w", err) + } + + teamID := auth.TeamIDFromContext(ctx) + + var id string + err = tx.QueryRowContext(ctx, + `INSERT INTO workflow_executions (workflow_name, workflow_version, status, inputs, started_at, team_id) + VALUES ($1, $2, $3, $4, NOW(), $5) RETURNING id`, - workflowName, version, inputsJSON, teamID, + workflowName, version, status, inputsJSON, teamID, ).Scan(&id) if err != nil { return "", err @@ -825,7 +980,7 @@ 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 { var completedAt any - if status == "completed" || status == "failed" || status == "cancelled" { + if status == "completed" || status == "failed" || status == "cancelled" || status == "timed_out" { completedAt = time.Now() } @@ -855,7 +1010,7 @@ func (e *Engine) recordStep(ctx context.Context, execID, stepName, status string _, err = e.DB.ExecContext(ctx, `INSERT INTO step_executions (execution_id, step_name, status, output, error, continue_on_error, started_at) VALUES ($1, $2, $3, $4, $5, $6, NOW()) - ON CONFLICT (execution_id, step_name, attempt) DO NOTHING`, + ON CONFLICT (execution_id, step_name, attempt) WHERE hook_block IS NULL DO NOTHING`, execID, stepName, status, outputJSON, errorVal, continueOnError, ) if err != nil { diff --git a/packages/engine/internal/engine/engine_test.go b/packages/engine/internal/engine/engine_test.go index 24bc847..dcc5d06 100644 --- a/packages/engine/internal/engine/engine_test.go +++ b/packages/engine/internal/engine/engine_test.go @@ -411,11 +411,11 @@ steps: t.Fatalf("Execute() error: %v", err) } - if result.Status != "failed" { - t.Errorf("status = %q, want %q", result.Status, "failed") + if result.Status != "timed_out" { + t.Errorf("status = %q, want %q", result.Status, "timed_out") } - if result.Steps["slow-step"].Status != "failed" { - t.Errorf("slow-step status = %q, want %q", result.Steps["slow-step"].Status, "failed") + if result.Steps["slow-step"].Status != "timed_out" { + t.Errorf("slow-step status = %q, want %q", result.Steps["slow-step"].Status, "timed_out") } } diff --git a/packages/engine/internal/engine/hooks.go b/packages/engine/internal/engine/hooks.go new file mode 100644 index 0000000..086e90f --- /dev/null +++ b/packages/engine/internal/engine/hooks.go @@ -0,0 +1,213 @@ +package engine + +import ( + "context" + "encoding/json" + "fmt" + "log" + "time" + + "github.com/dvflw/mantle/internal/audit" + mantleCEL "github.com/dvflw/mantle/internal/cel" + "github.com/dvflw/mantle/internal/metrics" + "github.com/dvflw/mantle/internal/workflow" +) + +// defaultHookTimeout is applied when the configured hooks timeout is unparseable. +const defaultHookTimeout = 5 * time.Minute + +// executeHooks runs lifecycle hooks after main workflow execution completes. +// Hook failures are best-effort — they never alter the workflow execution status. +func (e *Engine) executeHooks( + ctx context.Context, + execID string, + workflowName string, + wf *workflow.Workflow, + mainStatus string, // "completed", "failed", or "timed_out" + mainError string, + failedStep string, + celCtx *mantleCEL.Context, + sc StepContext, +) { + if wf.Hooks == nil { + return + } + + // Populate execution context for CEL expressions in hook steps. + // failed_in starts as "steps" when a main step failed; hook failures + // update it to the hook block name. Empty string when no failure. + failedIn := "" + if failedStep != "" { + failedIn = "steps" + } + celCtx.Execution = map[string]any{ + "status": mainStatus, + "error": mainError, + "failed_step": failedStep, + "failed_in": failedIn, + } + + // Initialize hooks namespace for inter-hook step references. + celCtx.Hooks = make(map[string]map[string]any) + + // Apply hooks-level timeout if configured. + if wf.Hooks.Timeout != "" { + dur, err := time.ParseDuration(wf.Hooks.Timeout) + if err != nil { + log.Printf("hooks: invalid timeout %q: %v — falling back to %s", wf.Hooks.Timeout, err, defaultHookTimeout) + dur = defaultHookTimeout + } + var cancel context.CancelFunc + ctx, cancel = context.WithTimeout(ctx, dur) + defer cancel() + } + + // Execute conditional block: on_success if completed, on_failure if failed/timed_out. + if mainStatus == "completed" && len(wf.Hooks.OnSuccess) > 0 { + e.executeHookBlock(ctx, execID, workflowName, "on_success", wf.Hooks.OnSuccess, celCtx, sc) + } else if (mainStatus == "failed" || mainStatus == "timed_out") && len(wf.Hooks.OnFailure) > 0 { + e.executeHookBlock(ctx, execID, workflowName, "on_failure", wf.Hooks.OnFailure, celCtx, sc) + } + + // Execute on_finish always. + if len(wf.Hooks.OnFinish) > 0 { + e.executeHookBlock(ctx, execID, workflowName, "on_finish", wf.Hooks.OnFinish, celCtx, sc) + } +} + +// executeHookBlock runs a single hook block (on_success, on_failure, or on_finish) sequentially. +func (e *Engine) executeHookBlock( + ctx context.Context, + execID string, + workflowName string, + blockName string, + steps []workflow.Step, + celCtx *mantleCEL.Context, + sc StepContext, +) { + for _, step := range steps { + // Check context cancellation (hooks timeout or parent cancellation). + if ctx.Err() != nil { + log.Printf("hooks: %s block cancelled (context done): %v", blockName, ctx.Err()) + return + } + + // Emit audit: hook step started. + e.Auditor.Emit(ctx, audit.Event{ + Timestamp: time.Now(), + Actor: "engine", + Action: audit.ActionHookStepStarted, + Resource: audit.Resource{Type: "hook_step_execution", ID: step.Name}, + Metadata: map[string]string{ + "execution_id": execID, + "hook_block": blockName, + "step": step.Name, + }, + }) + + // Record hook step as running. + if err := e.recordHookStep(ctx, execID, step.Name, blockName, "running", nil, ""); err != nil { + log.Printf("hooks: failed to record running state for %s/%s: %v", blockName, step.Name, err) + } + + // Execute the step using the shared step logic. + output, execErr := e.executeStepLogic(ctx, execID, step, celCtx, workflowName, sc) + + // Determine outcome. + if execErr != nil { + errMsg := execErr.Error() + + // Record failure in DB. + if err := e.recordHookStep(ctx, execID, step.Name, blockName, "failed", output, errMsg); err != nil { + log.Printf("hooks: failed to record failure for %s/%s: %v", blockName, step.Name, err) + } + + // Update CEL context with error info. + celCtx.Hooks[step.Name] = map[string]any{ + "output": output, + "error": errMsg, + } + + // Emit metrics. + metrics.HookStepsTotal.WithLabelValues(workflowName, blockName, step.Name, "failed").Inc() + + // Emit audit: hook step failed. + e.Auditor.Emit(ctx, audit.Event{ + Timestamp: time.Now(), + Actor: "engine", + Action: audit.ActionHookStepFailed, + Resource: audit.Resource{Type: "hook_step_execution", ID: step.Name}, + Metadata: map[string]string{ + "execution_id": execID, + "hook_block": blockName, + "step": step.Name, + "error": errMsg, + }, + }) + + // Update failed_in to track where the failure occurred. + celCtx.Execution["failed_in"] = blockName + + if step.ContinueOnError { + log.Printf("hooks: %s step %q failed (continue_on_error): %s", blockName, step.Name, errMsg) + continue + } + log.Printf("hooks: %s step %q failed, halting block: %s", blockName, step.Name, errMsg) + return + } + + // Success path. + if err := e.recordHookStep(ctx, execID, step.Name, blockName, "completed", output, ""); err != nil { + log.Printf("hooks: failed to record completion for %s/%s: %v", blockName, step.Name, err) + } + + // Update CEL context with output. + celCtx.Hooks[step.Name] = map[string]any{ + "output": output, + "error": nil, + } + + // Emit metrics. + metrics.HookStepsTotal.WithLabelValues(workflowName, blockName, step.Name, "completed").Inc() + + // Emit audit: hook step completed. + e.Auditor.Emit(ctx, audit.Event{ + Timestamp: time.Now(), + Actor: "engine", + Action: audit.ActionHookStepCompleted, + Resource: audit.Resource{Type: "hook_step_execution", ID: step.Name}, + Metadata: map[string]string{ + "execution_id": execID, + "hook_block": blockName, + "step": step.Name, + }, + }) + } +} + +// recordHookStep inserts or updates a hook step execution record in the database. +// The hook_block column disambiguates hook steps from main steps (and from each other), +// so step names are stored unprefixed. +func (e *Engine) recordHookStep(ctx context.Context, execID, stepName, hookBlock, status string, output map[string]any, errMsg string) error { + outputJSON, err := json.Marshal(output) + if err != nil { + return fmt.Errorf("marshaling hook step %s output: %w", stepName, err) + } + + var errorVal *string + if errMsg != "" { + errorVal = &errMsg + } + + _, err = e.DB.ExecContext(ctx, + `INSERT INTO step_executions (execution_id, step_name, attempt, status, output, error, hook_block, started_at) + VALUES ($1, $2, 1, $3, $4, $5, $6, NOW()) + ON CONFLICT (execution_id, step_name, attempt, hook_block) WHERE hook_block IS NOT NULL DO UPDATE + SET status = EXCLUDED.status, output = EXCLUDED.output, error = EXCLUDED.error, updated_at = NOW()`, + execID, stepName, status, outputJSON, errorVal, hookBlock, + ) + if err != nil { + return fmt.Errorf("recording hook step %s: %w", stepName, err) + } + return nil +} diff --git a/packages/engine/internal/engine/hooks_test.go b/packages/engine/internal/engine/hooks_test.go new file mode 100644 index 0000000..7c19e6d --- /dev/null +++ b/packages/engine/internal/engine/hooks_test.go @@ -0,0 +1,229 @@ +package engine + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" + + mantleCEL "github.com/dvflw/mantle/internal/cel" + "github.com/dvflw/mantle/internal/workflow" +) + +func TestExecuteHooks_NilHooksIsNoop(t *testing.T) { + database := setupTestDB(t) + + eng, err := New(database) + if err != nil { + t.Fatalf("New() error: %v", err) + } + + celCtx := &mantleCEL.Context{ + Steps: make(map[string]map[string]any), + Inputs: make(map[string]any), + } + + // wf.Hooks is nil — should return without panic. + wf := &workflow.Workflow{ + Name: "test-no-hooks", + Hooks: nil, + } + + execID := createTestExecution(t, database) + + // Should not panic. + eng.executeHooks(context.Background(), execID, "test-no-hooks", wf, "completed", "", "", celCtx, StepContext{}) +} + +func TestExecuteHooks_OnSuccessFiresOnCompletion(t *testing.T) { + database := setupTestDB(t) + + var requestCount atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + requestCount.Add(1) + w.WriteHeader(200) + json.NewEncoder(w).Encode(map[string]any{"notified": true}) + })) + defer server.Close() + + // Apply workflow with on_success hook. + wfYAML := []byte(`name: test-hooks-success +description: Test on_success hook +steps: + - name: main-step + action: http/request + params: + method: GET + url: "` + server.URL + `" +hooks: + on_success: + - name: notify-success + action: http/request + params: + method: GET + url: "` + server.URL + `" +`) + version := applyWorkflow(t, database, wfYAML) + + eng, err := New(database) + if err != nil { + t.Fatalf("New() error: %v", err) + } + + result, err := eng.Execute(context.Background(), "test-hooks-success", version, nil) + if err != nil { + t.Fatalf("Execute() error: %v", err) + } + + if result.Status != "completed" { + t.Errorf("status = %q, want %q (error: %s)", result.Status, "completed", result.Error) + } + if got := requestCount.Load(); got != 2 { + t.Errorf("request count = %d, want 2 (main step + on_success hook)", got) + } +} + +func TestExecuteHooks_OnFailureFiresOnFailure(t *testing.T) { + database := setupTestDB(t) + + var hookCalled atomic.Bool + hookServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + hookCalled.Store(true) + w.WriteHeader(200) + json.NewEncoder(w).Encode(map[string]any{"hook": "fired"}) + })) + defer hookServer.Close() + + failServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(500) + w.Write([]byte("fail")) + })) + defer failServer.Close() + + wfYAML := []byte(`name: test-hooks-failure +description: Test on_failure hook +steps: + - name: fail-step + action: http/request + params: + method: GET + url: "` + failServer.URL + `" +hooks: + on_failure: + - name: notify-failure + action: http/request + params: + method: GET + url: "` + hookServer.URL + `" +`) + version := applyWorkflow(t, database, wfYAML) + + eng, err := New(database) + if err != nil { + t.Fatalf("New() error: %v", err) + } + + result, err := eng.Execute(context.Background(), "test-hooks-failure", version, nil) + if err != nil { + t.Fatalf("Execute() error: %v", err) + } + + if result.Status != "failed" { + t.Errorf("status = %q, want %q", result.Status, "failed") + } + if !hookCalled.Load() { + t.Error("on_failure hook should have been called on failure") + } +} + +func TestExecuteHooks_OnFinishAlwaysRuns(t *testing.T) { + database := setupTestDB(t) + + var finishCallCount atomic.Int32 + finishServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + finishCallCount.Add(1) + w.WriteHeader(200) + json.NewEncoder(w).Encode(map[string]any{"finished": true}) + })) + defer finishServer.Close() + + successServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(200) + json.NewEncoder(w).Encode(map[string]any{"ok": true}) + })) + defer successServer.Close() + + failServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(500) + w.Write([]byte("fail")) + })) + defer failServer.Close() + + // Test 1: on_finish runs on success. + wfYAML := []byte(`name: test-hooks-finish-ok +description: Test on_finish runs on success +steps: + - name: ok-step + action: http/request + params: + method: GET + url: "` + successServer.URL + `" +hooks: + on_finish: + - name: cleanup + action: http/request + params: + method: GET + url: "` + finishServer.URL + `" +`) + version := applyWorkflow(t, database, wfYAML) + + eng, err := New(database) + if err != nil { + t.Fatalf("New() error: %v", err) + } + + result, err := eng.Execute(context.Background(), "test-hooks-finish-ok", version, nil) + if err != nil { + t.Fatalf("Execute() error: %v", err) + } + if result.Status != "completed" { + t.Errorf("status = %q, want %q", result.Status, "completed") + } + if got := finishCallCount.Load(); got != 1 { + t.Errorf("on_finish call count = %d, want 1 (on success)", got) + } + + // Test 2: on_finish runs on failure too. + finishCallCount.Store(0) + wfYAML2 := []byte(`name: test-hooks-finish-fail +description: Test on_finish runs on failure +steps: + - name: fail-step + action: http/request + params: + method: GET + url: "` + failServer.URL + `" +hooks: + on_finish: + - name: cleanup + action: http/request + params: + method: GET + url: "` + finishServer.URL + `" +`) + version2 := applyWorkflow(t, database, wfYAML2) + + result2, err := eng.Execute(context.Background(), "test-hooks-finish-fail", version2, nil) + if err != nil { + t.Fatalf("Execute() error: %v", err) + } + if result2.Status != "failed" { + t.Errorf("status = %q, want %q", result2.Status, "failed") + } + if got := finishCallCount.Load(); got != 1 { + t.Errorf("on_finish call count = %d, want 1 (on failure)", got) + } +} diff --git a/packages/engine/internal/engine/orchestrator.go b/packages/engine/internal/engine/orchestrator.go index f060faf..de29787 100644 --- a/packages/engine/internal/engine/orchestrator.go +++ b/packages/engine/internal/engine/orchestrator.go @@ -104,7 +104,7 @@ func (o *Orchestrator) CreatePendingSteps(ctx context.Context, executionID strin stmt, err := tx.PrepareContext(ctx, `INSERT INTO step_executions (execution_id, step_name, attempt, status, max_attempts) VALUES ($1, $2, 1, 'pending', $3) - ON CONFLICT (execution_id, step_name, attempt) DO NOTHING`, + ON CONFLICT (execution_id, step_name, attempt) WHERE hook_block IS NULL DO NOTHING`, ) if err != nil { return fmt.Errorf("preparing statement: %w", err) @@ -132,7 +132,7 @@ func (o *Orchestrator) CreateRetryStep(ctx context.Context, executionID, stepNam _, err := o.DB.ExecContext(ctx, `INSERT INTO step_executions (execution_id, step_name, attempt, status, max_attempts) VALUES ($1, $2, $3, 'pending', $4) - ON CONFLICT (execution_id, step_name, attempt) DO NOTHING`, + ON CONFLICT (execution_id, step_name, attempt) WHERE hook_block IS NULL DO NOTHING`, executionID, stepName, nextAttempt, maxAttempts, ) if err != nil { @@ -147,7 +147,7 @@ func (o *Orchestrator) GetStepStatuses(ctx context.Context, executionID string) rows, err := o.DB.QueryContext(ctx, `SELECT DISTINCT ON (step_name) step_name, status, attempt, max_attempts, output, error FROM step_executions - WHERE execution_id = $1 AND parent_step_id IS NULL + WHERE execution_id = $1 AND parent_step_id IS NULL AND hook_block IS NULL ORDER BY step_name, attempt DESC`, executionID, ) diff --git a/packages/engine/internal/engine/retry.go b/packages/engine/internal/engine/retry.go new file mode 100644 index 0000000..e14aa86 --- /dev/null +++ b/packages/engine/internal/engine/retry.go @@ -0,0 +1,277 @@ +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" +) + +// RetryExecution creates a new execution that resumes from the failure point of +// a previous execution. If fromStep is provided, execution resumes from that +// step; otherwise, the first failed step (in topological order) is used. +// If force is false, concurrency limits are checked before creating the execution; +// if limits are hit the execution is queued instead of started immediately. +func (e *Engine) RetryExecution(ctx context.Context, originalExecID string, fromStep string, force bool) (*ExecutionResult, error) { + teamID := auth.TeamIDFromContext(ctx) + + // 1. Load original execution metadata. + var workflowName string + var version int + var inputsJSON []byte + err := e.DB.QueryRowContext(ctx, + `SELECT workflow_name, workflow_version, inputs + FROM workflow_executions + WHERE id = $1 AND team_id = $2`, + originalExecID, teamID, + ).Scan(&workflowName, &version, &inputsJSON) + if err == sql.ErrNoRows { + return nil, fmt.Errorf("execution %q not found", originalExecID) + } + if err != nil { + return nil, fmt.Errorf("loading original execution: %w", err) + } + + var inputs map[string]any + if len(inputsJSON) > 0 { + if err := json.Unmarshal(inputsJSON, &inputs); err != nil { + return nil, fmt.Errorf("unmarshaling inputs: %w", err) + } + } + + // 2. Load the workflow definition (same version as original). + wf, err := e.loadWorkflow(ctx, workflowName, version) + if err != nil { + return nil, fmt.Errorf("loading workflow: %w", err) + } + + // 3. Determine retry point. + retryStep := fromStep + if retryStep == "" { + statuses, err := loadStepStatuses(ctx, e.DB, originalExecID) + if err != nil { + return nil, fmt.Errorf("loading step statuses: %w", err) + } + for _, step := range wf.Steps { + if s, ok := statuses[step.Name]; ok && s == "failed" { + retryStep = step.Name + break + } + } + if retryStep == "" { + return nil, fmt.Errorf("no failed step found in execution %q", originalExecID) + } + } else { + // Validate that the specified step exists. + found := false + for _, step := range wf.Steps { + if step.Name == retryStep { + found = true + break + } + } + if !found { + return nil, fmt.Errorf("step %q not found in workflow %q", retryStep, workflowName) + } + } + + // 4. Find all upstream steps (ancestors of the retry point). + upstream := findUpstream(wf.Steps, retryStep) + + // 5. Check concurrency limits (unless force-bypassed). + // When limits apply, check + insert happen in the same transaction + // to prevent a TOCTOU race on the advisory lock. + var newExecID string + initialStatus := "pending" + if !force && (wf.MaxParallelExecutions > 0 || e.MaxConcurrentExecutionsPerTeam > 0) { + tx, txErr := e.DB.BeginTx(ctx, nil) + if txErr != nil { + return nil, fmt.Errorf("starting concurrency check transaction: %w", txErr) + } + defer tx.Rollback() + + cr := CheckConcurrencyLimits(ctx, tx, workflowName, + wf.MaxParallelExecutions, wf.OnLimit, e.MaxConcurrentExecutionsPerTeam) + + if cr.Err != nil { + return nil, cr.Err + } + if cr.Queued { + initialStatus = "queued" + } + + // Insert execution row inside the same transaction. + newExecID, err = e.createExecutionTx(ctx, tx, workflowName, version, inputs, initialStatus) + if err != nil { + return nil, fmt.Errorf("creating retry execution: %w", err) + } + + if err := tx.Commit(); err != nil { + return nil, fmt.Errorf("committing concurrency check: %w", err) + } + } else { + // No concurrency limits — use implicit transaction. + newExecID, err = e.createExecution(ctx, workflowName, version, inputs, initialStatus) + if err != nil { + return nil, fmt.Errorf("creating retry execution: %w", err) + } + } + + // 7. Link new execution to original, copy upstream steps, and emit audit — + // all in a single transaction so the audit record is atomic with the mutation. + linkTx, err := e.DB.BeginTx(ctx, nil) + if err != nil { + return nil, fmt.Errorf("starting link transaction: %w", err) + } + defer linkTx.Rollback() + + _, err = linkTx.ExecContext(ctx, + `UPDATE workflow_executions SET retried_from_execution_id = $1 WHERE id = $2 AND team_id = $3`, + originalExecID, newExecID, teamID, + ) + if err != nil { + return nil, fmt.Errorf("linking retry execution: %w", err) + } + + // 8. Copy completed upstream step outputs from original (exclude hook steps). + upstreamSet := make(map[string]bool, len(upstream)) + for _, name := range upstream { + upstreamSet[name] = true + } + + rows, err := linkTx.QueryContext(ctx, + `SELECT DISTINCT ON (step_name) step_name, status, output, error + FROM step_executions + WHERE execution_id = $1 AND hook_block IS NULL + ORDER BY step_name, attempt DESC`, + originalExecID, + ) + if err != nil { + return nil, fmt.Errorf("loading original steps: %w", err) + } + defer rows.Close() + + for rows.Next() { + var stepName, status string + var outputJSON []byte + var stepErr *string + if err := rows.Scan(&stepName, &status, &outputJSON, &stepErr); err != nil { + return nil, fmt.Errorf("scanning step: %w", err) + } + // Only copy completed upstream steps. + if !upstreamSet[stepName] || status != "completed" { + continue + } + errVal := "" + if stepErr != nil { + errVal = *stepErr + } + var errPtr *string + if errVal != "" { + errPtr = &errVal + } + _, err := linkTx.ExecContext(ctx, + `INSERT INTO step_executions (execution_id, step_name, attempt, status, output, error, started_at) + VALUES ($1, $2, 1, $3, $4, $5, NOW()) + ON CONFLICT (execution_id, step_name, attempt) WHERE hook_block IS NULL DO NOTHING`, + newExecID, stepName, status, outputJSON, errPtr, + ) + if err != nil { + return nil, fmt.Errorf("copying step %q: %w", stepName, err) + } + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("iterating original steps: %w", err) + } + + // 9. Emit audit event inside the same transaction. + if err := audit.EmitTx(ctx, linkTx, audit.Event{ + Timestamp: time.Now(), + Actor: "engine", + Action: audit.ActionExecutionRetried, + Resource: audit.Resource{Type: "workflow_execution", ID: newExecID}, + Metadata: map[string]string{ + "original_execution_id": originalExecID, + "from_step": retryStep, + "workflow": workflowName, + "version": fmt.Sprintf("%d", version), + }, + }); err != nil { + return nil, fmt.Errorf("emitting retry audit event: %w", err) + } + + if err := linkTx.Commit(); err != nil { + return nil, fmt.Errorf("committing retry link: %w", err) + } + + // 10. If queued, return immediately with queued status. + if initialStatus == "queued" { + return &ExecutionResult{ + ExecutionID: newExecID, + Status: "queued", + }, nil + } + + // 11. Run the new execution (resumeExecution will skip already-completed steps). + return e.resumeExecution(ctx, newExecID, workflowName, version, inputs) +} + +// findUpstream returns the names of all steps that are strictly upstream +// (ancestors) of the target step in the dependency graph. +func findUpstream(steps []workflow.Step, target string) []string { + // Build dependency map. + deps := make(map[string][]string, len(steps)) + for _, s := range steps { + deps[s.Name] = s.DependsOn + } + + // Walk backwards from target, collecting all ancestors. + visited := make(map[string]bool) + var walk func(name string) + walk = func(name string) { + for _, dep := range deps[name] { + if !visited[dep] { + visited[dep] = true + walk(dep) + } + } + } + walk(target) + + result := make([]string, 0, len(visited)) + for name := range visited { + result = append(result, name) + } + return result +} + +// loadStepStatuses loads the latest status per main step (excluding hooks) +// for a given execution. +func loadStepStatuses(ctx context.Context, db *sql.DB, execID string) (map[string]string, error) { + rows, err := db.QueryContext(ctx, + `SELECT DISTINCT ON (step_name) step_name, status + FROM step_executions + WHERE execution_id = $1 AND hook_block IS NULL + ORDER BY step_name, attempt DESC`, + execID, + ) + if err != nil { + return nil, err + } + defer rows.Close() + + result := make(map[string]string) + for rows.Next() { + var name, status string + if err := rows.Scan(&name, &status); err != nil { + return nil, err + } + result[name] = status + } + return result, rows.Err() +} diff --git a/packages/engine/internal/engine/retry_test.go b/packages/engine/internal/engine/retry_test.go new file mode 100644 index 0000000..df644bb --- /dev/null +++ b/packages/engine/internal/engine/retry_test.go @@ -0,0 +1,107 @@ +package engine + +import ( + "context" + "sort" + "testing" + + "github.com/dvflw/mantle/internal/workflow" +) + +func TestFindUpstream_LinearChain(t *testing.T) { + // A → B → C + steps := []workflow.Step{ + {Name: "A"}, + {Name: "B", DependsOn: []string{"A"}}, + {Name: "C", DependsOn: []string{"B"}}, + } + + upstream := findUpstream(steps, "C") + sort.Strings(upstream) + + expected := []string{"A", "B"} + if len(upstream) != len(expected) { + t.Fatalf("findUpstream(C) = %v, want %v", upstream, expected) + } + for i, name := range expected { + if upstream[i] != name { + t.Errorf("upstream[%d] = %q, want %q", i, upstream[i], name) + } + } +} + +func TestFindUpstream_Diamond(t *testing.T) { + // A → B, A → C, B → D, C → D + steps := []workflow.Step{ + {Name: "A"}, + {Name: "B", DependsOn: []string{"A"}}, + {Name: "C", DependsOn: []string{"A"}}, + {Name: "D", DependsOn: []string{"B", "C"}}, + } + + upstream := findUpstream(steps, "D") + sort.Strings(upstream) + + expected := []string{"A", "B", "C"} + if len(upstream) != len(expected) { + t.Fatalf("findUpstream(D) = %v, want %v", upstream, expected) + } + for i, name := range expected { + if upstream[i] != name { + t.Errorf("upstream[%d] = %q, want %q", i, upstream[i], name) + } + } +} + +func TestFindUpstream_NoUpstream(t *testing.T) { + steps := []workflow.Step{ + {Name: "root"}, + {Name: "child", DependsOn: []string{"root"}}, + } + + upstream := findUpstream(steps, "root") + if len(upstream) != 0 { + t.Errorf("findUpstream(root) = %v, want empty slice", upstream) + } +} + +func TestLoadStepStatuses_ExcludesHookSteps(t *testing.T) { + database := setupTestDB(t) + ctx := context.Background() + + execID := createTestExecution(t, database) + + // Insert a main step (hook_block IS NULL). + _, err := database.ExecContext(ctx, + `INSERT INTO step_executions (execution_id, step_name, attempt, status, started_at) + VALUES ($1, 'main-step', 1, 'completed', NOW())`, + execID) + if err != nil { + t.Fatalf("inserting main step: %v", err) + } + + // Insert a hook step (hook_block = 'on_success'). + _, err = database.ExecContext(ctx, + `INSERT INTO step_executions (execution_id, step_name, attempt, status, hook_block, started_at) + VALUES ($1, 'hook-step', 1, 'completed', 'on_success', NOW())`, + execID) + if err != nil { + t.Fatalf("inserting hook step: %v", err) + } + + statuses, err := loadStepStatuses(ctx, database, execID) + if err != nil { + t.Fatalf("loadStepStatuses() error: %v", err) + } + + // Should contain main-step but not hook-step. + if _, ok := statuses["main-step"]; !ok { + t.Error("expected main-step in statuses, not found") + } + if statuses["main-step"] != "completed" { + t.Errorf("main-step status = %q, want %q", statuses["main-step"], "completed") + } + if _, ok := statuses["hook-step"]; ok { + t.Error("hook-step should be excluded from statuses but was found") + } +} diff --git a/packages/engine/internal/engine/toolsteps.go b/packages/engine/internal/engine/toolsteps.go index ad677de..0ce4318 100644 --- a/packages/engine/internal/engine/toolsteps.go +++ b/packages/engine/internal/engine/toolsteps.go @@ -21,7 +21,7 @@ func (ts *ToolSteps) CreateSubStep(ctx context.Context, executionID, parentStepI err := ts.DB.QueryRowContext(ctx, ` INSERT INTO step_executions (execution_id, step_name, attempt, status, parent_step_id) VALUES ($1, $2, 1, 'pending', $3) - ON CONFLICT (execution_id, step_name, attempt) DO NOTHING + ON CONFLICT (execution_id, step_name, attempt) WHERE hook_block IS NULL DO NOTHING RETURNING id `, executionID, stepName, parentStepID).Scan(&id) @@ -29,7 +29,7 @@ func (ts *ToolSteps) CreateSubStep(ctx context.Context, executionID, parentStepI // Row already existed — fetch the existing ID. err = ts.DB.QueryRowContext(ctx, ` SELECT id FROM step_executions - WHERE execution_id = $1 AND step_name = $2 AND attempt = 1 + WHERE execution_id = $1 AND step_name = $2 AND attempt = 1 AND hook_block IS NULL `, executionID, stepName).Scan(&id) if err != nil { return "", fmt.Errorf("fetching existing sub-step: %w", err) diff --git a/packages/engine/internal/metrics/metrics.go b/packages/engine/internal/metrics/metrics.go index 6781a33..13d856e 100644 --- a/packages/engine/internal/metrics/metrics.go +++ b/packages/engine/internal/metrics/metrics.go @@ -45,6 +45,14 @@ var ( }, []string{"workflow", "step"}) ) +// Hook metrics. +var ( + HookStepsTotal = promauto.NewCounterVec(prometheus.CounterOpts{ + Name: "mantle_hook_steps_total", + Help: "Total hook step executions", + }, []string{"workflow", "hook", "step", "status"}) +) + // Queue and distribution metrics. var ( QueueDepth = promauto.NewGauge(prometheus.GaugeOpts{ @@ -157,6 +165,30 @@ var ( }) ) +// Concurrency control metrics. +var ( + ExecutionsQueued = promauto.NewGaugeVec(prometheus.GaugeOpts{ + Name: "mantle_executions_queued", + Help: "Current queue depth per workflow", + }, []string{"workflow"}) + + ExecutionsRejectedTotal = promauto.NewCounterVec(prometheus.CounterOpts{ + Name: "mantle_executions_rejected_total", + Help: "Executions rejected due to concurrency limit", + }, []string{"workflow"}) + + QueueWaitDuration = promauto.NewHistogramVec(prometheus.HistogramOpts{ + Name: "mantle_queue_wait_duration_seconds", + Help: "Time spent in queue before promotion", + Buckets: []float64{0.1, 0.5, 1, 5, 10, 30, 60, 120, 300}, + }, []string{"workflow"}) + + PromotionFailuresTotal = promauto.NewCounterVec(prometheus.CounterOpts{ + Name: "mantle_promotion_failures_total", + Help: "Failed queue promotion attempts", + }, []string{"workflow", "scope"}) +) + // Budget metrics. var ( BudgetCheckTotal = promauto.NewCounterVec(prometheus.CounterOpts{ diff --git a/packages/engine/internal/secret/store.go b/packages/engine/internal/secret/store.go index 6b19cfe..d6ebd00 100644 --- a/packages/engine/internal/secret/store.go +++ b/packages/engine/internal/secret/store.go @@ -31,15 +31,28 @@ func (s *Store) Create(ctx context.Context, name, typeName string, data map[stri return nil, fmt.Errorf("marshaling credential data: %w", err) } + teamID := auth.TeamIDFromContext(ctx) + + // Use a transaction with the same advisory lock as RotateAll to serialize + // credential writes with key rotation. Encryption happens inside the lock + // to prevent a concurrent RotateAll from changing keys between encrypt and insert. + tx, err := s.DB.BeginTx(ctx, nil) + if err != nil { + return nil, fmt.Errorf("starting transaction: %w", err) + } + defer tx.Rollback() + + if _, err := tx.ExecContext(ctx, "SELECT pg_advisory_xact_lock($1)", int64(0x4D414E544C45)); err != nil { + return nil, fmt.Errorf("acquiring advisory lock: %w", err) + } + ciphertext, nonce, err := s.Encryptor.Encrypt(plaintext) if err != nil { return nil, fmt.Errorf("encrypting credential: %w", err) } - teamID := auth.TeamIDFromContext(ctx) - var cred Credential - err = s.DB.QueryRowContext(ctx, + err = tx.QueryRowContext(ctx, `INSERT INTO credentials (name, type, encrypted_data, nonce, team_id) VALUES ($1, $2, $3, $4, $5) RETURNING id, name, type, created_at, updated_at`, @@ -49,6 +62,10 @@ func (s *Store) Create(ctx context.Context, name, typeName string, data map[stri return nil, fmt.Errorf("storing credential: %w", err) } + if err := tx.Commit(); err != nil { + return nil, fmt.Errorf("committing credential: %w", err) + } + return &cred, nil } @@ -123,6 +140,9 @@ func (s *Store) Delete(ctx context.Context, name string) error { // ReEncryptAll decrypts all credentials with the current encryptor and // re-encrypts them with a new encryptor. Used for key rotation. +// +// Deprecated: Use the package-level RotateAll function instead, which accepts +// a caller-managed transaction and operates globally across all teams. func (s *Store) ReEncryptAll(ctx context.Context, newEncryptor *Encryptor) (int, error) { tx, err := s.DB.BeginTx(ctx, nil) if err != nil { @@ -186,3 +206,64 @@ func (s *Store) ReEncryptAll(ctx context.Context, newEncryptor *Encryptor) (int, return len(toUpdate), nil } + +// RotateAll decrypts all credentials with the old encryptor and re-encrypts +// them with the new encryptor. Operates globally across all teams. +// The caller provides an open transaction and is responsible for commit/rollback. +func RotateAll(ctx context.Context, tx *sql.Tx, oldEncryptor, newEncryptor *Encryptor) (int, error) { + // Acquire advisory lock to serialize with credential writers. + // The fixed key 0x4D414E544C45 ("MANTLE" in ASCII) is used for all credential operations. + if _, err := tx.ExecContext(ctx, "SELECT pg_advisory_xact_lock($1)", int64(0x4D414E544C45)); err != nil { + return 0, fmt.Errorf("acquiring advisory lock: %w", err) + } + + rows, err := tx.QueryContext(ctx, + `SELECT id, name, encrypted_data, nonce FROM credentials FOR UPDATE`) + if err != nil { + return 0, fmt.Errorf("querying credentials: %w", err) + } + + type row struct { + id string + name string + data []byte + nonce []byte + } + var toUpdate []row + for rows.Next() { + var r row + if err := rows.Scan(&r.id, &r.name, &r.data, &r.nonce); err != nil { + rows.Close() + return 0, fmt.Errorf("scanning credential: %w", err) + } + toUpdate = append(toUpdate, r) + } + rows.Close() + if err := rows.Err(); err != nil { + return 0, err + } + + for _, r := range toUpdate { + // Decrypt with old key. + plaintext, err := oldEncryptor.Decrypt(r.data, r.nonce) + if err != nil { + return 0, fmt.Errorf("decrypting credential %q (id: %s): %w", r.name, r.id, err) + } + + // Re-encrypt with new key. + newData, newNonce, err := newEncryptor.Encrypt(plaintext) + if err != nil { + return 0, fmt.Errorf("re-encrypting credential %q (id: %s): %w", r.name, r.id, err) + } + + _, err = tx.ExecContext(ctx, + `UPDATE credentials SET encrypted_data = $1, nonce = $2, updated_at = $3 WHERE id = $4`, + newData, newNonce, time.Now(), r.id, + ) + if err != nil { + return 0, fmt.Errorf("updating credential %q (id: %s): %w", r.name, r.id, err) + } + } + + return len(toUpdate), nil +} diff --git a/packages/engine/internal/secret/store_test.go b/packages/engine/internal/secret/store_test.go index d50cf4f..c4a314c 100644 --- a/packages/engine/internal/secret/store_test.go +++ b/packages/engine/internal/secret/store_test.go @@ -256,5 +256,187 @@ func containsBytes(haystack, needle []byte) bool { return false } +func TestRotateAll_ReEncryptsAllCredentials(t *testing.T) { + store := setupTestStore(t) + ctx := context.Background() + + store.Create(ctx, "cred-1", "generic", map[string]string{"key": "secret-1"}) + store.Create(ctx, "cred-2", "bearer", map[string]string{"token": "secret-2"}) + + oldEncryptor := store.Encryptor + + newEnc, err := NewEncryptor(testKey(t)) + if err != nil { + t.Fatalf("NewEncryptor() error: %v", err) + } + + tx, err := store.DB.BeginTx(ctx, nil) + if err != nil { + t.Fatalf("BeginTx() error: %v", err) + } + defer tx.Rollback() + + count, err := RotateAll(ctx, tx, oldEncryptor, newEnc) + if err != nil { + t.Fatalf("RotateAll() error: %v", err) + } + if count != 2 { + t.Fatalf("RotateAll() count = %d, want 2", count) + } + + if err := tx.Commit(); err != nil { + t.Fatalf("Commit() error: %v", err) + } + + // Old encryptor should no longer work. + _, err = store.Get(ctx, "cred-1") + if err == nil { + t.Error("Get() with old encryptor should fail after rotation") + } + + // Switch to new encryptor and verify both credentials. + store.Encryptor = newEnc + data, err := store.Get(ctx, "cred-1") + if err != nil { + t.Fatalf("Get() with new encryptor error: %v", err) + } + if data["key"] != "secret-1" { + t.Errorf("key = %q, want %q", data["key"], "secret-1") + } + + data2, err := store.Get(ctx, "cred-2") + if err != nil { + t.Fatalf("Get() cred-2 error: %v", err) + } + if data2["token"] != "secret-2" { + t.Errorf("token = %q, want %q", data2["token"], "secret-2") + } +} + +func TestRotateAll_RollbackOnDecryptFailure(t *testing.T) { + store := setupTestStore(t) + ctx := context.Background() + + store.Create(ctx, "cred-1", "generic", map[string]string{"key": "val-1"}) + + // Use a wrong encryptor as the "old" key — decryption should fail. + wrongEnc, err := NewEncryptor(testKey(t)) + if err != nil { + t.Fatalf("NewEncryptor() error: %v", err) + } + newEnc, err := NewEncryptor(testKey(t)) + if err != nil { + t.Fatalf("NewEncryptor() error: %v", err) + } + + tx, err := store.DB.BeginTx(ctx, nil) + if err != nil { + t.Fatalf("BeginTx() error: %v", err) + } + defer tx.Rollback() + + _, err = RotateAll(ctx, tx, wrongEnc, newEnc) + if err == nil { + t.Fatal("RotateAll() with wrong old key should fail") + } + + // Original credentials should still be intact with the original encryptor. + data, err := store.Get(ctx, "cred-1") + if err != nil { + t.Fatalf("Get() after rollback error: %v", err) + } + if data["key"] != "val-1" { + t.Errorf("key = %q, want %q after rollback", data["key"], "val-1") + } +} + +func TestRotateAll_GlobalScope(t *testing.T) { + store := setupTestStore(t) + ctx := context.Background() + + // Insert credentials with different team_ids directly via SQL + // to verify RotateAll operates across all teams. + plaintext1 := []byte(`{"key":"team-a-secret"}`) + ct1, nonce1, err := store.Encryptor.Encrypt(plaintext1) + if err != nil { + t.Fatalf("Encrypt() error: %v", err) + } + plaintext2 := []byte(`{"key":"team-b-secret"}`) + ct2, nonce2, err := store.Encryptor.Encrypt(plaintext2) + if err != nil { + t.Fatalf("Encrypt() error: %v", err) + } + + // Use the default team (created by migration) for team A. + teamA := "00000000-0000-0000-0000-000000000001" + // Create a second team for team B. + teamB := "00000000-0000-0000-0000-000000000002" + _, err = store.DB.ExecContext(ctx, + `INSERT INTO teams (id, name) VALUES ($1, $2)`, teamB, "team-b") + if err != nil { + t.Fatalf("insert team-b: %v", err) + } + + _, err = store.DB.ExecContext(ctx, + `INSERT INTO credentials (name, type, encrypted_data, nonce, team_id) VALUES ($1, $2, $3, $4, $5)`, + "cred-team-a", "generic", ct1, nonce1, teamA) + if err != nil { + t.Fatalf("insert team-a credential: %v", err) + } + _, err = store.DB.ExecContext(ctx, + `INSERT INTO credentials (name, type, encrypted_data, nonce, team_id) VALUES ($1, $2, $3, $4, $5)`, + "cred-team-b", "generic", ct2, nonce2, teamB) + if err != nil { + t.Fatalf("insert team-b credential: %v", err) + } + + oldEncryptor := store.Encryptor + newEnc, err := NewEncryptor(testKey(t)) + if err != nil { + t.Fatalf("NewEncryptor() error: %v", err) + } + + tx, err := store.DB.BeginTx(ctx, nil) + if err != nil { + t.Fatalf("BeginTx() error: %v", err) + } + defer tx.Rollback() + + count, err := RotateAll(ctx, tx, oldEncryptor, newEnc) + if err != nil { + t.Fatalf("RotateAll() error: %v", err) + } + if count != 2 { + t.Fatalf("RotateAll() count = %d, want 2 (both teams)", count) + } + + if err := tx.Commit(); err != nil { + t.Fatalf("Commit() error: %v", err) + } + + // Verify both team credentials are re-encrypted with new key. + for _, name := range []string{"cred-team-a", "cred-team-b"} { + var encData, nonce []byte + err := store.DB.QueryRowContext(ctx, + `SELECT encrypted_data, nonce FROM credentials WHERE name = $1`, name, + ).Scan(&encData, &nonce) + if err != nil { + t.Fatalf("query %s: %v", name, err) + } + + // Should decrypt with new encryptor. + _, err = newEnc.Decrypt(encData, nonce) + if err != nil { + t.Errorf("Decrypt(%s) with new key failed: %v", name, err) + } + + // Should NOT decrypt with old encryptor. + _, err = oldEncryptor.Decrypt(encData, nonce) + if err == nil { + t.Errorf("Decrypt(%s) with old key should fail after rotation", name) + } + } +} + // Ensure the store variable satisfies the need for *sql.DB (compile check). var _ *sql.DB diff --git a/packages/engine/internal/server/server.go b/packages/engine/internal/server/server.go index 6008f40..aa70d2a 100644 --- a/packages/engine/internal/server/server.go +++ b/packages/engine/internal/server/server.go @@ -210,16 +210,16 @@ func (s *Server) Start(ctx context.Context) error { } // Start artifact reaper if the artifact subsystem is configured. - if s.Engine.ArtifactStore != nil && s.Engine.TmpStorage != nil { + if s.Engine.ArtifactStore != nil && s.Engine.Storage != nil { retention := 24 * time.Hour // default - if cfg.Tmp.Retention != "" { - if d, err := time.ParseDuration(cfg.Tmp.Retention); err == nil && d > 0 { + if cfg.Storage.Retention != "" { + if d, err := time.ParseDuration(cfg.Storage.Retention); err == nil && d > 0 { retention = d } } s.artifactReaper = &artifact.Reaper{ - Store: s.Engine.ArtifactStore, - TmpStorage: s.Engine.TmpStorage, + Store: s.Engine.ArtifactStore, + Storage: s.Engine.Storage, Retention: retention, Logger: s.Logger, } diff --git a/packages/engine/internal/workflow/validate.go b/packages/engine/internal/workflow/validate.go index 7f66dc5..f98b59c 100644 --- a/packages/engine/internal/workflow/validate.go +++ b/packages/engine/internal/workflow/validate.go @@ -85,6 +85,25 @@ func Validate(result *ParseResult) []ValidationError { } } + // Validate max_parallel_executions. + if w.MaxParallelExecutions < 0 { + line, col := findFieldPosition(root, "max_parallel_executions") + errs = append(errs, ValidationError{ + Line: line, Column: col, Field: "max_parallel_executions", + Message: fmt.Sprintf("max_parallel_executions must be >= 0, got %d", w.MaxParallelExecutions), + }) + } + + // Validate on_limit. + validOnLimitValues := map[string]bool{"": true, "queue": true, "reject": true} + if !validOnLimitValues[w.OnLimit] { + line, col := findFieldPosition(root, "on_limit") + errs = append(errs, ValidationError{ + Line: line, Column: col, Field: "on_limit", + Message: fmt.Sprintf("on_limit must be one of: queue, reject (got %q)", w.OnLimit), + }) + } + // Validate workflow-level timeout. if w.Timeout != "" { d, err := time.ParseDuration(w.Timeout) @@ -216,6 +235,14 @@ func Validate(result *ParseResult) []ValidationError { } } + // Validate max_parallel. + if step.MaxParallel < 0 { + errs = append(errs, ValidationError{ + Field: prefix + ".max_parallel", + Message: fmt.Sprintf("max_parallel must be >= 0, got %d", step.MaxParallel), + }) + } + // Validate params for browser/run steps. if step.Action == "browser/run" && step.Params != nil { // Validate script is present and non-empty. @@ -377,6 +404,11 @@ func Validate(result *ParseResult) []ValidationError { } } + // Validate hooks. + if w.Hooks != nil { + errs = append(errs, validateHooks(w.Hooks)...) + } + // Validate CEL expression syntax in step params and if conditions. celEval, celErr := mantleCEL.NewEvaluator() if celErr == nil { @@ -396,6 +428,13 @@ func Validate(result *ParseResult) []ValidationError { // Check params recursively. errs = append(errs, validateCELInParams(celEval, step.Params, prefix+".params")...) } + + // Validate CEL expressions in hook steps. + if w.Hooks != nil { + errs = append(errs, validateCELInHookBlock(celEval, w.Hooks.OnSuccess, "hooks.on_success")...) + errs = append(errs, validateCELInHookBlock(celEval, w.Hooks.OnFailure, "hooks.on_failure")...) + errs = append(errs, validateCELInHookBlock(celEval, w.Hooks.OnFinish, "hooks.on_finish")...) + } } return errs @@ -476,6 +515,26 @@ func toInt(v any) (int, bool) { return 0, false } +// validateCELInHookBlock validates CEL expressions in a slice of hook steps. +func validateCELInHookBlock(eval *mantleCEL.Evaluator, steps []Step, blockPrefix string) []ValidationError { + var errs []ValidationError + for i, step := range steps { + prefix := fmt.Sprintf("%s[%d]", blockPrefix, i) + + if step.If != "" { + if err := eval.CompileCheck(step.If); err != nil { + errs = append(errs, ValidationError{ + Field: prefix + ".if", + Message: fmt.Sprintf("invalid CEL expression: %v", err), + }) + } + } + + errs = append(errs, validateCELInParams(eval, step.Params, prefix+".params")...) + } + return errs +} + // validateCELInParams recursively walks a params map and validates any CEL // expressions found inside {{ }} delimiters. func validateCELInParams(eval *mantleCEL.Evaluator, params map[string]any, prefix string) []ValidationError { @@ -535,6 +594,115 @@ func extractCELExpressions(s string) []string { return exprs } +// validateHooks validates the hooks configuration block. +func validateHooks(hooks *HooksConfig) []ValidationError { + var errs []ValidationError + + // Validate hooks timeout. + if hooks.Timeout != "" { + d, err := time.ParseDuration(hooks.Timeout) + if err != nil { + errs = append(errs, ValidationError{ + Field: "hooks.timeout", + Message: fmt.Sprintf("invalid duration: %v", err), + }) + } else if d <= 0 { + errs = append(errs, ValidationError{ + Field: "hooks.timeout", + Message: "timeout must be a positive duration", + }) + } + } + + // Validate each hook block. Step names must be unique across all hook blocks + // because they share the same CEL namespace (hooks.). + crossBlockSeen := make(map[string]string) // step name -> block that first declared it + errs = append(errs, validateHookSteps(hooks.OnSuccess, "hooks.on_success", crossBlockSeen)...) + errs = append(errs, validateHookSteps(hooks.OnFailure, "hooks.on_failure", crossBlockSeen)...) + errs = append(errs, validateHookSteps(hooks.OnFinish, "hooks.on_finish", crossBlockSeen)...) + + return errs +} + +// validateHookSteps validates a slice of hook steps within a single block. +// crossBlockSeen tracks step names across all hook blocks for cross-block uniqueness. +func validateHookSteps(steps []Step, blockPrefix string, crossBlockSeen map[string]string) []ValidationError { + var errs []ValidationError + + for i, step := range steps { + prefix := fmt.Sprintf("%s[%d]", blockPrefix, i) + + if step.Name == "" { + errs = append(errs, ValidationError{ + Field: prefix + ".name", + Message: "hook step name is required", + }) + } else { + if !namePattern.MatchString(step.Name) { + errs = append(errs, ValidationError{ + Field: prefix + ".name", + Message: "hook step name must match ^[a-z][a-z0-9-]*$", + }) + } + if prevBlock, exists := crossBlockSeen[step.Name]; exists { + errs = append(errs, ValidationError{ + Field: prefix + ".name", + Message: fmt.Sprintf("duplicate hook step name %q (also declared in %s)", step.Name, prevBlock), + }) + } + crossBlockSeen[step.Name] = blockPrefix + } + + if step.Action == "" { + errs = append(errs, ValidationError{ + Field: prefix + ".action", + Message: "hook step action is required", + }) + } + + // Validate timeout. + if step.Timeout != "" { + d, err := time.ParseDuration(step.Timeout) + if err != nil { + errs = append(errs, ValidationError{ + Field: prefix + ".timeout", + Message: fmt.Sprintf("invalid duration: %v", err), + }) + } else if d <= 0 { + errs = append(errs, ValidationError{ + Field: prefix + ".timeout", + Message: "timeout must be a positive duration", + }) + } + } + + // Validate retry policy. + if step.Retry != nil { + if step.Retry.MaxAttempts <= 0 { + errs = append(errs, ValidationError{ + Field: prefix + ".retry.max_attempts", + Message: "max_attempts must be greater than 0", + }) + } + if step.Retry.Backoff != "" && !validBackoffTypes[step.Retry.Backoff] { + errs = append(errs, ValidationError{ + Field: prefix + ".retry.backoff", + Message: fmt.Sprintf("backoff must be one of: fixed, exponential (got %q)", step.Retry.Backoff), + }) + } + } + + if len(step.DependsOn) > 0 { + errs = append(errs, ValidationError{ + Field: prefix + ".depends_on", + Message: "hook steps do not support depends_on — use a child workflow for complex error handling", + }) + } + } + + return errs +} + // findFieldPosition searches the root mapping node for a top-level key and // returns its line and column. Falls back to (0, 0) if not found. func findFieldPosition(root *yaml.Node, field string) (int, int) { diff --git a/packages/engine/internal/workflow/validate_hooks_test.go b/packages/engine/internal/workflow/validate_hooks_test.go new file mode 100644 index 0000000..c2a9000 --- /dev/null +++ b/packages/engine/internal/workflow/validate_hooks_test.go @@ -0,0 +1,259 @@ +package workflow + +import ( + "strings" + "testing" +) + +func TestValidate_HooksNoDependsOn(t *testing.T) { + result := mustParse(t, ` +name: my-workflow +steps: + - name: do-work + action: http/request + params: + url: "https://example.com" +hooks: + on_success: + - name: notify + action: slack/send + depends_on: + - do-work + params: + channel: "#alerts" + text: "done" +`) + errs := Validate(result) + found := false + for _, e := range errs { + if strings.Contains(e.Message, "hook steps do not support depends_on") { + found = true + } + } + if !found { + t.Errorf("expected error about hook steps not supporting depends_on, got: %v", errs) + } +} + +func TestValidate_HookStepNameUniqueness(t *testing.T) { + result := mustParse(t, ` +name: my-workflow +steps: + - name: do-work + action: http/request + params: + url: "https://example.com" +hooks: + on_failure: + - name: notify + action: slack/send + params: + channel: "#alerts" + text: "failed" + - name: notify + action: email/send + params: + to: "admin@example.com" + body: "failed" +`) + errs := Validate(result) + found := false + for _, e := range errs { + if strings.Contains(e.Message, "duplicate") && strings.Contains(e.Message, "notify") { + found = true + } + } + if !found { + t.Errorf("expected duplicate name error for hook step, got: %v", errs) + } +} + +func TestValidate_ConcurrencyFields(t *testing.T) { + result := mustParse(t, ` +name: my-workflow +max_parallel_executions: 5 +on_limit: queue +steps: + - name: do-work + action: http/request + max_parallel: 3 + params: + url: "https://example.com" +`) + errs := Validate(result) + assertNoErrors(t, errs) + if result.Workflow.MaxParallelExecutions != 5 { + t.Errorf("expected MaxParallelExecutions=5, got %d", result.Workflow.MaxParallelExecutions) + } + if result.Workflow.OnLimit != "queue" { + t.Errorf("expected OnLimit=%q, got %q", "queue", result.Workflow.OnLimit) + } + if result.Workflow.Steps[0].MaxParallel != 3 { + t.Errorf("expected MaxParallel=3, got %d", result.Workflow.Steps[0].MaxParallel) + } +} + +func TestValidate_InvalidOnLimit(t *testing.T) { + result := mustParse(t, ` +name: my-workflow +on_limit: drop +steps: + - name: do-work + action: http/request + params: + url: "https://example.com" +`) + errs := Validate(result) + assertHasError(t, errs, "on_limit") +} + +func TestValidate_HooksTimeout(t *testing.T) { + t.Run("valid timeout", func(t *testing.T) { + result := mustParse(t, ` +name: my-workflow +steps: + - name: do-work + action: http/request + params: + url: "https://example.com" +hooks: + timeout: 5m + on_success: + - name: notify + action: slack/send + params: + channel: "#alerts" + text: "done" +`) + errs := Validate(result) + assertNoErrors(t, errs) + }) + + t.Run("invalid timeout", func(t *testing.T) { + result := mustParse(t, ` +name: my-workflow +steps: + - name: do-work + action: http/request + params: + url: "https://example.com" +hooks: + timeout: not-a-duration + on_success: + - name: notify + action: slack/send + params: + channel: "#alerts" + text: "done" +`) + errs := Validate(result) + assertHasError(t, errs, "hooks.timeout") + }) +} + +func TestValidate_HookStepMissingName(t *testing.T) { + result := mustParse(t, ` +name: my-workflow +steps: + - name: do-work + action: http/request + params: + url: "https://example.com" +hooks: + on_success: + - action: slack/send + params: + channel: "#alerts" +`) + errs := Validate(result) + assertHasError(t, errs, "hooks.on_success[0].name") +} + +func TestValidate_HookStepMissingAction(t *testing.T) { + result := mustParse(t, ` +name: my-workflow +steps: + - name: do-work + action: http/request + params: + url: "https://example.com" +hooks: + on_finish: + - name: cleanup + params: + channel: "#alerts" +`) + errs := Validate(result) + assertHasError(t, errs, "hooks.on_finish[0].action") +} + +func TestValidate_NegativeMaxParallelExecutions(t *testing.T) { + result := mustParse(t, ` +name: my-workflow +max_parallel_executions: -1 +steps: + - name: do-work + action: http/request + params: + url: "https://example.com" +`) + errs := Validate(result) + assertHasError(t, errs, "max_parallel_executions") +} + +func TestValidate_NegativeStepMaxParallel(t *testing.T) { + result := mustParse(t, ` +name: my-workflow +steps: + - name: do-work + action: http/request + max_parallel: -1 + params: + url: "https://example.com" +`) + errs := Validate(result) + assertHasError(t, errs, "steps[0].max_parallel") +} + +func TestValidate_OnLimitReject(t *testing.T) { + result := mustParse(t, ` +name: my-workflow +on_limit: reject +steps: + - name: do-work + action: http/request + params: + url: "https://example.com" +`) + errs := Validate(result) + assertNoErrors(t, errs) +} + +func TestValidate_HookStepNameUniqueAcrossBlocks(t *testing.T) { + // Same name in different hook blocks is NOT OK — they share the CEL + // namespace (hooks.), so duplicates would overwrite each other. + result := mustParse(t, ` +name: my-workflow +steps: + - name: do-work + action: http/request + params: + url: "https://example.com" +hooks: + on_success: + - name: notify + action: slack/send + params: + text: "success" + on_failure: + - name: notify + action: slack/send + params: + text: "failure" +`) + errs := Validate(result) + if len(errs) != 1 { + t.Fatalf("expected 1 error, got %d: %v", len(errs), errs) + } + assertErrorContains(t, errs, "duplicate hook step name") +} diff --git a/packages/engine/internal/workflow/workflow.go b/packages/engine/internal/workflow/workflow.go index 326d267..b87b727 100644 --- a/packages/engine/internal/workflow/workflow.go +++ b/packages/engine/internal/workflow/workflow.go @@ -4,13 +4,24 @@ import "fmt" // Workflow represents a complete workflow definition parsed from YAML. type Workflow struct { - Name string `yaml:"name"` - Description string `yaml:"description"` - Inputs map[string]Input `yaml:"inputs"` - Triggers []Trigger `yaml:"triggers"` - Steps []Step `yaml:"steps"` - Timeout string `yaml:"timeout"` // Go duration string, e.g., "30m" - TokenBudget int64 `yaml:"token_budget"` // 0 = unlimited + Name string `yaml:"name"` + Description string `yaml:"description"` + Inputs map[string]Input `yaml:"inputs"` + Triggers []Trigger `yaml:"triggers"` + Steps []Step `yaml:"steps"` + Timeout string `yaml:"timeout"` // Go duration string, e.g., "30m" + TokenBudget int64 `yaml:"token_budget"` // 0 = unlimited + MaxParallelExecutions int `yaml:"max_parallel_executions"` + OnLimit string `yaml:"on_limit"` + Hooks *HooksConfig `yaml:"hooks"` +} + +// HooksConfig defines lifecycle hooks that run after main workflow steps complete. +type HooksConfig struct { + Timeout string `yaml:"timeout"` + OnSuccess []Step `yaml:"on_success"` + OnFailure []Step `yaml:"on_failure"` + OnFinish []Step `yaml:"on_finish"` } // Trigger defines an automatic execution trigger for a workflow. @@ -42,6 +53,7 @@ type Step struct { Timeout string `yaml:"timeout"` Credential string `yaml:"credential"` DependsOn []string `yaml:"depends_on"` + MaxParallel int `yaml:"max_parallel"` RegistryCredential string `yaml:"registry_credential"` Artifacts []ArtifactDecl `yaml:"artifacts"` ContinueOnError bool `yaml:"continue_on_error"` diff --git a/packages/site/src/content/docs/cli-reference/workflow-commands.md b/packages/site/src/content/docs/cli-reference/workflow-commands.md index c44065a..4906a03 100644 --- a/packages/site/src/content/docs/cli-reference/workflow-commands.md +++ b/packages/site/src/content/docs/cli-reference/workflow-commands.md @@ -273,6 +273,117 @@ $ mantle logs --workflow hello-world --status failed --since 24h --limit 10 --- +## mantle retry + +Retry a failed workflow execution. By default, resumes from the first failed step, reusing outputs from previously completed steps. + +```text +Usage: + mantle retry [flags] +``` + +**Arguments:** + +| Argument | Required | Description | +|---|---|---| +| `execution-id` | Yes | UUID of the failed execution to retry. | + +**Flags:** + +| Flag | Default | Description | +|---|---|---| +| `--from-step` | -- | Step name to retry from. Overrides the default behavior of resuming from the first failed step. All steps from this point forward are re-executed. | +| `--force` | `false` | Bypass per-workflow and per-team concurrency limits. | + +**Example -- retry from failed step:** + +```bash +$ mantle retry abc123-def456 +Retrying execution abc123-def456 from step "summarize"... +Execution def456-abc123: completed + fetch-data: reused (cached) + summarize: completed (4.1s) + post-result: completed (0.3s) +``` + +**Example -- retry from a specific step:** + +```bash +$ mantle retry abc123-def456 --from-step fetch-data +Retrying execution abc123-def456 from step "fetch-data"... +Execution ghi789-jkl012: completed + fetch-data: completed (1.1s) + summarize: completed (3.8s) + post-result: completed (0.2s) +``` + +**Errors:** + +If the execution is not in a failed state: + +```text +Error: execution abc123-def456 is not in a failed state (status: completed). +``` + +If `--from-step` references a step that does not exist: + +```text +Error: step "nonexistent" not found in workflow "fetch-and-summarize" +``` + +--- + +## mantle rollback + +Roll back a workflow to a previous version. The previously active version becomes the current version used by `mantle run` and triggers. + +```text +Usage: + mantle rollback [flags] +``` + +**Arguments:** + +| Argument | Required | Description | +|---|---|---| +| `workflow` | Yes | Name of the workflow to roll back. | + +**Flags:** + +| Flag | Default | Description | +|---|---|---| +| `--to-version` | -- | Specific version number to roll back to. If omitted, rolls back to the previous version (current - 1). | + +**Example -- rollback to previous version:** + +```bash +$ mantle rollback fetch-and-summarize +Rolled back fetch-and-summarize from version 3 to version 2 +``` + +**Example -- rollback to a specific version:** + +```bash +$ mantle rollback fetch-and-summarize --to-version 1 +Rolled back fetch-and-summarize from version 3 to version 1 +``` + +**Errors:** + +If the workflow has only one version: + +```text +Error: workflow "fetch-and-summarize" is at version 1 — nothing to roll back to +``` + +If the target version does not exist: + +```text +Error: workflow "fetch-and-summarize" version 5 not found +``` + +--- + ## mantle status View the current state of a workflow execution with a summary of step statuses. diff --git a/packages/site/src/content/docs/configuration.md b/packages/site/src/content/docs/configuration.md index dca93f4..ba06979 100644 --- a/packages/site/src/content/docs/configuration.md +++ b/packages/site/src/content/docs/configuration.md @@ -20,6 +20,8 @@ By default, Mantle looks for a file named `mantle.yaml` in the current working d ### Full Example ```yaml +version: 1 + database: url: postgres://mantle:mantle@localhost:5432/mantle?sslmode=disable max_open_conns: 25 @@ -46,8 +48,9 @@ engine: step_output_max_bytes: 1048576 default_max_tool_rounds: 10 default_max_tool_calls_per_round: 10 + max_concurrent_executions_per_team: 10 -tmp: +storage: type: filesystem path: /var/lib/mantle/artifacts retention: "24h" @@ -57,6 +60,7 @@ tmp: | Field | Type | Default | Description | |---|---|---|---| +| `version` | integer | -- | Config file version. Must be `1` when present. Required for new config files starting in v0.4.0. | | `database.url` | string | `postgres://mantle:mantle@localhost:5432/mantle?sslmode=disable` | Postgres connection URL. | | `api.address` | string | `:8080` | Listen address for `mantle serve`. Format: `host:port` or `:port`. Used by the HTTP API, webhook listener, and health endpoints. | | `log.level` | string | `info` | Log verbosity. One of: `debug`, `info`, `warn`, `error`. | @@ -75,11 +79,16 @@ tmp: | `engine.step_output_max_bytes` | integer | `1048576` | Maximum size in bytes for a single step's output. Outputs exceeding this limit are truncated. Default is 1 MB. | | `engine.default_max_tool_rounds` | integer | `10` | Default maximum number of LLM-tool interaction rounds for AI steps with tools. Can be overridden per step with `max_tool_rounds`. | | `engine.default_max_tool_calls_per_round` | integer | `10` | Default maximum number of tool calls the LLM can make per round. Can be overridden per step with `max_tool_calls_per_round`. | -| `tmp.type` | string | -- | Artifact storage backend. One of: `s3`, `filesystem`. Required if any workflow declares artifacts. | -| `tmp.bucket` | string | -- | S3 bucket name. Required when `tmp.type` is `s3`. | -| `tmp.prefix` | string | -- | S3 key prefix for artifact storage. Optional. | -| `tmp.path` | string | -- | Local directory path. Required when `tmp.type` is `filesystem`. | -| `tmp.retention` | duration | -- | How long to keep artifacts after workflow completion. Uses Go duration format (e.g., `24h`). Empty means no auto-cleanup. | +| `engine.max_concurrent_executions_per_team` | integer | `10` | Maximum number of concurrent workflow executions allowed per team. When the limit is reached, new executions are queued. Set to `0` for unlimited. | +| `storage.type` | string | -- | Artifact storage backend. One of: `s3`, `filesystem`. Required if any workflow declares artifacts. | +| `storage.bucket` | string | -- | S3 bucket name. Required when `storage.type` is `s3`. | +| `storage.prefix` | string | -- | S3 key prefix for artifact storage. Optional. | +| `storage.path` | string | -- | Local directory path. Required when `storage.type` is `filesystem`. | +| `storage.retention` | duration | -- | How long to keep artifacts after workflow completion. Uses Go duration format (e.g., `24h`). Empty means no auto-cleanup. | + +:::caution[Deprecation: `tmp` renamed to `storage`] +The `tmp` configuration section has been renamed to `storage` in v0.4.0. The old `tmp` key still works but is deprecated and will be removed in a future release. Update your `mantle.yaml` to use `storage` instead. +::: ### Config File Discovery @@ -111,11 +120,12 @@ All environment variables use the `MANTLE_` prefix with underscores replacing do | `MANTLE_ENGINE_STEP_OUTPUT_MAX_BYTES` | `engine.step_output_max_bytes` | `1048576` | | `MANTLE_ENGINE_DEFAULT_MAX_TOOL_ROUNDS` | `engine.default_max_tool_rounds` | `10` | | `MANTLE_ENGINE_DEFAULT_MAX_TOOL_CALLS_PER_ROUND` | `engine.default_max_tool_calls_per_round` | `10` | -| `MANTLE_TMP_TYPE` | `tmp.type` | -- | -| `MANTLE_TMP_BUCKET` | `tmp.bucket` | -- | -| `MANTLE_TMP_PREFIX` | `tmp.prefix` | -- | -| `MANTLE_TMP_PATH` | `tmp.path` | -- | -| `MANTLE_TMP_RETENTION` | `tmp.retention` | -- | +| `MANTLE_ENGINE_MAX_CONCURRENT_EXECUTIONS_PER_TEAM` | `engine.max_concurrent_executions_per_team` | `10` | +| `MANTLE_STORAGE_TYPE` | `storage.type` | -- | +| `MANTLE_STORAGE_BUCKET` | `storage.bucket` | -- | +| `MANTLE_STORAGE_PREFIX` | `storage.prefix` | -- | +| `MANTLE_STORAGE_PATH` | `storage.path` | -- | +| `MANTLE_STORAGE_RETENTION` | `storage.retention` | -- | **Example:** @@ -185,6 +195,8 @@ Use a `mantle.yaml` file with production values, or pass everything through envi ```yaml # mantle.yaml +version: 1 + database: url: postgres://mantle:${DB_PASSWORD}@db.internal:5432/mantle?sslmode=require @@ -213,6 +225,8 @@ When running `mantle serve`, the `api.address` setting controls which address th ```yaml # mantle.yaml +version: 1 + database: url: postgres://mantle:secret@db.internal:5432/mantle?sslmode=require diff --git a/packages/site/src/content/docs/workflow-reference/index.md b/packages/site/src/content/docs/workflow-reference/index.md index e669d4b..7c25943 100644 --- a/packages/site/src/content/docs/workflow-reference/index.md +++ b/packages/site/src/content/docs/workflow-reference/index.md @@ -7,6 +7,8 @@ This document describes the top-level fields and step configuration in a Mantle ```yaml name: fetch-and-summarize description: Fetch data from an API and summarize it with an LLM +max_parallel_executions: 3 +on_limit: queue inputs: url: @@ -22,10 +24,27 @@ triggers: - type: webhook path: "/hooks/fetch-and-summarize" +hooks: + on_success: + - name: notify-success + action: slack/send + credential: slack-token + params: + channel: "#ops" + text: "Workflow {{ execution.workflow_name }} completed successfully" + on_failure: + - name: alert-failure + action: slack/send + credential: slack-token + params: + channel: "#ops-alerts" + text: "Workflow {{ execution.workflow_name }} failed: {{ execution.error }}" + steps: - name: fetch-data action: http/request timeout: 30s + max_parallel: 5 retry: max_attempts: 3 backoff: exponential @@ -59,6 +78,9 @@ steps: | `description` | string | No | Human-readable description of what the workflow does. | | `inputs` | map | No | Input parameters the workflow accepts at runtime. | | `triggers` | list | No | Automatic triggers that start the workflow. See [Triggers](#triggers). | +| `max_parallel_executions` | integer | No | Maximum number of concurrent executions of this workflow. When the limit is reached, behavior is controlled by `on_limit`. Default: unlimited. | +| `on_limit` | string | No | What to do when `max_parallel_executions` is reached. One of: `queue` (wait until a slot opens), `reject` (fail immediately). Default: `queue`. | +| `hooks` | object | No | Lifecycle hooks that run after the workflow completes. See [Hooks](#hooks). | | `steps` | list | Yes | Ordered list of steps to execute. At least one step is required. | ### Name Rules @@ -136,6 +158,7 @@ steps: | `timeout` | string | No | Maximum duration for the step. Uses Go duration format (e.g., `30s`, `5m`, `1h`). | | `credential` | string | No | Name of a stored credential to inject into this step. See [Secrets Guide](/docs/secrets-guide). | | `depends_on` | list of strings | No | Declares explicit dependencies on other steps for parallel execution. See [Parallel Execution](#parallel-execution). | +| `max_parallel` | integer | No | Maximum number of concurrent instances of this step (useful when the step is generated dynamically from a fan-out). Default: unlimited. | | `continue_on_error` | boolean | No | When `true`, workflow execution continues even if this step fails after exhausting retries. Default is `false`. See [Error Handling](#error-handling). | ### Step Name Rules @@ -295,6 +318,71 @@ steps: body: "{{ steps['try-primary-api'].error == null ? steps['try-primary-api'].output.body : steps['try-backup-api'].output.body }}" ``` +## Hooks + +Lifecycle hooks run after the main workflow steps complete. Use hooks for notifications, cleanup, or post-processing that should happen regardless of whether the workflow succeeded or failed. + +```yaml +hooks: + on_success: + - name: notify-success + action: slack/send + credential: slack-token + params: + channel: "#ops" + text: "{{ execution.workflow_name }} v{{ execution.workflow_version }} completed" + + on_failure: + - name: alert-failure + action: slack/send + credential: slack-token + params: + channel: "#ops-alerts" + text: "{{ execution.workflow_name }} failed: {{ execution.error }}" + + on_finish: + - name: record-metrics + action: http/request + params: + method: POST + url: https://metrics.internal/workflow-complete + body: + workflow: "{{ execution.workflow_name }}" + status: "{{ execution.status }}" + duration_ms: "{{ execution.duration_ms }}" +``` + +### Hook Blocks + +| Block | When it runs | +|---|---| +| `on_success` | Only when all steps completed successfully. | +| `on_failure` | Only when the workflow failed (one or more steps failed without `continue_on_error`). | +| `on_finish` | Always runs after the workflow completes, regardless of success or failure. Runs after `on_success` or `on_failure`. | + +Each block contains a list of steps with the same structure as regular workflow steps (`name`, `action`, `params`, `credential`, `timeout`, etc.). Hook steps execute sequentially within their block. + +### Hook CEL Variables + +In addition to the standard CEL variables (`inputs`, `steps`, `env`, `trigger`), hook steps have access to execution metadata: + +| Variable | Type | Description | +|---|---|---| +| `execution.id` | string | UUID of the current execution. | +| `execution.workflow_name` | string | Name of the workflow. | +| `execution.workflow_version` | integer | Version of the workflow definition. | +| `execution.status` | string | Final execution status: `completed`, `failed`, `cancelled`. | +| `execution.error` | string | Error message if the workflow failed; empty string otherwise. | +| `execution.duration_ms` | integer | Total execution duration in milliseconds. | +| `execution.started_at` | string | ISO 8601 timestamp of when the execution started. | +| `hooks['hook-name'].output` | map | Output of a previously executed hook step (within the same block). | + +### Hook Error Behavior + +Hook failures do not change the workflow's final status. If a workflow succeeded but an `on_success` hook fails, the execution status remains `completed`. Hook errors are logged and recorded in the audit trail. + +--- + ## CEL Expressions Mantle uses [CEL (Common Expression Language)](https://cel.dev) for conditional logic and data access between steps. See the [Expressions guide](/docs/concepts/expressions) for practical examples. CEL expressions appear in two places: