feat: add canvas compatibility checking for connection validation

2026-09-05 01:01:01 -07:00
parent 2a3b080e29
commit 9624f0e18d
5 changed files with 704 additions and 19 deletions
@@ -0,0 +1,295 @@
+package action
+
+import (
+	"encoding/json"
+	"fmt"
+	"regexp"
+	"strings"
+
+	"github.com/rockliang/poimen/workflows/pkg/db"
+)
+
+// IncompatibilityWarning explains why two activities can't be connected
+type IncompatibilityWarning struct {
+	Source       string `json:"source"`        // Source node ID
+	Target       string `json:"target"`        // Target node ID
+	Reason       string `json:"reason"`        // Why they can't connect
+	SourceNeeds  string `json:"source_needs"`  // What source would need to output
+	TargetNeeds  string `json:"target_needs"`  // What target requires as input
+	Suggestion   string `json:"suggestion"`    // Suggestion to make it work
+}
+
+// ActivitySchema describes what an activity needs/provides
+type ActivitySchema struct {
+	ActivityType string                 `json:"activity_type"`
+	Inputs       map[string]InputField  `json:"inputs"`
+	Outputs      map[string]OutputField `json:"outputs"`
+}
+
+type InputField struct {
+	Type        string   `json:"type"`
+	Description string   `json:"description"`
+	Required    bool     `json:"required"`
+	Enum        []string `json:"enum,omitempty"`
+}
+
+type OutputField struct {
+	Type        string `json:"type"`
+	Description string `json:"description"`
+}
+
+// getActivitySchema returns schema from knowledge base
+func getActivitySchema(activityType string) (*ActivitySchema, error) {
+	kb := knowledgeBaseData()
+	if kb == nil {
+		return nil, fmt.Errorf("knowledge base not loaded")
+	}
+
+	var activities []map[string]interface{}
+	if err := json.Unmarshal([]byte(kb), &activities); err != nil {
+		// Try to extract activities from full KB structure
+		var fullKB map[string]interface{}
+		if err := json.Unmarshal([]byte(kb), &fullKB); err != nil {
+			return nil, fmt.Errorf("failed to parse knowledge base")
+		}
+		if activitiesRaw, ok := fullKB["activities"]; ok {
+			if b, err := json.Marshal(activitiesRaw); err == nil {
+				if err := json.Unmarshal(b, &activities); err != nil {
+					return nil, fmt.Errorf("failed to extract activities from KB")
+				}
+			}
+		}
+	}
+
+	// Find matching activity
+	for _, act := range activities {
+		if name, ok := act["name"].(string); ok {
+			if toActivityName(activityType) == name {
+				// Convert to ActivitySchema
+				schema := &ActivitySchema{
+					ActivityType: activityType,
+					Inputs:       make(map[string]InputField),
+					Outputs:      make(map[string]OutputField),
+				}
+
+				if inputs, ok := act["inputs"].(map[string]interface{}); ok {
+					for key, val := range inputs {
+						if field, ok := val.(map[string]interface{}); ok {
+							schema.Inputs[key] = parseInputField(field)
+						}
+					}
+				}
+
+				if outputs, ok := act["outputs"].(map[string]interface{}); ok {
+					for key, val := range outputs {
+						if field, ok := val.(map[string]interface{}); ok {
+							schema.Outputs[key] = parseOutputField(field)
+						}
+					}
+				}
+
+				return schema, nil
+			}
+		}
+	}
+
+	return nil, fmt.Errorf("activity %s not found in knowledge base", activityType)
+}
+
+func parseInputField(data map[string]interface{}) InputField {
+	field := InputField{}
+	if t, ok := data["type"].(string); ok {
+		field.Type = t
+	}
+	if d, ok := data["description"].(string); ok {
+		field.Description = d
+	}
+	if r, ok := data["required"].(bool); ok {
+		field.Required = r
+	}
+	return field
+}
+
+func parseOutputField(data map[string]interface{}) OutputField {
+	field := OutputField{}
+	if t, ok := data["type"].(string); ok {
+		field.Type = t
+	}
+	if d, ok := data["description"].(string); ok {
+		field.Description = d
+	}
+	return field
+}
+
+// CheckConnectionCompatibility validates if source can connect to target
+func CheckConnectionCompatibility(sourceNode, targetNode db.WorkflowNode) []IncompatibilityWarning {
+	warnings := []IncompatibilityWarning{}
+
+	sourceSchema, err := getActivitySchema(sourceNode.Type)
+	if err != nil {
+		warnings = append(warnings, IncompatibilityWarning{
+			Source:     sourceNode.ID,
+			Target:     targetNode.ID,
+			Reason:     fmt.Sprintf("Source activity schema not found: %v", err),
+			Suggestion: "Ensure source activity type is registered in knowledge base",
+		})
+		return warnings
+	}
+
+	targetSchema, err := getActivitySchema(targetNode.Type)
+	if err != nil {
+		warnings = append(warnings, IncompatibilityWarning{
+			Source:     sourceNode.ID,
+			Target:     targetNode.ID,
+			Reason:     fmt.Sprintf("Target activity schema not found: %v", err),
+			Suggestion: "Ensure target activity type is registered in knowledge base",
+		})
+		return warnings
+	}
+
+	// Check if source produces outputs that target can consume
+	if len(sourceSchema.Outputs) == 0 {
+		warnings = append(warnings, IncompatibilityWarning{
+			Source:     sourceNode.ID,
+			Target:     targetNode.ID,
+			Reason:     fmt.Sprintf("%s produces no outputs", sourceNode.Type),
+			SourceNeeds: "any output",
+			Suggestion: "Source activity must produce outputs",
+		})
+		return warnings
+	}
+
+	if len(targetSchema.Inputs) == 0 {
+		warnings = append(warnings, IncompatibilityWarning{
+			Source:     sourceNode.ID,
+			Target:     targetNode.ID,
+			Reason:     fmt.Sprintf("%s accepts no inputs", targetNode.Type),
+			TargetNeeds: "no input",
+			Suggestion: "Target activity must accept inputs. Check if it's a terminal activity.",
+		})
+		return warnings
+	}
+
+	// Match outputs to inputs
+	sourceOutputs := getOutputNames(sourceSchema.Outputs)
+	targetInputs := getInputNames(targetSchema.Inputs)
+
+	if len(sourceOutputs) == 0 || len(targetInputs) == 0 {
+		warnings = append(warnings, IncompatibilityWarning{
+			Source:       sourceNode.ID,
+			Target:       targetNode.ID,
+			Reason:       "No compatible output/input fields found",
+			SourceNeeds:  strings.Join(sourceOutputs, ", "),
+			TargetNeeds:  strings.Join(targetInputs, ", "),
+			Suggestion:   "Use LLM transformation to map outputs to inputs",
+		})
+	}
+
+	return warnings
+}
+
+// CheckCanvasConnectivity analyzes all suggested edges for compatibility
+func CheckCanvasConnectivity(nodes []db.WorkflowNode, suggestedEdges []db.WorkflowEdge) []IncompatibilityWarning {
+	warnings := []IncompatibilityWarning{}
+	nodeMap := make(map[string]db.WorkflowNode)
+	for _, n := range nodes {
+		nodeMap[n.ID] = n
+	}
+
+	for _, edge := range suggestedEdges {
+		sourceNode, ok := nodeMap[edge.Source]
+		if !ok {
+			continue
+		}
+		targetNode, ok := nodeMap[edge.Target]
+		if !ok {
+			continue
+		}
+
+		edgeWarnings := CheckConnectionCompatibility(sourceNode, targetNode)
+		warnings = append(warnings, edgeWarnings...)
+	}
+
+	return warnings
+}
+
+// IdentifyDisconnectedNodes finds nodes that can't connect to anything
+func IdentifyDisconnectedNodes(nodes []db.WorkflowNode, suggestedEdges []db.WorkflowEdge) []string {
+	edgeMap := make(map[string]bool)
+	for _, edge := range suggestedEdges {
+		edgeMap[edge.Source] = true
+		edgeMap[edge.Target] = true
+	}
+
+	var disconnected []string
+	for _, node := range nodes {
+		if !edgeMap[node.ID] {
+			disconnected = append(disconnected, node.ID)
+		}
+	}
+	return disconnected
+}
+
+// toActivityName converts canvas type to activity name (e.g., "clone-repo" -> "CloneRepoActivity")
+func toActivityName(canvasType string) string {
+	parts := strings.Split(canvasType, "-")
+	var result string
+	for _, part := range parts {
+		if part != "" {
+			result += strings.ToUpper(part[:1]) + strings.ToLower(part[1:])
+		}
+	}
+	return result + "Activity"
+}
+
+// getOutputNames extracts output field names
+func getOutputNames(outputs map[string]OutputField) []string {
+	var names []string
+	for name := range outputs {
+		names = append(names, name)
+	}
+	return names
+}
+
+// getInputNames extracts input field names (required ones highlighted)
+func getInputNames(inputs map[string]InputField) []string {
+	var names []string
+	for name, field := range inputs {
+		if field.Required {
+			names = append(names, name+"*")
+		} else {
+			names = append(names, name)
+		}
+	}
+	return names
+}
+
+// SuggestDataTransformation proposes how to connect incompatible activities
+func SuggestDataTransformation(sourceNode, targetNode db.WorkflowNode) string {
+	sourceSchema, _ := getActivitySchema(sourceNode.Type)
+	targetSchema, _ := getActivitySchema(targetNode.Type)
+
+	if sourceSchema == nil || targetSchema == nil {
+		return "Cannot analyze compatibility without schemas"
+	}
+
+	sourceOuts := getOutputNames(sourceSchema.Outputs)
+	targetIns := getInputNames(targetSchema.Inputs)
+
+	return fmt.Sprintf(
+		"To connect %s → %s:\n"+
+			"  %s outputs: %s\n"+
+			"  %s needs: %s\n"+
+			"  Solution: Use LLM transformation node to map outputs to inputs",
+		sourceNode.Label, targetNode.Label,
+		sourceNode.Type, strings.Join(sourceOuts, ", "),
+		targetNode.Type, strings.Join(targetIns, ", "),
+	)
+}
+
+// knowledgeBaseData returns raw KB JSON (stub - implement with actual KB loading)
+func knowledgeBaseData() string {
+	// This would load from activity_knowledge_base.json
+	// For now, return empty - real implementation loads from file
+	return ""
+}
@@ -21,10 +21,13 @@ type CanvasReasonerInput struct {

 // CanvasReasonerOutput returns suggested edges and reasoning
 type CanvasReasonerOutput struct {
-	SuggestedEdges []db.WorkflowEdge `json:"suggested_edges"`  // New edges to add
-	RemovedEdges   []db.WorkflowEdge `json:"removed_edges,omitempty"` // Edges to remove (if redesign)
-	Reasoning      string            `json:"reasoning"`        // LLM explanation
-	Confidence     float64           `json:"confidence"`       // 0.0-1.0
+	SuggestedEdges        []db.WorkflowEdge             `json:"suggested_edges"`         // New edges to add
+	RemovedEdges          []db.WorkflowEdge             `json:"removed_edges,omitempty"` // Edges to remove (if redesign)
+	Reasoning             string                        `json:"reasoning"`               // LLM explanation
+	Confidence            float64                       `json:"confidence"`              // 0.0-1.0
+	IncompatibleEdges     []IncompatibilityWarning      `json:"incompatible_edges,omitempty"` // Edges that can't be created
+	DisconnectedNodes     []string                      `json:"disconnected_nodes,omitempty"` // Nodes with no connections
+	UserAlerts            []string                      `json:"user_alerts,omitempty"`        // Human-readable warnings
 }

 // CanvasReasonerActivity uses LLM to infer connections between workflow activities
@@ -45,23 +48,29 @@ func CanvasReasonerActivity(ctx context.Context, in CanvasReasonerInput) (Canvas
 	nodeDesc := buildNodeDescriptions(in.Nodes)
 	edgeDesc := buildEdgeDescriptions(in.Edges)

-	// Create prompt for LLM reasoning
+	// Create prompt for LLM reasoning with compatibility guidance
 	systemPrompt := `You are a workflow automation expert. Analyze the following activities and suggest logical connections (edges) between them based on:
-1. Activity input/output compatibility
-2. Logical execution order
-3. Data flow requirements
+1. Activity input/output compatibility (CRITICAL - only connect if outputs match inputs)
+2. Logical execution order and data flow
+3. Required dependencies
 4. Common workflow patterns

+IMPORTANT: Only suggest edges where:
+- Source activity has outputs (check "outputs" fields)
+- Target activity has inputs (check "inputs" fields)
+- Data types are compatible (string→string, object→object, etc)
+- Connection makes semantic sense (don't connect a notifier to an analyzer)
+
 Respond with JSON containing:
 {
  "edges": [{"source": "node-1", "target": "node-2"}, ...],
-  "reasoning": "explanation of why these connections make sense",
+  "reasoning": "explanation of why these connections make sense and any type mismatches noted",
  "confidence": 0.85
 }`

 	userPrompt := fmt.Sprintf(`Canvas Analysis:

-Nodes:
+Nodes (including inputs/outputs):
 %s

 Current Edges:
@@ -69,8 +78,12 @@ Current Edges:

 Task: %s

-Preserve existing edges and suggest only NEW edges to add.
-If any existing edges don't make sense, note them but keep them unless explicitly wrong.
+KEY RULES:
+- Preserve existing edges and suggest only NEW edges to add
+- SKIP any connections where input/output types don't match
+- If an activity has no outputs, it cannot be a source
+- If an activity has no inputs, it cannot be a target
+- Note any activities that are hard to connect (terminal activities, generators, etc)

 Return ONLY valid JSON, no markdown code blocks.`, nodeDesc, edgeDesc, getReasoningTask(in.PreserveExisting))

@@ -143,20 +156,75 @@ Return ONLY valid JSON, no markdown code blocks.`, nodeDesc, edgeDesc, getReason
 	output.Reasoning = reasonerResp.Reasoning
 	output.Confidence = reasonerResp.Confidence

-	logger.logf("info", "LLM suggested %d edges with confidence %.2f", len(validEdges), output.Confidence)
+	// Check compatibility of suggested edges
+	incompatibilities := CheckCanvasConnectivity(in.Nodes, validEdges)
+	if len(incompatibilities) > 0 {
+		output.IncompatibleEdges = incompatibilities
+		logger.logf("warn", "Found %d incompatible edge connections", len(incompatibilities))
+
+		// Generate user-friendly alerts
+		for i, incompat := range incompatibilities {
+			if i < 5 { // Limit to 5 alerts to avoid spam
+				alert := fmt.Sprintf(
+					"⚠️  %s → %s: %s. %s",
+					incompat.Source, incompat.Target, incompat.Reason, incompat.Suggestion,
+				)
+				output.UserAlerts = append(output.UserAlerts, alert)
+			}
+		}
+	}
+
+	// Identify disconnected nodes
+	disconnected := IdentifyDisconnectedNodes(in.Nodes, validEdges)
+	if len(disconnected) > 0 {
+		output.DisconnectedNodes = disconnected
+		logger.logf("warn", "Found %d disconnected nodes", len(disconnected))
+
+		for _, nodeID := range disconnected {
+			var label string
+			for _, node := range in.Nodes {
+				if node.ID == nodeID {
+					label = node.Label
+					break
+				}
+			}
+			alert := fmt.Sprintf(
+				"🔌 Node '%s' has no connections. Consider adding edges or removing it.",
+				label,
+			)
+			output.UserAlerts = append(output.UserAlerts, alert)
+		}
+	}
+
+	logger.logf("info", "LLM suggested %d edges with confidence %.2f | %d incompatibilities | %d disconnected",
+		len(validEdges), output.Confidence, len(incompatibilities), len(disconnected))

 	return output, nil
 }

-// buildNodeDescriptions creates readable node descriptions for LLM
+// buildNodeDescriptions creates readable node descriptions for LLM (including schemas)
 func buildNodeDescriptions(nodes []db.WorkflowNode) string {
 	var desc string
 	for i, node := range nodes {
-		desc += fmt.Sprintf("%d. %s (type: %s)\n", i+1, node.ID, node.Type)
-		desc += fmt.Sprintf("   Label: %s\n", node.Label)
+		desc += fmt.Sprintf("%d. [%s] %s (type: %s)\n", i+1, node.ID, node.Label, node.Type)
+
+		// Add input/output schema info
+		if schema, err := getActivitySchema(node.Type); err == nil {
+			if len(schema.Inputs) > 0 {
+				desc += fmt.Sprintf("   INPUTS: %v\n", getInputNames(schema.Inputs))
+			} else {
+				desc += fmt.Sprintf("   INPUTS: none (generator/trigger)\n")
+			}
+			if len(schema.Outputs) > 0 {
+				desc += fmt.Sprintf("   OUTPUTS: %v\n", getOutputNames(schema.Outputs))
+			} else {
+				desc += fmt.Sprintf("   OUTPUTS: none (terminal/sink)\n")
+			}
+		}
+
 		if node.Data != nil {
 			if b, err := json.MarshalIndent(node.Data, "   ", "  "); err == nil {
-				desc += fmt.Sprintf("   Config: %s\n", string(b))
+				desc += fmt.Sprintf("   CONFIG: %s\n", string(b))
 			}
 		}
 	}