core/server/workflow/automated.go

package workflow

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"time"

	"armature/events"
	"armature/models"
	"armature/sandbox"

	"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(5)
	_ = 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))
	_ = ctxDict.SetKey(starlark.String("started_by"), starlark.String(inst.StartedBy))

	// Parse stage_data into Starlark dict
	var dataMap map[string]interface{}
	if json.Unmarshal(inst.StageData, &dataMap) == nil {
		_ = ctxDict.SetKey(starlark.String("stage_data"), sandbox.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 := sandbox.DictToMap(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"
}