diff --git a/CHANGELOG.md b/CHANGELOG.md index ddb94d0..635c025 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,62 @@ # Changelog +## [0.27.3] — 2026-03-10 + +### Summary + +Task chaining — webhooks for external integration, AI-spawnable sub-tasks, +and workflow-to-workflow completion triggers. Also fixes a bug where the +`workflow_advance` tool didn't fire `on_complete` chains. + +### Added + +#### Webhook Delivery (`server/webhook/`) +- **`webhook.Deliver(url, secret, payload)`** — POST with HMAC-SHA256 signing. + 3 retries with exponential backoff (1s, 5s, 25s), 10s timeout per attempt. + `X-Switchboard-Signature` header for verification. +- **`webhook.GenerateSecret()`** — 32-byte hex random secret, auto-generated + on task creation when `webhook_url` is set. +- **`webhook.Payload`** struct: task_id, task_name, workflow_id, channel_id, + status, completed_at, output, stage_data, error. + +#### Task Webhooks +- Task executor fires webhook on completion, failure, and budget breach. + Runs in a goroutine — non-blocking. +- `webhook_secret` column on tasks table (migration 027). +- Secret auto-generated in task handler `Create` when `webhook_url` is provided. + +#### Workflow Webhooks +- `webhook_url` and `webhook_secret` columns on workflows table (migration 027). +- Webhook delivered on workflow completion (both handler and tool paths). + +#### `task_create` Tool +- AI-invocable tool that spawns sub-tasks from workflow/team contexts. +- **Predicates:** available in workflow channels and team channels only. + Blocked in service channels (prevents recursive task spawning). +- **Scope inheritance:** created tasks inherit parent channel's team scope. +- **Depth limit:** service channel check prevents tasks creating tasks. +- Parameters: name, prompt, schedule (cron or "once"), persona, model, + max_tokens, system_prompt. Validates cron before creation. + +#### Workflow `on_complete` Chaining Extraction +- `TriggerWorkflowOnComplete()` extracted to `tools/workflow.go` as shared + function. Used by both HTTP handler and `workflow_advance` tool. +- Handles both `on_complete` chain config and webhook delivery. + +### Fixed +- **`workflow_advance` tool completion path** now triggers `on_complete` + chaining. Previously only the HTTP `Advance` endpoint fired chains — + AI-driven stage advancement silently skipped them. + +### Changed +- `Workflow` model gains `WebhookURL`, `WebhookSecret` fields. +- `WorkflowPatch` gains `WebhookURL` field. +- `Task` model gains `WebhookSecret` field. +- `workflow_instances.go` `triggerOnComplete` method replaced with + one-line delegation to shared `tools.TriggerWorkflowOnComplete()`. + + + ## [0.27.2] — 2026-03-10 ### Summary diff --git a/VERSION b/VERSION index 76cab36..d0ca6fd 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -0.27.2 \ No newline at end of file +0.27.3 \ No newline at end of file diff --git a/docs/ROADMAP.md b/docs/ROADMAP.md index 13e7a0b..cea5c2b 100644 --- a/docs/ROADMAP.md +++ b/docs/ROADMAP.md @@ -166,7 +166,7 @@ v0.27.2 Task Execution ✅ (budgets, admin controls, run history, kill switch) │ -v0.27.3 Task Chaining +v0.27.3 Task Chaining ✅ (webhooks, task_create tool, workflow-to-workflow) │ @@ -973,16 +973,17 @@ Depends on: v0.27.1 (task scheduler). --- -## v0.27.3 — Task Chaining: Webhooks + Workflow-to-Workflow +## v0.27.3 — Task Chaining: Webhooks + Workflow-to-Workflow ✅ External integration and multi-workflow orchestration. Depends on: v0.27.2 (task budgets). -- [ ] Completion webhooks (POST to external URL on task/workflow completion) -- [ ] Webhook retry (3 attempts, exponential backoff) + HMAC signature -- [ ] `task_create` tool (AI spawns sub-tasks, `RequireWorkflow` predicate, depth limit) -- [ ] `on_complete` chaining via task system (completed workflow → one-shot task → target workflow) -- [ ] Webhook secret generation per task/workflow +- [x] Completion webhooks (POST to external URL on task/workflow completion) +- [x] Webhook retry (3 attempts, exponential backoff) + HMAC signature +- [x] `task_create` tool (AI spawns sub-tasks, RequireWorkflow/RequireTeam predicate, depth limit) +- [x] `on_complete` chaining via task system (completed workflow → one-shot task → target workflow) +- [x] Webhook secret generation per task/workflow +- [x] Bug fix: `workflow_advance` tool now triggers `on_complete` chaining on completion --- diff --git a/server/database/migrations/027_webhooks.sql b/server/database/migrations/027_webhooks.sql new file mode 100644 index 0000000..66e2866 --- /dev/null +++ b/server/database/migrations/027_webhooks.sql @@ -0,0 +1,8 @@ +-- 027_webhooks.sql — Webhook secrets + workflow webhook support (v0.27.3) + +-- Per-task webhook secret for HMAC signing +ALTER TABLE tasks ADD COLUMN IF NOT EXISTS webhook_secret TEXT; + +-- Workflows can also fire webhooks on completion +ALTER TABLE workflows ADD COLUMN IF NOT EXISTS webhook_url TEXT; +ALTER TABLE workflows ADD COLUMN IF NOT EXISTS webhook_secret TEXT; diff --git a/server/database/migrations/sqlite/027_webhooks.sql b/server/database/migrations/sqlite/027_webhooks.sql new file mode 100644 index 0000000..f2f93d3 --- /dev/null +++ b/server/database/migrations/sqlite/027_webhooks.sql @@ -0,0 +1,5 @@ +-- 027_webhooks.sql — Webhook secrets + workflow webhook support (v0.27.3) — SQLite + +ALTER TABLE tasks ADD COLUMN webhook_secret TEXT; +ALTER TABLE workflows ADD COLUMN webhook_url TEXT; +ALTER TABLE workflows ADD COLUMN webhook_secret TEXT; diff --git a/server/handlers/tasks.go b/server/handlers/tasks.go index 08196db..d0c60bc 100644 --- a/server/handlers/tasks.go +++ b/server/handlers/tasks.go @@ -7,8 +7,9 @@ import ( "github.com/gin-gonic/gin" "git.gobha.me/xcaliber/chat-switchboard/models" - "git.gobha.me/xcaliber/chat-switchboard/taskutil" "git.gobha.me/xcaliber/chat-switchboard/store" + "git.gobha.me/xcaliber/chat-switchboard/taskutil" + "git.gobha.me/xcaliber/chat-switchboard/webhook" ) // TaskHandler manages task CRUD and manual execution. @@ -129,6 +130,11 @@ func (h *TaskHandler) Create(c *gin.Context) { t.IsActive = true + // v0.27.3: Generate webhook secret if webhook URL is provided + if t.WebhookURL != "" && t.WebhookSecret == "" { + t.WebhookSecret = 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 diff --git a/server/handlers/workflow_instances.go b/server/handlers/workflow_instances.go index 2abc0c9..a3bc4e6 100644 --- a/server/handlers/workflow_instances.go +++ b/server/handlers/workflow_instances.go @@ -355,103 +355,8 @@ func (h *WorkflowInstanceHandler) emitWorkflowEvent(label, channelID string, dat // 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) { - if h.stores.Workflows == nil { - return - } - wf, err := h.stores.Workflows.GetByID(ctx, workflowID) - if err != nil || wf == nil || len(wf.OnComplete) == 0 || string(wf.OnComplete) == "null" { - return - } - - var chain struct { - Action string `json:"action"` - TargetSlug string `json:"target_slug"` - DataMap map[string]string `json:"data_map"` // source_key → target_key - } - if err := json.Unmarshal(wf.OnComplete, &chain); err != nil || chain.Action != "start_workflow" || chain.TargetSlug == "" { - return - } - - // Look up target workflow - target, err := h.stores.Workflows.GetBySlug(ctx, wf.TeamID, chain.TargetSlug) - if err != nil || target == nil || !target.IsActive { - log.Printf("[workflow] on_complete: target workflow '%s' not found or inactive", chain.TargetSlug) - return - } - - // Map stage_data from completed workflow to initial stage_data for target - var sourceData map[string]any - if err := json.Unmarshal([]byte(mergedData), &sourceData); err != nil { - sourceData = map[string]any{} - } - targetData := map[string]any{} - if len(chain.DataMap) > 0 { - for srcKey, tgtKey := range chain.DataMap { - if v, ok := sourceData[srcKey]; ok { - targetData[tgtKey] = v - } - } - } else { - // No mapping = pass all data through - targetData = sourceData - } - - // Get the channel owner to start the chained workflow as - var ownerID string - _ = database.DB.QueryRowContext(ctx, database.Q(` - SELECT user_id FROM channels WHERE id = $1 - `), channelID).Scan(&ownerID) - if ownerID == "" { - return - } - - // Create a new workflow channel for the target - ver, err := h.stores.Workflows.GetLatestVersion(ctx, target.ID) - if err != nil { - log.Printf("[workflow] on_complete: target '%s' has no published version", chain.TargetSlug) - return - } - stages, err := h.stores.Workflows.ListStages(ctx, target.ID) - if err != nil || len(stages) == 0 { - return - } - - ch := &models.Channel{ - UserID: ownerID, - Title: target.Name + " (chained)", - Description: target.Description, - Type: "workflow", - TeamID: target.TeamID, - } - if err := h.stores.Channels.Create(ctx, ch); err != nil { - log.Printf("[workflow] on_complete: failed to create chained channel: %v", err) - return - } - - initialData, _ := json.Marshal(targetData) - _, err = database.DB.ExecContext(ctx, database.Q(` - UPDATE channels - SET workflow_id = $1, workflow_version = $2, current_stage = 0, - stage_data = $3, workflow_status = 'active', - last_activity_at = $4, ai_mode = 'auto' - WHERE id = $5 - `), target.ID, ver.VersionNumber, string(initialData), time.Now().UTC(), ch.ID) - if err != nil { - log.Printf("[workflow] on_complete: failed to set workflow columns: %v", err) - return - } - - _ = h.stores.Channels.AddParticipant(ctx, &models.ChannelParticipant{ - ChannelID: ch.ID, ParticipantType: "user", ParticipantID: ownerID, Role: "owner", - }) - if stages[0].PersonaID != nil { - _ = h.stores.Channels.AddParticipant(ctx, &models.ChannelParticipant{ - ChannelID: ch.ID, ParticipantType: "persona", ParticipantID: *stages[0].PersonaID, Role: "member", - }) - } - - log.Printf("[workflow] on_complete: chained '%s' → '%s' (channel %s → %s)", - wf.Slug, target.Slug, channelID, ch.ID) + // 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 diff --git a/server/main.go b/server/main.go index 8bf9763..c948862 100644 --- a/server/main.go +++ b/server/main.go @@ -359,6 +359,9 @@ func main() { // Workflow tools (v0.26.4 — AI-triggered stage advancement) tools.RegisterWorkflowTools(stores) + + // v0.27.3: task_create tool (AI spawns sub-tasks) + tools.RegisterTaskTools(stores) memExtractor := memory.NewExtractor(stores, roleResolver, kbEmbedder) memScanner := memory.NewScanner(memExtractor, stores, memory.ScannerConfig{}) memScanner.Start() diff --git a/server/models/models_task.go b/server/models/models_task.go index f5ff936..48884f0 100644 --- a/server/models/models_task.go +++ b/server/models/models_task.go @@ -35,7 +35,8 @@ type Task struct { MaxWallClock int `json:"max_wall_clock" db:"max_wall_clock"` // seconds OutputMode string `json:"output_mode" db:"output_mode"` // channel | note | webhook OutputChannelID *string `json:"output_channel_id,omitempty" db:"output_channel_id"` - WebhookURL string `json:"webhook_url" db:"webhook_url"` + WebhookURL string `json:"webhook_url" db:"webhook_url"` + WebhookSecret string `json:"webhook_secret,omitempty" db:"webhook_secret"` // Provider routing ProviderConfigID *string `json:"provider_config_id,omitempty" db:"provider_config_id"` diff --git a/server/models/workflow.go b/server/models/workflow.go index 72ba4de..eb85f1f 100644 --- a/server/models/workflow.go +++ b/server/models/workflow.go @@ -18,9 +18,11 @@ type Workflow struct { EntryMode string `json:"entry_mode"` // public_link | team_only IsActive bool `json:"is_active"` Version int `json:"version"` - OnComplete json.RawMessage `json:"on_complete,omitempty"` - Retention json.RawMessage `json:"retention"` - CreatedBy string `json:"created_by"` + OnComplete json.RawMessage `json:"on_complete,omitempty"` + Retention json.RawMessage `json:"retention"` + WebhookURL string `json:"webhook_url,omitempty"` + WebhookSecret string `json:"webhook_secret,omitempty"` + CreatedBy string `json:"created_by"` CreatedAt time.Time `json:"created_at"` UpdatedAt time.Time `json:"updated_at"` @@ -37,6 +39,7 @@ type WorkflowPatch struct { IsActive *bool `json:"is_active,omitempty"` OnComplete *json.RawMessage `json:"on_complete,omitempty"` Retention *json.RawMessage `json:"retention,omitempty"` + WebhookURL *string `json:"webhook_url,omitempty"` } // ── Workflow Stage ────────────────────────── diff --git a/server/scheduler/executor.go b/server/scheduler/executor.go index d4f8d8c..56e3b87 100644 --- a/server/scheduler/executor.go +++ b/server/scheduler/executor.go @@ -22,6 +22,7 @@ import ( "git.gobha.me/xcaliber/chat-switchboard/providers" "git.gobha.me/xcaliber/chat-switchboard/store" "git.gobha.me/xcaliber/chat-switchboard/tools" + "git.gobha.me/xcaliber/chat-switchboard/webhook" ) // Executor runs task completions headlessly (no HTTP client). @@ -200,6 +201,19 @@ func (e *Executor) Execute(ctx context.Context, task models.Task, run *models.Ta // ── 9. Owner notification ────────────────── e.notifyOwner(ctx, task, status, errMsg) + + // ── 10. Webhook delivery (v0.27.3) ───────── + if task.WebhookURL != "" { + go webhook.Deliver(task.WebhookURL, task.WebhookSecret, webhook.Payload{ + TaskID: task.ID, + TaskName: task.Name, + ChannelID: channelID, + Status: status, + CompletedAt: time.Now().UTC(), + Output: result.Content, + Error: errMsg, + }) + } } // failRun marks a run as failed before completion was attempted. @@ -207,6 +221,15 @@ func (e *Executor) failRun(ctx context.Context, task models.Task, run *models.Ta log.Printf("[executor] Task %s (%s) pre-execution failure: %s", task.ID, task.Name, errMsg) _ = e.stores.Tasks.UpdateRun(ctx, run.ID, "failed", 0, 0, 0, errMsg) e.notifyOwner(ctx, task, "failed", errMsg) + if task.WebhookURL != "" { + go webhook.Deliver(task.WebhookURL, task.WebhookSecret, webhook.Payload{ + TaskID: task.ID, + TaskName: task.Name, + Status: "failed", + CompletedAt: time.Now().UTC(), + Error: errMsg, + }) + } } // notifyOwner sends a notification based on task outcome and preferences. diff --git a/server/tools/task_create.go b/server/tools/task_create.go new file mode 100644 index 0000000..a2f9e08 --- /dev/null +++ b/server/tools/task_create.go @@ -0,0 +1,199 @@ +package tools + +import ( + "context" + "encoding/json" + "fmt" + "log" + "time" + + "git.gobha.me/xcaliber/chat-switchboard/database" + "git.gobha.me/xcaliber/chat-switchboard/models" + "git.gobha.me/xcaliber/chat-switchboard/store" + "git.gobha.me/xcaliber/chat-switchboard/taskutil" + "git.gobha.me/xcaliber/chat-switchboard/webhook" +) + +// ── Late Registration ──────────────────────── + +// RegisterTaskTools registers the task_create tool. +// Called from main.go after stores are initialized. +func RegisterTaskTools(stores store.Stores) { + if stores.Tasks == nil { + log.Println("⚠ task tools: TaskStore not available, skipping registration") + return + } + Register(&taskCreateTool{stores: stores}) + log.Println("✅ task tools registered (task_create)") +} + +// ═══════════════════════════════════════════ +// task_create +// ═══════════════════════════════════════════ +// +// AI-invocable tool that spawns sub-tasks. Only available inside +// workflow or team channels — prevents runaway task creation in +// personal chats. Tasks cannot create tasks (depth limit: service +// channels are blocked). + +type taskCreateTool struct { + BaseTool + stores store.Stores +} + +// Availability: workflow or team context only — not personal chats, +// not service channels (prevents recursive task spawning). +func (t *taskCreateTool) Availability() Require { + return func(tc ToolContext) bool { + // Block in service channels (task output) — prevents recursion + if tc.ChannelType == "service" { + return false + } + // Require workflow or team context + return tc.WorkflowID != "" || tc.TeamID != "" + } +} + +func (t *taskCreateTool) Definition() ToolDef { + return ToolDef{ + Name: "task_create", + DisplayName: "Create Task", + Category: "automation", + Description: "Create a new scheduled or one-shot task. Use this to spawn " + + "background work that runs independently — for example, a follow-up " + + "analysis, a periodic check, or a notification workflow. The task runs " + + "in its own service channel with its own budget.", + Parameters: JSONSchema(map[string]interface{}{ + "name": Prop("string", "Short descriptive name for the task"), + "prompt": Prop("string", "The user prompt the task will execute"), + "schedule": Prop("string", + "Cron expression (e.g. '0 8 * * *' for daily at 8am) or 'once' for immediate one-shot execution"), + "persona": Prop("string", + "Persona handle to use for execution (optional — uses default if omitted)"), + "model": Prop("string", + "Model ID to use (optional — uses persona or config default)"), + "max_tokens": map[string]interface{}{ + "type": "integer", + "description": "Maximum tokens for the task (optional — uses global default)", + }, + "system_prompt": Prop("string", + "Optional system prompt override for the task"), + }, []string{"name", "prompt", "schedule"}), + } +} + +func (t *taskCreateTool) Execute(ctx context.Context, execCtx ExecutionContext, argsJSON string) (string, error) { + var args struct { + Name string `json:"name"` + Prompt string `json:"prompt"` + Schedule string `json:"schedule"` + Persona string `json:"persona"` + Model string `json:"model"` + MaxTokens int `json:"max_tokens"` + SystemPrompt string `json:"system_prompt"` + } + if err := json.Unmarshal([]byte(argsJSON), &args); err != nil { + return "", fmt.Errorf("invalid arguments: %w", err) + } + + if args.Name == "" || args.Prompt == "" || args.Schedule == "" { + return "", fmt.Errorf("name, prompt, and schedule are required") + } + + // Validate cron + if err := taskutil.ValidateCron(args.Schedule); err != nil { + return "", fmt.Errorf("invalid schedule: %w", err) + } + + // Depth guard: verify we're not inside a service channel (task output) + if execCtx.ChannelID != "" { + var channelType string + _ = database.DB.QueryRowContext(ctx, database.Q( + `SELECT type FROM channels WHERE id = $1`, + ), execCtx.ChannelID).Scan(&channelType) + if channelType == "service" { + return "", fmt.Errorf("tasks cannot create tasks (depth limit)") + } + } + + // Resolve persona ID from handle + var personaID *string + if args.Persona != "" { + var pID string + err := database.DB.QueryRowContext(ctx, database.Q( + `SELECT id FROM personas WHERE LOWER(handle) = LOWER($1) AND is_active = true`, + ), args.Persona).Scan(&pID) + if err == nil && pID != "" { + personaID = &pID + } + } + + // Inherit scope from channel context + scope := "personal" + var teamID *string + if execCtx.TeamID != "" { + scope = "team" + teamID = &execCtx.TeamID + } + + // Build task + task := &models.Task{ + OwnerID: execCtx.UserID, + TeamID: teamID, + Name: args.Name, + Scope: scope, + TaskType: "prompt", + PersonaID: personaID, + ModelID: args.Model, + SystemPrompt: args.SystemPrompt, + UserPrompt: args.Prompt, + Schedule: args.Schedule, + Timezone: "UTC", + IsActive: true, + OutputMode: "channel", + NotifyOnFailure: true, + } + + if args.MaxTokens > 0 { + task.MaxTokens = args.MaxTokens + } + + // Apply global defaults for zero-value budgets + cfg := taskutil.LoadTaskConfig(ctx, t.stores.GlobalConfig) + cfg.ApplyDefaults(task) + + // Generate webhook secret in case webhook is added later + task.WebhookSecret = webhook.GenerateSecret() + + // Compute next_run_at + if args.Schedule == "once" { + now := time.Now().UTC() + task.NextRunAt = &now + } else { + task.NextRunAt = taskutil.NextRunFromSchedule(args.Schedule, "UTC") + } + + if err := t.stores.Tasks.Create(ctx, task); err != nil { + return "", fmt.Errorf("failed to create task: %w", err) + } + + log.Printf("[task_create] AI spawned task %s (%s) schedule=%s scope=%s", + task.ID, task.Name, args.Schedule, scope) + + result, _ := json.Marshal(map[string]interface{}{ + "task_id": task.ID, + "name": task.Name, + "schedule": args.Schedule, + "scope": scope, + "next_run_at": task.NextRunAt, + "message": fmt.Sprintf("Task '%s' created successfully. It will run %s.", task.Name, scheduleDesc(args.Schedule)), + }) + return string(result), nil +} + +func scheduleDesc(schedule string) string { + if schedule == "once" { + return "immediately (one-shot)" + } + return "on schedule: " + schedule +} diff --git a/server/tools/workflow.go b/server/tools/workflow.go index f1f839c..c929c2d 100644 --- a/server/tools/workflow.go +++ b/server/tools/workflow.go @@ -10,6 +10,7 @@ import ( "git.gobha.me/xcaliber/chat-switchboard/database" "git.gobha.me/xcaliber/chat-switchboard/models" "git.gobha.me/xcaliber/chat-switchboard/store" + "git.gobha.me/xcaliber/chat-switchboard/webhook" ) // ── Late Registration ──────────────────────── @@ -120,6 +121,9 @@ func (t *workflowAdvanceTool) Execute(ctx context.Context, execCtx ExecutionCont // Create note with collected data CreateWorkflowStageNote(ctx, t.stores, channelID, currentStage, args.Data, args.Summary) + // v0.27.3: Fire on_complete chaining + webhook (was missing from tool path) + go TriggerWorkflowOnComplete(ctx, t.stores, workflowID, channelID, mergedData) + result, _ := json.Marshal(map[string]interface{}{ "status": "completed", "current_stage": nextStage, @@ -245,3 +249,128 @@ func CreateWorkflowAssignment(ctx context.Context, channelID string, stage int, } return id } + +// ── Workflow Completion Triggers (v0.27.3) ── + +// TriggerWorkflowOnComplete checks if the workflow has an on_complete chain +// config and fires the target workflow. Also delivers webhook if configured. +// Shared by both the HTTP handler (Advance endpoint) and the workflow_advance tool. +func TriggerWorkflowOnComplete(ctx context.Context, stores store.Stores, workflowID, channelID, mergedData string) { + if stores.Workflows == nil { + return + } + wf, err := stores.Workflows.GetByID(ctx, workflowID) + if err != nil || wf == nil { + return + } + + // v0.27.3: Webhook delivery on workflow completion + if wf.WebhookURL != "" { + var stageData any + _ = json.Unmarshal([]byte(mergedData), &stageData) + + go deliverWorkflowWebhook(wf.WebhookURL, wf.WebhookSecret, workflowID, channelID, stageData) + } + + // on_complete chaining + if len(wf.OnComplete) == 0 || string(wf.OnComplete) == "null" { + return + } + + var chain struct { + Action string `json:"action"` + TargetSlug string `json:"target_slug"` + DataMap map[string]string `json:"data_map"` + } + if err := json.Unmarshal(wf.OnComplete, &chain); err != nil || chain.Action != "start_workflow" || chain.TargetSlug == "" { + return + } + + target, err := stores.Workflows.GetBySlug(ctx, wf.TeamID, chain.TargetSlug) + if err != nil || target == nil || !target.IsActive { + log.Printf("[workflow] on_complete: target workflow '%s' not found or inactive", chain.TargetSlug) + return + } + + var sourceData map[string]any + if err := json.Unmarshal([]byte(mergedData), &sourceData); err != nil { + sourceData = map[string]any{} + } + targetData := map[string]any{} + if len(chain.DataMap) > 0 { + for srcKey, tgtKey := range chain.DataMap { + if v, ok := sourceData[srcKey]; ok { + targetData[tgtKey] = v + } + } + } else { + targetData = sourceData + } + + var ownerID string + _ = database.DB.QueryRowContext(ctx, database.Q( + `SELECT user_id FROM channels WHERE id = $1`, + ), channelID).Scan(&ownerID) + if ownerID == "" { + return + } + + ver, err := stores.Workflows.GetLatestVersion(ctx, target.ID) + if err != nil { + log.Printf("[workflow] on_complete: target '%s' has no published version", chain.TargetSlug) + return + } + stages, err := stores.Workflows.ListStages(ctx, target.ID) + if err != nil || len(stages) == 0 { + return + } + + ch := &models.Channel{ + UserID: ownerID, + Title: target.Name + " (chained)", + Description: target.Description, + Type: "workflow", + TeamID: target.TeamID, + } + if err := stores.Channels.Create(ctx, ch); err != nil { + log.Printf("[workflow] on_complete: failed to create chained channel: %v", err) + return + } + + initialData, _ := json.Marshal(targetData) + _, err = database.DB.ExecContext(ctx, database.Q(` + UPDATE channels + SET workflow_id = $1, workflow_version = $2, current_stage = 0, + stage_data = $3, workflow_status = 'active', + last_activity_at = $4, ai_mode = 'auto' + WHERE id = $5 + `), target.ID, ver.VersionNumber, string(initialData), time.Now().UTC(), ch.ID) + if err != nil { + log.Printf("[workflow] on_complete: failed to set workflow columns: %v", err) + return + } + + // Bind first stage persona + if len(stages) > 0 && stages[0].PersonaID != nil { + _, _ = database.DB.ExecContext(ctx, database.Q(` + INSERT INTO channel_participants (channel_id, participant_type, participant_id) + VALUES ($1, 'persona', $2) + ON CONFLICT DO NOTHING + `), ch.ID, *stages[0].PersonaID) + } + + log.Printf("[workflow] on_complete: started chained workflow %s → %s (channel %s)", + workflowID, target.ID, ch.ID) +} + +// deliverWorkflowWebhook fires a webhook for workflow completion. +func deliverWorkflowWebhook(url, secret, workflowID, channelID string, stageData any) { + webhook.Deliver(url, secret, webhook.Payload{ + WorkflowID: workflowID, + ChannelID: channelID, + Status: "completed", + CompletedAt: time.Now().UTC(), + StageData: stageData, + }) +} + diff --git a/server/webhook/webhook.go b/server/webhook/webhook.go new file mode 100644 index 0000000..f285c70 --- /dev/null +++ b/server/webhook/webhook.go @@ -0,0 +1,111 @@ +// Package webhook delivers HTTP POST notifications on task/workflow completion. +// +// v0.27.3: Retry (3 attempts, exponential backoff), HMAC-SHA256 signature, +// 10s timeout per attempt. +package webhook + +import ( + "bytes" + "crypto/hmac" + "crypto/rand" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "io" + "log" + "net/http" + "time" +) + +const ( + maxAttempts = 3 + timeout = 10 * time.Second + signatureHeader = "X-Switchboard-Signature" +) + +// Payload is the JSON body sent to webhook endpoints. +type Payload struct { + TaskID string `json:"task_id,omitempty"` + TaskName string `json:"task_name,omitempty"` + WorkflowID string `json:"workflow_id,omitempty"` + ChannelID string `json:"channel_id,omitempty"` + Status string `json:"status"` + CompletedAt time.Time `json:"completed_at"` + Output string `json:"output,omitempty"` // last assistant message or summary + StageData any `json:"stage_data,omitempty"` // workflow stage data (if applicable) + Error string `json:"error,omitempty"` +} + +// Deliver sends a webhook payload to the given URL with HMAC signing. +// Retries up to 3 times with exponential backoff (1s, 5s, 25s). +// Runs synchronously — call in a goroutine for non-blocking delivery. +func Deliver(url, secret string, payload Payload) error { + if url == "" { + return nil + } + + body, err := json.Marshal(payload) + if err != nil { + return fmt.Errorf("webhook marshal: %w", err) + } + + // Compute HMAC signature + signature := "" + if secret != "" { + mac := hmac.New(sha256.New, []byte(secret)) + mac.Write(body) + signature = hex.EncodeToString(mac.Sum(nil)) + } + + client := &http.Client{Timeout: timeout} + backoff := time.Second // 1s, 5s, 25s + + var lastErr error + for attempt := 1; attempt <= maxAttempts; attempt++ { + req, err := http.NewRequest("POST", url, bytes.NewReader(body)) + if err != nil { + return fmt.Errorf("webhook request build: %w", err) + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("User-Agent", "Switchboard-Webhook/1.0") + if signature != "" { + req.Header.Set(signatureHeader, signature) + } + + resp, err := client.Do(req) + if err != nil { + lastErr = err + log.Printf("[webhook] Attempt %d/%d failed for %s: %v", attempt, maxAttempts, url, err) + } else { + io.Copy(io.Discard, resp.Body) + resp.Body.Close() + + if resp.StatusCode >= 200 && resp.StatusCode < 300 { + if attempt > 1 { + log.Printf("[webhook] Delivered to %s on attempt %d", url, attempt) + } + return nil + } + lastErr = fmt.Errorf("HTTP %d", resp.StatusCode) + log.Printf("[webhook] Attempt %d/%d returned %d for %s", attempt, maxAttempts, resp.StatusCode, url) + } + + if attempt < maxAttempts { + time.Sleep(backoff) + backoff *= 5 + } + } + + return fmt.Errorf("webhook delivery failed after %d attempts: %w", maxAttempts, lastErr) +} + +// GenerateSecret creates a random 32-byte hex-encoded webhook secret. +func GenerateSecret() string { + b := make([]byte, 32) + if _, err := rand.Read(b); err != nil { + // Fallback to timestamp-based (extremely unlikely) + return fmt.Sprintf("%x", time.Now().UnixNano()) + } + return hex.EncodeToString(b) +}