From 8474d7e49429a609eca1e0b87e88c5f21bb372b9 Mon Sep 17 00:00:00 2001 From: Test Date: Sat, 5 Sep 2026 05:58:53 -0700 Subject: [PATCH] chore: remove docker-compose (use k8s + CI/CD only) --- build-push.sh | 34 - docker-compose.yml | 138 --- docs/CANVAS_REASONER_LOGIC.md | 322 ------- docs/TEMPORAL_GRAPH_RAG_WIRING.md | 1013 ----------------------- docs/WORKFLOWS_GRAPH_RAG_INTEGRATION.md | 712 ---------------- k8s/git-commit.yaml | 2 +- k8s/worker-deployment.yaml | 2 +- 7 files changed, 2 insertions(+), 2221 deletions(-) delete mode 100755 build-push.sh delete mode 100644 docker-compose.yml delete mode 100644 docs/CANVAS_REASONER_LOGIC.md delete mode 100644 docs/TEMPORAL_GRAPH_RAG_WIRING.md delete mode 100644 docs/WORKFLOWS_GRAPH_RAG_INTEGRATION.md diff --git a/build-push.sh b/build-push.sh deleted file mode 100755 index de730ea..0000000 --- a/build-push.sh +++ /dev/null @@ -1,34 +0,0 @@ -#!/bin/bash -set -e - -REGISTRY="forgejo.riotpiao.com/rock" -TAG="${1:-latest}" - -echo "πŸ”¨ Building Poimen Application Images" -echo " Registry: $REGISTRY" -echo " Tag: $TAG" -echo "" - -# Build memory service -echo "πŸ“¦ Building poimen-memory..." -docker build -t $REGISTRY/poimen-memory:$TAG ./memory -docker push $REGISTRY/poimen-memory:$TAG -echo "βœ“ Pushed poimen-memory:$TAG" - -# Build workflows service -echo "πŸ“¦ Building poimen-workflows..." -docker build -t $REGISTRY/poimen-workflows:$TAG ./workflows -docker push $REGISTRY/poimen-workflows:$TAG -echo "βœ“ Pushed poimen-workflows:$TAG" - -# Build frontend service -echo "πŸ“¦ Building poimen-frontend..." -docker build -t $REGISTRY/poimen-frontend:$TAG ./poimen-frontend -docker push $REGISTRY/poimen-frontend:$TAG -echo "βœ“ Pushed poimen-frontend:$TAG" - -echo "" -echo "βœ… All images built and pushed" -echo "" -echo "To deploy:" -echo " cd k8s && ./deploy.sh -a" diff --git a/docker-compose.yml b/docker-compose.yml deleted file mode 100644 index 4daf34e..0000000 --- a/docker-compose.yml +++ /dev/null @@ -1,138 +0,0 @@ -version: '3.8' - -services: - postgres: - image: postgres:15-alpine - environment: - POSTGRES_USER: poimen - POSTGRES_PASSWORD: poimen - POSTGRES_INITDB_ARGS: -c max_connections=200 - ports: - - "5432:5432" - volumes: - - postgres-data:/var/lib/postgresql/data - healthcheck: - test: ["CMD-SHELL", "pg_isready -U poimen"] - interval: 5s - timeout: 5s - retries: 5 - - redis: - image: redis:7-alpine - ports: - - "6379:6379" - healthcheck: - test: ["CMD", "redis-cli", "ping"] - interval: 5s - timeout: 5s - retries: 5 - - temporal: - image: temporalio/auto-setup:latest - ports: - - "7233:7233" - - "8233:8233" - environment: - DB: postgresql - DB_PORT: 5432 - POSTGRES_USER: poimen - POSTGRES_PASSWORD: poimen - POSTGRES_SEEDS: postgres - depends_on: - postgres: - condition: service_healthy - healthcheck: - test: ["CMD", "tctl", "workflow", "list"] - interval: 10s - timeout: 5s - retries: 5 - - poimen-memory: - build: - context: ./memory - dockerfile: Dockerfile - ports: - - "8000:8000" - environment: - DATABASE_URL: postgresql://poimen:poimen@postgres:5432/poimen_memory - REDIS_URL: redis://redis:6379 - JWT_SECRET: dev-secret-key - LOG_LEVEL: debug - depends_on: - postgres: - condition: service_healthy - redis: - condition: service_healthy - volumes: - - ./memory:/app - - poimen-workflows: - build: - context: ./workflows - dockerfile: Dockerfile - ports: - - "8080:8080" - environment: - DATABASE_URL: postgresql://poimen:poimen@postgres:5432/poimen_workflows - TEMPORAL_HOST: temporal:7233 - MEMORY_SERVICE_URL: http://poimen-memory:8000 - JWT_SECRET: dev-secret-key - LOG_LEVEL: debug - depends_on: - postgres: - condition: service_healthy - temporal: - condition: service_healthy - poimen-memory: - condition: service_started - volumes: - - ./workflows:/app - command: /app/workflows server - - poimen-workflows-worker: - build: - context: ./workflows - dockerfile: Dockerfile - environment: - DATABASE_URL: postgresql://poimen:poimen@postgres:5432/poimen_workflows - TEMPORAL_HOST: temporal:7233 - MEMORY_SERVICE_URL: http://poimen-memory:8000 - JWT_SECRET: dev-secret-key - LOG_LEVEL: debug - depends_on: - postgres: - condition: service_healthy - temporal: - condition: service_healthy - poimen-memory: - condition: service_started - volumes: - - ./workflows:/app - command: /app/workflows worker - - poimen-frontend: - build: - context: ./poimen-frontend - dockerfile: Dockerfile - ports: - - "3000:3000" - environment: - NEXT_PUBLIC_WORKFLOWS_API: http://localhost:8080 - NEXT_PUBLIC_MEMORY_API: http://localhost:8000 - NEXT_PUBLIC_AUTH_URL: http://localhost:8000/auth - OAUTH_CLIENT_ID: dev-client-id - OAUTH_CLIENT_SECRET: dev-client-secret - NODE_ENV: development - depends_on: - - poimen-workflows - - poimen-memory - volumes: - - ./poimen-frontend:/app - - /app/node_modules - -volumes: - postgres-data: - -networks: - default: - name: poimen-network diff --git a/docs/CANVAS_REASONER_LOGIC.md b/docs/CANVAS_REASONER_LOGIC.md deleted file mode 100644 index b8a36a1..0000000 --- a/docs/CANVAS_REASONER_LOGIC.md +++ /dev/null @@ -1,322 +0,0 @@ -# Canvas Reasoner: Auto-Inferring Workflow Connections - -## Overview - -The **CanvasReasonerActivity** uses LLM reasoning to automatically suggest connections between workflow activities when users drop new nodes onto the canvas. It analyzes input/output compatibility and detects connection problems. - -## Connection Logic - -### How It Works - -1. **Analyze Node Schemas** - - Get each activity's input/output fields from knowledge base - - Activities are classified as: - - **Generators**: No inputs, has outputs (e.g., API call, trigger) - - **Processors**: Has inputs and outputs (e.g., analyze code, security scan) - - **Sinks/Terminals**: Has inputs, no outputs (e.g., notification, approval) - -2. **LLM Reasoning** - - Pass all nodes + their schemas to reasoning model - - Ask LLM to suggest edges based on: - - Type compatibility (stringβ†’string, objectβ†’object) - - Logical execution order - - Data flow requirements - - Common workflow patterns - -3. **Validate Suggestions** - - Check all suggested edges exist in node map - - Skip self-loops - - Remove duplicates - -4. **Compatibility Checking** - - For each suggested edge: `source β†’ target` - - Verify source produces outputs - - Verify target accepts inputs - - Check output/input type compatibility - - Flag incompatible connections - -5. **Identify Issues** - - Collect all incompatible edges - - Identify disconnected nodes (no edges in/out) - - Generate user alerts for problems - -## Connection Impossibility Detection - -### Why Connections Fail - -1. **Missing Outputs** - ``` - NotifyStatusActivity β†’ AnalyzeCodeActivity - ⚠️ NotifyStatusActivity produces no outputs - Reason: Notification is terminal activity (sink) - Solution: Add an intermediate processor that has outputs - ``` - -2. **Missing Inputs** - ``` - CloneRepoActivity β†’ ApproveWorkflowActivity - ⚠️ ApproveWorkflowActivity accepts no inputs - Reason: Approval is a terminal activity (sink) - Solution: ApproveWorkflowActivity only works as final step - ``` - -3. **Type Mismatch** - ``` - LLMInferenceActivity (output: string) β†’ DeploymentPreCheckActivity (input: object) - ⚠️ String output cannot satisfy object input requirement - Reason: Incompatible data types - Solution: Use LLM transformation node to convert stringβ†’object - ``` - -4. **Semantic Incompatibility** - ``` - NotifyStatusActivity β†’ CloneRepoActivity - ⚠️ No logical connection between these activities - Reason: Notification cannot be input to clone operation - Solution: Ensure data flow makes semantic sense - ``` - -### Incompatibility Data Structure - -```json -{ - "incompatible_edges": [ - { - "source": "node-1", - "target": "node-2", - "reason": "Source activity produces no outputs", - "source_needs": "any output", - "target_needs": "path, depth", - "suggestion": "Use LLM transformation to map outputs to inputs" - } - ], - "disconnected_nodes": ["node-5", "node-8"], - "user_alerts": [ - "⚠️ node-1 β†’ node-2: Source activity produces no outputs. Use LLM transformation to map outputs to inputs", - "πŸ”Œ Node 'NotifyStatus-1' has no connections. Consider adding edges or removing it." - ] -} -``` - -## User Alerts - -### Alert Types - -1. **Incompatibility Warnings** (⚠️) - ``` - ⚠️ source β†’ target: reason. suggestion. - ``` - - Highlighted in red on canvas - - Shows in error sidebar - - Prevents workflow execution until fixed - -2. **Disconnection Warnings** (πŸ”Œ) - ``` - πŸ”Œ Node 'label' has no connections. Consider adding edges or removing it. - ``` - - Highlighted in yellow - - Nodes with no input/output edges - - May be valid (first step, last step) or indicate design error - -3. **Type Mismatch Info** (ℹ️) - ``` - ℹ️ To connect source β†’ target, use transformer to map: {source_outputs} β†’ {target_inputs} - ``` - - Suggestion to use intermediate LLM node - - Provides mapping information - -## Frontend Integration - -### Canvas UI Feedback - -When CanvasReasonerActivity returns incompatibilities: - -1. **Visual Markers** - - Incompatible suggested edges: ❌ red dashed line (don't auto-add) - - Disconnected nodes: ⚠️ yellow border - -2. **Sidebar Alerts** - ``` - 🚨 Connection Issues (3) - - ⚠️ CloneRepo β†’ ApproveWorkflow - Reason: ApproveWorkflow is terminal (no outputs) - Suggestion: Place ApproveWorkflow at end of workflow - - ⚠️ LLMInference β†’ DeploymentPreCheck - Reason: Type mismatch (string β‰  object) - Suggestion: Add LLM transformation node - - πŸ”Œ SecurityScan-1 has no incoming edges - Suggestion: Connect CloneRepo β†’ SecurityScan - ``` - -3. **User Actions** - - βœ… Accept suggestions (green edges) - - ❌ Reject incompatible edges - - πŸ”§ Add transformer nodes - - πŸ—‘οΈ Remove disconnected nodes - -### API Response Example - -```json -{ - "suggested_edges": [ - {"source": "clone-1", "target": "analyze-1"}, - {"source": "analyze-1", "target": "security-1"}, - {"source": "security-1", "target": "report-1"} - ], - "reasoning": "Standard code review workflow: clone β†’ analyze β†’ scan β†’ report", - "confidence": 0.92, - "incompatible_edges": [ - { - "source": "report-1", - "target": "approve-1", - "reason": "ReportGenerator has no outputs (terminal activity)", - "suggestion": "ApproveWorkflow can only be a final step" - } - ], - "disconnected_nodes": [], - "user_alerts": [ - "⚠️ report-1 β†’ approve-1: ReportGenerator has no outputs (terminal activity). ApproveWorkflow can only be a final step" - ] -} -``` - -## Knowledge Base Schema - -Each activity in `activity_knowledge_base.json` defines: - -```json -{ - "name": "CloneRepoActivity", - "inputs": { - "repo": {"type": "string", "required": true}, - "branch": {"type": "string", "required": false} - }, - "outputs": { - "path": {"type": "string"}, - "commit": {"type": "string"} - } -} -``` - -### Classification Rules - -- **Generator** (0 inputs): trigger, API call, schedule -- **Processor** (1+ inputs, 1+ outputs): analysis, transformation, scan -- **Sink** (1+ inputs, 0 outputs): notification, approval, archive -- **Bypass** (0 inputs, 0 outputs): rare - usually error - -## Common Patterns - -### βœ… Valid Chains - -``` -CloneRepo β†’ Analyze β†’ SecurityScan β†’ Report -(generator) β†’ (processor) β†’ (processor) β†’ (sink) -``` - -``` -Trigger β†’ LLMInference β†’ Decision β†’ (Branch: Notify OR Approve) -(gen) β†’ (processor) β†’ (processor) β†’ (sink) -``` - -### ❌ Invalid Chains - -``` -Notify β†’ CloneRepo ❌ -(sink) β†’ (generator) - backward flow - -CloneRepo β†’ CloneRepo β†’ Analyze ❌ -self-loop - no benefit - -Analyze β†’ Approve β†’ Notify ❌ -Approve is terminal (sink), can't output to Notify -``` - -## Edge Cases - -### Multiple Outputs β†’ Single Input -``` -SecurityScan β†’ Report -SecurityScan outputs: [issues, metrics, severity] -Report inputs: [report_data] - -LLM must infer: bundle all outputs into single report_data object -Confidence: 0.7 (requires transformation) -``` - -### Terminal Activities -- **ApproveWorkflowActivity**: Must be last (blocks workflow) -- **NotifyStatusActivity**: Can be mid-workflow (async notify) -- **ArchiveResultsActivity**: Should be last (persistence) - -### Data Transformation -When source outputs don't match target inputs: - -```python -# User can insert transformer node: -LLMInference β†’ [LLMTransformer] β†’ DeploymentPreCheck - -# Transformer: -# - Input: LLMInference.output (string) -# - Output: DeploymentPreCheck.requirements (object) -# - Action: Call LLM to convert format -``` - -## Testing Incompatibility Detection - -### Test Case 1: Terminal Activity as Source -```go -source := db.WorkflowNode{ID: "n1", Type: "notify-status", Label: "Notify"} -target := db.WorkflowNode{ID: "n2", Type: "clone-repo", Label: "Clone"} - -warnings := CheckConnectionCompatibility(source, target) -// Should warn: NotifyStatusActivity produces no outputs -``` - -### Test Case 2: Type Mismatch -```go -source := db.WorkflowNode{ID: "n1", Type: "llm-inference", ...} -target := db.WorkflowNode{ID: "n2", Type: "deployment-check", ...} - -warnings := CheckConnectionCompatibility(source, target) -// Should warn: string output β‰  object input -``` - -### Test Case 3: Disconnected Node -```go -nodes := []db.WorkflowNode{n1, n2, n3} -edges := []db.WorkflowEdge{{Source: "n1", Target: "n2"}} - -disconnected := IdentifyDisconnectedNodes(nodes, edges) -// Should return ["n3"] -``` - -## Future Enhancements - -1. **Automatic Transformer Insertion** - - Detect incompatibilities - - Auto-suggest LLM transformer nodes - - Chain transformers if needed - -2. **Confidence Scoring** - - Increase when types match perfectly - - Decrease for semantic mismatches - - Factor in activity dependencies - -3. **Learning from History** - - Track successful workflows - - Remember user edits to suggestions - - Improve LLM prompts over time - -4. **Multi-Path Analysis** - - Suggest multiple connection topologies - - Show cost/efficiency of each - - Rank by execution time/cost - -5. **Dry-Run Validation** - - Execute suggested workflow in simulation - - Catch runtime errors early - - Show data flow through each node diff --git a/docs/TEMPORAL_GRAPH_RAG_WIRING.md b/docs/TEMPORAL_GRAPH_RAG_WIRING.md deleted file mode 100644 index fed6341..0000000 --- a/docs/TEMPORAL_GRAPH_RAG_WIRING.md +++ /dev/null @@ -1,1013 +0,0 @@ -# Temporal.io + Graph RAG Integration for Workflows - -## Architecture - -``` -β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” -β”‚ Workflow API Request β”‚ -β”‚ POST /workflows/{id}/query β”‚ -β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”¬β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ - β”‚ {query, search_type, relation_type, version} - β–Ό -β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” -β”‚ Temporal Workflow Orchestratorβ”‚ -β”‚ - Start execution β”‚ -β”‚ - Route to activities β”‚ -β”‚ - Aggregate results β”‚ -β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”¬β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ - β”‚ - β”Œβ”€β”€β”€β”€β”€β”€β”΄β”€β”€β”€β”€β”€β”€β” - β–Ό β–Ό - β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” - β”‚ Activity 1 β”‚ β”‚ Activity 2 β”‚ - β”‚ Retrieve β”‚ β”‚ Query Graph RAG β”‚ - β”‚ Entities β”‚ β”‚ for Relations β”‚ - β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ - β”‚ β”‚ - β–Ό β–Ό - β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” - β”‚ Memory API β”‚ - β”‚ /memory/query/semantic β”‚ - β”‚ /memory/entities/{id}/v.. β”‚ - β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ - β”‚ - β–Ό - β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” - β”‚ Graph RAG (PostgreSQL) β”‚ - β”‚ - workflow_relations β”‚ - β”‚ - workflow_versions β”‚ - β”‚ - relation_changes β”‚ - β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ -``` - ---- - -## 1. Temporal Workflow Definition - -**File:** `statemachine/workflow_graph_query.go` - -```go -package statemachine - -import ( - "context" - "time" - "go.temporal.io/sdk/workflow" - "github.com/rockliang/poimen/workflows/action" - "github.com/rockliang/poimen/workflows/pkg/db" -) - -// WorkflowGraphQueryInput defines query parameters -type WorkflowGraphQueryInput struct { - WorkflowID string `json:"workflow_id"` - Query string `json:"query"` - SearchType string `json:"search_type"` // entities, edges, all - RelationType string `json:"relation_type"` // data-flow, dependency, etc - Version int `json:"version"` // canvas 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"` // default, recency, accuracy - IncludeReasoning bool `json:"include_reasoning"` -} - -// WorkflowGraphQueryOutput aggregates results -type WorkflowGraphQueryOutput struct { - WorkflowID string `json:"workflow_id"` - Query string `json:"query"` - Version int `json:"version"` - ExecutionTimeMs int64 `json:"execution_time_ms"` - Results []EdgeWithWording `json:"results"` - Paths []QueryPath `json:"paths"` - TotalCount int `json:"total_count"` - HasMore bool `json:"has_more"` - RankingProfile string `json:"ranking_profile"` -} - -type QueryPath struct { - SourceID string `json:"source_id"` - TargetID string `json:"target_id"` - PathCount int `json:"path_count"` - Distance int `json:"distance"` - Confidence float64 `json:"total_confidence"` - PathNodes []string `json:"node_ids"` -} - -// WorkflowGraphQuery is the main Temporal workflow -func WorkflowGraphQuery(ctx workflow.Context, input WorkflowGraphQueryInput) (WorkflowGraphQueryOutput, error) { - startTime := time.Now() - output := WorkflowGraphQueryOutput{ - WorkflowID: input.WorkflowID, - Query: input.Query, - Version: input.Version, - RankingProfile: input.RankingProfile, - Results: []EdgeWithWording{}, - } - - // Activity options - opts := workflow.ActivityOptions{ - StartToCloseTimeout: 120 * time.Second, - RetryPolicy: &temporal.RetryPolicy{ - InitialInterval: 2 * time.Second, - BackoffCoefficient: 2.0, - MaxInterval: 10 * time.Second, - MaxAttempts: 3, - }, - } - ctx = workflow.WithActivityOptions(ctx, opts) - - // Step 1: Fetch Canvas + Relations from Database - var canvasData action.CanvasWithRelationsData - err := workflow.ExecuteActivity( - ctx, - action.FetchCanvasRelationsActivity, - action.FetchCanvasRelationsInput{ - WorkflowID: input.WorkflowID, - Version: input.Version, - }, - ).Get(ctx, &canvasData) - if err != nil { - return output, err - } - - // Step 2: Query Graph RAG based on search_type - var graphResults action.GraphRAGQueryOutput - graphQuery := action.GraphRAGQueryInput{ - WorkflowID: input.WorkflowID, - Query: input.Query, - SearchType: input.SearchType, - RelationType: input.RelationType, - ConfidenceFloor: input.ConfidenceFloor, - TopK: input.TopK, - RankingProfile: input.RankingProfile, - Canvas: canvasData, - } - - err = workflow.ExecuteActivity( - ctx, - action.QueryGraphRAGActivity, - graphQuery, - ).Get(ctx, &graphResults) - if err != nil { - return output, err - } - - output.Results = graphResults.Edges - output.TotalCount = graphResults.TotalCount - output.HasMore = graphResults.HasMore - - // Step 3: If find_paths requested, calculate paths - if input.FindPaths && input.TargetNodeID != "" { - var pathResults action.PathfindingOutput - err = workflow.ExecuteActivity( - ctx, - action.FindRelationPathsActivity, - action.FindRelationPathsInput{ - WorkflowID: input.WorkflowID, - Version: input.Version, - SourceNodes: extractSourceNodes(graphResults.Edges), - TargetNodeID: input.TargetNodeID, - MaxDepth: input.MaxPathDepth, - Canvas: canvasData, - Relations: graphResults.Edges, - }, - ).Get(ctx, &pathResults) - if err != nil { - return output, err - } - output.Paths = pathResults.Paths - } - - // Step 4: If reasoning requested, explain results - if input.IncludeReasoning { - var reasoning action.ReasoningOutput - err = workflow.ExecuteActivity( - ctx, - action.ExplainGraphResultsActivity, - action.ExplainGraphResultsInput{ - Query: input.Query, - Results: output.Results, - Paths: output.Paths, - Confidence: calculateAverageConfidence(output.Results), - }, - ).Get(ctx, &reasoning) - if err != nil { - // Log but don't fail if reasoning fails - workflow.GetLogger(ctx).Warn("Reasoning failed", "error", err) - } - } - - output.ExecutionTimeMs = time.Since(startTime).Milliseconds() - return output, nil -} - -func extractSourceNodes(edges []EdgeWithWording) []string { - nodeSet := make(map[string]bool) - for _, edge := range edges { - nodeSet[edge.Source] = true - } - var nodes []string - for node := range nodeSet { - nodes = append(nodes, node) - } - return nodes -} - -func calculateAverageConfidence(edges []EdgeWithWording) float64 { - if len(edges) == 0 { - return 0.0 - } - sum := 0.0 - for _, edge := range edges { - sum += edge.RelationWording.Confidence - } - return sum / float64(len(edges)) -} -``` - ---- - -## 2. Activity: Fetch Canvas + Relations - -**File:** `action/fetch_canvas_relations.go` - -```go -package action - -import ( - "context" - "fmt" - "github.com/rockliang/poimen/workflows/pkg/db" -) - -type CanvasWithRelationsData struct { - WorkflowID string `json:"workflow_id"` - Version int `json:"version"` - Nodes []db.WorkflowNode `json:"nodes"` - Edges []EdgeWithWording `json:"edges"` - CreatedAt string `json:"created_at"` - Metadata map[string]interface{} `json:"metadata"` -} - -type FetchCanvasRelationsInput struct { - WorkflowID string `json:"workflow_id"` - Version int `json:"version"` -} - -// FetchCanvasRelationsActivity retrieves canvas + versioned relations -func FetchCanvasRelationsActivity(ctx context.Context, input FetchCanvasRelationsInput) (CanvasWithRelationsData, error) { - logger := newActivityLogger(ctx) - output := CanvasWithRelationsData{ - WorkflowID: input.WorkflowID, - Version: input.Version, - Nodes: []db.WorkflowNode{}, - Edges: []EdgeWithWording{}, - } - - logger.logf("info", "Fetching canvas relations for workflow %s version %d", input.WorkflowID, input.Version) - - // Get database connection (injected via Temporal context) - dbConn, err := getDBFromContext(ctx) - if err != nil { - return output, fmt.Errorf("failed to get DB connection: %w", err) - } - - // Fetch workflow version + canvas - var canvas db.Canvas - query := ` - SELECT workflow_id, version, nodes, edges, created_at, metadata - FROM workflow_versions - WHERE workflow_id = $1 AND version = $2 - ` - rows, err := dbConn.Query(ctx, query, input.WorkflowID, input.Version) - if err != nil { - return output, fmt.Errorf("failed to query workflow_versions: %w", err) - } - defer rows.Close() - - if !rows.Next() { - return output, fmt.Errorf("workflow version not found: %s v%d", input.WorkflowID, input.Version) - } - - // Parse canvas JSONB - if err := rows.Scan(&canvas.WorkflowID, &canvas.Version, &canvas.Nodes, &canvas.Edges, &canvas.CreatedAt, &canvas.Metadata); err != nil { - return output, fmt.Errorf("failed to scan canvas: %w", err) - } - - output.Nodes = canvas.Nodes - output.CreatedAt = canvas.CreatedAt.String() - - // Fetch relations (edges with wording) - relQuery := ` - SELECT id, source_node_id, target_node_id, relation_type, relation_label, - relation_wording, metadata, created_at - FROM workflow_relations - WHERE workflow_id = $1 AND version = $2 - ORDER BY created_at - ` - - relRows, err := dbConn.Query(ctx, relQuery, input.WorkflowID, input.Version) - if err != nil { - return output, fmt.Errorf("failed to query workflow_relations: %w", err) - } - defer relRows.Close() - - for relRows.Next() { - var edge EdgeWithWording - if err := relRows.Scan( - &edge.ID, - &edge.Source, - &edge.Target, - &edge.RelationType, - &edge.RelationLabel, - &edge.RelationWording, - &edge.Metadata, - &edge.CreatedAt, - ); err != nil { - logger.logf("warn", "Failed to scan relation: %v", err) - continue - } - output.Edges = append(output.Edges, edge) - } - - logger.logf("info", "Fetched %d nodes, %d relations", len(output.Nodes), len(output.Edges)) - return output, nil -} -``` - ---- - -## 3. Activity: Query Graph RAG - -**File:** `action/query_graph_rag.go` - -```go -package action - -import ( - "bytes" - "context" - "encoding/json" - "fmt" - "io" - "net/http" -) - -type GraphRAGQueryInput struct { - WorkflowID string `json:"workflow_id"` - Query string `json:"query"` - SearchType string `json:"search_type"` // entities, edges, all - RelationType string `json:"relation_type"` - ConfidenceFloor float64 `json:"confidence_floor"` - TopK int `json:"top_k"` - RankingProfile string `json:"ranking_profile"` - Canvas CanvasWithRelationsData `json:"canvas"` -} - -type GraphRAGQueryOutput struct { - WorkflowID string `json:"workflow_id"` - Query string `json:"query"` - Edges []EdgeWithWording `json:"results"` - TotalCount int `json:"total_count"` - HasMore bool `json:"has_more"` - ExecutionMs int64 `json:"execution_time_ms"` -} - -// QueryGraphRAGActivity calls Memory System unified query -func QueryGraphRAGActivity(ctx context.Context, input GraphRAGQueryInput) (GraphRAGQueryOutput, error) { - logger := newActivityLogger(ctx) - output := GraphRAGQueryOutput{ - WorkflowID: input.WorkflowID, - Query: input.Query, - Edges: []EdgeWithWording{}, - } - - logger.logf("info", "Querying Graph RAG: %s (type: %s, confidence: %.2f)", input.Query, input.SearchType, input.ConfidenceFloor) - - // Build unified query for Memory System - memoryQuery := buildMemoryUnifiedQuery(input) - - // Call Memory System /memory/query endpoint - payload, err := json.Marshal(memoryQuery) - if err != nil { - return output, fmt.Errorf("failed to marshal query: %w", err) - } - - // Get JWT token from context (set by worker) - token := ctx.Value("jwt_token").(string) - - req, err := http.NewRequestWithContext(ctx, "POST", "http://localhost:8080/memory/query", bytes.NewReader(payload)) - 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: 30 * time.Second} - resp, err := client.Do(req) - if err != nil { - return output, fmt.Errorf("failed to query Memory System: %w", err) - } - defer resp.Body.Close() - - if resp.StatusCode != 200 { - body, _ := io.ReadAll(resp.Body) - return output, fmt.Errorf("Memory System returned %d: %s", resp.StatusCode, string(body)) - } - - // Parse response - var memoryResp struct { - Results []map[string]interface{} `json:"results"` - TotalCount int `json:"total_count"` - HasMore bool `json:"has_more"` - ExecutionMs int64 `json:"execution_time_ms"` - } - - if err := json.NewDecoder(resp.Body).Decode(&memoryResp); err != nil { - return output, fmt.Errorf("failed to decode Memory response: %w", err) - } - - // Map results to edges with wording - for _, result := range memoryResp.Results { - edge := EdgeWithWording{} - - // Extract edge data from Memory result - if sourceID, ok := result["source_entity_id"].(string); ok { - edge.Source = sourceID - } - if targetID, ok := result["target_entity_id"].(string); ok { - edge.Target = targetID - } - if relType, ok := result["relation_type"].(string); ok { - edge.RelationType = relType - } - if fact, ok := result["fact"].(string); ok { - edge.RelationLabel = fact - } - if conf, ok := result["confidence"].(float64); ok { - edge.RelationWording.Confidence = conf - } - - output.Edges = append(output.Edges, edge) - } - - output.TotalCount = memoryResp.TotalCount - output.HasMore = memoryResp.HasMore - output.ExecutionMs = memoryResp.ExecutionMs - - logger.logf("info", "Graph RAG returned %d edges", len(output.Edges)) - return output, nil -} - -// buildMemoryUnifiedQuery constructs Memory System unified query format -func buildMemoryUnifiedQuery(input GraphRAGQueryInput) map[string]interface{} { - query := map[string]interface{}{ - "query": input.Query, - "search_type": "edges", // Default to edges for workflow relations - "confidence_floor": input.ConfidenceFloor, - "top_k": input.TopK, - "find_paths": input.RelationType != "", - "ranking_profile": input.RankingProfile, - "detect_communities": false, - "discover_facets": false, - } - - // Add relation type filter if specified - if input.RelationType != "" { - if query["facet_filters"] == nil { - query["facet_filters"] = map[string]interface{}{} - } - filters := query["facet_filters"].(map[string]interface{}) - filters["relation_type"] = input.RelationType - } - - return query -} -``` - ---- - -## 4. Activity: Find Relation Paths - -**File:** `action/find_relation_paths.go` - -```go -package action - -import ( - "context" - "fmt" -) - -type FindRelationPathsInput struct { - WorkflowID string `json:"workflow_id"` - Version int `json:"version"` - SourceNodes []string `json:"source_nodes"` - TargetNodeID string `json:"target_node_id"` - MaxDepth int `json:"max_depth"` - Canvas CanvasWithRelationsData `json:"canvas"` - Relations []EdgeWithWording `json:"relations"` -} - -type PathfindingOutput struct { - Paths []QueryPath `json:"paths"` -} - -// FindRelationPathsActivity computes shortest paths in relation graph -func FindRelationPathsActivity(ctx context.Context, input FindRelationPathsInput) (PathfindingOutput, error) { - logger := newActivityLogger(ctx) - output := PathfindingOutput{Paths: []QueryPath{}} - - if input.TargetNodeID == "" { - return output, fmt.Errorf("target_node_id required for pathfinding") - } - - logger.logf("info", "Finding paths to %s (max depth: %d)", input.TargetNodeID, input.MaxDepth) - - // Build adjacency list from relations - graph := buildRelationGraph(input.Relations) - - // BFS from each source to target - for _, sourceNode := range input.SourceNodes { - if sourceNode == input.TargetNodeID { - continue - } - - paths := bfsShortestPaths(graph, sourceNode, input.TargetNodeID, input.MaxDepth) - for _, path := range paths { - output.Paths = append(output.Paths, path) - } - } - - logger.logf("info", "Found %d paths", len(output.Paths)) - return output, nil -} - -func buildRelationGraph(relations []EdgeWithWording) map[string][]string { - graph := make(map[string][]string) - for _, edge := range relations { - graph[edge.Source] = append(graph[edge.Source], edge.Target) - } - return graph -} - -func bfsShortestPaths(graph map[string][]string, source, target string, maxDepth int) []QueryPath { - var paths []QueryPath - - queue := [][]string{{source}} - visited := make(map[string]bool) - visited[source] = true - - for len(queue) > 0 { - path := queue[0] - queue = queue[1:] - - if len(path) > maxDepth { - continue - } - - current := path[len(path)-1] - if current == target { - paths = append(paths, QueryPath{ - SourceID: source, - TargetID: target, - Distance: len(path) - 1, - PathCount: len(paths) + 1, - PathNodes: path, - }) - continue - } - - for _, neighbor := range graph[current] { - newPath := append([]string{}, path...) - newPath = append(newPath, neighbor) - queue = append(queue, newPath) - } - } - - return paths -} -``` - ---- - -## 5. API Handler: Wire Everything Together - -**File:** `internal/api/workflows_graph_query.go` - -```go -package api - -import ( - "context" - "encoding/json" - "net/http" - "time" - - "go.temporal.io/sdk/client" - "github.com/rockliang/poimen/workflows/statemachine" -) - -// QueryWorkflowGraph handles POST /workflows/{id}/query -func (s *Server) QueryWorkflowGraph(w http.ResponseWriter, r *http.Request) { - if r.Method != http.MethodPost { - http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) - return - } - - workflowID := r.PathValue("id") - if workflowID == "" { - http.Error(w, "Missing workflow ID", http.StatusBadRequest) - return - } - - // Parse request - var input statemachine.WorkflowGraphQueryInput - if err := json.NewDecoder(r.Body).Decode(&input); err != nil { - http.Error(w, "Invalid request body", http.StatusBadRequest) - return - } - - input.WorkflowID = workflowID - - // Set defaults - if input.SearchType == "" { - input.SearchType = "edges" - } - if input.TopK == 0 { - input.TopK = 10 - } - if input.ConfidenceFloor == 0 { - input.ConfidenceFloor = 0.5 - } - if input.MaxPathDepth == 0 { - input.MaxPathDepth = 3 - } - if input.RankingProfile == "" { - input.RankingProfile = "default" - } - - // Start Temporal workflow - temporalClient := s.TemporalClient // injected in NewServer - - ctx, cancel := context.WithTimeout(r.Context(), 30*time.Second) - defer cancel() - - we, err := temporalClient.ExecuteWorkflow( - ctx, - client.StartWorkflowOptions{ - ID: workflowID + "-query-" + time.Now().Format("20060102150405"), - TaskQueue: "workflow-tasks", - }, - statemachine.WorkflowGraphQuery, - input, - ) - if err != nil { - http.Error(w, "Failed to start workflow", http.StatusInternalServerError) - return - } - - // Wait for result - var result statemachine.WorkflowGraphQueryOutput - if err := we.Get(ctx, &result); err != nil { - http.Error(w, "Workflow failed", http.StatusInternalServerError) - return - } - - // Return result - w.Header().Set("Content-Type", "application/json") - w.WriteHeader(http.StatusOK) - json.NewEncoder(w).Encode(result) -} - -// GetWorkflowRelationVersions handles GET /workflows/{id}/relations/{edge_id}/versions -func (s *Server) GetWorkflowRelationVersions(w http.ResponseWriter, r *http.Request) { - if r.Method != http.MethodGet { - http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) - return - } - - workflowID := r.PathValue("id") - edgeID := r.PathValue("edge_id") - - if workflowID == "" || edgeID == "" { - http.Error(w, "Missing workflow or edge ID", http.StatusBadRequest) - return - } - - // Query database for edge versions - query := ` - SELECT 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 - ` - - rows, err := s.DB.Query(r.Context(), query, workflowID, edgeID) - if err != nil { - http.Error(w, "Failed to query versions", http.StatusInternalServerError) - return - } - defer rows.Close() - - type Version struct { - VersionNum int `json:"version_num"` - Operation string `json:"operation"` - Snapshot interface{} `json:"snapshot"` - ChangedAt string `json:"changed_at"` - ChangedBy string `json:"changed_by"` - FieldsChanged []string `json:"fields_changed"` - } - - versions := []Version{} - for rows.Next() { - v := Version{} - if err := rows.Scan(&v.VersionNum, &v.Operation, &v.Snapshot, &v.ChangedAt, &v.ChangedBy, &v.FieldsChanged); err != nil { - continue - } - versions = append(versions, v) - } - - response := map[string]interface{}{ - "edge_id": edgeID, - "workflow_id": workflowID, - "versions": versions, - "total_versions": len(versions), - "current_version": len(versions), - } - - w.Header().Set("Content-Type", "application/json") - json.NewEncoder(w).Encode(response) -} -``` - ---- - -## 6. Database Schema for Graph RAG - -**File:** `migrations/004_graph_rag_relations.sql` - -```sql --- Versioned relations with wording -CREATE TABLE workflow_relations ( - id UUID PRIMARY KEY DEFAULT gen_random_uuid(), - workflow_id UUID NOT NULL, - version INT NOT NULL, - source_node_id VARCHAR(255) NOT NULL, - target_node_id VARCHAR(255) NOT NULL, - relation_type VARCHAR(100), -- data-flow, dependency, conditional, parallel - relation_label TEXT, - relation_wording JSONB, -- {verb, source_output, target_input, confidence, etc} - metadata JSONB, - created_at TIMESTAMP DEFAULT NOW(), - - FOREIGN KEY (workflow_id, version) REFERENCES workflow_versions(workflow_id, version), - INDEX (workflow_id, version), - INDEX (relation_type), - UNIQUE (workflow_id, version, source_node_id, target_node_id) -); - --- Relation change history for versioning -CREATE TABLE workflow_relation_versions ( - id UUID PRIMARY KEY DEFAULT gen_random_uuid(), - workflow_id UUID NOT NULL, - edge_id UUID NOT NULL REFERENCES workflow_relations(id), - version_num INT NOT NULL, - operation VARCHAR(50), -- CREATE, UPDATE, DELETE - snapshot JSONB, -- Full relation state at this version - changed_at TIMESTAMP DEFAULT NOW(), - changed_by VARCHAR(255), - fields_changed TEXT[], - - INDEX (workflow_id, version_num), - UNIQUE (workflow_id, edge_id, version_num) -); - --- Graph RAG indexing metadata -CREATE TABLE workflow_rag_index ( - id UUID PRIMARY KEY DEFAULT gen_random_uuid(), - workflow_id UUID NOT NULL, - version INT NOT NULL, - indexed_entities JSONB, -- {node_ids, count} - indexed_edges JSONB, -- {edge_ids, count} - index_status VARCHAR(50), -- indexed, pending, failed - last_indexed_at TIMESTAMP, - embedding_model VARCHAR(100), - - INDEX (workflow_id, version) -); -``` - ---- - -## 7. Worker Configuration - -**File:** `cmd/server/main.go` (additions) - -```go -import ( - "context" - "go.temporal.io/sdk/client" - "go.temporal.io/sdk/worker" - "github.com/rockliang/poimen/workflows/statemachine" - "github.com/rockliang/poimen/workflows/action" -) - -func main() { - // ... existing setup ... - - // Temporal client - c, err := client.Dial(client.Options{ - HostPort: os.Getenv("TEMPORAL_HOST_PORT"), // localhost:7233 - }) - if err != nil { - log.Fatal(err) - } - defer c.Close() - - // Worker - w := worker.New(c, "workflow-tasks", worker.Options{}) - - // Register workflows - w.RegisterWorkflow(statemachine.WorkflowGraphQuery) - - // Register activities with DB connection in context - w.RegisterActivityWithOptions( - func(ctx context.Context, input interface{}) (interface{}, error) { - // Inject DB into context - ctx = context.WithValue(ctx, "db", dbConn) - ctx = context.WithValue(ctx, "jwt_token", os.Getenv("JWT_TOKEN")) - return nil, nil - }, - activity.RegisterOptions{Name: "setup"}, - ) - - w.RegisterActivity(action.FetchCanvasRelationsActivity) - w.RegisterActivity(action.QueryGraphRAGActivity) - w.RegisterActivity(action.FindRelationPathsActivity) - w.RegisterActivity(action.ExplainGraphResultsActivity) - - // Start worker - if err := w.Start(); err != nil { - log.Fatal(err) - } - defer w.Stop() - - // Register API handler - server := api.NewServer(dbConn, c) - http.HandleFunc("POST /workflows/{id}/query", server.QueryWorkflowGraph) - http.HandleFunc("GET /workflows/{id}/relations/{edge_id}/versions", server.GetWorkflowRelationVersions) - - log.Fatal(http.ListenAndServe(":8081", nil)) -} -``` - ---- - -## 8. Request/Response Flow Example - -### Request -```bash -curl -X POST http://localhost:8081/workflows/workflow-1/query \ - -H "Content-Type: application/json" \ - -d '{ - "query": "how does code analysis flow into security scanning", - "search_type": "edges", - "relation_type": "data-flow", - "version": 3, - "confidence_floor": 0.7, - "top_k": 10, - "find_paths": true, - "target_node_id": "security-scan-1", - "max_path_depth": 3, - "ranking_profile": "default", - "include_reasoning": true - }' -``` - -### Execution Flow (Temporal) - -1. **API Handler** β†’ Starts `WorkflowGraphQuery` workflow -2. **Workflow** β†’ Calls `FetchCanvasRelationsActivity` - - Query DB: `SELECT * FROM workflow_versions WHERE workflow_id=? AND version=?` - - Query DB: `SELECT * FROM workflow_relations WHERE workflow_id=? AND version=?` - - Return: Canvas + Relations data - -3. **Workflow** β†’ Calls `QueryGraphRAGActivity` - - Build unified query for Memory System - - POST to `/memory/query` with edges search - - Return: Ranked edges with wording - -4. **Workflow** β†’ If find_paths, calls `FindRelationPathsActivity` - - BFS pathfinding in relation graph - - Return: Query paths from sources to target - -5. **Workflow** β†’ If reasoning, calls `ExplainGraphResultsActivity` - - LLM explains why results match query - - Return: Reasoning narrative - -6. **Workflow** β†’ Aggregates results, returns to API -7. **API Handler** β†’ Returns 200 + JSON - -### Response -```json -{ - "workflow_id": "workflow-1", - "query": "how does code analysis flow into security scanning", - "version": 3, - "execution_time_ms": 245, - "results": [ - { - "id": "edge_analyze_scan", - "source": "analyze-code-1", - "target": "security-scan-1", - "relation_type": "data-flow", - "relation_label": "AnalyzeCode outputs metrics β†’ SecurityScan requires code structure", - "relation_wording": { - "verb": "provides-input-for", - "source_output": "metrics (object)", - "target_input": "path (string)", - "confidence": 0.85 - } - } - ], - "paths": [ - { - "source_id": "analyze-code-1", - "target_id": "security-scan-1", - "distance": 1, - "node_ids": ["analyze-code-1", "security-scan-1"] - } - ], - "total_count": 1, - "has_more": false, - "ranking_profile": "default" -} -``` - ---- - -## 9. Error Handling - -```go -// Temporal retry policy -RetryPolicy: &temporal.RetryPolicy{ - InitialInterval: 2 * time.Second, - BackoffCoefficient: 2.0, - MaxInterval: 10 * time.Second, - MaxAttempts: 3, -} - -// Memory System errors propagate -if resp.StatusCode != 200 { - return fmt.Errorf("Memory System %d: %s", resp.StatusCode, body) -} - -// Activity timeouts -StartToCloseTimeout: 120 * time.Second -``` - ---- - -## 10. Testing - -**File:** `tests/workflow_graph_query_test.go` - -```go -func TestWorkflowGraphQuery(t *testing.T) { - // Setup Temporal test - testSuite := &testsuite.WorkflowTestSuite{} - env := testSuite.NewTestWorkflowEnvironment() - - // Mock activities - env.OnActivity(action.FetchCanvasRelationsActivity, mock.MatchedBy(func(input interface{}) bool { - return true - })).Return(CanvasWithRelationsData{ - Nodes: []db.WorkflowNode{...}, - Edges: []EdgeWithWording{...}, - }, nil) - - env.OnActivity(action.QueryGraphRAGActivity, mock.MatchedBy(func(input interface{}) bool { - return true - })).Return(GraphRAGQueryOutput{ - Edges: []EdgeWithWording{...}, - TotalCount: 1, - }, nil) - - // Execute - env.ExecuteWorkflow(statemachine.WorkflowGraphQuery, input) - - // Assert - require.True(t, env.IsWorkflowCompleted()) - require.NoError(t, env.GetWorkflowError()) - - var result statemachine.WorkflowGraphQueryOutput - err := env.GetWorkflowResult(&result) - require.NoError(t, err) - require.Equal(t, 1, result.TotalCount) -} -``` - diff --git a/docs/WORKFLOWS_GRAPH_RAG_INTEGRATION.md b/docs/WORKFLOWS_GRAPH_RAG_INTEGRATION.md deleted file mode 100644 index 5ff8dd6..0000000 --- a/docs/WORKFLOWS_GRAPH_RAG_INTEGRATION.md +++ /dev/null @@ -1,712 +0,0 @@ -# Workflows + Graph RAG Integration - -Align Poimen Workflows with Poimen Memory System API for versioned relations, semantic queries, and intelligent canvas reasoning. - -**Key Features:** -- Versioned workflow relations following `/memory/entities/{id}/versions` pattern -- Semantic edge queries via `/workflows/{id}/query/semantic/edges` -- Relation wording as Facts (matching Memory System edges schema) -- Point-in-time canvas reconstruction via `?as_of=timestamp` -- Ranking profiles for relation importance -- Automatic Graph RAG indexing - ---- - -## Architecture - -``` -β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” -β”‚ Workflow Canvas (React Flow) β”‚ -β”‚ - Nodes (activities) β”‚ -β”‚ - Edges (connections) β”‚ -β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”¬β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ - β”‚ PUT /workflows/{id} - β–Ό -β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” -β”‚ CanvasReasonerActivity β”‚ -β”‚ - Suggest edges β”‚ -β”‚ - Validate compatibility β”‚ -β”‚ - Generate change reasoning β”‚ -β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”¬β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ - β”‚ suggested_edges + reasoning - β–Ό -β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” -β”‚ Workflows API Server β”‚ -β”‚ - Update canvas in DB β”‚ -β”‚ - Create version entry β”‚ -β”‚ - Store change metadata β”‚ -β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”¬β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ - β”‚ canvas_version, relations - β–Ό -β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” -β”‚ Graph RAG Backend β”‚ -β”‚ - Store versioned relations β”‚ -β”‚ - Index relation wording β”‚ -β”‚ - Enable semantic queries β”‚ -β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ -``` - ---- - -## Data Model - -### Workflow Canvas Version - -```sql --- In memory.workflows_versions (new table) -CREATE TABLE workflow_versions ( - id UUID PRIMARY KEY, - workflow_id UUID NOT NULL REFERENCES workflows(id), - customer_id UUID NOT NULL, - version INT NOT NULL, - canvas JSONB NOT NULL, -- {nodes, edges} - changed_by UUID, - change_reason TEXT, - change_type VARCHAR(50), -- 'manual', 'auto_reasoned', 'import' - reasoner_confidence FLOAT, - reasoning_metadata JSONB, -- LLM reasoning output - created_at TIMESTAMP DEFAULT NOW(), - - UNIQUE(workflow_id, version), - FOREIGN KEY(workflow_id, customer_id) - REFERENCES workflows(id, customer_id) -); - --- In memory.workflow_relations (replaces simple edges) -CREATE TABLE workflow_relations ( - id UUID PRIMARY KEY, - workflow_id UUID NOT NULL, - version INT NOT NULL, - source_node_id VARCHAR(255), - target_node_id VARCHAR(255), - relation_type VARCHAR(100), -- 'data-flow', 'dependency', 'conditional', 'parallel' - relation_label TEXT, -- Human-readable: "SecurityScan outputs issues β†’ Report inputs requirements" - relation_wording JSONB, -- {verb, object, context} - metadata JSONB, -- {source_output_type, target_input_type, compatibility_score} - created_at TIMESTAMP, - - FOREIGN KEY(workflow_id, version) - REFERENCES workflow_versions(id, version), - INDEX (workflow_id, version) -); - --- In memory.relation_changes (for Graph RAG indexing) -CREATE TABLE relation_changes ( - id UUID PRIMARY KEY, - workflow_id UUID, - version_from INT, - version_to INT, - change_type VARCHAR(50), -- 'added', 'removed', 'modified' - relation_id UUID REFERENCES workflow_relations(id), - source_node_id VARCHAR(255), - target_node_id VARCHAR(255), - old_wording JSONB, - new_wording JSONB, - change_reason TEXT, - change_timestamp TIMESTAMP, - reasoner_confidence FLOAT, - - INDEX (workflow_id, version_to) -); -``` - -### Relation Wording Schema - -```json -{ - "id": "edge-1", - "source": "clone-repo-1", - "target": "analyze-code-1", - "relation_type": "data-flow", - - "relation_label": "CloneRepo outputs path β†’ AnalyzeCode requires path", - - "relation_wording": { - "verb": "outputs", - "source_output": "path (string): Local filesystem path where repo was cloned", - "target_input": "path (string, required): Local filesystem path to analyze", - "connection_type": "direct-map", - "confidence": 0.98, - "notes": "Perfect type match between CloneRepo.path and AnalyzeCode.path" - }, - - "metadata": { - "source_activity": "CloneRepoActivity", - "target_activity": "AnalyzeCodeActivity", - "output_type": "string", - "input_type": "string", - "compatibility_score": 0.98, - "requires_transformation": false, - "semantic_match": "File path passes directly" - }, - - "change_history": [ - { - "version": 2, - "action": "added", - "reason": "LLM reasoner suggested data-flow connection", - "confidence": 0.98, - "timestamp": "2025-09-05T10:00:00Z" - } - ] -} -``` - ---- - -## Unified API Endpoints - -All workflow queries follow the unified endpoint pattern from Memory System. - -### 1. Unified Workflow Query - -**Endpoint:** `POST /workflows/{id}/query` - -This is the primary endpoint for all workflow canvas queries (replaces separate search endpoints). - -**Request:** -```json -{ - "query": "how does code analysis flow into security scanning", - "search_type": "edges|entities|all", - "version": 3, - "relation_type": "data-flow", - "confidence_floor": 0.7, - "top_k": 10, - "find_paths": true, - "target_node_id": "security-scan-1", - "max_path_depth": 3, - "ranking_profile": "default", - "include_reasoning": true -} -``` - -**Response (200 OK):** -```json -{ - "workflow_id": "workflow-1", - "query": "how does code analysis flow into security scanning", - "search_type": "edges", - "version": 3, - "execution_time_ms": 145, - "results": [ - { - "id": "edge_analyze_scan", - "source_node_id": "analyze-code-1", - "source_name": "AnalyzeCodeActivity", - "target_node_id": "security-scan-1", - "target_name": "SecurityScanActivity", - "relation_type": "data-flow", - "relation_label": "AnalyzeCode outputs metrics β†’ SecurityScan requires code structure", - "relation_wording": { - "verb": "provides-input-for", - "source_output": "metrics (object): Code quality and structural metrics", - "target_input": "path (string): Directory to scan", - "connection_type": "requires-transformer", - "confidence": 0.85, - "semantic_match": "Analysis metrics can guide security scan prioritization" - }, - "similarity_score": 0.92, - "confidence": 0.85, - "created_at": "2025-09-05T10:00:00Z", - "metadata": { - "source": "canvas://workflow-1:v3", - "tags": ["code-review", "security"] - } - } - ], - "paths": [ - { - "source_id": "analyze-code-1", - "target_id": "security-scan-1", - "path_count": 1, - "shortest_distance": 1, - "paths_found": [ - { - "node_ids": ["analyze-code-1", "security-scan-1"], - "relation_types": ["data-flow"], - "distance": 1, - "total_confidence": 0.85 - } - ] - } - ], - "total_count": 1, - "has_more": false, - "ranking_profile": "default" -} -``` - ---- - -### 2. Update Workflow Canvas (with Versioning) - -**Endpoint:** `PUT /workflows/{id}` - -**Request:** -```json -{ - "nodes": [...], - "edges": [...], - "auto_reason": true, - "change_reason": "User connected CloneRepo to AnalyzeCode", - "user_id": "uuid" -} -``` - -**Response:** -```json -{ - "id": "workflow-1", - "version": 3, - "canvas": { - "nodes": [...], - "edges": [...] - }, - "version_info": { - "version_number": 3, - "created_at": "2025-09-05T10:05:00Z", - "created_by": "user-uuid", - "change_reason": "User connected CloneRepo to AnalyzeCode", - "change_type": "manual" - }, - "relation_updates": { - "added": [ - { - "source": "clone-repo-1", - "target": "analyze-code-1", - "relation_wording": { - "verb": "outputs", - "source_output": "path: Local filesystem path where repo was cloned", - "target_input": "path (required): Local filesystem path to analyze", - "confidence": 0.98 - } - } - ], - "removed": [], - "modified": [] - } -} -``` - ---- - -### 2. Get Relation Version History (Memory System Pattern) - -**Endpoint:** `GET /workflows/{id}/relations/{edge_id}/versions` - -Follows `/memory/entities/{id}/versions` pattern from Memory System. - -**Response:** -```json -{ - "edge_id": "edge_1", - "workflow_id": "workflow-1", - "source": "clone-repo-1", - "target": "analyze-code-1", - "versions": [ - { - "version_num": 1, - "operation": "CREATE", - "snapshot": { - "relation_type": "data-flow", - "relation_label": "CloneRepo β†’ AnalyzeCode", - "relation_wording": { - "verb": "connects-to", - "confidence": 0.75 - } - }, - "changed_at": "2025-09-04T12:00:00Z", - "changed_by": "system", - "fields_changed": ["relation_type", "relation_wording"] - }, - { - "version_num": 2, - "operation": "UPDATE", - "snapshot": { - "relation_type": "data-flow", - "relation_label": "CloneRepo outputs path β†’ AnalyzeCode requires path", - "relation_wording": { - "verb": "outputs", - "source_output": "path (string)", - "target_input": "path (string, required)", - "confidence": 0.98 - } - }, - "changed_at": "2025-09-05T10:00:00Z", - "changed_by": "reasoner-activity", - "fields_changed": ["relation_wording", "relation_label"] - } - ], - "total_versions": 2, - "current_version": 2 -} -``` - ---- - -### 3. Edge Diff (Versioning Pattern) - -**Endpoint:** `POST /workflows/{id}/relations/diff` - -Follows `/memory/entities/diff` pattern. - -**Request:** -```json -{ - "edge_id": "edge_1", - "from_version": 1, - "to_version": 2 -} -``` - -**Response:** -```json -{ - "edge_id": "edge_1", - "from_version": 1, - "to_version": 2, - "source": "clone-repo-1", - "target": "analyze-code-1", - "diff": { - "added_fields": {}, - "removed_fields": {}, - "modified_fields": { - "relation_wording": { - "old": { - "verb": "connects-to", - "confidence": 0.75 - }, - "new": { - "verb": "outputs", - "source_output": "path (string)", - "target_input": "path (string, required)", - "confidence": 0.98 - } - } - } - }, - "change_timeline": [ - { - "version": 1, - "confidence": 0.75, - "changed_at": "2025-09-04T12:00:00Z" - }, - { - "version": 2, - "confidence": 0.98, - "changed_at": "2025-09-05T10:00:00Z" - } - ], - "editors_involved": ["system", "reasoner-activity"] -} -``` - ---- - -### 4. Point-in-Time Canvas (Versioning Pattern) - -**Endpoint:** `GET /workflows/{id}?at_version={v}` or `?as_of=2025-09-05T10:00:00Z` - -Follows `/memory/entities/at` pattern. - -**Response:** -```json -{ - "workflow_id": "workflow-1", - "version": 2, - "as_of_timestamp": "2025-09-05T10:00:00Z", - "canvas": { - "nodes": [...], - "edges": [...] - }, - "relations": [ - { - "id": "edge_1", - "source": "clone-repo-1", - "target": "analyze-code-1", - "relation_type": "data-flow", - "relation_label": "CloneRepo outputs path β†’ AnalyzeCode requires path", - "relation_wording": { - "verb": "outputs", - "source_output": "path (string)", - "target_input": "path (string, required)", - "confidence": 0.98 - }, - "version": 2 - } - ], - "metadata": { - "version_number": 2, - "created_at": "2025-09-05T10:00:00Z", - "changed_by": "reasoner-activity", - "change_reason": "LLM reasoner refined relation wording" - } -} -``` - ---- - -### 5. Bulk Import Canvas (with Relations) - -**Endpoint:** `POST /workflows/{id}/import` - -**Request:** -```json -{ - "canvas": { - "nodes": [...], - "edges": [...] - }, - "relations": [ - { - "source": "n1", - "target": "n2", - "relation_type": "data-flow", - "relation_wording": { - "verb": "outputs", - "source_output": "result (string)", - "target_input": "input (string, required)" - } - } - ], - "change_reason": "Imported from external workflow system" -} -``` - -**Response:** -```json -{ - "workflow_id": "workflow-1", - "version": 4, - "canvas": {...}, - "relations": {...}, - "import_metadata": { - "imported_nodes": 5, - "imported_edges": 4, - "validation_status": "success", - "indexing_status": "queued_for_graph_rag" - } -} -``` - ---- - -## Integration with CanvasReasonerActivity - -### Flow - -``` -User edits canvas - ↓ -PUT /workflows/{id} with auto_reason=true - ↓ -Backend calls CanvasReasonerActivity - ↓ -LLM suggests edges + reasoning - ↓ -Generate relation_wording from suggestions - ↓ -Create workflow_version entry - ↓ -Create workflow_relations entries - ↓ -Index relations in Graph RAG - ↓ -Response includes suggested_edges + relation_wording -``` - -### Response Structure - -```json -{ - "version": 3, - "suggested_edges": [ - { - "source": "clone-1", - "target": "analyze-1" - } - ], - "reasoning": "Standard code review workflow", - "confidence": 0.92, - "relation_wordings": [ - { - "source": "clone-1", - "target": "analyze-1", - "relation_wording": { - "verb": "outputs", - "source_output": "path (string): Cloned repository path", - "target_input": "path (string, required): Directory to analyze", - "connection_type": "direct-map", - "confidence": 0.98, - "semantic_description": "Repository path flows from clone operation to analysis" - } - } - ] -} -``` - ---- - -## Graph RAG Indexing - -### Relations Indexed - -Each workflow relation creates Graph RAG entities: - -``` -Node: { - id: "clone-repo-1", - type: "activity", - name: "CloneRepoActivity", - workflow_id: "workflow-1", - version: 3 -} - -Node: { - id: "analyze-code-1", - type: "activity", - ... -} - -Edge: { - id: "rel-1", - source: "clone-repo-1", - target: "analyze-code-1", - type: "data-flow", - label: "outputs path", - wording: {...}, - version: 3, - created_at: "2025-09-05T10:00:00Z" -} -``` - -### Semantic Queries - -Users can query like: - -- *"Which activities receive data from CloneRepo?"* -- *"What's the data flow from Analyze to Report?"* -- *"Which relations were added in version 3?"* -- *"Show me all type-mismatched connections"* -- *"Find workflows with Security Scan that require approval"* - ---- - -## Versioning Strategy - -### Version Numbering - -- Increment on every canvas change -- Track change_type: `manual`, `auto_reasoned`, `import`, `rag_query_applied` -- Store reasoner_confidence for auto changes - -### Relation Wording Versions - -- Each relation has independent wording history -- Confidence scores tracked per version -- Sources: user input, LLM reasoner, import, RAG query - -### Changelog - -```json -{ - "workflow_id": "workflow-1", - "total_versions": 5, - "changes": [ - { - "version": 1, - "type": "created", - "timestamp": "2025-09-04T10:00:00Z", - "user": "system", - "reason": "Workflow initialized" - }, - { - "version": 2, - "type": "auto_reasoned", - "timestamp": "2025-09-04T12:00:00Z", - "reasoner_confidence": 0.89, - "changes": { - "edges_added": 4, - "edges_modified": 0 - } - }, - { - "version": 3, - "type": "manual", - "timestamp": "2025-09-05T10:05:00Z", - "user": "user-uuid", - "reason": "Connected SecurityScan to Report manually" - } - ] -} -``` - ---- - -## Data Flow Example - -### Scenario: User drops SecurityScan node, system suggests connection - -1. **User Action:** Drops SecurityScan node onto existing workflow -2. **Frontend Call:** `PUT /workflows/{id}` with new nodes + `auto_reason=true` -3. **Backend:** - - Saves canvas to `workflow_versions` v.3 - - Calls CanvasReasonerActivity -4. **LLM Reasoning:** - - Analyzes: AnalyzeCode (outputs: quality, metrics) β†’ SecurityScan (inputs: path, depth) - - Suggests: Add transformer node OR use metrics for decision - - Confidence: 0.85 (type mismatch, requires transformation) -5. **Relation Wording:** - ```json - { - "source": "analyze-code-1", - "target": "security-scan-1", - "relation_wording": { - "verb": "provides-context-for", - "source_output": "quality (object): Code quality metrics", - "target_input": "path (string): Directory to scan", - "connection_type": "requires-transformer", - "reasoning": "Metrics inform which files to prioritize in scanning" - } - } - ``` -6. **Graph RAG:** Relations indexed automatically -7. **Response:** Frontend shows suggestions with natural language descriptions - ---- - -## Implementation Checklist - -- [ ] Create `workflow_versions` table in memory DB -- [ ] Create `workflow_relations` table with wording schema -- [ ] Create `relation_changes` table for tracking modifications -- [ ] Update CanvasReasonerActivity to generate relation wordings -- [ ] Implement versioned CRUD endpoints (PUT, GET, DIFF) -- [ ] Add Graph RAG indexing on relation creation -- [ ] Implement semantic query endpoint -- [ ] Add changelog view -- [ ] Test point-in-time reconstruction -- [ ] Document API in homelab-frontend/API.md - ---- - -## Relation Wording Language - -### Verbs (relation_type β†’ verb mapping) - -| Type | Verbs | Example | -|------|-------|---------| -| `data-flow` | outputs, inputs, receives, provides | "CloneRepo outputs path β†’ AnalyzeCode inputs path" | -| `dependency` | must-complete-before, depends-on, requires | "SecurityScan depends-on AnalyzeCode completion" | -| `conditional` | triggers-if, branches-on, routes-to | "ApproveWorkflow branches-on result approval" | -| `parallel` | runs-alongside, concurrent-with, independent-of | "Notify runs-alongside Report generation" | -| `transformation` | transforms, converts, maps, adapts | "LLMTransform converts metrics β†’ deployment plan" | - -### Confidence Scoring - -- **0.9-1.0:** Perfect match (type-compatible, direct data flow) -- **0.7-0.9:** Good match (semantic fit, minor transformation needed) -- **0.5-0.7:** Possible match (requires user confirmation) -- **<0.5:** Poor match (suggest removal or transformer) - diff --git a/k8s/git-commit.yaml b/k8s/git-commit.yaml index 95c3c60..7be31ea 100644 --- a/k8s/git-commit.yaml +++ b/k8s/git-commit.yaml @@ -9,6 +9,6 @@ metadata: app.kubernetes.io/name: poimen app.kubernetes.io/component: orchestrator data: - GIT_COMMIT: "eb4c4e989" # Updated automatically by CI/CD + GIT_COMMIT: "84b4ca120" # Updated automatically by CI/CD GIT_BRANCH: "main" DEPLOYMENT_DATE: "2026-09-05" diff --git a/k8s/worker-deployment.yaml b/k8s/worker-deployment.yaml index a28024d..71253e5 100644 --- a/k8s/worker-deployment.yaml +++ b/k8s/worker-deployment.yaml @@ -13,7 +13,7 @@ spec: labels: app: poimen-worker annotations: - git-commit: "eb4c4e989" # βœ… Updated on each push, triggers rolling restart + git-commit: "84b4ca120" # βœ… Updated on each push, triggers rolling restart deployment-date: "2026-09-05" spec: containers: