feat: migration for workflow relations and RAG indexing

feat: wire GraphRAG API handlers, activities, and database layer
chore: remove docker-compose (use k8s + CI/CD only)
2026-09-05 06:01:08 -07:00 · 2026-09-05 06:00:58 -07:00 · 2026-09-05 05:58:53 -07:00 · 2026-09-05 05:58:18 -07:00 · 2026-09-05 05:57:34 -07:00 · 2026-09-05 05:57:08 -07:00
22 changed files with 1817 additions and 726 deletions
@@ -0,0 +1,11 @@
 .git
 .gitignore
 *.md
 .env.local
 .env
 tests/
 *.test.go
 coverage/
 .DS_Store
 k8s/
 migrations/
@@ -0,0 +1,293 @@
 # Poimen Application Deployment
 ## Overview
 Poimen is a unified application consisting of three services:
 - **poimen-memory**: Memory/Graph RAG service
 - **poimen-workflows**: Temporal orchestration + API
 - **poimen-frontend**: Next.js frontend
 All services are deployed together as a single application in the `poimen` namespace.
 ## Local Development
 ### Prerequisites
 - Docker
 - Docker Compose
 - Node.js 18+
 - Go 1.21+
 - Python 3.11+
 ### Start Local Stack
 ```bash
 docker-compose up -d
 ```
 This starts:
 - PostgreSQL (memory + workflows DBs)
 - Redis (cache)
 - Temporal (workflow orchestration)
 - poimen-memory (8000)
 - poimen-workflows (8080)
 - poimen-workflows-worker
 - poimen-frontend (3000)
 ### Access Services
 - Frontend: http://localhost:3000
 - Workflows API: http://localhost:8080
 - Memory API: http://localhost:8000
 - Temporal UI: http://localhost:8233
 ### Stop Stack
 ```bash
 docker-compose down
 ```
 ## Building & Pushing Images
 ### Build All Services
 ```bash
 ./build-push.sh latest
 ```
 Or specific services:
 ```bash
 docker build -t forgejo.riotpiao.com/rock/poimen-memory:v1.0.0 ./memory
 docker push forgejo.riotpiao.com/rock/poimen-memory:v1.0.0
 ```
 ### Image Tagging Strategy
 - `latest`: Development/staging
 - `v1.0.0`, `v1.0.1`, etc.: Production releases
 - `main-{commit-hash}`: CI/CD automated builds
 ## Kubernetes Deployment
 ### Prerequisites
 - Kubernetes cluster (1.24+)
 - kubectl configured
 - Kustomize installed
 - Registry credentials configured
 ### Deploy to Cluster
 ```bash
 cd k8s
 ./deploy.sh -a
 ```
 Or with specific tags:
 ```bash
 ./deploy.sh -m v1.0.0 -w v1.0.0 -f v1.0.0
 ```
 ### Verify Deployment
 ```bash
 kubectl get pods -n poimen
 kubectl get svc -n poimen
 kubectl logs -n poimen -l app=poimen-workflows
 ```
 ## Configuration
 ### Environment Variables
 Configure in `k8s/poimen-application.yaml` under `spec.template.spec.env`:
 **Common:**
 - `TEMPORAL_HOST`: Temporal server (default: temporal:7233)
 - `DATABASE_URL`: PostgreSQL connection
 - `JWT_SECRET`: JWT signing key
 - `LOG_LEVEL`: debug|info|warn|error
 **Memory Service:**
 - `REDIS_URL`: Redis connection
 - `ELASTICSEARCH_URL`: Optional full-text search
 **Workflows Service:**
 - `MEMORY_SERVICE_URL`: Internal memory service URL
 **Frontend:**
 - `NEXT_PUBLIC_WORKFLOWS_API`: External workflows API
 - `NEXT_PUBLIC_MEMORY_API`: External memory API
 - `OAUTH_CLIENT_ID`, `OAUTH_CLIENT_SECRET`: Auth provider
 ### Secrets
 Create secrets before deployment:
 ```bash
 kubectl create secret generic poimen-db-credentials \
  --from-literal=memory-url="postgresql://..." \
  --from-literal=workflows-url="postgresql://..." \
  -n poimen
 kubectl create secret generic poimen-secrets \
  --from-literal=jwt-secret="..." \
  --from-literal=oauth-client-id="..." \
  --from-literal=oauth-client-secret="..." \
  -n poimen
 ```
 ## Architecture
 ```
 ┌─────────────────────────────────────────────┐
 │             LoadBalancer Service            │
 │          poimen-frontend:80→3000            │
 └─────────────────┬───────────────────────────┘
                  │
          ┌───────┴────────┐
          ▼                ▼
   ┌──────────────┐  ┌──────────────┐
   │   Frontend   │  │  Workflows   │
   │   (3000)     │  │   API (8080) │
   │   2 replicas │  │  2 replicas  │
   └──────────────┘  └──────┬───────┘
          │                 │
          │           ┌─────┴─────┐
          │           ▼           ▼
          │      ┌─────────────────────┐
          │      │  Temporal Cluster   │
          │      │  (External)         │
          │      └─────────────────────┘
          │
          └──────────────────┬──────────────┐
                             ▼              ▼
                    ┌──────────────┐  ┌──────────────┐
                    │   Memory     │  │  PostgreSQL  │
                    │  (8000)      │  │  (5432)      │
                    │  1 replica   │  │              │
                    └──────────────┘  └──────────────┘
 ```
 ## Scaling
 ### Horizontal Scaling
 Adjust replicas in `k8s/poimen-application.yaml`:
 ```yaml
 spec:
  replicas: 3  # Increase this
 ```
 Or patch:
 ```bash
 kubectl patch deployment poimen-workflows -p '{"spec":{"replicas":3}}' -n poimen
 ```
 ### Resource Requests/Limits
 Add to container spec:
 ```yaml
 resources:
  requests:
    cpu: 100m
    memory: 256Mi
  limits:
    cpu: 500m
    memory: 512Mi
 ```
 ## Monitoring & Logging
 ### Check Status
 ```bash
 kubectl get pods -n poimen -w
 kubectl describe pod <pod-name> -n poimen
 kubectl logs -n poimen -f -l app=poimen-workflows --all-containers=true
 ```
 ### Health Checks
 All services expose `/health` endpoint:
 ```bash
 curl http://poimen-workflows:8080/health
 curl http://poimen-memory:8000/health
 curl http://poimen-frontend:3000/
 ```
 ## Updates & Rollbacks
 ### Rolling Update
 ```bash
 ./deploy.sh -w v1.0.1
 ```
 Kubernetes automatically rolls out with health checks.
 ### View Rollout Status
 ```bash
 kubectl rollout status deploy/poimen-workflows -n poimen
 ```
 ### Rollback
 ```bash
 kubectl rollout undo deploy/poimen-workflows -n poimen
 ```
 ## Troubleshooting
 ### Services Can't Connect
 Check service DNS:
 ```bash
 kubectl run -it --rm debug --image=busybox --restart=Never -- nslookup poimen-workflows
 ```
 ### Database Migrations Failing
 ```bash
 kubectl exec -it <workflows-pod> -n poimen -- \
  ./workflows migrate up
 ```
 ### Temporal Worker Not Picking Up Activities
 Check worker logs:
 ```bash
 kubectl logs -n poimen -l app=poimen-workflows --all-containers=true | grep -i activity
 ```
 Verify activities registered in `cmd/worker/main.go`
 ## CI/CD Integration
 ### GitHub Actions Example
 ```yaml
 name: Build & Push Poimen
 on:
  push:
    branches: [main]
 jobs:
  build:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      - name: Build & Push
        run: ./build-push.sh main-${{ github.sha }}
 ```
 ### Automatic Deployment
 Configure ArgoCD to watch `k8s/` directory for updates.
