poimen-workflows/action/canvas_reasoner.go

package action

import (
	"context"
	"encoding/json"
	"fmt"

	"github.com/rockliang/poimen/workflows/action/llm"
	"github.com/rockliang/poimen/workflows/pkg/db"
	"github.com/rockliang/poimen/workflows/statemachine"
)

// CanvasReasonerInput infers connections between nodes using LLM reasoning
type CanvasReasonerInput struct {
	Nodes []db.WorkflowNode `json:"nodes"`       // Canvas nodes
	Edges []db.WorkflowEdge `json:"edges"`       // Existing edges
	// If true, only suggest new edges; if false, redesign entire canvas
	PreserveExisting bool `json:"preserve_existing,omitempty"`
	AuthToken        string `json:"auth_token,omitempty"` // JWT for LLM calls
}

// 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
	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
func CanvasReasonerActivity(ctx context.Context, in CanvasReasonerInput) (CanvasReasonerOutput, error) {
	logger := newActivityLogger(ctx)

	output := CanvasReasonerOutput{
		SuggestedEdges: []db.WorkflowEdge{},
	}

	if len(in.Nodes) == 0 {
		return output, fmt.Errorf("no nodes provided")
	}

	logger.logf("info", "Analyzing canvas with %d nodes, %d edges", len(in.Nodes), len(in.Edges))

	// Build activity descriptions for LLM context
	nodeDesc := buildNodeDescriptions(in.Nodes)
	edgeDesc := buildEdgeDescriptions(in.Edges)

	// 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 (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 and any type mismatches noted",
  "confidence": 0.85
}`

	userPrompt := fmt.Sprintf(`Canvas Analysis:

Nodes (including inputs/outputs):
%s

Current Edges:
%s

Task: %s

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))

	logger.logf("info", "Calling LLM reasoning (preserve_existing=%v)", in.PreserveExisting)

	// Call LLM
	client, err := llm.NewClient()
	if err != nil {
		return output, fmt.Errorf("failed to create LLM client: %w", err)
	}

	response, err := client.CreateMessage(ctx, llm.MessageInput{
		Model: statemachine.ModelSpec{
			ModelID: "reasoning", // Use reasoning model for complex analysis
		},
		SystemPrompt: systemPrompt,
		Messages: []llm.MessageParam{
			{
				Role:    "user",
				Content: userPrompt,
			},
		},
		AuthToken: in.AuthToken,
	})

	if err != nil {
		return output, fmt.Errorf("LLM reasoning failed: %w", err)
	}

	// Parse LLM response
	var reasonerResp struct {
		Edges      []db.WorkflowEdge `json:"edges"`
		Reasoning  string            `json:"reasoning"`
		Confidence float64           `json:"confidence"`
	}

	if err := json.Unmarshal([]byte(response), &reasonerResp); err != nil {
		logger.logf("warn", "Failed to parse LLM response as JSON: %v", err)
		// Try to extract from response text
		output.Reasoning = response
		output.Confidence = 0.5
		return output, fmt.Errorf("failed to parse LLM response: %w", err)
	}

	// Validate suggested edges
	nodeMap := make(map[string]bool)
	for _, n := range in.Nodes {
		nodeMap[n.ID] = true
	}

	validEdges := []db.WorkflowEdge{}
	for _, edge := range reasonerResp.Edges {
		if !nodeMap[edge.Source] {
			logger.logf("warn", "Suggested edge references unknown source: %s", edge.Source)
			continue
		}
		if !nodeMap[edge.Target] {
			logger.logf("warn", "Suggested edge references unknown target: %s", edge.Target)
			continue
		}
		// Don't suggest self-loops
		if edge.Source == edge.Target {
			logger.logf("warn", "Skipping self-loop: %s", edge.Source)
			continue
		}
		validEdges = append(validEdges, edge)
	}

	output.SuggestedEdges = validEdges
	output.Reasoning = reasonerResp.Reasoning
	output.Confidence = reasonerResp.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 (including schemas)
func buildNodeDescriptions(nodes []db.WorkflowNode) string {
	var desc string
	for i, node := range nodes {
		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))
			}
		}
	}
	return desc
}

// buildEdgeDescriptions creates readable edge descriptions for LLM
func buildEdgeDescriptions(edges []db.WorkflowEdge) string {
	if len(edges) == 0 {
		return "None"
	}
	var desc string
	for i, edge := range edges {
		desc += fmt.Sprintf("%d. %s → %s\n", i+1, edge.Source, edge.Target)
	}
	return desc
}

// getReasoningTask returns task description based on preservation mode
func getReasoningTask(preserveExisting bool) string {
	if preserveExisting {
		return "Keep all existing edges and suggest ONLY NEW edges to improve workflow"
	}
	return "Design optimal workflow by suggesting all connections and noting any redundant edges"
}