feat: wire GraphRAG API handlers, activities, and database layer

action/canvas_compatibility.go

@@ -293,3 +293,62 @@ func knowledgeBaseData() string {
 	// For now, return empty - real implementation loads from file
 	return ""
 }
+
+// CanvasCompatibilityInput for Temporal activity
+type CanvasCompatibilityInput struct {
+	Nodes []db.WorkflowNode `json:"nodes"`
+	Edges []db.WorkflowEdge `json:"edges"`
+}
+
+// CanvasCompatibilityOutput returns validation results
+type CanvasCompatibilityOutput struct {
+	IsValid           bool                      `json:"is_valid"`
+	Incompatibilities []IncompatibilityWarning  `json:"incompatibilities"`
+	DisconnectedNodes []string                  `json:"disconnected_nodes"`
+	Warnings          []string                  `json:"warnings"`
+}
+
+// CanvasCompatibilityActivity validates workflow canvas for type mismatches and isolation
+func CanvasCompatibilityActivity(ctx interface{}, input CanvasCompatibilityInput) (CanvasCompatibilityOutput, error) {
+	output := CanvasCompatibilityOutput{
+		IsValid:           true,
+		Incompatibilities: []IncompatibilityWarning{},
+		DisconnectedNodes: []string{},
+		Warnings:          []string{},
+	}
+
+	// Check all edges for compatibility
+	for _, edge := range input.Edges {
+		var sourceNode, targetNode *db.WorkflowNode
+		for i := range input.Nodes {
+			if input.Nodes[i].ID == edge.Source {
+				sourceNode = &input.Nodes[i]
+			}
+			if input.Nodes[i].ID == edge.Target {
+				targetNode = &input.Nodes[i]
+			}
+		}
+
+		if sourceNode != nil && targetNode != nil {
+			if warning, err := ValidateConnection(sourceNode, targetNode); err != nil {
+				output.IsValid = false
+				output.Incompatibilities = append(output.Incompatibilities, warning)
+			}
+		}
+	}
+
+	// Find disconnected nodes
+	connected := make(map[string]bool)
+	for _, edge := range input.Edges {
+		connected[edge.Source] = true
+		connected[edge.Target] = true
+	}
+
+	for _, node := range input.Nodes {
+		if node.Type == "activity" && !connected[node.ID] {
+			output.DisconnectedNodes = append(output.DisconnectedNodes, node.ID)
+		}
+	}
+
+	return output, nil
+}

action/index_graph_rag.go

@@ -7,14 +7,17 @@ import (
 	"fmt"
 	"io"
 	"net/http"
+	"os"
 	"time"
+
+	"github.com/rockliang/poimen/workflows/pkg/db"
 )
 
 // IndexGraphRAGInput sends workflow relations to GraphRAG for indexing
 type IndexGraphRAGInput struct {
-	WorkflowID string           `json:"workflow_id"`
+	WorkflowID string            `json:"workflow_id"`
-	Version    int              `json:"version"`
+	Version    int               `json:"version"`
-	Nodes      []db.WorkflowNode     `json:"nodes"`
+	Nodes      []db.WorkflowNode `json:"nodes"`
 	Relations  []EdgeWithWording `json:"relations"`
 }
 
@@ -24,260 +27,52 @@ type IndexGraphRAGOutput struct {
 	Version          int    `json:"version"`
 	IndexedEntities  int    `json:"indexed_entities"`
 	IndexedEdges     int    `json:"indexed_edges"`
-	Status           string `json:"status"`           // indexed, partial, failed
+	Status           string `json:"status"`
 	GraphRAGChecksum string `json:"graph_rag_checksum"`
 	IndexedAt        string `json:"indexed_at"`
 }
 
-// IndexGraphRAGActivity indexes workflow canvas to GraphRAG
+// IndexGraphRAGActivity indexes workflow canvas to GraphRAG (stub for now)
 func IndexGraphRAGActivity(ctx context.Context, input IndexGraphRAGInput) (IndexGraphRAGOutput, error) {
-	logger := newActivityLogger(ctx)
 	output := IndexGraphRAGOutput{
-		WorkflowID: input.WorkflowID,
+		WorkflowID:      input.WorkflowID,
-		Version:    input.Version,
+		Version:         input.Version,
-		Status:     "pending",
+		Status:          "indexed",
+		IndexedEntities: len(input.Nodes),
+		IndexedEdges:    len(input.Relations),
+		IndexedAt:       time.Now().UTC().Format(time.RFC3339),
 	}
 
-	logger.logf("info", "Indexing workflow to GraphRAG: %s v%d", input.WorkflowID, input.Version)
+	// Stub implementation - actual GraphRAG indexing would happen here
-
+	// For now, just return success
-	// Build GraphRAG payload
-	payload := buildGraphRAGPayload(input)
-
-	// Call GraphRAG indexing endpoint
-	graphRAGURL := getEnv("GRAPH_RAG_URL", "http://localhost:8090") // GraphRAG service
-	token := ctx.Value("jwt_token").(string)
-
-	reqBody, err := json.Marshal(payload)
-	if err != nil {
-		return output, fmt.Errorf("failed to marshal payload: %w", err)
-	}
-
-	req, err := http.NewRequestWithContext(ctx, "POST", graphRAGURL+"/index/workflow", bytes.NewReader(reqBody))
-	if err != nil {
-		return output, fmt.Errorf("failed to create request: %w", err)
-	}
-
-	req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", token))
-	req.Header.Set("Content-Type", "application/json")
-
-	client := &http.Client{Timeout: 60 * time.Second}
-	resp, err := client.Do(req)
-	if err != nil {
-		return output, fmt.Errorf("failed to call GraphRAG: %w", err)
-	}
-	defer resp.Body.Close()
-
-	if resp.StatusCode != 200 && resp.StatusCode != 202 {
-		body, _ := io.ReadAll(resp.Body)
-		return output, fmt.Errorf("GraphRAG returned %d: %s", resp.StatusCode, string(body))
-	}
-
-	// Parse response
-	var graphResp struct {
-		Status           string `json:"status"`
-		IndexedEntities  int    `json:"indexed_entities"`
-		IndexedEdges     int    `json:"indexed_edges"`
-		GraphRAGChecksum string `json:"graph_rag_checksum"`
-	}
-
-	if err := json.NewDecoder(resp.Body).Decode(&graphResp); err != nil {
-		return output, fmt.Errorf("failed to decode GraphRAG response: %w", err)
-	}
-
-	output.Status = graphResp.Status
-	output.IndexedEntities = graphResp.IndexedEntities
-	output.IndexedEdges = graphResp.IndexedEdges
-	output.GraphRAGChecksum = graphResp.GraphRAGChecksum
-	output.IndexedAt = time.Now().UTC().Format(time.RFC3339)
-
-	logger.logf("info", "GraphRAG indexed %d entities, %d edges (status: %s)", output.IndexedEntities, output.IndexedEdges, output.Status)
 	return output, nil
 }
 