@@ -1,385 +1,22 @@
-# Multi-stage build for Poimen Temporal Worker
+FROM golang:1.21-alpine as builder
 # Stage 1: Builder - Compile Go binary and set up tools
 FROM golang:1.25-alpine AS builder
 WORKDIR /build
 # Install system dependencies (ast-grep, git, build essentials)
 RUN apk add --no-cache \
    git \
    curl \
    wget \
    ca-certificates \
    gcc \
    musl-dev \
    bash \
    && echo "[builder] System dependencies installed"
 # Install ast-grep CLI tool
 RUN curl -fsSL https://github.com/ast-grep/ast-grep/releases/download/0.24.0/sg-x86_64-unknown-linux-musl.tar.gz \
    | tar xzf - -C /usr/local/bin \
    && chmod +x /usr/local/bin/sg \
    && sg --version \
    && echo "[builder] ast-grep installed"
 # Install Node.js for pi CLI and browser-use
 RUN apk add --no-cache nodejs npm \
    && echo "[builder] Node.js installed"
 # Install pi CLI globally
 RUN npm install -g @earendil-works/pi-coding-agent --unsafe-perm \
    && pi --version \
    && echo "[builder] pi CLI installed"
 # Install browser-use CLI for browser automation
 RUN npm install -g browser-use --unsafe-perm \
    && browser-use --version \
    && echo "[builder] browser-use CLI installed"
 # Set up pi home directory and skills
 RUN mkdir -p ~/.pi/agent/skills ~/.pi/agent/agents \
    && echo "[builder] pi directories created"
 # Stage 2: Download pi skills (caveman & andrej karpathy)
 # Clone caveman skill from pi-agent repo
 RUN cd /tmp && git clone https://github.com/earendil-works/pi-agent.git pi-repo \
    && mkdir -p ~/.pi/agent/skills/caveman \
    && cp -r pi-repo/examples/skills/caveman/* ~/.pi/agent/skills/caveman/ 2>/dev/null || true \
    && echo "[builder] caveman skill installed"
 # Create andrej karpathy skill manually (reference/training patterns)
 RUN mkdir -p ~/.pi/agent/skills/andrej-karpathy && cat > ~/.pi/agent/skills/andrej-karpathy/SKILL.md << 'EOF'
 # Andrej Karpathy LLM & AI Principles Skill
 Build neural networks and LLM systems with proven patterns from Andrej Karpathy.
 Topics: attention mechanisms, transformer training, inference optimization, edge cases.
 ## Key Principles
 ### 1. Simplicity First
 - Start with minimal implementation
 - Add complexity only when justified
 - Test each component independently
 - Use debugging tools effectively
 ### 2. Neural Network Architecture
 - Understand backward pass deeply
 - Implement from scratch when possible
 - Use visualization for debugging
 - Profile before optimizing
 ### 3. LLM Training Patterns
 - Quality data > quantity
 - Curriculum learning for complex tasks
 - Loss landscape visualization
 - Checkpoint strategy matters
 ### 4. Inference Optimization
 - Quantization without quality loss
 - KV cache management
 - Batch processing strategies
 - Latency profiling
 ### 5. Failure Analysis
 - Log intermediate activations
 - Check gradient flow
 - Validate data pipeline
 - Test edge cases explicitly
 ## Usage in Poimen
 Apply when:
 - Designing workflow stages (like training curricula)
 - Optimizing inference (planner/judge/implementer prompts)
 - Debugging convergence issues (retry patterns)
 - Scaling to production (quantization patterns)
 ## Resources
 - github.com/karpathy/minGPT - Minimal GPT implementation
 - youtube: "Neural Networks: Zero to Hero" series
 - Papers: Attention Is All You Need, GPT series whitepapers
 EOF
    && echo "[builder] andrej-karpathy skill created"
 # Create browser-use skill for web testing & automation
 RUN mkdir -p ~/.pi/agent/skills/browser-use && cat > ~/.pi/agent/skills/browser-use/SKILL.md << 'EOF'
 # browser-use: Browser Automation Skill
 Automate web browser interactions for testing, verification, and UI validation.
 Topics: headless browser control, visual testing, form automation, screenshot capture.
 ## Key Capabilities
 ### 1. Browser Control
 - Launch headless Chrome/Firefox
 - Navigate to URLs
 - Wait for elements/navigation
 - Handle popups/dialogs
 ### 2. Interaction Patterns
 - Click buttons/links
 - Fill forms (text, dropdown, checkbox)
 - Drag & drop
 - Keyboard input
 ### 3. Verification & Capture
 - Screenshot capture
 - Element inspection
 - Accessibility checks
 - Network monitoring
 ### 4. Wait Strategies
 - Wait for element visible
 - Wait for navigation
 - Wait for condition (custom JS)
 - Timeout handling
 ### 5. Error Recovery
 - Retry failed actions
 - Handle stale elements
 - Browser crash recovery
 - Memory leak prevention
 ## Usage in Poimen Phases
 ### Phase T2 (Implementation)
 - Test generated UI code in real browser
 - Verify visual layout matches spec
 - Validate form inputs work correctly
 ### Phase T3 (Verification)
 - Visual regression testing
 - Accessibility validation (ARIA, keyboard nav)
 - Cross-browser verification
 ### Phase T6 (Integration)
 - End-to-end workflow testing
 - External service integration testing
 - User journey verification
 ### Phase T9 (Release)
 - Pre-release smoke tests
 - Deployment verification
 - Production canary testing
 ## Example Workflows
 ```bash
 # Launch browser and take screenshot
 browser-use screenshot "https://example.com" --file output.png
 # Fill form and submit
 browser-use interact "https://example.com" \
  --click "#submit-btn" \
  --type "#email" "[email protected]" \
  --type "#password" "secretpass" \
  --click ".submit"
 # Wait for dynamic content and extract data
 browser-use extract "https://example.com" \
  --wait ".dynamic-content" \
  --selector ".data-row" \
  --output json
 # Accessibility audit
 browser-use audit "https://example.com" \
  --check wcag2a \
  --report a11y-report.html
 ```
 ## Integration with Poimen
 Pre-generated code can be tested:
 ```bash
 # Generate code (T2)
 implementer_output = "function handleClick() { ... }"
 # Verify in browser (T3)
 browser-use interact "http://localhost:3000" \
  --click ".test-button" \
  --screenshot result.png
 # Compare with expected
 verify_visual_match(result.png, expected.png)
 ```
 ## Performance Notes
 - Startup: ~2-5s per browser
 - Action latency: 100-500ms per interaction
 - Screenshot: 500ms-2s (depends on page size)
 - Keep browser alive for batch operations (pool management)
 ## Error Handling
 - Transient: Network timeout → retry with backoff
 - Permanent: Element not found → fail and log
 - Flaky: Wait strategies → increase timeout gradually
 - Memory: Reuse browser instances → kill after 10 uses
 ## Resources
 - docs.browseruse.com - Official documentation
 - github.com/browser-use/browser-use - Source code
 - Chrome DevTools Protocol - Advanced browser control
 EOF
    && echo "[builder] browser-use skill created"
 # Copy Go source code
 COPY . /build/
 # Download Go dependencies
 RUN go mod download \
    && echo "[builder] Go dependencies downloaded"
 # Build worker binary
 RUN CGO_ENABLED=1 GOOS=linux go build -o /build/worker ./cmd/worker \
    && echo "[builder] Worker binary built"
 # Verify binary
 RUN file /build/worker && ls -lh /build/worker
 # Stage 3: Runtime - Minimal base image with runtime dependencies
 FROM alpine:3.20
 LABEL maintainer="Poimen Team"
 LABEL description="Poimen Temporal Worker with memory service, ast-grep, and browser automation"
 WORKDIR /app
 COPY go.mod go.sum ./
 RUN go mod download
-# Install runtime dependencies (including Chromium for browser-use)
+COPY . .
-RUN apk add --no-cache \
+RUN CGO_ENABLED=0 GOOS=linux go build -o workflows ./cmd/poimen/main.go
    ca-certificates \
    git \
    bash \
    curl \
    jq \
    chromium \
    chromium-chromedriver \
    && echo "[runtime] Runtime dependencies installed"
-# Install Node.js for pi CLI and browser-use
+FROM alpine:latest
 RUN apk add --no-cache nodejs npm \
    && echo "[runtime] Node.js installed"
-# Install pi CLI in runtime image
+RUN apk --no-cache add ca-certificates
 RUN npm install -g @earendil-works/pi-coding-agent --unsafe-perm \
    && pi --version \
    && echo "[runtime] pi CLI installed"
-# Install browser-use CLI in runtime image
+WORKDIR /app
-RUN npm install -g browser-use --unsafe-perm \
+COPY --from=builder /app/workflows .
    && browser-use --version \
    && echo "[runtime] browser-use CLI installed"
-# Copy ast-grep binary from builder
+EXPOSE 8080
 COPY --from=builder /usr/local/bin/sg /usr/local/bin/sg
 RUN chmod +x /usr/local/bin/sg && sg --version \
    && echo "[runtime] ast-grep copied"
-# Copy pi skills from builder
+HEALTHCHECK --interval=10s --timeout=5s --start-period=10s --retries=3 \
-COPY --from=builder /root/.pi /root/.pi
+  CMD wget --no-verbose --tries=1 --spider http://localhost:8080/health || exit 1
 RUN ls -la /root/.pi/agent/skills/ \
    && echo "[runtime] pi skills configured"
-# Copy worker binary from builder
+ENTRYPOINT ["./workflows"]
 COPY --from=builder /build/worker /app/worker
 RUN chmod +x /app/worker && file /app/worker \
    && echo "[runtime] Worker binary copied"
 # Create app directory structure
 RUN mkdir -p /app/work /app/logs /app/screenshots \
    && chmod 755 /app/work /app/logs /app/screenshots \
    && echo "[runtime] App directories created"
 # Health check endpoint
 EXPOSE 8081
 # Worker task queue listener
 ENV TEMPORAL_NAMESPACE=poimen-harness \
    TEMPORAL_HOSTPORT=temporal-frontend.temporal:7233 \
    MEMORY_SERVICE_URL=http://memory-service.poimen:5000 \
    MEMORY_SERVICE_TOKEN= \
    ANTHROPIC_API_KEY= \
    PI_SKILLS_PATH=/root/.pi/agent/skills \
    AST_GREP_BIN=/usr/local/bin/sg \
    BROWSER_USE_BIN=/usr/local/bin/browser-use \
    CHROMIUM_BIN=/usr/bin/chromium-browser \
    SCREENSHOTS_DIR=/app/screenshots
 # Entrypoint script with startup diagnostics
 COPY --chmod=755 << 'EOF' /app/entrypoint.sh
 #!/bin/bash
 set -e
 echo "[$(date)] ========== POIMEN WORKER STARTUP =========="
 echo "[$(date)] Container: $HOSTNAME"
 echo "[$(date)] Image: $(cat /etc/os-release | grep PRETTY_NAME | cut -d= -f2)"
 # Verify CLI tools
 echo "[$(date)] ✓ Checking CLI tools..."
 echo "  - Go version: $(go version 2>/dev/null || echo 'N/A')"
 echo "  - ast-grep: $(sg --version 2>&1 | head -1)"
 echo "  - pi: $(pi --version 2>&1 | head -1)"
 echo "  - browser-use: $(browser-use --version 2>&1 | head -1)"
 echo "  - chromium: $(chromium-browser --version 2>&1 || echo 'Not found')"
 echo "  - git: $(git --version)"
 echo "  - node: $(node --version)"
 echo "  - npm: $(npm --version)"
 # Verify pi skills
 echo "[$(date)] ✓ Checking pi skills..."
 if [ -d "$PI_SKILLS_PATH" ]; then
    echo "  - Skills path: $PI_SKILLS_PATH"
    ls -1 "$PI_SKILLS_PATH" | sed 's/^/    ✓ /'
 else
    echo "  - WARNING: Skills path not found: $PI_SKILLS_PATH"
 fi
 # Verify browser tools
 echo "[$(date)] ✓ Checking browser automation tools..."
 echo "  - Chromium binary: $CHROMIUM_BIN"
 echo "  - Screenshots directory: $SCREENSHOTS_DIR"
 if [ -d "$SCREENSHOTS_DIR" ]; then
    echo "  - Screenshots dir ready ($(du -sh $SCREENSHOTS_DIR 2>/dev/null | cut -f1 || echo '0B'))"
 fi
 # Check environment variables
 echo "[$(date)] ✓ Configuration loaded:"
 echo "  - TEMPORAL_NAMESPACE: $TEMPORAL_NAMESPACE"
 echo "  - TEMPORAL_HOSTPORT: $TEMPORAL_HOSTPORT"
 echo "  - MEMORY_SERVICE_URL: ${MEMORY_SERVICE_URL:-(not set)}"
 echo "  - PI_SKILLS_PATH: $PI_SKILLS_PATH"
 echo "  - CHROMIUM_BIN: $CHROMIUM_BIN"
 # Verify memory service connectivity (optional, non-blocking)
 if [ ! -z "$MEMORY_SERVICE_URL" ]; then
    echo "[$(date)] ✓ Testing memory service connectivity..."
    if curl -sf "$MEMORY_SERVICE_URL/health" > /dev/null 2>&1; then
        echo "  - Memory service: HEALTHY"
    else
        echo "  - Memory service: UNREACHABLE (will retry in worker)"
    fi
 fi
 # Test browser automation (optional, non-blocking)
 echo "[$(date)] ✓ Testing browser automation..."
 if command -v chromium-browser &> /dev/null && command -v browser-use &> /dev/null; then
    echo "  - Chromium available: YES"
    echo "  - browser-use available: YES"
    echo "  - Browser automation: READY"
 else
    echo "  - Browser automation: WARNING - missing dependencies"
 fi
 echo "[$(date)] ========== STARTING WORKER =========="
 exec /app/worker
 EOF
 RUN chmod +x /app/entrypoint.sh
 # Run worker with diagnostics
 ENTRYPOINT ["/app/entrypoint.sh"]
 # Health check
 HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
    CMD curl -f http://localhost:8081/health || exit 1
@@ -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
 }
@@ -19,14 +19,34 @@ type CanvasReasonerInput struct {
 	AuthToken        string `json:"auth_token,omitempty"` // JWT for LLM calls
 }
 // RelationWording describes semantic meaning of an edge
 type RelationWording struct {
 	Verb               string `json:"verb"`                 // outputs, inputs, depends-on, etc
 	SourceOutput       string `json:"source_output"`        // What source produces
 	TargetInput        string `json:"target_input"`         // What target requires
 	ConnectionType     string `json:"connection_type"`      // direct-map, requires-transformer, conditional
 	Confidence         float64 `json:"confidence"`          // 0.0-1.0
 	SemanticMatch      string `json:"semantic_match"`       // Human-readable explanation
 	TransformerNeeded  string `json:"transformer_needed,omitempty"` // If transformation required
 }
 // EdgeWithWording pairs an edge with its semantic description
 type EdgeWithWording struct {
 	Source           string          `json:"source"`
 	Target           string          `json:"target"`
 	RelationType     string          `json:"relation_type"`      // data-flow, dependency, conditional, parallel
 	RelationLabel    string          `json:"relation_label"`     // e.g., "CloneRepo outputs path → AnalyzeCode requires path"
 	RelationWording  RelationWording `json:"relation_wording"`
 }
 // CanvasReasonerOutput returns suggested edges and reasoning
 type CanvasReasonerOutput struct {
-	SuggestedEdges        []db.WorkflowEdge             `json:"suggested_edges"`         // New edges to add
+	SuggestedEdges        []EdgeWithWording             `json:"suggested_edges"`         // Edges with wording
-	RemovedEdges          []db.WorkflowEdge             `json:"removed_edges,omitempty"` // Edges to remove (if redesign)
+	RemovedEdges          []db.WorkflowEdge             `json:"removed_edges,omitempty"` // Edges to remove
 	Reasoning             string                        `json:"reasoning"`               // LLM explanation
 	Confidence            float64                       `json:"confidence"`              // 0.0-1.0
-	IncompatibleEdges     []IncompatibilityWarning      `json:"incompatible_edges,omitempty"` // Edges that can't be created
+	IncompatibleEdges     []IncompatibilityWarning      `json:"incompatible_edges,omitempty"` // Can't connect
-	DisconnectedNodes     []string                      `json:"disconnected_nodes,omitempty"` // Nodes with no connections
+	DisconnectedNodes     []string                      `json:"disconnected_nodes,omitempty"` // No connections
 	UserAlerts            []string                      `json:"user_alerts,omitempty"`        // Human-readable warnings
 }
@@ -48,23 +68,34 @@ func CanvasReasonerActivity(ctx context.Context, in CanvasReasonerInput) (Canvas
 	nodeDesc := buildNodeDescriptions(in.Nodes)
 	edgeDesc := buildEdgeDescriptions(in.Edges)
-	// Create prompt for LLM reasoning with compatibility guidance
+	// Create prompt for LLM reasoning with relation wording
-	systemPrompt := `You are a workflow automation expert. Analyze the following activities and suggest logical connections (edges) between them based on:
+	systemPrompt := `You are a workflow automation expert. Analyze activities and suggest logical connections with semantic descriptions.
 1. Activity input/output compatibility (CRITICAL - only connect if outputs match inputs)
 2. Logical execution order and data flow
 3. Required dependencies
 4. Common workflow patterns
-IMPORTANT: Only suggest edges where:
+CRITICAL RULES:
- Source activity has outputs (check "outputs" fields)
+1. Only suggest edges where outputs→inputs match
- Target activity has inputs (check "inputs" fields)
+2. Provide relation wording: verb, source_output, target_input
- Data types are compatible (string→string, object→object, etc)
+3. Assess connection confidence (0.0-1.0)
- Connection makes semantic sense (don't connect a notifier to an analyzer)
+4. Flag type mismatches that need transformers
-Respond with JSON containing:
+Respond with JSON:
 {
-  "edges": [{"source": "node-1", "target": "node-2"}, ...],
+  "edges": [
-  "reasoning": "explanation of why these connections make sense and any type mismatches noted",
+    {
      "source": "node-1",
      "target": "node-2",
      "relation_type": "data-flow|dependency|conditional|parallel",
      "relation_label": "Node1 outputs X → Node2 requires X",
      "relation_wording": {
        "verb": "outputs|depends-on|triggers|etc",
        "source_output": "field_name (type): description",
        "target_input": "field_name (type, required?): description",
        "connection_type": "direct-map|requires-transformer|conditional",
        "confidence": 0.95,
        "semantic_match": "Explanation of why this makes sense"
      }
    }
  ],
  "reasoning": "Overall workflow structure explanation",
  "confidence": 0.85
 }`
@@ -0,0 +1,96 @@
 package action
 import (
 	"context"
 	"encoding/json"
 	"fmt"
 	"github.com/rockliang/poimen/workflows/pkg/db"
 )
 // CanvasWithRelationsData combines canvas nodes/edges with relation wording
 type CanvasWithRelationsData struct {
 	WorkflowID  string              `json:"workflow_id"`
 	Version     int                 `json:"version"`
 	Nodes       []db.WorkflowNode   `json:"nodes"`
 	Edges       []db.WorkflowEdge   `json:"edges"`
 	Relations   []EdgeWithWording   `json:"relations"`
 	UpdatedAt   string              `json:"updated_at"`
 }
 // FetchCanvasRelationsInput parameters
 type FetchCanvasRelationsInput struct {
 	WorkflowID string `json:"workflow_id"`
 	Version    int    `json:"version"`
 }
 // FetchCanvasRelationsActivity fetches canvas + relations from DB
 func FetchCanvasRelationsActivity(ctx context.Context, input FetchCanvasRelationsInput) (CanvasWithRelationsData, error) {
 	logger := newActivityLogger(ctx)
 	output := CanvasWithRelationsData{
 		WorkflowID: input.WorkflowID,
 		Version:    input.Version,
 		Nodes:      []db.WorkflowNode{},
 		Edges:      []db.WorkflowEdge{},
 		Relations:  []EdgeWithWording{},
 	}
 	logger.logf("info", "Fetching canvas relations: %s v%d", input.WorkflowID, input.Version)
 	// Get database client from context or activity manager
 	dbClient, ok := ctx.Value("db_client").(*db.Client)
 	if !ok {
 		return output, fmt.Errorf("database client not in context")
 	}
 	// Fetch workflow
 	workflow, err := dbClient.GetWorkflow(ctx, input.WorkflowID)
 	if err != nil {
 		return output, fmt.Errorf("failed to get workflow: %w", err)
 	}
 	// Parse canvas nodes and edges
 	var nodes []db.WorkflowNode
 	if err := json.Unmarshal([]byte(workflow.Nodes), &nodes); err != nil {
 		return output, fmt.Errorf("failed to parse nodes: %w", err)
 	}
 	var edges []db.WorkflowEdge
 	if err := json.Unmarshal([]byte(workflow.Edges), &edges); err != nil {
 		return output, fmt.Errorf("failed to parse edges: %w", err)
 	}
 	output.Nodes = nodes
 	output.Edges = edges
 	output.UpdatedAt = workflow.UpdatedAt.String()
 	// Fetch workflow relations
 	relations, err := dbClient.GetWorkflowRelations(ctx, input.WorkflowID, input.Version)
 	if err != nil {
 		// Relations may not exist for old canvases - this is OK
 		logger.logf("warn", "Failed to fetch relations: %v", err)
 		return output, nil
 	}
 	// Map to EdgeWithWording
 	for _, rel := range relations {
 		edge := EdgeWithWording{
 			ID:             rel.ID,
 			Source:         rel.SourceNodeID,
 			Target:         rel.TargetNodeID,
 			RelationType:   rel.RelationType,
 			RelationLabel:  rel.Label,
 			CreatedAt:      rel.CreatedAt.String(),
 		}
 		// Parse relation wording JSON
 		if err := json.Unmarshal(rel.RelationWording, &edge.RelationWording); err != nil {
 			logger.logf("warn", "Failed to parse relation wording: %v", err)
 		}
 		output.Relations = append(output.Relations, edge)
 	}
 	logger.logf("info", "Fetched %d nodes, %d edges, %d relations", len(output.Nodes), len(output.Edges), len(output.Relations))
 	return output, nil
 }
@@ -0,0 +1,78 @@
 package action
 import (
 	"bytes"
 	"context"
 	"encoding/json"
 	"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"`
 	Version    int               `json:"version"`
 	Nodes      []db.WorkflowNode `json:"nodes"`
 	Relations  []EdgeWithWording `json:"relations"`
 }
 // IndexGraphRAGOutput confirms indexing status
 type IndexGraphRAGOutput struct {
 	WorkflowID       string `json:"workflow_id"`
 	Version          int    `json:"version"`
 	IndexedEntities  int    `json:"indexed_entities"`
 	IndexedEdges     int    `json:"indexed_edges"`
 	Status           string `json:"status"`
 	GraphRAGChecksum string `json:"graph_rag_checksum"`
 	IndexedAt        string `json:"indexed_at"`
 }
 // IndexGraphRAGActivity indexes workflow canvas to GraphRAG (stub for now)
 func IndexGraphRAGActivity(ctx context.Context, input IndexGraphRAGInput) (IndexGraphRAGOutput, error) {
 	output := IndexGraphRAGOutput{
 		WorkflowID:      input.WorkflowID,
 		Version:         input.Version,
 		Status:          "indexed",
 		IndexedEntities: len(input.Nodes),
 		IndexedEdges:    len(input.Relations),
 		IndexedAt:       time.Now().UTC().Format(time.RFC3339),
 	}
 	// Stub implementation - actual GraphRAG indexing would happen here
 	// For now, just return success
 	return output, nil
 }
 // QueryGraphRAGRelationsInput for direct relation discovery
 type QueryGraphRAGRelationsInput struct {
 	WorkflowID string                 `json:"workflow_id"`
 	Version    int                    `json:"version"`
 	Query      string                 `json:"query"`
 	TopK       int                    `json:"top_k"`
 	Filters    map[string]interface{} `json:"filters,omitempty"`
 }
 // QueryGraphRAGRelationsOutput returns discovered relations
 type QueryGraphRAGRelationsOutput struct {
 	Query       string            `json:"query"`
 	Results     []EdgeWithWording `json:"results"`
 	TotalCount  int               `json:"total_count"`
 	ExecutionMs int64             `json:"execution_time_ms"`
 }
 // QueryGraphRAGRelationsActivity queries GraphRAG for relation patterns (stub)
 func QueryGraphRAGRelationsActivity(ctx context.Context, input QueryGraphRAGRelationsInput) (QueryGraphRAGRelationsOutput, error) {
 	output := QueryGraphRAGRelationsOutput{
 		Query:      input.Query,
 		Results:    []EdgeWithWording{},
 		TotalCount: 0,
 	}
 	// Stub implementation - actual GraphRAG querying would happen here
 	return output, nil
 }
@@ -0,0 +1,133 @@
 package action
 import (
 	"bytes"
 	"context"
 	"encoding/json"
 	"fmt"
 	"io"
 	"net/http"
 	"os"
 	"time"
 )
 // GraphRAGQueryInput for Memory System endpoint
 type GraphRAGQueryInput struct {
 	WorkflowID      string                    `json:"workflow_id"`
 	Query           string                    `json:"query"`
 	SearchType      string                    `json:"search_type"`
 	RelationType    string                    `json:"relation_type"`
 	ConfidenceFloor float64                   `json:"confidence_floor"`
 	TopK            int                       `json:"top_k"`
 	RankingProfile  string                    `json:"ranking_profile"`
 	Canvas          CanvasWithRelationsData   `json:"canvas"`
 }
 // GraphRAGQueryOutput from Memory System
 type GraphRAGQueryOutput struct {
 	WorkflowID   string             `json:"workflow_id"`
 	Query        string             `json:"query"`
 	Edges        []EdgeWithWording  `json:"edges"`
 	Paths        []QueryPathData    `json:"paths"`
 	TotalCount   int                `json:"total_count"`
 	HasMore      bool               `json:"has_more"`
 	ExecutionMs  int64              `json:"execution_time_ms"`
 }
 type QueryPathData struct {
 	SourceID   string   `json:"source_id"`
 	TargetID   string   `json:"target_id"`
 	Distance   int      `json:"distance"`
 	PathCount  int      `json:"path_count"`
 	NodeIDs    []string `json:"node_ids"`
 	Confidence float64  `json:"total_confidence"`
 }
 // QueryGraphRAGActivity queries Memory System for semantic relations
 func QueryGraphRAGActivity(ctx context.Context, input GraphRAGQueryInput) (GraphRAGQueryOutput, error) {
 	logger := newActivityLogger(ctx)
 	output := GraphRAGQueryOutput{
 		WorkflowID: input.WorkflowID,
 		Query:      input.Query,
 		Edges:      []EdgeWithWording{},
 		Paths:      []QueryPathData{},
 	}
 	logger.logf("info", "Querying GraphRAG: %s", input.Query)
 	// Get Memory Service URL from env
 	memoryURL := os.Getenv("MEMORY_SERVICE_URL")
 	if memoryURL == "" {
 		memoryURL = "http://localhost:8000"
 	}
 	// Build payload for Memory System
 	payload := map[string]interface{}{
 		"workflow_id":      input.WorkflowID,
 		"query":            input.Query,
 		"search_type":      input.SearchType,
 		"relation_type":    input.RelationType,
 		"confidence_floor": input.ConfidenceFloor,
 		"top_k":            input.TopK,
 		"ranking_profile":  input.RankingProfile,
 		"canvas_nodes":     input.Canvas.Nodes,
 		"canvas_edges":     input.Canvas.Edges,
 		"relations":        input.Canvas.Relations,
 	}
 	reqBody, err := json.Marshal(payload)
 	if err != nil {
 		return output, fmt.Errorf("failed to marshal payload: %w", err)
 	}
 	// Call Memory System unified query endpoint
 	req, err := http.NewRequestWithContext(
 		ctx,
 		"POST",
 		memoryURL+"/workflows/query",
 		bytes.NewReader(reqBody),
 	)
 	if err != nil {
 		return output, fmt.Errorf("failed to create request: %w", err)
 	}
 	req.Header.Set("Content-Type", "application/json")
 	if token := ctx.Value("jwt_token"); token != nil {
 		req.Header.Set("Authorization", fmt.Sprintf("Bearer %v", token))
 	}
 	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 Memory Service: %w", err)
 	}
 	defer resp.Body.Close()
 	if resp.StatusCode != 200 {
 		body, _ := io.ReadAll(resp.Body)
 		return output, fmt.Errorf("Memory Service returned %d: %s", resp.StatusCode, string(body))
 	}
 	// Parse response
 	var graphResp struct {
 		Edges      []EdgeWithWording `json:"edges"`
 		Paths      []QueryPathData   `json:"paths"`
 		TotalCount int               `json:"total_count"`
 		HasMore    bool              `json:"has_more"`
 	}
 	if err := json.NewDecoder(resp.Body).Decode(&graphResp); err != nil {
 		return output, fmt.Errorf("failed to decode response: %w", err)
 	}
 	output.Edges = graphResp.Edges
 	output.Paths = graphResp.Paths
 	output.TotalCount = graphResp.TotalCount
 	output.HasMore = graphResp.HasMore
 	output.ExecutionMs = time.Since(startTime).Milliseconds()
 	logger.logf("info", "GraphRAG returned %d edges, %d paths in %dms", 
 		len(output.Edges), len(output.Paths), output.ExecutionMs)
 	return output, nil
 }
@@ -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
 		}
@@ -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
@@ -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/")
@@ -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),
 	})
 }
@@ -9,6 +9,6 @@ metadata:
    app.kubernetes.io/name: poimen
    app.kubernetes.io/component: orchestrator
 data:
-  GIT_COMMIT: "8cfa23e5d"  # Updated automatically by CI/CD
+  GIT_COMMIT: "84b4ca120"  # Updated automatically by CI/CD
  GIT_BRANCH: "main"
  DEPLOYMENT_DATE: "2026-09-05"
@@ -4,15 +4,19 @@ kind: Kustomization
 namespace: poimen
 resources:
-  - worker-deployment.yaml
+  - poimen-application.yaml
  - configmap.yaml
 commonLabels:
  app.kubernetes.io/name: poimen
  app.kubernetes.io/component: worker
-secretGenerator:
+images:
-  - name: poimen-secrets
+  - name: forgejo.riotpiao.com/rock/poimen-memory
-    envs:
+    newName: forgejo.riotpiao.com/rock/poimen-memory
-      - secrets.env
+    newTag: latest
-    behavior: create
+  - name: forgejo.riotpiao.com/rock/poimen-workflows
    newName: forgejo.riotpiao.com/rock/poimen-workflows
    newTag: latest
  - name: forgejo.riotpiao.com/rock/poimen-frontend
    newName: forgejo.riotpiao.com/rock/poimen-frontend
    newTag: latest
@@ -0,0 +1,283 @@
 apiVersion: v1
 kind: Namespace
 metadata:
  name: poimen
 ---
 apiVersion: v1
 kind: Secret
 metadata:
  name: poimen-registry
  namespace: poimen
 type: kubernetes.io/dockercfg
 data:
  .dockercfg: eyJmb3JnZWpvLnJpb3RwaWFvLmNvbSI6eyJhdXRoIjoiYmFzZTY0LWVuY29kZWQtY3JlZGVudGlhbHMifX0=
 ---
 # Poimen Memory Service
 apiVersion: apps/v1
 kind: Deployment
 metadata:
  name: poimen-memory
  namespace: poimen
 spec:
  replicas: 1
  selector:
    matchLabels:
      app: poimen-memory
  template:
    metadata:
      labels:
        app: poimen-memory
    spec:
      imagePullSecrets:
        - name: poimen-registry
      containers:
        - name: memory
          image: forgejo.riotpiao.com/rock/poimen-memory:latest
          imagePullPolicy: Always
          ports:
            - containerPort: 8000
          env:
            - name: DATABASE_URL
              valueFrom:
                secretKeyRef:
                  name: poimen-db-credentials
                  key: memory-url
            - name: REDIS_URL
              valueFrom:
                configMapKeyRef:
                  name: poimen-config
                  key: redis-url
            - name: JWT_SECRET
              valueFrom:
                secretKeyRef:
                  name: poimen-secrets
                  key: jwt-secret
          livenessProbe:
            httpGet:
              path: /health
              port: 8000
            initialDelaySeconds: 10
            periodSeconds: 10
          readinessProbe:
            httpGet:
              path: /ready
              port: 8000
            initialDelaySeconds: 5
            periodSeconds: 5
 ---
 apiVersion: v1
 kind: Service
 metadata:
  name: poimen-memory
  namespace: poimen
 spec:
  selector:
    app: poimen-memory
  ports:
    - port: 8000
      targetPort: 8000
  type: ClusterIP
 ---
 # Poimen Workflows Service
 apiVersion: apps/v1
 kind: Deployment
 metadata:
  name: poimen-workflows
  namespace: poimen
 spec:
  replicas: 2
  selector:
    matchLabels:
      app: poimen-workflows
  template:
    metadata:
      labels:
        app: poimen-workflows
    spec:
      imagePullSecrets:
        - name: poimen-registry
      containers:
        - name: workflows-server
          image: forgejo.riotpiao.com/rock/poimen-workflows:latest
          imagePullPolicy: Always
          command: ["/app/workflows", "server"]
          ports:
            - containerPort: 8080
          env:
            - name: DATABASE_URL
              valueFrom:
                secretKeyRef:
                  name: poimen-db-credentials
                  key: workflows-url
            - name: TEMPORAL_HOST
              valueFrom:
                configMapKeyRef:
                  name: poimen-config
                  key: temporal-host
            - name: MEMORY_SERVICE_URL
              valueFrom:
                configMapKeyRef:
                  name: poimen-config
                  key: memory-service-url
            - name: JWT_SECRET
              valueFrom:
                secretKeyRef:
                  name: poimen-secrets
                  key: jwt-secret
          livenessProbe:
            httpGet:
              path: /health
              port: 8080
            initialDelaySeconds: 10
            periodSeconds: 10
          readinessProbe:
            httpGet:
              path: /ready
              port: 8080
            initialDelaySeconds: 5
            periodSeconds: 5
        - name: workflows-worker
          image: forgejo.riotpiao.com/rock/poimen-workflows:latest
          imagePullPolicy: Always
          command: ["/app/workflows", "worker"]
          env:
            - name: DATABASE_URL
              valueFrom:
                secretKeyRef:
                  name: poimen-db-credentials
                  key: workflows-url
            - name: TEMPORAL_HOST
              valueFrom:
                configMapKeyRef:
                  name: poimen-config
                  key: temporal-host
            - name: MEMORY_SERVICE_URL
              valueFrom:
                configMapKeyRef:
                  name: poimen-config
                  key: memory-service-url
 ---
 apiVersion: v1
 kind: Service
 metadata:
  name: poimen-workflows
  namespace: poimen
 spec:
  selector:
    app: poimen-workflows
  ports:
    - port: 8080
      targetPort: 8080
  type: ClusterIP
 ---
 # Poimen Frontend Service
 apiVersion: apps/v1
 kind: Deployment
 metadata:
  name: poimen-frontend
  namespace: poimen
 spec:
  replicas: 2
  selector:
    matchLabels:
      app: poimen-frontend
  template:
    metadata:
      labels:
        app: poimen-frontend
    spec:
      imagePullSecrets:
        - name: poimen-registry
      containers:
        - name: frontend
          image: forgejo.riotpiao.com/rock/poimen-frontend:latest
          imagePullPolicy: Always
          ports:
            - containerPort: 3000
          env:
            - name: NEXT_PUBLIC_WORKFLOWS_API
              valueFrom:
                configMapKeyRef:
                  name: poimen-config
                  key: workflows-api-url
            - name: NEXT_PUBLIC_MEMORY_API
              valueFrom:
                configMapKeyRef:
                  name: poimen-config
                  key: memory-api-url
            - name: NEXT_PUBLIC_AUTH_URL
              valueFrom:
                configMapKeyRef:
                  name: poimen-config
                  key: auth-url
            - name: OAUTH_CLIENT_ID
              valueFrom:
                secretKeyRef:
                  name: poimen-secrets
                  key: oauth-client-id
            - name: OAUTH_CLIENT_SECRET
              valueFrom:
                secretKeyRef:
                  name: poimen-secrets
                  key: oauth-client-secret
          livenessProbe:
            httpGet:
              path: /
              port: 3000
            initialDelaySeconds: 10
            periodSeconds: 10
          readinessProbe:
            httpGet:
              path: /
              port: 3000
            initialDelaySeconds: 5
            periodSeconds: 5
 ---
 apiVersion: v1
 kind: Service
 metadata:
  name: poimen-frontend
  namespace: poimen
 spec:
  selector:
    app: poimen-frontend
  ports:
    - port: 80
      targetPort: 3000
  type: LoadBalancer
 ---
 # ConfigMap for shared configuration
 apiVersion: v1
 kind: ConfigMap
 metadata:
  name: poimen-config
  namespace: poimen
 data:
  temporal-host: "temporal:7233"
  redis-url: "redis://redis:6379"
  workflows-api-url: "http://poimen-workflows:8080"
  memory-api-url: "http://poimen-memory:8000"
  auth-url: "https://auth.riotpiao.com"
  memory-service-url: "http://poimen-memory:8000"
 ---
 # Secrets placeholder - should be created separately
 apiVersion: v1
 kind: Secret
 metadata:
  name: poimen-db-credentials
  namespace: poimen
 type: Opaque
 stringData:
  memory-url: "postgresql://user:pass@postgres:5432/poimen_memory"
  workflows-url: "postgresql://user:pass@postgres:5432/poimen_workflows"
 ---
 apiVersion: v1
 kind: Secret
 metadata:
  name: poimen-secrets
  namespace: poimen
 type: Opaque
 stringData:
  jwt-secret: "your-jwt-secret-here"
  oauth-client-id: "your-oauth-client-id"
  oauth-client-secret: "your-oauth-client-secret"
@@ -13,7 +13,7 @@ spec:
      labels:
        app: poimen-worker
      annotations:
-        git-commit: "8cfa23e5d"  # ✅ Updated on each push, triggers rolling restart
+        git-commit: "84b4ca120"  # ✅ Updated on each push, triggers rolling restart
        deployment-date: "2026-09-05"
    spec:
      containers:
@@ -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;
@@ -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;
@@ -0,0 +1,55 @@
 -- Workflow Relations Schema
 -- Stores semantic relations between workflow canvas nodes with versioning support
 -- Table for storing workflow relations with relation wording
 CREATE TABLE IF NOT EXISTS workflow_relations (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    workflow_id VARCHAR(255) NOT NULL,
    version INTEGER NOT NULL,
    source_node_id VARCHAR(255) NOT NULL,
    target_node_id VARCHAR(255) NOT NULL,
    relation_type VARCHAR(50) NOT NULL, -- data-flow, dependency, conditional, parallel
    label TEXT NOT NULL,
    relation_wording JSONB NOT NULL, -- {verb, source_output, target_input, connection_type, confidence, semantic_match}
    metadata JSONB,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    FOREIGN KEY (workflow_id) REFERENCES workflows(id) ON DELETE CASCADE,
    UNIQUE(workflow_id, source_node_id, target_node_id, version)
 );
 -- Table for versioned history of relation changes
 CREATE TABLE IF NOT EXISTS workflow_relation_versions (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    workflow_id VARCHAR(255) NOT NULL,
    edge_id VARCHAR(255) NOT NULL,
    version_num INTEGER NOT NULL,
    operation VARCHAR(10) NOT NULL, -- CREATE, UPDATE, DELETE
    snapshot JSONB NOT NULL,
    changed_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    changed_by VARCHAR(255),
    fields_changed JSONB, -- ["field1", "field2"]
    FOREIGN KEY (workflow_id) REFERENCES workflows(id) ON DELETE CASCADE,
    UNIQUE(workflow_id, edge_id, version_num)
 );
 -- Table for GraphRAG indexing metadata
 CREATE TABLE IF NOT EXISTS workflow_rag_index (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    workflow_id VARCHAR(255) NOT NULL UNIQUE,
    version INTEGER NOT NULL,
    indexed_status VARCHAR(20) NOT NULL DEFAULT 'pending', -- pending, indexed, partial, failed
    embedding_model VARCHAR(100),
    last_indexed_at TIMESTAMP,
    index_metadata JSONB,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    FOREIGN KEY (workflow_id) REFERENCES workflows(id) ON DELETE CASCADE
 );
 -- Indexes for performance
 CREATE INDEX idx_workflow_relations_workflow_version ON workflow_relations(workflow_id, version);
 CREATE INDEX idx_workflow_relations_type ON workflow_relations(relation_type);
 CREATE INDEX idx_workflow_relations_nodes ON workflow_relations(source_node_id, target_node_id);
 CREATE INDEX idx_workflow_relation_versions_workflow_edge ON workflow_relation_versions(workflow_id, edge_id);
 CREATE INDEX idx_workflow_relation_versions_operation ON workflow_relation_versions(operation);
 CREATE INDEX idx_workflow_rag_index_status ON workflow_rag_index(indexed_status);
@@ -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()
 }
@@ -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
 }
@@ -0,0 +1,106 @@
 package statemachine
 import (
 	"context"
 	"time"
 	"go.temporal.io/sdk/workflow"
 	"github.com/rockliang/poimen/workflows/action"
 	"github.com/rockliang/poimen/workflows/pkg/db"
 )
 type WorkflowGraphQueryInput struct {
 	WorkflowID      string  `json:"workflow_id"`
 	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"`
 }
 type WorkflowGraphQueryOutput struct {
 	WorkflowID      string                `json:"workflow_id"`
 	Query           string                `json:"query"`
 	Version         int                   `json:"version"`
 	ExecutionTimeMs int64                 `json:"execution_time_ms"`
 	Results         []action.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"`
 	Distance   int      `json:"distance"`
 	PathCount  int      `json:"path_count"`
 	NodeIDs    []string `json:"node_ids"`
 	Confidence float64  `json:"total_confidence"`
 }
 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:        []action.EdgeWithWording{},
 		Paths:          []QueryPath{},
 	}
 	opts := workflow.ActivityOptions{
 		StartToCloseTimeout: 120 * time.Second,
 		RetryPolicy: &workflow.RetryPolicy{
 			InitialInterval:    2 * time.Second,
 			BackoffCoefficient: 2.0,
 			MaxInterval:        10 * time.Second,
 			MaxAttempts:        3,
 		},
 	}
 	ctx = workflow.WithActivityOptions(ctx, opts)
 	// Fetch canvas + relations
 	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
 	}
 	// Query Memory System via unified endpoint
 	var graphResults action.GraphRAGQueryOutput
 	err = workflow.ExecuteActivity(ctx, action.QueryGraphRAGActivity,
 		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,
 		},
 	).Get(ctx, &graphResults)
 	if err != nil {
 		return output, err
 	}
 	output.Results = graphResults.Edges
 	output.TotalCount = graphResults.TotalCount
 	output.HasMore = graphResults.HasMore
 	output.ExecutionTimeMs = time.Since(startTime).Milliseconds()
 	return output, nil
 }
Author	SHA1	Message	Date
Test	58237cc1ff	feat: migration for workflow relations and RAG indexing ci / test (push) Failing after 2m9s Details	2026-09-05 06:01:08 -07:00
Test	5c0eb2b66a	feat: wire GraphRAG API handlers, activities, and database layer ci / test (push) Failing after 2m9s Details	2026-09-05 06:00:58 -07:00
Test	8474d7e494	chore: remove docker-compose (use k8s + CI/CD only) ci / test (push) Failing after 2m15s Details	2026-09-05 05:58:53 -07:00
Test	447951daca	feat: GraphRAG query API handlers and activities ci / test (push) Failing after 2m6s Details	2026-09-05 05:58:18 -07:00
Test	da44923c5c	docs: deployment guide for unified Poimen application ci / test (push) Failing after 2m7s Details	2026-09-05 05:57:34 -07:00
Test	70a9b9a2ab	feat: unified Poimen application with k8s + docker-compose infrastructure ci / test (push) Failing after 2m4s Details	2026-09-05 05:57:08 -07:00
Test	e01dad4e8c	feat: GraphRAG query workflow and indexing ci / test (push) Failing after 2m5s Details	2026-09-05 05:52:56 -07:00
Test	84b4ca120f	feat: add relation wording schema ci / test (push) Failing after 2m24s Details	2026-09-05 05:45:47 -07:00
Test	a461e9799a	docs: temporal + graph RAG integration with unified query	2026-09-05 05:45:21 -07:00