Feat v0.3.2 workflow engine + handlers
Some checks failed
CI/CD / detect-changes (pull_request) Successful in 4s
CI/CD / test-frontend (pull_request) Has been skipped
CI/CD / test-go-pg (pull_request) Failing after 2m43s
CI/CD / test-sqlite (pull_request) Successful in 2m57s
CI/CD / build-and-deploy (pull_request) Has been skipped
Some checks failed
CI/CD / detect-changes (pull_request) Successful in 4s
CI/CD / test-frontend (pull_request) Has been skipped
CI/CD / test-go-pg (pull_request) Failing after 2m43s
CI/CD / test-sqlite (pull_request) Successful in 2m57s
CI/CD / build-and-deploy (pull_request) Has been skipped
Workflow execution engine with Start/Advance/Cancel lifecycle, automated stage processor with Starlark hook firing and cycle guard (max 10), instance and assignment HTTP handlers, store round-trip tests for all v0.3.1 methods, and Starlark module expansion (get_instance, list_instances). New routes: - POST/GET /workflows/:id/instances (start, list) - GET /workflows/:id/instances/:iid (get) - POST /workflows/:id/instances/:iid/advance (advance) - POST /workflows/:id/instances/:iid/cancel (cancel) - POST /assignments/:id/claim|unclaim|complete|cancel - GET /assignments/mine - Team-scoped mirrors under /teams/:teamId/... Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
237
server/workflow/automated.go
Normal file
237
server/workflow/automated.go
Normal file
@@ -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()
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user