step 5 (partial): strip dropped packages from production code

Removed all references to dropped packages in non-test code.
Zero dropped imports, store refs, or model types remaining.

Major removals:
  - scheduler/ package (entire dir) — tasks moved to extension track
  - taskutil/ package (entire dir)
  - health/ package (entire dir) — provider/tool health
  - sandbox/provider_module.go — provider.complete module
  - handlers: workflow_entry, workflow_instances, workflow_forms,
    workflow_assignments, workflow_monitor, health_admin (6 files)
  - auth/session.go, middleware/session_auth.go
  - store/{postgres,sqlite}/sessions.go, tasks.go

Rewrites:
  - store/{postgres,sqlite}/health.go — kernel-only Prune
    (ws_tickets, rate_limit_counters, stale presence)
  - handlers/admin.go: 847→467 lines (stripped provider CRUD)
  - handlers/teams.go: 520→411 lines (stripped model listing)
  - sandbox/runner.go: removed ProviderResolver
  - sandbox/workflow_module.go: 216→83 lines (definition-only)
  - main.go: 1579→1325 lines (stripped dropped package init)
  - pages/pages.go: stripped channel/session lookups
  - events/types.go: stripped chat/channel/workspace event routes
  - models: stripped stale types, constants, notification types

Store interface cleanup:
  - Removed SessionStore, TaskStore from Stores struct
  - Stripped workflow assignment methods from WorkflowStore iface
  - Stripped assignment methods from PG+SQLite workflow stores
  - Updated testhelper.go table list to kernel-only

Migrations: removed 008_tasks.sql (both dialects)

-9665/+69 lines across 51 files.
This commit is contained in:
2026-03-25 21:13:01 -04:00
parent ebea16344c
commit e4b7ee98a5
55 changed files with 95 additions and 9981 deletions

View File

@@ -16,7 +16,6 @@ import (
"switchboard-core/crypto"
"switchboard-core/database"
"switchboard-core/models"
"switchboard-core/providers"
"switchboard-core/storage"
"switchboard-core/store"
"switchboard-core/tools/search"
@@ -438,325 +437,6 @@ func (h *AdminHandler) VaultStatus(c *gin.Context) {
c.JSON(http.StatusOK, status)
}
// ── Provider Configs (Global) ───────────────
func (h *AdminHandler) ListGlobalConfigs(c *gin.Context) {
cfgs, err := h.stores.Providers.ListGlobal(c.Request.Context())
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to list configs"})
return
}
// Redact API keys but expose has_key flag for UI
type configWithKey struct {
models.ProviderConfig
HasKey bool `json:"has_key"`
}
out := make([]configWithKey, len(cfgs))
for i, cfg := range cfgs {
out[i] = configWithKey{
ProviderConfig: cfg,
HasKey: cfg.HasKey(),
}
}
c.JSON(http.StatusOK, gin.H{"data": out})
}
func (h *AdminHandler) CreateGlobalConfig(c *gin.Context) {
// Wrapper struct: models.ProviderConfig has APIKeyEnc tagged json:"-"
// so ShouldBindJSON would silently drop the api_key field.
var req struct {
Name string `json:"name" binding:"required"`
Provider string `json:"provider" binding:"required"`
Endpoint string `json:"endpoint" binding:"required"`
APIKey string `json:"api_key"`
ModelDefault string `json:"model_default,omitempty"`
Config map[string]interface{} `json:"config,omitempty"`
Headers map[string]interface{} `json:"headers,omitempty"`
Settings map[string]interface{} `json:"settings,omitempty"`
IsPrivate bool `json:"is_private,omitempty"`
}
if err := c.ShouldBindJSON(&req); err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
return
}
if _, err := providers.Get(req.Provider); err != nil {
c.JSON(http.StatusBadRequest, gin.H{
"error": "unsupported provider: " + req.Provider,
"supported_providers": providers.List(),
})
return
}
cfg := &models.ProviderConfig{
Name: req.Name,
Provider: req.Provider,
Endpoint: req.Endpoint,
ModelDefault: req.ModelDefault,
Config: models.JSONMap(req.Config),
Headers: models.JSONMap(req.Headers),
Settings: models.JSONMap(req.Settings),
Scope: models.ScopeGlobal,
KeyScope: models.ScopeGlobal,
IsActive: true,
IsPrivate: req.IsPrivate,
}
// Encrypt the API key
if req.APIKey != "" {
if h.vault != nil {
enc, nonce, err := h.vault.EncryptForScope(req.APIKey, "global", "")
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to encrypt API key"})
return
}
cfg.APIKeyEnc = enc
cfg.KeyNonce = nonce
} else {
// No vault configured — store raw bytes (unencrypted fallback)
cfg.APIKeyEnc = []byte(req.APIKey)
}
}
if err := h.stores.Providers.Create(c.Request.Context(), cfg); err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to create config"})
return
}
h.auditLog(c, "provider.create", "provider_config", cfg.ID, gin.H{"name": cfg.Name, "provider": cfg.Provider})
c.JSON(http.StatusCreated, gin.H{
"id": cfg.ID,
"scope": cfg.Scope,
"name": cfg.Name,
"provider": cfg.Provider,
"endpoint": cfg.Endpoint,
"model_default": cfg.ModelDefault,
"config": cfg.Config,
"headers": cfg.Headers,
"settings": cfg.Settings,
"is_active": cfg.IsActive,
"is_private": cfg.IsPrivate,
"has_key": cfg.HasKey(),
"created_at": cfg.CreatedAt,
"updated_at": cfg.UpdatedAt,
})
}
func (h *AdminHandler) UpdateGlobalConfig(c *gin.Context) {
id := c.Param("id")
// Wrapper struct: ProviderConfigPatch has APIKeyEnc tagged json:"-"
var req struct {
models.ProviderConfigPatch
APIKey *string `json:"api_key,omitempty"`
}
if err := c.ShouldBindJSON(&req); err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
return
}
patch := req.ProviderConfigPatch
// Transfer api_key → encrypted fields
if req.APIKey != nil && *req.APIKey != "" {
if h.vault != nil {
enc, nonce, err := h.vault.EncryptForScope(*req.APIKey, "global", "")
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to encrypt API key"})
return
}
patch.APIKeyEnc = enc
patch.KeyNonce = nonce
} else {
patch.APIKeyEnc = []byte(*req.APIKey)
}
}
if err := h.stores.Providers.Update(c.Request.Context(), id, patch); err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to update config"})
return
}
h.auditLog(c, "provider.update", "provider_config", id, nil)
c.JSON(http.StatusOK, gin.H{"message": "config updated"})
}
func (h *AdminHandler) DeleteGlobalConfig(c *gin.Context) {
id := c.Param("id")
// Delete associated catalog entries first
h.stores.Catalog.DeleteForProvider(c.Request.Context(), id)
if err := h.stores.Providers.Delete(c.Request.Context(), id); err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to delete config"})
return
}
h.auditLog(c, "provider.delete", "provider_config", id, nil)
c.JSON(http.StatusOK, gin.H{"message": "config deleted"})
}
// ── Model Catalog ───────────────────────────
func (h *AdminHandler) ListModelConfigs(c *gin.Context) {
entries, err := h.stores.Catalog.ListAllGlobal(c.Request.Context())
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to list models"})
return
}
c.JSON(http.StatusOK, gin.H{"data": entries})
}
func (h *AdminHandler) FetchModels(c *gin.Context) {
var req struct {
ProviderConfigID string `json:"provider_config_id"`
}
c.ShouldBindJSON(&req)
// If no specific provider, fetch from ALL global providers
if req.ProviderConfigID == "" {
configs, err := h.stores.Providers.ListGlobal(c.Request.Context())
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to list providers"})
return
}
if len(configs) == 0 {
c.JSON(http.StatusBadRequest, gin.H{"error": "no providers configured — add one first"})
return
}
totalAdded, totalUpdated, totalFetched := 0, 0, 0
var errs []string
for _, cfg := range configs {
if !cfg.IsActive {
continue
}
added, updated, fetched, err := h.fetchModelsForProvider(c, &cfg)
if err != nil {
errs = append(errs, cfg.Provider+": "+err.Error())
continue
}
totalAdded += added
totalUpdated += updated
totalFetched += fetched
}
result := gin.H{
"message": "models synced",
"added": totalAdded,
"updated": totalUpdated,
"total": totalFetched,
}
if len(errs) > 0 {
result["errors"] = errs
}
c.JSON(http.StatusOK, result)
return
}
// Single provider fetch
cfg, err := h.stores.Providers.GetByID(c.Request.Context(), req.ProviderConfigID)
if err != nil {
c.JSON(http.StatusNotFound, gin.H{"error": "provider config not found"})
return
}
added, updated, fetched, err := h.fetchModelsForProvider(c, cfg)
if err != nil {
if errors.Is(err, ErrUpstreamTimeout) {
c.JSON(http.StatusGatewayTimeout, gin.H{"error": err.Error()})
return
}
c.JSON(http.StatusBadGateway, gin.H{"error": "failed to fetch models: " + err.Error()})
return
}
c.JSON(http.StatusOK, gin.H{
"message": "models synced",
"added": added,
"updated": updated,
"total": fetched,
})
}
// fetchModelsForProvider fetches and syncs models for a single provider config.
func (h *AdminHandler) fetchModelsForProvider(c *gin.Context, cfg *models.ProviderConfig) (added, updated, total int, err error) {
// Decrypt the API key for the provider API call
apiKey := ""
if len(cfg.APIKeyEnc) > 0 {
if h.vault != nil {
apiKey, err = h.vault.Decrypt(cfg.APIKeyEnc, cfg.KeyNonce, cfg.KeyScope, "")
if err != nil {
return 0, 0, 0, fmt.Errorf("failed to decrypt API key: %w", err)
}
} else {
// No vault — key stored as raw bytes (unencrypted fallback)
apiKey = string(cfg.APIKeyEnc)
}
}
result, err := syncProviderModels(c.Request.Context(), h.stores, cfg, apiKey)
if err != nil {
return 0, 0, 0, err
}
h.auditLog(c, "models.fetch", "provider_config", cfg.ID, gin.H{
"added": result.Added, "updated": result.Updated, "total": result.Total,
})
return result.Added, result.Updated, result.Total, nil
}
func (h *AdminHandler) UpdateModelConfig(c *gin.Context) {
id := c.Param("id")
var req struct {
Visibility *string `json:"visibility"`
}
if err := c.ShouldBindJSON(&req); err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
return
}
if req.Visibility != nil {
if err := h.stores.Catalog.SetVisibility(c.Request.Context(), id, *req.Visibility); err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to update visibility"})
return
}
}
c.JSON(http.StatusOK, gin.H{"message": "model updated"})
}
func (h *AdminHandler) BulkUpdateModels(c *gin.Context) {
var req struct {
ProviderConfigID string `json:"provider_config_id"`
Visibility string `json:"visibility" binding:"required"`
}
if err := c.ShouldBindJSON(&req); err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
return
}
var err error
if req.ProviderConfigID != "" {
err = h.stores.Catalog.BulkSetVisibility(c.Request.Context(), req.ProviderConfigID, req.Visibility)
} else {
err = h.stores.Catalog.BulkSetVisibilityAll(c.Request.Context(), req.Visibility)
}
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to bulk update"})
return
}
c.JSON(http.StatusOK, gin.H{"message": "bulk update complete"})
}
func (h *AdminHandler) DeleteModelConfig(c *gin.Context) {
id := c.Param("id")
if err := h.stores.Catalog.Delete(c.Request.Context(), id); err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to delete model"})
return
}
c.JSON(http.StatusOK, gin.H{"message": "model deleted"})
}
// ── Stats ───────────────────────────────────
func (h *AdminHandler) GetStats(c *gin.Context) {
@@ -764,12 +444,10 @@ func (h *AdminHandler) GetStats(c *gin.Context) {
stats := gin.H{}
userCount, _ := h.stores.Users.CountAll(ctx)
channelCount, _ := h.stores.Channels.CountAll(ctx)
messageCount, _ := h.stores.Messages.CountAll(ctx)
teamCount, _ := h.stores.Teams.CountAll(ctx)
stats["users"] = userCount
stats["channels"] = channelCount
stats["messages"] = messageCount
stats["teams"] = teamCount
c.JSON(http.StatusOK, stats)
}
@@ -786,62 +464,3 @@ func (h *AdminHandler) auditLog(c *gin.Context, action, resourceType, resourceID
}
AuditLog(h.stores.Audit, c, action, resourceType, resourceID, meta)
}
// ── Archived Channels (v0.23.2) ──────────────
func (h *AdminHandler) ListArchivedChannels(c *gin.Context) {
page, perPage, offset := parsePagination(c)
channels, total, err := h.stores.Channels.ListArchived(c.Request.Context(), store.ListOptions{
Limit: perPage,
Offset: offset,
})
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "query failed"})
return
}
// Retention config
retentionTTL := 0
if ttlCfg, err := h.stores.GlobalConfig.Get(c.Request.Context(), "retention_ttl_days"); err == nil {
if v, ok := ttlCfg["value"].(float64); ok {
retentionTTL = int(v)
}
}
c.JSON(http.StatusOK, gin.H{
"data": channels,
"total": total,
"page": page,
"per_page": perPage,
"retention_ttl_days": retentionTTL,
})
}
func (h *AdminHandler) PurgeChannel(c *gin.Context) {
channelID := c.Param("id")
// Clean up file storage blobs before CASCADE deletes the DB references
if h.objStore != nil {
prefix := fmt.Sprintf("files/%s", channelID)
if err := h.objStore.DeletePrefix(c.Request.Context(), prefix); err != nil {
log.Printf("⚠️ storage cleanup for purged channel %s failed: %v", channelID, err)
}
}
if err := h.stores.Channels.Purge(c.Request.Context(), channelID); err != nil {
msg := err.Error()
switch {
case msg == "channel not found":
c.JSON(http.StatusNotFound, gin.H{"error": msg})
case msg == "channel must be archived before purging":
c.JSON(http.StatusConflict, gin.H{"error": msg})
default:
c.JSON(http.StatusInternalServerError, gin.H{"error": "purge failed"})
}
return
}
h.auditLog(c, "channel.purged", "channel", channelID, nil)
c.JSON(http.StatusOK, gin.H{"ok": true})
}

View File

