Feat v0.3.2 workflow engine (#16)

Co-authored-by: Jeffrey Smith <jasafpro@gmail.com> Co-committed-by: Jeffrey Smith <jasafpro@gmail.com>
2026-03-27 21:37:44 +00:00
parent 01ee9e668b
commit ab28e4b784
13 changed files with 1652 additions and 24 deletions
--- a/server/workflow/automated.go
+++ b/server/workflow/automated.go
@@ -0,0 +1,237 @@
+package workflow
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"log"
+	"time"
+
+	"switchboard-core/events"
+	"switchboard-core/models"
+
+	"go.starlark.net/starlark"
+)
+
+// processAutomatedStage fires the Starlark hook for an automated stage
+// and optionally auto-advances the instance.
+func (e *Engine) processAutomatedStage(ctx context.Context, inst *models.WorkflowInstance, stages []models.WorkflowStage, autoDepth int) error {
+	if autoDepth >= MaxConsecutiveAutomated {
+		// Cycle guard: mark instance as error
+		errInst := *inst
+		errInst.Status = models.InstanceStatusError
+		errInst.Metadata = json.RawMessage(fmt.Sprintf(
+			`{"error":"exceeded %d consecutive automated stages","depth":%d}`,
+			MaxConsecutiveAutomated, autoDepth))
+		if err := e.stores.Workflows.UpdateInstance(ctx, &errInst); err != nil {
+			log.Printf("[workflow-automated] failed to set error status: %v", err)
+		}
+		e.emitError(ctx, inst.ID, "cycle guard: exceeded max consecutive automated stages")
+		return fmt.Errorf("cycle guard: %d consecutive automated stages", autoDepth)
+	}
+
+	// Re-read instance to get current state
+	inst, err := e.stores.Workflows.GetInstance(ctx, inst.ID)
+	if err != nil {
+		return err
+	}
+	if inst.Status != models.InstanceStatusActive {
+		return nil // instance was completed/cancelled during processing
+	}
+
+	// Find current stage definition
+	var stage *models.WorkflowStage
+	for i := range stages {
+		if stages[i].Name == inst.CurrentStage {
+			stage = &stages[i]
+			break
+		}
+	}
+	if stage == nil {
+		return fmt.Errorf("stage %q not found", inst.CurrentStage)
+	}
+
+	if stage.StageMode != models.StageModeAutomated || stage.StarlarkHook == nil {
+		return nil
+	}
+
+	if e.runner == nil {
+		log.Printf("[workflow-automated] no Starlark runner, skipping hook for stage %q", stage.Name)
+		return nil
+	}
+
+	// Parse hook reference: "package_id:entry_point" or just "package_id"
+	hookPkgID, hookEntry := parseHookRef(*stage.StarlarkHook)
+	if hookPkgID == "" {
+		return fmt.Errorf("invalid starlark_hook: %q", *stage.StarlarkHook)
+	}
+
+	pkg, err := e.stores.Packages.Get(ctx, hookPkgID)
+	if err != nil || pkg == nil {
+		log.Printf("[workflow-automated] package %q not found for hook", hookPkgID)
+		return fmt.Errorf("hook package not found: %s", hookPkgID)
+	}
+
+	// Build context dict for the hook
+	ctxDict := starlark.NewDict(4)
+	_ = ctxDict.SetKey(starlark.String("instance_id"), starlark.String(inst.ID))
+	_ = ctxDict.SetKey(starlark.String("current_stage"), starlark.String(inst.CurrentStage))
+	_ = ctxDict.SetKey(starlark.String("workflow_id"), starlark.String(inst.WorkflowID))
+
+	// Parse stage_data into Starlark dict
+	var dataMap map[string]interface{}
+	if json.Unmarshal(inst.StageData, &dataMap) == nil {
+		_ = ctxDict.SetKey(starlark.String("stage_data"), goToStarlark(dataMap))
+	} else {
+		_ = ctxDict.SetKey(starlark.String("stage_data"), starlark.NewDict(0))
+	}
+
+	val, _, err := e.runner.CallEntryPoint(ctx, pkg, hookEntry,
+		starlark.Tuple{ctxDict}, nil, nil)
+	if err != nil {
+		log.Printf("[workflow-automated] hook error: %v", err)
+		e.emitError(ctx, inst.ID, fmt.Sprintf("automated hook error: %v", err))
+		return err
+	}
+
+	return e.handleHookResult(ctx, inst, val, autoDepth)
+}
+
+// handleHookResult interprets the Starlark hook return value.
+func (e *Engine) handleHookResult(ctx context.Context, inst *models.WorkflowInstance, val starlark.Value, autoDepth int) error {
+	if val == nil || val == starlark.None {
+		return nil // no action — leave instance at current stage
+	}
+
+	d, ok := val.(*starlark.Dict)
+	if !ok {
+		return nil
+	}
+
+	// Check for error
+	if errVal, found, _ := d.Get(starlark.String("error")); found {
+		if s, ok := errVal.(starlark.String); ok {
+			errInst := *inst
+			errInst.Status = models.InstanceStatusError
+			errInst.Metadata = json.RawMessage(fmt.Sprintf(`{"error":%q}`, string(s)))
+			e.stores.Workflows.UpdateInstance(ctx, &errInst)
+			e.emitError(ctx, inst.ID, string(s))
+			return fmt.Errorf("hook returned error: %s", string(s))
+		}
+	}
+
+	// Check for advance
+	if advVal, found, _ := d.Get(starlark.String("advance")); found {
+		if advVal == starlark.True {
+			// Extract enriched data if present
+			var enrichedData json.RawMessage
+			if dataVal, found, _ := d.Get(starlark.String("data")); found {
+				if sd, ok := dataVal.(*starlark.Dict); ok {
+					goMap := starlarkDictToMap(sd)
+					if data, err := json.Marshal(goMap); err == nil {
+						enrichedData = data
+					}
+				}
+			}
+			_, err := e.advanceInternal(ctx, inst.ID, enrichedData, inst.StartedBy, autoDepth)
+			return err
+		}
+	}
+
+	return nil // advance: false or not set — leave at current stage
+}
+
+// emitError publishes a workflow.error event.
+func (e *Engine) emitError(ctx context.Context, instanceID, message string) {
+	if e.bus == nil {
+		return
+	}
+	e.bus.Publish(events.Event{
+		Label:   "workflow.error",
+		Room:    instanceID,
+		Payload: events.MustJSON(map[string]any{"instance_id": instanceID, "error": message}),
+		Ts:      time.Now().UnixMilli(),
+	})
+}
+
+// parseHookRef splits "package_id:entry_point" or returns (pkg, "on_run").
+func parseHookRef(ref string) (string, string) {
+	for i, c := range ref {
+		if c == ':' {
+			return ref[:i], ref[i+1:]
+		}
+	}
+	return ref, "on_run"
+}
+
+// ── Starlark conversion helpers ─────────────
+
+func goToStarlark(v any) starlark.Value {
+	switch val := v.(type) {
+	case nil:
+		return starlark.None
+	case bool:
+		return starlark.Bool(val)
+	case float64:
+		if val == float64(int(val)) {
+			return starlark.MakeInt(int(val))
+		}
+		return starlark.Float(val)
+	case string:
+		return starlark.String(val)
+	case map[string]interface{}:
+		d := starlark.NewDict(len(val))
+		for k, v := range val {
+			_ = d.SetKey(starlark.String(k), goToStarlark(v))
+		}
+		return d
+	case []interface{}:
+		elems := make([]starlark.Value, len(val))
+		for i, v := range val {
+			elems[i] = goToStarlark(v)
+		}
+		return starlark.NewList(elems)
+	default:
+		return starlark.String(fmt.Sprintf("%v", v))
+	}
+}
+
+func starlarkDictToMap(d *starlark.Dict) map[string]any {
+	result := make(map[string]any, d.Len())
+	for _, item := range d.Items() {
+		k, ok := item[0].(starlark.String)
+		if !ok {
+			continue
+		}
+		result[string(k)] = starlarkToGo(item[1])
+	}
+	return result
+}
+
+func starlarkToGo(v starlark.Value) any {
+	switch val := v.(type) {
+	case starlark.NoneType:
+		return nil
+	case starlark.Bool:
+		return bool(val)
+	case starlark.Int:
+		if i, ok := val.Int64(); ok {
+			return i
+		}
+		return val.String()
+	case starlark.Float:
+		return float64(val)
+	case starlark.String:
+		return string(val)
+	case *starlark.List:
+		result := make([]any, val.Len())
+		for i := 0; i < val.Len(); i++ {
+			result[i] = starlarkToGo(val.Index(i))
+		}
+		return result
+	case *starlark.Dict:
+		return starlarkDictToMap(val)
+	default:
+		return v.String()
+	}
+}
--- a/server/workflow/engine.go
+++ b/server/workflow/engine.go
@@ -0,0 +1,282 @@
+package workflow
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"log"
+	"time"
+
+	"switchboard-core/events"
+	"switchboard-core/models"
+	"switchboard-core/sandbox"
+	"switchboard-core/store"
+
+	"github.com/google/uuid"
+)
+
+// MaxConsecutiveAutomated is the cycle guard limit for automated stages.
+const MaxConsecutiveAutomated = 10
+
+// Engine orchestrates workflow instance lifecycle.
+type Engine struct {
+	stores store.Stores
+	bus    *events.Bus
+	runner *sandbox.Runner
+}
+
+// NewEngine creates a workflow engine.
+func NewEngine(stores store.Stores, bus *events.Bus, runner *sandbox.Runner) *Engine {
+	return &Engine{stores: stores, bus: bus, runner: runner}
+}
+
+// Start creates a new workflow instance and kicks off the first stage.
+func (e *Engine) Start(ctx context.Context, workflowID string, initialData json.RawMessage, userID string) (*models.WorkflowInstance, error) {
+	wf, err := e.stores.Workflows.GetByID(ctx, workflowID)
+	if err != nil {
+		return nil, fmt.Errorf("workflow not found: %w", err)
+	}
+	if !wf.IsActive {
+		return nil, fmt.Errorf("workflow %q is not active", wf.Name)
+	}
+
+	ver, err := e.stores.Workflows.GetLatestVersion(ctx, workflowID)
+	if err != nil {
+		return nil, fmt.Errorf("no published version: %w", err)
+	}
+
+	var stages []models.WorkflowStage
+	if err := json.Unmarshal(ver.Snapshot, &stages); err != nil {
+		return nil, fmt.Errorf("corrupt version snapshot: %w", err)
+	}
+	if len(stages) == 0 {
+		return nil, fmt.Errorf("workflow has no stages")
+	}
+
+	inst := &models.WorkflowInstance{
+		WorkflowID:      workflowID,
+		WorkflowVersion: ver.VersionNumber,
+		CurrentStage:    stages[0].Name,
+		StageData:       initialData,
+		Status:          models.InstanceStatusActive,
+		StartedBy:       userID,
+		Metadata:        json.RawMessage(`{}`),
+	}
+
+	if wf.EntryMode == "public_link" {
+		tok := uuid.New().String()
+		inst.EntryToken = &tok
+	}
+
+	if err := e.stores.Workflows.CreateInstance(ctx, inst); err != nil {
+		return nil, fmt.Errorf("create instance: %w", err)
+	}
+
+	// Create initial assignment if first stage has a team
+	if stages[0].AssignmentTeamID != nil {
+		a := &models.WorkflowAssignment{
+			InstanceID: inst.ID,
+			Stage:      stages[0].Name,
+			TeamID:     *stages[0].AssignmentTeamID,
+		}
+		if err := e.stores.Workflows.CreateAssignment(ctx, a); err != nil {
+			log.Printf("[workflow-engine] create initial assignment: %v", err)
+		}
+	}
+
+	e.emit(ctx, "workflow.started", inst.ID, map[string]any{
+		"workflow_id": workflowID,
+		"instance_id": inst.ID,
+		"started_by":  userID,
+	})
+
+	// If first stage is automated, process it
+	if stages[0].StageMode == models.StageModeAutomated {
+		if err := e.processAutomatedStage(ctx, inst, stages, 0); err != nil {
+			log.Printf("[workflow-engine] automated stage error on start: %v", err)
+		}
+		// Re-read instance after automated processing
+		inst, _ = e.stores.Workflows.GetInstance(ctx, inst.ID)
+	}
+
+	return inst, nil
+}
+
+// Advance moves an instance to the next stage.
+func (e *Engine) Advance(ctx context.Context, instanceID string, stageData json.RawMessage, userID string) (*models.WorkflowInstance, error) {
+	return e.advanceInternal(ctx, instanceID, stageData, userID, 0)
+}
+
+func (e *Engine) advanceInternal(ctx context.Context, instanceID string, stageData json.RawMessage, userID string, autoDepth int) (*models.WorkflowInstance, error) {
+	inst, err := e.stores.Workflows.GetInstance(ctx, instanceID)
+	if err != nil {
+		return nil, fmt.Errorf("instance not found: %w", err)
+	}
+	if inst.Status != models.InstanceStatusActive {
+		return nil, fmt.Errorf("instance is %s, not active", inst.Status)
+	}
+
+	ver, err := e.stores.Workflows.GetVersion(ctx, inst.WorkflowID, inst.WorkflowVersion)
+	if err != nil {
+		return nil, fmt.Errorf("version not found: %w", err)
+	}
+
+	var stages []models.WorkflowStage
+	if err := json.Unmarshal(ver.Snapshot, &stages); err != nil {
+		return nil, fmt.Errorf("corrupt snapshot: %w", err)
+	}
+
+	// Find current stage ordinal
+	currentOrdinal := -1
+	for _, s := range stages {
+		if s.Name == inst.CurrentStage {
+			currentOrdinal = s.Ordinal
+			break
+		}
+	}
+	if currentOrdinal < 0 {
+		return nil, fmt.Errorf("current stage %q not found in snapshot", inst.CurrentStage)
+	}
+	currentStage := stages[currentOrdinal]
+
+	// Merge stage data
+	merged := mergeJSON(inst.StageData, stageData)
+
+	// Resolve next stage
+	nextOrdinal, err := ResolveNextStage(stages, currentOrdinal, merged)
+	if err != nil {
+		return nil, fmt.Errorf("resolve next stage: %w", err)
+	}
+
+	// Terminal — complete the instance
+	if nextOrdinal >= len(stages) {
+		if err := e.stores.Workflows.CompleteInstance(ctx, instanceID); err != nil {
+			return nil, fmt.Errorf("complete: %w", err)
+		}
+		e.emit(ctx, "workflow.completed", instanceID, map[string]any{
+			"instance_id": instanceID,
+		})
+		return e.stores.Workflows.GetInstance(ctx, instanceID)
+	}
+
+	nextStage := stages[nextOrdinal]
+
+	// Advance in store
+	if err := e.stores.Workflows.AdvanceStage(ctx, instanceID, nextStage.Name, merged); err != nil {
+		return nil, fmt.Errorf("advance store: %w", err)
+	}
+
+	// Cancel old assignments for the previous stage
+	oldAssignments, _ := e.stores.Workflows.ListAssignmentsByInstance(ctx, instanceID)
+	for _, a := range oldAssignments {
+		if a.Stage == currentStage.Name && (a.Status == models.AssignmentStatusUnassigned || a.Status == models.AssignmentStatusClaimed) {
+			e.stores.Workflows.CancelAssignment(ctx, a.ID)
+		}
+	}
+
+	// Create new assignment if next stage has a team
+	if nextStage.AssignmentTeamID != nil {
+		a := &models.WorkflowAssignment{
+			InstanceID: instanceID,
+			Stage:      nextStage.Name,
+			TeamID:     *nextStage.AssignmentTeamID,
+		}
+		if err := e.stores.Workflows.CreateAssignment(ctx, a); err != nil {
+			log.Printf("[workflow-engine] create assignment: %v", err)
+		} else {
+			e.emit(ctx, "workflow.assigned", a.TeamID, map[string]any{
+				"instance_id":   instanceID,
+				"stage":         nextStage.Name,
+				"assignment_id": a.ID,
+				"team_id":       a.TeamID,
+			})
+		}
+	}
+
+	e.emit(ctx, "workflow.advanced", instanceID, map[string]any{
+		"instance_id":    instanceID,
+		"previous_stage": currentStage.Name,
+		"current_stage":  nextStage.Name,
+	})
+
+	// If next stage is automated, process it
+	if nextStage.StageMode == models.StageModeAutomated {
+		if err := e.processAutomatedStage(ctx, inst, stages, autoDepth+1); err != nil {
+			log.Printf("[workflow-engine] automated stage error: %v", err)
+		}
+	}
+
+	return e.stores.Workflows.GetInstance(ctx, instanceID)
+}
+
+// Cancel terminates an active instance and all its open assignments.
+func (e *Engine) Cancel(ctx context.Context, instanceID string, userID string) error {
+	inst, err := e.stores.Workflows.GetInstance(ctx, instanceID)
+	if err != nil {
+		return fmt.Errorf("instance not found: %w", err)
+	}
+	if inst.Status != models.InstanceStatusActive {
+		return fmt.Errorf("instance is %s, not active", inst.Status)
+	}
+
+	if err := e.stores.Workflows.CancelInstance(ctx, instanceID); err != nil {
+		return fmt.Errorf("cancel: %w", err)
+	}
+
+	// Cancel all open assignments
+	assignments, _ := e.stores.Workflows.ListAssignmentsByInstance(ctx, instanceID)
+	for _, a := range assignments {
+		if a.Status == models.AssignmentStatusUnassigned || a.Status == models.AssignmentStatusClaimed {
+			e.stores.Workflows.CancelAssignment(ctx, a.ID)
+		}
+	}
+
+	e.emit(ctx, "workflow.cancelled", instanceID, map[string]any{
+		"instance_id": instanceID,
+		"cancelled_by": userID,
+	})
+
+	return nil
+}
+
+// emit publishes an event on the bus.
+func (e *Engine) emit(_ context.Context, label, room string, payload map[string]any) {
+	if e.bus == nil {
+		return
+	}
+	e.bus.Publish(events.Event{
+		Label:   label,
+		Room:    room,
+		Payload: events.MustJSON(payload),
+		Ts:      time.Now().UnixMilli(),
+	})
+}
+
+// mergeJSON merges b into a (shallow). Returns a if b is empty.
+func mergeJSON(a, b json.RawMessage) json.RawMessage {
+	if len(b) == 0 || string(b) == "{}" || string(b) == "null" {
+		if len(a) == 0 {
+			return json.RawMessage(`{}`)
+		}
+		return a
+	}
+	if len(a) == 0 || string(a) == "{}" || string(a) == "null" {
+		return b
+	}
+
+	var ma, mb map[string]json.RawMessage
+	if json.Unmarshal(a, &ma) != nil {
+		return b
+	}
+	if json.Unmarshal(b, &mb) != nil {
+		return a
+	}
+	for k, v := range mb {
+		ma[k] = v
+	}
+	merged, err := json.Marshal(ma)
+	if err != nil {
+		return b
+	}
+	return merged
+}