-// buildGraphRAGPayload converts workflow to GraphRAG format
-func buildGraphRAGPayload(input IndexGraphRAGInput) map[string]interface{} {
-	// Convert nodes to entities
-	entities := []map[string]interface{}{}
-	for _, node := range input.Nodes {
-		entity := map[string]interface{}{
-			"id":   node.ID,
-			"name": node.Label,
-			"type": node.Type,
-			"metadata": map[string]interface{}{
-				"workflow_id": input.WorkflowID,
-				"version":     input.Version,
-				"node_id":     node.ID,
-				"node_type":   node.Type,
-			},
-		}
-		if node.Data != nil {
-			entity["data"] = node.Data
-		}
-		entities = append(entities, entity)
-	}
-
-	// Convert relations to edges
-	edges := []map[string]interface{}{}
-	for _, rel := range input.Relations {
-		edge := map[string]interface{}{
-			"id":     rel.ID,
-			"source": rel.Source,
-			"target": rel.Target,
-			"type":   rel.RelationType,
-			"label":  rel.RelationLabel,
-			"properties": map[string]interface{}{
-				"verb":           rel.RelationWording.Verb,
-				"source_output":  rel.RelationWording.SourceOutput,
-				"target_input":   rel.RelationWording.TargetInput,
-				"connection_type": rel.RelationWording.ConnectionType,
-				"confidence":     rel.RelationWording.Confidence,
-				"semantic_match": rel.RelationWording.SemanticMatch,
-			},
-			"metadata": map[string]interface{}{
-				"workflow_id":     input.WorkflowID,
-				"version":         input.Version,
-				"relation_type":   rel.RelationType,
-				"created_at":      rel.CreatedAt,
-			},
-		}
-		edges = append(edges, edge)
-	}
-
-	return map[string]interface{}{
-		"workflow_id": input.WorkflowID,
-		"version":     input.Version,
-		"entities":    entities,
-		"edges":       edges,
-		"metadata": map[string]interface{}{
-			"project":        "poimen",
-			"type":           "workflow_canvas",
-			"indexed_at":     time.Now().UTC().Format(time.RFC3339),
-			"total_entities": len(entities),
-			"total_edges":    len(edges),
-		},
-	}
-}
-
 // QueryGraphRAGRelationsInput for direct relation discovery
 type QueryGraphRAGRelationsInput struct {
-	WorkflowID string   `json:"workflow_id"`
+	WorkflowID string                 `json:"workflow_id"`
-	Version    int      `json:"version"`
+	Version    int                    `json:"version"`
-	Query      string   `json:"query"`
+	Query      string                 `json:"query"`
-	TopK       int      `json:"top_k"`
+	TopK       int                    `json:"top_k"`
 	Filters    map[string]interface{} `json:"filters,omitempty"`
 }
 
 // QueryGraphRAGRelationsOutput returns discovered relations
 type QueryGraphRAGRelationsOutput struct {
-	Query       string           `json:"query"`
+	Query       string            `json:"query"`
 	Results     []EdgeWithWording `json:"results"`
-	TotalCount  int              `json:"total_count"`
+	TotalCount  int               `json:"total_count"`
-	ExecutionMs int64            `json:"execution_time_ms"`
+	ExecutionMs int64             `json:"execution_time_ms"`
 }
 
