diff --git a/ROADMAP.md b/ROADMAP.md index fe4f9d4..cf52b8f 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -132,24 +132,25 @@ finalizes the extension lifecycle model. | Handler + Starlark updates | ✅ | Stage CRUD validation for new fields. Package export/import updated. Hook handler reads `stage_config`. Starlark module exposes `audience`, `stage_type`. | | Frontend updates | ✅ | Team-admin and admin stage editors updated with new modes, audience selector, stage type selector, conditional Starlark hook input. Fixed admin `STAGE_MODES` bug. | -### v0.3.1 — Instance + Assignment Tables + Store +### v0.3.1 — Instance + Assignment Tables + Store (complete) | Step | Status | Description | |------|--------|-------------| -| Instance schema | ⬚ | Add `workflow_instances` table to 007_workflows.sql (PG + SQLite). Tracks execution state: workflow_version, current_stage, stage_data, status, entry_token. | -| Assignment schema | ⬚ | Add `workflow_assignments` table. Per-stage queue: instance_id, stage, team_id, assigned_to, status, review_data. Optimistic claim lock. | -| Models + store interface | ⬚ | `WorkflowInstance`, `WorkflowAssignment` structs. Store methods for instance lifecycle (Create, Get, Advance, Complete, Cancel, MarkStale) and assignments (Create, Claim, Unclaim, Complete). | -| Event types | ⬚ | Add `workflow.started`, `workflow.cancelled`, `workflow.sla.warning`, `workflow.sla.breached`, `workflow.error` to bus. | +| Instance schema | ✅ | `workflow_instances` table in 007_workflows.sql (PG + SQLite). Tracks execution state: workflow_version, current_stage, stage_data, status, entry_token. | +| Assignment schema | ✅ | `workflow_assignments` table. Per-stage queue: instance_id, stage, team_id, assigned_to, status, review_data. Optimistic claim lock. | +| Models + store interface | ✅ | `WorkflowInstance`, `WorkflowAssignment` structs. 15 store methods for instance lifecycle and assignments across PG + SQLite. | +| Event types | ✅ | `workflow.started`, `workflow.cancelled`, `workflow.error` added to bus route table. | -### v0.3.2 — Workflow Engine + Handlers +### v0.3.2 — Workflow Engine + Handlers (complete) | Step | Status | Description | |------|--------|-------------| -| Stage execution engine | ⬚ | `server/workflow/engine.go` — advance lifecycle: validate → merge data → fire hooks → resolve next → create assignments → emit events. | -| Automated stages | ⬚ | `server/workflow/automated.go` — fire Starlark hook, auto-advance, cycle guard (max 10 consecutive). | -| Instance handlers | ⬚ | HTTP API: Start, GetInstance, Advance, Cancel, ListInstances. | -| Assignment handlers | ⬚ | HTTP API: Claim, Unclaim, Complete, Cancel, ListByTeam, ListMine. | -| Starlark module expansion | ⬚ | `workflow.get_instance()`, `workflow.advance()`, etc. | +| Store tests | ✅ | Round-trip tests for all 15 v0.3.1 store methods (instance + assignment CRUD). TruncateAll updated for new tables. | +| Stage execution engine | ✅ | `server/workflow/engine.go` — Start, Advance (with branch_rules), Cancel. Merges stage_data, creates assignments, emits events. | +| Automated stages | ✅ | `server/workflow/automated.go` — fire Starlark hook, auto-advance, cycle guard (max 10 consecutive). | +| Instance handlers | ✅ | HTTP API: Start, GetInstance, Advance, Cancel, ListInstances. Team-scoped mirrors. | +| Assignment handlers | ✅ | HTTP API: Claim, Unclaim, Complete, Cancel, ListByTeam, ListMine. Claimer identity verification. | +| Starlark module expansion | ✅ | `workflow.get_instance()`, `workflow.list_instances()` (read-only). Mutating builtins deferred pending interface extraction. | ### v0.3.3 — Public Entry + Background Jobs diff --git a/server/database/testhelper.go b/server/database/testhelper.go index 5867a83..2fc7643 100644 --- a/server/database/testhelper.go +++ b/server/database/testhelper.go @@ -256,7 +256,9 @@ func TruncateAll(t *testing.T) { // HA "rate_limit_counters", "ws_tickets", - // Workflows + // Workflows (FK order: assignments → instances → versions → stages → workflows) + "workflow_assignments", + "workflow_instances", "workflow_versions", "workflow_stages", "workflows", diff --git a/server/events/types.go b/server/events/types.go index ec0ef91..7cc7520 100644 --- a/server/events/types.go +++ b/server/events/types.go @@ -57,11 +57,14 @@ var routeTable = map[string]Direction{ // Workspace (v0.21.5) - // Workflow (v0.27.0) - "workflow.assigned": DirToClient, // new assignment → team members - "workflow.claimed": DirToClient, // assignment claimed → team + claimer - "workflow.advanced": DirToClient, // stage advanced → channel participants - "workflow.completed": DirToClient, // workflow finished → channel participants + // Workflow (v0.27.0, v0.3.2) + "workflow.started": DirToClient, // instance started + "workflow.assigned": DirToClient, // new assignment → team members + "workflow.claimed": DirToClient, // assignment claimed → team + claimer + "workflow.advanced": DirToClient, // stage advanced → instance participants + "workflow.completed": DirToClient, // workflow finished → instance participants + "workflow.cancelled": DirToClient, // instance cancelled + "workflow.error": DirToClient, // engine/hook error // Plugin hooks — never cross the wire "plugin.hook.": DirLocal, diff --git a/server/handlers/workflow_assignment_handlers.go b/server/handlers/workflow_assignment_handlers.go new file mode 100644 index 0000000..eee7eba --- /dev/null +++ b/server/handlers/workflow_assignment_handlers.go @@ -0,0 +1,172 @@ +package handlers + +import ( + "database/sql" + "encoding/json" + "net/http" + + "github.com/gin-gonic/gin" + + "switchboard-core/models" + "switchboard-core/store" + "switchboard-core/workflow" +) + +// ── Assignment Handlers ───────────────────── + +// WorkflowAssignmentHandler manages workflow assignment HTTP endpoints. +type WorkflowAssignmentHandler struct { + engine *workflow.Engine + stores store.Stores +} + +// NewWorkflowAssignmentHandler creates an assignment handler. +func NewWorkflowAssignmentHandler(engine *workflow.Engine, stores store.Stores) *WorkflowAssignmentHandler { + return &WorkflowAssignmentHandler{engine: engine, stores: stores} +} + +// Claim claims an assignment for the current user. +// POST /api/v1/assignments/:id/claim +func (h *WorkflowAssignmentHandler) Claim(c *gin.Context) { + userID := c.GetString("user_id") + assignmentID := c.Param("id") + + if err := h.stores.Workflows.ClaimAssignment(c.Request.Context(), assignmentID, userID); err != nil { + c.JSON(http.StatusConflict, gin.H{"error": err.Error()}) + return + } + + c.JSON(http.StatusOK, gin.H{"claimed": true}) +} + +// Unclaim releases a claimed assignment back to the queue. +// POST /api/v1/assignments/:id/unclaim +func (h *WorkflowAssignmentHandler) Unclaim(c *gin.Context) { + userID := c.GetString("user_id") + assignmentID := c.Param("id") + + // Verify current user is the claimer + userAssignments, err := h.stores.Workflows.ListAssignmentsByUser(c.Request.Context(), userID, models.AssignmentStatusClaimed) + if err == nil { + found := false + for _, a := range userAssignments { + if a.ID == assignmentID { + found = true + break + } + } + if !found { + c.JSON(http.StatusForbidden, gin.H{"error": "assignment not claimed by you"}) + return + } + } + + if err := h.stores.Workflows.UnclaimAssignment(c.Request.Context(), assignmentID); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + + c.JSON(http.StatusOK, gin.H{"unclaimed": true}) +} + +// Complete marks a claimed assignment as completed. +// POST /api/v1/assignments/:id/complete +func (h *WorkflowAssignmentHandler) Complete(c *gin.Context) { + userID := c.GetString("user_id") + assignmentID := c.Param("id") + + var body struct { + ReviewData json.RawMessage `json:"review_data"` + } + if err := c.ShouldBindJSON(&body); err != nil && err.Error() != "EOF" { + c.JSON(http.StatusBadRequest, gin.H{"error": "invalid request body"}) + return + } + + // Verify current user is the claimer + userAssignments, err := h.stores.Workflows.ListAssignmentsByUser(c.Request.Context(), userID, models.AssignmentStatusClaimed) + if err == nil { + found := false + for _, a := range userAssignments { + if a.ID == assignmentID { + found = true + break + } + } + if !found { + c.JSON(http.StatusForbidden, gin.H{"error": "assignment not claimed by you"}) + return + } + } + + if err := h.stores.Workflows.CompleteAssignment(c.Request.Context(), assignmentID, body.ReviewData); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + + // If the stage has auto_transition, advance the instance. + // Look up the assignment's instance to get context. + userAssignments, _ = h.stores.Workflows.ListAssignmentsByUser(c.Request.Context(), userID, models.AssignmentStatusCompleted) + for _, a := range userAssignments { + if a.ID == assignmentID { + // Try to advance — engine will handle validation + h.engine.Advance(c.Request.Context(), a.InstanceID, body.ReviewData, userID) + break + } + } + + c.JSON(http.StatusOK, gin.H{"completed": true}) +} + +// Cancel cancels an open assignment. +// POST /api/v1/assignments/:id/cancel +func (h *WorkflowAssignmentHandler) Cancel(c *gin.Context) { + assignmentID := c.Param("id") + + if err := h.stores.Workflows.CancelAssignment(c.Request.Context(), assignmentID); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + + c.JSON(http.StatusOK, gin.H{"cancelled": true}) +} + +// ListByTeam returns assignments for a team. +// GET /api/v1/teams/:teamId/assignments?status= +func (h *WorkflowAssignmentHandler) ListByTeam(c *gin.Context) { + teamID := c.Param("teamId") + status := c.Query("status") + + result, err := h.stores.Workflows.ListAssignmentsByTeam(c.Request.Context(), teamID, status) + if err != nil { + if err == sql.ErrNoRows { + c.JSON(http.StatusOK, gin.H{"data": []struct{}{}}) + return + } + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to list assignments"}) + return + } + if result == nil { + c.JSON(http.StatusOK, gin.H{"data": []struct{}{}}) + return + } + c.JSON(http.StatusOK, gin.H{"data": result}) +} + +// ListMine returns assignments claimed by or assigned to the current user. +// GET /api/v1/assignments/mine?status= +func (h *WorkflowAssignmentHandler) ListMine(c *gin.Context) { + userID := c.GetString("user_id") + status := c.Query("status") + + result, err := h.stores.Workflows.ListAssignmentsByUser(c.Request.Context(), userID, status) + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to list assignments"}) + return + } + if result == nil { + c.JSON(http.StatusOK, gin.H{"data": []struct{}{}}) + return + } + c.JSON(http.StatusOK, gin.H{"data": result}) +} diff --git a/server/handlers/workflow_instance_handlers.go b/server/handlers/workflow_instance_handlers.go new file mode 100644 index 0000000..b233231 --- /dev/null +++ b/server/handlers/workflow_instance_handlers.go @@ -0,0 +1,134 @@ +package handlers + +import ( + "database/sql" + "encoding/json" + "net/http" + "strconv" + + "github.com/gin-gonic/gin" + + "switchboard-core/store" + "switchboard-core/workflow" +) + +// ── Instance Handlers ─────────────────────── + +// WorkflowInstanceHandler manages workflow instance HTTP endpoints. +type WorkflowInstanceHandler struct { + engine *workflow.Engine + stores store.Stores +} + +// NewWorkflowInstanceHandler creates an instance handler. +func NewWorkflowInstanceHandler(engine *workflow.Engine, stores store.Stores) *WorkflowInstanceHandler { + return &WorkflowInstanceHandler{engine: engine, stores: stores} +} + +// Start creates a new workflow instance. +// POST /api/v1/workflows/:id/instances +func (h *WorkflowInstanceHandler) Start(c *gin.Context) { + workflowID := c.Param("id") + userID := c.GetString("user_id") + + var body struct { + Data json.RawMessage `json:"data"` + Metadata json.RawMessage `json:"metadata"` + } + if err := c.ShouldBindJSON(&body); err != nil && err.Error() != "EOF" { + c.JSON(http.StatusBadRequest, gin.H{"error": "invalid request body"}) + return + } + if len(body.Data) == 0 { + body.Data = json.RawMessage(`{}`) + } + + inst, err := h.engine.Start(c.Request.Context(), workflowID, body.Data, userID) + if err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + + c.JSON(http.StatusCreated, inst) +} + +// ListInstances returns instances for a workflow. +// GET /api/v1/workflows/:id/instances?status=&limit=&offset= +func (h *WorkflowInstanceHandler) ListInstances(c *gin.Context) { + workflowID := c.Param("id") + status := c.Query("status") + + opts := store.ListOptions{} + if l := c.Query("limit"); l != "" { + opts.Limit, _ = strconv.Atoi(l) + } + if o := c.Query("offset"); o != "" { + opts.Offset, _ = strconv.Atoi(o) + } + + result, err := h.stores.Workflows.ListInstances(c.Request.Context(), workflowID, status, opts) + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to list instances"}) + return + } + if result == nil { + c.JSON(http.StatusOK, gin.H{"data": []struct{}{}}) + return + } + c.JSON(http.StatusOK, gin.H{"data": result}) +} + +// GetInstance returns a single instance. +// GET /api/v1/workflows/:id/instances/:iid +func (h *WorkflowInstanceHandler) GetInstance(c *gin.Context) { + inst, err := h.stores.Workflows.GetInstance(c.Request.Context(), c.Param("iid")) + if err != nil { + if err == sql.ErrNoRows { + c.JSON(http.StatusNotFound, gin.H{"error": "instance not found"}) + return + } + c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } + // Verify it belongs to the requested workflow + if inst.WorkflowID != c.Param("id") { + c.JSON(http.StatusNotFound, gin.H{"error": "instance not found"}) + return + } + c.JSON(http.StatusOK, inst) +} + +// Advance moves an instance to the next stage. +// POST /api/v1/workflows/:id/instances/:iid/advance +func (h *WorkflowInstanceHandler) Advance(c *gin.Context) { + userID := c.GetString("user_id") + + var body struct { + Data json.RawMessage `json:"data"` + } + if err := c.ShouldBindJSON(&body); err != nil && err.Error() != "EOF" { + c.JSON(http.StatusBadRequest, gin.H{"error": "invalid request body"}) + return + } + + inst, err := h.engine.Advance(c.Request.Context(), c.Param("iid"), body.Data, userID) + if err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + + c.JSON(http.StatusOK, inst) +} + +// Cancel terminates an active instance. +// POST /api/v1/workflows/:id/instances/:iid/cancel +func (h *WorkflowInstanceHandler) Cancel(c *gin.Context) { + userID := c.GetString("user_id") + + if err := h.engine.Cancel(c.Request.Context(), c.Param("iid"), userID); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + + c.JSON(http.StatusOK, gin.H{"cancelled": true}) +} diff --git a/server/handlers/workflow_store_test.go b/server/handlers/workflow_store_test.go new file mode 100644 index 0000000..2aa7d7c --- /dev/null +++ b/server/handlers/workflow_store_test.go @@ -0,0 +1,633 @@ +package handlers + +import ( + "context" + "encoding/json" + "testing" + "time" + + "switchboard-core/database" + "switchboard-core/models" + "switchboard-core/store" + "switchboard-core/store/postgres" + "switchboard-core/store/sqlite" +) + +// testStores returns an appropriate Stores for the current dialect. +func testStores(t *testing.T) store.Stores { + t.Helper() + database.RequireTestDB(t) + if database.IsSQLite() { + return sqlite.NewStores(database.TestDB) + } + return postgres.NewStores(database.TestDB) +} + +// seedWorkflowFixture creates a workflow + 2 stages + published version. +// Returns (workflowID, versionNumber, stage1Name, stage2Name). +func seedWorkflowFixture(t *testing.T, s store.Stores, userID, teamID string) (string, int, string, string) { + t.Helper() + ctx := context.Background() + + wf := &models.Workflow{ + TeamID: &teamID, + Name: "Test Workflow", + Slug: "test-workflow", + EntryMode: "team_only", + IsActive: true, + CreatedBy: userID, + } + if err := s.Workflows.Create(ctx, wf); err != nil { + t.Fatalf("create workflow: %v", err) + } + + s1 := &models.WorkflowStage{ + WorkflowID: wf.ID, + Ordinal: 0, + Name: "intake", + StageMode: models.StageModeForm, + Audience: models.AudienceTeam, + StageType: models.StageTypeSimple, + } + s2 := &models.WorkflowStage{ + WorkflowID: wf.ID, + Ordinal: 1, + Name: "review", + StageMode: models.StageModeReview, + Audience: models.AudienceTeam, + StageType: models.StageTypeSimple, + AssignmentTeamID: &teamID, + } + if err := s.Workflows.CreateStage(ctx, s1); err != nil { + t.Fatalf("create stage 1: %v", err) + } + if err := s.Workflows.CreateStage(ctx, s2); err != nil { + t.Fatalf("create stage 2: %v", err) + } + + // Publish version 1 + snapshot, _ := json.Marshal([]models.WorkflowStage{*s1, *s2}) + ver := &models.WorkflowVersion{ + WorkflowID: wf.ID, + VersionNumber: 1, + Snapshot: snapshot, + } + if err := s.Workflows.Publish(ctx, ver); err != nil { + t.Fatalf("publish version: %v", err) + } + + return wf.ID, 1, s1.Name, s2.Name +} + +// ── Instance Store Tests ──────────────────── + +func TestWorkflowInstance_CreateAndGet(t *testing.T) { + database.RequireTestDB(t) + database.TruncateAll(t) + s := testStores(t) + ctx := context.Background() + + userID := database.SeedTestUser(t, "Alice", "alice@test.com") + teamID := database.SeedTestTeam(t, "Team A", userID) + wfID, ver, stage1, _ := seedWorkflowFixture(t, s, userID, teamID) + + token := "tok-abc-123" + inst := &models.WorkflowInstance{ + WorkflowID: wfID, + WorkflowVersion: ver, + CurrentStage: stage1, + StageData: json.RawMessage(`{"key":"val"}`), + Status: models.InstanceStatusActive, + StartedBy: userID, + EntryToken: &token, + Metadata: json.RawMessage(`{"source":"test"}`), + } + if err := s.Workflows.CreateInstance(ctx, inst); err != nil { + t.Fatalf("create instance: %v", err) + } + if inst.ID == "" { + t.Fatal("expected ID to be set") + } + + got, err := s.Workflows.GetInstance(ctx, inst.ID) + if err != nil { + t.Fatalf("get instance: %v", err) + } + if got.WorkflowID != wfID { + t.Errorf("workflow_id = %q, want %q", got.WorkflowID, wfID) + } + if got.CurrentStage != stage1 { + t.Errorf("current_stage = %q, want %q", got.CurrentStage, stage1) + } + if got.Status != models.InstanceStatusActive { + t.Errorf("status = %q, want active", got.Status) + } + if got.StartedBy != userID { + t.Errorf("started_by = %q, want %q", got.StartedBy, userID) + } + if got.EntryToken == nil || *got.EntryToken != token { + t.Errorf("entry_token = %v, want %q", got.EntryToken, token) + } + if string(got.StageData) != `{"key":"val"}` { + t.Errorf("stage_data = %s, want {\"key\":\"val\"}", got.StageData) + } + if got.CreatedAt.IsZero() { + t.Error("created_at is zero") + } +} + +func TestWorkflowInstance_GetByToken(t *testing.T) { + database.RequireTestDB(t) + database.TruncateAll(t) + s := testStores(t) + ctx := context.Background() + + userID := database.SeedTestUser(t, "Bob", "bob@test.com") + teamID := database.SeedTestTeam(t, "Team B", userID) + wfID, ver, stage1, _ := seedWorkflowFixture(t, s, userID, teamID) + + token := "unique-entry-token" + inst := &models.WorkflowInstance{ + WorkflowID: wfID, + WorkflowVersion: ver, + CurrentStage: stage1, + StartedBy: userID, + EntryToken: &token, + } + if err := s.Workflows.CreateInstance(ctx, inst); err != nil { + t.Fatalf("create instance: %v", err) + } + + got, err := s.Workflows.GetInstanceByToken(ctx, token) + if err != nil { + t.Fatalf("get by token: %v", err) + } + if got.ID != inst.ID { + t.Errorf("id = %q, want %q", got.ID, inst.ID) + } +} + +func TestWorkflowInstance_Update(t *testing.T) { + database.RequireTestDB(t) + database.TruncateAll(t) + s := testStores(t) + ctx := context.Background() + + userID := database.SeedTestUser(t, "Carol", "carol@test.com") + teamID := database.SeedTestTeam(t, "Team C", userID) + wfID, ver, stage1, stage2 := seedWorkflowFixture(t, s, userID, teamID) + + inst := &models.WorkflowInstance{ + WorkflowID: wfID, + WorkflowVersion: ver, + CurrentStage: stage1, + StageData: json.RawMessage(`{}`), + StartedBy: userID, + Metadata: json.RawMessage(`{}`), + } + if err := s.Workflows.CreateInstance(ctx, inst); err != nil { + t.Fatalf("create: %v", err) + } + + inst.CurrentStage = stage2 + inst.StageData = json.RawMessage(`{"updated":true}`) + inst.StageEnteredAt = time.Now().UTC().Add(time.Minute) + if err := s.Workflows.UpdateInstance(ctx, inst); err != nil { + t.Fatalf("update: %v", err) + } + + got, _ := s.Workflows.GetInstance(ctx, inst.ID) + if got.CurrentStage != stage2 { + t.Errorf("current_stage = %q, want %q", got.CurrentStage, stage2) + } + if string(got.StageData) != `{"updated":true}` { + t.Errorf("stage_data = %s", got.StageData) + } +} + +func TestWorkflowInstance_List(t *testing.T) { + database.RequireTestDB(t) + database.TruncateAll(t) + s := testStores(t) + ctx := context.Background() + + userID := database.SeedTestUser(t, "Dave", "dave@test.com") + teamID := database.SeedTestTeam(t, "Team D", userID) + wfID, ver, stage1, _ := seedWorkflowFixture(t, s, userID, teamID) + + // Create 3 instances: 2 active, 1 completed + for i := 0; i < 3; i++ { + inst := &models.WorkflowInstance{ + WorkflowID: wfID, + WorkflowVersion: ver, + CurrentStage: stage1, + StartedBy: userID, + } + if err := s.Workflows.CreateInstance(ctx, inst); err != nil { + t.Fatalf("create %d: %v", i, err) + } + if i == 2 { + s.Workflows.CompleteInstance(ctx, inst.ID) + } + } + + // List all + all, err := s.Workflows.ListInstances(ctx, wfID, "", store.ListOptions{}) + if err != nil { + t.Fatalf("list all: %v", err) + } + if len(all) != 3 { + t.Fatalf("len = %d, want 3", len(all)) + } + + // Filter by status + active, err := s.Workflows.ListInstances(ctx, wfID, "active", store.ListOptions{}) + if err != nil { + t.Fatalf("list active: %v", err) + } + if len(active) != 2 { + t.Errorf("active count = %d, want 2", len(active)) + } + + // Pagination + page, err := s.Workflows.ListInstances(ctx, wfID, "", store.ListOptions{Limit: 2, Offset: 1}) + if err != nil { + t.Fatalf("list paged: %v", err) + } + if len(page) != 2 { + t.Errorf("paged count = %d, want 2", len(page)) + } +} + +func TestWorkflowInstance_AdvanceStage(t *testing.T) { + database.RequireTestDB(t) + database.TruncateAll(t) + s := testStores(t) + ctx := context.Background() + + userID := database.SeedTestUser(t, "Eve", "eve@test.com") + teamID := database.SeedTestTeam(t, "Team E", userID) + wfID, ver, stage1, stage2 := seedWorkflowFixture(t, s, userID, teamID) + + inst := &models.WorkflowInstance{ + WorkflowID: wfID, + WorkflowVersion: ver, + CurrentStage: stage1, + StageData: json.RawMessage(`{"step":"one"}`), + StartedBy: userID, + } + if err := s.Workflows.CreateInstance(ctx, inst); err != nil { + t.Fatalf("create: %v", err) + } + + // Re-read to get the stored (possibly truncated) timestamp + before, _ := s.Workflows.GetInstance(ctx, inst.ID) + originalEnteredAt := before.StageEnteredAt + + // Sleep >1s to ensure SQLite second-precision timestamp differs + time.Sleep(1100 * time.Millisecond) + + newData := json.RawMessage(`{"step":"two"}`) + if err := s.Workflows.AdvanceStage(ctx, inst.ID, stage2, newData); err != nil { + t.Fatalf("advance: %v", err) + } + + got, _ := s.Workflows.GetInstance(ctx, inst.ID) + if got.CurrentStage != stage2 { + t.Errorf("current_stage = %q, want %q", got.CurrentStage, stage2) + } + if string(got.StageData) != `{"step":"two"}` { + t.Errorf("stage_data = %s", got.StageData) + } + if !got.StageEnteredAt.After(originalEnteredAt) { + t.Errorf("stage_entered_at should have advanced; original=%v, got=%v", + originalEnteredAt, got.StageEnteredAt) + } +} + +func TestWorkflowInstance_Complete(t *testing.T) { + database.RequireTestDB(t) + database.TruncateAll(t) + s := testStores(t) + ctx := context.Background() + + userID := database.SeedTestUser(t, "Frank", "frank@test.com") + teamID := database.SeedTestTeam(t, "Team F", userID) + wfID, ver, stage1, _ := seedWorkflowFixture(t, s, userID, teamID) + + inst := &models.WorkflowInstance{ + WorkflowID: wfID, + WorkflowVersion: ver, + CurrentStage: stage1, + StartedBy: userID, + } + s.Workflows.CreateInstance(ctx, inst) + + if err := s.Workflows.CompleteInstance(ctx, inst.ID); err != nil { + t.Fatalf("complete: %v", err) + } + + got, _ := s.Workflows.GetInstance(ctx, inst.ID) + if got.Status != models.InstanceStatusCompleted { + t.Errorf("status = %q, want completed", got.Status) + } + + // Complete again — should be no-op (already completed, WHERE status='active' won't match) + if err := s.Workflows.CompleteInstance(ctx, inst.ID); err != nil { + t.Fatalf("complete again: %v", err) + } +} + +func TestWorkflowInstance_Cancel(t *testing.T) { + database.RequireTestDB(t) + database.TruncateAll(t) + s := testStores(t) + ctx := context.Background() + + userID := database.SeedTestUser(t, "Grace", "grace@test.com") + teamID := database.SeedTestTeam(t, "Team G", userID) + wfID, ver, stage1, _ := seedWorkflowFixture(t, s, userID, teamID) + + inst := &models.WorkflowInstance{ + WorkflowID: wfID, + WorkflowVersion: ver, + CurrentStage: stage1, + StartedBy: userID, + } + s.Workflows.CreateInstance(ctx, inst) + + if err := s.Workflows.CancelInstance(ctx, inst.ID); err != nil { + t.Fatalf("cancel: %v", err) + } + + got, _ := s.Workflows.GetInstance(ctx, inst.ID) + if got.Status != models.InstanceStatusCancelled { + t.Errorf("status = %q, want cancelled", got.Status) + } +} + +// ── Assignment Store Tests ────────────────── + +func TestWorkflowAssignment_CreateAndList(t *testing.T) { + database.RequireTestDB(t) + database.TruncateAll(t) + s := testStores(t) + ctx := context.Background() + + userID := database.SeedTestUser(t, "Hank", "hank@test.com") + teamID := database.SeedTestTeam(t, "Team H", userID) + wfID, ver, stage1, _ := seedWorkflowFixture(t, s, userID, teamID) + + inst := &models.WorkflowInstance{ + WorkflowID: wfID, + WorkflowVersion: ver, + CurrentStage: stage1, + StartedBy: userID, + } + s.Workflows.CreateInstance(ctx, inst) + + a := &models.WorkflowAssignment{ + InstanceID: inst.ID, + Stage: stage1, + TeamID: teamID, + } + if err := s.Workflows.CreateAssignment(ctx, a); err != nil { + t.Fatalf("create assignment: %v", err) + } + if a.ID == "" { + t.Fatal("expected assignment ID") + } + if a.Status != models.AssignmentStatusUnassigned { + t.Errorf("default status = %q, want unassigned", a.Status) + } + + list, err := s.Workflows.ListAssignmentsByInstance(ctx, inst.ID) + if err != nil { + t.Fatalf("list by instance: %v", err) + } + if len(list) != 1 { + t.Fatalf("len = %d, want 1", len(list)) + } + if list[0].Stage != stage1 { + t.Errorf("stage = %q, want %q", list[0].Stage, stage1) + } +} + +func TestWorkflowAssignment_Claim(t *testing.T) { + database.RequireTestDB(t) + database.TruncateAll(t) + s := testStores(t) + ctx := context.Background() + + userID := database.SeedTestUser(t, "Ivy", "ivy@test.com") + teamID := database.SeedTestTeam(t, "Team I", userID) + wfID, ver, stage1, _ := seedWorkflowFixture(t, s, userID, teamID) + + inst := &models.WorkflowInstance{ + WorkflowID: wfID, WorkflowVersion: ver, + CurrentStage: stage1, StartedBy: userID, + } + s.Workflows.CreateInstance(ctx, inst) + + a := &models.WorkflowAssignment{InstanceID: inst.ID, Stage: stage1, TeamID: teamID} + s.Workflows.CreateAssignment(ctx, a) + + if err := s.Workflows.ClaimAssignment(ctx, a.ID, userID); err != nil { + t.Fatalf("claim: %v", err) + } + + list, _ := s.Workflows.ListAssignmentsByInstance(ctx, inst.ID) + got := list[0] + if got.Status != models.AssignmentStatusClaimed { + t.Errorf("status = %q, want claimed", got.Status) + } + if got.AssignedTo == nil || *got.AssignedTo != userID { + t.Errorf("assigned_to = %v, want %q", got.AssignedTo, userID) + } + if got.ClaimedAt == nil { + t.Error("claimed_at should be set") + } +} + +func TestWorkflowAssignment_ClaimAlreadyClaimed(t *testing.T) { + database.RequireTestDB(t) + database.TruncateAll(t) + s := testStores(t) + ctx := context.Background() + + userID := database.SeedTestUser(t, "Jack", "jack@test.com") + teamID := database.SeedTestTeam(t, "Team J", userID) + wfID, ver, stage1, _ := seedWorkflowFixture(t, s, userID, teamID) + + inst := &models.WorkflowInstance{ + WorkflowID: wfID, WorkflowVersion: ver, + CurrentStage: stage1, StartedBy: userID, + } + s.Workflows.CreateInstance(ctx, inst) + + a := &models.WorkflowAssignment{InstanceID: inst.ID, Stage: stage1, TeamID: teamID} + s.Workflows.CreateAssignment(ctx, a) + s.Workflows.ClaimAssignment(ctx, a.ID, userID) + + // Second claim should fail + err := s.Workflows.ClaimAssignment(ctx, a.ID, userID) + if err == nil { + t.Fatal("expected error on double claim") + } +} + +func TestWorkflowAssignment_Unclaim(t *testing.T) { + database.RequireTestDB(t) + database.TruncateAll(t) + s := testStores(t) + ctx := context.Background() + + userID := database.SeedTestUser(t, "Kate", "kate@test.com") + teamID := database.SeedTestTeam(t, "Team K", userID) + wfID, ver, stage1, _ := seedWorkflowFixture(t, s, userID, teamID) + + inst := &models.WorkflowInstance{ + WorkflowID: wfID, WorkflowVersion: ver, + CurrentStage: stage1, StartedBy: userID, + } + s.Workflows.CreateInstance(ctx, inst) + + a := &models.WorkflowAssignment{InstanceID: inst.ID, Stage: stage1, TeamID: teamID} + s.Workflows.CreateAssignment(ctx, a) + s.Workflows.ClaimAssignment(ctx, a.ID, userID) + + if err := s.Workflows.UnclaimAssignment(ctx, a.ID); err != nil { + t.Fatalf("unclaim: %v", err) + } + + list, _ := s.Workflows.ListAssignmentsByInstance(ctx, inst.ID) + got := list[0] + if got.Status != models.AssignmentStatusUnassigned { + t.Errorf("status = %q, want unassigned", got.Status) + } + if got.AssignedTo != nil { + t.Errorf("assigned_to = %v, want nil", got.AssignedTo) + } + if got.ClaimedAt != nil { + t.Errorf("claimed_at = %v, want nil", got.ClaimedAt) + } +} + +func TestWorkflowAssignment_Complete(t *testing.T) { + database.RequireTestDB(t) + database.TruncateAll(t) + s := testStores(t) + ctx := context.Background() + + userID := database.SeedTestUser(t, "Leo", "leo@test.com") + teamID := database.SeedTestTeam(t, "Team L", userID) + wfID, ver, stage1, _ := seedWorkflowFixture(t, s, userID, teamID) + + inst := &models.WorkflowInstance{ + WorkflowID: wfID, WorkflowVersion: ver, + CurrentStage: stage1, StartedBy: userID, + } + s.Workflows.CreateInstance(ctx, inst) + + a := &models.WorkflowAssignment{InstanceID: inst.ID, Stage: stage1, TeamID: teamID} + s.Workflows.CreateAssignment(ctx, a) + s.Workflows.ClaimAssignment(ctx, a.ID, userID) + + reviewData := json.RawMessage(`{"approved":true,"notes":"LGTM"}`) + if err := s.Workflows.CompleteAssignment(ctx, a.ID, reviewData); err != nil { + t.Fatalf("complete: %v", err) + } + + list, _ := s.Workflows.ListAssignmentsByInstance(ctx, inst.ID) + got := list[0] + if got.Status != models.AssignmentStatusCompleted { + t.Errorf("status = %q, want completed", got.Status) + } + if string(got.ReviewData) != `{"approved":true,"notes":"LGTM"}` { + t.Errorf("review_data = %s", got.ReviewData) + } + if got.CompletedAt == nil { + t.Error("completed_at should be set") + } +} + +func TestWorkflowAssignment_Cancel(t *testing.T) { + database.RequireTestDB(t) + database.TruncateAll(t) + s := testStores(t) + ctx := context.Background() + + userID := database.SeedTestUser(t, "Mia", "mia@test.com") + teamID := database.SeedTestTeam(t, "Team M", userID) + wfID, ver, stage1, _ := seedWorkflowFixture(t, s, userID, teamID) + + inst := &models.WorkflowInstance{ + WorkflowID: wfID, WorkflowVersion: ver, + CurrentStage: stage1, StartedBy: userID, + } + s.Workflows.CreateInstance(ctx, inst) + + // Cancel from unassigned + a1 := &models.WorkflowAssignment{InstanceID: inst.ID, Stage: stage1, TeamID: teamID} + s.Workflows.CreateAssignment(ctx, a1) + if err := s.Workflows.CancelAssignment(ctx, a1.ID); err != nil { + t.Fatalf("cancel unassigned: %v", err) + } + list, _ := s.Workflows.ListAssignmentsByInstance(ctx, inst.ID) + if list[0].Status != models.AssignmentStatusCancelled { + t.Errorf("status = %q, want cancelled", list[0].Status) + } + + // Cancel from claimed + a2 := &models.WorkflowAssignment{InstanceID: inst.ID, Stage: stage1, TeamID: teamID} + s.Workflows.CreateAssignment(ctx, a2) + s.Workflows.ClaimAssignment(ctx, a2.ID, userID) + if err := s.Workflows.CancelAssignment(ctx, a2.ID); err != nil { + t.Fatalf("cancel claimed: %v", err) + } +} + +func TestWorkflowAssignment_ListByTeam(t *testing.T) { + database.RequireTestDB(t) + database.TruncateAll(t) + s := testStores(t) + ctx := context.Background() + + userID := database.SeedTestUser(t, "Nora", "nora@test.com") + teamA := database.SeedTestTeam(t, "Team Alpha", userID) + teamB := database.SeedTestTeam(t, "Team Beta", userID) + wfID, ver, stage1, _ := seedWorkflowFixture(t, s, userID, teamA) + + inst := &models.WorkflowInstance{ + WorkflowID: wfID, WorkflowVersion: ver, + CurrentStage: stage1, StartedBy: userID, + } + s.Workflows.CreateInstance(ctx, inst) + + // 2 assignments for teamA, 1 for teamB + for i := 0; i < 2; i++ { + a := &models.WorkflowAssignment{InstanceID: inst.ID, Stage: stage1, TeamID: teamA} + s.Workflows.CreateAssignment(ctx, a) + } + a3 := &models.WorkflowAssignment{InstanceID: inst.ID, Stage: stage1, TeamID: teamB} + s.Workflows.CreateAssignment(ctx, a3) + + // Claim one of teamA's so we can test status filter + listA, _ := s.Workflows.ListAssignmentsByTeam(ctx, teamA, "") + if len(listA) != 2 { + t.Fatalf("teamA count = %d, want 2", len(listA)) + } + s.Workflows.ClaimAssignment(ctx, listA[0].ID, userID) + + // Filter teamA unassigned only + unassigned, _ := s.Workflows.ListAssignmentsByTeam(ctx, teamA, "unassigned") + if len(unassigned) != 1 { + t.Errorf("teamA unassigned = %d, want 1", len(unassigned)) + } + + // teamB should have 1 + listB, _ := s.Workflows.ListAssignmentsByTeam(ctx, teamB, "") + if len(listB) != 1 { + t.Errorf("teamB count = %d, want 1", len(listB)) + } +} diff --git a/server/main.go b/server/main.go index 3349e48..c43feb2 100644 --- a/server/main.go +++ b/server/main.go @@ -30,6 +30,7 @@ import ( "switchboard-core/storage" "switchboard-core/store" "switchboard-core/triggers" + "switchboard-core/workflow" postgres "switchboard-core/store/postgres" sqliteStore "switchboard-core/store/sqlite" ) @@ -399,6 +400,21 @@ func main() { protected.POST("/workflows/:id/publish", middleware.RequirePermission(auth.PermWorkflowCreate, stores), wfH.Publish) protected.GET("/workflows/:id/versions/:version", wfH.GetVersion) + // Workflow instances + assignments (v0.3.2) + wfEngine := workflow.NewEngine(stores, bus, starlarkRunner) + wfInstH := handlers.NewWorkflowInstanceHandler(wfEngine, stores) + protected.POST("/workflows/:id/instances", middleware.RequirePermission(auth.PermWorkflowSubmit, stores), wfInstH.Start) + protected.GET("/workflows/:id/instances", wfInstH.ListInstances) + protected.GET("/workflows/:id/instances/:iid", wfInstH.GetInstance) + protected.POST("/workflows/:id/instances/:iid/advance", middleware.RequirePermission(auth.PermWorkflowSubmit, stores), wfInstH.Advance) + protected.POST("/workflows/:id/instances/:iid/cancel", middleware.RequirePermission(auth.PermWorkflowCreate, stores), wfInstH.Cancel) + + wfAssignH := handlers.NewWorkflowAssignmentHandler(wfEngine, stores) + protected.POST("/assignments/:id/claim", wfAssignH.Claim) + protected.POST("/assignments/:id/unclaim", wfAssignH.Unclaim) + protected.POST("/assignments/:id/complete", wfAssignH.Complete) + protected.POST("/assignments/:id/cancel", middleware.RequirePermission(auth.PermWorkflowCreate, stores), wfAssignH.Cancel) + protected.GET("/assignments/mine", wfAssignH.ListMine) // Surface discovery (v0.25.0, v0.28.7: unified packages) pkgH := handlers.NewPackageHandler(stores) @@ -522,6 +538,17 @@ func main() { teamScoped.POST("/workflows/:id/publish", teamWfH.PublishTeamWorkflow) teamScoped.GET("/workflows/:id/versions/:version", teamWfH.GetTeamWorkflowVersion) + // Team workflow instances + assignments (v0.3.2) + teamWfInstH := handlers.NewWorkflowInstanceHandler(wfEngine, stores) + teamScoped.POST("/workflows/:id/instances", teamWfInstH.Start) + teamScoped.GET("/workflows/:id/instances", teamWfInstH.ListInstances) + teamScoped.GET("/workflows/:id/instances/:iid", teamWfInstH.GetInstance) + teamScoped.POST("/workflows/:id/instances/:iid/advance", teamWfInstH.Advance) + teamScoped.POST("/workflows/:id/instances/:iid/cancel", teamWfInstH.Cancel) + + teamWfAssignH := handlers.NewWorkflowAssignmentHandler(wfEngine, stores) + teamScoped.GET("/assignments", teamWfAssignH.ListByTeam) + } // Public global settings (non-admin users can read safe subset) diff --git a/server/sandbox/workflow_module.go b/server/sandbox/workflow_module.go index cfba1e6..c75a465 100644 --- a/server/sandbox/workflow_module.go +++ b/server/sandbox/workflow_module.go @@ -1,18 +1,21 @@ package sandbox -// workflow_module.go — v0.30.2 CS1 +// workflow_module.go — v0.30.2 CS1, v0.3.2 instance API // -// The workflow module lets extensions read workflow definitions and -// stage data, and programmatically advance or reject workflow stages. +// The workflow module lets extensions read workflow definitions, +// manage instances, and programmatically advance workflow stages. // // Starlark API: -// wf = workflow.get_definition(workflow_id) # returns dict -// data = workflow.get_stage_data(channel_id) # returns dict -// workflow.advance(channel_id) # advance to next stage -// workflow.reject(channel_id, reason) # reject current stage +// wf = workflow.get_definition(workflow_id) # returns dict +// inst = workflow.get_instance(instance_id) # returns dict +// inst = workflow.start(workflow_id, data={}) # start new instance +// inst = workflow.advance(instance_id, data={}) # advance to next stage +// workflow.cancel(instance_id) # cancel instance +// instances = workflow.list_instances(workflow_id) # list instances import ( "context" + "encoding/json" "fmt" "go.starlark.net/starlark" @@ -23,9 +26,16 @@ import ( // BuildWorkflowModule creates the "workflow" Starlark module for a package. // Requires the workflow.access permission. +// +// v0.3.2: added get_instance and list_instances for read-only instance access. +// Mutating operations (start/advance/cancel) are available via HTTP API; +// Starlark builtins for them will be added when the engine interface is +// extracted to a shared package to avoid circular imports. func BuildWorkflowModule(ctx context.Context, stores store.Stores) *starlarkstruct.Module { return MakeModule("workflow", starlark.StringDict{ "get_definition": starlark.NewBuiltin("workflow.get_definition", workflowGetDef(ctx, stores)), + "get_instance": starlark.NewBuiltin("workflow.get_instance", workflowGetInstance(ctx, stores)), + "list_instances": starlark.NewBuiltin("workflow.list_instances", workflowListInstances(ctx, stores)), }) } @@ -77,5 +87,105 @@ func workflowGetDef(ctx context.Context, stores store.Stores) func(*starlark.Thr } } +// ── Instance Read API (v0.3.2) ────────────── + +func workflowGetInstance(ctx context.Context, stores store.Stores) func(*starlark.Thread, *starlark.Builtin, starlark.Tuple, []starlark.Tuple) (starlark.Value, error) { + return func(_ *starlark.Thread, _ *starlark.Builtin, args starlark.Tuple, kwargs []starlark.Tuple) (starlark.Value, error) { + var instanceID string + if err := starlark.UnpackPositionalArgs("workflow.get_instance", args, kwargs, 1, &instanceID); err != nil { + return nil, err + } + inst, err := stores.Workflows.GetInstance(ctx, instanceID) + if err != nil { + return starlark.None, fmt.Errorf("workflow.get_instance: %w", err) + } + + d := starlark.NewDict(8) + d.SetKey(starlark.String("id"), starlark.String(inst.ID)) + d.SetKey(starlark.String("workflow_id"), starlark.String(inst.WorkflowID)) + d.SetKey(starlark.String("workflow_version"), starlark.MakeInt(inst.WorkflowVersion)) + d.SetKey(starlark.String("current_stage"), starlark.String(inst.CurrentStage)) + d.SetKey(starlark.String("status"), starlark.String(inst.Status)) + d.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 { + starlarkData := starlark.NewDict(len(dataMap)) + for k, v := range dataMap { + starlarkData.SetKey(starlark.String(k), goValToStarlark(v)) + } + d.SetKey(starlark.String("stage_data"), starlarkData) + } else { + d.SetKey(starlark.String("stage_data"), starlark.NewDict(0)) + } + + if inst.EntryToken != nil { + d.SetKey(starlark.String("entry_token"), starlark.String(*inst.EntryToken)) + } + + return d, nil + } +} + +func workflowListInstances(ctx context.Context, stores store.Stores) func(*starlark.Thread, *starlark.Builtin, starlark.Tuple, []starlark.Tuple) (starlark.Value, error) { + return func(_ *starlark.Thread, _ *starlark.Builtin, args starlark.Tuple, kwargs []starlark.Tuple) (starlark.Value, error) { + var workflowID string + var status string + if err := starlark.UnpackPositionalArgs("workflow.list_instances", args, kwargs, 1, &workflowID, &status); err != nil { + return nil, err + } + + instances, err := stores.Workflows.ListInstances(ctx, workflowID, status, store.ListOptions{Limit: 100}) + if err != nil { + return starlark.None, fmt.Errorf("workflow.list_instances: %w", err) + } + + result := make([]starlark.Value, 0, len(instances)) + for _, inst := range instances { + d := starlark.NewDict(6) + d.SetKey(starlark.String("id"), starlark.String(inst.ID)) + d.SetKey(starlark.String("current_stage"), starlark.String(inst.CurrentStage)) + d.SetKey(starlark.String("status"), starlark.String(inst.Status)) + d.SetKey(starlark.String("started_by"), starlark.String(inst.StartedBy)) + d.SetKey(starlark.String("workflow_version"), starlark.MakeInt(inst.WorkflowVersion)) + result = append(result, d) + } + + return starlark.NewList(result), nil + } +} + +// goValToStarlark converts a Go value to a Starlark value. +func goValToStarlark(v interface{}) 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), goValToStarlark(v)) + } + return d + case []interface{}: + elems := make([]starlark.Value, len(val)) + for i, v := range val { + elems[i] = goValToStarlark(v) + } + return starlark.NewList(elems) + default: + return starlark.String(fmt.Sprintf("%v", v)) + } +} + // workflowRoute routes the workflow to a named stage (v0.35.0). // Starlark: workflow.route(channel_id, target_stage, reason) diff --git a/server/store/postgres/workflows.go b/server/store/postgres/workflows.go index f8be11b..1f69fc0 100644 --- a/server/store/postgres/workflows.go +++ b/server/store/postgres/workflows.go @@ -609,6 +609,19 @@ func (s *WorkflowStore) ListAssignmentsByInstance(ctx context.Context, instanceI ORDER BY created_at ASC`, instanceID) } +func (s *WorkflowStore) ListAssignmentsByUser(ctx context.Context, userID string, status string) ([]models.WorkflowAssignment, error) { + q := `SELECT id, instance_id, stage, team_id, assigned_to, status, + review_data, claimed_at, completed_at, created_at + FROM workflow_assignments WHERE assigned_to = $1` + args := []interface{}{userID} + if status != "" { + q += " AND status = $2" + args = append(args, status) + } + q += " ORDER BY created_at DESC" + return s.queryAssignments(ctx, q, args...) +} + func (s *WorkflowStore) queryAssignments(ctx context.Context, q string, args ...interface{}) ([]models.WorkflowAssignment, error) { rows, err := DB.QueryContext(ctx, q, args...) if err != nil { diff --git a/server/store/sqlite/workflows.go b/server/store/sqlite/workflows.go index 48ac8a5..22ca9d4 100644 --- a/server/store/sqlite/workflows.go +++ b/server/store/sqlite/workflows.go @@ -633,6 +633,19 @@ func (s *WorkflowStore) ListAssignmentsByInstance(ctx context.Context, instanceI ORDER BY created_at ASC`, instanceID) } +func (s *WorkflowStore) ListAssignmentsByUser(ctx context.Context, userID string, status string) ([]models.WorkflowAssignment, error) { + q := `SELECT id, instance_id, stage, team_id, assigned_to, status, + review_data, claimed_at, completed_at, created_at + FROM workflow_assignments WHERE assigned_to = ?` + args := []interface{}{userID} + if status != "" { + q += " AND status = ?" + args = append(args, status) + } + q += " ORDER BY created_at DESC" + return s.queryAssignments(ctx, q, args...) +} + func (s *WorkflowStore) queryAssignments(ctx context.Context, q string, args ...interface{}) ([]models.WorkflowAssignment, error) { rows, err := DB.QueryContext(ctx, q, args...) if err != nil { diff --git a/server/store/workflow_iface.go b/server/store/workflow_iface.go index bc41882..3bfe9f5 100644 --- a/server/store/workflow_iface.go +++ b/server/store/workflow_iface.go @@ -49,4 +49,5 @@ type WorkflowStore interface { CancelAssignment(ctx context.Context, id string) error ListAssignmentsByTeam(ctx context.Context, teamID string, status string) ([]models.WorkflowAssignment, error) ListAssignmentsByInstance(ctx context.Context, instanceID string) ([]models.WorkflowAssignment, error) + ListAssignmentsByUser(ctx context.Context, userID string, status string) ([]models.WorkflowAssignment, error) } diff --git a/server/workflow/automated.go b/server/workflow/automated.go new file mode 100644 index 0000000..a6875d9 --- /dev/null +++ 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() + } +} diff --git a/server/workflow/engine.go b/server/workflow/engine.go new file mode 100644 index 0000000..ae8fdba --- /dev/null +++ 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 +}