@@ -349,17 +349,12 @@ func hashToken(token string) string {
//
// Does NOT evict from UEK cache or write audit logs — callers handle that.
// v0.29.0: accepts stores instead of using database.DB directly.
func DestroyVaultDB(ctx context.Context, stores store.Stores, userID string) (providersDeleted int64) {
func DestroyVaultDB(ctx context.Context, stores store.Stores, userID string) (deleted int64) {
if err := stores.Users.ClearVaultKeys(ctx, userID); err != nil {
log.Printf("⚠ DestroyVaultDB: failed to clear vault columns for user %s: %v", userID, err)
}
rows, err := stores.Providers.DeletePersonalByOwner(ctx, userID)
if err != nil {
log.Printf("⚠ DestroyVaultDB: failed to delete personal providers for user %s: %v", userID, err)
return 0
}
return rows
return 0
}
// ProbeAndRepairVault checks whether a user's vault can be unlocked with

View File

@@ -1,294 +0,0 @@
package handlers
import (
"net/http"
"strconv"
"github.com/gin-gonic/gin"
capspkg "switchboard-core/capabilities"
"switchboard-core/health"
"switchboard-core/models"
"switchboard-core/providers"
"switchboard-core/store"
)
// ── Health Admin Handler ────────────────────
type HealthAdminHandler struct {
healthStore health.Store
stores store.Stores
}
func NewHealthAdminHandler(hs health.Store, stores store.Stores) *HealthAdminHandler {
return &HealthAdminHandler{healthStore: hs, stores: stores}
}
// GetProviderHealth returns health metrics for a single provider.
// GET /api/v1/admin/providers/:id/health?hours=24
func (h *HealthAdminHandler) GetProviderHealth(c *gin.Context) {
providerID := c.Param("id")
hours, _ := strconv.Atoi(c.DefaultQuery("hours", "24"))
if hours < 1 || hours > 168 {
hours = 24
}
windows, err := h.healthStore.ListWindows(c.Request.Context(), providerID, hours)
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to fetch health data"})
return
}
// Derive current status from most recent window
var summary models.ProviderHealthSummary
summary.ProviderConfigID = providerID
if len(windows) > 0 {
latest := windows[0]
summary.RequestCount = latest.RequestCount
summary.ErrorRate = latest.ErrorRate()
summary.AvgLatencyMs = latest.AvgLatencyMs()
summary.MaxLatencyMs = latest.MaxLatencyMs
summary.LastError = latest.LastError
summary.LastErrorAt = latest.LastErrorAt
summary.Status = health.DeriveStatus(summary.ErrorRate, summary.RequestCount)
} else {
summary.Status = models.StatusUnknown
}
SafeJSON(c, http.StatusOK, gin.H{
"summary": summary,
"windows": windows,
})
}
// GetAllProviderHealth returns current health for all providers.
// GET /api/v1/admin/providers/health
func (h *HealthAdminHandler) GetAllProviderHealth(c *gin.Context) {
windows, err := h.healthStore.ListAllCurrent(c.Request.Context())
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to fetch health data"})
return
}
// Build summaries
summaries := make([]models.ProviderHealthSummary, 0, len(windows))
for _, w := range windows {
s := models.ProviderHealthSummary{
ProviderConfigID: w.ProviderConfigID,
Status: health.DeriveStatus(w.ErrorRate(), w.RequestCount),
RequestCount: w.RequestCount,
ErrorRate: w.ErrorRate(),
ErrorCount: w.ErrorCount,
RateLimitCount: w.RateLimitCount,
TimeoutCount: w.TimeoutCount,
AvgLatencyMs: w.AvgLatencyMs(),
MaxLatencyMs: w.MaxLatencyMs,
LastError: w.LastError,
LastErrorAt: w.LastErrorAt,
}
summaries = append(summaries, s)
}
SafeJSON(c, http.StatusOK, gin.H{
"data": summaries,
})
}
// ── Capability Override Admin Handler ────────
type CapOverrideAdminHandler struct {
stores store.Stores
}
func NewCapOverrideAdminHandler(stores store.Stores) *CapOverrideAdminHandler {
return &CapOverrideAdminHandler{stores: stores}
}
// GetModelCapabilities returns resolved capabilities with source annotation.
// GET /api/v1/admin/models/:id/capabilities?provider_config_id=...
func (h *CapOverrideAdminHandler) GetModelCapabilities(c *gin.Context) {
modelID := c.Param("id")
providerConfigID := c.Query("provider_config_id")
// Get catalog capabilities
var catalogCaps *models.ModelCapabilities
if providerConfigID != "" {
if entry, err := h.stores.Catalog.GetByModelID(c.Request.Context(), providerConfigID, modelID); err == nil {
catalogCaps = &entry.Capabilities
}
} else {
if entry, err := h.stores.Catalog.GetByModelIDAny(c.Request.Context(), modelID); err == nil {
catalogCaps = &entry.Capabilities
}
}
// Get admin overrides
var overrides []models.CapabilityOverride
var err error
if providerConfigID != "" {
overrides, err = h.stores.CapOverrides.ListForProviderModel(c.Request.Context(), providerConfigID, modelID)
} else {
overrides, err = h.stores.CapOverrides.ListForModel(c.Request.Context(), modelID)
}
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to fetch overrides"})
return
}
// Resolve through three tiers
heuristic := capspkg.InferCapabilities(modelID)
resolved := capspkg.ResolveIntrinsic(modelID, catalogCaps, overrides)
// Build source annotations
sources := buildSourceAnnotations(catalogCaps, &heuristic, overrides)
SafeJSON(c, http.StatusOK, gin.H{
"model_id": modelID,
"resolved": resolved,
"catalog": catalogCaps,
"heuristic": heuristic,
"overrides": overrides,
"sources": sources,
})
}
// SetModelCapability sets an admin override for a specific capability.
// PUT /api/v1/admin/models/:id/capabilities
func (h *CapOverrideAdminHandler) SetModelCapability(c *gin.Context) {
modelID := c.Param("id")
userID := c.GetString("user_id")
var req struct {
ProviderConfigID *string `json:"provider_config_id"`
Field string `json:"field" binding:"required"`
Value string `json:"value" binding:"required"`
}
if err := c.ShouldBindJSON(&req); err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
return
}
// Validate field name
validFields := map[string]bool{
"streaming": true, "tool_calling": true, "vision": true,
"thinking": true, "reasoning": true, "code_optimized": true,
"web_search": true, "max_context": true, "max_output_tokens": true,
}
if !validFields[req.Field] {
c.JSON(http.StatusBadRequest, gin.H{"error": "invalid capability field: " + req.Field})
return
}
// Validate value
switch req.Field {
case "max_context", "max_output_tokens":
if _, err := strconv.Atoi(req.Value); err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": "value must be a number for " + req.Field})
return
}
default:
if req.Value != "true" && req.Value != "false" {
c.JSON(http.StatusBadRequest, gin.H{"error": "value must be 'true' or 'false' for " + req.Field})
return
}
}
override := &models.CapabilityOverride{
ProviderConfigID: req.ProviderConfigID,
ModelID: modelID,
Field: req.Field,
Value: req.Value,
SetBy: &userID,
}
if err := h.stores.CapOverrides.Set(c.Request.Context(), override); err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to set override"})
return
}
c.JSON(http.StatusOK, gin.H{"status": "ok"})
}
// DeleteModelCapability removes a specific override.
// DELETE /api/v1/admin/models/:id/capabilities/:overrideId
func (h *CapOverrideAdminHandler) DeleteModelCapability(c *gin.Context) {
overrideID := c.Param("overrideId")
if err := h.stores.CapOverrides.Delete(c.Request.Context(), overrideID); err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to delete override"})
return
}
c.JSON(http.StatusOK, gin.H{"status": "ok"})
}
// ListAllOverrides returns all capability overrides (admin view).
// GET /api/v1/admin/capability-overrides
func (h *CapOverrideAdminHandler) ListAllOverrides(c *gin.Context) {
overrides, err := h.stores.CapOverrides.ListAll(c.Request.Context())
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to fetch overrides"})
return
}
SafeJSON(c, http.StatusOK, gin.H{
"data": overrides,
"total": len(overrides),
})
}
// ── Source Annotation Helpers ────────────────
type capSource struct {
Field string `json:"field"`
Source string `json:"source"` // "catalog", "heuristic", "override"
}
func buildSourceAnnotations(catalog *models.ModelCapabilities, heuristic *models.ModelCapabilities, overrides []models.CapabilityOverride) []capSource {
// Build override set for quick lookup
overrideSet := make(map[string]bool, len(overrides))
for _, o := range overrides {
overrideSet[o.Field] = true
}
fields := []struct {
name string
hasCatalog bool
hasHeur bool
}{
{"streaming", catalog != nil && catalog.Streaming, heuristic.Streaming},
{"tool_calling", catalog != nil && catalog.ToolCalling, heuristic.ToolCalling},
{"vision", catalog != nil && catalog.Vision, heuristic.Vision},
{"thinking", catalog != nil && catalog.Thinking, heuristic.Thinking},
{"reasoning", catalog != nil && catalog.Reasoning, heuristic.Reasoning},
{"code_optimized", catalog != nil && catalog.CodeOptimized, heuristic.CodeOptimized},
{"web_search", catalog != nil && catalog.WebSearch, heuristic.WebSearch},
{"max_context", catalog != nil && catalog.MaxContext > 0, heuristic.MaxContext > 0},
{"max_output_tokens", catalog != nil && catalog.MaxOutputTokens > 0, heuristic.MaxOutputTokens > 0},
}
var sources []capSource
for _, f := range fields {
src := "default"
if overrideSet[f.name] {
src = "override"
} else if f.hasCatalog {
src = "catalog"
} else if f.hasHeur {
src = "heuristic"
}
sources = append(sources, capSource{Field: f.name, Source: src})
}
return sources
}
// ── Provider Types Endpoint ─────────────────
// GetProviderTypes returns metadata and profile schemas for all registered
// provider types. Used by the admin UI to render provider creation forms
// and show available settings.
func GetProviderTypes(c *gin.Context) {
types := providers.ListTypes()
c.JSON(http.StatusOK, gin.H{"data": types})
}

File diff suppressed because it is too large Load Diff

View File

@@ -1,590 +0,0 @@
package handlers
import (
"io"
"net/http"
"time"
"github.com/gin-gonic/gin"
"switchboard-core/auth"
"switchboard-core/middleware"
"switchboard-core/models"
"switchboard-core/store"
"switchboard-core/taskutil"
"switchboard-core/webhook"
)
// TaskHandler manages task CRUD and manual execution.
type TaskHandler struct {
stores store.Stores
}
func NewTaskHandler(stores store.Stores) *TaskHandler {
return &TaskHandler{stores: stores}
}
// ── List endpoints ──────────────────────────────
// ListMine returns tasks owned by the current user.
// GET /api/v1/tasks
func (h *TaskHandler) ListMine(c *gin.Context) {
userID := c.GetString("user_id")
tasks, err := h.stores.Tasks.ListByOwner(c.Request.Context(), userID)
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to list tasks"})
return
}
if tasks == nil {
tasks = []models.Task{}
}
c.JSON(http.StatusOK, gin.H{"data": tasks})
}
// ListAll returns all tasks (admin only).
// GET /api/v1/admin/tasks
func (h *TaskHandler) ListAll(c *gin.Context) {
tasks, err := h.stores.Tasks.ListAll(c.Request.Context())
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to list tasks"})
return
}
if tasks == nil {
tasks = []models.Task{}
}
c.JSON(http.StatusOK, gin.H{"data": tasks})
}
// ListTeamTasks returns tasks scoped to the given team.
// Any team member can view; RequireTeamMember middleware enforces access.
// GET /api/v1/teams/:teamId/tasks
func (h *TaskHandler) ListTeamTasks(c *gin.Context) {
teamID := c.Param("teamId")
tasks, err := h.stores.Tasks.ListByTeam(c.Request.Context(), teamID)
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to list team tasks"})
return
}
if tasks == nil {
tasks = []models.Task{}
}
c.JSON(http.StatusOK, gin.H{"data": tasks})
}
// ── Create ──────────────────────────────────────
// CreateTeamTask creates a task scoped to the team.
// Team admins only (RequireTeamAdmin middleware).
// POST /api/v1/teams/:teamId/tasks
func (h *TaskHandler) CreateTeamTask(c *gin.Context) {
teamID := c.Param("teamId")
// Inject team context, then delegate to Create
c.Set("force_team_id", teamID)
c.Set("force_scope", "team")
h.Create(c)
}
// Create creates a new task.
// POST /api/v1/tasks
func (h *TaskHandler) Create(c *gin.Context) {
ctx := c.Request.Context()
// v0.27.2: Check global task configuration
taskCfg := taskutil.LoadTaskConfig(ctx, h.stores.GlobalConfig)
if !taskCfg.Enabled {
c.JSON(http.StatusForbidden, gin.H{"error": "tasks are disabled by the administrator"})
return
}
var t models.Task
if err := c.ShouldBindJSON(&t); err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
return
}
t.OwnerID = c.GetString("user_id")
// v0.27.5: Team task context override (set by CreateTeamTask)
if forceTeam, ok := c.Get("force_team_id"); ok {
teamID := forceTeam.(string)
t.TeamID = &teamID
t.Scope = "team"
}
if forceScope, ok := c.Get("force_scope"); ok {
t.Scope = forceScope.(string)
}
// v0.27.2: Personal task check — non-admin users need tasks.allow_personal
if t.Scope == "personal" || t.Scope == "" {
if c.GetString("role") != "admin" && !taskCfg.AllowPersonal {
c.JSON(http.StatusForbidden, gin.H{"error": "personal tasks are disabled — contact your administrator"})
return
}
}
// v0.28.0: Action tasks require task.action permission
if t.TaskType == "action" {
if c.GetString("role") != "admin" {
perms := middleware.GetResolvedPermissions(c)
if perms == nil || !perms[auth.PermTaskAction] {
c.JSON(http.StatusForbidden, gin.H{"error": "permission required: " + auth.PermTaskAction})
return
}
}
}
// v0.29.0: Starlark tasks — run sandboxed extension scripts.
// Requires task.starlark permission. system_function holds package ID.
if t.TaskType == "starlark" {
if c.GetString("role") != "admin" {
perms := middleware.GetResolvedPermissions(c)
if perms == nil || !perms[auth.PermTaskStarlark] {
c.JSON(http.StatusForbidden, gin.H{"error": "permission required: " + auth.PermTaskStarlark})
return
}
}
if t.SystemFunction == "" {
c.JSON(http.StatusBadRequest, gin.H{"error": "system_function (package_id) is required for starlark tasks"})
return
}
// Verify package exists and is starlark tier
pkg, err := h.stores.Packages.Get(c.Request.Context(), t.SystemFunction)
if err != nil || pkg == nil {
c.JSON(http.StatusBadRequest, gin.H{"error": "package not found: " + t.SystemFunction})
return
}
if pkg.Tier != models.ExtTierStarlark {
c.JSON(http.StatusBadRequest, gin.H{"error": "package " + t.SystemFunction + " is not a starlark package"})
return
}
}
// v0.28.0-audit: Workflow task execution is not yet implemented.
// Reject at the API boundary to prevent silent wrong behavior
// (workflow tasks would fall through to the prompt pipeline).
if t.TaskType == "workflow" {
c.JSON(http.StatusBadRequest, gin.H{"error": "workflow task execution is not yet implemented — use task_type 'prompt' or 'action'"})
return
}
// v0.28.6: System tasks — admin-only, requires valid system_function
if t.TaskType == "system" {
if c.GetString("role") != "admin" {
c.JSON(http.StatusForbidden, gin.H{"error": "system tasks are admin-only"})
return
}
if t.SystemFunction == "" {
c.JSON(http.StatusBadRequest, gin.H{"error": "system_function is required for system tasks"})
return
}
if err := taskutil.ValidateSystemFunc(t.SystemFunction); err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
return
}
}
// Validate required fields
if t.Name == "" {
c.JSON(http.StatusBadRequest, gin.H{"error": "name is required"})
return
}
if t.Schedule == "" {
c.JSON(http.StatusBadRequest, gin.H{"error": "schedule is required"})
return
}
if t.TaskType == "prompt" && t.UserPrompt == "" {
c.JSON(http.StatusBadRequest, gin.H{"error": "user_prompt is required for prompt tasks"})
return
}
// v0.28.0: Webhook schedule validation — cannot use cron for webhook-triggered tasks
if t.Schedule == "webhook" {
// Webhook tasks never have a cron schedule — they fire on inbound POST
} else {
// v0.27.2: Validate cron expression before persisting
if err := taskutil.ValidateCron(t.Schedule); err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": "invalid schedule: " + err.Error()})
return
}
}
// Defaults
if t.Scope == "" {
t.Scope = "personal"
}
if t.TaskType == "" {
t.TaskType = "prompt"
}
if t.Timezone == "" {
t.Timezone = "UTC"
}
if t.OutputMode == "" {
t.OutputMode = "channel"
}
// v0.27.2: Apply global default budgets for zero-value fields
taskCfg.ApplyDefaults(&t)
// Compute initial next_run_at (webhook tasks have no schedule-based next_run)
if t.Schedule == "webhook" {
// No initial next_run_at — triggered externally
t.NextRunAt = nil
} else if t.Schedule == "once" {
now := time.Now().UTC()
t.NextRunAt = &now
} else {
// v0.27.2: Full cron parsing via robfig/cron/v3
t.NextRunAt = taskutil.NextRunFromSchedule(t.Schedule, t.Timezone)
}
t.IsActive = true
// v0.27.3: Generate webhook secret if webhook URL is provided
if t.WebhookURL != "" && t.WebhookSecret == "" {
t.WebhookSecret = webhook.GenerateSecret()
}
// v0.28.0: Generate trigger token for webhook-scheduled tasks
if t.Schedule == "webhook" {
t.TriggerToken = webhook.GenerateSecret()
}
if err := h.stores.Tasks.Create(ctx, &t); err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to create task: " + err.Error()})
return
}
c.JSON(http.StatusCreated, t)
}
// ── Access control helpers ──────────────────────
// canAccessTask returns true if the user can view this task.
// Access: owner, system admin, or team member (for team-scoped tasks).
func (h *TaskHandler) canAccessTask(c *gin.Context, t *models.Task) bool {
if c.GetString("role") == "admin" {
return true
}
if t.OwnerID == c.GetString("user_id") {
return true
}
// Team members can view team-scoped tasks
if t.Scope == "team" && t.TeamID != nil {
ok, _ := h.stores.Teams.IsMember(c.Request.Context(), *t.TeamID, c.GetString("user_id"))
return ok
}
return false
}
// canMutateTask returns true if the user can edit/delete/run this task.
// Mutation: owner, system admin, or team admin (for team-scoped tasks).
func (h *TaskHandler) canMutateTask(c *gin.Context, t *models.Task) bool {
if c.GetString("role") == "admin" {
return true
}
if t.OwnerID == c.GetString("user_id") {
return true
}
if t.Scope == "team" && t.TeamID != nil {
ok, _ := h.stores.Teams.IsTeamAdmin(c.Request.Context(), *t.TeamID, c.GetString("user_id"))
return ok
}
return false
}
// ── Get / Update / Delete ───────────────────────
// Get returns a single task.
// GET /api/v1/tasks/:id
func (h *TaskHandler) Get(c *gin.Context) {
id := c.Param("id")
t, err := h.stores.Tasks.GetByID(c.Request.Context(), id)
if err != nil {
c.JSON(http.StatusNotFound, gin.H{"error": "task not found"})
return
}
if !h.canAccessTask(c, t) {
c.JSON(http.StatusNotFound, gin.H{"error": "task not found"})
return
}
c.JSON(http.StatusOK, t)
}
// Update patches a task.
// PUT /api/v1/tasks/:id
func (h *TaskHandler) Update(c *gin.Context) {
id := c.Param("id")
ctx := c.Request.Context()
// Ownership check
t, err := h.stores.Tasks.GetByID(ctx, id)
if err != nil {
c.JSON(http.StatusNotFound, gin.H{"error": "task not found"})
return
}
if !h.canMutateTask(c, t) {
c.JSON(http.StatusNotFound, gin.H{"error": "task not found"})
return
}
var patch models.TaskPatch
if err := c.ShouldBindJSON(&patch); err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
return
}
// v0.27.2: Validate new schedule if provided
if patch.Schedule != nil && *patch.Schedule != "webhook" {
if err := taskutil.ValidateCron(*patch.Schedule); err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": "invalid schedule: " + err.Error()})
return
}
}
if err := h.stores.Tasks.Update(ctx, id, patch); err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to update task"})
return
}
updated, _ := h.stores.Tasks.GetByID(ctx, id)
// Recompute next_run_at if schedule or timezone changed
if patch.Schedule != nil || patch.Timezone != nil {
if updated.Schedule == "webhook" {
// Webhook tasks have no cron-based next_run
_ = h.stores.Tasks.SetNextRun(ctx, id, nil)
updated.NextRunAt = nil
} else {
tz := updated.Timezone
next := taskutil.NextRunFromSchedule(updated.Schedule, tz)
_ = h.stores.Tasks.SetNextRun(ctx, id, next)
updated.NextRunAt = next
}
}
c.JSON(http.StatusOK, updated)
}
// Delete removes a task.
// DELETE /api/v1/tasks/:id
func (h *TaskHandler) Delete(c *gin.Context) {
id := c.Param("id")
ctx := c.Request.Context()
t, err := h.stores.Tasks.GetByID(ctx, id)
if err != nil {
c.JSON(http.StatusNotFound, gin.H{"error": "task not found"})
return
}
if !h.canMutateTask(c, t) {
c.JSON(http.StatusNotFound, gin.H{"error": "task not found"})
return
}
if err := h.stores.Tasks.Delete(ctx, id); err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to delete task"})
return
}
c.JSON(http.StatusOK, gin.H{"deleted": true, "id": id})
}
// ── Run History ─────────────────────────────────
// ListRuns returns run history for a task.
// GET /api/v1/tasks/:id/runs
func (h *TaskHandler) ListRuns(c *gin.Context) {
id := c.Param("id")
ctx := c.Request.Context()
// Access check (v0.27.5)
t, err := h.stores.Tasks.GetByID(ctx, id)
if err != nil {
c.JSON(http.StatusNotFound, gin.H{"error": "task not found"})
return
}
if !h.canAccessTask(c, t) {
c.JSON(http.StatusNotFound, gin.H{"error": "task not found"})
return
}
runs, err := h.stores.Tasks.ListRuns(ctx, id, 50)
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to list runs"})
return
}
if runs == nil {
runs = []models.TaskRun{}
}
c.JSON(http.StatusOK, gin.H{"data": runs})
}
// ── RunNow / KillRun ────────────────────────────
// RunNow triggers immediate execution of a task (sets next_run_at to now).
// POST /api/v1/tasks/:id/run
func (h *TaskHandler) RunNow(c *gin.Context) {
id := c.Param("id")
ctx := c.Request.Context()
// v0.27.2: Check tasks enabled
taskCfg := taskutil.LoadTaskConfig(ctx, h.stores.GlobalConfig)
if !taskCfg.Enabled {
c.JSON(http.StatusForbidden, gin.H{"error": "tasks are disabled by the administrator"})
return
}
t, err := h.stores.Tasks.GetByID(ctx, id)
if err != nil {
c.JSON(http.StatusNotFound, gin.H{"error": "task not found"})
return
}
if !h.canMutateTask(c, t) {
c.JSON(http.StatusNotFound, gin.H{"error": "task not found"})
return
}
// Check for already-running execution
active, _ := h.stores.Tasks.GetActiveRun(ctx, id)
if active != nil {
c.JSON(http.StatusConflict, gin.H{"error": "task already has an active run"})
return
}
// Also check for queued runs
queued, _ := h.stores.Tasks.GetQueuedRun(ctx, id)
if queued != nil {
c.JSON(http.StatusConflict, gin.H{"error": "task already has a queued run"})
return
}
now := time.Now().UTC()
if err := h.stores.Tasks.SetNextRun(ctx, id, now); err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to schedule run"})
return
}
c.JSON(http.StatusOK, gin.H{"scheduled": true, "next_run_at": now})
}
// KillRun cancels the active run of a task.
// POST /api/v1/tasks/:id/kill
func (h *TaskHandler) KillRun(c *gin.Context) {
id := c.Param("id")
ctx := c.Request.Context()
t, err := h.stores.Tasks.GetByID(ctx, id)
if err != nil {
c.JSON(http.StatusNotFound, gin.H{"error": "task not found"})
return
}
if !h.canMutateTask(c, t) {
c.JSON(http.StatusNotFound, gin.H{"error": "task not found"})
return
}
active, _ := h.stores.Tasks.GetActiveRun(ctx, id)
if active == nil {
c.JSON(http.StatusNotFound, gin.H{"error": "no active run to cancel"})
return
}
if err := h.stores.Tasks.UpdateRun(ctx, active.ID, "cancelled", active.TokensUsed, active.ToolCalls, active.WallClock, "killed by user"); err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to cancel run"})
return
}
c.JSON(http.StatusOK, gin.H{"killed": true, "run_id": active.ID})
}
// ════════════════════════════════════════════════
// Webhook Trigger Handler (v0.28.0)
// ════════════════════════════════════════════════
//
// POST /api/v1/hooks/t/:token — unauthenticated, token-based auth.
// External systems (CI, task chaining, etc.) POST here to fire a
// webhook-triggered task. The request body is stored as trigger_payload
// and forwarded to the executor.
// TriggerHandler handles inbound webhook triggers.
type TriggerHandler struct {
stores store.Stores
}
func NewTriggerHandler(stores store.Stores) *TriggerHandler {
return &TriggerHandler{stores: stores}
}
// Handle processes an inbound webhook trigger.
// POST /api/v1/hooks/t/:token
func (h *TriggerHandler) Handle(c *gin.Context) {
token := c.Param("token")
ctx := c.Request.Context()
// Look up task by trigger token
task, err := h.stores.Tasks.GetByTriggerToken(ctx, token)
if err != nil {
c.JSON(http.StatusNotFound, gin.H{"error": "not found"})
return
}
// Validate task state
if !task.IsActive {
c.JSON(http.StatusGone, gin.H{"error": "task is inactive"})
return
}
if task.Schedule != "webhook" {
c.JSON(http.StatusBadRequest, gin.H{"error": "task is not webhook-triggered"})
return
}
// Check for already-active or queued run
active, _ := h.stores.Tasks.GetActiveRun(ctx, task.ID)
if active != nil {
c.JSON(http.StatusConflict, gin.H{"error": "task already has an active run"})
return
}
queued, _ := h.stores.Tasks.GetQueuedRun(ctx, task.ID)
if queued != nil {
c.JSON(http.StatusConflict, gin.H{"error": "task already has a queued run"})
return
}
// Read trigger payload (optional)
var triggerPayload string
if c.Request.Body != nil {
body, err := io.ReadAll(io.LimitReader(c.Request.Body, 1<<20)) // 1MB limit
if err == nil && len(body) > 0 {
triggerPayload = string(body)
}
}
// Create a queued run with the trigger payload
run := &models.TaskRun{
TaskID: task.ID,
Status: "queued",
TriggerPayload: triggerPayload,
}
if err := h.stores.Tasks.CreateRun(ctx, run); err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to create run"})
return
}
// Set next_run_at to now so the scheduler picks it up
now := time.Now().UTC()
if err := h.stores.Tasks.SetNextRun(ctx, task.ID, now); err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to schedule run"})
return
}
c.JSON(http.StatusAccepted, gin.H{
"triggered": true,
"run_id": run.ID,
"task_id": task.ID,
})
}
// ListSystemFunctions returns the available system function names and descriptions.
// GET /admin/system-functions
func (h *TaskHandler) ListSystemFunctions(c *gin.Context) {
c.JSON(http.StatusOK, gin.H{"data": taskutil.ListSystemFuncs()})
}

View File

@@ -14,7 +14,6 @@ import (
"switchboard-core/crypto"
"switchboard-core/database"
"switchboard-core/models"
"switchboard-core/providers"
"switchboard-core/store"
)
@@ -346,91 +345,6 @@ func (h *TeamHandler) MyTeams(c *gin.Context) {
c.JSON(http.StatusOK, gin.H{"data": teams})
}
// ── Team Models: Available for Personas ──────
// ListAvailableModels returns models with visibility 'enabled' or 'team'
// for team admins building personas. Requires RequireTeamAdmin middleware.
// GET /api/v1/teams/:teamId/models
func (h *TeamHandler) ListAvailableModels(c *gin.Context) {
teamID := getTeamID(c)
ctx := c.Request.Context()
type availableModel struct {
ID string `json:"id"`
ModelID string `json:"model_id"`
DisplayName *string `json:"display_name"`
Visibility string `json:"visibility"`
Provider string `json:"provider"`
ProviderName string `json:"provider_name"`
Source string `json:"source"`
}
result := make([]availableModel, 0)
// ── 1. Global admin models (synced in model_catalog) ──
catalogModels, err := h.stores.Catalog.ListTeamAvailable(ctx)
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "query failed"})
return
}
for _, cm := range catalogModels {
result = append(result, availableModel{
ID: cm.ID,
ModelID: cm.ModelID,
DisplayName: cm.DisplayName,
Visibility: cm.Visibility,
Provider: cm.Provider,
ProviderName: cm.ProviderName,
Source: "global",
})
}
// ── 2. Team provider models (live query) ──
teamConfigs, err := h.stores.Providers.ListForTeam(ctx, teamID)
if err == nil {
for _, cfg := range teamConfigs {
provider, pErr := providers.Get(cfg.Provider)
if pErr != nil {
continue
}
key := ""
if cfg.HasKey() {
key = string(cfg.APIKeyEnc)
}
var customHeaders map[string]string
if cfg.Headers != nil {
b, _ := json.Marshal(cfg.Headers)
_ = json.Unmarshal(b, &customHeaders)
}
provModels, lErr := provider.ListModels(ctx, providers.ProviderConfig{
Endpoint: cfg.Endpoint,
APIKey: key,
CustomHeaders: customHeaders,
})
if lErr != nil {
log.Printf("[models] team provider %q list failed: %v", cfg.Name, lErr)
continue
}
for _, pm := range provModels {
result = append(result, availableModel{
ID: pm.ID,
ModelID: pm.ID,
Provider: cfg.Provider,
ProviderName: cfg.Name,
Visibility: "enabled",
Source: "team",
})
}
}
}
c.JSON(http.StatusOK, gin.H{"data": result})
}
// ── Helpers ─────────────────────────────────
// getTeamID extracts team ID from either :id (admin routes) or :teamId (team-scoped routes).
@@ -441,30 +355,6 @@ func getTeamID(c *gin.Context) string {
return c.Param("id")
}
// enforcePrivateProviderPolicy checks if a user belongs to any team that
// requires private providers, and if so, verifies the resolved config is
// marked as private. Returns nil if allowed, error if blocked.
func enforcePrivateProviderPolicy(ctx context.Context, stores store.Stores, userID, configID string) error {
if configID == "" {
return nil
}
requiresPrivate, err := stores.Teams.HasPrivateProviderRequirement(ctx, userID)
if err != nil || !requiresPrivate {
return nil
}
// User is in a restricted team — verify the config is private
cfg, err := stores.Providers.GetByID(ctx, configID)
if err != nil {
return nil // config lookup failed, allow (fail open)
}
if !cfg.IsPrivate {
return fmt.Errorf("your team requires private providers — this provider sends data externally")
}
return nil
}
// ── Team Audit Log (scoped to team members) ─
func (h *TeamHandler) ListTeamAuditLog(c *gin.Context) {

View File

@@ -1,346 +0,0 @@
package handlers
// workflow_assignments.go — Assignment queue for human review stages.
//
// v0.29.0: Raw SQL replaced with WorkflowStore + ChannelStore methods.
import (
"encoding/json"
"fmt"
"net/http"
"time"
"github.com/gin-gonic/gin"
"switchboard-core/events"
"switchboard-core/notifications"
"switchboard-core/store"
)
// ── Workflow Assignment Handler ─────────────
type WorkflowAssignmentHandler struct {
stores store.Stores
hub *events.Hub
}
func NewWorkflowAssignmentHandler(stores store.Stores, hub ...*events.Hub) *WorkflowAssignmentHandler {
h := &WorkflowAssignmentHandler{stores: stores}
if len(hub) > 0 {
h.hub = hub[0]
}
return h
}
// ListForTeam returns unassigned + claimed assignments for a team.
// GET /api/v1/teams/:teamId/assignments
func (h *WorkflowAssignmentHandler) ListForTeam(c *gin.Context) {
teamID := c.Param("teamId")
status := c.DefaultQuery("status", "unassigned")
result, err := h.stores.Workflows.ListAssignmentsForTeam(c.Request.Context(), teamID, status)
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to list assignments"})
return
}
c.JSON(http.StatusOK, gin.H{"data": result})
}
// ListMine returns assignments claimed by the current user plus
// unassigned assignments for teams the user belongs to.
// GET /api/v1/workflow-assignments/mine
func (h *WorkflowAssignmentHandler) ListMine(c *gin.Context) {
userID := c.GetString("user_id")
result, err := h.stores.Workflows.ListAssignmentsMine(c.Request.Context(), userID)
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to list assignments"})
return
}
c.JSON(http.StatusOK, gin.H{"data": result})
}
// Claim assigns a workflow to the current user via optimistic lock.
// POST /api/v1/workflow-assignments/:id/claim
func (h *WorkflowAssignmentHandler) Claim(c *gin.Context) {
assignmentID := c.Param("id")
userID := c.GetString("user_id")
now := time.Now().UTC()
n, err := h.stores.Workflows.ClaimAssignment(c.Request.Context(), assignmentID, userID)
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to claim assignment"})
return
}
if n == 0 {
c.JSON(http.StatusConflict, gin.H{"error": "assignment already claimed or not found"})
return
}
// Look up channel for WS delivery + notification
channelID, _ := h.stores.Workflows.GetAssignmentChannelID(c.Request.Context(), assignmentID)
// Emit workflow.claimed WS event to all user participants in the channel
if h.hub != nil && channelID != "" {
payload, _ := json.Marshal(map[string]any{
"assignment_id": assignmentID,
"claimed_by": userID,
"channel_id": channelID,
})
evt := events.Event{
Label: "workflow.claimed",
Payload: payload,
Ts: now.UnixMilli(),
}
pids, _ := h.stores.Channels.ListUserParticipantIDs(c.Request.Context(), channelID, "")
for _, uid := range pids {
h.hub.PublishToUser(uid, evt)
}
}
// Persist notification for bell/inbox
if svc := notifications.Default(); svc != nil && channelID != "" {
notifications.NotifyWorkflowClaimed(svc, userID, assignmentID, channelID)
}
c.JSON(http.StatusOK, gin.H{"claimed": true, "assignment_id": assignmentID})
}
// Complete marks an assignment as completed.
// POST /api/v1/workflow-assignments/:id/complete
func (h *WorkflowAssignmentHandler) Complete(c *gin.Context) {
assignmentID := c.Param("id")
n, err := h.stores.Workflows.CompleteAssignment(c.Request.Context(), assignmentID)
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to complete assignment"})
return
}
if n == 0 {
c.JSON(http.StatusConflict, gin.H{"error": "assignment not in claimed state"})
return
}
c.JSON(http.StatusOK, gin.H{"completed": true, "assignment_id": assignmentID})
}
// ── Lifecycle operations (v0.37.15) ──
// Unclaim returns a claimed assignment to unassigned.
// POST /api/v1/workflow-assignments/:id/unclaim
func (h *WorkflowAssignmentHandler) Unclaim(c *gin.Context) {
assignmentID := c.Param("id")
userID := c.GetString("user_id")
role, _ := c.Get("role")
// Auth: claimer or team admin or global admin
a, err := h.stores.Workflows.GetAssignmentByID(c.Request.Context(), assignmentID)
if err != nil || a == nil {
c.JSON(http.StatusNotFound, gin.H{"error": "assignment not found"})
return
}
if role != "admin" && (a.AssignedTo == nil || *a.AssignedTo != userID) {
isTA := false
if h.stores.Teams != nil {
isTA, _ = h.stores.Teams.IsTeamAdmin(c.Request.Context(), a.TeamID, userID)
}
if !isTA {
c.JSON(http.StatusForbidden, gin.H{"error": "only the claimer or a team admin can unclaim"})
return
}
}
n, err := h.stores.Workflows.UnclaimAssignment(c.Request.Context(), assignmentID)
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to unclaim assignment"})
return
}
if n == 0 {
c.JSON(http.StatusConflict, gin.H{"error": "assignment not in claimed state"})
return
}
// Emit WS event
channelID, _ := h.stores.Workflows.GetAssignmentChannelID(c.Request.Context(), assignmentID)
if h.hub != nil && channelID != "" {
payload, _ := json.Marshal(map[string]any{
"assignment_id": assignmentID,
"unclaimed_by": userID,
"channel_id": channelID,
})
evt := events.Event{
Label: "workflow.unclaimed",
Payload: payload,
Ts: time.Now().UnixMilli(),
}
pids, _ := h.stores.Channels.ListUserParticipantIDs(c.Request.Context(), channelID, "")
for _, uid := range pids {
h.hub.PublishToUser(uid, evt)
}
}
c.JSON(http.StatusOK, gin.H{"unclaimed": true, "assignment_id": assignmentID})
}
// Reassign changes the assigned_to on a claimed assignment.
// POST /api/v1/workflow-assignments/:id/reassign
func (h *WorkflowAssignmentHandler) Reassign(c *gin.Context) {
assignmentID := c.Param("id")
userID := c.GetString("user_id")
role, _ := c.Get("role")
var body struct {
UserID string `json:"user_id"`
}
if err := c.ShouldBindJSON(&body); err != nil || body.UserID == "" {
c.JSON(http.StatusBadRequest, gin.H{"error": "user_id is required"})
return
}
// Auth: team admin or global admin
a, err := h.stores.Workflows.GetAssignmentByID(c.Request.Context(), assignmentID)
if err != nil || a == nil {
c.JSON(http.StatusNotFound, gin.H{"error": "assignment not found"})
return
}
if role != "admin" {
isTA := false
if h.stores.Teams != nil {
isTA, _ = h.stores.Teams.IsTeamAdmin(c.Request.Context(), a.TeamID, userID)
}
if !isTA {
c.JSON(http.StatusForbidden, gin.H{"error": "team admin access required"})
return
}
}
n, err := h.stores.Workflows.ReassignAssignment(c.Request.Context(), assignmentID, body.UserID)
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to reassign assignment"})
return
}
if n == 0 {
c.JSON(http.StatusConflict, gin.H{"error": "assignment not in claimed state"})
return
}
// Emit WS event
channelID, _ := h.stores.Workflows.GetAssignmentChannelID(c.Request.Context(), assignmentID)
if h.hub != nil && channelID != "" {
payload, _ := json.Marshal(map[string]any{
"assignment_id": assignmentID,
"reassigned_by": userID,
"new_assignee": body.UserID,
"channel_id": channelID,
})
evt := events.Event{
Label: "workflow.reassigned",
Payload: payload,
Ts: time.Now().UnixMilli(),
}
pids, _ := h.stores.Channels.ListUserParticipantIDs(c.Request.Context(), channelID, "")
for _, uid := range pids {
h.hub.PublishToUser(uid, evt)
}
}
c.JSON(http.StatusOK, gin.H{"reassigned": true, "assignment_id": assignmentID, "new_assignee": body.UserID})
}
// CancelAssignment cancels a single assignment.
// POST /api/v1/workflow-assignments/:id/cancel
func (h *WorkflowAssignmentHandler) CancelAssignment(c *gin.Context) {
assignmentID := c.Param("id")
userID := c.GetString("user_id")
role, _ := c.Get("role")
// Auth: team admin or global admin
a, err := h.stores.Workflows.GetAssignmentByID(c.Request.Context(), assignmentID)
if err != nil || a == nil {
c.JSON(http.StatusNotFound, gin.H{"error": "assignment not found"})
return
}
if role != "admin" {
isTA := false
if h.stores.Teams != nil {
isTA, _ = h.stores.Teams.IsTeamAdmin(c.Request.Context(), a.TeamID, userID)
}
if !isTA {
c.JSON(http.StatusForbidden, gin.H{"error": "team admin access required"})
return
}
}
n, err := h.stores.Workflows.CancelAssignment(c.Request.Context(), assignmentID)
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to cancel assignment"})
return
}
if n == 0 {
c.JSON(http.StatusConflict, gin.H{"error": "assignment not in cancellable state"})
return
}
c.JSON(http.StatusOK, gin.H{"cancelled": true, "assignment_id": assignmentID})
}
// CommentOnAssignment adds a review comment to an assignment.
// POST /api/v1/workflow-assignments/:id/comment
func (h *WorkflowAssignmentHandler) CommentOnAssignment(c *gin.Context) {
assignmentID := c.Param("id")
userID := c.GetString("user_id")
var body struct {
Text string `json:"text"`
}
if err := c.ShouldBindJSON(&body); err != nil || body.Text == "" {
c.JSON(http.StatusBadRequest, gin.H{"error": "text is required"})
return
}
comment := store.ReviewComment{
Text: body.Text,
UserID: userID,
CreatedAt: time.Now().UTC().Format(time.RFC3339),
}
if err := h.stores.Workflows.AddReviewComment(c.Request.Context(), assignmentID, comment); err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to add comment: " + err.Error()})
return
}
c.JSON(http.StatusOK, gin.H{"ok": true, "comment": comment})
}
// GetAssignment returns a single assignment with review comments.
// GET /api/v1/workflow-assignments/:id
func (h *WorkflowAssignmentHandler) GetAssignment(c *gin.Context) {
assignmentID := c.Param("id")
a, err := h.stores.Workflows.GetAssignmentByID(c.Request.Context(), assignmentID)
if err != nil {
c.JSON(http.StatusNotFound, gin.H{"error": fmt.Sprintf("assignment not found: %v", err)})
return
}
// Parse review_comments for the response
var comments []store.ReviewComment
if len(a.ReviewComments) > 0 {
_ = json.Unmarshal(a.ReviewComments, &comments)
}
c.JSON(http.StatusOK, gin.H{
"id": a.ID,
"channel_id": a.ChannelID,
"stage": a.Stage,
"team_id": a.TeamID,
"assigned_to": a.AssignedTo,
"status": a.Status,
"review_comments": comments,
"created_at": a.CreatedAt,
"claimed_at": a.ClaimedAt,
"completed_at": a.CompletedAt,
})
}

