Minimal pipeline engine that loads steps from YAML config and runs them sequentially with a global timeout.

pipeline_runner.go

package pipeline

import (
	"context"
	"fmt"
	"log"
	"time"
)

// Step defines a single pipeline step from YAML config.
type Step struct {
	Name   string            `yaml:"name"`
	Type   string            `yaml:"type"`   // "deterministic", "ai", "approval"
	Action string            `yaml:"action"` // for deterministic steps
	Skill  string            `yaml:"skill"`  // for AI steps
	Vars   map[string]string `yaml:"vars"`
}

// Pipeline defines a named sequence of steps.
type Pipeline struct {
	Name    string        `yaml:"name"`
	Timeout time.Duration `yaml:"timeout"`
	Steps   []Step        `yaml:"steps"`
}

// StepHandler processes a single step. Receives and returns a shared data map.
type StepHandler func(ctx context.Context, step Step, data map[string]interface{}) error

// Runner executes pipelines with registered step handlers.
type Runner struct {
	handlers map[string]StepHandler
}

func NewRunner() *Runner {
	return &Runner{handlers: make(map[string]StepHandler)}
}

// RegisterHandler maps a step type to its handler.
func (r *Runner) RegisterHandler(stepType string, h StepHandler) {
	r.handlers[stepType] = h
}

// Run executes all steps in sequence with a pipeline-level timeout.
func (r *Runner) Run(p Pipeline, data map[string]interface{}) error {
	timeout := p.Timeout
	if timeout == 0 {
		timeout = 5 * time.Minute
	}
	ctx, cancel := context.WithTimeout(context.Background(), timeout)
	defer cancel()

	log.Printf("[pipeline:%s] starting (%d steps, timeout %s)", p.Name, len(p.Steps), timeout)

	for _, step := range p.Steps {
		select {
		case <-ctx.Done():
			return fmt.Errorf("pipeline %s timed out at step %s", p.Name, step.Name)
		default:
		}

		handler, ok := r.handlers[step.Type]
		if !ok {
			return fmt.Errorf("no handler for step type %q in step %s", step.Type, step.Name)
		}

		log.Printf("[pipeline:%s][step:%s] type=%s", p.Name, step.Name, step.Type)
		if err := handler(ctx, step, data); err != nil {
			return fmt.Errorf("step %s failed: %w", step.Name, err)
		}
	}

	log.Printf("[pipeline:%s] completed", p.Name)
	return nil
}

// ────────────────────────────────────────────────────────────
// Full implementation: https://github.com/renezander030/draftcat (Go, MIT)
// Production pipeline engine with Gmail / GoHighLevel connectors,
// PDF cite-verification, and Telegram / Slack approval gates.