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/webhook" ) // ── Late Registration ──────────────────────── // RegisterWorkflowTools registers the workflow_advance tool. // Called from main.go after stores are initialized. func RegisterWorkflowTools(stores store.Stores) { if stores.Workflows == nil { log.Println("⚠ workflow tools: WorkflowStore not available, skipping registration") return } Register(&workflowAdvanceTool{stores: stores}) log.Println("✅ workflow tools registered (workflow_advance)") } // ═══════════════════════════════════════════ // workflow_advance // ═══════════════════════════════════════════ // // Called by the stage persona when it has collected all required form // data from the visitor. Triggers a stage transition — same effect as // the human-triggered POST /channels/:id/workflow/advance endpoint. type workflowAdvanceTool struct { BaseTool stores store.Stores } // Override availability: only available inside workflow channels. func (t *workflowAdvanceTool) Availability() Require { return RequireWorkflow } func (t *workflowAdvanceTool) Definition() ToolDef { return ToolDef{ Name: "workflow_advance", DisplayName: "Advance Workflow", Category: "workflow", Description: "Advance the workflow to the next stage. Call this when you have " + "collected all required information from the visitor. Pass the collected " + "data as a JSON object — it will be saved and made available to the next stage.", Parameters: JSONSchema(map[string]interface{}{ "data": map[string]interface{}{ "type": "object", "description": "Collected form data as key-value pairs. Must include all fields defined in the stage's form template.", }, "summary": Prop("string", "Brief summary of what was collected (1-2 sentences, shown in stage notes)"), }, []string{"data"}), } } func (t *workflowAdvanceTool) Execute(ctx context.Context, execCtx ExecutionContext, argsJSON string) (string, error) { var args struct { Data json.RawMessage `json:"data"` Summary string `json:"summary"` } if err := json.Unmarshal([]byte(argsJSON), &args); err != nil { return "", fmt.Errorf("invalid arguments: %w", err) } if len(args.Data) == 0 || string(args.Data) == "null" { return "", fmt.Errorf("data is required") } channelID := execCtx.ChannelID if channelID == "" { return "", fmt.Errorf("no channel context") } // Read current workflow state var workflowID string var currentStage int var status string var wfID *string err := database.DB.QueryRowContext(ctx, database.Q(` SELECT workflow_id, COALESCE(current_stage, 0), COALESCE(workflow_status, 'active') FROM channels WHERE id = $1 AND type = 'workflow' `), channelID).Scan(&wfID, ¤tStage, &status) if err != nil || wfID == nil { return "", fmt.Errorf("not a workflow channel") } workflowID = *wfID if status != "active" { return "", fmt.Errorf("workflow is %s, cannot advance", status) } // Load stages stages, err := t.stores.Workflows.ListStages(ctx, workflowID) if err != nil { return "", fmt.Errorf("failed to load stages: %w", err) } // Merge collected data into stage_data mergedData := MergeWorkflowStageData(ctx, channelID, args.Data) nextStage := currentStage + 1 if nextStage >= len(stages) { // Workflow complete _, err = database.DB.ExecContext(ctx, database.Q(` UPDATE channels SET current_stage = $1, workflow_status = 'completed', stage_data = $2, last_activity_at = $3, ai_mode = 'off' WHERE id = $4 `), nextStage, mergedData, time.Now().UTC(), channelID) if err != nil { return "", fmt.Errorf("failed to complete workflow: %w", err) } // 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, "message": "Workflow completed successfully.", }) return string(result), nil } // Advance to next stage _, err = database.DB.ExecContext(ctx, database.Q(` UPDATE channels SET current_stage = $1, stage_data = $2, last_activity_at = $3 WHERE id = $4 `), nextStage, mergedData, time.Now().UTC(), channelID) if err != nil { return "", fmt.Errorf("failed to advance: %w", err) } // Create note with collected data CreateWorkflowStageNote(ctx, t.stores, channelID, currentStage, args.Data, args.Summary) // Create assignment if next stage has an assignment team nextStageDef := stages[nextStage] if nextStageDef.AssignmentTeamID != nil { CreateWorkflowAssignment(ctx, channelID, nextStage, *nextStageDef.AssignmentTeamID) } result, _ := json.Marshal(map[string]interface{}{ "status": "active", "current_stage": nextStage, "stage_name": nextStageDef.Name, "message": fmt.Sprintf("Advanced to stage %d: %s", nextStage, nextStageDef.Name), }) return string(result), nil } // ── Shared helpers (used by both tool and handler) ─ // MergeWorkflowStageData reads current stage_data, merges new data in Go, returns JSON string. func MergeWorkflowStageData(ctx context.Context, channelID string, newData json.RawMessage) string { var existing []byte _ = database.DB.QueryRowContext(ctx, database.Q(` SELECT COALESCE(stage_data, '{}') FROM channels WHERE id = $1 `), channelID).Scan(&existing) if len(newData) == 0 || string(newData) == "null" { if len(existing) == 0 { return "{}" } return string(existing) } var base map[string]interface{} if err := json.Unmarshal(existing, &base); err != nil { base = map[string]interface{}{} } var incoming map[string]interface{} if err := json.Unmarshal(newData, &incoming); err == nil { for k, v := range incoming { base[k] = v } } merged, _ := json.Marshal(base) return string(merged) } // CreateWorkflowStageNote persists collected form data as a channel-scoped note. func CreateWorkflowStageNote(ctx context.Context, stores store.Stores, channelID string, stageOrdinal int, data json.RawMessage, summary string) { if stores.Notes == nil || len(data) == 0 || string(data) == "null" { return } title := fmt.Sprintf("Stage %d collected data", stageOrdinal) if summary != "" { title = fmt.Sprintf("Stage %d: %s", stageOrdinal, summary) } content := string(data) // Find workflow channel owner for the note's user_id var userID string _ = database.DB.QueryRowContext(ctx, database.Q(` SELECT user_id FROM channels WHERE id = $1 `), channelID).Scan(&userID) if userID == "" { return } note := &models.Note{ Title: title, Content: content, SourceChannelID: &channelID, } note.UserID = userID if err := stores.Notes.Create(ctx, note); err != nil { log.Printf("⚠️ Failed to create stage note: %v", err) } } // CreateWorkflowAssignment creates an unassigned workflow assignment entry. // Returns the assignment ID (for round-robin auto-assignment). func CreateWorkflowAssignment(ctx context.Context, channelID string, stage int, teamID string) string { id := store.NewID() if database.IsSQLite() { _, err := database.DB.ExecContext(ctx, ` INSERT INTO workflow_assignments (id, channel_id, stage, team_id) VALUES (?, ?, ?, ?) `, id, channelID, stage, teamID) if err != nil { log.Printf("⚠️ Failed to create workflow assignment: %v", err) return "" } return id } _, err := database.DB.ExecContext(ctx, ` INSERT INTO workflow_assignments (id, channel_id, stage, team_id) VALUES ($1, $2, $3, $4) `, id, channelID, stage, teamID) if err != nil { log.Printf("⚠️ Failed to create workflow assignment: %v", err) return "" } 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, }) }