View File

@@ -1,137 +0,0 @@
package handlers
import (
"fmt"
"log"
"net/http"
"github.com/gin-gonic/gin"
"switchboard-core/models"
"switchboard-core/store"
)
// ── Workflow Entry Handler ──────────────────
// Handles the visitor-facing workflow start flow.
// Landing page rendering is in pages/pages.go (RenderWorkflowLanding).
type WorkflowEntryHandler struct {
stores store.Stores
}
func NewWorkflowEntryHandler(stores store.Stores) *WorkflowEntryHandler {
return &WorkflowEntryHandler{stores: stores}
}
// StartVisitor creates an anonymous session + workflow instance for a visitor.
// POST /api/v1/workflow-entry/:scope/:slug
func (h *WorkflowEntryHandler) StartVisitor(c *gin.Context) {
ctx := c.Request.Context()
scope := c.Param("scope")
slug := c.Param("slug")
var teamID *string
if scope != "global" {
teamID = &scope
}
wf, err := h.stores.Workflows.GetBySlug(ctx, teamID, slug)
if err != nil || wf == nil {
c.JSON(http.StatusNotFound, gin.H{"error": "workflow not found"})
return
}
if !wf.IsActive {
c.JSON(http.StatusBadRequest, gin.H{"error": "workflow not active"})
return
}
if wf.EntryMode != "public_link" {
c.JSON(http.StatusForbidden, gin.H{"error": "workflow requires authentication"})
return
}
ver, err := h.stores.Workflows.GetLatestVersion(ctx, wf.ID)
if err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": "workflow has no published version"})
return
}
stages, err := h.stores.Workflows.ListStages(ctx, wf.ID)
if err != nil || len(stages) == 0 {
c.JSON(http.StatusBadRequest, gin.H{"error": "workflow has no stages"})
return
}
// Create the workflow channel (owned by workflow creator)
ch := &models.Channel{
UserID: wf.CreatedBy,
Title: wf.Name,
Description: wf.Description,
Type: "workflow",
TeamID: wf.TeamID,
}
if err := h.stores.Channels.Create(ctx, ch); err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to create channel"})
return
}
// Set workflow columns
err = h.stores.Channels.SetWorkflowInstance(ctx, ch.ID, wf.ID, ver.VersionNumber, []byte("{}"), "active")
if err != nil {
log.Printf("Failed to set workflow columns: %v", err)
}
// Enable anonymous access + auto AI mode for visitor entry
_ = h.stores.Channels.Update(ctx, ch.ID, map[string]interface{}{
"allow_anonymous": true,
"ai_mode": "auto",
})
// Create anonymous session
sessionToken := store.NewID()
visitorCount, _ := h.stores.Sessions.CountForChannel(ctx, ch.ID)
displayName := "Visitor"
if visitorCount > 0 {
displayName = fmt.Sprintf("Visitor %d", visitorCount+1)
}
sess := &models.SessionParticipant{
SessionToken: sessionToken,
ChannelID: ch.ID,
DisplayName: displayName,
}
if err := h.stores.Sessions.Create(ctx, sess); err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to create session"})
return
}
// Add session as channel participant
_ = h.stores.Channels.AddParticipant(ctx, &models.ChannelParticipant{
ChannelID: ch.ID,
ParticipantType: "session",
ParticipantID: sess.ID,
Role: "visitor",
})
// Bind stage 0 persona
if stages[0].PersonaID != nil {
_ = h.stores.Channels.AddParticipant(ctx, &models.ChannelParticipant{
ChannelID: ch.ID,
ParticipantType: "persona",
ParticipantID: *stages[0].PersonaID,
Role: "member",
})
}
// Set session cookie (30 day expiry, matching v0.24.3)
c.SetCookie("sb_session", sessionToken, 30*24*60*60, "/", "", false, true)
// Redirect to workflow page — visitor lands on existing /w/:channelId
stageMode := stages[0].StageMode
if stageMode == "" {
stageMode = "chat_only"
}
c.JSON(http.StatusCreated, gin.H{
"channel_id": ch.ID,
"session_id": sess.ID,
"redirect_to": "/w/" + ch.ID,
"stage_mode": stageMode,
})
}

View File

@@ -1,288 +0,0 @@
package handlers
import (
"context"
"encoding/json"
"log"
"net/http"
"github.com/gin-gonic/gin"
"switchboard-core/events"
"switchboard-core/models"
"switchboard-core/sandbox"
"switchboard-core/store"
"switchboard-core/tools"
"switchboard-core/workflow"
"go.starlark.net/starlark"
)
// ── Workflow Form Handler ─────────────────────
// Handles typed form submission for workflow stages.
type WorkflowFormHandler struct {
stores store.Stores
runner *sandbox.Runner
hub *events.Hub
}
func NewWorkflowFormHandler(stores store.Stores, runner *sandbox.Runner, hub *events.Hub) *WorkflowFormHandler {
return &WorkflowFormHandler{stores: stores, runner: runner, hub: hub}
}
// GetFormTemplate returns the current stage's typed form template.
// GET /api/v1/w/:id/form
func (h *WorkflowFormHandler) GetFormTemplate(c *gin.Context) {
ctx := c.Request.Context()
channelID := c.Param("id")
ws, err := h.stores.Channels.GetWorkflowStatus(ctx, channelID)
if err != nil || ws == nil {
c.JSON(http.StatusNotFound, gin.H{"error": "workflow channel not found"})
return
}
if ws.WorkflowID == nil || *ws.WorkflowID == "" {
c.JSON(http.StatusNotFound, gin.H{"error": "not a workflow channel"})
return
}
stages, err := h.stores.Workflows.ListStages(ctx, *ws.WorkflowID)
if err != nil || ws.CurrentStage >= len(stages) {
c.JSON(http.StatusNotFound, gin.H{"error": "stage not found"})
return
}
stage := stages[ws.CurrentStage]
tpl := models.ParseTypedFormTemplate(stage.FormTemplate)
// Check if form was already submitted for this stage
var formSubmitted bool
if len(ws.StageData) > 0 {
var sd map[string]interface{}
if json.Unmarshal(ws.StageData, &sd) == nil {
_, formSubmitted = sd["_form_submitted"]
}
}
c.JSON(http.StatusOK, gin.H{
"stage_name": stage.Name,
"stage_mode": stage.StageMode,
"form_template": tpl,
"form_submitted": formSubmitted,
"current_stage": ws.CurrentStage,
"status": ws.Status,
})
}
// SubmitForm handles form data submission for a workflow stage.
// POST /api/v1/w/:id/form-submit
func (h *WorkflowFormHandler) SubmitForm(c *gin.Context) {
ctx := c.Request.Context()
channelID := c.Param("id")
// 1. Verify workflow channel
ws, err := h.stores.Channels.GetWorkflowStatus(ctx, channelID)
if err != nil || ws == nil {
c.JSON(http.StatusNotFound, gin.H{"error": "workflow channel not found"})
return
}
if ws.Status != "active" {
c.JSON(http.StatusBadRequest, gin.H{"error": "workflow is " + ws.Status + ", cannot submit form"})
return
}
if ws.WorkflowID == nil || *ws.WorkflowID == "" {
c.JSON(http.StatusBadRequest, gin.H{"error": "not a workflow channel"})
return
}
// 2. Load stage definition
stages, err := h.stores.Workflows.ListStages(ctx, *ws.WorkflowID)
if err != nil || ws.CurrentStage >= len(stages) {
c.JSON(http.StatusBadRequest, gin.H{"error": "stage not found"})
return
}
stage := stages[ws.CurrentStage]
if stage.StageMode == models.StageModeChatOnly {
c.JSON(http.StatusBadRequest, gin.H{"error": "this stage does not accept form submissions"})
return
}
// 3. Parse typed form template
tpl := models.ParseTypedFormTemplate(stage.FormTemplate)
if tpl == nil {
c.JSON(http.StatusBadRequest, gin.H{"error": "stage has no typed form template"})
return
}
// 4. Parse submitted data (flat fields per ICD spec)
var formData map[string]interface{}
if err := c.ShouldBindJSON(&formData); err != nil || len(formData) == 0 {
c.JSON(http.StatusBadRequest, gin.H{"error": "form data is required"})
return
}
// 5. Go-level validation
if errs := models.ValidateFormData(tpl, formData); len(errs) > 0 {
c.JSON(http.StatusBadRequest, gin.H{"error": "validation failed", "errors": errs})
return
}
// 6. Starlark validate hook (if configured)
if tpl.Hooks != nil && tpl.Hooks.PackageID != "" && tpl.Hooks.Validate != "" {
if hookErrs := h.runValidateHook(c, tpl.Hooks, channelID, formData, ws.StageData, stage.Name); hookErrs != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": "validation failed", "errors": hookErrs})
return
}
}
// 7. Merge into stage_data
formData["_form_submitted"] = true
dataJSON, _ := json.Marshal(formData)
mergedData := tools.MergeWorkflowStageData(ctx, h.stores.Channels, channelID, dataJSON)
// 8. Starlark on_submit hook (fire-and-forget)
if tpl.Hooks != nil && tpl.Hooks.PackageID != "" && tpl.Hooks.OnSubmit != "" {
go h.runOnSubmitHook(tpl.Hooks, channelID, formData, json.RawMessage(mergedData), stage.Name)
}
// 9. Auto-advance if form_only + auto_transition
if stage.StageMode == models.StageModeFormOnly && stage.AutoTransition {
nextStage, _ := workflow.ResolveNextStage(stages, ws.CurrentStage, json.RawMessage(mergedData))
if nextStage >= len(stages) {
// Complete
_ = h.stores.Channels.CompleteWorkflow(ctx, channelID, nextStage, json.RawMessage(mergedData))
tools.CreateWorkflowStageNote(ctx, h.stores, channelID, ws.CurrentStage, dataJSON, "")
h.emitFormEvent(channelID, "workflow.completed", ws.CurrentStage)
c.JSON(http.StatusOK, gin.H{"status": "completed", "current_stage": nextStage})
return
}
_ = h.stores.Channels.AdvanceWorkflowStage(ctx, channelID, nextStage, json.RawMessage(mergedData))
tools.CreateWorkflowStageNote(ctx, h.stores, channelID, ws.CurrentStage, dataJSON, "")
h.emitFormEvent(channelID, "workflow.advanced", nextStage)
c.JSON(http.StatusOK, gin.H{"status": "advanced", "current_stage": nextStage, "stage": stages[nextStage]})
return
}
// 10. Not auto-advancing — persist stage_data and confirm submission
_ = h.stores.Channels.AdvanceWorkflowStage(ctx, channelID, ws.CurrentStage, json.RawMessage(mergedData))
tools.CreateWorkflowStageNote(ctx, h.stores, channelID, ws.CurrentStage, dataJSON, "")
h.emitFormEvent(channelID, "workflow.form_submitted", ws.CurrentStage)
c.JSON(http.StatusOK, gin.H{"status": "submitted", "current_stage": ws.CurrentStage})
}
// ── Starlark Hooks ──────────────────────────
func (h *WorkflowFormHandler) runValidateHook(c *gin.Context, hooks *models.FormHooks, channelID string, data map[string]interface{}, stageData json.RawMessage, stageName string) []models.FieldError {
if h.runner == nil {
return nil
}
pkg, err := h.stores.Packages.Get(c.Request.Context(), hooks.PackageID)
if err != nil || pkg == nil {
log.Printf("[workflow-forms] validate hook: package %s not found", hooks.PackageID)
return nil
}
ctxDict := starlark.NewDict(4)
_ = ctxDict.SetKey(starlark.String("data"), jsonToStarlark(data))
_ = ctxDict.SetKey(starlark.String("stage_data"), starlark.String(string(stageData)))
_ = ctxDict.SetKey(starlark.String("stage_name"), starlark.String(stageName))
_ = ctxDict.SetKey(starlark.String("channel_id"), starlark.String(channelID))
val, _, err := h.runner.CallEntryPoint(c.Request.Context(), pkg, hooks.Validate,
starlark.Tuple{ctxDict}, nil, nil)
if err != nil {
log.Printf("[workflow-forms] validate hook error: %v", err)
return nil
}
return parseValidationResult(val)
}
func (h *WorkflowFormHandler) runOnSubmitHook(hooks *models.FormHooks, channelID string, data map[string]interface{}, stageData json.RawMessage, stageName string) {
if h.runner == nil {
return
}
ctx := context.Background()
pkg, err := h.stores.Packages.Get(ctx, hooks.PackageID)
if err != nil || pkg == nil {
return
}
ctxDict := starlark.NewDict(4)
_ = ctxDict.SetKey(starlark.String("data"), jsonToStarlark(data))
_ = ctxDict.SetKey(starlark.String("stage_data"), starlark.String(string(stageData)))
_ = ctxDict.SetKey(starlark.String("stage_name"), starlark.String(stageName))
_ = ctxDict.SetKey(starlark.String("channel_id"), starlark.String(channelID))
_, _, err = h.runner.CallEntryPoint(ctx, pkg, hooks.OnSubmit,
starlark.Tuple{ctxDict}, nil, nil)
if err != nil {
log.Printf("[workflow-forms] on_submit hook error: %v", err)
}
}
// ── Helpers ─────────────────────────────────
func (h *WorkflowFormHandler) emitFormEvent(channelID, label string, stage int) {
if h.hub == nil {
return
}
payload, _ := json.Marshal(map[string]interface{}{
"channel_id": channelID,
"stage": stage,
})
ctx := context.Background()
pids, err := h.stores.Channels.ListUserParticipantIDs(ctx, channelID, "")
if err != nil {
return
}
evt := events.Event{Label: label, Payload: payload}
for _, uid := range pids {
h.hub.PublishToUser(uid, evt)
}
}
// parseValidationResult extracts field errors from a Starlark return value.
// Expected: None (pass) or {"errors": [{"key": "...", "message": "..."}, ...]}
func parseValidationResult(val starlark.Value) []models.FieldError {
if val == nil || val == starlark.None {
return nil
}
d, ok := val.(*starlark.Dict)
if !ok {
return nil
}
errVal, found, _ := d.Get(starlark.String("errors"))
if !found {
return nil
}
list, ok := errVal.(*starlark.List)
if !ok {
return nil
}
var errs []models.FieldError
iter := list.Iterate()
defer iter.Done()
var item starlark.Value
for iter.Next(&item) {
ed, ok := item.(*starlark.Dict)
if !ok {
continue
}
keyVal, _, _ := ed.Get(starlark.String("key"))
msgVal, _, _ := ed.Get(starlark.String("message"))
k, _ := keyVal.(starlark.String)
m, _ := msgVal.(starlark.String)
if k != "" && m != "" {
errs = append(errs, models.FieldError{Key: string(k), Message: string(m)})
}
}
if len(errs) == 0 {
return nil
}
return errs
}