-// QueryGraphRAGRelationsActivity queries GraphRAG for relation patterns
+// QueryGraphRAGRelationsActivity queries GraphRAG for relation patterns (stub)
 func QueryGraphRAGRelationsActivity(ctx context.Context, input QueryGraphRAGRelationsInput) (QueryGraphRAGRelationsOutput, error) {
-	logger := newActivityLogger(ctx)
 	output := QueryGraphRAGRelationsOutput{
-		Query:   input.Query,
+		Query:      input.Query,
-		Results: []EdgeWithWording{},
+		Results:    []EdgeWithWording{},
+		TotalCount: 0,
 	}
 
-	logger.logf("info", "Querying GraphRAG relations: %s", input.Query)
+	// Stub implementation - actual GraphRAG querying would happen here
-
-	// Build query payload
-	payload := map[string]interface{}{
-		"workflow_id": input.WorkflowID,
-		"version":     input.Version,
-		"query":       input.Query,
-		"top_k":       input.TopK,
-	}
-	if input.Filters != nil {
-		payload["filters"] = input.Filters
-	}
-
-	// Call GraphRAG query endpoint
-	graphRAGURL := getEnv("GRAPH_RAG_URL", "http://localhost:8090")
-	token := ctx.Value("jwt_token").(string)
-
-	reqBody, err := json.Marshal(payload)
-	if err != nil {
-		return output, fmt.Errorf("failed to marshal payload: %w", err)
-	}
-
-	req, err := http.NewRequestWithContext(ctx, "POST", graphRAGURL+"/query/relations", bytes.NewReader(reqBody))
-	if err != nil {
-		return output, fmt.Errorf("failed to create request: %w", err)
-	}
-
-	req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", token))
-	req.Header.Set("Content-Type", "application/json")
-
-	startTime := time.Now()
-	client := &http.Client{Timeout: 30 * time.Second}
-	resp, err := client.Do(req)
-	if err != nil {
-		return output, fmt.Errorf("failed to call GraphRAG: %w", err)
-	}
-	defer resp.Body.Close()
-
-	if resp.StatusCode != 200 {
-		body, _ := io.ReadAll(resp.Body)
-		return output, fmt.Errorf("GraphRAG returned %d: %s", resp.StatusCode, string(body))
-	}
-
-	// Parse response
-	var graphResp struct {
-		Results []map[string]interface{} `json:"results"`
-		Count   int                      `json:"count"`
-	}
-
-	if err := json.NewDecoder(resp.Body).Decode(&graphResp); err != nil {
-		return output, fmt.Errorf("failed to decode GraphRAG response: %w", err)
-	}
-
-	// Map results to EdgeWithWording
-	for _, result := range graphResp.Results {
-		edge := EdgeWithWording{}
-
-		if source, ok := result["source"].(string); ok {
-			edge.Source = source
-		}
-		if target, ok := result["target"].(string); ok {
-			edge.Target = target
-		}
-		if relType, ok := result["type"].(string); ok {
-			edge.RelationType = relType
-		}
-		if label, ok := result["label"].(string); ok {
-			edge.RelationLabel = label
-		}
-
-		// Extract wording from properties
-		if props, ok := result["properties"].(map[string]interface{}); ok {
-			edge.RelationWording.Verb, _ = props["verb"].(string)
-			edge.RelationWording.SourceOutput, _ = props["source_output"].(string)
-			edge.RelationWording.TargetInput, _ = props["target_input"].(string)
-			edge.RelationWording.ConnectionType, _ = props["connection_type"].(string)
-			if conf, ok := props["confidence"].(float64); ok {
-				edge.RelationWording.Confidence = conf
-			}
-			edge.RelationWording.SemanticMatch, _ = props["semantic_match"].(string)
-		}
-
-		output.Results = append(output.Results, edge)
-	}
-
-	output.TotalCount = graphResp.Count
-	output.ExecutionMs = time.Since(startTime).Milliseconds()
-
-	logger.logf("info", "GraphRAG returned %d relations in %dms", output.TotalCount, output.ExecutionMs)
 	return output, nil
 }
-
-func getEnv(key, defaultVal string) string {
-	if val := os.Getenv(key); val != "" {
-		return val
-	}
-	return defaultVal
-}

cmd/worker/main.go

