diff --git a/examples/docker-data-transform.yaml b/examples/docker-data-transform.yaml new file mode 100644 index 0000000..805b305 --- /dev/null +++ b/examples/docker-data-transform.yaml @@ -0,0 +1,32 @@ +name: docker-data-transform +description: > + Query a database for recent records, process them through + a containerized tool, and post the results to Slack. + Demonstrates the docker/run connector with stdin/stdout + data passing. + +steps: + - name: fetch-records + action: postgres/query + credential: my-db + timeout: "15s" + params: + query: "SELECT id, name, status, updated_at FROM tasks WHERE updated_at > now() - interval '24 hours' ORDER BY updated_at DESC" + + - name: transform-data + action: docker/run + credential: my-docker + timeout: "30s" + params: + image: giantswarm/tiny-tools + cmd: ["jq", "[.[] | {id, name, status}]"] + stdin: "{{ jsonEncode(steps['fetch-records'].output.rows) }}" + memory: "128m" + + - name: post-results + action: slack/send + credential: slack-token + if: "steps['transform-data'].output.exit_code == 0" + params: + channel: "#daily-report" + text: "Daily task summary:\n{{ steps['transform-data'].output.stdout }}" diff --git a/examples/docker-volume-backup.yaml b/examples/docker-volume-backup.yaml new file mode 100644 index 0000000..7ad8c3a --- /dev/null +++ b/examples/docker-volume-backup.yaml @@ -0,0 +1,56 @@ +name: docker-volume-backup +description: > + Back up a Docker volume to S3 on a daily schedule. + Compresses the volume contents, uploads to S3, and + sends a Slack alert on success or failure. + +triggers: + - type: cron + schedule: "0 2 * * *" + +steps: + - name: backup-volume + action: docker/run + credential: my-docker + timeout: "10m" + params: + image: alpine + cmd: ["tar", "-czf", "/mantle/artifacts/backup.tar.gz", "-C", "/data", "."] + mounts: + - source: "my-app-data" + target: "/data" + readonly: true + memory: "256m" + remove: true + artifacts: + - path: /mantle/artifacts/backup.tar.gz + name: backup-archive + + - name: upload-to-s3 + action: s3/put + credential: aws-prod + timeout: "5m" + params: + bucket: my-backups + key: "volumes/my-app-data/{{ date }}/backup.tar.gz" + content: "{{ artifacts['backup-archive'].url }}" + content_type: "application/gzip" + + - name: notify-success + action: slack/send + credential: slack-token + if: "steps['upload-to-s3'].error == null && steps['upload-to-s3'].output.size > 0" + params: + channel: "#ops-alerts" + text: "Volume backup completed — {{ artifacts['backup-archive'].size }} bytes uploaded to s3://my-backups/volumes/my-app-data/" + + # NOTE: notify-failure requires continue_on_error support (see issue #29). + # Without it, the engine halts on the first failed step and this step + # will not execute. This is included to show the intended pattern. + - name: notify-failure + action: slack/send + credential: slack-token + if: "steps['upload-to-s3'].error != null || steps['backup-volume'].error != null" + params: + channel: "#ops-alerts" + text: "Volume backup FAILED" diff --git a/go.mod b/go.mod index 233bd86..1e3c060 100644 --- a/go.mod +++ b/go.mod @@ -22,6 +22,7 @@ require ( github.com/spf13/cobra v1.10.2 github.com/spf13/viper v1.21.0 github.com/stretchr/testify v1.11.1 + github.com/docker/docker v28.5.2+incompatible github.com/testcontainers/testcontainers-go v0.41.0 github.com/testcontainers/testcontainers-go/modules/postgres v0.41.0 golang.org/x/oauth2 v0.35.0 @@ -68,7 +69,6 @@ require ( github.com/cpuguy83/dockercfg v0.3.2 // indirect github.com/davecgh/go-spew v1.1.1 // indirect github.com/distribution/reference v0.6.0 // indirect - github.com/docker/docker v28.5.2+incompatible // indirect github.com/docker/go-connections v0.6.0 // indirect github.com/docker/go-units v0.5.0 // indirect github.com/ebitengine/purego v0.10.0 // indirect diff --git a/internal/artifact/context.go b/internal/artifact/context.go new file mode 100644 index 0000000..8c9df72 --- /dev/null +++ b/internal/artifact/context.go @@ -0,0 +1,16 @@ +package artifact + +import "context" + +type artifactsDirKey struct{} + +// WithArtifactsDir returns a context carrying the artifacts directory path. +func WithArtifactsDir(ctx context.Context, dir string) context.Context { + return context.WithValue(ctx, artifactsDirKey{}, dir) +} + +// ArtifactsDirFromContext extracts the artifacts directory from context. +func ArtifactsDirFromContext(ctx context.Context) string { + dir, _ := ctx.Value(artifactsDirKey{}).(string) + return dir +} diff --git a/internal/artifact/reaper.go b/internal/artifact/reaper.go new file mode 100644 index 0000000..cc1b4f2 --- /dev/null +++ b/internal/artifact/reaper.go @@ -0,0 +1,60 @@ +package artifact + +import ( + "context" + "fmt" + "log/slog" + "time" +) + +// Reaper cleans up expired artifacts from both tmp storage and the database. +type Reaper struct { + Store *Store + TmpStorage TmpStorage + Retention time.Duration + Logger *slog.Logger +} + +// Sweep finds and removes artifacts older than the retention period. +// Returns the number of artifacts cleaned up. +func (r *Reaper) Sweep(ctx context.Context) (int, error) { + if r.Retention <= 0 { + return 0, nil // cleanup disabled + } + + logger := r.Logger + if logger == nil { + logger = slog.Default() + } + + expired, err := r.Store.ListExpired(ctx, r.Retention) + if err != nil { + return 0, fmt.Errorf("listing expired artifacts: %w", err) + } + + if len(expired) == 0 { + return 0, nil + } + + cleaned := 0 + for _, a := range expired { + // Delete file from tmp storage. + if delErr := r.TmpStorage.Delete(ctx, a.URL); delErr != nil { + logger.Error("failed to delete artifact file", + "artifact", a.Name, "url", a.URL, "error", delErr) + continue // skip metadata deletion if file delete failed + } + + // Delete metadata from database. + if err := r.Store.DeleteByID(ctx, a.ID); err != nil { + logger.Error("failed to delete artifact metadata", + "artifact", a.Name, "id", a.ID, "error", err) + continue + } + cleaned++ + logger.Info("cleaned expired artifact", + "artifact", a.Name, "execution_id", a.ExecutionID) + } + + return cleaned, nil +} diff --git a/internal/artifact/reaper_test.go b/internal/artifact/reaper_test.go new file mode 100644 index 0000000..18bd94c --- /dev/null +++ b/internal/artifact/reaper_test.go @@ -0,0 +1,90 @@ +package artifact + +import ( + "context" + "os" + "path/filepath" + "testing" + "time" +) + +func TestReaper_CleansExpiredArtifacts(t *testing.T) { + dir := t.TempDir() + fs := &FilesystemTmpStorage{BasePath: dir} + db := setupTestDB(t) + store := &Store{DB: db} + ctx := context.Background() + execID := createTestExecution(t, db) + + // Write a file and create metadata + srcPath := filepath.Join(dir, "src.bin") + if err := os.WriteFile(srcPath, []byte("data"), 0644); err != nil { + t.Fatalf("os.WriteFile: %v", err) + } + key := "wf/" + execID + "/artifact/file.bin" + url, err := fs.Put(ctx, key, srcPath) + if err != nil { + t.Fatalf("fs.Put: %v", err) + } + + if err := store.Create(ctx, &Artifact{ + ExecutionID: execID, + StepName: "step-a", + Name: "test-artifact", + URL: url, + Size: 4, + }); err != nil { + t.Fatalf("store.Create: %v", err) + } + + // Backdate the artifact so it appears expired + _, err = db.ExecContext(ctx, ` + UPDATE execution_artifacts SET created_at = NOW() - interval '48 hours' + WHERE execution_id = $1 + `, execID) + if err != nil { + t.Fatalf("backdate: %v", err) + } + + reaper := &Reaper{ + Store: store, + TmpStorage: fs, + Retention: 24 * time.Hour, + } + + cleaned, err := reaper.Sweep(ctx) + if err != nil { + t.Fatalf("Sweep: %v", err) + } + if cleaned != 1 { + t.Errorf("cleaned = %d, want 1", cleaned) + } + + // Metadata should be gone + arts, listErr := store.ListByExecution(ctx, execID) + if listErr != nil { + t.Fatalf("ListByExecution: %v", listErr) + } + if len(arts) != 0 { + t.Errorf("expected 0 artifacts after sweep, got %d", len(arts)) + } +} + +func TestReaper_SkipsWhenRetentionZero(t *testing.T) { + dir := t.TempDir() + fs := &FilesystemTmpStorage{BasePath: dir} + + reaper := &Reaper{ + Store: &Store{}, // safe: Sweep returns early when Retention <= 0, before accessing Store.DB + TmpStorage: fs, + Retention: 0, + } + + cleaned, err := reaper.Sweep(context.Background()) + if err != nil { + t.Fatalf("Sweep: %v", err) + } + if cleaned != 0 { + t.Errorf("cleaned = %d, want 0 when retention is 0", cleaned) + } +} diff --git a/internal/artifact/store.go b/internal/artifact/store.go new file mode 100644 index 0000000..fbd5655 --- /dev/null +++ b/internal/artifact/store.go @@ -0,0 +1,124 @@ +package artifact + +import ( + "context" + "database/sql" + "fmt" + "time" +) + +// Artifact represents metadata for a persisted artifact. +type Artifact struct { + ID string + ExecutionID string + StepName string + Name string + URL string + Size int64 + CreatedAt time.Time +} + +// Store manages artifact metadata in Postgres. +type Store struct { + DB *sql.DB +} + +// Create inserts artifact metadata into the database and populates a.ID and a.CreatedAt. +func (s *Store) Create(ctx context.Context, a *Artifact) error { + err := s.DB.QueryRowContext(ctx, ` + INSERT INTO execution_artifacts (execution_id, step_name, name, url, size) + VALUES ($1, $2, $3, $4, $5) + RETURNING id, created_at + `, a.ExecutionID, a.StepName, a.Name, a.URL, a.Size).Scan(&a.ID, &a.CreatedAt) + if err != nil { + return fmt.Errorf("artifact create: %w", err) + } + return nil +} + +// GetByName retrieves a single artifact by execution ID and artifact name. +func (s *Store) GetByName(ctx context.Context, executionID, name string) (*Artifact, error) { + var a Artifact + err := s.DB.QueryRowContext(ctx, ` + SELECT id, execution_id, step_name, name, url, size, created_at + FROM execution_artifacts + WHERE execution_id = $1 AND name = $2 + `, executionID, name).Scan(&a.ID, &a.ExecutionID, &a.StepName, &a.Name, &a.URL, &a.Size, &a.CreatedAt) + if err == sql.ErrNoRows { + return nil, fmt.Errorf("artifact %q not found for execution %s", name, executionID) + } + if err != nil { + return nil, fmt.Errorf("artifact get: %w", err) + } + return &a, nil +} + +// ListByExecution returns all artifacts associated with the given execution, ordered by creation time. +func (s *Store) ListByExecution(ctx context.Context, executionID string) ([]Artifact, error) { + rows, err := s.DB.QueryContext(ctx, ` + SELECT id, execution_id, step_name, name, url, size, created_at + FROM execution_artifacts + WHERE execution_id = $1 + ORDER BY created_at ASC + `, executionID) + if err != nil { + return nil, fmt.Errorf("artifact list: %w", err) + } + defer rows.Close() + + var artifacts []Artifact + for rows.Next() { + var a Artifact + if err := rows.Scan(&a.ID, &a.ExecutionID, &a.StepName, &a.Name, &a.URL, &a.Size, &a.CreatedAt); err != nil { + return nil, fmt.Errorf("artifact scan: %w", err) + } + artifacts = append(artifacts, a) + } + return artifacts, rows.Err() +} + +// DeleteByExecution removes all artifact metadata for an execution. +func (s *Store) DeleteByExecution(ctx context.Context, executionID string) error { + _, err := s.DB.ExecContext(ctx, ` + DELETE FROM execution_artifacts WHERE execution_id = $1 + `, executionID) + if err != nil { + return fmt.Errorf("artifact delete: %w", err) + } + return nil +} + +// DeleteByID removes a single artifact metadata row. +func (s *Store) DeleteByID(ctx context.Context, id string) error { + _, err := s.DB.ExecContext(ctx, ` + DELETE FROM execution_artifacts WHERE id = $1 + `, id) + if err != nil { + return fmt.Errorf("artifact delete by id: %w", err) + } + return nil +} + +// ListExpired returns artifacts older than the given duration, using the DB clock for comparison. +func (s *Store) ListExpired(ctx context.Context, olderThan time.Duration) ([]Artifact, error) { + rows, err := s.DB.QueryContext(ctx, ` + SELECT id, execution_id, step_name, name, url, size, created_at + FROM execution_artifacts + WHERE created_at < NOW() - $1::interval + ORDER BY created_at ASC + `, olderThan.String()) + if err != nil { + return nil, fmt.Errorf("artifact list expired: %w", err) + } + defer rows.Close() + + var artifacts []Artifact + for rows.Next() { + var a Artifact + if err := rows.Scan(&a.ID, &a.ExecutionID, &a.StepName, &a.Name, &a.URL, &a.Size, &a.CreatedAt); err != nil { + return nil, fmt.Errorf("artifact scan: %w", err) + } + artifacts = append(artifacts, a) + } + return artifacts, rows.Err() +} diff --git a/internal/artifact/store_test.go b/internal/artifact/store_test.go new file mode 100644 index 0000000..5a1698a --- /dev/null +++ b/internal/artifact/store_test.go @@ -0,0 +1,137 @@ +package artifact + +import ( + "context" + "strings" + "testing" +) + +func TestStore_CreateAndList(t *testing.T) { + database := setupTestDB(t) + ctx := context.Background() + execID := createTestExecution(t, database) + + store := &Store{DB: database} + + a := &Artifact{ + ExecutionID: execID, + StepName: "fetch", + Name: "output.json", + URL: "s3://bucket/output.json", + Size: 1024, + } + if err := store.Create(ctx, a); err != nil { + t.Fatalf("Create() error = %v", err) + } + + artifacts, err := store.ListByExecution(ctx, execID) + if err != nil { + t.Fatalf("ListByExecution() error = %v", err) + } + if len(artifacts) != 1 { + t.Fatalf("ListByExecution() returned %d artifacts, want 1", len(artifacts)) + } + + got := artifacts[0] + if got.ExecutionID != execID { + t.Errorf("ExecutionID = %q, want %q", got.ExecutionID, execID) + } + if got.StepName != "fetch" { + t.Errorf("StepName = %q, want %q", got.StepName, "fetch") + } + if got.Name != "output.json" { + t.Errorf("Name = %q, want %q", got.Name, "output.json") + } + if got.URL != "s3://bucket/output.json" { + t.Errorf("URL = %q, want %q", got.URL, "s3://bucket/output.json") + } + if got.Size != 1024 { + t.Errorf("Size = %d, want 1024", got.Size) + } + if got.ID == "" { + t.Error("ID is empty, want a UUID") + } +} + +func TestStore_GetByName(t *testing.T) { + database := setupTestDB(t) + ctx := context.Background() + execID := createTestExecution(t, database) + + store := &Store{DB: database} + + a := &Artifact{ + ExecutionID: execID, + StepName: "transform", + Name: "result.csv", + URL: "s3://bucket/result.csv", + Size: 2048, + } + if err := store.Create(ctx, a); err != nil { + t.Fatalf("Create() error = %v", err) + } + + got, err := store.GetByName(ctx, execID, "result.csv") + if err != nil { + t.Fatalf("GetByName() error = %v", err) + } + + if got.ExecutionID != execID { + t.Errorf("ExecutionID = %q, want %q", got.ExecutionID, execID) + } + if got.StepName != "transform" { + t.Errorf("StepName = %q, want %q", got.StepName, "transform") + } + if got.Name != "result.csv" { + t.Errorf("Name = %q, want %q", got.Name, "result.csv") + } + if got.URL != "s3://bucket/result.csv" { + t.Errorf("URL = %q, want %q", got.URL, "s3://bucket/result.csv") + } + if got.Size != 2048 { + t.Errorf("Size = %d, want 2048", got.Size) + } + if got.ID == "" { + t.Error("ID is empty, want a UUID") + } + if got.CreatedAt.IsZero() { + t.Error("CreatedAt is zero, want a non-zero time") + } +} + +func TestStore_DuplicateNameFails(t *testing.T) { + database := setupTestDB(t) + ctx := context.Background() + execID := createTestExecution(t, database) + + store := &Store{DB: database} + + a := &Artifact{ + ExecutionID: execID, + StepName: "step1", + Name: "report.pdf", + URL: "s3://bucket/report.pdf", + Size: 512, + } + if err := store.Create(ctx, a); err != nil { + t.Fatalf("Create() first artifact error = %v", err) + } + + duplicate := &Artifact{ + ExecutionID: execID, + StepName: "step2", + Name: "report.pdf", + URL: "s3://bucket/report-v2.pdf", + Size: 768, + } + err := store.Create(ctx, duplicate) + if err == nil { + t.Fatal("Create() second artifact with duplicate name expected error, got nil") + } + if !strings.Contains(err.Error(), "artifact create") { + t.Errorf("error = %q, want it to contain %q", err.Error(), "artifact create") + } + if !strings.Contains(err.Error(), "duplicate key") && !strings.Contains(err.Error(), "unique") { + t.Errorf("error = %q, want it to indicate a uniqueness violation", err.Error()) + } +} diff --git a/internal/artifact/test_helpers_test.go b/internal/artifact/test_helpers_test.go new file mode 100644 index 0000000..a948d86 --- /dev/null +++ b/internal/artifact/test_helpers_test.go @@ -0,0 +1,73 @@ +package artifact + +import ( + "context" + "database/sql" + "os" + "testing" + "time" + + "github.com/dvflw/mantle/internal/config" + "github.com/dvflw/mantle/internal/db" + "github.com/dvflw/mantle/internal/dbdefaults" + "github.com/testcontainers/testcontainers-go" + "github.com/testcontainers/testcontainers-go/modules/postgres" + "github.com/testcontainers/testcontainers-go/wait" +) + +func setupTestDB(t *testing.T) *sql.DB { + t.Helper() + ctx := context.Background() + pgContainer, err := postgres.Run(ctx, + dbdefaults.PostgresImage, + postgres.WithDatabase(dbdefaults.TestDatabase), + postgres.WithUsername(dbdefaults.User), + postgres.WithPassword(dbdefaults.Password), + testcontainers.WithWaitStrategy( + wait.ForLog("database system is ready to accept connections"). + WithOccurrence(2). + WithStartupTimeout(30*time.Second), + ), + ) + if err != nil { + if os.Getenv("CI") != "" { + t.Fatalf("Could not start Postgres container (CI): %v", err) + } + t.Skipf("Could not start Postgres container: %v", err) + } + t.Cleanup(func() { + if err := pgContainer.Terminate(ctx); err != nil { + t.Logf("Failed to terminate container: %v", err) + } + }) + connStr, err := pgContainer.ConnectionString(ctx, "sslmode=disable") + if err != nil { + t.Fatalf("Failed to get connection string: %v", err) + } + + database, err := db.Open(config.DatabaseConfig{URL: connStr}) + if err != nil { + t.Fatalf("Failed to open database: %v", err) + } + t.Cleanup(func() { database.Close() }) + + if err := db.Migrate(ctx, database); err != nil { + t.Fatalf("Failed to run migrations: %v", err) + } + + return database +} + +func createTestExecution(t *testing.T, database *sql.DB) string { + t.Helper() + var id string + err := database.QueryRow(` + INSERT INTO workflow_executions (workflow_name, workflow_version, status) + VALUES ('test-workflow', 1, 'running') + RETURNING id + `).Scan(&id) + if err != nil { + t.Fatalf("createTestExecution: %v", err) + } + return id +} diff --git a/internal/artifact/tmp.go b/internal/artifact/tmp.go new file mode 100644 index 0000000..76ffa11 --- /dev/null +++ b/internal/artifact/tmp.go @@ -0,0 +1,117 @@ +package artifact + +import ( + "context" + "fmt" + "io" + "os" + "path/filepath" + "strings" +) + +// TmpStorage persists and retrieves artifact files. +type TmpStorage interface { + // Put copies a local file to tmp 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) + + // Delete removes a single artifact by the URL/path returned from Put. + Delete(ctx context.Context, url string) error + + // DeleteByPrefix removes all files under the given key prefix. + DeleteByPrefix(ctx context.Context, prefix string) error +} + +// FilesystemTmpStorage stores artifacts on the local filesystem. +type FilesystemTmpStorage 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) { + destPath := filepath.Join(fs.BasePath, key) + // Validate destination is within BasePath to prevent path traversal. + absBase, err := filepath.Abs(fs.BasePath) + if err != nil { + return "", fmt.Errorf("resolving base path: %w", err) + } + absDest, err := filepath.Abs(destPath) + if err != nil { + return "", fmt.Errorf("resolving dest path: %w", err) + } + rel, err := filepath.Rel(absBase, absDest) + 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 { + return "", fmt.Errorf("stat source file: %w", err) + } + if !fi.Mode().IsRegular() { + return "", fmt.Errorf("source file %q is not a regular file (mode: %s)", localPath, fi.Mode()) + } + + src, err := os.Open(localPath) + if err != nil { + return "", fmt.Errorf("opening source file: %w", err) + } + defer src.Close() + + dst, err := os.Create(destPath) + if err != nil { + return "", fmt.Errorf("creating dest file: %w", err) + } + defer dst.Close() + + if _, err := io.Copy(dst, src); err != nil { + dst.Close() + os.Remove(destPath) + return "", fmt.Errorf("copying artifact: %w", err) + } + + return destPath, nil +} + +// DeleteByPrefix removes all files stored under the given key prefix. +func (fs *FilesystemTmpStorage) DeleteByPrefix(ctx context.Context, prefix string) error { + target := filepath.Join(fs.BasePath, prefix) + absBase, err := filepath.Abs(fs.BasePath) + if err != nil { + return fmt.Errorf("resolving base path: %w", err) + } + absTarget, err := filepath.Abs(target) + if err != nil { + return fmt.Errorf("resolving target path: %w", err) + } + rel, err := filepath.Rel(absBase, absTarget) + if err != nil || strings.HasPrefix(rel, "..") { + return fmt.Errorf("prefix escapes base path") + } + return os.RemoveAll(target) +} + +// 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) + if err != nil { + return fmt.Errorf("resolving base path: %w", err) + } + absURL, err := filepath.Abs(url) + if err != nil { + return fmt.Errorf("resolving artifact path: %w", err) + } + rel, err := filepath.Rel(absBase, absURL) + if err != nil || strings.HasPrefix(rel, "..") { + return fmt.Errorf("artifact path escapes base path") + } + if err := os.Remove(absURL); err != nil && !os.IsNotExist(err) { + return err + } + return nil +} diff --git a/internal/artifact/tmp_test.go b/internal/artifact/tmp_test.go new file mode 100644 index 0000000..7162a4c --- /dev/null +++ b/internal/artifact/tmp_test.go @@ -0,0 +1,95 @@ +package artifact + +import ( + "context" + "os" + "path/filepath" + "testing" +) + +func TestFilesystemTmpStorage_PutAndGet(t *testing.T) { + dir := t.TempDir() + fs := &FilesystemTmpStorage{BasePath: dir} + + ctx := context.Background() + key := "test-workflow/exec-123/my-artifact/data.tar.gz" + data := []byte("fake tar content") + + // Write source file first. + srcPath := filepath.Join(dir, "source.tar.gz") + if err := os.WriteFile(srcPath, data, 0644); err != nil { + t.Fatal(err) + } + + url, err := fs.Put(ctx, key, srcPath) + if err != nil { + t.Fatalf("Put: %v", err) + } + + if url == "" { + t.Fatal("url should not be empty") + } + + // Verify file exists at the stored location. + stored, err := os.ReadFile(url) + if err != nil { + t.Fatalf("reading stored file: %v", err) + } + if string(stored) != string(data) { + t.Errorf("stored content = %q, want %q", stored, data) + } +} + +func TestFilesystemTmpStorage_Delete(t *testing.T) { + dir := t.TempDir() + fs := &FilesystemTmpStorage{BasePath: dir} + ctx := context.Background() + + // Write a file + key := "test-workflow/exec-123/artifact/file.txt" + srcPath := filepath.Join(dir, "src.txt") + if err := os.WriteFile(srcPath, []byte("data"), 0644); err != nil { + t.Fatal(err) + } + url, err := fs.Put(ctx, key, srcPath) + if err != nil { + t.Fatalf("Put: %v", err) + } + + // Delete by prefix + err = fs.DeleteByPrefix(ctx, "test-workflow/exec-123/") + if err != nil { + t.Fatalf("DeleteByPrefix: %v", err) + } + + // File should be gone + if _, err := os.Stat(url); !os.IsNotExist(err) { + t.Error("file should have been deleted") + } +} + +func TestFilesystemTmpStorage_DeleteEscapeProtection(t *testing.T) { + dir := t.TempDir() + fs := &FilesystemTmpStorage{BasePath: dir} + ctx := context.Background() + + err := fs.DeleteByPrefix(ctx, "../../etc") + if err == nil { + t.Error("expected error for path traversal") + } +} + +func TestArtifactsDirContext(t *testing.T) { + ctx := context.Background() + + // Empty by default + if dir := ArtifactsDirFromContext(ctx); dir != "" { + t.Errorf("expected empty, got %q", dir) + } + + // Set and retrieve + ctx = WithArtifactsDir(ctx, "/tmp/artifacts") + if dir := ArtifactsDirFromContext(ctx); dir != "/tmp/artifacts" { + t.Errorf("expected '/tmp/artifacts', got %q", dir) + } +} diff --git a/internal/audit/audit.go b/internal/audit/audit.go index 396bd77..a911f78 100644 --- a/internal/audit/audit.go +++ b/internal/audit/audit.go @@ -15,7 +15,8 @@ const ( ActionStepCompleted Action = "step.completed" ActionStepFailed Action = "step.failed" ActionStepSkipped Action = "step.skipped" - ActionExecutionCancelled Action = "execution.cancelled" + ActionExecutionCancelled Action = "execution.cancelled" + ActionArtifactPersisted Action = "artifact.persisted" // Admin operations. ActionUserCreated Action = "user.created" diff --git a/internal/cel/cel.go b/internal/cel/cel.go index f54f3b4..f665aca 100644 --- a/internal/cel/cel.go +++ b/internal/cel/cel.go @@ -14,9 +14,10 @@ import ( // Context holds the runtime data available to CEL expressions. type Context struct { - Steps map[string]map[string]any // steps..output → step outputs - Inputs map[string]any // inputs. → workflow inputs - Trigger map[string]any // trigger.payload → webhook trigger data + Steps map[string]map[string]any // steps..output → step outputs + 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} } // Evaluator evaluates CEL expressions against a runtime context. @@ -33,6 +34,7 @@ func NewEvaluator() (*Evaluator, error) { cel.Variable("inputs", cel.MapType(cel.StringType, cel.DynType)), 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)), } opts = append(opts, customFunctions()...) @@ -56,11 +58,17 @@ func (e *Evaluator) Eval(expression string, ctx *Context) (any, error) { trigger = map[string]any{} } + artifacts := ctx.Artifacts + if artifacts == nil { + artifacts = map[string]map[string]any{} + } + vars := map[string]any{ - "steps": ctx.Steps, - "inputs": ctx.Inputs, - "env": e.envCache, - "trigger": trigger, + "steps": ctx.Steps, + "inputs": ctx.Inputs, + "env": e.envCache, + "trigger": trigger, + "artifacts": artifacts, } out, _, err := prog.Eval(vars) diff --git a/internal/cel/cel_test.go b/internal/cel/cel_test.go index 17618b6..b89882e 100644 --- a/internal/cel/cel_test.go +++ b/internal/cel/cel_test.go @@ -317,3 +317,40 @@ func TestEnvVars_PrefixStripping(t *testing.T) { t.Error("envVars() should not contain DATABASE_URL (MANTLE_ prefix without ENV_)") } } + +func TestEval_ArtifactsAccess(t *testing.T) { + eval, err := NewEvaluator() + if err != nil { + t.Fatal(err) + } + + ctx := &Context{ + Steps: map[string]map[string]any{}, + Inputs: map[string]any{}, + Artifacts: map[string]map[string]any{ + "backup-archive": { + "name": "backup-archive", + "url": "s3://bucket/path/backup.tar.gz", + "size": int64(1048576), + }, + }, + } + + // Test accessing artifact URL + result, err := eval.Eval("artifacts['backup-archive'].url", ctx) + if err != nil { + t.Fatalf("Eval url: %v", err) + } + if result != "s3://bucket/path/backup.tar.gz" { + t.Errorf("url result = %v, want s3 URL", result) + } + + // Test accessing artifact size + result, err = eval.Eval("artifacts['backup-archive'].size", ctx) + if err != nil { + t.Fatalf("Eval size: %v", err) + } + if result != int64(1048576) { + t.Errorf("size result = %v, want 1048576", result) + } +} diff --git a/internal/config/config.go b/internal/config/config.go index 8d05196..3252fe4 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -29,6 +29,7 @@ type Config struct { AWS AWSConfig `mapstructure:"aws"` GCP GCPConfig `mapstructure:"gcp"` Azure AzureConfig `mapstructure:"azure"` + Tmp TmpConfig `mapstructure:"tmp"` } // RetentionConfig holds data retention settings. @@ -53,6 +54,15 @@ type AzureConfig struct { Region string `mapstructure:"region"` } +// TmpConfig configures ephemeral storage for workflow artifacts. +type TmpConfig 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) + Path string `mapstructure:"path"` // Local directory (for type: filesystem) + Retention string `mapstructure:"retention"` // Duration string, e.g. "24h". Empty = no auto-cleanup. +} + // AuthConfig holds authentication configuration. type AuthConfig struct { OIDC OIDCConfig `mapstructure:"oidc"` @@ -219,6 +229,13 @@ 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") + // Engine env var bindings _ = v.BindEnv("engine.node_id", "MANTLE_ENGINE_NODE_ID") _ = v.BindEnv("engine.worker_poll_interval", "MANTLE_ENGINE_WORKER_POLL_INTERVAL") diff --git a/internal/config/config_test.go b/internal/config/config_test.go index ae1572a..310ad7a 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -247,3 +247,65 @@ func TestLoad_BudgetResetDay_CalendarClampsUpperBound(t *testing.T) { require.NoError(t, err) 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") + + 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) +} + +func TestLoad_TmpConfigFromConfigFile(t *testing.T) { + dir := t.TempDir() + configFile := filepath.Join(dir, "mantle.yaml") + err := os.WriteFile(configFile, []byte(` +tmp: + type: "filesystem" + path: "/tmp/mantle-artifacts" + retention: "48h" +`), 0644) + require.NoError(t, err) + + cmd := newTestCommand() + _ = cmd.Flags().Set("config", configFile) + + 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) +} + +func TestLoad_TmpConfigEnvVarOverridesFile(t *testing.T) { + dir := t.TempDir() + configFile := filepath.Join(dir, "mantle.yaml") + _ = os.WriteFile(configFile, []byte(` +tmp: + type: "filesystem" + path: "/tmp/file-path" +`), 0644) + + t.Setenv("MANTLE_TMP_TYPE", "s3") + t.Setenv("MANTLE_TMP_BUCKET", "env-bucket") + + cmd := newTestCommand() + _ = cmd.Flags().Set("config", configFile) + + cfg, err := Load(cmd) + require.NoError(t, err) + + assert.Equal(t, "s3", cfg.Tmp.Type) + assert.Equal(t, "env-bucket", cfg.Tmp.Bucket) + // path from file should still be present since env var didn't override it + assert.Equal(t, "/tmp/file-path", cfg.Tmp.Path) +} diff --git a/internal/connector/connector.go b/internal/connector/connector.go index 5d51c03..cd928ca 100644 --- a/internal/connector/connector.go +++ b/internal/connector/connector.go @@ -34,6 +34,7 @@ func NewRegistry() *Registry { r.Register("s3/put", &S3PutConnector{}) r.Register("s3/get", &S3GetConnector{}) r.Register("s3/list", &S3ListConnector{}) + r.Register("docker/run", &DockerRunConnector{}) return r } diff --git a/internal/connector/docker.go b/internal/connector/docker.go new file mode 100644 index 0000000..8d6d332 --- /dev/null +++ b/internal/connector/docker.go @@ -0,0 +1,446 @@ +package connector + +import ( + "context" + "encoding/base64" + "encoding/json" + "fmt" + "io" + "log" + "os" + "path/filepath" + "strconv" + "strings" + "time" + + "github.com/docker/docker/api/types/container" + "github.com/docker/docker/api/types/image" + "github.com/docker/docker/api/types/mount" + "github.com/docker/docker/api/types/network" + "github.com/docker/docker/client" + "github.com/docker/docker/pkg/stdcopy" + "github.com/dvflw/mantle/internal/artifact" +) + +// DockerRunConnector runs a container to completion and captures output. +type DockerRunConnector struct{} + +type dockerConfig struct { + Image string + Cmd []string + Env map[string]string + Stdin string + Mounts []dockerMount + Network string + Pull string // "always", "missing", "never" + Memory string + CPUs float64 + Remove bool +} + +type dockerMount struct { + Source string + Target string + ReadOnly bool +} + +func parseDockerParams(params map[string]any) (*dockerConfig, error) { + img, _ := params["image"].(string) + if img == "" { + return nil, fmt.Errorf("docker/run: image is required") + } + + cfg := &dockerConfig{ + Image: img, + Network: "bridge", + Pull: "missing", + Remove: true, + } + + // cmd + if raw, ok := params["cmd"].([]any); ok { + for _, v := range raw { + if s, ok := v.(string); ok { + cfg.Cmd = append(cfg.Cmd, s) + } + } + } + + // env + if raw, ok := params["env"].(map[string]any); ok { + cfg.Env = make(map[string]string, len(raw)) + for k, v := range raw { + cfg.Env[k] = fmt.Sprintf("%v", v) + } + } + + // stdin + cfg.Stdin, _ = params["stdin"].(string) + + // mounts + if raw, ok := params["mounts"].([]any); ok { + for _, item := range raw { + m, ok := item.(map[string]any) + if !ok { + continue + } + dm := dockerMount{} + dm.Source, _ = m["source"].(string) + dm.Target, _ = m["target"].(string) + dm.ReadOnly, _ = m["readonly"].(bool) + if dm.Source == "" || dm.Target == "" { + return nil, fmt.Errorf("docker/run: mount requires both source and target") + } + cfg.Mounts = append(cfg.Mounts, dm) + } + } + + // network + if n, ok := params["network"].(string); ok && n != "" { + cfg.Network = n + } + + // pull + if p, ok := params["pull"].(string); ok && p != "" { + cfg.Pull = p + } + + // memory + cfg.Memory, _ = params["memory"].(string) + + // cpus + switch v := params["cpus"].(type) { + case float64: + if v < 0 { + return nil, fmt.Errorf("docker/run: cpus must be non-negative, got %v", v) + } + cfg.CPUs = v + case int: + if v < 0 { + return nil, fmt.Errorf("docker/run: cpus must be non-negative, got %v", v) + } + cfg.CPUs = float64(v) + case int64: + if v < 0 { + return nil, fmt.Errorf("docker/run: cpus must be non-negative, got %v", v) + } + cfg.CPUs = float64(v) + } + + // remove + if r, ok := params["remove"].(bool); ok { + cfg.Remove = r + } + + return cfg, nil +} + +// parseMemoryString converts strings like "256m", "1g" to bytes. +func parseMemoryString(s string) (int64, error) { + s = strings.TrimSpace(strings.ToLower(s)) + if s == "" { + return 0, nil + } + var multiplier int64 = 1 + switch { + case strings.HasSuffix(s, "g"): + multiplier = 1024 * 1024 * 1024 + s = s[:len(s)-1] + case strings.HasSuffix(s, "m"): + multiplier = 1024 * 1024 + s = s[:len(s)-1] + case strings.HasSuffix(s, "k"): + multiplier = 1024 + s = s[:len(s)-1] + } + n, err := strconv.ParseInt(s, 10, 64) + if err != nil { + return 0, fmt.Errorf("invalid memory value %q: %w", s, err) + } + if n < 0 { + return 0, fmt.Errorf("negative memory value %q", s) + } + return n * multiplier, nil +} + +// limitWriter wraps a writer to limit bytes written. Logs a warning on first truncation. +type limitWriter struct { + w io.Writer + limit int64 + n int64 + truncated bool +} + +func (lw *limitWriter) Write(p []byte) (int, error) { + remaining := lw.limit - lw.n + if remaining <= 0 { + if !lw.truncated { + lw.truncated = true + log.Printf("docker/run: output truncated at %d bytes", lw.limit) + } + return len(p), nil + } + if int64(len(p)) > remaining { + p = p[:remaining] + lw.truncated = true + log.Printf("docker/run: output truncated at %d bytes", lw.limit) + } + n, err := lw.w.Write(p) + lw.n += int64(n) + return n, err +} + +// Execute runs a Docker container to completion and returns its exit code, stdout, and stderr. +func (c *DockerRunConnector) Execute(ctx context.Context, params map[string]any) (map[string]any, error) { + cfg, err := parseDockerParams(params) + if err != nil { + return nil, err + } + + // Build client options from credential. + clientOpts := []client.Opt{client.WithAPIVersionNegotiation()} + if cred, ok := params["_credential"].(map[string]string); ok { + if host := cred["host"]; host != "" { + clientOpts = append(clientOpts, client.WithHost(host)) + } + // TLS configuration. + if cred["ca_cert"] != "" && cred["client_cert"] != "" && cred["client_key"] != "" { + tmpDir, tmpErr := os.MkdirTemp("", "mantle-docker-tls-*") + if tmpErr != nil { + return nil, fmt.Errorf("docker/run: creating TLS temp dir: %w", tmpErr) + } + defer os.RemoveAll(tmpDir) + caPath := filepath.Join(tmpDir, "ca.pem") + certPath := filepath.Join(tmpDir, "cert.pem") + keyPath := filepath.Join(tmpDir, "key.pem") + if err := os.WriteFile(caPath, []byte(cred["ca_cert"]), 0600); err != nil { + return nil, fmt.Errorf("docker/run: writing CA cert: %w", err) + } + if err := os.WriteFile(certPath, []byte(cred["client_cert"]), 0600); err != nil { + return nil, fmt.Errorf("docker/run: writing client cert: %w", err) + } + if err := os.WriteFile(keyPath, []byte(cred["client_key"]), 0600); err != nil { + return nil, fmt.Errorf("docker/run: writing client key: %w", err) + } + clientOpts = append(clientOpts, client.WithTLSClientConfig(caPath, certPath, keyPath)) + } + } else { + clientOpts = append(clientOpts, client.FromEnv) + } + + cli, err := client.NewClientWithOpts(clientOpts...) + if err != nil { + return nil, fmt.Errorf("docker/run: creating client: %w", err) + } + defer cli.Close() + + // Build registry auth for private image pulls. + var registryAuth string + if regCred, ok := params["_registry_credential"].(map[string]string); ok { + authConfig := map[string]string{ + "username": regCred["username"], + "password": regCred["password"], + } + authJSON, _ := json.Marshal(authConfig) + registryAuth = base64.URLEncoding.EncodeToString(authJSON) + } + + // Pull image. + if err := pullImage(ctx, cli, cfg, registryAuth); err != nil { + return nil, err + } + + // Build container config. + containerCfg := &container.Config{ + Image: cfg.Image, + Cmd: cfg.Cmd, + } + + // Env vars. + for k, v := range cfg.Env { + containerCfg.Env = append(containerCfg.Env, k+"="+v) + } + + // Stdin. + if cfg.Stdin != "" { + containerCfg.OpenStdin = true + containerCfg.StdinOnce = true + containerCfg.AttachStdin = true + } + + // Host config (mounts, resources). + hostCfg := &container.HostConfig{ + NetworkMode: container.NetworkMode(cfg.Network), + } + + // Mounts from params. + for _, m := range cfg.Mounts { + hostCfg.Mounts = append(hostCfg.Mounts, mount.Mount{ + Type: mount.TypeVolume, + Source: m.Source, + Target: m.Target, + ReadOnly: m.ReadOnly, + }) + } + + // Artifacts dir mount (from context). + if artDir := artifact.ArtifactsDirFromContext(ctx); artDir != "" { + // Bind mounts only work with local Docker daemons. + daemonHost := cli.DaemonHost() + if daemonHost != "" && !strings.HasPrefix(daemonHost, "unix://") { + return nil, fmt.Errorf("docker/run: artifact mounts are not supported with remote Docker daemons (host: %s); use s3/put inside the container instead", daemonHost) + } + hostCfg.Mounts = append(hostCfg.Mounts, mount.Mount{ + Type: mount.TypeBind, + Source: artDir, + Target: "/mantle/artifacts", + }) + } + + // Resource limits. + if cfg.Memory != "" { + mem, parseErr := parseMemoryString(cfg.Memory) + if parseErr != nil { + return nil, fmt.Errorf("docker/run: invalid memory limit: %w", parseErr) + } + hostCfg.Resources.Memory = mem + } + if cfg.CPUs > 0 { + hostCfg.Resources.NanoCPUs = int64(cfg.CPUs * 1e9) + } + + // Create container. + resp, err := cli.ContainerCreate(ctx, containerCfg, hostCfg, &network.NetworkingConfig{}, nil, "") + if err != nil { + return nil, fmt.Errorf("docker/run: creating container: %w", err) + } + containerID := resp.ID + + // Ensure cleanup. + defer func() { + if cfg.Remove { + cleanupCtx, cleanupCancel := context.WithTimeout(context.Background(), 30*time.Second) // #nosec G118 -- intentional: parent ctx may be cancelled + defer cleanupCancel() + if err := cli.ContainerRemove(cleanupCtx, containerID, container.RemoveOptions{Force: true}); err != nil { + log.Printf("docker/run: failed to remove container %s: %v", containerID, err) + } + } + }() + + // Attach for stdin if needed. + if cfg.Stdin != "" { + attach, attachErr := cli.ContainerAttach(ctx, containerID, container.AttachOptions{ + Stream: true, + Stdin: true, + }) + if attachErr != nil { + return nil, fmt.Errorf("docker/run: attaching stdin: %w", attachErr) + } + go func() { + defer attach.Close() + if _, err := io.Copy(attach.Conn, strings.NewReader(cfg.Stdin)); err != nil { + log.Printf("docker/run: failed writing stdin to container: %v", err) + } + attach.CloseWrite() + }() + } + + // Start container. + if err := cli.ContainerStart(ctx, containerID, container.StartOptions{}); err != nil { + return nil, fmt.Errorf("docker/run: starting container: %w", err) + } + + // Handle context cancellation → container stop. + doneCh := make(chan struct{}) + defer close(doneCh) + go func() { // #nosec G118 -- intentional context.Background: parent ctx is cancelled, need a live context to stop the container + select { + case <-ctx.Done(): + gracePeriod := 10 + stopCtx, stopCancel := context.WithTimeout(context.Background(), 30*time.Second) + defer stopCancel() + if err := cli.ContainerStop(stopCtx, containerID, container.StopOptions{ + Timeout: &gracePeriod, + }); err != nil { + log.Printf("docker/run: failed to stop container %s: %v", containerID, err) + } + case <-doneCh: + } + }() + + // Wait for container to finish. + statusCh, errCh := cli.ContainerWait(ctx, containerID, container.WaitConditionNotRunning) + var exitCode int64 + select { + case err := <-errCh: + if err != nil { + return nil, fmt.Errorf("docker/run: waiting for container: %w", err) + } + case status := <-statusCh: + exitCode = status.StatusCode + // Non-blocking check for a late error. + select { + case err := <-errCh: + if err != nil { + return nil, fmt.Errorf("docker/run: waiting for container: %w", err) + } + default: + } + } + + // Capture logs (stdout + stderr). + logReader, err := cli.ContainerLogs(ctx, containerID, container.LogsOptions{ + ShowStdout: true, + ShowStderr: true, + }) + if err != nil { + return nil, fmt.Errorf("docker/run: reading logs: %w", err) + } + defer logReader.Close() + + var stdoutBuf, stderrBuf strings.Builder + if _, err := stdcopy.StdCopy( + &limitWriter{w: &stdoutBuf, limit: DefaultMaxResponseBytes}, + &limitWriter{w: &stderrBuf, limit: DefaultMaxResponseBytes}, + logReader, + ); err != nil { + log.Printf("docker/run: failed to demultiplex container logs: %v", err) + } + + return map[string]any{ + "exit_code": exitCode, + "stdout": stdoutBuf.String(), + "stderr": stderrBuf.String(), + }, nil +} + +// pullImage handles image pulling based on the pull policy. +func pullImage(ctx context.Context, cli *client.Client, cfg *dockerConfig, registryAuth string) error { + switch cfg.Pull { + case "never": + return nil + case "always": + // Always pull — fall through. + case "missing": + // Check if image exists locally. + _, err := cli.ImageInspect(ctx, cfg.Image) + if err == nil { + return nil // already present + } + default: + return fmt.Errorf("docker/run: invalid pull policy %q (must be always, missing, or never)", cfg.Pull) + } + + reader, err := cli.ImagePull(ctx, cfg.Image, image.PullOptions{ + RegistryAuth: registryAuth, + }) + if err != nil { + return fmt.Errorf("docker/run: pulling image %q: %w", cfg.Image, err) + } + defer reader.Close() + // Drain the pull output to complete the pull. + if _, err := io.Copy(io.Discard, reader); err != nil { + return fmt.Errorf("docker/run: draining pull output: %w", err) + } + return nil +} diff --git a/internal/connector/docker_test.go b/internal/connector/docker_test.go new file mode 100644 index 0000000..705ca33 --- /dev/null +++ b/internal/connector/docker_test.go @@ -0,0 +1,241 @@ +package connector + +import ( + "context" + "strings" + "testing" + + "github.com/docker/docker/client" +) + +func TestDockerRun_MissingImage(t *testing.T) { + _, err := parseDockerParams(map[string]any{ + "cmd": []any{"echo", "hello"}, + }) + if err == nil { + t.Fatal("expected error for missing image") + } + if !strings.Contains(err.Error(), "image is required") { + t.Errorf("error = %q, want 'image is required'", err) + } +} + +func TestDockerRun_ParseParams(t *testing.T) { + params := map[string]any{ + "image": "alpine:latest", + "cmd": []any{"echo", "hello"}, + "env": map[string]any{"FOO": "bar"}, + "network": "host", + "pull": "never", + "memory": "256m", + "cpus": 1.5, + "remove": false, + } + + cfg, err := parseDockerParams(params) + if err != nil { + t.Fatalf("parseDockerParams: %v", err) + } + if cfg.Image != "alpine:latest" { + t.Errorf("image = %q", cfg.Image) + } + if len(cfg.Cmd) != 2 || cfg.Cmd[0] != "echo" { + t.Errorf("cmd = %v", cfg.Cmd) + } + if cfg.Env["FOO"] != "bar" { + t.Errorf("env = %v", cfg.Env) + } + if cfg.Network != "host" { + t.Errorf("network = %q", cfg.Network) + } + if cfg.Pull != "never" { + t.Errorf("pull = %q", cfg.Pull) + } + if cfg.Memory != "256m" { + t.Errorf("memory = %q", cfg.Memory) + } + if cfg.CPUs != 1.5 { + t.Errorf("cpus = %f", cfg.CPUs) + } + if cfg.Remove { + t.Errorf("remove = %v", cfg.Remove) + } +} + +func TestDockerRun_DefaultParams(t *testing.T) { + cfg, err := parseDockerParams(map[string]any{ + "image": "alpine", + }) + if err != nil { + t.Fatalf("parseDockerParams: %v", err) + } + if cfg.Network != "bridge" { + t.Errorf("default network = %q, want 'bridge'", cfg.Network) + } + if cfg.Pull != "missing" { + t.Errorf("default pull = %q, want 'missing'", cfg.Pull) + } + if cfg.Remove != true { + t.Errorf("default remove = %v, want true", cfg.Remove) + } +} + +func TestDockerRun_ParseMounts(t *testing.T) { + cfg, err := parseDockerParams(map[string]any{ + "image": "alpine", + "mounts": []any{ + map[string]any{ + "source": "my-volume", + "target": "/data", + "readonly": true, + }, + }, + }) + if err != nil { + t.Fatalf("parseDockerParams: %v", err) + } + if len(cfg.Mounts) != 1 { + t.Fatalf("mounts = %d, want 1", len(cfg.Mounts)) + } + if cfg.Mounts[0].Source != "my-volume" { + t.Errorf("source = %q", cfg.Mounts[0].Source) + } + if cfg.Mounts[0].Target != "/data" { + t.Errorf("target = %q", cfg.Mounts[0].Target) + } + if cfg.Mounts[0].ReadOnly != true { + t.Errorf("readonly = %v", cfg.Mounts[0].ReadOnly) + } +} + +func TestParseMemoryString(t *testing.T) { + tests := []struct { + input string + want int64 + }{ + {"256m", 256 * 1024 * 1024}, + {"1g", 1024 * 1024 * 1024}, + {"512k", 512 * 1024}, + {"1024", 1024}, + {"", 0}, + } + for _, tt := range tests { + got, err := parseMemoryString(tt.input) + if err != nil { + t.Errorf("parseMemoryString(%q): %v", tt.input, err) + continue + } + if got != tt.want { + t.Errorf("parseMemoryString(%q) = %d, want %d", tt.input, got, tt.want) + } + } +} + +func TestDockerRun_Integration_EchoStdout(t *testing.T) { + cli, err := client.NewClientWithOpts(client.FromEnv, client.WithAPIVersionNegotiation()) + if err != nil { + t.Skipf("Docker not available: %v", err) + } + defer cli.Close() + if _, err := cli.Ping(context.Background()); err != nil { + t.Skipf("Docker not reachable: %v", err) + } + + c := &DockerRunConnector{} + output, err := c.Execute(context.Background(), map[string]any{ + "image": "alpine:latest", + "cmd": []any{"echo", "hello mantle"}, + "pull": "missing", + }) + if err != nil { + t.Fatalf("Execute: %v", err) + } + if output["exit_code"] != int64(0) { + t.Errorf("exit_code = %v, want 0", output["exit_code"]) + } + stdout, _ := output["stdout"].(string) + if !strings.Contains(stdout, "hello mantle") { + t.Errorf("stdout = %q, want to contain 'hello mantle'", stdout) + } +} + +func TestDockerRun_Integration_NonZeroExitCode(t *testing.T) { + cli, err := client.NewClientWithOpts(client.FromEnv, client.WithAPIVersionNegotiation()) + if err != nil { + t.Skipf("Docker not available: %v", err) + } + defer cli.Close() + if _, err := cli.Ping(context.Background()); err != nil { + t.Skipf("Docker not reachable: %v", err) + } + + c := &DockerRunConnector{} + output, err := c.Execute(context.Background(), map[string]any{ + "image": "alpine:latest", + "cmd": []any{"sh", "-c", "echo error-output >&2; exit 42"}, + "pull": "missing", + }) + // Non-zero exit code is NOT a step failure. + if err != nil { + t.Fatalf("Execute should not error on non-zero exit: %v", err) + } + if output["exit_code"] != int64(42) { + t.Errorf("exit_code = %v, want 42", output["exit_code"]) + } + stderr, _ := output["stderr"].(string) + if !strings.Contains(stderr, "error-output") { + t.Errorf("stderr = %q, want to contain 'error-output'", stderr) + } +} + +func TestDockerRun_Integration_Stdin(t *testing.T) { + cli, err := client.NewClientWithOpts(client.FromEnv, client.WithAPIVersionNegotiation()) + if err != nil { + t.Skipf("Docker not available: %v", err) + } + defer cli.Close() + if _, err := cli.Ping(context.Background()); err != nil { + t.Skipf("Docker not reachable: %v", err) + } + + c := &DockerRunConnector{} + output, err := c.Execute(context.Background(), map[string]any{ + "image": "alpine:latest", + "cmd": []any{"cat"}, + "stdin": "piped input data", + "pull": "missing", + }) + if err != nil { + t.Fatalf("Execute: %v", err) + } + stdout, _ := output["stdout"].(string) + if !strings.Contains(stdout, "piped input data") { + t.Errorf("stdout = %q, want to contain 'piped input data'", stdout) + } +} + +func TestDockerRun_Integration_EnvVars(t *testing.T) { + cli, err := client.NewClientWithOpts(client.FromEnv, client.WithAPIVersionNegotiation()) + if err != nil { + t.Skipf("Docker not available: %v", err) + } + defer cli.Close() + if _, err := cli.Ping(context.Background()); err != nil { + t.Skipf("Docker not reachable: %v", err) + } + + c := &DockerRunConnector{} + output, err := c.Execute(context.Background(), map[string]any{ + "image": "alpine:latest", + "cmd": []any{"sh", "-c", "echo $MY_VAR"}, + "env": map[string]any{"MY_VAR": "hello-from-env"}, + "pull": "missing", + }) + if err != nil { + t.Fatalf("Execute: %v", err) + } + stdout, _ := output["stdout"].(string) + if !strings.Contains(stdout, "hello-from-env") { + t.Errorf("stdout = %q, want 'hello-from-env'", stdout) + } +} diff --git a/internal/db/migrations/013_execution_artifacts.sql b/internal/db/migrations/013_execution_artifacts.sql new file mode 100644 index 0000000..43d4a8b --- /dev/null +++ b/internal/db/migrations/013_execution_artifacts.sql @@ -0,0 +1,16 @@ +-- +goose Up +CREATE TABLE execution_artifacts ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + execution_id UUID NOT NULL REFERENCES workflow_executions(id) ON DELETE CASCADE, + step_name TEXT NOT NULL, + name TEXT NOT NULL, + url TEXT NOT NULL, + size BIGINT NOT NULL CHECK (size >= 0), + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + UNIQUE (execution_id, name) +); + +CREATE INDEX idx_execution_artifacts_created_at ON execution_artifacts(created_at); + +-- +goose Down +DROP TABLE IF EXISTS execution_artifacts; diff --git a/internal/engine/engine.go b/internal/engine/engine.go index 55cf7a1..3891ae0 100644 --- a/internal/engine/engine.go +++ b/internal/engine/engine.go @@ -6,9 +6,12 @@ import ( "encoding/json" "fmt" "log" + "os" + "path/filepath" "strings" "time" + "github.com/dvflw/mantle/internal/artifact" "github.com/dvflw/mantle/internal/audit" "github.com/dvflw/mantle/internal/auth" "github.com/dvflw/mantle/internal/budget" @@ -26,9 +29,11 @@ type Engine struct { 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 + 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 } // StepContext carries workflow-level metadata needed by step execution. @@ -138,6 +143,9 @@ func (e *Engine) resumeExecution(ctx context.Context, execID string, workflowNam celCtx.Steps[name] = map[string]any{"output": output} } + // Load artifacts for CEL context. + e.loadArtifactsIntoCELContext(ctx, execID, celCtx) + result := &ExecutionResult{ ExecutionID: execID, Status: "completed", @@ -175,6 +183,7 @@ func (e *Engine) resumeExecution(ctx context.Context, execID string, workflowNam if stepResult.Status == "completed" { celCtx.Steps[step.Name] = map[string]any{"output": stepResult.Output} sc.CompletedSteps[step.Name] = stepResult.Output + e.loadArtifactsIntoCELContext(ctx, execID, celCtx) } else if stepResult.Status == "skipped" { celCtx.Steps[step.Name] = map[string]any{"output": map[string]any{}} } else if stepResult.Status == "failed" { @@ -390,6 +399,23 @@ func (e *Engine) executeStepLogic(ctx context.Context, execID string, step workf resolvedParams["_credential"] = credData } + // Resolve registry credential for docker/run private image pulls. + if step.RegistryCredential != "" { + regCredData, regCredErr := e.Resolver.Resolve(ctx, step.RegistryCredential) + if regCredErr != nil { + return nil, fmt.Errorf("resolving registry credential %q: %v", step.RegistryCredential, regCredErr) + } + resolvedParams["_registry_credential"] = regCredData + } + + // 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) + } + } + // Inject workflow/step context for AI observability metrics. resolvedParams["_workflow"] = workflowName resolvedParams["_step"] = step.Name @@ -418,6 +444,20 @@ func (e *Engine) executeStepLogic(ctx context.Context, execID string, step workf var lastErr error for attempt := 1; attempt <= maxAttempts; attempt++ { + // Create a fresh artifacts scratch dir per attempt. + if len(step.Artifacts) > 0 && e.TmpStorage != nil { + if artifactsDir != "" { + os.RemoveAll(artifactsDir) // clean previous attempt's scratch + } + var tmpErr error + artifactsDir, tmpErr = os.MkdirTemp("", "mantle-artifacts-*") + if tmpErr != nil { + return nil, fmt.Errorf("creating artifacts dir: %v", tmpErr) + } + defer os.RemoveAll(artifactsDir) + ctx = artifact.WithArtifactsDir(ctx, artifactsDir) + } + // Apply per-step timeout. execCtx := ctx var cancel context.CancelFunc @@ -473,6 +513,77 @@ 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 != "" { + type persistedArtifact struct { + id string + url string + } + var persisted []persistedArtifact + + rollback := func() { + for _, p := range persisted { + 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 { + log.Printf("warning: rollback: failed to delete artifact blob %q: %v", p.url, delErr) + } + } + } + + for _, artDecl := range step.Artifacts { + relPath := filepath.Base(artDecl.Path) + localPath := filepath.Join(artifactsDir, relPath) + info, statErr := os.Stat(localPath) + if statErr != nil { + rollback() + return nil, fmt.Errorf("declared artifact %q not found at %q: %v", artDecl.Name, localPath, statErr) + } + + key := fmt.Sprintf("%s/%s/%s/%s", workflowName, execID, artDecl.Name, relPath) + url, putErr := e.TmpStorage.Put(ctx, key, localPath) + if putErr != nil { + rollback() + return nil, fmt.Errorf("persisting artifact %q: %v", artDecl.Name, putErr) + } + + art := &artifact.Artifact{ + ExecutionID: execID, + StepName: step.Name, + Name: artDecl.Name, + URL: url, + Size: info.Size(), + } + if createErr := e.ArtifactStore.Create(ctx, art); createErr != nil { + if delErr := e.TmpStorage.Delete(ctx, url); delErr != nil { + log.Printf("warning: failed to clean up orphaned artifact blob %q: %v", url, delErr) + } + rollback() + return nil, fmt.Errorf("recording artifact %q: %v", artDecl.Name, createErr) + } + + persisted = append(persisted, persistedArtifact{id: art.ID, url: url}) + + // Emit audit event for artifact persistence. + if err := e.Auditor.Emit(ctx, audit.Event{ + Timestamp: time.Now(), + Actor: "engine", + Action: audit.ActionArtifactPersisted, + Resource: audit.Resource{Type: "execution_artifact", ID: execID}, + Metadata: map[string]string{ + "step": step.Name, + "artifact_name": artDecl.Name, + "url": url, + "size": fmt.Sprintf("%d", info.Size()), + }, + TeamID: sc.TeamID, + }); err != nil { + log.Printf("warning: failed to emit artifact audit event: %v", err) + } + } + } + return output, lastErr } @@ -497,6 +608,10 @@ func (e *Engine) MakeStepExecutor(wf *workflow.Workflow, execID string, celCtx * sc.CompletedSteps[name] = out } } + + // Load artifacts for CEL context. + e.loadArtifactsIntoCELContext(ctx, execID, celCtx) + output, err := e.executeStepLogic(ctx, execID, *step, celCtx, wf.Name, sc) if err != nil { return output, err @@ -568,6 +683,9 @@ func (e *Engine) MakeGlobalStepExecutor() StepExecutor { celCtx.Steps[name] = map[string]any{"output": output} } + // Load artifacts for CEL context. + e.loadArtifactsIntoCELContext(ctx, execID, celCtx) + sc := StepContext{ WorkflowTokenBudget: wf.TokenBudget, CompletedSteps: completedSteps, @@ -586,6 +704,26 @@ func (e *Engine) MakeGlobalStepExecutor() StepExecutor { } } +// loadArtifactsIntoCELContext loads artifact metadata into the CEL context. +func (e *Engine) loadArtifactsIntoCELContext(ctx context.Context, execID string, celCtx *mantleCEL.Context) { + if e.ArtifactStore == nil { + return + } + arts, err := e.ArtifactStore.ListByExecution(ctx, execID) + if err != nil { + log.Printf("warning: loading artifacts for execution %s: %v", execID, err) + return + } + celCtx.Artifacts = make(map[string]map[string]any, len(arts)) + for _, a := range arts { + celCtx.Artifacts[a.Name] = map[string]any{ + "name": a.Name, + "url": a.URL, + "size": a.Size, + } + } +} + // loadWorkflow retrieves a workflow definition from the database. func (e *Engine) loadWorkflow(ctx context.Context, name string, version int) (*workflow.Workflow, error) { teamID := auth.TeamIDFromContext(ctx) diff --git a/internal/secret/types.go b/internal/secret/types.go index 6590e39..e954ddd 100644 --- a/internal/secret/types.go +++ b/internal/secret/types.go @@ -1,6 +1,10 @@ package secret -import "fmt" +import ( + "fmt" + "sort" + "strings" +) // FieldDef defines a credential field with its validation rules. type FieldDef struct { @@ -73,13 +77,27 @@ var builtinTypes = map[string]*CredentialType{ {Name: "session_token", Required: false}, }, }, + "docker": { + Name: "docker", + Fields: []FieldDef{ + {Name: "host", Required: false}, + {Name: "ca_cert", Required: false}, + {Name: "client_cert", Required: false}, + {Name: "client_key", Required: false}, + }, + }, } // GetType returns the credential type definition, or an error if unknown. func GetType(name string) (*CredentialType, error) { ct, ok := builtinTypes[name] if !ok { - return nil, fmt.Errorf("unknown credential type %q (available: generic, bearer, openai, basic, aws)", name) + available := make([]string, 0, len(builtinTypes)) + for k := range builtinTypes { + available = append(available, k) + } + sort.Strings(available) + return nil, fmt.Errorf("unknown credential type %q (available: %s)", name, strings.Join(available, ", ")) } return ct, nil } diff --git a/internal/secret/types_test.go b/internal/secret/types_test.go index b02c98a..923c5ff 100644 --- a/internal/secret/types_test.go +++ b/internal/secret/types_test.go @@ -51,3 +51,28 @@ func TestGetType_AllBuiltins(t *testing.T) { } } } + +func TestCredentialType_Docker(t *testing.T) { + ct, err := GetType("docker") + if err != nil { + t.Fatalf("GetType('docker'): %v", err) + } + if ct.Name != "docker" { + t.Errorf("name = %q, want 'docker'", ct.Name) + } + + // All fields are optional — empty data should be valid. + if err := ct.Validate(map[string]string{}); err != nil { + t.Errorf("empty data should be valid: %v", err) + } + + // Full TLS config should also be valid. + if err := ct.Validate(map[string]string{ + "host": "tcp://docker:2376", + "ca_cert": "ca-data", + "client_cert": "cert-data", + "client_key": "key-data", + }); err != nil { + t.Errorf("full TLS data should be valid: %v", err) + } +} diff --git a/internal/workflow/parse_test.go b/internal/workflow/parse_test.go index 969b386..38a536b 100644 --- a/internal/workflow/parse_test.go +++ b/internal/workflow/parse_test.go @@ -307,3 +307,39 @@ steps: t.Fatal("tool[1] input_schema is nil") } } + +func TestParse_ArtifactsAndRegistryCredential(t *testing.T) { + path := writeTestFile(t, ` +name: test-artifacts +description: test artifacts parsing +steps: + - name: build + action: docker/run + credential: my-docker + registry_credential: my-registry + params: + image: alpine + cmd: ["tar", "-czf", "/mantle/artifacts/out.tar.gz", "/data"] + artifacts: + - path: /mantle/artifacts/out.tar.gz + name: build-output +`) + result, err := Parse(path) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + step := result.Workflow.Steps[0] + if step.RegistryCredential != "my-registry" { + t.Errorf("registry_credential = %q, want %q", step.RegistryCredential, "my-registry") + } + if len(step.Artifacts) != 1 { + t.Fatalf("len(artifacts) = %d, want 1", len(step.Artifacts)) + } + if step.Artifacts[0].Name != "build-output" { + t.Errorf("artifact name = %q, want %q", step.Artifacts[0].Name, "build-output") + } + if step.Artifacts[0].Path != "/mantle/artifacts/out.tar.gz" { + t.Errorf("artifact path = %q, want %q", step.Artifacts[0].Path, "/mantle/artifacts/out.tar.gz") + } +} diff --git a/internal/workflow/validate.go b/internal/workflow/validate.go index 39d66c0..830d1b9 100644 --- a/internal/workflow/validate.go +++ b/internal/workflow/validate.go @@ -243,6 +243,37 @@ func Validate(result *ParseResult) []ValidationError { } } + // Validate artifact declarations. + artifactNames := make(map[string]string) // name -> step that declared it + for i, step := range w.Steps { + prefix := fmt.Sprintf("steps[%d]", i) + for j, art := range step.Artifacts { + artPrefix := fmt.Sprintf("%s.artifacts[%d]", prefix, j) + if art.Name == "" { + errs = append(errs, ValidationError{ + Field: artPrefix + ".name", + Message: "artifact name is required", + }) + } + if art.Path == "" { + errs = append(errs, ValidationError{ + Field: artPrefix + ".path", + Message: "artifact path is required", + }) + } + if art.Name != "" { + if prevStep, exists := artifactNames[art.Name]; exists { + errs = append(errs, ValidationError{ + Field: artPrefix + ".name", + Message: fmt.Sprintf("duplicate artifact name %q (also declared in step %q)", art.Name, prevStep), + }) + } else { + artifactNames[art.Name] = step.Name + } + } + } + } + // Validate dependency references (explicit + implicit from CEL expressions). for i, step := range w.Steps { prefix := fmt.Sprintf("steps[%d]", i) diff --git a/internal/workflow/validate_test.go b/internal/workflow/validate_test.go index 9b4ad37..ecb123a 100644 --- a/internal/workflow/validate_test.go +++ b/internal/workflow/validate_test.go @@ -515,3 +515,83 @@ steps: errs := Validate(result) assertNoErrors(t, errs) } + +func TestValidate_ArtifactDuplicateNames(t *testing.T) { + result := mustParse(t, ` +name: test-dup-artifacts +description: test +steps: + - name: step-a + action: docker/run + params: + image: alpine + artifacts: + - path: /mantle/artifacts/out.tar.gz + name: my-artifact + - name: step-b + action: docker/run + params: + image: alpine + artifacts: + - path: /mantle/artifacts/other.tar.gz + name: my-artifact +`) + errs := Validate(result) + if len(errs) == 0 { + t.Fatal("expected validation error for duplicate artifact name") + } + found := false + for _, e := range errs { + if strings.Contains(e.Message, "duplicate artifact name") { + found = true + } + } + if !found { + t.Errorf("expected 'duplicate artifact name' error, got: %v", errs) + } +} + +func TestValidate_ArtifactMissingFields(t *testing.T) { + result := mustParse(t, ` +name: test-missing-artifact-fields +description: test +steps: + - name: step-a + action: docker/run + params: + image: alpine + artifacts: + - path: "" + name: "" +`) + errs := Validate(result) + if len(errs) < 2 { + t.Fatalf("expected at least 2 errors for missing artifact fields, got %d: %v", len(errs), errs) + } +} + +func TestValidate_ArtifactValidDeclaration(t *testing.T) { + result := mustParse(t, ` +name: test-valid-artifacts +description: test +steps: + - name: step-a + action: docker/run + params: + image: alpine + artifacts: + - path: /mantle/artifacts/out.tar.gz + name: my-artifact +`) + errs := Validate(result) + // Filter to artifact-related errors only + var artifactErrs []ValidationError + for _, e := range errs { + if strings.Contains(e.Field, "artifact") { + artifactErrs = append(artifactErrs, e) + } + } + if len(artifactErrs) > 0 { + t.Errorf("unexpected artifact validation errors: %v", artifactErrs) + } +} diff --git a/internal/workflow/workflow.go b/internal/workflow/workflow.go index e9187a4..4a84c86 100644 --- a/internal/workflow/workflow.go +++ b/internal/workflow/workflow.go @@ -30,14 +30,16 @@ type Input struct { // Step defines a single step within a workflow. type Step struct { - Name string `yaml:"name"` - Action string `yaml:"action"` - Params map[string]any `yaml:"params"` - If string `yaml:"if"` - Retry *RetryPolicy `yaml:"retry"` - Timeout string `yaml:"timeout"` - Credential string `yaml:"credential"` - DependsOn []string `yaml:"depends_on"` + Name string `yaml:"name"` + Action string `yaml:"action"` + Params map[string]any `yaml:"params"` + If string `yaml:"if"` + Retry *RetryPolicy `yaml:"retry"` + Timeout string `yaml:"timeout"` + Credential string `yaml:"credential"` + DependsOn []string `yaml:"depends_on"` + RegistryCredential string `yaml:"registry_credential"` + Artifacts []ArtifactDecl `yaml:"artifacts"` } // FindStep returns a pointer to the step with the given name, or nil if not found. @@ -65,6 +67,19 @@ type Tool struct { Params map[string]any `yaml:"params"` } +// ArtifactDecl declares a file that a step will produce. +type ArtifactDecl struct { + Path string `yaml:"path"` // path inside the artifacts dir + Name string `yaml:"name"` // unique name for CEL reference +} + +// ArtifactRef is the runtime representation available in CEL expressions. +type ArtifactRef struct { + Name string `json:"name"` + URL string `json:"url"` // S3 URI or local filesystem path + Size int64 `json:"size"` +} + // ParseTools extracts typed Tool structs from an AI step's params["tools"]. // Returns nil, nil if no tools key is present. Returns an error if the tools // value is not an array or if any item is not a map. diff --git a/site/src/components/Connectors.astro b/site/src/components/Connectors.astro index 38f9912..b4d60b2 100644 --- a/site/src/components/Connectors.astro +++ b/site/src/components/Connectors.astro @@ -7,12 +7,13 @@ const connectors = [ { icon: ``, name: 'Email', desc: 'Send via SMTP, plaintext and HTML' }, { icon: ``, name: 'Postgres', desc: 'Parameterized SQL against external databases' }, { icon: ``, name: 'S3', desc: 'Put, get, list objects (S3-compatible)' }, + { icon: ``, name: 'Docker', desc: 'Run containers to completion, capture output' }, ]; ---

Built-in Connectors

-

7 connectors with 9 actions ship with the binary — plus a plugin system for anything else.

+

8 connectors with 10 actions ship with the binary — plus a plugin system for anything else.

{connectors.map((c) => ( diff --git a/site/src/components/Hero.astro b/site/src/components/Hero.astro index edc9b1b..f59c034 100644 --- a/site/src/components/Hero.astro +++ b/site/src/components/Hero.astro @@ -19,7 +19,7 @@

- Single binary. Postgres only. Bring your own keys. 9 connectors, infinite plugins. + Single binary. Postgres only. Bring your own keys. 8 connectors, infinite plugins.

@@ -35,8 +35,8 @@
- 9 built-in connectors - 17 example workflows + 8 built-in connectors + 22 example workflows Source-available (BSL 1.1)
diff --git a/site/src/components/Nav.astro b/site/src/components/Nav.astro index 38561e8..0e0c217 100644 --- a/site/src/components/Nav.astro +++ b/site/src/components/Nav.astro @@ -14,7 +14,7 @@ const isDocs = Astro.url.pathname === '/docs' || Astro.url.pathname.startsWith(' -