View File

@@ -1,845 +0,0 @@
package handlers
import (
"encoding/json"
"fmt"
"net/http"
"testing"
"github.com/gin-gonic/gin"
"switchboard-core/config"
authpkg "switchboard-core/auth"
"switchboard-core/database"
"switchboard-core/middleware"
"switchboard-core/store"
postgres "switchboard-core/store/postgres"
sqlite "switchboard-core/store/sqlite"
)
// ═══════════════════════════════════════════════════════════
// Extended harness: CRUD + instances + assignments + entry
// ═══════════════════════════════════════════════════════════
type workflowInstanceHarness struct {
*testHarness
stores store.Stores
adminToken string
adminID string
}
func setupWorkflowInstanceHarness(t *testing.T) *workflowInstanceHarness {
t.Helper()
database.RequireTestDB(t)
database.TruncateAll(t)
cfg := &config.Config{
JWTSecret: testJWTSecret,
BasePath: "",
}
var stores store.Stores
if database.IsSQLite() {
stores = sqlite.NewStores(database.TestDB)
} else {
stores = postgres.NewStores(database.TestDB)
}
userCache := middleware.NewUserStatusCache()
r := gin.New()
api := r.Group("/api/v1")
auth := NewAuthHandler(cfg, stores, nil, authpkg.NewBuiltinProvider())
api.POST("/auth/login", auth.Login)
api.POST("/auth/register", auth.Register)
protected := api.Group("")
protected.Use(middleware.Auth(cfg, stores.Users, userCache))
// Workflow CRUD
wfH := NewWorkflowHandler(stores)
protected.GET("/workflows", wfH.List)
protected.POST("/workflows", wfH.Create)
protected.GET("/workflows/:id", wfH.Get)
protected.PATCH("/workflows/:id", wfH.Update)
protected.DELETE("/workflows/:id", wfH.Delete)
protected.GET("/workflows/:id/stages", wfH.ListStages)
protected.POST("/workflows/:id/stages", wfH.CreateStage)
protected.PUT("/workflows/:id/stages/:sid", wfH.UpdateStage)
protected.DELETE("/workflows/:id/stages/:sid", wfH.DeleteStage)
protected.PATCH("/workflows/:id/stages/reorder", wfH.ReorderStages)
protected.POST("/workflows/:id/publish", wfH.Publish)
protected.GET("/workflows/:id/versions/:version", wfH.GetVersion)
// Workflow instances
wfInstH := NewWorkflowInstanceHandler(stores, nil, nil, nil)
protected.POST("/workflows/:id/start", wfInstH.Start)
protected.GET("/channels/:id/workflow/status", wfInstH.GetStatus)
protected.POST("/channels/:id/workflow/advance", wfInstH.Advance)
protected.POST("/channels/:id/workflow/reject", wfInstH.Reject)
// Assignments
wfAssignH := NewWorkflowAssignmentHandler(stores)
protected.GET("/workflow-assignments/mine", wfAssignH.ListMine)
protected.POST("/workflow-assignments/:id/claim", wfAssignH.Claim)
protected.POST("/workflow-assignments/:id/complete", wfAssignH.Complete)
// Teams (admin routes for creating teams + adding members)
teamH := NewTeamHandler(stores, nil)
admin := api.Group("/admin")
admin.Use(middleware.Auth(cfg, stores.Users, userCache), middleware.RequireAdmin())
admin.POST("/teams", teamH.CreateTeam)
admin.POST("/teams/:id/members", teamH.AddMember)
// Team-scoped assignment listing (mirrors main.go teamScoped)
teamScoped := protected.Group("/teams/:teamId")
teamAssignH := NewWorkflowAssignmentHandler(stores)
teamScoped.GET("/assignments", teamAssignH.ListForTeam)
// Visitor entry
wfEntry := NewWorkflowEntryHandler(stores)
r.POST("/api/v1/workflow-entry/:scope/:slug", wfEntry.StartVisitor)
// Seed admin user
adminID := seedInsertReturningID(t,
`INSERT INTO users (username, email, password_hash, role, handle, auth_source) VALUES ($1, $2, $3, $4, $5, $6) RETURNING id`,
"wfinst-admin", "wfinst-admin@test.com", "$2a$10$test", "admin", "wfinst-admin", "builtin",
)
token := makeToken(adminID, "wfinst-admin@test.com", "admin")
return &workflowInstanceHarness{
testHarness: &testHarness{router: r, t: t},
stores: stores,
adminToken: token,
adminID: adminID,
}
}
// createPublishedWorkflow is a helper that creates a workflow with N stages,
// activates it, and publishes it. Returns workflow ID + stage IDs.
func (h *workflowInstanceHarness) createPublishedWorkflow(name string, numStages int) (string, []string) {
h.t.Helper()
resp := h.request("POST", "/api/v1/workflows", h.adminToken, map[string]interface{}{
"name": name,
"entry_mode": "public_link",
})
if resp.Code != http.StatusCreated {
h.t.Fatalf("create workflow: %d: %s", resp.Code, resp.Body.String())
}
var wf struct{ ID string `json:"id"` }
json.Unmarshal(resp.Body.Bytes(), &wf)
stageIDs := make([]string, numStages)
for i := 0; i < numStages; i++ {
resp = h.request("POST", "/api/v1/workflows/"+wf.ID+"/stages", h.adminToken, map[string]interface{}{
"name": fmt.Sprintf("Stage %d", i),
"ordinal": i,
"history_mode": "full",
})
if resp.Code != http.StatusCreated {
h.t.Fatalf("create stage %d: %d: %s", i, resp.Code, resp.Body.String())
}
var st struct{ ID string `json:"id"` }
json.Unmarshal(resp.Body.Bytes(), &st)
stageIDs[i] = st.ID
}
// Activate + publish
h.request("PATCH", "/api/v1/workflows/"+wf.ID, h.adminToken, map[string]interface{}{
"is_active": true,
})
resp = h.request("POST", "/api/v1/workflows/"+wf.ID+"/publish", h.adminToken, nil)
if resp.Code != http.StatusCreated {
h.t.Fatalf("publish: %d: %s", resp.Code, resp.Body.String())
}
return wf.ID, stageIDs
}
// ═══════════════════════════════════════════════════════════
// #15 — Instance Lifecycle (currently untested)
// ═══════════════════════════════════════════════════════════
// TestWorkflowInstanceLifecycle exercises: start → status → advance → complete.
func TestWorkflowInstanceLifecycle(t *testing.T) {
h := setupWorkflowInstanceHarness(t)
wfID, _ := h.createPublishedWorkflow("Lifecycle Test", 2)
// ── Start ───────────────────────────────
resp := h.request("POST", "/api/v1/workflows/"+wfID+"/start", h.adminToken, nil)
if resp.Code != http.StatusCreated {
t.Fatalf("Start: got %d, body: %s", resp.Code, resp.Body.String())
}
var startResp struct {
ChannelID string `json:"channel_id"`
WorkflowVersion int `json:"workflow_version"`
CurrentStage int `json:"current_stage"`
}
json.Unmarshal(resp.Body.Bytes(), &startResp)
if startResp.ChannelID == "" {
t.Fatal("Start: channel_id empty")
}
if startResp.CurrentStage != 0 {
t.Errorf("Start: current_stage = %d, want 0", startResp.CurrentStage)
}
chID := startResp.ChannelID
// ── Status ──────────────────────────────
resp = h.request("GET", "/api/v1/channels/"+chID+"/workflow/status", h.adminToken, nil)
if resp.Code != http.StatusOK {
t.Fatalf("Status: got %d, body: %s", resp.Code, resp.Body.String())
}
var status struct {
WorkflowID *string `json:"workflow_id"`
Status string `json:"status"`
Stage int `json:"current_stage"`
StageData json.RawMessage `json:"stage_data"`
}
json.Unmarshal(resp.Body.Bytes(), &status)
if status.Status != "active" {
t.Errorf("Status: got %q, want \"active\"", status.Status)
}
if status.WorkflowID == nil || *status.WorkflowID != wfID {
t.Errorf("Status: workflow_id mismatch")
}
// ── Advance (stage 0 → 1) ───────────────
resp = h.request("POST", "/api/v1/channels/"+chID+"/workflow/advance", h.adminToken, map[string]interface{}{
"data": map[string]string{"name": "Jane", "email": "jane@test.com"},
})
if resp.Code != http.StatusOK {
t.Fatalf("Advance 0→1: got %d, body: %s", resp.Code, resp.Body.String())
}
var advResp struct {
Status string `json:"status"`
CurrentStage int `json:"current_stage"`
}
json.Unmarshal(resp.Body.Bytes(), &advResp)
if advResp.Status != "active" {
t.Errorf("Advance 0→1: status = %q, want \"active\"", advResp.Status)
}
if advResp.CurrentStage != 1 {
t.Errorf("Advance 0→1: current_stage = %d, want 1", advResp.CurrentStage)
}
// ── Verify merged data persisted ────────
resp = h.request("GET", "/api/v1/channels/"+chID+"/workflow/status", h.adminToken, nil)
json.Unmarshal(resp.Body.Bytes(), &status)
var data map[string]string
json.Unmarshal(status.StageData, &data)
if data["name"] != "Jane" || data["email"] != "jane@test.com" {
t.Errorf("Stage data after advance: got %v", data)
}
// ── Advance (stage 1 → complete) ────────
resp = h.request("POST", "/api/v1/channels/"+chID+"/workflow/advance", h.adminToken, map[string]interface{}{
"data": map[string]string{"resolution": "approved"},
})
if resp.Code != http.StatusOK {
t.Fatalf("Advance 1→complete: got %d, body: %s", resp.Code, resp.Body.String())
}
json.Unmarshal(resp.Body.Bytes(), &advResp)
if advResp.Status != "completed" {
t.Errorf("Completion: status = %q, want \"completed\"", advResp.Status)
}
// ── Status after completion ──────────────
resp = h.request("GET", "/api/v1/channels/"+chID+"/workflow/status", h.adminToken, nil)
json.Unmarshal(resp.Body.Bytes(), &status)
if status.Status != "completed" {
t.Errorf("Post-complete status: got %q, want \"completed\"", status.Status)
}
// ── Advance on completed should fail ────
resp = h.request("POST", "/api/v1/channels/"+chID+"/workflow/advance", h.adminToken, map[string]interface{}{
"data": map[string]string{"extra": "data"},
})
if resp.Code != http.StatusBadRequest {
t.Errorf("Advance after complete: got %d, want 400", resp.Code)
}
}
// TestWorkflowInstanceStart_InactiveWorkflow verifies start fails on inactive workflow.
func TestWorkflowInstanceStart_InactiveWorkflow(t *testing.T) {
h := setupWorkflowInstanceHarness(t)
// Create workflow with a stage but don't activate
resp := h.request("POST", "/api/v1/workflows", h.adminToken, map[string]interface{}{
"name": "Inactive WF",
})
var wf struct{ ID string `json:"id"` }
json.Unmarshal(resp.Body.Bytes(), &wf)
h.request("POST", "/api/v1/workflows/"+wf.ID+"/stages", h.adminToken, map[string]interface{}{
"name": "Stage 0", "ordinal": 0, "history_mode": "full",
})
resp = h.request("POST", "/api/v1/workflows/"+wf.ID+"/start", h.adminToken, nil)
if resp.Code != http.StatusBadRequest {
t.Errorf("Start inactive: got %d, want 400", resp.Code)
}
}
// TestWorkflowInstanceStart_NoPublishedVersion verifies start fails without a published version.
func TestWorkflowInstanceStart_NoPublishedVersion(t *testing.T) {
h := setupWorkflowInstanceHarness(t)
resp := h.request("POST", "/api/v1/workflows", h.adminToken, map[string]interface{}{
"name": "Unpublished WF",
})
var wf struct{ ID string `json:"id"` }
json.Unmarshal(resp.Body.Bytes(), &wf)
h.request("POST", "/api/v1/workflows/"+wf.ID+"/stages", h.adminToken, map[string]interface{}{
"name": "Stage 0", "ordinal": 0, "history_mode": "full",
})
// Activate but don't publish
h.request("PATCH", "/api/v1/workflows/"+wf.ID, h.adminToken, map[string]interface{}{
"is_active": true,
})
resp = h.request("POST", "/api/v1/workflows/"+wf.ID+"/start", h.adminToken, nil)
if resp.Code != http.StatusBadRequest {
t.Errorf("Start unpublished: got %d, want 400", resp.Code)
}
}
// ═══════════════════════════════════════════════════════════
// #15 — Reject validation
// ═══════════════════════════════════════════════════════════
// TestWorkflowReject exercises reject validation: reason required, stage-0 guard.
func TestWorkflowReject(t *testing.T) {
h := setupWorkflowInstanceHarness(t)
wfID, _ := h.createPublishedWorkflow("Reject Test", 3)
// Start
resp := h.request("POST", "/api/v1/workflows/"+wfID+"/start", h.adminToken, nil)
var start struct{ ChannelID string `json:"channel_id"` }
json.Unmarshal(resp.Body.Bytes(), &start)
chID := start.ChannelID
// ── Reject from stage 0 should fail ─────
resp = h.request("POST", "/api/v1/channels/"+chID+"/workflow/reject", h.adminToken, map[string]interface{}{
"reason": "want to go back",
})
if resp.Code != http.StatusBadRequest {
t.Errorf("Reject at stage 0: got %d, want 400", resp.Code)
}
// ── Reject without reason should fail ───
// Advance to stage 1 first
h.request("POST", "/api/v1/channels/"+chID+"/workflow/advance", h.adminToken, map[string]interface{}{
"data": map[string]string{"step": "0"},
})
resp = h.request("POST", "/api/v1/channels/"+chID+"/workflow/reject", h.adminToken, nil)
if resp.Code != http.StatusBadRequest {
t.Errorf("Reject no reason: got %d, want 400", resp.Code)
}
// ── Reject with empty reason should fail ─
resp = h.request("POST", "/api/v1/channels/"+chID+"/workflow/reject", h.adminToken, map[string]interface{}{
"reason": "",
})
if resp.Code != http.StatusBadRequest {
t.Errorf("Reject empty reason: got %d, want 400", resp.Code)
}
// ── Valid reject ────────────────────────
resp = h.request("POST", "/api/v1/channels/"+chID+"/workflow/reject", h.adminToken, map[string]interface{}{
"reason": "Missing email field",
})
if resp.Code != http.StatusOK {
t.Fatalf("Valid reject: got %d, body: %s", resp.Code, resp.Body.String())
}
var rejResp struct {
CurrentStage int `json:"current_stage"`
Reason string `json:"reason"`
}
json.Unmarshal(resp.Body.Bytes(), &rejResp)
if rejResp.CurrentStage != 0 {
t.Errorf("After reject: current_stage = %d, want 0", rejResp.CurrentStage)
}
if rejResp.Reason != "Missing email field" {
t.Errorf("Reject reason: got %q", rejResp.Reason)
}
}
// ═══════════════════════════════════════════════════════════
// #1 — Advance body field is "data" (not "stage_data")
// ═══════════════════════════════════════════════════════════
// TestWorkflowAdvance_BodyFieldName verifies the advance endpoint reads "data"
// from the request body, not "stage_data" (which the old ICD incorrectly documented).
func TestWorkflowAdvance_BodyFieldName(t *testing.T) {
h := setupWorkflowInstanceHarness(t)
wfID, _ := h.createPublishedWorkflow("Field Name Test", 2)
resp := h.request("POST", "/api/v1/workflows/"+wfID+"/start", h.adminToken, nil)
var start struct{ ChannelID string `json:"channel_id"` }
json.Unmarshal(resp.Body.Bytes(), &start)
chID := start.ChannelID
// Send data under the correct "data" key
h.request("POST", "/api/v1/channels/"+chID+"/workflow/advance", h.adminToken, map[string]interface{}{
"data": map[string]string{"collected": "yes"},
})
// Verify data was merged
resp = h.request("GET", "/api/v1/channels/"+chID+"/workflow/status", h.adminToken, nil)
var status struct{ StageData json.RawMessage `json:"stage_data"` }
json.Unmarshal(resp.Body.Bytes(), &status)
var data map[string]string
json.Unmarshal(status.StageData, &data)
if data["collected"] != "yes" {
t.Errorf("Data sent under 'data' key not merged: got %v", data)
}
}
// ═══════════════════════════════════════════════════════════
// #2 — GetStatus response uses "status" (not "workflow_status")
// ═══════════════════════════════════════════════════════════
// TestWorkflowGetStatus_ResponseShape verifies the exact JSON field names
// returned by the status endpoint.
func TestWorkflowGetStatus_ResponseShape(t *testing.T) {
h := setupWorkflowInstanceHarness(t)
wfID, _ := h.createPublishedWorkflow("Status Shape", 1)
resp := h.request("POST", "/api/v1/workflows/"+wfID+"/start", h.adminToken, nil)
var start struct{ ChannelID string `json:"channel_id"` }
json.Unmarshal(resp.Body.Bytes(), &start)
resp = h.request("GET", "/api/v1/channels/"+start.ChannelID+"/workflow/status", h.adminToken, nil)
if resp.Code != http.StatusOK {
t.Fatalf("GetStatus: %d", resp.Code)
}
// Parse as raw map to check exact field names
var raw map[string]interface{}
json.Unmarshal(resp.Body.Bytes(), &raw)
if _, ok := raw["status"]; !ok {
t.Error("Response missing 'status' field — should be 'status', not 'workflow_status'")
}
if _, ok := raw["workflow_status"]; ok {
t.Error("Response has 'workflow_status' — should be 'status' per ICD")
}
if _, ok := raw["workflow_id"]; !ok {
t.Error("Response missing 'workflow_id'")
}
if _, ok := raw["current_stage"]; !ok {
t.Error("Response missing 'current_stage'")
}
if _, ok := raw["last_activity_at"]; !ok {
t.Error("Response missing 'last_activity_at'")
}
}
// ═══════════════════════════════════════════════════════════
// #4 — StartVisitor response shape
// ═══════════════════════════════════════════════════════════
// TestWorkflowVisitorEntry_ResponseShape verifies the visitor start endpoint
// returns the correct field names (session_id + redirect_to, not session_token).
func TestWorkflowVisitorEntry_ResponseShape(t *testing.T) {
h := setupWorkflowInstanceHarness(t)
wfID, _ := h.createPublishedWorkflow("Visitor Test", 1)
// Read slug
resp := h.request("GET", "/api/v1/workflows/"+wfID, h.adminToken, nil)
var wf struct{ Slug string `json:"slug"` }
json.Unmarshal(resp.Body.Bytes(), &wf)
// Visitor start (no auth, no body)
resp = h.request("POST", "/api/v1/workflow-entry/global/"+wf.Slug, "", nil)
if resp.Code != http.StatusCreated {
t.Fatalf("Visitor start: got %d, body: %s", resp.Code, resp.Body.String())
}
var raw map[string]interface{}
json.Unmarshal(resp.Body.Bytes(), &raw)
if _, ok := raw["channel_id"]; !ok {
t.Error("Response missing 'channel_id'")
}
if _, ok := raw["session_id"]; !ok {
t.Error("Response missing 'session_id' — should be 'session_id', not 'session_token'")
}
if _, ok := raw["redirect_to"]; !ok {
t.Error("Response missing 'redirect_to'")
}
// These should NOT be present
if _, ok := raw["session_token"]; ok {
t.Error("Response has 'session_token' — field was renamed to 'session_id'")
}
if _, ok := raw["workflow"]; ok {
t.Error("Response has 'workflow' object — not in actual response")
}
}
// TestWorkflowVisitorEntry_TeamOnlyBlocked verifies team_only workflows reject visitors.
func TestWorkflowVisitorEntry_TeamOnlyBlocked(t *testing.T) {
h := setupWorkflowInstanceHarness(t)
resp := h.request("POST", "/api/v1/workflows", h.adminToken, map[string]interface{}{
"name": "Team Only WF",
"entry_mode": "team_only",
})
var wf struct {
ID string `json:"id"`
Slug string `json:"slug"`
}
json.Unmarshal(resp.Body.Bytes(), &wf)
h.request("POST", "/api/v1/workflows/"+wf.ID+"/stages", h.adminToken, map[string]interface{}{
"name": "S0", "ordinal": 0, "history_mode": "full",
})
h.request("PATCH", "/api/v1/workflows/"+wf.ID, h.adminToken, map[string]interface{}{
"is_active": true,
})
h.request("POST", "/api/v1/workflows/"+wf.ID+"/publish", h.adminToken, nil)
resp = h.request("POST", "/api/v1/workflow-entry/global/"+wf.Slug, "", nil)
if resp.Code != http.StatusForbidden {
t.Errorf("Visitor on team_only: got %d, want 403", resp.Code)
}
}
// ═══════════════════════════════════════════════════════════
// #6/#7 — webhook_url/webhook_secret round-trip
// EXPECTED TO FAIL until stores are fixed to SELECT/INSERT these columns.
// ═══════════════════════════════════════════════════════════
// TestWorkflowWebhookFields verifies webhook_url and webhook_secret survive
// create → get round-trip. This WILL FAIL until the Postgres and SQLite
// workflow stores are updated to include these columns in their queries.
func TestWorkflowWebhookFields(t *testing.T) {
h := setupWorkflowInstanceHarness(t)
resp := h.request("POST", "/api/v1/workflows", h.adminToken, map[string]interface{}{
"name": "Webhook WF",
"webhook_url": "https://hooks.example.com/wf",
})
if resp.Code != http.StatusCreated {
t.Fatalf("Create with webhook_url: %d: %s", resp.Code, resp.Body.String())
}
var wf struct {
ID string `json:"id"`
WebhookURL string `json:"webhook_url"`
}
json.Unmarshal(resp.Body.Bytes(), &wf)
// BUG #6: webhook_url is not included in the store's INSERT/SELECT,
// so it will be empty on the create response.
if wf.WebhookURL != "https://hooks.example.com/wf" {
t.Errorf("Create response webhook_url: got %q, want %q (BUG #6: store doesn't persist webhook_url)",
wf.WebhookURL, "https://hooks.example.com/wf")
}
// Also verify via GET
resp = h.request("GET", "/api/v1/workflows/"+wf.ID, h.adminToken, nil)
json.Unmarshal(resp.Body.Bytes(), &wf)
if wf.WebhookURL != "https://hooks.example.com/wf" {
t.Errorf("GET webhook_url: got %q, want %q (BUG #6: store doesn't SELECT webhook_url)",
wf.WebhookURL, "https://hooks.example.com/wf")
}
// BUG #7: PATCH webhook_url should work
resp = h.request("PATCH", "/api/v1/workflows/"+wf.ID, h.adminToken, map[string]interface{}{
"webhook_url": "https://hooks.example.com/updated",
})
if resp.Code != http.StatusOK {
t.Fatalf("PATCH webhook_url: %d: %s", resp.Code, resp.Body.String())
}
json.Unmarshal(resp.Body.Bytes(), &wf)
if wf.WebhookURL != "https://hooks.example.com/updated" {
t.Errorf("PATCH webhook_url: got %q, want %q (BUG #7: Update() ignores webhook_url)",
wf.WebhookURL, "https://hooks.example.com/updated")
}
}
// ═══════════════════════════════════════════════════════════
// #18 — ListMine should include unassigned-for-my-teams
// EXPECTED TO FAIL until ListMine query is fixed.
// ═══════════════════════════════════════════════════════════
// TestWorkflowAssignment_ListMineIncludesTeamUnassigned verifies that
// GET /workflow-assignments/mine returns both claimed assignments AND
// unassigned assignments for the user's teams.
func TestWorkflowAssignment_ListMineIncludesTeamUnassigned(t *testing.T) {
h := setupWorkflowInstanceHarness(t)
// Create a team and add a member
memberID := seedInsertReturningID(t,
`INSERT INTO users (username, email, password_hash, role, handle, auth_source) VALUES ($1, $2, $3, $4, $5, $6) RETURNING id`,
"wf-member", "wf-member@test.com", "$2a$10$test", "user", "wf-member", "builtin",
)
memberToken := makeToken(memberID, "wf-member@test.com", "user")
resp := h.request("POST", "/api/v1/admin/teams", h.adminToken, map[string]string{
"name": "Review Team", "description": "Test team",
})
if resp.Code != http.StatusCreated {
t.Fatalf("Create team: %d: %s", resp.Code, resp.Body.String())
}
var team map[string]interface{}
decode(resp, &team)
teamID := team["id"].(string)
h.request("POST", fmt.Sprintf("/api/v1/admin/teams/%s/members", teamID), h.adminToken,
map[string]string{"user_id": memberID, "role": "member"})
// Create a workflow with an assignment stage
wfResp := h.request("POST", "/api/v1/workflows", h.adminToken, map[string]interface{}{
"name": "Assignment WF",
"entry_mode": "team_only",
})
var wf struct{ ID string `json:"id"` }
json.Unmarshal(wfResp.Body.Bytes(), &wf)
h.request("POST", "/api/v1/workflows/"+wf.ID+"/stages", h.adminToken, map[string]interface{}{
"name": "Intake", "ordinal": 0, "history_mode": "full",
})
h.request("POST", "/api/v1/workflows/"+wf.ID+"/stages", h.adminToken, map[string]interface{}{
"name": "Review",
"ordinal": 1,
"history_mode": "full",
"assignment_team_id": teamID,
})
h.request("PATCH", "/api/v1/workflows/"+wf.ID, h.adminToken, map[string]interface{}{
"is_active": true,
})
h.request("POST", "/api/v1/workflows/"+wf.ID+"/publish", h.adminToken, nil)
// Start instance and advance to stage 1 (creates an assignment)
resp = h.request("POST", "/api/v1/workflows/"+wf.ID+"/start", h.adminToken, nil)
var start struct{ ChannelID string `json:"channel_id"` }
json.Unmarshal(resp.Body.Bytes(), &start)
h.request("POST", "/api/v1/channels/"+start.ChannelID+"/workflow/advance", h.adminToken, map[string]interface{}{
"data": map[string]string{"intake": "done"},
})
// BUG #18: ListMine should include the unassigned assignment for
// the member's team, but currently only returns claimed assignments.
resp = h.request("GET", "/api/v1/workflow-assignments/mine", memberToken, nil)
if resp.Code != http.StatusOK {
t.Fatalf("ListMine: %d: %s", resp.Code, resp.Body.String())
}
var listResp struct {
Data []struct {
ID string `json:"id"`
Status string `json:"status"`
TeamID string `json:"team_id"`
To *string `json:"assigned_to"`
} `json:"data"`
}
json.Unmarshal(resp.Body.Bytes(), &listResp)
if len(listResp.Data) == 0 {
t.Error("ListMine returned 0 assignments — BUG #18: should include unassigned assignments for user's teams")
}
// If we do get results, verify the shape
for _, a := range listResp.Data {
if a.TeamID != teamID {
t.Errorf("Assignment team_id: got %q, want %q", a.TeamID, teamID)
}
}
}
// ═══════════════════════════════════════════════════════════
// Assignment Claim + Complete (basic happy path)
// ═══════════════════════════════════════════════════════════
// TestWorkflowAssignment_ClaimAndComplete exercises the claim → complete flow.
func TestWorkflowAssignment_ClaimAndComplete(t *testing.T) {
h := setupWorkflowInstanceHarness(t)
// Seed assignment directly (simulates the advance handler creating one)
wfID, _ := h.createPublishedWorkflow("Assign WF", 2)
// Start + advance to create assignment stage
resp := h.request("POST", "/api/v1/workflows/"+wfID+"/start", h.adminToken, nil)
var start struct{ ChannelID string `json:"channel_id"` }
json.Unmarshal(resp.Body.Bytes(), &start)
// Manually insert an assignment (since our stages don't have assignment_team_id)
teamID := seedInsertReturningID(t,
`INSERT INTO teams (name, description, created_by) VALUES ($1, $2, $3) RETURNING id`,
"Claim Team", "Test", h.adminID,
)
assignID := seedInsertReturningID(t,
`INSERT INTO workflow_assignments (channel_id, stage, team_id) VALUES ($1, $2, $3) RETURNING id`,
start.ChannelID, 0, teamID,
)
// Claim
resp = h.request("POST", "/api/v1/workflow-assignments/"+assignID+"/claim", h.adminToken, nil)
if resp.Code != http.StatusOK {
t.Fatalf("Claim: got %d, body: %s", resp.Code, resp.Body.String())
}
var claimResp struct {
Claimed bool `json:"claimed"`
AssignmentID string `json:"assignment_id"`
}
json.Unmarshal(resp.Body.Bytes(), &claimResp)
if !claimResp.Claimed {
t.Error("Claim: claimed should be true")
}
// Double-claim should conflict
resp = h.request("POST", "/api/v1/workflow-assignments/"+assignID+"/claim", h.adminToken, nil)
if resp.Code != http.StatusConflict {
t.Errorf("Double claim: got %d, want 409", resp.Code)
}
// Complete
resp = h.request("POST", "/api/v1/workflow-assignments/"+assignID+"/complete", h.adminToken, nil)
if resp.Code != http.StatusOK {
t.Fatalf("Complete: got %d, body: %s", resp.Code, resp.Body.String())
}
// Double-complete should conflict
resp = h.request("POST", "/api/v1/workflow-assignments/"+assignID+"/complete", h.adminToken, nil)
if resp.Code != http.StatusConflict {
t.Errorf("Double complete: got %d, want 409", resp.Code)
}
}
// ═══════════════════════════════════════════════════════════
// #15 — Status on non-workflow channel returns 404
// ═══════════════════════════════════════════════════════════
func TestWorkflowGetStatus_NonWorkflowChannel(t *testing.T) {
h := setupWorkflowInstanceHarness(t)
resp := h.request("GET", "/api/v1/channels/00000000-0000-0000-0000-000000000000/workflow/status", h.adminToken, nil)
if resp.Code != http.StatusNotFound {
t.Errorf("Status on non-existent channel: got %d, want 404", resp.Code)
}
}
// ═══════════════════════════════════════════════════════════
// #15 — Data merge across multiple advances
// ═══════════════════════════════════════════════════════════
// TestWorkflowAdvance_DataMerge verifies that stage data accumulates
// across multiple advance calls (merge, not replace).
func TestWorkflowAdvance_DataMerge(t *testing.T) {
h := setupWorkflowInstanceHarness(t)
wfID, _ := h.createPublishedWorkflow("Merge Test", 3)
resp := h.request("POST", "/api/v1/workflows/"+wfID+"/start", h.adminToken, nil)
var start struct{ ChannelID string `json:"channel_id"` }
json.Unmarshal(resp.Body.Bytes(), &start)
chID := start.ChannelID
// Advance stage 0 → 1 with key "a"
h.request("POST", "/api/v1/channels/"+chID+"/workflow/advance", h.adminToken, map[string]interface{}{
"data": map[string]string{"a": "1"},
})
// Advance stage 1 → 2 with key "b"
h.request("POST", "/api/v1/channels/"+chID+"/workflow/advance", h.adminToken, map[string]interface{}{
"data": map[string]string{"b": "2"},
})
// Check accumulated data contains both keys
resp = h.request("GET", "/api/v1/channels/"+chID+"/workflow/status", h.adminToken, nil)
var status struct{ StageData json.RawMessage `json:"stage_data"` }
json.Unmarshal(resp.Body.Bytes(), &status)
var merged map[string]string
json.Unmarshal(status.StageData, &merged)
if merged["a"] != "1" {
t.Errorf("Merged data missing key 'a': got %v", merged)
}
if merged["b"] != "2" {
t.Errorf("Merged data missing key 'b': got %v", merged)
}
}
// ═══════════════════════════════════════════════════════════
// #17 — PATCH non-existent workflow
// ═══════════════════════════════════════════════════════════
// TestWorkflowPatchNonExistent verifies PATCH to a non-existent ID returns 404.
// Documents the confusing code path: Update affects 0 rows (no error),
// then GetByID returns sql.ErrNoRows → 404.
func TestWorkflowPatchNonExistent(t *testing.T) {
h := setupWorkflowInstanceHarness(t)
resp := h.request("PATCH", "/api/v1/workflows/00000000-0000-0000-0000-000000000000", h.adminToken, map[string]interface{}{
"name": "ghost",
})
if resp.Code != http.StatusNotFound {
t.Errorf("PATCH non-existent: got %d, want 404", resp.Code)
}
}
// ═══════════════════════════════════════════════════════════
// #19 — ListForTeam with ?status= filter
// ═══════════════════════════════════════════════════════════
// TestWorkflowAssignment_ListForTeamStatusFilter verifies the ?status=
// query param on GET /teams/:teamId/assignments.
func TestWorkflowAssignment_ListForTeamStatusFilter(t *testing.T) {
h := setupWorkflowInstanceHarness(t)
// Create team
resp := h.request("POST", "/api/v1/admin/teams", h.adminToken, map[string]string{
"name": "Filter Team", "description": "Test",
})
if resp.Code != http.StatusCreated {
t.Fatalf("Create team: %d: %s", resp.Code, resp.Body.String())
}
var team map[string]interface{}
decode(resp, &team)
teamID := team["id"].(string)
// Add self as member (needed for access)
h.request("POST", fmt.Sprintf("/api/v1/admin/teams/%s/members", teamID), h.adminToken,
map[string]string{"user_id": h.adminID, "role": "admin"})
// Create real channels for FK satisfaction
ch1ID := database.SeedTestChannel(t, h.adminID, "filter-ch-1")
ch2ID := database.SeedTestChannel(t, h.adminID, "filter-ch-2")
// Seed two assignments: one unassigned, one claimed
seedExec(t,
`INSERT INTO workflow_assignments (channel_id, stage, team_id, status) VALUES ($1, $2, $3, $4)`,
ch1ID, 0, teamID, "unassigned",
)
seedExec(t,
`INSERT INTO workflow_assignments (channel_id, stage, team_id, status, assigned_to) VALUES ($1, $2, $3, $4, $5)`,
ch2ID, 1, teamID, "claimed", h.adminID,
)
// Default (unassigned)
resp = h.request("GET", "/api/v1/teams/"+teamID+"/assignments", h.adminToken, nil)
if resp.Code != http.StatusOK {
t.Fatalf("ListForTeam default: %d: %s", resp.Code, resp.Body.String())
}
var lr struct {
Data []struct{ Status string `json:"status"` } `json:"data"`
}
json.Unmarshal(resp.Body.Bytes(), &lr)
if len(lr.Data) != 1 {
t.Errorf("Default filter: expected 1 unassigned, got %d", len(lr.Data))
} else if lr.Data[0].Status != "unassigned" {
t.Errorf("Default filter: expected status 'unassigned', got %q", lr.Data[0].Status)
}
// Explicit ?status=claimed
resp = h.request("GET", "/api/v1/teams/"+teamID+"/assignments?status=claimed", h.adminToken, nil)
json.Unmarshal(resp.Body.Bytes(), &lr)
if len(lr.Data) != 1 {
t.Errorf("?status=claimed: expected 1, got %d", len(lr.Data))
} else if lr.Data[0].Status != "claimed" {
t.Errorf("?status=claimed: expected status 'claimed', got %q", lr.Data[0].Status)
}
}

