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() } }