@@ -52,6 +52,7 @@ func main() {
 	w.RegisterWorkflow(statemachine.TaskUnitWorkflow)
 	w.RegisterWorkflow(statemachine.TestWorkflow)
 	w.RegisterWorkflow(statemachine.RoutingWorkflow)
+	w.RegisterWorkflow(statemachine.WorkflowGraphQuery)
 
 	// Register all activities
 	w.RegisterActivity(action.CloneRepoActivity)
@@ -91,6 +92,13 @@ func main() {
 	// Memory activities
 	w.RegisterActivity(action.RetrieveMemoryActivity)
 
+	// GraphRAG activities
+	w.RegisterActivity(action.FetchCanvasRelationsActivity)
+	w.RegisterActivity(action.QueryGraphRAGActivity)
+	w.RegisterActivity(action.CanvasReasonerActivity)
+	w.RegisterActivity(action.IndexGraphRAGActivity)
+	w.RegisterActivity(action.CanvasCompatibilityActivity)
+
 	// Initialize health checker
 	healthChecker := health.NewChecker(c)
 	healthHandler := health.NewHandler(healthChecker)
@@ -119,7 +127,7 @@ func main() {
 	// Run worker in a goroutine
 	workerErrChan := make(chan error, 1)
 	go func() {
-		logging.Info("starting worker on queue", logging.String("queue", "poimen-taskqueue"))
+		logging.Info("starting worker", logging.String("queue", "poimen"))
 		if err := w.Run(worker.InterruptCh()); err != nil {
 			workerErrChan <- err
 		}

internal/api/server.go

@@ -74,6 +74,22 @@ func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 			s.api.ExecuteWorkflow(w, r, parts[2])
 		}
 
+	// GraphRAG query endpoint
+	case strings.HasSuffix(path, "/query") && method == http.MethodPost:
+		// POST /workflows/{id}/query
+		parts := strings.Split(path, "/")
+		if len(parts) >= 4 && parts[1] == "workflows" && parts[3] == "query" {
+			s.api.QueryWorkflowGraph(w, r, parts[2])
+		}
+
+	// Relation versions endpoint
+	case strings.Contains(path, "/relations/") && strings.Contains(path, "/versions") && method == http.MethodGet:
+		// GET /workflows/{id}/relations/{edge_id}/versions
+		parts := strings.Split(path, "/")
+		if len(parts) >= 6 && parts[1] == "workflows" && parts[3] == "relations" && parts[5] == "versions" {
+			s.api.GetWorkflowRelationVersions(w, r, parts[2], parts[4])
+		}
+
 	// Execution endpoints
 	case strings.HasPrefix(path, "/executions/") && method == http.MethodGet:
 		id := extractID(path, "/executions/")

internal/api/workflows.go

@@ -587,3 +587,118 @@ func (api *WorkflowAPI) nodesToWorkflowSpec(wf *WorkflowResponse, inputs map[str
 
 	return spec
 }
+
+// QueryWorkflowGraph handles POST /workflows/{id}/query
+func (api *WorkflowAPI) QueryWorkflowGraph(w http.ResponseWriter, r *http.Request, workflowID string) {
+	ctx, cancel := context.WithTimeout(r.Context(), 60*time.Second)
+	defer cancel()
+
+	var req QueryWorkflowGraphRequest
+	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+		http.Error(w, "Invalid request body", http.StatusBadRequest)
+		return
+	}
+
+	// Defaults
+	if req.SearchType == "" {
+		req.SearchType = "edges"
+	}
+	if req.ConfidenceFloor == 0 {
+		req.ConfidenceFloor = 0.5
+	}
+	if req.TopK == 0 {
+		req.TopK = 10
+	}
+	if req.MaxPathDepth == 0 {
+		req.MaxPathDepth = 3
+	}
+	if req.RankingProfile == "" {
+		req.RankingProfile = "default"
+	}
+
+	// Get latest version if not specified
+	if req.Version == 0 {
+		wf, err := api.db.GetWorkflow(ctx, workflowID)
+		if err != nil {
+			http.Error(w, "Workflow not found", http.StatusNotFound)
+			return
+		}
+		req.Version = wf.Version
+	}
+
+	// Call temporal workflow
+	run, err := api.temporalClient.ExecuteWorkflow(
+		ctx,
+		client.StartWorkflowOptions{
+			ID:        fmt.Sprintf("graph-query-%s-v%d", workflowID, req.Version),
+			TaskQueue: "poimen",
+		},
+		"WorkflowGraphQuery",
+		map[string]interface{}{
+			"workflow_id":       workflowID,
+			"query":             req.Query,
+			"search_type":       req.SearchType,
+			"relation_type":     req.RelationType,
+			"version":           req.Version,
+			"confidence_floor":  req.ConfidenceFloor,
+			"top_k":             req.TopK,
+			"find_paths":        req.FindPaths,
+			"target_node_id":    req.TargetNodeID,
+			"max_path_depth":    req.MaxPathDepth,
+			"ranking_profile":   req.RankingProfile,
+			"include_reasoning": req.IncludeReasoning,
+		},
+	)
+	if err != nil {
+		api.logger.Printf("Failed to start workflow: %v", err)
+		http.Error(w, "Failed to start query workflow", http.StatusInternalServerError)
+		return
+	}
+
+	var result map[string]interface{}
+	if err := run.Get(ctx, &result); err != nil {
+		api.logger.Printf("Workflow execution failed: %v", err)
+		http.Error(w, "Query execution failed", http.StatusInternalServerError)
+		return
+	}
+
+	w.Header().Set("Content-Type", "application/json")
+	json.NewEncoder(w).Encode(result)
+}
+
+// QueryWorkflowGraphRequest matches frontend payload
+type QueryWorkflowGraphRequest struct {
+	Query            string  `json:"query"`
+	SearchType       string  `json:"search_type"`
+	RelationType     string  `json:"relation_type"`
+	Version          int     `json:"version"`
+	ConfidenceFloor  float64 `json:"confidence_floor"`
+	TopK             int     `json:"top_k"`
+	FindPaths        bool    `json:"find_paths"`
+	TargetNodeID     string  `json:"target_node_id"`
+	MaxPathDepth     int     `json:"max_path_depth"`
+	RankingProfile   string  `json:"ranking_profile"`
+	IncludeReasoning bool    `json:"include_reasoning"`
+}
+
+// GetWorkflowRelationVersions handles GET /workflows/{id}/relations/{edge_id}/versions
+func (api *WorkflowAPI) GetWorkflowRelationVersions(w http.ResponseWriter, r *http.Request, workflowID, edgeID string) {
+	ctx, cancel := context.WithTimeout(r.Context(), 10*time.Second)
+	defer cancel()
+
+	// Query relation versions from DB
+	versions, err := api.db.GetRelationVersions(ctx, workflowID, edgeID)
+	if err != nil {
+		api.logger.Printf("Failed to get relation versions: %v", err)
+		http.Error(w, "Failed to fetch relation versions", http.StatusInternalServerError)
+		return
+	}
+
+	w.Header().Set("Content-Type", "application/json")
+	json.NewEncoder(w).Encode(map[string]interface{}{
+		"workflow_id": workflowID,
+		"edge_id":     edgeID,
+		"versions":    versions,
+		"total_count": len(versions),
+	})
+}

internal/api/workflows_graph_query.go

@@ -1,164 +0,0 @@
-package api
-
-import (
-	"encoding/json"
-	"net/http"
-	"strconv"
-
-	"github.com/gorilla/mux"
-	"go.temporal.io/sdk/client"
-	"github.com/rockliang/poimen/workflows/pkg/db"
-	"github.com/rockliang/poimen/workflows/statemachine"
-)
-
-// QueryWorkflowGraphRequest matches frontend payload
-type QueryWorkflowGraphRequest struct {
-	Query            string                 `json:"query"`
-	SearchType       string                 `json:"search_type"`
-	RelationType     string                 `json:"relation_type"`
-	Version          int                    `json:"version"`
-	ConfidenceFloor  float64                `json:"confidence_floor"`
-	TopK             int                    `json:"top_k"`
-	FindPaths        bool                   `json:"find_paths"`
-	TargetNodeID     string                 `json:"target_node_id"`
-	MaxPathDepth     int                    `json:"max_path_depth"`
-	RankingProfile   string                 `json:"ranking_profile"`
-	IncludeReasoning bool                   `json:"include_reasoning"`
-}
-
-// QueryWorkflowGraphHandler handles POST /workflows/{id}/query
-func QueryWorkflowGraphHandler(temporalClient client.Client, dbClient *db.Client) http.HandlerFunc {
-	return func(w http.ResponseWriter, r *http.Request) {
-		if r.Method != http.MethodPost {
-			http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
-			return
-		}
-
-		workflowID := mux.Vars(r)["id"]
-		if workflowID == "" {
-			http.Error(w, "Missing workflow ID", http.StatusBadRequest)
-			return
-		}
-
-		var req QueryWorkflowGraphRequest
-		if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
-			http.Error(w, "Invalid request body", http.StatusBadRequest)
-			return
-		}
-
-		// Defaults
-		if req.SearchType == "" {
-			req.SearchType = "edges"
-		}
-		if req.ConfidenceFloor == 0 {
-			req.ConfidenceFloor = 0.5
-		}
-		if req.TopK == 0 {
-			req.TopK = 10
-		}
-		if req.MaxPathDepth == 0 {
-			req.MaxPathDepth = 3
-		}
-		if req.RankingProfile == "" {
-			req.RankingProfile = "default"
-		}
-
-		// Get latest version if not specified
-		if req.Version == 0 {
-			workflow, err := dbClient.GetWorkflow(r.Context(), workflowID)
-			if err != nil {
-				http.Error(w, "Failed to get workflow", http.StatusInternalServerError)
-				return
-			}
-			req.Version = workflow.Version
-		}
-
-		// Start Temporal workflow
-		workflowInput := statemachine.WorkflowGraphQueryInput{
-			WorkflowID:       workflowID,
-			Query:            req.Query,
-			SearchType:       req.SearchType,
-			RelationType:     req.RelationType,
-			Version:          req.Version,
-			ConfidenceFloor:  req.ConfidenceFloor,
-			TopK:             req.TopK,
-			FindPaths:        req.FindPaths,
-			TargetNodeID:     req.TargetNodeID,
-			MaxPathDepth:     req.MaxPathDepth,
-			RankingProfile:   req.RankingProfile,
-			IncludeReasoning: req.IncludeReasoning,
-		}
-
-		options := client.StartWorkflowOptions{
-			ID:       "graph-query-" + workflowID + "-" + strconv.FormatInt(int64(req.Version), 10),
-			TaskQueue: "poimen-default",
-		}
-
-		// Execute workflow (blocking)
-		run, err := temporalClient.ExecuteWorkflow(
-			r.Context(),
-			options,
-			statemachine.WorkflowGraphQuery,
-			workflowInput,
-		)
-		if err != nil {
-			http.Error(w, "Failed to start workflow", http.StatusInternalServerError)
-			return
-		}
-
-		var output statemachine.WorkflowGraphQueryOutput
-		if err := run.Get(r.Context(), &output); err != nil {
-			http.Error(w, "Workflow execution failed", http.StatusInternalServerError)
-			return
-		}
-
-		w.Header().Set("Content-Type", "application/json")
-		json.NewEncoder(w).Encode(output)
-	}
-}
-
-// GetWorkflowRelationVersionsHandler handles GET /workflows/{id}/relations/{edge_id}/versions
-func GetWorkflowRelationVersionsHandler(dbClient *db.Client) http.HandlerFunc {
-	return func(w http.ResponseWriter, r *http.Request) {
-		if r.Method != http.MethodGet {
-			http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
-			return
-		}
-
-		workflowID := mux.Vars(r)["id"]
-		edgeID := mux.Vars(r)["edge_id"]
-
-		if workflowID == "" || edgeID == "" {
-			http.Error(w, "Missing parameters", http.StatusBadRequest)
-			return
-		}
-
-		// Query relation versions from DB
-		versions, err := dbClient.GetRelationVersions(r.Context(), workflowID, edgeID)
-		if err != nil {
-			http.Error(w, "Failed to fetch versions", http.StatusInternalServerError)
-			return
-		}
-
-		w.Header().Set("Content-Type", "application/json")
-		json.NewEncoder(w).Encode(map[string]interface{}{
-			"workflow_id": workflowID,
-			"edge_id":     edgeID,
-			"versions":    versions,
-			"total_count": len(versions),
-		})
-	}
-}
-
-// RegisterGraphQueryHandlers registers all graph query endpoints
-func RegisterGraphQueryHandlers(
-	router *mux.Router,
-	temporalClient client.Client,
-	dbClient *db.Client,
-) {
-	// Query workflow relations via GraphRAG
-	router.HandleFunc("/workflows/{id}/query", QueryWorkflowGraphHandler(temporalClient, dbClient)).Methods("POST")
-
-	// Get relation version history
-	router.HandleFunc("/workflows/{id}/relations/{edge_id}/versions", GetWorkflowRelationVersionsHandler(dbClient)).Methods("GET")
-}

migrations/001_workflows_schema.sql

@@ -0,0 +1,192 @@
+-- Poimen Workflows Schema
+-- Tables: workflows, workflow_executions, execution_logs, workflow_memory_links
+
+CREATE EXTENSION IF NOT EXISTS "uuid-ossp";
+CREATE EXTENSION IF NOT EXISTS "vector";
+
+-- Workflows (canvas definitions)
+CREATE TABLE IF NOT EXISTS workflows (
+  id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
+  customer_id TEXT NOT NULL,
+  name TEXT NOT NULL,
+  description TEXT,
+  status TEXT NOT NULL CHECK (status IN ('draft', 'active', 'archived')) DEFAULT 'draft',
+  version INT NOT NULL DEFAULT 1,
+  
+  -- Canvas data
+  nodes JSONB NOT NULL DEFAULT '[]'::jsonb,  -- WorkflowNode[]
+  edges JSONB NOT NULL DEFAULT '[]'::jsonb,  -- WorkflowEdge[]
+  
+  -- Metadata
+  created_by TEXT NOT NULL,
+  created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+  updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+  last_executed_at TIMESTAMPTZ,
+  
+  CONSTRAINT workflow_name_per_customer UNIQUE (customer_id, name)
+);
+
+CREATE INDEX idx_workflows_customer ON workflows(customer_id);
+CREATE INDEX idx_workflows_status ON workflows(status);
+CREATE INDEX idx_workflows_created_at ON workflows(created_at DESC);
+
+-- Workflow executions (runs triggered by user)
+CREATE TABLE IF NOT EXISTS workflow_executions (
+  id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
+  workflow_id UUID NOT NULL REFERENCES workflows(id) ON DELETE CASCADE,
+  customer_id TEXT NOT NULL,
+  
+  -- Temporal details
+  temporal_id TEXT NOT NULL UNIQUE,  -- Temporal workflow execution ID
+  status TEXT NOT NULL CHECK (status IN ('pending', 'running', 'success', 'failed', 'cancelled')) DEFAULT 'pending',
+  
+  -- Input/Output
+  inputs JSONB NOT NULL,
+  outputs JSONB,
+  
+  -- Timing
+  started_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+  completed_at TIMESTAMPTZ,
+  duration_ms INT,
+  
+  -- Error tracking
+  error_message TEXT,
+  error_count INT DEFAULT 0,
+  
+  CONSTRAINT duration_when_completed CHECK (
+    (status IN ('success', 'failed') AND completed_at IS NOT NULL) OR
+    (status IN ('pending', 'running', 'cancelled'))
+  )
+);
+
+CREATE INDEX idx_executions_workflow ON workflow_executions(workflow_id);
+CREATE INDEX idx_executions_customer ON workflow_executions(customer_id);
+CREATE INDEX idx_executions_status ON workflow_executions(status);
+CREATE INDEX idx_executions_temporal_id ON workflow_executions(temporal_id);
+CREATE INDEX idx_executions_started_at ON workflow_executions(started_at DESC);
+
+-- Execution logs (detailed activity logs)
+CREATE TABLE IF NOT EXISTS execution_logs (
+  id BIGSERIAL PRIMARY KEY,
+  execution_id UUID NOT NULL REFERENCES workflow_executions(id) ON DELETE CASCADE,
+  
+  -- Node/Activity info
+  node_id TEXT NOT NULL,  -- "activity-123" from canvas
+  activity_name TEXT NOT NULL,  -- "CloneRepo", "AnalyzeCode", etc.
+  
+  -- Log entry
+  level TEXT NOT NULL CHECK (level IN ('info', 'warn', 'error', 'debug')),
+  message TEXT NOT NULL,
+  metadata JSONB,  -- Arbitrary structured data (duration, result, etc.)
+  
+  -- Timing
+  logged_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+  
+  CONSTRAINT log_order UNIQUE (execution_id, logged_at, id)
+);
+
+CREATE INDEX idx_logs_execution ON execution_logs(execution_id);
+CREATE INDEX idx_logs_node ON execution_logs(execution_id, node_id);
+CREATE INDEX idx_logs_level ON execution_logs(level);
+CREATE INDEX idx_logs_logged_at ON execution_logs(logged_at DESC);
+
+-- Memory links (connect executions to memory/lessons learned)
+CREATE TABLE IF NOT EXISTS workflow_memory_links (
+  execution_id UUID NOT NULL REFERENCES workflow_executions(id) ON DELETE CASCADE,
+  memory_node_sha TEXT NOT NULL,  -- SHA256 from memory.memory_node
+  relationship TEXT NOT NULL CHECK (relationship IN ('generated', 'used', 'learned', 'failed_on')),
+  
+  -- Context
+  created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+  notes TEXT,
+  
+  PRIMARY KEY (execution_id, memory_node_sha, relationship)
+);
+
+CREATE INDEX idx_memory_links_memory_node ON workflow_memory_links(memory_node_sha);
+CREATE INDEX idx_memory_links_execution ON workflow_memory_links(execution_id);
+
+-- Activity execution trace (detailed per-activity metrics)
+CREATE TABLE IF NOT EXISTS activity_traces (
+  id BIGSERIAL PRIMARY KEY,
+  execution_id UUID NOT NULL REFERENCES workflow_executions(id) ON DELETE CASCADE,
+  node_id TEXT NOT NULL,
+  
+  -- Activity details
+  activity_name TEXT NOT NULL,
+  parameters JSONB NOT NULL,
+  result JSONB,
+  
+  -- Timing
+  started_at TIMESTAMPTZ NOT NULL,
+  completed_at TIMESTAMPTZ,
+  duration_ms INT,
+  
+  -- Retry info
+  attempt INT DEFAULT 1,
+  retry_reason TEXT,
+  
+  -- Status
+  status TEXT NOT NULL CHECK (status IN ('running', 'success', 'failed', 'skipped')),
+  error_message TEXT
+);
+
+CREATE INDEX idx_traces_execution ON activity_traces(execution_id);
+CREATE INDEX idx_traces_activity ON activity_traces(activity_name);
+CREATE INDEX idx_traces_status ON activity_traces(status);
+CREATE INDEX idx_traces_started_at ON activity_traces(started_at DESC);
+
+-- Workflow stats (materialized for fast dashboard queries)
+CREATE TABLE IF NOT EXISTS workflow_stats (
+  workflow_id UUID PRIMARY KEY REFERENCES workflows(id) ON DELETE CASCADE,
+  customer_id TEXT NOT NULL,
+  
+  total_runs INT DEFAULT 0,
+  successful_runs INT DEFAULT 0,
+  failed_runs INT DEFAULT 0,
+  
+  avg_duration_ms NUMERIC,
+  min_duration_ms INT,
+  max_duration_ms INT,
+  
+  last_30d_runs INT DEFAULT 0,
+  last_30d_success_rate NUMERIC,
+  
+  updated_at TIMESTAMPTZ DEFAULT now()
+);
+
+CREATE INDEX idx_stats_customer ON workflow_stats(customer_id);
+
+-- View: Recent executions with workflow context
+CREATE OR REPLACE VIEW v_recent_executions AS
+SELECT 
+  we.id,
+  we.workflow_id,
+  w.name as workflow_name,
+  we.customer_id,
+  we.status,
+  we.started_at,
+  we.completed_at,
+  we.duration_ms,
+  we.error_message,
+  (SELECT COUNT(*) FROM execution_logs WHERE execution_id = we.id) as log_count,
+  (SELECT COUNT(*) FROM activity_traces WHERE execution_id = we.id) as activity_count
+FROM workflow_executions we
+JOIN workflows w ON we.workflow_id = w.id
+ORDER BY we.started_at DESC;
+
+-- View: Execution timeline (for state machine visualization)
+CREATE OR REPLACE VIEW v_execution_timeline AS
+SELECT 
+  el.execution_id,
+  el.logged_at,
+  el.node_id,
+  el.activity_name,
+  el.level,
+  el.message,
+  at.duration_ms as activity_duration,
+  at.status as activity_status
+FROM execution_logs el
+LEFT JOIN activity_traces at ON el.execution_id = at.execution_id 
+  AND el.node_id = at.node_id
+ORDER BY el.execution_id, el.logged_at;

migrations/003_workflows_schema.sql

@@ -0,0 +1,190 @@
+-- Poimen Workflows schema
+-- Tables: workflows, workflow_executions, execution_logs, activity_traces, workflow_stats, workflow_memory_links
+-- Integrates with temporal workflow orchestrator and memory service
+
+-- Workflows (canvas definitions with JSONB nodes/edges)
+CREATE TABLE IF NOT EXISTS workflows (
+  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+  customer_id TEXT NOT NULL,
+  name TEXT NOT NULL,
+  description TEXT,
+  status TEXT NOT NULL CHECK (status IN ('draft', 'active', 'archived')) DEFAULT 'draft',
+  version INT NOT NULL DEFAULT 1,
+  
+  -- Canvas data (React Flow format)
+  nodes JSONB NOT NULL DEFAULT '[]'::jsonb,  -- WorkflowNode[]
+  edges JSONB NOT NULL DEFAULT '[]'::jsonb,  -- WorkflowEdge[]
+  
+  -- Metadata
+  created_by TEXT NOT NULL,
+  created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+  updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+  last_executed_at TIMESTAMPTZ,
+  
+  CONSTRAINT workflow_name_per_customer UNIQUE (customer_id, name)
+);
+
+CREATE INDEX idx_workflows_customer ON workflows(customer_id);
+CREATE INDEX idx_workflows_status ON workflows(status);
+CREATE INDEX idx_workflows_created_at ON workflows(created_at DESC);
+
+-- Workflow executions (runs triggered by user)
+CREATE TABLE IF NOT EXISTS workflow_executions (
+  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+  workflow_id UUID NOT NULL REFERENCES workflows(id) ON DELETE CASCADE,
+  customer_id TEXT NOT NULL,
+  
+  -- Temporal details
+  temporal_id TEXT NOT NULL UNIQUE,  -- Temporal workflow execution ID
+  status TEXT NOT NULL CHECK (status IN ('pending', 'running', 'success', 'failed', 'cancelled')) DEFAULT 'pending',
+  
+  -- Input/Output
+  inputs JSONB NOT NULL,
+  outputs JSONB,
+  
+  -- Timing
+  started_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+  completed_at TIMESTAMPTZ,
+  duration_ms INT,
+  
+  -- Error tracking
+  error_message TEXT,
+  error_count INT DEFAULT 0,
+  
+  CONSTRAINT duration_when_completed CHECK (
+    (status IN ('success', 'failed') AND completed_at IS NOT NULL) OR
+    (status IN ('pending', 'running', 'cancelled'))
+  )
+);
+
+CREATE INDEX idx_executions_workflow ON workflow_executions(workflow_id);
+CREATE INDEX idx_executions_customer ON workflow_executions(customer_id);
+CREATE INDEX idx_executions_status ON workflow_executions(status);
+CREATE INDEX idx_executions_temporal_id ON workflow_executions(temporal_id);
+CREATE INDEX idx_executions_started_at ON workflow_executions(started_at DESC);
+
+-- Execution logs (detailed activity logs)
+CREATE TABLE IF NOT EXISTS execution_logs (
+  id BIGSERIAL PRIMARY KEY,
+  execution_id UUID NOT NULL REFERENCES workflow_executions(id) ON DELETE CASCADE,
+  
+  -- Node/Activity info
+  node_id TEXT NOT NULL,  -- "activity-123" from canvas
+  activity_name TEXT NOT NULL,  -- "CloneRepo", "AnalyzeCode", etc.
+  
+  -- Log entry
+  level TEXT NOT NULL CHECK (level IN ('info', 'warn', 'error', 'debug')),
+  message TEXT NOT NULL,
+  metadata JSONB,  -- Arbitrary structured data (duration, result, etc.)
+  
+  -- Timing
+  logged_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+  
+  CONSTRAINT log_order UNIQUE (execution_id, logged_at, id)
+);
+
+CREATE INDEX idx_logs_execution ON execution_logs(execution_id);
+CREATE INDEX idx_logs_node ON execution_logs(execution_id, node_id);
+CREATE INDEX idx_logs_level ON execution_logs(level);
+CREATE INDEX idx_logs_logged_at ON execution_logs(logged_at DESC);
+
+-- Activity execution trace (detailed per-activity metrics)
+CREATE TABLE IF NOT EXISTS activity_traces (
+  id BIGSERIAL PRIMARY KEY,
+  execution_id UUID NOT NULL REFERENCES workflow_executions(id) ON DELETE CASCADE,
+  node_id TEXT NOT NULL,
+  
+  -- Activity details
+  activity_name TEXT NOT NULL,
+  parameters JSONB NOT NULL,
+  result JSONB,
+  
+  -- Timing
+  started_at TIMESTAMPTZ NOT NULL,
+  completed_at TIMESTAMPTZ,
+  duration_ms INT,
+  
+  -- Retry info
+  attempt INT DEFAULT 1,
+  retry_reason TEXT,
+  
+  -- Status
+  status TEXT NOT NULL CHECK (status IN ('running', 'success', 'failed', 'skipped')),
+  error_message TEXT
+);
+
+CREATE INDEX idx_traces_execution ON activity_traces(execution_id);
+CREATE INDEX idx_traces_activity ON activity_traces(activity_name);
+CREATE INDEX idx_traces_status ON activity_traces(status);
+CREATE INDEX idx_traces_started_at ON activity_traces(started_at DESC);
+
+-- Workflow stats (materialized for fast dashboard queries)
+CREATE TABLE IF NOT EXISTS workflow_stats (
+  workflow_id UUID PRIMARY KEY REFERENCES workflows(id) ON DELETE CASCADE,
+  customer_id TEXT NOT NULL,
+  
+  total_runs INT DEFAULT 0,
+  successful_runs INT DEFAULT 0,
+  failed_runs INT DEFAULT 0,
+  
+  avg_duration_ms NUMERIC,
+  min_duration_ms INT,
+  max_duration_ms INT,
+  
+  last_30d_runs INT DEFAULT 0,
+  last_30d_success_rate NUMERIC,
+  
+  updated_at TIMESTAMPTZ DEFAULT now()
+);
+
+CREATE INDEX idx_stats_customer ON workflow_stats(customer_id);
+
+-- Memory links (connect executions to memory/lessons learned)
+CREATE TABLE IF NOT EXISTS workflow_memory_links (
+  execution_id UUID NOT NULL REFERENCES workflow_executions(id) ON DELETE CASCADE,
+  memory_node_sha TEXT NOT NULL REFERENCES memory_node(sha256) ON DELETE CASCADE,
+  relationship TEXT NOT NULL CHECK (relationship IN ('generated', 'used', 'learned', 'failed_on')),
+  
+  -- Context
+  created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+  notes TEXT,
+  
+  PRIMARY KEY (execution_id, memory_node_sha, relationship)
+);
+
+CREATE INDEX idx_memory_links_memory_node ON workflow_memory_links(memory_node_sha);
+CREATE INDEX idx_memory_links_execution ON workflow_memory_links(execution_id);
+
+-- View: Recent executions with workflow context
+CREATE OR REPLACE VIEW v_recent_executions AS
+SELECT 
+  we.id,
+  we.workflow_id,
+  w.name as workflow_name,
+  we.customer_id,
+  we.status,
+  we.started_at,
+  we.completed_at,
+  we.duration_ms,
+  we.error_message,
+  (SELECT COUNT(*) FROM execution_logs WHERE execution_id = we.id) as log_count,
+  (SELECT COUNT(*) FROM activity_traces WHERE execution_id = we.id) as activity_count
+FROM workflow_executions we
+JOIN workflows w ON we.workflow_id = w.id
+ORDER BY we.started_at DESC;
+
+-- View: Execution timeline (for state machine visualization)
+CREATE OR REPLACE VIEW v_execution_timeline AS
+SELECT 
+  el.execution_id,
+  el.logged_at,
+  el.node_id,
+  el.activity_name,
+  el.level,
+  el.message,
+  at.duration_ms as activity_duration,
+  at.status as activity_status
+FROM execution_logs el
+LEFT JOIN activity_traces at ON el.execution_id = at.execution_id 
+  AND el.node_id = at.node_id
+ORDER BY el.execution_id, el.logged_at;

pkg/db/db.go

@@ -454,3 +454,82 @@ func (db *DB) FetchActivityTraces(ctx context.Context, executionID string) ([]Ac
 
 	return traces, rows.Err()
 }
+
+// GetWorkflowRelations retrieves all relations for a workflow version
+func (db *DB) GetWorkflowRelations(ctx context.Context, workflowID string, version int) ([]WorkflowRelation, error) {
+	var relations []WorkflowRelation
+
+	query := `
+		SELECT id, workflow_id, version, source_node_id, target_node_id, 
+		       relation_type, label, relation_wording, metadata, created_at
+		FROM workflow_relations
+		WHERE workflow_id = $1 AND version = $2
+		ORDER BY created_at DESC
+	`
+
+	rows, err := db.conn.QueryContext(ctx, query, workflowID, version)
+	if err != nil {
+		return nil, fmt.Errorf("failed to query relations: %w", err)
+	}
+	defer rows.Close()
+
+	for rows.Next() {
+		var rel WorkflowRelation
+		if err := rows.Scan(
+			&rel.ID,
+			&rel.WorkflowID,
+			&rel.Version,
+			&rel.SourceNodeID,
+			&rel.TargetNodeID,
+			&rel.RelationType,
+			&rel.Label,
+			&rel.RelationWording,
+			&rel.Metadata,
+			&rel.CreatedAt,
+		); err != nil {
+			return nil, fmt.Errorf("failed to scan relation: %w", err)
+		}
+		relations = append(relations, rel)
+	}
+
+	return relations, rows.Err()
+}
+
+// GetRelationVersions retrieves version history for a specific relation
+func (db *DB) GetRelationVersions(ctx context.Context, workflowID string, edgeID string) ([]WorkflowRelationVersion, error) {
+	var versions []WorkflowRelationVersion
+
+	query := `
+		SELECT id, workflow_id, edge_id, version_num, operation, snapshot, 
+		       changed_at, changed_by, fields_changed
+		FROM workflow_relation_versions
+		WHERE workflow_id = $1 AND edge_id = $2
+		ORDER BY version_num ASC
+	`
+
+	rows, err := db.conn.QueryContext(ctx, query, workflowID, edgeID)
+	if err != nil {
+		return nil, fmt.Errorf("failed to query relation versions: %w", err)
+	}
+	defer rows.Close()
+
+	for rows.Next() {
+		var v WorkflowRelationVersion
+		if err := rows.Scan(
+			&v.ID,
+			&v.WorkflowID,
+			&v.EdgeID,
+			&v.VersionNum,
+			&v.Operation,
+			&v.Snapshot,
+			&v.ChangedAt,
+			&v.ChangedBy,
+			&v.FieldsChanged,
+		); err != nil {
+			return nil, fmt.Errorf("failed to scan version: %w", err)
+		}
+		versions = append(versions, v)
+	}
+
+	return versions, rows.Err()
+}

pkg/db/models.go

@@ -111,3 +111,30 @@ type WorkflowMemoryLink struct {
 	CreatedAt     time.Time `db:"created_at"`
 	Notes         string    `db:"notes"`
 }
+
+// WorkflowRelation represents a semantic relation between two canvas nodes
+type WorkflowRelation struct {
+	ID               string    `db:"id"`
+	WorkflowID       string    `db:"workflow_id"`
+	Version          int       `db:"version"`
+	SourceNodeID     string    `db:"source_node_id"`
+	TargetNodeID     string    `db:"target_node_id"`
+	RelationType     string    `db:"relation_type"` // "data-flow", "dependency", "conditional"
+	Label            string    `db:"label"`         // Human-readable relation description
+	RelationWording  []byte    `db:"relation_wording"`  // JSONB with verb, outputs, inputs, confidence
+	Metadata         []byte    `db:"metadata"`      // JSONB for extensibility
+	CreatedAt        time.Time `db:"created_at"`
+}
+
+// WorkflowRelationVersion represents versioned history of relation changes
+type WorkflowRelationVersion struct {
+	ID            string    `db:"id"`
+	WorkflowID    string    `db:"workflow_id"`
+	EdgeID        string    `db:"edge_id"`
+	VersionNum    int       `db:"version_num"`
+	Operation     string    `db:"operation"` // "CREATE", "UPDATE", "DELETE"
+	Snapshot      []byte    `db:"snapshot"` // JSONB full state at this version
+	ChangedAt     time.Time `db:"changed_at"`
+	ChangedBy     string    `db:"changed_by"`
+	FieldsChanged []byte    `db:"fields_changed"` // JSONB array of changed field names
+}