View File

@@ -1,551 +0,0 @@
package handlers
import (
"context"
"encoding/json"
"fmt"
"log"
"net/http"
"time"
"github.com/gin-gonic/gin"
"switchboard-core/events"
"switchboard-core/models"
"switchboard-core/notifications"
"switchboard-core/sandbox"
"switchboard-core/store"
"switchboard-core/tools"
"switchboard-core/workflow"
)
// ── Workflow Instance Handler ───────────────
// Manages the runtime lifecycle of workflow channels: starting instances,
// advancing/rejecting stages, and querying status.
type WorkflowInstanceHandler struct {
stores store.Stores
hub *events.Hub
notifSvc *notifications.Service
runner *sandbox.Runner
}
func NewWorkflowInstanceHandler(stores store.Stores, hub *events.Hub, notifSvc *notifications.Service, runner *sandbox.Runner) *WorkflowInstanceHandler {
return &WorkflowInstanceHandler{stores: stores, hub: hub, notifSvc: notifSvc, runner: runner}
}
// ── Start ───────────────────────────────────
// Start creates a new workflow channel from a published workflow version.
// POST /api/v1/workflows/:id/start
func (h *WorkflowInstanceHandler) Start(c *gin.Context) {
ctx := c.Request.Context()
wfID := c.Param("id")
userID := c.GetString("user_id")
wf, err := h.stores.Workflows.GetByID(ctx, wfID)
if err != nil {
c.JSON(http.StatusNotFound, gin.H{"error": "workflow not found"})
return
}
if !wf.IsActive {
c.JSON(http.StatusBadRequest, gin.H{"error": "workflow is not active"})
return
}
ver, err := h.stores.Workflows.GetLatestVersion(ctx, wfID)
if err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": "workflow has no published version — publish first"})
return
}
stages, err := h.stores.Workflows.ListStages(ctx, wfID)
if err != nil || len(stages) == 0 {
c.JSON(http.StatusBadRequest, gin.H{"error": "workflow has no stages"})
return
}
// Create the workflow channel
ch := &models.Channel{
UserID: userID,
Title: wf.Name,
Description: wf.Description,
Type: "workflow",
TeamID: wf.TeamID,
}
if err := h.stores.Channels.Create(ctx, ch); err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to create channel: " + err.Error()})
return
}
// Set workflow-specific columns (not part of base Channel.Create)
allowAnon := wf.EntryMode == "public_link"
err = h.stores.Channels.SetWorkflowInstance(ctx, ch.ID, wfID, ver.VersionNumber, []byte("{}"), "active")
if err != nil {
log.Printf("Failed to set workflow columns on channel %s: %v", ch.ID, err)
}
_ = h.stores.Channels.Update(ctx, ch.ID, map[string]interface{}{
"allow_anonymous": allowAnon,
"ai_mode": "auto",
})
// Add caller as channel owner
_ = h.stores.Channels.AddParticipant(ctx, &models.ChannelParticipant{
ChannelID: ch.ID,
ParticipantType: "user",
ParticipantID: userID,
Role: "owner",
})
// Bind stage 0 persona as participant
firstStage := stages[0]
if firstStage.PersonaID != nil {
_ = h.stores.Channels.AddParticipant(ctx, &models.ChannelParticipant{
ChannelID: ch.ID,
ParticipantType: "persona",
ParticipantID: *firstStage.PersonaID,
Role: "member",
})
}
c.JSON(http.StatusCreated, gin.H{
"channel_id": ch.ID,
"workflow_id": wfID,
"workflow_version": ver.VersionNumber,
"current_stage": 0,
"stage": firstStage,
})
}
// ── Status ──────────────────────────────────
// GetStatus returns the workflow state for a channel.
// GET /api/v1/channels/:id/workflow/status
func (h *WorkflowInstanceHandler) GetStatus(c *gin.Context) {
channelID := c.Param("id")
ws, err := h.stores.Channels.GetWorkflowStatus(c.Request.Context(), channelID)
if err != nil || ws == nil {
c.JSON(http.StatusNotFound, gin.H{"error": "workflow channel not found"})
return
}
c.JSON(http.StatusOK, ws)
}
// ── Advance ─────────────────────────────────
// Advance moves the workflow to the next stage.
// POST /api/v1/channels/:id/workflow/advance
func (h *WorkflowInstanceHandler) Advance(c *gin.Context) {
ctx := c.Request.Context()
channelID := c.Param("id")
workflowID, currentStage, status, err := h.readWorkflowState(ctx, channelID)
if err != nil {
c.JSON(http.StatusNotFound, gin.H{"error": "workflow channel not found"})
return
}
if status != "active" {
c.JSON(http.StatusBadRequest, gin.H{"error": "workflow is " + status + ", cannot advance"})
return
}
stages, err := h.stores.Workflows.ListStages(ctx, workflowID)
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to load stages"})
return
}
var body struct {
Data json.RawMessage `json:"data,omitempty"`
}
_ = c.ShouldBindJSON(&body)
mergedData := tools.MergeWorkflowStageData(ctx, h.stores.Channels, channelID, body.Data)
nextStage, err := workflow.ResolveNextStage(stages, currentStage, json.RawMessage(mergedData))
if err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": "routing error: " + err.Error()})
return
}
if nextStage >= len(stages) {
// Workflow complete
err = h.stores.Channels.CompleteWorkflow(ctx, channelID, nextStage, json.RawMessage(mergedData))
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to complete workflow"})
return
}
tools.CreateWorkflowStageNote(ctx, h.stores, channelID, currentStage, body.Data, "")
// v0.27.0: Emit workflow.completed WS event
h.emitWorkflowEvent("workflow.completed", channelID, map[string]any{
"channel_id": channelID, "workflow_id": workflowID, "stage": nextStage,
})
// v0.27.0: on_complete chaining — trigger target workflow if configured
h.triggerOnComplete(ctx, workflowID, channelID, mergedData)
c.JSON(http.StatusOK, gin.H{"status": "completed", "current_stage": nextStage})
return
}
// Advance to next stage
err = h.stores.Channels.AdvanceWorkflowStage(ctx, channelID, nextStage, json.RawMessage(mergedData))
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to advance stage"})
return
}
tools.CreateWorkflowStageNote(ctx, h.stores, channelID, currentStage, body.Data, "")
// v0.35.0: Fire on_advance hook (can enrich stage_data)
currentStageDef := stages[currentStage]
if hookResult := FireOnAdvanceHook(ctx, h.stores, h.runner, currentStageDef.TransitionRules,
channelID, currentStage, nextStage, json.RawMessage(mergedData)); hookResult != nil {
if hookResult.Error != "" {
log.Printf("[workflow] on_advance hook rejected: %s", hookResult.Error)
} else if hookResult.EnrichedData != nil {
_ = h.stores.Channels.AdvanceWorkflowStage(ctx, channelID, nextStage, hookResult.EnrichedData)
}
}
nextStageDef := stages[nextStage]
// Bind next persona (skip for form_only — no LLM needed)
if nextStageDef.StageMode != models.StageModeFormOnly && nextStageDef.PersonaID != nil {
alreadyIn, _ := h.stores.Channels.IsParticipant(ctx, channelID, "persona", *nextStageDef.PersonaID)
if !alreadyIn {
_ = h.stores.Channels.AddParticipant(ctx, &models.ChannelParticipant{
ChannelID: channelID,
ParticipantType: "persona",
ParticipantID: *nextStageDef.PersonaID,
Role: "member",
})
}
}
// v0.27.0: Assignment + round-robin + WS notifications
if nextStageDef.AssignmentTeamID != nil {
assignmentID := tools.CreateWorkflowAssignment(ctx, h.stores, channelID, nextStage, *nextStageDef.AssignmentTeamID)
// Round-robin auto-assignment if configured
assignedTo := h.tryRoundRobin(ctx, nextStageDef, assignmentID)
// Notify team members about new assignment
h.notifyAssignment(ctx, *nextStageDef.AssignmentTeamID, channelID, nextStageDef.Name, assignedTo)
}
// v0.27.0: Emit workflow.advanced WS event
h.emitWorkflowEvent("workflow.advanced", channelID, map[string]any{
"channel_id": channelID, "workflow_id": workflowID,
"stage": nextStage, "stage_name": nextStageDef.Name,
})
c.JSON(http.StatusOK, gin.H{
"status": "active",
"current_stage": nextStage,
"stage": nextStageDef,
})
}
// ── Reject ──────────────────────────────────
// Reject returns the workflow to the previous stage with a reason.
// POST /api/v1/channels/:id/workflow/reject
func (h *WorkflowInstanceHandler) Reject(c *gin.Context) {
ctx := c.Request.Context()
channelID := c.Param("id")
var body struct {
Reason string `json:"reason"`
}
if err := c.ShouldBindJSON(&body); err != nil || body.Reason == "" {
c.JSON(http.StatusBadRequest, gin.H{"error": "reason is required"})
return
}
_, currentStage, status, err := h.readWorkflowState(ctx, channelID)
if err != nil {
c.JSON(http.StatusNotFound, gin.H{"error": "workflow channel not found"})
return
}
if status != "active" {
c.JSON(http.StatusBadRequest, gin.H{"error": "workflow is " + status + ", cannot reject"})
return
}
if currentStage <= 0 {
c.JSON(http.StatusBadRequest, gin.H{"error": "cannot reject from the first stage"})
return
}
prevStage := currentStage - 1
err = h.stores.Channels.RejectWorkflowToStage(ctx, channelID, prevStage)
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to reject stage"})
return
}
// Persist rejection as system message
if h.stores.Messages != nil {
_ = h.stores.Messages.Create(ctx, &models.Message{
ChannelID: channelID,
Role: "system",
Content: "Stage rejected: " + body.Reason,
ParticipantType: "user",
ParticipantID: c.GetString("user_id"),
})
}
c.JSON(http.StatusOK, gin.H{
"status": "active",
"current_stage": prevStage,
"reason": body.Reason,
})
}
// ── Cancel (v0.37.15) ────────────────────────
// CancelInstance cancels a workflow instance and all its open assignments.
// POST /api/v1/channels/:id/workflow/cancel
func (h *WorkflowInstanceHandler) CancelInstance(c *gin.Context) {
ctx := c.Request.Context()
channelID := c.Param("id")
userID := c.GetString("user_id")
role, _ := c.Get("role")
workflowID, _, status, err := h.readWorkflowState(ctx, channelID)
if err != nil {
c.JSON(http.StatusNotFound, gin.H{"error": "workflow channel not found"})
return
}
if status != "active" {
c.JSON(http.StatusBadRequest, gin.H{"error": "workflow is " + status + ", cannot cancel"})
return
}
// Auth: instance owner, team admin, or global admin
if role != "admin" {
ch, err := h.stores.Channels.GetByID(ctx, channelID)
if err != nil {
c.JSON(http.StatusNotFound, gin.H{"error": "channel not found"})
return
}
isOwner := ch.UserID == userID
isTeamAdmin := false
if ch.TeamID != nil && h.stores.Teams != nil {
isTeamAdmin, _ = h.stores.Teams.IsTeamAdmin(ctx, *ch.TeamID, userID)
}
if !isOwner && !isTeamAdmin {
c.JSON(http.StatusForbidden, gin.H{"error": "only instance owner, team admin, or global admin can cancel"})
return
}
}
// Cancel the workflow instance
if err := h.stores.Channels.CancelWorkflow(ctx, channelID); err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to cancel workflow"})
return
}
// Cancel all open assignments
cancelled, _ := h.stores.Workflows.CancelAssignmentsForChannel(ctx, channelID)
// Emit WS event
h.emitWorkflowEvent("workflow.cancelled", channelID, map[string]any{
"channel_id": channelID,
"workflow_id": workflowID,
"cancelled_by": userID,
"assignments_cancelled": cancelled,
})
c.JSON(http.StatusOK, gin.H{
"cancelled": true,
"channel_id": channelID,
"assignments_cancelled": cancelled,
})
}
// CancelTeamInstance cancels a workflow instance via the team-admin monitor.
// POST /api/v1/teams/:teamId/workflows/monitor/instances/:channelId/cancel
func (h *WorkflowInstanceHandler) CancelTeamInstance(c *gin.Context) {
teamID := c.Param("teamId")
channelID := c.Param("channelId")
userID := c.GetString("user_id")
role, _ := c.Get("role")
ctx := c.Request.Context()
// Auth: team admin or global admin
if role != "admin" {
if h.stores.Teams == nil {
c.JSON(http.StatusForbidden, gin.H{"error": "team admin access required"})
return
}
isTA, _ := h.stores.Teams.IsTeamAdmin(ctx, teamID, userID)
if !isTA {
c.JSON(http.StatusForbidden, gin.H{"error": "team admin access required"})
return
}
}
// Verify the channel belongs to this team
ch, err := h.stores.Channels.GetByID(ctx, channelID)
if err != nil || ch.TeamID == nil || *ch.TeamID != teamID {
c.JSON(http.StatusNotFound, gin.H{"error": "workflow instance not found for this team"})
return
}
_, _, status, err := h.readWorkflowState(ctx, channelID)
if err != nil {
c.JSON(http.StatusNotFound, gin.H{"error": "workflow channel not found"})
return
}
if status != "active" {
c.JSON(http.StatusBadRequest, gin.H{"error": "workflow is " + status + ", cannot cancel"})
return
}
if err := h.stores.Channels.CancelWorkflow(ctx, channelID); err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to cancel workflow"})
return
}
cancelled, _ := h.stores.Workflows.CancelAssignmentsForChannel(ctx, channelID)
h.emitWorkflowEvent("workflow.cancelled", channelID, map[string]any{
"channel_id": channelID,
"cancelled_by": userID,
"assignments_cancelled": cancelled,
})
c.JSON(http.StatusOK, gin.H{
"cancelled": true,
"channel_id": channelID,
"assignments_cancelled": cancelled,
})
}
// ── Helpers ─────────────────────────────────
func (h *WorkflowInstanceHandler) readWorkflowState(ctx context.Context, channelID string) (workflowID string, currentStage int, status string, err error) {
ws, err := h.stores.Channels.GetWorkflowStatus(ctx, channelID)
if err != nil || ws == nil {
return "", 0, "", fmt.Errorf("workflow channel not found")
}
if ws.WorkflowID != nil {
workflowID = *ws.WorkflowID
}
return workflowID, ws.CurrentStage, ws.Status, nil
}
// emitWorkflowEvent pushes a workflow event to all user participants in the channel.
// Uses PublishToUser (not room-scoped Bus.Publish) because room subscriptions are
// not yet wired on the client side. See websocket.md § Room Model.
func (h *WorkflowInstanceHandler) emitWorkflowEvent(label, channelID string, data map[string]any) {
if h.hub == nil {
return
}
payload, _ := json.Marshal(data)
evt := events.Event{
Label: label,
Payload: payload,
Ts: time.Now().UnixMilli(),
}
// Send to all user participants in the channel
pids, err := h.stores.Channels.ListUserParticipantIDs(context.Background(), channelID, "")
if err != nil {
log.Printf("[ws] %s: failed to query participants for channel %s: %v", label, channelID[:min(8, len(channelID))], err)
return
}
for _, uid := range pids {
h.hub.PublishToUser(uid, evt)
}
}
// triggerOnComplete checks if the workflow has an on_complete chain config
// and starts the target workflow if so.
func (h *WorkflowInstanceHandler) triggerOnComplete(ctx context.Context, workflowID, channelID, mergedData string) {
// v0.27.3: Delegate to shared implementation (also handles webhooks)
go tools.TriggerWorkflowOnComplete(ctx, h.stores, workflowID, channelID, mergedData)
}
// tryRoundRobin checks if the stage has auto_assign:"round_robin" in
// transition_rules and assigns the newly created assignment to the
// least-recently-assigned team member. Returns the assigned user ID or "".
func (h *WorkflowInstanceHandler) tryRoundRobin(ctx context.Context, stage models.WorkflowStage, assignmentID string) string {
if assignmentID == "" || stage.AssignmentTeamID == nil {
return ""
}
// Check transition_rules for auto_assign
var rules struct {
AutoAssign string `json:"auto_assign"`
}
if len(stage.TransitionRules) > 0 {
_ = json.Unmarshal(stage.TransitionRules, &rules)
}
if rules.AutoAssign != "round_robin" {
return ""
}
assignedTo, err := h.stores.Workflows.TryRoundRobin(ctx, *stage.AssignmentTeamID, assignmentID)
if err != nil {
log.Printf("[workflow] round-robin: failed to auto-assign %s: %v", assignmentID, err)
return ""
}
if assignedTo != "" {
log.Printf("[workflow] round-robin: auto-assigned %s to user %s", assignmentID, assignedTo)
}
return assignedTo
}
// notifyAssignment sends notifications to team members about a new workflow assignment.
func (h *WorkflowInstanceHandler) notifyAssignment(ctx context.Context, teamID, channelID, stageName, assignedTo string) {
if h.notifSvc == nil || h.stores.Teams == nil {
return
}
members, err := h.stores.Teams.ListMembers(ctx, teamID)
if err != nil {
return
}
for _, m := range members {
title := "New workflow assignment"
body := "Stage '" + stageName + "' needs review"
if assignedTo != "" && m.UserID == assignedTo {
title = "Workflow assigned to you"
body = "Stage '" + stageName + "' has been assigned to you (round-robin)"
}
n := &models.Notification{
UserID: m.UserID,
Type: "workflow.assigned",
Title: title,
Body: body,
ResourceType: models.ResourceTypeChannel,
ResourceID: channelID,
}
if err := h.notifSvc.Notify(ctx, n); err != nil {
log.Printf("[workflow] notify assignment: %v", err)
}
}
// Also emit targeted WS event for immediate UI update
if h.hub != nil {
payload, _ := json.Marshal(map[string]any{
"channel_id": channelID, "team_id": teamID,
"stage_name": stageName, "assigned_to": assignedTo,
})
for _, m := range members {
h.hub.PublishToUser(m.UserID, events.Event{
Label: "workflow.assigned",
Payload: payload,
Ts: time.Now().UnixMilli(),
})
}
}
}
// createStageNote and mergeStageData are now in tools/workflow.go
// (shared between handler and workflow_advance tool).

View File

@@ -1,239 +0,0 @@
package handlers
// workflow_monitor.go — v0.35.0 Monitoring Dashboard
//
// Provides admin and team-scoped views of active workflow instances,
// stage funnels, and stale instance detection for SLA tracking.
import (
"context"
"net/http"
"strconv"
"time"
"github.com/gin-gonic/gin"
"switchboard-core/store"
)
// ── Workflow Monitor Handler ─────────────────
type WorkflowMonitorHandler struct {
stores store.Stores
}
func NewWorkflowMonitorHandler(stores store.Stores) *WorkflowMonitorHandler {
return &WorkflowMonitorHandler{stores: stores}
}
// ── Instance types ──────────────────────────
// MonitorInstance is a single active workflow instance for the monitoring dashboard.
type MonitorInstance struct {
ChannelID string `json:"channel_id"`
ChannelTitle string `json:"channel_title"`
WorkflowID string `json:"workflow_id"`
WorkflowName string `json:"workflow_name"`
CurrentStage int `json:"current_stage"`
StageName string `json:"stage_name"`
AssignedTo *string `json:"assigned_to,omitempty"`
AgeSeconds int64 `json:"age_seconds"`
StageAgeSeconds int64 `json:"stage_age_seconds"`
SLASeconds *int `json:"sla_seconds,omitempty"`
SLARemaining *int64 `json:"sla_remaining_seconds,omitempty"`
SLABreached bool `json:"sla_breached"`
LastActivityAt string `json:"last_activity_at"`
}
// FunnelEntry is a count of instances at a given stage.
type FunnelEntry struct {
StageOrdinal int `json:"stage_ordinal"`
StageName string `json:"stage_name"`
Count int `json:"count"`
}
// ── Endpoints ───────────────────────────────
// ListActiveInstances returns all active workflow instances.
// GET /api/v1/admin/workflows/monitor/instances
func (h *WorkflowMonitorHandler) ListActiveInstances(c *gin.Context) {
ctx := c.Request.Context()
instances, err := h.queryActiveInstances(ctx, "")
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to query instances: " + err.Error()})
return
}
c.JSON(http.StatusOK, gin.H{"data": instances})
}
// ListTeamActiveInstances returns active workflow instances for a team.
// GET /api/v1/teams/:teamId/workflows/monitor/instances
func (h *WorkflowMonitorHandler) ListTeamActiveInstances(c *gin.Context) {
teamID := c.Param("teamId")
ctx := c.Request.Context()
instances, err := h.queryActiveInstances(ctx, teamID)
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to query instances: " + err.Error()})
return
}
c.JSON(http.StatusOK, gin.H{"data": instances})
}
// GetFunnel returns stage-by-stage instance counts for a workflow.
// GET /api/v1/admin/workflows/monitor/funnel/:id
func (h *WorkflowMonitorHandler) GetFunnel(c *gin.Context) {
workflowID := c.Param("id")
ctx := c.Request.Context()
stages, err := h.stores.Workflows.ListStages(ctx, workflowID)
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to load stages"})
return
}
instances, err := h.queryActiveInstances(ctx, "")
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to query instances"})
return
}
// Count instances at each stage
counts := make(map[int]int)
for _, inst := range instances {
if inst.WorkflowID == workflowID {
counts[inst.CurrentStage]++
}
}
funnel := make([]FunnelEntry, len(stages))
for i, s := range stages {
funnel[i] = FunnelEntry{
StageOrdinal: s.Ordinal,
StageName: s.Name,
Count: counts[s.Ordinal],
}
}
c.JSON(http.StatusOK, gin.H{"data": funnel})
}
// ListStaleInstances returns workflow instances that haven't been touched recently.
// GET /api/v1/admin/workflows/monitor/stale
func (h *WorkflowMonitorHandler) ListStaleInstances(c *gin.Context) {
thresholdHours, _ := strconv.Atoi(c.DefaultQuery("threshold_hours", "48"))
if thresholdHours <= 0 {
thresholdHours = 48
}
ctx := c.Request.Context()
instances, err := h.queryActiveInstances(ctx, "")
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to query instances"})
return
}
now := time.Now().UTC()
threshold := time.Duration(thresholdHours) * time.Hour
var stale []MonitorInstance
for _, inst := range instances {
if inst.AgeSeconds > 0 && time.Duration(inst.AgeSeconds)*time.Second > threshold {
stale = append(stale, inst)
} else {
// Also check last_activity_at
if t, err := time.Parse(time.RFC3339, inst.LastActivityAt); err == nil {
if now.Sub(t) > threshold {
stale = append(stale, inst)
}
}
}
}
if stale == nil {
stale = []MonitorInstance{}
}
c.JSON(http.StatusOK, gin.H{"data": stale, "threshold_hours": thresholdHours})
}
// ── Helpers ─────────────────────────────────
func (h *WorkflowMonitorHandler) queryActiveInstances(ctx context.Context, teamIDFilter string) ([]MonitorInstance, error) {
// Query all active workflow channels via store
// This is a new query we build from existing store methods
channels, err := h.stores.Channels.ListByType(ctx, "workflow")
if err != nil {
return nil, err
}
now := time.Now().UTC()
var result []MonitorInstance
for _, ch := range channels {
ws, err := h.stores.Channels.GetWorkflowStatus(ctx, ch.ID)
if err != nil || ws == nil || ws.WorkflowID == nil || ws.Status != "active" {
continue
}
// Team filter
if teamIDFilter != "" {
if ch.TeamID == nil || *ch.TeamID != teamIDFilter {
continue
}
}
wf, err := h.stores.Workflows.GetByID(ctx, *ws.WorkflowID)
if err != nil || wf == nil {
continue
}
stages, _ := h.stores.Workflows.ListStages(ctx, *ws.WorkflowID)
var stageName string
var slaSeconds *int
if ws.CurrentStage < len(stages) {
stageName = stages[ws.CurrentStage].Name
slaSeconds = stages[ws.CurrentStage].SLASeconds
}
inst := MonitorInstance{
ChannelID: ch.ID,
ChannelTitle: ch.Title,
WorkflowID: *ws.WorkflowID,
WorkflowName: wf.Name,
CurrentStage: ws.CurrentStage,
StageName: stageName,
SLASeconds: slaSeconds,
LastActivityAt: "",
}
// Compute ages
inst.AgeSeconds = int64(now.Sub(ch.CreatedAt).Seconds())
if ws.LastActivityAt != nil {
inst.LastActivityAt = *ws.LastActivityAt
}
// Stage age from stage_entered_at
if ws.StageEnteredAt != nil {
if t, err := time.Parse(time.RFC3339, *ws.StageEnteredAt); err == nil {
inst.StageAgeSeconds = int64(now.Sub(t).Seconds())
// SLA computation
if slaSeconds != nil {
remaining := int64(*slaSeconds) - inst.StageAgeSeconds
inst.SLARemaining = &remaining
inst.SLABreached = remaining < 0
}
}
}
result = append(result, inst)
}
if result == nil {
result = []MonitorInstance{}
}
return result, nil
}

View File

@@ -219,7 +219,7 @@ func InstallWorkflowFromManifest(ctx *gin.Context, stores store.Stores, pkgID st
SurfacePkgID: s.SurfacePkgID,
}
if st.StageMode == "" {
st.StageMode = models.StageModeChatOnly
st.StageMode = models.StageModeCustom
}
if st.HistoryMode == "" {
st.HistoryMode = "full"

View File

@@ -195,10 +195,10 @@ func (h *WorkflowHandler) CreateStage(c *gin.Context) {
return
}
if st.StageMode == "" {
st.StageMode = models.StageModeChatOnly
st.StageMode = models.StageModeCustom
}
if !models.ValidStageModes[st.StageMode] {
c.JSON(http.StatusBadRequest, gin.H{"error": "stage_mode must be chat_only, form_only, form_chat, or review"})
c.JSON(http.StatusBadRequest, gin.H{"error": "stage_mode must be custom, form_only, form_chat, or review"})
return
}
if st.Ordinal == 0 {
@@ -227,7 +227,7 @@ func (h *WorkflowHandler) UpdateStage(c *gin.Context) {
return
}
if st.StageMode != "" && !models.ValidStageModes[st.StageMode] {
c.JSON(http.StatusBadRequest, gin.H{"error": "stage_mode must be chat_only, form_only, form_chat, or review"})
c.JSON(http.StatusBadRequest, gin.H{"error": "stage_mode must be custom, form_only, form_chat, or review"})
return
}
if err := h.stores.Workflows.UpdateStage(c.Request.Context(), &st); err != nil {