ci: fix golang runner - use go 1.26, remove -mod=readonly, add go mod tidy

- Add pkg/db models and CRUD methods for workflows
- Add internal/routing canvas validator (DAG check, connectivity)
- Add internal/routing canvas converter (Canvas → WorkflowSpec)
- Register LLMInferenceActivity and LLMBatchInferenceActivity
- Update api/server and cmd/server with database integration
- Add K8s environment variable support
- Update activity knowledge base with LLM activities
- Add .env.example configuration template

.env.example

@@ -0,0 +1,22 @@
+# Workflows Backend Configuration
+
+# Database (memory-db CNPG in K8s)
+# Option A: Direct DATABASE_URL
+DATABASE_URL=postgresql://app:[email protected]:5432/memory?sslmode=disable
+
+# Option B: Individual env vars (used if DATABASE_URL is empty)
+DATABASE_HOST=memory-db-rw.poimen.svc.cluster.local
+DATABASE_PORT=5432
+DATABASE_NAME=memory
+DATABASE_USER=app
+DATABASE_PASSWORD=PASSWORD
+
+# Temporal
+TEMPORAL_HOST_PORT=localhost:7233
+TEMPORAL_NAMESPACE=default
+
+# API Server
+API_PORT=8080
+
+# Logging
+VERBOSE=false

.gitea/workflows/build-push.yml

@@ -0,0 +1,46 @@
+name: Build & Push Workflows Image
+
+on:
+  push:
+    branches:
+      - main
+
+jobs:
+  build-push:
+    runs-on: golang
+    env:
+      REGISTRY: forgejo.riotpiao.com
+      IMAGE: forgejo.riotpiao.com/rock/poimen-workflows
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v4
+
+      - name: Download dependencies
+        run: go mod download
+
+      - name: Get short SHA
+        id: sha
+        run: |
+          SHORT_SHA=$(git rev-parse --short HEAD)
+          echo "short_sha=${SHORT_SHA}" >> $GITHUB_OUTPUT
+
+      - name: Registry login
+        run: |
+          echo "${REGISTRY_TOKEN}" | docker login "${REGISTRY}" \
+            --username "${REGISTRY_USER}" --password-stdin
+        env:
+          REGISTRY_USER: ${{ secrets.FORGEJO_REGISTRY_USER }}
+          REGISTRY_TOKEN: ${{ secrets.FORGEJO_REGISTRY_TOKEN }}
+
+      - name: Build image
+        run: |
+          docker build --no-cache \
+            -t "${IMAGE}:${{ steps.sha.outputs.short_sha }}" \
+            -t "${IMAGE}:latest" \
+            .
+
+      - name: Push image
+        run: |
+          docker push "${IMAGE}:${{ steps.sha.outputs.short_sha }}"
+          docker push "${IMAGE}:latest"
+          echo "✓ Image pushed: ${IMAGE}:${{ steps.sha.outputs.short_sha }}"

.gitea/workflows/ci.yaml

@@ -1,66 +1,43 @@
-name: CI
+name: ci
 
 on:
   push:
     branches: [main]
   pull_request:
-    branches: [main]
-  workflow_dispatch:
-
-env:
-  GOPRIVATE: forgejo.riotpiao.com
-  REGISTRY: forgejo.riotpiao.com
-  IMAGE: forgejo.riotpiao.com/rock/poimen-workflows
-  DOCKER_HOST: tcp://localhost:2375
 
 jobs:
-  ci:
-    name: CI
+  test:
     runs-on: golang
+    container:
+      image: golang:1.26
+    env:
+      GOPRIVATE: forgejo.riotpiao.com
+      GITHUB_TOKEN: ${{ secrets.REGISTRY_PAT }}
     steps:
-      - name: Install Node.js and Docker
+      - name: Configure git authentication
         run: |
-          apt-get update
-          apt-get install -y nodejs docker.io
+          git config --global url."https://oauth2:${{ secrets.REGISTRY_PAT }}@forgejo.riotpiao.com".insteadOf "https://forgejo.riotpiao.com"
+          git config --global credential.helper store
+          echo "https://oauth2:${{ secrets.REGISTRY_PAT }}@forgejo.riotpiao.com" >> ~/.git-credentials
 
       - name: Checkout code
-        uses: actions/checkout@v4
+        run: |
+          git init
+          git remote add origin https://forgejo.riotpiao.com/rock/poimen-workflows.git
+          git fetch origin ${{ github.ref_name }} --depth=1
+          git checkout FETCH_HEAD
 
       - name: Download dependencies
         run: go mod download
 
-      - name: Go vet
+      - name: Tidy modules
+        run: go mod tidy
+
+      - name: Test
+        run: go test -v ./...
+
+      - name: Build
+        run: go build -o /tmp/poimen-bin/ ./cmd/...
+
+      - name: Vet
         run: go vet ./...
-
-      - name: Go test
-        run: go test ./...
-
-      - name: Build binary
-        run: CGO_ENABLED=0 GOOS=linux go build -o /tmp/poimen-worker ./cmd/worker
-
-      - name: Get short SHA
-        id: sha
-        run: echo "short_sha=$(git rev-parse --short HEAD)" >> $GITHUB_OUTPUT
-
-      - name: Registry login
-        run: |
-          echo "${REGISTRY_TOKEN}" | docker login "${REGISTRY}" \
-            --username "${REGISTRY_USER}" --password-stdin
-        env:
-          REGISTRY_USER: ${{ secrets.FORGEJO_REGISTRY_USER }}
-          REGISTRY_TOKEN: ${{ secrets.FORGEJO_REGISTRY_TOKEN }}
-
-      - name: Build Docker image
-        run: |
-          docker build --no-cache \
-            -t "${IMAGE}:${{ steps.sha.outputs.short_sha }}" \
-            -t "${IMAGE}:latest" .
-
-      - name: Push Docker image
-        run: |
-          docker push "${IMAGE}:${{ steps.sha.outputs.short_sha }}"
-          docker push "${IMAGE}:latest"
-          echo "✓ Pushed: ${IMAGE}:${{ steps.sha.outputs.short_sha }}"
-
-      - name: Prune unused images
-        run: docker image prune -a --force 2>&1 | tail -3 || true

.sops.yaml

@@ -1,3 +0,0 @@
-creation_rules:
-  - path_regex: k8s/.*\.enc\.ya?ml
-    age: age1e5fq3hwxy78psus2nfvmtmua36g0u3suk78ephw6246l974d2utsvn0hla

Dockerfile

@@ -1,20 +1,22 @@
-FROM golang:1.26-alpine AS builder
+FROM golang:1.21-alpine as builder
 
 WORKDIR /app
 COPY go.mod go.sum ./
 RUN go mod download
+
 COPY . .
+RUN CGO_ENABLED=0 GOOS=linux go build -o workflows ./cmd/poimen/main.go
 
-# Re-use CI-built binary if present, otherwise build
-ARG BINARY_PATH=
-RUN if [ -n "$BINARY_PATH" ] && [ -f "$BINARY_PATH" ]; then \
-      cp "$BINARY_PATH" worker; \
-    else \
-      CGO_ENABLED=0 GOOS=linux go build -o worker ./cmd/worker; \
-    fi
+FROM alpine:latest
 
-FROM alpine:3.20
 RUN apk --no-cache add ca-certificates
+
 WORKDIR /app
-COPY --from=builder /app/worker .
-ENTRYPOINT ["./worker"]
+COPY --from=builder /app/workflows .
+
+EXPOSE 8080
+
+HEALTHCHECK --interval=10s --timeout=5s --start-period=10s --retries=3 \
+  CMD wget --no-verbose --tries=1 --spider http://localhost:8080/health || exit 1
+
+ENTRYPOINT ["./workflows"]

action/analysis.go

@@ -1,4 +1,4 @@
-package activity
+package action
 
 import (
 	"context"

action/analysis_test.go

@@ -1,4 +1,4 @@
-package activity
+package action
 
 import (
 	"context"

action/assume_role.go

@@ -1,4 +1,4 @@
-package activity
+package action
 
 import (
 	"bytes"

action/canvas_compatibility.go

@@ -1,21 +1,14 @@
-package activity
+package action
 
 import (
 	"encoding/json"
 	"fmt"
+	"regexp"
 	"strings"
 
 	"github.com/rockliang/poimen/workflows/pkg/db"
 )
 
-// CanvasCompatibilityOutput validation results
-type CanvasCompatibilityOutput struct {
-	IsValid           bool                    `json:"is_valid"`
-	Incompatibilities []IncompatibilityWarning `json:"incompatibilities"`
-	DisconnectedNodes []string                 `json:"disconnected_nodes"`
-	Warnings          []string                 `json:"warnings"`
-}
-
 // IncompatibilityWarning explains why two activities can't be connected
 type IncompatibilityWarning struct {
 	Source       string `json:"source"`        // Source node ID
@@ -48,7 +41,7 @@ type OutputField struct {
 // getActivitySchema returns schema from knowledge base
 func getActivitySchema(activityType string) (*ActivitySchema, error) {
 	kb := knowledgeBaseData()
-	if kb == "" {
+	if kb == nil {
 		return nil, fmt.Errorf("knowledge base not loaded")
 	}
 
@@ -196,7 +189,7 @@ func CheckConnectionCompatibility(sourceNode, targetNode db.WorkflowNode) []Inco
 }
 
 // CheckCanvasConnectivity analyzes all suggested edges for compatibility
-func CheckCanvasConnectivity(nodes []db.WorkflowNode, suggestedEdges []EdgeWithWording) []IncompatibilityWarning {
+func CheckCanvasConnectivity(nodes []db.WorkflowNode, suggestedEdges []db.WorkflowEdge) []IncompatibilityWarning {
 	warnings := []IncompatibilityWarning{}
 	nodeMap := make(map[string]db.WorkflowNode)
 	for _, n := range nodes {
@@ -221,7 +214,7 @@ func CheckCanvasConnectivity(nodes []db.WorkflowNode, suggestedEdges []EdgeWithW
 }
 
 // IdentifyDisconnectedNodes finds nodes that can't connect to anything
-func IdentifyDisconnectedNodes(nodes []db.WorkflowNode, suggestedEdges []EdgeWithWording) []string {
+func IdentifyDisconnectedNodes(nodes []db.WorkflowNode, suggestedEdges []db.WorkflowEdge) []string {
 	edgeMap := make(map[string]bool)
 	for _, edge := range suggestedEdges {
 		edgeMap[edge.Source] = true
@@ -301,6 +294,20 @@ func knowledgeBaseData() string {
 	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{
@@ -345,15 +352,3 @@ func CanvasCompatibilityActivity(ctx interface{}, input CanvasCompatibilityInput
 
 	return output, nil
 }
-
-// ValidateConnection checks if two nodes can be connected based on their types.
-func ValidateConnection(source, target *db.WorkflowNode) (IncompatibilityWarning, error) {
-	if source.Type != "activity" || target.Type != "activity" {
-		return IncompatibilityWarning{
-			Source: source.ID,
-			Target: target.ID,
-			Reason: fmt.Sprintf("Cannot connect %s to %s: both must be activity type", source.Type, target.Type),
-		}, fmt.Errorf("type mismatch")
-	}
-	return IncompatibilityWarning{}, nil
-}

action/canvas_reasoner.go

@@ -1,14 +1,44 @@
-package activity
+package action
 
 import (
 	"context"
 	"encoding/json"
 	"fmt"
 
-	"github.com/rockliang/poimen/workflows/activity/llm"
+	"github.com/rockliang/poimen/workflows/action/llm"
 	"github.com/rockliang/poimen/workflows/pkg/db"
+	"github.com/rockliang/poimen/workflows/statemachine"
 )
 
+// CanvasReasonerInput infers connections between nodes using LLM reasoning
+type CanvasReasonerInput struct {
+	Nodes []db.WorkflowNode `json:"nodes"`       // Canvas nodes
+	Edges []db.WorkflowEdge `json:"edges"`       // Existing edges
+	// If true, only suggest new edges; if false, redesign entire canvas
+	PreserveExisting bool `json:"preserve_existing,omitempty"`
+	AuthToken        string `json:"auth_token,omitempty"` // JWT for LLM calls
+}
+
+// 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        []EdgeWithWording             `json:"suggested_edges"`         // Edges with wording
@@ -25,14 +55,14 @@ func CanvasReasonerActivity(ctx context.Context, in CanvasReasonerInput) (Canvas
 	logger := newActivityLogger(ctx)
 
 	output := CanvasReasonerOutput{
-		SuggestedEdges: []EdgeWithWording{},
+		SuggestedEdges: []db.WorkflowEdge{},
 	}
 
 	if len(in.Nodes) == 0 {
 		return output, fmt.Errorf("no nodes provided")
 	}
 
-	logger.Info("Analyzing canvas with %d nodes, %d edges", len(in.Nodes), len(in.Edges))
+	logger.logf("info", "Analyzing canvas with %d nodes, %d edges", len(in.Nodes), len(in.Edges))
 
 	// Build activity descriptions for LLM context
 	nodeDesc := buildNodeDescriptions(in.Nodes)
@@ -88,7 +118,7 @@ KEY RULES:
 
 Return ONLY valid JSON, no markdown code blocks.`, nodeDesc, edgeDesc, getReasoningTask(in.PreserveExisting))
 
-	logger.Info("Calling LLM reasoning (preserve_existing=%v)", in.PreserveExisting)
+	logger.logf("info", "Calling LLM reasoning (preserve_existing=%v)", in.PreserveExisting)
 
 	// Call LLM
 	client, err := llm.NewClient()
@@ -97,7 +127,7 @@ Return ONLY valid JSON, no markdown code blocks.`, nodeDesc, edgeDesc, getReason
 	}
 
 	response, err := client.CreateMessage(ctx, llm.MessageInput{
-		Model: ModelSpec{
+		Model: statemachine.ModelSpec{
 			ModelID: "reasoning", // Use reasoning model for complex analysis
 		},
 		SystemPrompt: systemPrompt,
@@ -116,13 +146,13 @@ Return ONLY valid JSON, no markdown code blocks.`, nodeDesc, edgeDesc, getReason
 
 	// Parse LLM response
 	var reasonerResp struct {
-		Edges      []EdgeWithWording `json:"edges"`
+		Edges      []db.WorkflowEdge `json:"edges"`
 		Reasoning  string            `json:"reasoning"`
 		Confidence float64           `json:"confidence"`
 	}
 
 	if err := json.Unmarshal([]byte(response), &reasonerResp); err != nil {
-		logger.Warn("Failed to parse LLM response as JSON: %v", err)
+		logger.logf("warn", "Failed to parse LLM response as JSON: %v", err)
 		// Try to extract from response text
 		output.Reasoning = response
 		output.Confidence = 0.5
@@ -135,19 +165,19 @@ Return ONLY valid JSON, no markdown code blocks.`, nodeDesc, edgeDesc, getReason
 		nodeMap[n.ID] = true
 	}
 
-	validEdges := []EdgeWithWording{}
+	validEdges := []db.WorkflowEdge{}
 	for _, edge := range reasonerResp.Edges {
 		if !nodeMap[edge.Source] {
-			logger.Warn("Suggested edge references unknown source: %s", edge.Source)
+			logger.logf("warn", "Suggested edge references unknown source: %s", edge.Source)
 			continue
 		}
 		if !nodeMap[edge.Target] {
-			logger.Warn("Suggested edge references unknown target: %s", edge.Target)
+			logger.logf("warn", "Suggested edge references unknown target: %s", edge.Target)
 			continue
 		}
 		// Don't suggest self-loops
 		if edge.Source == edge.Target {
-			logger.Warn("Skipping self-loop: %s", edge.Source)
+			logger.logf("warn", "Skipping self-loop: %s", edge.Source)
 			continue
 		}
 		validEdges = append(validEdges, edge)
@@ -161,7 +191,7 @@ Return ONLY valid JSON, no markdown code blocks.`, nodeDesc, edgeDesc, getReason
 	incompatibilities := CheckCanvasConnectivity(in.Nodes, validEdges)
 	if len(incompatibilities) > 0 {
 		output.IncompatibleEdges = incompatibilities
-		logger.Warn("Found %d incompatible edge connections", len(incompatibilities))
+		logger.logf("warn", "Found %d incompatible edge connections", len(incompatibilities))
 
 		// Generate user-friendly alerts
 		for i, incompat := range incompatibilities {
@@ -179,7 +209,7 @@ Return ONLY valid JSON, no markdown code blocks.`, nodeDesc, edgeDesc, getReason
 	disconnected := IdentifyDisconnectedNodes(in.Nodes, validEdges)
 	if len(disconnected) > 0 {
 		output.DisconnectedNodes = disconnected
-		logger.Warn("Found %d disconnected nodes", len(disconnected))
+		logger.logf("warn", "Found %d disconnected nodes", len(disconnected))
 
 		for _, nodeID := range disconnected {
 			var label string
@@ -197,7 +227,7 @@ Return ONLY valid JSON, no markdown code blocks.`, nodeDesc, edgeDesc, getReason
 		}
 	}
 
-	logger.Info("LLM suggested %d edges with confidence %.2f | %d incompatibilities | %d disconnected",
+	logger.logf("info", "LLM suggested %d edges with confidence %.2f | %d incompatibilities | %d disconnected",
 		len(validEdges), output.Confidence, len(incompatibilities), len(disconnected))
 
 	return output, nil

action/fetch_canvas_relations.go

@@ -1,4 +1,4 @@
-package activity
+package action
 
 import (
 	"context"
@@ -8,6 +8,22 @@ import (
 	"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)
@@ -19,16 +35,16 @@ func FetchCanvasRelationsActivity(ctx context.Context, input FetchCanvasRelation
 		Relations:  []EdgeWithWording{},
 	}
 
-	logger.Info("Fetching canvas relations: %s v%d", input.WorkflowID, input.Version)
+	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.DB)
+	dbClient, ok := ctx.Value("db_client").(*db.Client)
 	if !ok {
 		return output, fmt.Errorf("database client not in context")
 	}
 
 	// Fetch workflow
-	workflow, err := dbClient.FetchWorkflow(ctx, input.WorkflowID, "")
+	workflow, err := dbClient.GetWorkflow(ctx, input.WorkflowID)
 	if err != nil {
 		return output, fmt.Errorf("failed to get workflow: %w", err)
 	}
@@ -52,7 +68,7 @@ func FetchCanvasRelationsActivity(ctx context.Context, input FetchCanvasRelation
 	relations, err := dbClient.GetWorkflowRelations(ctx, input.WorkflowID, input.Version)
 	if err != nil {
 		// Relations may not exist for old canvases - this is OK
-		logger.Warn("Failed to fetch relations: %v", err)
+		logger.logf("warn", "Failed to fetch relations: %v", err)
 		return output, nil
 	}
 
@@ -69,12 +85,12 @@ func FetchCanvasRelationsActivity(ctx context.Context, input FetchCanvasRelation
 
 		// Parse relation wording JSON
 		if err := json.Unmarshal(rel.RelationWording, &edge.RelationWording); err != nil {
-			logger.Warn("Failed to parse relation wording: %v", err)
+			logger.logf("warn", "Failed to parse relation wording: %v", err)
 		}
 
 		output.Relations = append(output.Relations, edge)
 	}
 
-	logger.Info("Fetched %d nodes, %d edges, %d relations", len(output.Nodes), len(output.Edges), len(output.Relations))
+	logger.logf("info", "Fetched %d nodes, %d edges, %d relations", len(output.Nodes), len(output.Edges), len(output.Relations))
 	return output, nil
 }

action/git.go

@@ -1,4 +1,4 @@
-package activity
+package action
 
 import (
 	"context"

action/implementer.go

@@ -0,0 +1,93 @@
+package action
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/rockliang/poimen/workflows/action/llm"
+	"github.com/rockliang/poimen/workflows/prompts"
+	"github.com/rockliang/poimen/workflows/statemachine"
+	"go.temporal.io/sdk/activity"
+)
+
+// ImplementerInput is input to ImplementerActivity.
+type ImplementerInput struct {
+	Config      statemachine.OrchestratorConfig
+	TaskID      string
+	WorktreePath string
+	Lessons     string // "known errors — do not repeat" section
+}
+
+// ImplementerOutput is the output of ImplementerActivity.
+type ImplementerOutput struct {
+	Success bool
+	Changes string // summary of changes made
+}
+
+// ImplementerActivity calls the Implementer LLM to implement the task.
+func ImplementerActivity(ctx context.Context, in ImplementerInput) (ImplementerOutput, error) {
+	// Record heartbeat
+	activity.RecordHeartbeat(ctx, "starting implementer for "+in.TaskID)
+
+	// Get LLM client
+	client, err := llm.NewClient()
+	if err != nil {
+		return ImplementerOutput{}, fmt.Errorf("failed to create LLM client: %w", err)
+	}
+
+	// Get implementer spec
+	implementerSpec, exists := in.Config.RolePrompts["implementer"]
+	if !exists {
+		return ImplementerOutput{}, fmt.Errorf("implementer role prompt not configured")
+	}
+
+	// Build variables for template
+	templateVars := map[string]any{
+		"SystemPrompt": in.Config.SystemPrompt,
+		"Task":         in.TaskID,
+		"WorktreePath": in.WorktreePath,
+	}
+
+	// Inject lessons if provided
+	if in.Lessons != "" {
+		templateVars["Lessons"] = in.Lessons
+	}
+
+	// Render template
+	var templateContent string
+	if implementerSpec.RawTemplate != "" {
+		templateContent = implementerSpec.RawTemplate
+	} else {
+		// Parse and render the embedded template
+		templateContent, err = prompts.Render(implementerSpec.TemplateRef, templateVars)
+		if err != nil {
+			return ImplementerOutput{}, fmt.Errorf("failed to render implementer template: %w", err)
+		}
+	}
+
+	// Call LLM
+	messages := []llm.MessageParam{
+		{
+			Role:    "user",
+			Content: templateContent,
+		},
+	}
+
+	response, err := client.CreateMessage(ctx, llm.MessageInput{
+		Model:        implementerSpec.Model,
+		SystemPrompt: in.Config.SystemPrompt,
+		Messages:     messages,
+	})
+	if err != nil {
+		return ImplementerOutput{}, fmt.Errorf("implementer LLM call failed: %w", err)
+	}
+
+	// Record progress
+	activity.RecordHeartbeat(ctx, "implementer completed for "+in.TaskID)
+
+	// Return success (in full implementation would parse response and execute tool calls)
+	return ImplementerOutput{
+		Success: true,
+		Changes: response,
+	}, nil
+}

action/index_graph_rag.go

@@ -1,10 +1,37 @@
-package activity
+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{

action/integration.go

@@ -1,4 +1,4 @@
-package activity
+package action
 
 import (
 	"context"

action/judge.go

@@ -0,0 +1,78 @@
+package action
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/rockliang/poimen/workflows/action/llm"
+	"github.com/rockliang/poimen/workflows/prompts"
+	"github.com/rockliang/poimen/workflows/statemachine"
+)
+
+// JudgeInput is input to JudgeActivity.
+type JudgeInput struct {
+	Config              statemachine.OrchestratorConfig
+	Diff                string // git diff output
+	IntegrationTestLogs string // test output
+}
+
+// JudgeOutput is the output of JudgeActivity.
+type JudgeOutput struct {
+	Verdict  string // "pass" or "fail"
+	Critique string // explanation if fail
+}
+
+// JudgeActivity calls the Judge LLM to review correctness.
+func JudgeActivity(ctx context.Context, in JudgeInput) (JudgeOutput, error) {
+	// Get LLM client
+	client, err := llm.NewClient()
+	if err != nil {
+		return JudgeOutput{}, fmt.Errorf("failed to create LLM client: %w", err)
+	}
+
+	// Get judge spec
+	judgeSpec, exists := in.Config.RolePrompts["judge"]
+	if !exists {
+		return JudgeOutput{}, fmt.Errorf("judge role prompt not configured")
+	}
+
+	// Render template
+	var templateContent string
+	if judgeSpec.RawTemplate != "" {
+		templateContent = judgeSpec.RawTemplate
+	} else {
+		// Parse and render the embedded template
+		templateContent, err = prompts.Render(judgeSpec.TemplateRef, map[string]any{
+			"SystemPrompt": in.Config.SystemPrompt,
+			"Diff":         in.Diff,
+			"TestResult":   in.IntegrationTestLogs,
+		})
+		if err != nil {
+			return JudgeOutput{}, fmt.Errorf("failed to render judge template: %w", err)
+		}
+	}
+
+	// Call LLM
+	messages := []llm.MessageParam{
+		{
+			Role:    "user",
+			Content: templateContent,
+		},
+	}
+
+	response, err := client.CreateMessage(ctx, llm.MessageInput{
+		Model:        judgeSpec.Model,
+		SystemPrompt: in.Config.SystemPrompt,
+		Messages:     messages,
+	})
+	if err != nil {
+		return JudgeOutput{}, fmt.Errorf("judge LLM call failed: %w", err)
+	}
+
+	// For now, return a default pass verdict
+	// In full implementation, would parse LLM response
+	return JudgeOutput{
+		Verdict:  "pass",
+		Critique: response,
+	}, nil
+}

action/lessons.go

@@ -1,4 +1,4 @@
-package activity
+package action
 
 import (
 	"context"

action/llm/client.go

@@ -9,7 +9,7 @@ import (
 	"net/http"
 	"os"
 
-	"github.com/rockliang/poimen/workflows/pkg/types"
+	"github.com/rockliang/poimen/workflows/statemachine"
 )
 
 var (
@@ -54,7 +54,7 @@ func NewClient() (*OpenAIClient, error) {
 
 // MessageInput is the input to CreateMessage.
 type MessageInput struct {
-	Model        types.ModelSpec
+	Model        statemachine.ModelSpec
 	SystemPrompt string
 	Messages     []MessageParam
 	AuthToken    string // Optional JWT token for authenticated endpoints

action/llm/client_test.go

@@ -4,7 +4,7 @@ import (
 	"context"
 	"testing"
 
-	"github.com/rockliang/poimen/workflows/pkg/types"
+	"github.com/rockliang/poimen/workflows/statemachine"
 )
 
 func TestNewClient(t *testing.T) {
@@ -70,7 +70,7 @@ func TestCreateMessageValidation(t *testing.T) {
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			in := MessageInput{
-				Model: types.ModelSpec{
+				Model: statemachine.ModelSpec{
 					ModelID: tt.modelID,
 				},
 				SystemPrompt: "test",

action/llm_inference.go

@@ -0,0 +1,159 @@
+package action
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/rockliang/poimen/workflows/action/llm"
+	"github.com/rockliang/poimen/workflows/statemachine"
+)
+
+// LLMInferenceInput is input for LLMInferenceActivity
+type LLMInferenceInput struct {
+	Model         string                 `json:"model"`          // Model ID (reasoning, ornith:35b, etc)
+	SystemPrompt  string                 `json:"system_prompt"`  // System instruction
+	UserPrompt    string                 `json:"user_prompt"`    // User message
+	Temperature   float64                `json:"temperature,omitempty"`   // LLM temperature (0-1)
+	MaxTokens     int                    `json:"max_tokens,omitempty"`    // Max output tokens
+	AuthToken     string                 `json:"auth_token,omitempty"`    // JWT token for authenticated endpoints
+}
+
+// LLMInferenceOutput is output from LLMInferenceActivity
+type LLMInferenceOutput struct {
+	Response     string `json:"response"`      // LLM response text
+	Model        string `json:"model"`         // Model used
+	StopReason   string `json:"stop_reason"`   // How inference stopped (stop_sequence, length, etc)
+	TokensUsed   int    `json:"tokens_used"`   // Total tokens consumed
+	ErrorMessage string `json:"error,omitempty"`
+}
+
+// LLMInferenceActivity calls LLM API with given prompt and returns response
+func LLMInferenceActivity(ctx context.Context, in LLMInferenceInput) (LLMInferenceOutput, error) {
+	logger := newActivityLogger(ctx)
+
+	output := LLMInferenceOutput{
+		Model: in.Model,
+	}
+
+	// Validate input
+	if in.Model == "" {
+		return output, fmt.Errorf("model not specified")
+	}
+
+	if in.UserPrompt == "" {
+		return output, fmt.Errorf("user_prompt not specified")
+	}
+
+	logger.logf("info", "Starting LLM inference with model: %s", in.Model)
+
+	// Create LLM client
+	client, err := llm.NewClient()
+	if err != nil {
+		output.ErrorMessage = err.Error()
+		return output, fmt.Errorf("failed to create LLM client: %w", err)
+	}
+
+	// Call LLM
+	logger.logf("info", "Calling LLM API (model=%s, prompt_len=%d, auth=%v)", in.Model, len(in.UserPrompt), in.AuthToken != "")
+
+	response, err := client.CreateMessage(ctx, llm.MessageInput{
+		Model: statemachine.ModelSpec{
+			ModelID: in.Model,
+		},
+		SystemPrompt: in.SystemPrompt,
+		Messages: []llm.MessageParam{
+			{
+				Role:    "user",
+				Content: in.UserPrompt,
+			},
+		},
+		AuthToken: in.AuthToken,
+	})
+
+	if err != nil {
+		output.ErrorMessage = err.Error()
+		logger.logf("error", "LLM API call failed: %v", err)
+		return output, fmt.Errorf("LLM inference failed: %w", err)
+	}
+
+	output.Response = response
+	output.StopReason = "stop_sequence"
+
+	logger.logf("info", "LLM inference completed (response_len=%d)", len(response))
+
+	return output, nil
+}
+
+// LLMBatchInferenceInput is input for batch inference
+type LLMBatchInferenceInput struct {
+	Model        string                   `json:"model"`
+	SystemPrompt string                   `json:"system_prompt"`
+	Prompts      []string                 `json:"prompts"`       // List of user prompts
+	Temperature  float64                  `json:"temperature,omitempty"`
+	AuthToken    string                   `json:"auth_token,omitempty"`    // JWT token for authenticated endpoints
+}
+
+// LLMBatchInferenceOutput is output from batch inference
+type LLMBatchInferenceOutput struct {
+	Responses []string `json:"responses"`     // LLM responses (parallel to input Prompts)
+	Model     string   `json:"model"`
+	Errors    []string `json:"errors,omitempty"`
+}
+
+// LLMBatchInferenceActivity calls LLM multiple times in sequence
+func LLMBatchInferenceActivity(ctx context.Context, in LLMBatchInferenceInput) (LLMBatchInferenceOutput, error) {
+	logger := newActivityLogger(ctx)
+
+	output := LLMBatchInferenceOutput{
+		Model:     in.Model,
+		Responses: []string{},
+		Errors:    []string{},
+	}
+
+	if in.Model == "" {
+		return output, fmt.Errorf("model not specified")
+	}
+
+	if len(in.Prompts) == 0 {
+		return output, fmt.Errorf("no prompts provided")
+	}
+
+	logger.logf("info", "Starting batch LLM inference (model=%s, count=%d)", in.Model, len(in.Prompts))
+
+	// Create LLM client
+	client, err := llm.NewClient()
+	if err != nil {
+		return output, fmt.Errorf("failed to create LLM client: %w", err)
+	}
+
+	// Process each prompt
+	for i, prompt := range in.Prompts {
+		logger.logf("info", "Processing prompt %d/%d", i+1, len(in.Prompts))
+
+		response, err := client.CreateMessage(ctx, llm.MessageInput{
+			Model: statemachine.ModelSpec{
+				ModelID: in.Model,
+			},
+			SystemPrompt: in.SystemPrompt,
+			Messages: []llm.MessageParam{
+				{
+					Role:    "user",
+					Content: prompt,
+				},
+			},
+		})
+
+		if err != nil {
+			output.Errors = append(output.Errors, fmt.Sprintf("prompt %d: %v", i, err))
+			output.Responses = append(output.Responses, "")
+			logger.logf("warn", "Failed to process prompt %d: %v", i, err)
+		} else {
+			output.Responses = append(output.Responses, response)
+		}
+	}
+
+	logger.logf("info", "Batch inference completed (responses=%d, errors=%d)", 
+		len(output.Responses), len(output.Errors))
+
+	return output, nil
+}

action/logger.go

@@ -1,4 +1,4 @@
-package activity
+package action
 
 import (
 	"context"

action/memory.go

@@ -1,4 +1,4 @@
-package activity
+package action
 
 import (
 	"context"
@@ -92,9 +92,9 @@ func RetrieveMemoryActivity(ctx context.Context, in RetrieveMemoryInput) (Retrie
 	}
 
 	// Get memory service URL and token
-	baseURL := os.Getenv("MEMORY_SERVICE_URL")
+	baseURL := os.Getenv("POIMEN_MEMORY_URL")
 	if baseURL == "" {
-		baseURL = "http://localhost:8080"
+		baseURL = "http://poimen-memory.poimen.svc.cluster.local:8080"
 	}
 
 	token := os.Getenv("POIMEN_MEMORY_TOKEN")

action/memory_test.go

@@ -1,4 +1,4 @@
-package activity
+package action
 
 import (
 	"context"
@@ -39,7 +39,7 @@ func TestRetrieveMemoryActivity_Query(t *testing.T) {
 	defer server.Close()
 
 	// Set env for test
-	t.Setenv("MEMORY_SERVICE_URL", server.URL)
+	t.Setenv("POIMEN_MEMORY_URL", server.URL)
 
 	output, err := RetrieveMemoryActivity(context.Background(), RetrieveMemoryInput{
 		Query:   "security scanning",
@@ -93,7 +93,7 @@ func TestRetrieveMemoryActivity_Context(t *testing.T) {
 	}))
 	defer server.Close()
 
-	t.Setenv("MEMORY_SERVICE_URL", server.URL)
+	t.Setenv("POIMEN_MEMORY_URL", server.URL)
 
 	output, err := RetrieveMemoryActivity(context.Background(), RetrieveMemoryInput{
 		Query: "security scan repo",

action/notification.go

@@ -1,4 +1,4 @@
-package activity
+package action
 
 import (
 	"bytes"

action/notification_test.go

@@ -1,4 +1,4 @@
-package activity
+package action
 
 import (
 	"context"

action/planner.go

@@ -0,0 +1,91 @@
+package action
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/rockliang/poimen/workflows/action/llm"
+	"github.com/rockliang/poimen/workflows/prompts"
+	"github.com/rockliang/poimen/workflows/statemachine"
+)
+
+// PlanningInput is input to PlanningActivity.
+type PlanningInput struct {
+	Config        statemachine.OrchestratorConfig
+	BoardState    string                            // JSON or markdown of task board
+	RepoPath      string                            // Path to target repository
+	Milestone     string                            // e.g., "T0"
+	TaskResults   []statemachine.TaskUnitOutput     // Results from completed tasks
+}
+
+// TaskDispatch represents a dispatched task.
+type TaskDispatch struct {
+	TaskID       string
+	PromptSpec   statemachine.PromptSpec
+	BaseTimeout  *int64 // optional override in milliseconds
+}
+
+// PlanningOutput is the output of PlanningActivity.
+type PlanningOutput struct {
+	TasksToDispatch        []string // Task IDs to dispatch in this cycle
+	CompletedBranches      []string // Branches to squash merge (when milestone complete)
+	SubmilestoneComplete   bool     // Whether the milestone is complete
+}
+
+// PlanningActivity calls the Planner LLM to decide which tasks to dispatch.
+func PlanningActivity(ctx context.Context, in PlanningInput) (PlanningOutput, error) {
+	// Get LLM client
+	client, err := llm.NewClient()
+	if err != nil {
+		return PlanningOutput{}, fmt.Errorf("failed to create LLM client: %w", err)
+	}
+
+	// Get planner spec
+	plannerSpec, exists := in.Config.RolePrompts["planner"]
+	if !exists {
+		return PlanningOutput{}, fmt.Errorf("planner role prompt not configured")
+	}
+
+	// Render template
+	var templateContent string
+	if plannerSpec.RawTemplate != "" {
+		templateContent = plannerSpec.RawTemplate
+	} else {
+		// Parse and render the embedded template
+		templateContent, err = prompts.Render(plannerSpec.TemplateRef, map[string]any{
+			"SystemPrompt": in.Config.SystemPrompt,
+			"BoardState":   in.BoardState,
+			"Milestone":    in.Milestone,
+			"Config":       in.Config,
+		})
+		if err != nil {
+			return PlanningOutput{}, fmt.Errorf("failed to render planner template: %w", err)
+		}
+	}
+
+	// Call LLM
+	messages := []llm.MessageParam{
+		{
+			Role:    "user",
+			Content: templateContent,
+		},
+	}
+
+	response, err := client.CreateMessage(ctx, llm.MessageInput{
+		Model:        plannerSpec.Model,
+		SystemPrompt: in.Config.SystemPrompt,
+		Messages:     messages,
+	})
+	if err != nil {
+		return PlanningOutput{}, fmt.Errorf("planner LLM call failed: %w", err)
+	}
+
+	// For now, return empty dispatch (will be parsed from LLM response in full implementation)
+	// This is a stub that allows the test to verify the activity is called
+	_ = response
+	return PlanningOutput{
+		TasksToDispatch:      []string{},
+		CompletedBranches:    []string{},
+		SubmilestoneComplete: false,
+	}, nil
+}

action/query_graph_rag.go

@@ -1,4 +1,4 @@
-package activity
+package action
 
 import (
 	"bytes"
@@ -11,6 +11,38 @@ import (
 	"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)
@@ -21,7 +53,7 @@ func QueryGraphRAGActivity(ctx context.Context, input GraphRAGQueryInput) (Graph
 		Paths:      []QueryPathData{},
 	}
 
-	logger.Info("Querying GraphRAG: %s", input.Query)
+	logger.logf("info", "Querying GraphRAG: %s", input.Query)
 
 	// Get Memory Service URL from env
 	memoryURL := os.Getenv("MEMORY_SERVICE_URL")
@@ -95,7 +127,7 @@ func QueryGraphRAGActivity(ctx context.Context, input GraphRAGQueryInput) (Graph
 	output.HasMore = graphResp.HasMore
 	output.ExecutionMs = time.Since(startTime).Milliseconds()
 
-	logger.Info("GraphRAG returned %d edges, %d paths in %dms", 
+	logger.logf("info", "GraphRAG returned %d edges, %d paths in %dms", 
 		len(output.Edges), len(output.Paths), output.ExecutionMs)
 	return output, nil
 }

action/router.go

@@ -1,4 +1,4 @@
-package activity
+package action
 
 import (
 	"context"

action/skills.go

@@ -1,4 +1,4 @@
-package activity
+package action
 
 import (
 	"context"
@@ -10,11 +10,12 @@ import (
 	"time"
 
 	"go.temporal.io/sdk/activity"
+	"github.com/rockliang/poimen/workflows/statemachine"
 )
 
 // PrepareSkillsInput is input to PrepareSkillsActivity.
 type PrepareSkillsInput struct {
-	Skills        []SkillRef
+	Skills        []statemachine.SkillRef
 	StreamTimeout time.Duration
 	Provider      string // pi provider name (e.g. "homelab-reasoning"); required, pi has no usable default provider
 }

activity/implementer.go

@@ -1,41 +0,0 @@
-package activity
-
-import (
-	"context"
-
-	"github.com/rockliang/poimen/workflows/pkg/types"
-	"go.temporal.io/sdk/activity"
-)
-
-type ImplementerInput struct {
-	Config       types.OrchestratorConfig
-	TaskID       string
-	WorktreePath string
-	Lessons      string
-}
-
-type ImplementerOutput struct {
-	Success bool
-	Changes string
-}
-
-func ImplementerActivity(ctx context.Context, in ImplementerInput) (ImplementerOutput, error) {
-	activity.RecordHeartbeat(ctx, "starting implementer for "+in.TaskID)
-
-	vars := map[string]any{
-		"SystemPrompt": in.Config.SystemPrompt,
-		"Task":         in.TaskID,
-		"WorktreePath": in.WorktreePath,
-	}
-	if in.Lessons != "" {
-		vars["Lessons"] = in.Lessons
-	}
-
-	response, err := CallRoleLLM(ctx, in.Config, "implementer", vars)
-	if err != nil {
-		return ImplementerOutput{}, err
-	}
-
-	activity.RecordHeartbeat(ctx, "implementer completed for "+in.TaskID)
-	return ImplementerOutput{Success: true, Changes: response}, nil
-}

activity/judge.go

@@ -1,32 +0,0 @@
-package activity
-
-import (
-	"context"
-
-	"github.com/rockliang/poimen/workflows/pkg/types"
-)
-
-type JudgeInput struct {
-	Config              types.OrchestratorConfig
-	Diff                string
-	IntegrationTestLogs string
-}
-
-type JudgeOutput struct {
-	Verdict  string
-	Critique string
-}
-
-func JudgeActivity(ctx context.Context, in JudgeInput) (JudgeOutput, error) {
-	response, err := CallRoleLLM(ctx, in.Config, "judge", map[string]any{
-		"SystemPrompt": in.Config.SystemPrompt,
-		"Diff":         in.Diff,
-		"TestResult":   in.IntegrationTestLogs,
-	})
-	if err != nil {
-		return JudgeOutput{}, err
-	}
-
-	// TODO: parse LLM response for verdict
-	return JudgeOutput{Verdict: "pass", Critique: response}, nil
-}

activity/llm_helper.go

@@ -1,47 +0,0 @@
-package activity
-
-import (
-	"context"
-	"fmt"
-
-	"github.com/rockliang/poimen/workflows/activity/llm"
-	"github.com/rockliang/poimen/workflows/pkg/types"
-	"github.com/rockliang/poimen/workflows/prompts"
-)
-
-// CallRoleLLM is the shared pattern for calling an LLM with a role-based prompt.
-// Used by planner, implementer, and judge activities (DRY extraction).
-func CallRoleLLM(ctx context.Context, config types.OrchestratorConfig, role string, vars map[string]any) (string, error) {
-	client, err := llm.NewClient()
-	if err != nil {
-		return "", fmt.Errorf("failed to create LLM client: %w", err)
-	}
-
-	spec, exists := config.RolePrompts[role]
-	if !exists {
-		return "", fmt.Errorf("%s role prompt not configured", role)
-	}
-
-	// Render template
-	var content string
-	if spec.RawTemplate != "" {
-		content = spec.RawTemplate
-	} else {
-		content, err = prompts.Render(spec.TemplateRef, vars)
-		if err != nil {
-			return "", fmt.Errorf("failed to render %s template: %w", role, err)
-		}
-	}
-
-	// Call LLM
-	response, err := client.CreateMessage(ctx, llm.MessageInput{
-		Model:        spec.Model,
-		SystemPrompt: config.SystemPrompt,
-		Messages:     []llm.MessageParam{{Role: "user", Content: content}},
-	})
-	if err != nil {
-		return "", fmt.Errorf("%s LLM call failed: %w", role, err)
-	}
-
-	return response, nil
-}

activity/llm_inference.go

@@ -1,114 +0,0 @@
-package activity
-
-import (
-	"context"
-	"fmt"
-
-	"github.com/rockliang/poimen/workflows/activity/llm"
-	"github.com/rockliang/poimen/workflows/pkg/types"
-)
-
-type LLMInferenceInput struct {
-	Model        string  `json:"model"`
-	SystemPrompt string  `json:"system_prompt"`
-	UserPrompt   string  `json:"user_prompt"`
-	Temperature  float64 `json:"temperature,omitempty"`
-	MaxTokens    int     `json:"max_tokens,omitempty"`
-	AuthToken    string  `json:"auth_token,omitempty"`
-}
-
-type LLMInferenceOutput struct {
-	Response     string `json:"response"`
-	Model        string `json:"model"`
-	StopReason   string `json:"stop_reason"`
-	TokensUsed   int    `json:"tokens_used"`
-	ErrorMessage string `json:"error,omitempty"`
-}
-
-func LLMInferenceActivity(ctx context.Context, in LLMInferenceInput) (LLMInferenceOutput, error) {
-	logger := newActivityLogger(ctx)
-	output := LLMInferenceOutput{Model: in.Model}
-
-	if in.Model == "" {
-		return output, fmt.Errorf("model not specified")
-	}
-	if in.UserPrompt == "" {
-		return output, fmt.Errorf("user_prompt not specified")
-	}
-
-	logger.Info("Starting LLM inference", "model", in.Model)
-
-	client, err := llm.NewClient()
-	if err != nil {
-		output.ErrorMessage = err.Error()
-		return output, fmt.Errorf("failed to create LLM client: %w", err)
-	}
-
-	response, err := client.CreateMessage(ctx, llm.MessageInput{
-		Model:        types.ModelSpec{ModelID: in.Model},
-		SystemPrompt: in.SystemPrompt,
-		Messages:     []llm.MessageParam{{Role: "user", Content: in.UserPrompt}},
-		AuthToken:    in.AuthToken,
-	})
-	if err != nil {
-		output.ErrorMessage = err.Error()
-		logger.Warn("LLM API call failed", "error", err)
-		return output, fmt.Errorf("LLM inference failed: %w", err)
-	}
-
-	output.Response = response
-	output.StopReason = "stop_sequence"
-	logger.Info("LLM inference completed", "response_len", len(response))
-	return output, nil
-}
-
-type LLMBatchInferenceInput struct {
-	Model        string   `json:"model"`
-	SystemPrompt string   `json:"system_prompt"`
-	Prompts      []string `json:"prompts"`
-	Temperature  float64  `json:"temperature,omitempty"`
-	AuthToken    string   `json:"auth_token,omitempty"`
-}
-
-type LLMBatchInferenceOutput struct {
-	Responses []string `json:"responses"`
-	Model     string   `json:"model"`
-	Errors    []string `json:"errors,omitempty"`
-}
-
-func LLMBatchInferenceActivity(ctx context.Context, in LLMBatchInferenceInput) (LLMBatchInferenceOutput, error) {
-	logger := newActivityLogger(ctx)
-	output := LLMBatchInferenceOutput{Model: in.Model, Responses: []string{}, Errors: []string{}}
-
-	if in.Model == "" {
-		return output, fmt.Errorf("model not specified")
-	}
-	if len(in.Prompts) == 0 {
-		return output, fmt.Errorf("no prompts provided")
-	}
-
-	logger.Info("Starting batch inference", "model", in.Model, "count", len(in.Prompts))
-
-	client, err := llm.NewClient()
-	if err != nil {
-		return output, fmt.Errorf("failed to create LLM client: %w", err)
-	}
-
-	for i, prompt := range in.Prompts {
-		response, err := client.CreateMessage(ctx, llm.MessageInput{
-			Model:        types.ModelSpec{ModelID: in.Model},
-			SystemPrompt: in.SystemPrompt,
-			Messages:     []llm.MessageParam{{Role: "user", Content: prompt}},
-		})
-		if err != nil {
-			output.Errors = append(output.Errors, fmt.Sprintf("prompt %d: %v", i, err))
-			output.Responses = append(output.Responses, "")
-			logger.Warn("Failed prompt", "index", i, "error", err)
-		} else {
-			output.Responses = append(output.Responses, response)
-		}
-	}
-
-	logger.Info("Batch inference completed", "responses", len(output.Responses), "errors", len(output.Errors))
-	return output, nil
-}

activity/planner.go

@@ -1,47 +0,0 @@
-package activity
-
-import (
-	"context"
-
-	"github.com/rockliang/poimen/workflows/pkg/types"
-)
-
-type PlanningInput struct {
-	Config      types.OrchestratorConfig
-	BoardState  string
-	RepoPath    string
-	Milestone   string
-	TaskResults []types.TaskUnitOutput
-}
-
-type TaskDispatch struct {
-	TaskID      string
-	PromptSpec  types.PromptSpec
-	BaseTimeout *int64
-}
-
-type PlanningOutput struct {
-	TasksToDispatch      []string
-	CompletedBranches    []string
-	SubmilestoneComplete bool
-}
-
-func PlanningActivity(ctx context.Context, in PlanningInput) (PlanningOutput, error) {
-	response, err := CallRoleLLM(ctx, in.Config, "planner", map[string]any{
-		"SystemPrompt": in.Config.SystemPrompt,
-		"BoardState":   in.BoardState,
-		"Milestone":    in.Milestone,
-		"Config":       in.Config,
-	})
-	if err != nil {
-		return PlanningOutput{}, err
-	}
-
-	// TODO: parse LLM response into task dispatch list
-	_ = response
-	return PlanningOutput{
-		TasksToDispatch:      []string{},
-		CompletedBranches:    []string{},
-		SubmilestoneComplete: false,
-	}, nil
-}

activity/synthesis.go

@@ -1,346 +0,0 @@
-package activity
-
-import (
-	"context"
-	"crypto/sha256"
-	"encoding/hex"
-	"fmt"
-	"log/slog"
-	"regexp"
-	"strings"
-
-	"github.com/rockliang/poimen/workflows/internal/memory"
-	"github.com/rockliang/poimen/workflows/pkg/types"
-)
-
-// SynthesisActivities holds dependencies for synthesis pipeline activities.
-type SynthesisActivities struct {
-	memClient *memory.Client
-}
-
-// NewSynthesisActivities creates synthesis activities with a memory service client.
-func NewSynthesisActivities(memClient *memory.Client) *SynthesisActivities {
-	return &SynthesisActivities{memClient: memClient}
-}
-
-// Re-export shared types from pkg/types
-type SynthesisInput = types.SynthesisInput
-type ExtractedEntity = types.ExtractedEntity
-type ExtractedFact = types.ExtractedFact
-type ContradictionResult = types.ContradictionResult
-type PersistInput = types.PersistInput
-
-// ChunkAndEmbedActivity chunks text and generates a chunk ID.
-// Stage 1: Creates a deterministic chunk ID from content hash,
-// then ingests via memory service for embedding generation.
-func (s *SynthesisActivities) ChunkAndEmbedActivity(ctx context.Context, input SynthesisInput) (string, error) {
-	logger := slog.Default()
-
-	// Generate deterministic chunk ID from content
-	hash := sha256.Sum256([]byte(input.Text))
-	chunkID := "chunk-" + hex.EncodeToString(hash[:8])
-
-	logger.Info("chunking text", "chunk_id", chunkID, "text_len", len(input.Text))
-
-	// Ingest via memory service (generates embedding)
-	_, err := s.memClient.Ingest(ctx, &memory.IngestRequest{
-		Project: input.Project,
-		Source:  input.Source,
-		Kind:    input.Kind,
-		Text:    input.Text,
-		Metadata: map[string]interface{}{
-			"chunk_id": chunkID,
-			"tags":     input.Tags,
-		},
-	})
-	if err != nil {
-		return "", fmt.Errorf("ingest chunk: %w", err)
-	}
-
-	return chunkID, nil
-}
-
-// ExtractEntitiesActivity extracts entities from text using pattern matching
-// and wiki-link detection. LLM extraction is a future enhancement.
-// Stage 2: Returns entities with confidence scores.
-func (s *SynthesisActivities) ExtractEntitiesActivity(ctx context.Context, chunkID string, text string) ([]ExtractedEntity, error) {
-	logger := slog.Default()
-	logger.Info("extracting entities", "chunk_id", chunkID)
-
-	entities := make([]ExtractedEntity, 0)
-	seen := make(map[string]bool)
-
-	// Pattern 1: Wiki-link extraction [[EntityName]]
-	wikiPattern := regexp.MustCompile(`\[\[([^\]]+)\]\]`)
-	for _, match := range wikiPattern.FindAllStringSubmatch(text, -1) {
-		name := strings.TrimSpace(match[1])
-		if !seen[name] {
-			entities = append(entities, ExtractedEntity{
-				Name:       name,
-				EntityType: "reference",
-				Confidence: 0.95,
-			})
-			seen[name] = true
-		}
-	}
-
-	// Pattern 2: Capitalized proper nouns (simple NER)
-	properNounPattern := regexp.MustCompile(`\b([A-Z][a-z]+(?:\s+[A-Z][a-z]+)*)\b`)
-	for _, match := range properNounPattern.FindAllStringSubmatch(text, -1) {
-		name := match[1]
-		if !seen[name] && !isCommonWord(name) && len(name) > 2 {
-			entities = append(entities, ExtractedEntity{
-				Name:       name,
-				EntityType: classifyEntity(name),
-				Confidence: 0.70,
-			})
-			seen[name] = true
-		}
-	}
-
-	// Pattern 3: Technical terms (ALL_CAPS or camelCase)
-	techPattern := regexp.MustCompile(`\b([A-Z][A-Z_]{2,}|[a-z]+[A-Z][a-zA-Z]+)\b`)
-	for _, match := range techPattern.FindAllStringSubmatch(text, -1) {
-		name := match[1]
-		if !seen[name] {
-			entities = append(entities, ExtractedEntity{
-				Name:       name,
-				EntityType: "technical",
-				Confidence: 0.65,
-			})
-			seen[name] = true
-		}
-	}
-
-	logger.Info("entities extracted", "count", len(entities))
-	return entities, nil
-}
-
-// ExtractFactsActivity extracts subject-predicate-object facts from text.
-// Stage 3: Pattern-based extraction with entity context.
-func (s *SynthesisActivities) ExtractFactsActivity(ctx context.Context, chunkID string, text string, entities []ExtractedEntity) ([]ExtractedFact, error) {
-	logger := slog.Default()
-	logger.Info("extracting facts", "chunk_id", chunkID, "entity_count", len(entities))
-
-	facts := make([]ExtractedFact, 0)
-
-	// Build entity name set for matching
-	entityNames := make(map[string]bool)
-	for _, e := range entities {
-		entityNames[strings.ToLower(e.Name)] = true
-	}
-
-	// Pattern: "X uses/runs/has Y"
-	verbPatterns := []struct {
-		pattern   *regexp.Regexp
-		predicate string
-	}{
-		{regexp.MustCompile(`(?i)(\w+(?:\s+\w+)?)\s+uses?\s+(.+?)(?:\.|,|$)`), "uses"},
-		{regexp.MustCompile(`(?i)(\w+(?:\s+\w+)?)\s+runs?\s+(?:on\s+)?(.+?)(?:\.|,|$)`), "runs_on"},
-		{regexp.MustCompile(`(?i)(\w+(?:\s+\w+)?)\s+(?:has|have)\s+(.+?)(?:\.|,|$)`), "has"},
-		{regexp.MustCompile(`(?i)(\w+(?:\s+\w+)?)\s+(?:is|are)\s+(.+?)(?:\.|,|$)`), "is"},
-		{regexp.MustCompile(`(?i)(\w+(?:\s+\w+)?)\s+(?:depends?\s+on|requires?)\s+(.+?)(?:\.|,|$)`), "depends_on"},
-		{regexp.MustCompile(`(?i)(\w+(?:\s+\w+)?)\s+(?:connects?\s+to|talks?\s+to)\s+(.+?)(?:\.|,|$)`), "connects_to"},
-	}
-
-	for _, vp := range verbPatterns {
-		for _, match := range vp.pattern.FindAllStringSubmatch(text, -1) {
-			subject := strings.TrimSpace(match[1])
-			object := strings.TrimSpace(match[2])
-
-			// Validation: skip empty or invalid extracts
-			if len(subject) == 0 || len(object) == 0 {
-				continue // Skip empty subject/object
-			}
-
-			// Truncate overly long objects (avoid capturing entire sentence)
-			if len(object) > 500 {
-				logger.Info("truncating long object", "original_len", len(object), "subject", subject, "predicate", vp.predicate)
-				object = object[:500]
-			}
-
-			// Truncate overly long subjects
-			if len(subject) > 200 {
-				logger.Info("truncating long subject", "original_len", len(subject), "predicate", vp.predicate)
-				subject = subject[:200]
-			}
-
-			// Boost confidence if subject/object are known entities
-			confidence := 0.60
-			if entityNames[strings.ToLower(subject)] {
-				confidence += 0.15
-			}
-			if entityNames[strings.ToLower(object)] {
-				confidence += 0.15
-			}
-
-			facts = append(facts, ExtractedFact{
-				Subject:    subject,
-				Predicate:  vp.predicate,
-				Object:     object,
-				Confidence: confidence,
-			})
-		}
-	}
-
-	logger.Info("facts extracted", "count", len(facts))
-	return facts, nil
-}
-
-// DetectContradictionsActivity detects contradictions between new facts
-// and existing knowledge. Uses pre-filter to avoid unnecessary comparisons.
-// Stage 4: Returns contradictions with severity and review status.
-func (s *SynthesisActivities) DetectContradictionsActivity(ctx context.Context, project string, facts []ExtractedFact) ([]ContradictionResult, error) {
-	logger := slog.Default()
-	logger.Info("detecting contradictions", "project", project, "fact_count", len(facts))
-
-	contradictions := make([]ContradictionResult, 0)
-
-	for _, fact := range facts {
-		// Query existing facts about the same subject
-		query := fmt.Sprintf("%s %s", fact.Subject, fact.Predicate)
-		results, err := s.memClient.Query(ctx, &memory.QueryRequest{
-			Project:     project,
-			Query:       query,
-			LevelFilter: []string{"L1", "L2"},
-			Floor:       0.7,
-			Limit:       5,
-		})
-		if err != nil {
-			logger.Warn("query for contradictions failed", "error", err, "subject", fact.Subject)
-			continue // Non-fatal: skip this fact
-		}
-
-		for _, r := range results.Results {
-			// Pre-filter: check if result mentions same subject + different object
-			if containsSubject(r.Text, fact.Subject) && contradicts(r.Text, fact) {
-				severity := "low"
-				if r.Score > 0.9 {
-					severity = "high"
-				} else if r.Score > 0.8 {
-					severity = "medium"
-				}
-
-				autoResolved := severity == "low"
-				contradictions = append(contradictions, ContradictionResult{
-					FactA: ExtractedFact{
-						Subject:   fact.Subject,
-						Predicate: fact.Predicate,
-						Object:    r.Text,
-					},
-					FactB:        fact,
-					Severity:     severity,
-					AutoResolved: autoResolved,
-					QueuedReview: !autoResolved,
-				})
-			}
-		}
-	}
-
-	logger.Info("contradictions detected", "count", len(contradictions))
-	return contradictions, nil
-}
-
-// PersistSynthesisActivity saves all synthesis results to the memory service.
-// Stage 5: Persists entities, facts, and queues contradictions for review.
-// Returns error if any persistence fails (fail-safe semantics).
-func (s *SynthesisActivities) PersistSynthesisActivity(ctx context.Context, input PersistInput) error {
-	logger := slog.Default()
-	logger.Info("persisting synthesis results",
-		"chunk_id", input.ChunkID,
-		"entities", len(input.Entities),
-		"facts", len(input.Facts),
-		"contradictions", len(input.Contradictions),
-	)
-
-	var errs []error
-
-	// Persist entities as knowledge records
-	for _, entity := range input.Entities {
-		_, err := s.memClient.Ingest(ctx, &memory.IngestRequest{
-			Project: input.Project,
-			Source:  input.Source,
-			Kind:    "L1",
-			Text:    fmt.Sprintf("Entity: %s (type: %s, confidence: %.2f)", entity.Name, entity.EntityType, entity.Confidence),
-			Metadata: map[string]interface{}{
-				"chunk_id":    input.ChunkID,
-				"entity_type": entity.EntityType,
-				"entity_name": entity.Name,
-			},
-		})
-		if err != nil {
-			logger.Error("failed to persist entity", "entity", entity.Name, "error", err)
-			errs = append(errs, fmt.Errorf("persist entity %s: %w", entity.Name, err))
-		}
-	}
-
-	// Persist facts
-	for _, fact := range input.Facts {
-		_, err := s.memClient.Ingest(ctx, &memory.IngestRequest{
-			Project: input.Project,
-			Source:  input.Source,
-			Kind:    "L1",
-			Text:    fmt.Sprintf("%s %s %s", fact.Subject, fact.Predicate, fact.Object),
-			Metadata: map[string]interface{}{
-				"chunk_id":  input.ChunkID,
-				"subject":   fact.Subject,
-				"predicate": fact.Predicate,
-				"object":    fact.Object,
-			},
-		})
-		if err != nil {
-			logger.Error("failed to persist fact", "subject", fact.Subject, "predicate", fact.Predicate, "error", err)
-			errs = append(errs, fmt.Errorf("persist fact %s %s: %w", fact.Subject, fact.Predicate, err))
-		}
-	}
-
-	// Return all accumulated errors (fail-safe semantics)
-	if len(errs) > 0 {
-		logger.Error("persistence failed with errors", "error_count", len(errs), "chunk_id", input.ChunkID)
-		return fmt.Errorf("persist synthesis: %d errors - %v", len(errs), errs)
-	}
-
-	logger.Info("synthesis persisted successfully", "chunk_id", input.ChunkID)
-	return nil
-}
-
-// --- helpers ---
-
-func isCommonWord(word string) bool {
-	common := map[string]bool{
-		"The": true, "This": true, "That": true, "These": true,
-		"There": true, "When": true, "Where": true, "What": true,
-		"Which": true, "How": true, "But": true, "And": true,
-		"For": true, "Not": true, "You": true, "All": true,
-		"Can": true, "Her": true, "Was": true, "One": true,
-		"Our": true, "Out": true, "Are": true, "Has": true,
-		"Its": true, "May": true, "New": true, "Now": true,
-		"Old": true, "See": true, "Way": true, "Who": true,
-	}
-	return common[word]
-}
-
-func classifyEntity(name string) string {
-	toolPatterns := []string{"Kubernetes", "Docker", "Nginx", "Redis", "Postgres", "ArgoCD", "Terraform", "Helm"}
-	for _, t := range toolPatterns {
-		if strings.EqualFold(name, t) {
-			return "tool"
-		}
-	}
-	return "concept"
-}
-
-func containsSubject(text, subject string) bool {
-	return strings.Contains(strings.ToLower(text), strings.ToLower(subject))
-}
-
-func contradicts(existingText string, newFact ExtractedFact) bool {
-	// Simple heuristic: if existing text mentions subject with a different value
-	// for the same predicate pattern, it might contradict
-	lower := strings.ToLower(existingText)
-	subjectLower := strings.ToLower(newFact.Subject)
-	objectLower := strings.ToLower(newFact.Object)
-
-	// If text mentions subject but NOT the same object, potential contradiction
-	return strings.Contains(lower, subjectLower) && !strings.Contains(lower, objectLower)
-}

activity/synthesis_test.go

@@ -1,183 +0,0 @@
-package activity
-
-import (
-	"context"
-	"strings"
-	"testing"
-
-	"github.com/stretchr/testify/assert"
-)
-
-var testCtx = context.Background()
-
-func TestExtractEntities_WikiLinks(t *testing.T) {
-	sa := NewSynthesisActivities(nil) // No client needed for extraction
-	entities, err := sa.ExtractEntitiesActivity(testCtx, "chunk-1", "Deploy [[Kubernetes]] with [[ArgoCD]]")
-	assert.NoError(t, err)
-
-	names := entityNames(entities)
-	assert.Contains(t, names, "Kubernetes")
-	assert.Contains(t, names, "ArgoCD")
-
-	// Wiki links get high confidence
-	for _, e := range entities {
-		if e.Name == "Kubernetes" || e.Name == "ArgoCD" {
-			assert.Equal(t, 0.95, e.Confidence)
-			assert.Equal(t, "reference", e.EntityType)
-		}
-	}
-}
-
-func TestExtractEntities_ProperNouns(t *testing.T) {
-	sa := NewSynthesisActivities(nil)
-	entities, err := sa.ExtractEntitiesActivity(testCtx, "chunk-2", "Redis runs on Ubuntu Server")
-	assert.NoError(t, err)
-
-	names := entityNames(entities)
-	assert.Contains(t, names, "Redis")
-	assert.Contains(t, names, "Ubuntu Server")
-}
-
-func TestExtractEntities_TechnicalTerms(t *testing.T) {
-	sa := NewSynthesisActivities(nil)
-	entities, err := sa.ExtractEntitiesActivity(testCtx, "chunk-3", "Set MAX_RETRIES and use camelCase variables")
-	assert.NoError(t, err)
-
-	names := entityNames(entities)
-	assert.Contains(t, names, "MAX_RETRIES")
-	assert.Contains(t, names, "camelCase")
-}
-
-func TestExtractEntities_Deduplication(t *testing.T) {
-	sa := NewSynthesisActivities(nil)
-	entities, err := sa.ExtractEntitiesActivity(testCtx, "chunk-4", "[[Redis]] uses Redis for caching")
-	assert.NoError(t, err)
-
-	count := 0
-	for _, e := range entities {
-		if e.Name == "Redis" {
-			count++
-		}
-	}
-	assert.Equal(t, 1, count, "Redis should appear only once")
-}
-
-func TestExtractFacts_VerbPatterns(t *testing.T) {
-	sa := NewSynthesisActivities(nil)
-	entities := []ExtractedEntity{
-		{Name: "Kubernetes", EntityType: "tool"},
-		{Name: "Docker", EntityType: "tool"},
-	}
-	facts, err := sa.ExtractFactsActivity(testCtx, "chunk-5",
-		"Kubernetes uses Docker for container runtime. Redis depends on TCP",
-		entities)
-	assert.NoError(t, err)
-	assert.Greater(t, len(facts), 0)
-
-	// Find the "uses" fact
-	found := false
-	for _, f := range facts {
-		if f.Predicate == "uses" && f.Subject == "Kubernetes" {
-			found = true
-			assert.Greater(t, f.Confidence, 0.7) // Boosted by known entities
-		}
-	}
-	assert.True(t, found, "should find Kubernetes uses Docker fact")
-}
-
-func TestExtractFacts_EmptyText(t *testing.T) {
-	sa := NewSynthesisActivities(nil)
-	facts, err := sa.ExtractFactsActivity(testCtx, "chunk-6", "", nil)
-	assert.NoError(t, err)
-	assert.Empty(t, facts)
-}
-
-func TestContradicts(t *testing.T) {
-	assert.True(t, contradicts("Kubernetes uses port 8080", ExtractedFact{
-		Subject: "Kubernetes", Predicate: "uses_port", Object: "9090",
-	}))
-
-	assert.False(t, contradicts("Kubernetes uses port 8080", ExtractedFact{
-		Subject: "Kubernetes", Predicate: "uses_port", Object: "8080",
-	}))
-}
-
-func TestContainsSubject(t *testing.T) {
-	assert.True(t, containsSubject("Kubernetes runs on Linux", "kubernetes"))
-	assert.False(t, containsSubject("Docker runs on Linux", "kubernetes"))
-}
-
-func TestIsCommonWord(t *testing.T) {
-	assert.True(t, isCommonWord("The"))
-	assert.True(t, isCommonWord("This"))
-	assert.False(t, isCommonWord("Kubernetes"))
-	assert.False(t, isCommonWord("Redis"))
-}
-
-func TestClassifyEntity(t *testing.T) {
-	assert.Equal(t, "tool", classifyEntity("Kubernetes"))
-	assert.Equal(t, "tool", classifyEntity("Docker"))
-	assert.Equal(t, "tool", classifyEntity("Redis"))
-	assert.Equal(t, "concept", classifyEntity("SomeRandomThing"))
-}
-
-// --- Tests for ExtractFactsActivity Validation ---
-
-func TestExtractFacts_WithValidation(t *testing.T) {
-	sa := NewSynthesisActivities(nil)
-	// Test with very long object that should be truncated
-	longText := "Kubernetes uses " + strings.Repeat("very long object name that should be truncated ", 20)
-	entities := []ExtractedEntity{}
-
-	facts, err := sa.ExtractFactsActivity(testCtx, "chunk-1", longText, entities)
-	assert.NoError(t, err)
-
-	// Verify no fact has object > 500 chars
-	for _, f := range facts {
-		assert.LessOrEqual(t, len(f.Object), 500, "object should be truncated to 500 chars")
-	}
-}
-
-func TestExtractFacts_SkipsEmpty(t *testing.T) {
-	sa := NewSynthesisActivities(nil)
-	// Text with empty patterns that would extract nothing
-	text := "Something uses  and other things"
-	entities := []ExtractedEntity{}
-
-	facts, err := sa.ExtractFactsActivity(testCtx, "chunk-1", text, entities)
-	assert.NoError(t, err)
-
-	// Verify no empty facts
-	for _, f := range facts {
-		assert.NotEmpty(t, f.Subject, "subject should not be empty")
-		assert.NotEmpty(t, f.Object, "object should not be empty")
-	}
-}
-
-func TestPersistSynthesis_EmptyInput(t *testing.T) {
-	// Test that empty input is handled (no entities or facts to persist)
-	// Note: This test requires a mock memory service; for now we just test structure
-	input := PersistInput{
-		ChunkID:        "chunk-123",
-		Project:        "test",
-		Source:         "test://1",
-		Kind:           "L1",
-		Entities:       []ExtractedEntity{}, // Empty
-		Facts:          []ExtractedFact{},  // Empty
-		Contradictions: []ContradictionResult{},
-	}
-
-	// Verify input structure is valid
-	assert.Equal(t, "chunk-123", input.ChunkID)
-	assert.Equal(t, 0, len(input.Entities))
-	assert.Equal(t, 0, len(input.Facts))
-}
-
-// helper
-func entityNames(entities []ExtractedEntity) []string {
-	names := make([]string, len(entities))
-	for i, e := range entities {
-		names[i] = e.Name
-	}
-	return names
-}

activity/types.go

@@ -1,21 +0,0 @@
-package activity
-
-import "github.com/rockliang/poimen/workflows/pkg/types"
-
-// Re-export from pkg/types for convenience within activity package.
-type ModelSpec = types.ModelSpec
-type PromptSpec = types.PromptSpec
-type SkillRef = types.SkillRef
-type OrchestratorConfig = types.OrchestratorConfig
-type TaskUnitOutput = types.TaskUnitOutput
-type EdgeWithWording = types.EdgeWithWording
-type RelationWording = types.RelationWording
-type CanvasWithRelationsData = types.CanvasWithRelationsData
-type FetchCanvasRelationsInput = types.FetchCanvasRelationsInput
-type CanvasReasonerInput = types.CanvasReasonerInput
-type GraphRAGQueryInput = types.GraphRAGQueryInput
-type GraphRAGQueryOutput = types.GraphRAGQueryOutput
-type QueryPathData = types.QueryPathData
-type CanvasCompatibilityInput = types.CanvasCompatibilityInput
-type IndexGraphRAGInput = types.IndexGraphRAGInput
-type IndexGraphRAGOutput = types.IndexGraphRAGOutput

cmd/server/main.go

@@ -0,0 +1,125 @@
+package main
+
+import (
+	"context"
+	"flag"
+	"log"
+	"os"
+	"os/signal"
+	"sync"
+	"syscall"
+
+	"go.temporal.io/sdk/client"
+	"go.temporal.io/sdk/worker"
+
+	"github.com/rockliang/poimen/workflows/action"
+	"github.com/rockliang/poimen/workflows/internal/api"
+	"github.com/rockliang/poimen/workflows/internal/config"
+	"github.com/rockliang/poimen/workflows/pkg/db"
+	"github.com/rockliang/poimen/workflows/statemachine"
+)
+
+func main() {
+	var (
+		apiPort = flag.Int("port", 8080, "HTTP API port")
+		verbose = flag.Bool("verbose", false, "verbose logging")
+	)
+	flag.Parse()
+
+	logger := log.New(os.Stdout, "[poimen-server] ", log.LstdFlags|log.Lshortfile)
+
+	// Load configuration
+	cfg, err := config.LoadConfig()
+	if err != nil {
+		logger.Fatalf("failed to load config: %v", err)
+	}
+
+	// Connect to database (memory-db via K8s CNPG)
+	logger.Println("connecting to database...")
+	database, err := db.New(os.Getenv("DATABASE_URL"))
+	if err != nil {
+		logger.Fatalf("failed to connect to database: %v", err)
+	}
+	defer database.Close()
+	logger.Println("✓ Connected to database")
+
+	// Connect to Temporal
+	logger.Printf("connecting to Temporal at %s", cfg.Temporal.HostPort)
+	c, err := client.Dial(client.Options{
+		HostPort:  cfg.Temporal.HostPort,
+		Namespace: cfg.Temporal.Namespace,
+	})
+	if err != nil {
+		logger.Fatalf("failed to connect to temporal: %v", err)
+	}
+	defer c.Close()
+
+	logger.Println("✓ Connected to Temporal")
+
+	// Create and start Temporal worker
+	w := worker.New(c, "default", worker.Options{})
+
+	// Register RoutingWorkflow
+	w.RegisterWorkflow(statemachine.RoutingWorkflow)
+
+	// Register activities
+	w.RegisterActivity(action.CloneRepoActivity)
+	w.RegisterActivity(action.AnalyzeCodeActivity)
+	w.RegisterActivity(action.SecurityScanActivity)
+	w.RegisterActivity(action.GenerateReportActivity)
+	w.RegisterActivity(action.DeploymentPreCheckActivity)
+	w.RegisterActivity(action.NotifyStatusActivity)
+	w.RegisterActivity(action.ApproveWorkflowActivity)
+	w.RegisterActivity(action.ArchiveResultsActivity)
+	w.RegisterActivity(action.RetrieveMemoryActivity)
+	w.RegisterActivity(action.AssumeRoleActivity)
+	w.RegisterActivity(action.LLMInferenceActivity)
+	w.RegisterActivity(action.LLMBatchInferenceActivity)
+	w.RegisterActivity(action.CanvasReasonerActivity)
+
+	var wg sync.WaitGroup
+	errChan := make(chan error, 2)
+
+	// Start Temporal worker
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		logger.Println("starting Temporal worker...")
+		if err := w.Run(worker.InterruptCh()); err != nil {
+			errChan <- err
+		}
+	}()
+
+	// Start HTTP API server
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		server := api.NewServer(database, c, logger)
+		logger.Printf("starting API server on port %d", *apiPort)
+		if err := server.Start(*apiPort); err != nil {
+			errChan <- err
+		}
+	}()
+
+	// Wait for interrupt signal
+	sigChan := make(chan os.Signal, 1)
+	signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)
+
+	go func() {
+		sig := <-sigChan
+		logger.Printf("received signal: %v", sig)
+		w.Stop()
+	}()
+
+	// Monitor for errors
+	go func() {
+		err := <-errChan
+		if err != nil {
+			logger.Printf("error: %v", err)
+			w.Stop()
+		}
+	}()
+
+	wg.Wait()
+	logger.Println("✓ Server stopped gracefully")
+}

cmd/starter/main.go

@@ -11,12 +11,12 @@ import (
 	"time"
 
 	"go.temporal.io/sdk/client"
-	"github.com/rockliang/poimen/workflows/activity/llm"
+	"github.com/rockliang/poimen/workflows/action/llm"
 	"github.com/rockliang/poimen/workflows/internal/config"
 	"github.com/rockliang/poimen/workflows/internal/health"
 	"github.com/rockliang/poimen/workflows/internal/logging"
 	"github.com/rockliang/poimen/workflows/internal/routing"
-	"github.com/rockliang/poimen/workflows/workflow"
+	"github.com/rockliang/poimen/workflows/statemachine"
 )
 
 func main() {
@@ -89,20 +89,20 @@ func main() {
 
 
 	// Build OrchestratorInput
-	input := workflow.OrchestratorInput{
+	input := statemachine.OrchestratorInput{
 		TargetRepoPath:     *repoPath,
 		RemoteURL:          *remoteURL,
 		Milestone:          *milestone,
 		DryRun:             *dryRun,
 		MaxCyclesBeforeCAN: 100,
 		PiProvider:         *piProvider,
-		Config: workflow.OrchestratorConfig{
+		Config: statemachine.OrchestratorConfig{
 			SystemPrompt: "You are an expert software developer orchestrating multi-agent work.",
-			Skills:       []workflow.SkillRef{},
-			RolePrompts: map[string]workflow.PromptSpec{
+			Skills:       []statemachine.SkillRef{},
+			RolePrompts: map[string]statemachine.PromptSpec{
 				"planner": {
 					TemplateRef: "planner/default.tmpl",
-					Model: workflow.ModelSpec{
+					Model: statemachine.ModelSpec{
 						ModelID:  *plannerModel,
 						Thinking: "adaptive",
 						Effort:   "high",
@@ -110,7 +110,7 @@ func main() {
 				},
 				"judge": {
 					TemplateRef: "judge/default.tmpl",
-					Model: workflow.ModelSpec{
+					Model: statemachine.ModelSpec{
 						ModelID:  *judgeModel,
 						Thinking: "adaptive",
 						Effort:   "high",
@@ -118,12 +118,12 @@ func main() {
 				},
 				"implementer": {
 					TemplateRef: "implementer/default.tmpl",
-					Model: workflow.ModelSpec{
+					Model: statemachine.ModelSpec{
 						ModelID: *implementerModel,
 					},
 				},
 			},
-			Tuning: workflow.NewActivityTuning(),
+			Tuning: statemachine.NewActivityTuning(),
 		},
 	}
 
@@ -144,7 +144,7 @@ func main() {
 	run, err := c.ExecuteWorkflow(context.Background(), client.StartWorkflowOptions{
 		ID:        workflowID,
 		TaskQueue: "poimen-taskqueue",
-	}, workflow.OrchestratorWorkflow, input)
+	}, statemachine.OrchestratorWorkflow, input)
 	if err != nil {
 		logging.Fatal("failed to start workflow", logging.Err(err))
 	}
@@ -164,7 +164,7 @@ func main() {
 	ctx, cancel := context.WithTimeout(context.Background(), 1*time.Minute)
 	defer cancel()
 
-	var result workflow.OrchestratorOutput
+	var result statemachine.OrchestratorOutput
 	if err := run.Get(ctx, &result); err != nil {
 		fmt.Printf("\nWorkflow initiated (execution in progress).\n")
 		fmt.Printf("Check the Web UI for real-time status updates.\n")
@@ -267,12 +267,12 @@ func runRoutingWorkflow(c client.Client, routeMsg, specFile string, isCron, dryR
 	}
 
 	workflowID := "routing-" + spec.Name + "-" + time.Now().Format("20060102-150405")
-	input := workflow.RoutingWorkflowInput{Spec: spec}
+	input := statemachine.RoutingWorkflowInput{Spec: spec}
 
 	run, err := c.ExecuteWorkflow(ctx, client.StartWorkflowOptions{
 		ID:        workflowID,
 		TaskQueue: "poimen-taskqueue",
-	}, workflow.RoutingWorkflow, input)
+	}, statemachine.RoutingWorkflow, input)
 	if err != nil {
 		logging.Fatal("failed to start routing workflow", logging.Err(err))
 	}
@@ -285,7 +285,7 @@ func runRoutingWorkflow(c client.Client, routeMsg, specFile string, isCron, dryR
 	waitCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
 	defer cancel()
 
-	var result workflow.RoutingWorkflowOutput
+	var result statemachine.RoutingWorkflowOutput
 	if err := run.Get(waitCtx, &result); err != nil {
 		fmt.Printf("\nWorkflow running (check Temporal UI for status)\n")
 	} else {

cmd/worker/main.go

@@ -11,11 +11,11 @@ import (
 
 	"go.temporal.io/sdk/client"
 	"go.temporal.io/sdk/worker"
-	"github.com/rockliang/poimen/workflows/activity"
+	"github.com/rockliang/poimen/workflows/action"
 	"github.com/rockliang/poimen/workflows/internal/config"
 	"github.com/rockliang/poimen/workflows/internal/health"
 	"github.com/rockliang/poimen/workflows/internal/logging"
-	"github.com/rockliang/poimen/workflows/workflow"
+	"github.com/rockliang/poimen/workflows/statemachine"
 )
 
 func main() {
@@ -48,56 +48,56 @@ func main() {
 	}
 
 	// Register all workflows
-	w.RegisterWorkflow(workflow.OrchestratorWorkflow)
-	w.RegisterWorkflow(workflow.TaskUnitWorkflow)
-	w.RegisterWorkflow(workflow.TestWorkflow)
-	w.RegisterWorkflow(workflow.RoutingWorkflow)
-	w.RegisterWorkflow(workflow.WorkflowGraphQuery)
+	w.RegisterWorkflow(statemachine.OrchestratorWorkflow)
+	w.RegisterWorkflow(statemachine.TaskUnitWorkflow)
+	w.RegisterWorkflow(statemachine.TestWorkflow)
+	w.RegisterWorkflow(statemachine.RoutingWorkflow)
+	w.RegisterWorkflow(statemachine.WorkflowGraphQuery)
 
 	// Register all activities
-	w.RegisterActivity(activity.CloneRepoActivity)
-	w.RegisterActivity(activity.GitWorktreeAddActivity)
-	w.RegisterActivity(activity.GitCommitActivity)
-	w.RegisterActivity(activity.GitPushActivity)
-	w.RegisterActivity(activity.GitSquashMergeActivity)
-	w.RegisterActivity(activity.GitDiffActivity)
-	w.RegisterActivity(activity.PrepareSkillsActivity)
-	w.RegisterActivity(activity.PlanningActivity)
-	w.RegisterActivity(activity.ImplementerActivity)
-	w.RegisterActivity(activity.JudgeActivity)
+	w.RegisterActivity(action.CloneRepoActivity)
+	w.RegisterActivity(action.GitWorktreeAddActivity)
+	w.RegisterActivity(action.GitCommitActivity)
+	w.RegisterActivity(action.GitPushActivity)
+	w.RegisterActivity(action.GitSquashMergeActivity)
+	w.RegisterActivity(action.GitDiffActivity)
+	w.RegisterActivity(action.PrepareSkillsActivity)
+	w.RegisterActivity(action.PlanningActivity)
+	w.RegisterActivity(action.ImplementerActivity)
+	w.RegisterActivity(action.JudgeActivity)
 	// Integration and lessons activities - register when fully tested
-	w.RegisterActivity(activity.RunIntegrationTestActivity)
-	// w.RegisterActivity(activity.UpdateLessonsActivity)
-	// w.RegisterActivity(activity.ReadLessonsActivity)
+	w.RegisterActivity(action.RunIntegrationTestActivity)
+	// w.RegisterActivity(action.UpdateLessonsActivity)
+	// w.RegisterActivity(action.ReadLessonsActivity)
 
 	// Routing workflow activities
-	w.RegisterActivity(activity.LLMRouterActivity)
-	w.RegisterActivity(activity.ValidateWorkflowSpecActivity)
-	w.RegisterActivity(activity.ValidateCronWorkflowSpecActivity)
+	w.RegisterActivity(action.LLMRouterActivity)
+	w.RegisterActivity(action.ValidateWorkflowSpecActivity)
+	w.RegisterActivity(action.ValidateCronWorkflowSpecActivity)
 
 	// Analysis activities
-	w.RegisterActivity(activity.AnalyzeCodeActivity)
-	w.RegisterActivity(activity.SecurityScanActivity)
-	w.RegisterActivity(activity.GenerateReportActivity)
+	w.RegisterActivity(action.AnalyzeCodeActivity)
+	w.RegisterActivity(action.SecurityScanActivity)
+	w.RegisterActivity(action.GenerateReportActivity)
 
 	// Notification and utility activities
-	w.RegisterActivity(activity.NotifyStatusActivity)
-	w.RegisterActivity(activity.ArchiveResultsActivity)
-	w.RegisterActivity(activity.DeploymentPreCheckActivity)
-	w.RegisterActivity(activity.ApproveWorkflowActivity)
+	w.RegisterActivity(action.NotifyStatusActivity)
+	w.RegisterActivity(action.ArchiveResultsActivity)
+	w.RegisterActivity(action.DeploymentPreCheckActivity)
+	w.RegisterActivity(action.ApproveWorkflowActivity)
 
 	// Authentication activities
-	w.RegisterActivity(activity.AssumeRoleActivity)
+	w.RegisterActivity(action.AssumeRoleActivity)
 
 	// Memory activities
-	w.RegisterActivity(activity.RetrieveMemoryActivity)
+	w.RegisterActivity(action.RetrieveMemoryActivity)
 
 	// GraphRAG activities
-	w.RegisterActivity(activity.FetchCanvasRelationsActivity)
-	w.RegisterActivity(activity.QueryGraphRAGActivity)
-	w.RegisterActivity(activity.CanvasReasonerActivity)
-	w.RegisterActivity(activity.IndexGraphRAGActivity)
-	w.RegisterActivity(activity.CanvasCompatibilityActivity)
+	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)

internal/api/server.go

@@ -0,0 +1,117 @@
+package api
+
+import (
+	"fmt"
+	"log"
+	"net/http"
+	"strings"
+
+	"go.temporal.io/sdk/client"
+
+	"github.com/rockliang/poimen/workflows/pkg/db"
+)
+
+// Server handles HTTP routing for workflow APIs
+type Server struct {
+	api    *WorkflowAPI
+	logger *log.Logger
+}
+
+// NewServer creates new HTTP server with database connection
+func NewServer(database *db.DB, temporalClient client.Client, logger *log.Logger) *Server {
+	return &Server{
+		api:    NewWorkflowAPI(database, temporalClient, logger),
+		logger: logger,
+	}
+}
+
+// ServeHTTP dispatches HTTP requests to appropriate handler
+func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	// Enable CORS
+	w.Header().Set("Access-Control-Allow-Origin", "*")
+	w.Header().Set("Access-Control-Allow-Methods", "GET, POST, PUT, DELETE, OPTIONS")
+	w.Header().Set("Access-Control-Allow-Headers", "Content-Type, Authorization")
+
+	if r.Method == http.MethodOptions {
+		w.WriteHeader(http.StatusOK)
+		return
+	}
+
+	path := r.URL.Path
+	method := r.Method
+
+	s.logger.Printf("%s %s", method, path)
+
+	// Route requests
+	switch {
+	// Workflow endpoints
+	case path == "/workflows" && method == http.MethodPost:
+		s.api.CreateWorkflow(w, r)
+	case path == "/workflows" && method == http.MethodGet:
+		s.api.ListWorkflows(w, r)
+	case strings.HasPrefix(path, "/workflows/") && method == http.MethodGet:
+		id := strings.TrimPrefix(path, "/workflows/")
+		// Exclude special paths
+		if !strings.Contains(id, "/") {
+			s.api.GetWorkflow(w, r, id)
+		} else if strings.HasSuffix(id, "/executions") {
+			// GET /workflows/{id}/executions
+			workflowID := strings.TrimSuffix(id, "/executions")
+			s.api.ListExecutions(w, r, workflowID)
+		}
+	case strings.HasPrefix(path, "/workflows/") && method == http.MethodPut:
+		id := extractID(path, "/workflows/")
+		s.api.UpdateWorkflow(w, r, id)
+	case strings.HasPrefix(path, "/workflows/") && method == http.MethodDelete:
+		id := extractID(path, "/workflows/")
+		s.api.DeleteWorkflow(w, r, id)
+
+	// Execute workflow
+	case strings.HasSuffix(path, "/execute") && method == http.MethodPost:
+		// POST /workflows/{id}/execute
+		parts := strings.Split(path, "/")
+		if len(parts) >= 4 && parts[1] == "workflows" && parts[3] == "execute" {
+			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/")
+		s.api.GetExecution(w, r, id)
+
+	default:
+		http.Error(w, "Not found", http.StatusNotFound)
+	}
+}
+
+// extractID extracts resource ID from path
+func extractID(path, prefix string) string {
+	id := strings.TrimPrefix(path, prefix)
+	if idx := strings.Index(id, "/"); idx != -1 {
+		return id[:idx]
+	}
+	return id
+}
+
+// Start starts the HTTP server
+func (s *Server) Start(port int) error {
+	addr := fmt.Sprintf(":%d", port)
+	s.logger.Printf("Starting API server on %s", addr)
+	return http.ListenAndServe(addr, s)
+}

internal/api/workflows.go

@@ -0,0 +1,704 @@
+package api
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"log"
+	"net/http"
+	"time"
+
+	"github.com/google/uuid"
+	"go.temporal.io/sdk/client"
+
+	"github.com/rockliang/poimen/workflows/internal/routing"
+	"github.com/rockliang/poimen/workflows/pkg/db"
+)
+
+// WorkflowNode matches frontend node type
+type WorkflowNode struct {
+	ID       string                 `json:"id"`
+	Type     string                 `json:"type"` // "activity", "start", "end"
+	Position map[string]interface{} `json:"position"`
+	Data     struct {
+		Label    string                 `json:"label"`
+		Activity string                 `json:"activity"`
+		Config   map[string]interface{} `json:"config"`
+	} `json:"data"`
+}
+
+// WorkflowEdge matches frontend edge type
+type WorkflowEdge struct {
+	ID     string                 `json:"id"`
+	Source string                 `json:"source"`
+	Target string                 `json:"target"`
+	Data   map[string]interface{} `json:"data,omitempty"`
+}
+
+// WorkflowDef is the request body for creating/updating workflows
+type WorkflowDef struct {
+	Name        string          `json:"name"`
+	Description string          `json:"description"`
+	Nodes       []WorkflowNode  `json:"nodes"`
+	Edges       []WorkflowEdge  `json:"edges"`
+	Status      string          `json:"status"` // "draft", "active"
+}
+
+// WorkflowResponse is the workflow with metadata
+type WorkflowResponse struct {
+	ID          string          `json:"id"`
+	Name        string          `json:"name"`
+	Description string          `json:"description"`
+	Status      string          `json:"status"`
+	Version     int             `json:"version"`
+	Nodes       []WorkflowNode  `json:"nodes"`
+	Edges       []WorkflowEdge  `json:"edges"`
+	CreatedAt   string          `json:"createdAt"`
+	UpdatedAt   string          `json:"updatedAt"`
+	CreatedBy   string          `json:"createdBy"`
+}
+
+// ExecutionRequest is the request to execute a workflow
+type ExecutionRequest struct {
+	Inputs map[string]interface{} `json:"inputs"`
+}
+
+// ExecutionResponse is the execution result
+type ExecutionResponse struct {
+	ID           string                 `json:"id"`
+	WorkflowID   string                 `json:"workflowId"`
+	Status       string                 `json:"status"` // "pending", "running", "success", "failed"
+	StartedAt    string                 `json:"startedAt"`
+	CompletedAt  string                 `json:"completedAt,omitempty"`
+	Inputs       map[string]interface{} `json:"inputs"`
+	Outputs      map[string]interface{} `json:"outputs,omitempty"`
+	Errors       []string               `json:"errors,omitempty"`
+	Logs         []ExecutionLog         `json:"logs"`
+}
+
+// ExecutionLog is a log entry from execution
+type ExecutionLog struct {
+	Timestamp string `json:"timestamp"`
+	NodeID    string `json:"nodeId"`
+	Level     string `json:"level"` // "info", "warn", "error"
+	Message   string `json:"message"`
+}
+
+// WorkflowAPI handles workflow endpoints
+type WorkflowAPI struct {
+	db             *db.DB
+	temporalClient client.Client
+	logger         *log.Logger
+	customerID     string // TODO: Extract from JWT token
+}
+
+// NewWorkflowAPI creates new API handler
+func NewWorkflowAPI(database *db.DB, tc client.Client, logger *log.Logger) *WorkflowAPI {
+	return &WorkflowAPI{
+		db:             database,
+		temporalClient: tc,
+		logger:         logger,
+		customerID:     "default-customer", // TODO: From auth context
+	}
+}
+
+// CreateWorkflow handles POST /workflows
+func (api *WorkflowAPI) CreateWorkflow(w http.ResponseWriter, r *http.Request) {
+	if r.Method != http.MethodPost {
+		http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
+		return
+	}
+
+	var req WorkflowDef
+	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+		http.Error(w, fmt.Sprintf("Invalid request: %v", err), http.StatusBadRequest)
+		return
+	}
+
+	if req.Name == "" {
+		http.Error(w, "Workflow name required", http.StatusBadRequest)
+		return
+	}
+
+	// Create workflow in database
+	id := uuid.New().String()
+	now := time.Now()
+
+	// Convert nodes and edges to JSONB
+	nodesJSON, err := json.Marshal(req.Nodes)
+	if err != nil {
+		http.Error(w, fmt.Sprintf("Failed to marshal nodes: %v", err), http.StatusBadRequest)
+		return
+	}
+
+	edgesJSON, err := json.Marshal(req.Edges)
+	if err != nil {
+		http.Error(w, fmt.Sprintf("Failed to marshal edges: %v", err), http.StatusBadRequest)
+		return
+	}
+
+	status := req.Status
+	if status == "" {
+		status = "draft"
+	}
+
+	workflow := &db.Workflow{
+		ID:          id,
+		CustomerID:  api.customerID,
+		Name:        req.Name,
+		Description: req.Description,
+		Status:      status,
+		Version:     1,
+		Nodes:       nodesJSON,
+		Edges:       edgesJSON,
+		CreatedBy:   "anonymous", // Use JWT claim in real implementation
+		CreatedAt:   now,
+		UpdatedAt:   now,
+	}
+
+	if err := api.db.SaveWorkflow(r.Context(), workflow); err != nil {
+		api.logger.Printf("Failed to save workflow: %v", err)
+		http.Error(w, "Failed to create workflow", http.StatusInternalServerError)
+		return
+	}
+
+	response := WorkflowResponse{
+		ID:          workflow.ID,
+		Name:        workflow.Name,
+		Description: workflow.Description,
+		Status:      workflow.Status,
+		Version:     workflow.Version,
+		Nodes:       req.Nodes,
+		Edges:       req.Edges,
+		CreatedAt:   workflow.CreatedAt.Format(time.RFC3339),
+		UpdatedAt:   workflow.UpdatedAt.Format(time.RFC3339),
+		CreatedBy:   workflow.CreatedBy,
+	}
+
+	w.Header().Set("Content-Type", "application/json")
+	w.WriteHeader(http.StatusCreated)
+	json.NewEncoder(w).Encode(response)
+}
+
+// ListWorkflows handles GET /workflows
+func (api *WorkflowAPI) ListWorkflows(w http.ResponseWriter, r *http.Request) {
+	if r.Method != http.MethodGet {
+		http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
+		return
+	}
+
+	page := 1
+	limit := 10
+	// Parse pagination params if needed
+
+	workflows, err := api.db.ListWorkflows(r.Context(), api.customerID, limit, (page-1)*limit)
+	if err != nil {
+		api.logger.Printf("Failed to list workflows: %v", err)
+		http.Error(w, "Failed to list workflows", http.StatusInternalServerError)
+		return
+	}
+
+	list := make([]WorkflowResponse, 0)
+	for _, wf := range workflows {
+		var nodes []WorkflowNode
+		var edges []WorkflowEdge
+
+		json.Unmarshal(wf.Nodes, &nodes)
+		json.Unmarshal(wf.Edges, &edges)
+
+		list = append(list, WorkflowResponse{
+			ID:          wf.ID,
+			Name:        wf.Name,
+			Description: wf.Description,
+			Status:      wf.Status,
+			Version:     wf.Version,
+			Nodes:       nodes,
+			Edges:       edges,
+			CreatedAt:   wf.CreatedAt.Format(time.RFC3339),
+			UpdatedAt:   wf.UpdatedAt.Format(time.RFC3339),
+			CreatedBy:   wf.CreatedBy,
+		})
+	}
+
+	response := map[string]interface{}{
+		"workflows": list,
+		"total":     len(list),
+		"page":      page,
+		"limit":     limit,
+	}
+
+	w.Header().Set("Content-Type", "application/json")
+	json.NewEncoder(w).Encode(response)
+}
+
+// GetWorkflow handles GET /workflows/{id}
+func (api *WorkflowAPI) GetWorkflow(w http.ResponseWriter, r *http.Request, id string) {
+	if r.Method != http.MethodGet {
+		http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
+		return
+	}
+
+	workflow, err := api.db.FetchWorkflow(r.Context(), id, api.customerID)
+	if err != nil {
+		http.Error(w, "Workflow not found", http.StatusNotFound)
+		return
+	}
+
+	var nodes []WorkflowNode
+	var edges []WorkflowEdge
+
+	json.Unmarshal(workflow.Nodes, &nodes)
+	json.Unmarshal(workflow.Edges, &edges)
+
+	response := WorkflowResponse{
+		ID:          workflow.ID,
+		Name:        workflow.Name,
+		Description: workflow.Description,
+		Status:      workflow.Status,
+		Version:     workflow.Version,
+		Nodes:       nodes,
+		Edges:       edges,
+		CreatedAt:   workflow.CreatedAt.Format(time.RFC3339),
+		UpdatedAt:   workflow.UpdatedAt.Format(time.RFC3339),
+		CreatedBy:   workflow.CreatedBy,
+	}
+
+	w.Header().Set("Content-Type", "application/json")
+	json.NewEncoder(w).Encode(response)
+}
+
+// UpdateWorkflow handles PUT /workflows/{id}
+func (api *WorkflowAPI) UpdateWorkflow(w http.ResponseWriter, r *http.Request, id string) {
+	if r.Method != http.MethodPut {
+		http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
+		return
+	}
+
+	// Fetch existing workflow
+	workflow, err := api.db.FetchWorkflow(r.Context(), id, api.customerID)
+	if err != nil {
+		http.Error(w, "Workflow not found", http.StatusNotFound)
+		return
+	}
+
+	var req WorkflowDef
+	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+		http.Error(w, fmt.Sprintf("Invalid request: %v", err), http.StatusBadRequest)
+		return
+	}
+
+	// Update fields
+	if req.Name != "" {
+		workflow.Name = req.Name
+	}
+	if req.Description != "" {
+		workflow.Description = req.Description
+	}
+	if req.Nodes != nil {
+		nodesJSON, _ := json.Marshal(req.Nodes)
+		workflow.Nodes = nodesJSON
+	}
+	if req.Edges != nil {
+		edgesJSON, _ := json.Marshal(req.Edges)
+		workflow.Edges = edgesJSON
+	}
+	if req.Status != "" {
+		workflow.Status = req.Status
+	}
+
+	workflow.Version++
+	workflow.UpdatedAt = time.Now()
+
+	if err := api.db.SaveWorkflow(r.Context(), workflow); err != nil {
+		api.logger.Printf("Failed to update workflow: %v", err)
+		http.Error(w, "Failed to update workflow", http.StatusInternalServerError)
+		return
+	}
+
+	var nodes []WorkflowNode
+	var edges []WorkflowEdge
+
+	json.Unmarshal(workflow.Nodes, &nodes)
+	json.Unmarshal(workflow.Edges, &edges)
+
+	response := WorkflowResponse{
+		ID:          workflow.ID,
+		Name:        workflow.Name,
+		Description: workflow.Description,
+		Status:      workflow.Status,
+		Version:     workflow.Version,
+		Nodes:       nodes,
+		Edges:       edges,
+		CreatedAt:   workflow.CreatedAt.Format(time.RFC3339),
+		UpdatedAt:   workflow.UpdatedAt.Format(time.RFC3339),
+		CreatedBy:   workflow.CreatedBy,
+	}
+
+	w.Header().Set("Content-Type", "application/json")
+	json.NewEncoder(w).Encode(response)
+}
+
+// DeleteWorkflow handles DELETE /workflows/{id}
+func (api *WorkflowAPI) DeleteWorkflow(w http.ResponseWriter, r *http.Request, id string) {
+	if r.Method != http.MethodDelete {
+		http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
+		return
+	}
+
+	if err := api.db.DeleteWorkflow(r.Context(), id, api.customerID); err != nil {
+		http.Error(w, "Workflow not found", http.StatusNotFound)
+		return
+	}
+
+	w.WriteHeader(http.StatusNoContent)
+}
+
+// ExecuteWorkflow handles POST /workflows/{id}/execute
+func (api *WorkflowAPI) ExecuteWorkflow(w http.ResponseWriter, r *http.Request, id string) {
+	if r.Method != http.MethodPost {
+		http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
+		return
+	}
+
+	workflow, err := api.db.FetchWorkflow(r.Context(), id, api.customerID)
+	if err != nil {
+		http.Error(w, "Workflow not found", http.StatusNotFound)
+		return
+	}
+
+	var req ExecutionRequest
+	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+		http.Error(w, fmt.Sprintf("Invalid request: %v", err), http.StatusBadRequest)
+		return
+	}
+
+	// Unmarshal nodes and edges
+	var nodes []WorkflowNode
+	var edges []WorkflowEdge
+	json.Unmarshal(workflow.Nodes, &nodes)
+	json.Unmarshal(workflow.Edges, &edges)
+
+	// Convert to workflow response for spec conversion
+	workflowResp := &WorkflowResponse{
+		ID:          workflow.ID,
+		Name:        workflow.Name,
+		Description: workflow.Description,
+		Status:      workflow.Status,
+		Version:     workflow.Version,
+		Nodes:       nodes,
+		Edges:       edges,
+		CreatedBy:   workflow.CreatedBy,
+	}
+
+	// Convert nodes/edges to WorkflowSpec
+	spec := api.nodesToWorkflowSpec(workflowResp, req.Inputs)
+
+	// Execute via Temporal RoutingWorkflow
+	execID := uuid.New().String()
+	workflowOptions := client.StartWorkflowOptions{
+		ID:        execID,
+		TaskQueue: "default",
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+
+	_, err = api.temporalClient.ExecuteWorkflow(ctx, workflowOptions, "RoutingWorkflow", spec)
+	if err != nil {
+		api.logger.Printf("Failed to execute workflow: %v", err)
+		http.Error(w, fmt.Sprintf("Execution failed: %v", err), http.StatusInternalServerError)
+		return
+	}
+
+	// Save execution to database
+	inputsJSON, _ := json.Marshal(req.Inputs)
+	now := time.Now()
+
+	execution := &db.WorkflowExecution{
+		ID:         execID,
+		WorkflowID: id,
+		CustomerID: api.customerID,
+		TemporalID: execID,
+		Status:     "running",
+		Inputs:     inputsJSON,
+		StartedAt:  now,
+	}
+
+	if err := api.db.SaveExecution(r.Context(), execution); err != nil {
+		api.logger.Printf("Failed to save execution: %v", err)
+		http.Error(w, "Failed to save execution", http.StatusInternalServerError)
+		return
+	}
+
+	// Create execution response
+	execResp := ExecutionResponse{
+		ID:        execID,
+		WorkflowID: id,
+		Status:    "running",
+		StartedAt: now.Format(time.RFC3339),
+		Inputs:    req.Inputs,
+		Outputs:   make(map[string]interface{}),
+		Logs:      []ExecutionLog{},
+	}
+
+	w.Header().Set("Content-Type", "application/json")
+	w.WriteHeader(http.StatusCreated)
+	json.NewEncoder(w).Encode(execResp)
+}
+
+// GetExecution handles GET /executions/{id}
+func (api *WorkflowAPI) GetExecution(w http.ResponseWriter, r *http.Request, id string) {
+	if r.Method != http.MethodGet {
+		http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
+		return
+	}
+
+	execution, err := api.db.FetchExecution(r.Context(), id)
+	if err != nil {
+		http.Error(w, "Execution not found", http.StatusNotFound)
+		return
+	}
+
+	// Get logs from database
+	logs, err := api.db.FetchExecutionLogs(r.Context(), id)
+	if err != nil {
+		api.logger.Printf("Failed to fetch logs: %v", err)
+	}
+
+	execLogs := make([]ExecutionLog, 0)
+	for _, log := range logs {
+		execLogs = append(execLogs, ExecutionLog{
+			Timestamp: log.LoggedAt.Format(time.RFC3339),
+			NodeID:    log.NodeID,
+			Level:     log.Level,
+			Message:   log.Message,
+		})
+	}
+
+	// Parse inputs/outputs
+	var inputs map[string]interface{}
+	var outputs map[string]interface{}
+	json.Unmarshal(execution.Inputs, &inputs)
+	if execution.Outputs != nil {
+		json.Unmarshal(execution.Outputs, &outputs)
+	}
+
+	// Check Temporal workflow status
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	desc, err := api.temporalClient.DescribeWorkflowExecution(ctx, execution.TemporalID, "")
+	status := execution.Status
+	if err == nil && desc != nil {
+		switch desc.Status.String() {
+		case "RUNNING":
+			status = "running"
+		case "COMPLETED":
+			status = "success"
+		case "FAILED":
+			status = "failed"
+		}
+	}
+
+	completedAtStr := ""
+	if execution.CompletedAt != nil {
+		completedAtStr = execution.CompletedAt.Format(time.RFC3339)
+	}
+
+	execResp := ExecutionResponse{
+		ID:          execution.ID,
+		WorkflowID:  execution.WorkflowID,
+		Status:      status,
+		StartedAt:   execution.StartedAt.Format(time.RFC3339),
+		CompletedAt: completedAtStr,
+		Inputs:      inputs,
+		Outputs:     outputs,
+		Logs:        execLogs,
+	}
+
+	w.Header().Set("Content-Type", "application/json")
+	json.NewEncoder(w).Encode(execResp)
+}
+
+// ListExecutions handles GET /workflows/{id}/executions
+func (api *WorkflowAPI) ListExecutions(w http.ResponseWriter, r *http.Request, workflowID string) {
+	if r.Method != http.MethodGet {
+		http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
+		return
+	}
+
+	// TODO: Implement query by workflow_id in database
+	// For now, return empty list (needs DB method for filtering by workflow_id)
+	list := make([]ExecutionResponse, 0)
+
+	w.Header().Set("Content-Type", "application/json")
+	json.NewEncoder(w).Encode(list)
+}
+
+// nodesToWorkflowSpec converts frontend nodes/edges to routing.WorkflowSpec
+func (api *WorkflowAPI) nodesToWorkflowSpec(wf *WorkflowResponse, inputs map[string]interface{}) *routing.WorkflowSpec {
+	spec := &routing.WorkflowSpec{
+		Name:   wf.Name,
+		Input:  inputs,
+		States: []routing.State{},
+	}
+
+	// Build states from nodes
+	stateMap := make(map[string]*routing.State)
+
+	// Create all states
+	for _, node := range wf.Nodes {
+		if node.Type == "activity" {
+			state := &routing.State{
+				Name:     node.ID,
+				Type:     routing.StateTypeTask,
+				Resource: node.Data.Activity,
+				Parameters: node.Data.Config,
+				End:      false,
+			}
+			stateMap[node.ID] = state
+			spec.States = append(spec.States, *state)
+		}
+	}
+
+	// Wire edges (transitions)
+	for _, edge := range wf.Edges {
+		if state, exists := stateMap[edge.Source]; exists {
+			state.Next = edge.Target
+		}
+	}
+
+	// Mark last state as End
+	if len(spec.States) > 0 {
+		// Find state with no outgoing edge
+		for i := range spec.States {
+			hasNext := false
+			for _, edge := range wf.Edges {
+				if edge.Source == spec.States[i].Name {
+					hasNext = true
+					break
+				}
+			}
+			if !hasNext {
+				spec.States[i].End = true
+			}
+		}
+	}
+
+	return spec
+}
+
+// QueryWorkflowGraph handles POST /workflows/{id}/query
+func (api *WorkflowAPI) QueryWorkflowGraph(w http.ResponseWriter, r *http.Request, workflowID string) {
+	ctx, cancel := context.WithTimeout(r.Context(), 60*time.Second)
+	defer cancel()
+
+	var req QueryWorkflowGraphRequest
+	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+		http.Error(w, "Invalid request body", http.StatusBadRequest)
+		return
+	}
+
+	// Defaults
+	if req.SearchType == "" {
+		req.SearchType = "edges"
+	}
+	if req.ConfidenceFloor == 0 {
+		req.ConfidenceFloor = 0.5
+	}
+	if req.TopK == 0 {
+		req.TopK = 10
+	}
+	if req.MaxPathDepth == 0 {
+		req.MaxPathDepth = 3
+	}
+	if req.RankingProfile == "" {
+		req.RankingProfile = "default"
+	}
+
+	// Get latest version if not specified
+	if req.Version == 0 {
+		wf, err := api.db.GetWorkflow(ctx, workflowID)
+		if err != nil {
+			http.Error(w, "Workflow not found", http.StatusNotFound)
+			return
+		}
+		req.Version = wf.Version
+	}
+
+	// Call temporal workflow
+	run, err := api.temporalClient.ExecuteWorkflow(
+		ctx,
+		client.StartWorkflowOptions{
+			ID:        fmt.Sprintf("graph-query-%s-v%d", workflowID, req.Version),
+			TaskQueue: "poimen",
+		},
+		"WorkflowGraphQuery",
+		map[string]interface{}{
+			"workflow_id":       workflowID,
+			"query":             req.Query,
+			"search_type":       req.SearchType,
+			"relation_type":     req.RelationType,
+			"version":           req.Version,
+			"confidence_floor":  req.ConfidenceFloor,
+			"top_k":             req.TopK,
+			"find_paths":        req.FindPaths,
+			"target_node_id":    req.TargetNodeID,
+			"max_path_depth":    req.MaxPathDepth,
+			"ranking_profile":   req.RankingProfile,
+			"include_reasoning": req.IncludeReasoning,
+		},
+	)
+	if err != nil {
+		api.logger.Printf("Failed to start workflow: %v", err)
+		http.Error(w, "Failed to start query workflow", http.StatusInternalServerError)
+		return
+	}
+
+	var result map[string]interface{}
+	if err := run.Get(ctx, &result); err != nil {
+		api.logger.Printf("Workflow execution failed: %v", err)
+		http.Error(w, "Query execution failed", http.StatusInternalServerError)
+		return
+	}
+
+	w.Header().Set("Content-Type", "application/json")
+	json.NewEncoder(w).Encode(result)
+}
+
+// QueryWorkflowGraphRequest matches frontend payload
+type QueryWorkflowGraphRequest struct {
+	Query            string  `json:"query"`
+	SearchType       string  `json:"search_type"`
+	RelationType     string  `json:"relation_type"`
+	Version          int     `json:"version"`
+	ConfidenceFloor  float64 `json:"confidence_floor"`
+	TopK             int     `json:"top_k"`
+	FindPaths        bool    `json:"find_paths"`
+	TargetNodeID     string  `json:"target_node_id"`
+	MaxPathDepth     int     `json:"max_path_depth"`
+	RankingProfile   string  `json:"ranking_profile"`
+	IncludeReasoning bool    `json:"include_reasoning"`
+}
+
+// GetWorkflowRelationVersions handles GET /workflows/{id}/relations/{edge_id}/versions
+func (api *WorkflowAPI) GetWorkflowRelationVersions(w http.ResponseWriter, r *http.Request, workflowID, edgeID string) {
+	ctx, cancel := context.WithTimeout(r.Context(), 10*time.Second)
+	defer cancel()
+
+	// Query relation versions from DB
+	versions, err := api.db.GetRelationVersions(ctx, workflowID, edgeID)
+	if err != nil {
+		api.logger.Printf("Failed to get relation versions: %v", err)
+		http.Error(w, "Failed to fetch relation versions", http.StatusInternalServerError)
+		return
+	}
+
+	w.Header().Set("Content-Type", "application/json")
+	json.NewEncoder(w).Encode(map[string]interface{}{
+		"workflow_id": workflowID,
+		"edge_id":     edgeID,
+		"versions":    versions,
+		"total_count": len(versions),
+	})
+}

internal/routing/llm_client.go

@@ -8,6 +8,7 @@ import (
 	"io"
 	"net/http"
 	"os"
+	"strings"
 )
 
 var (

internal/routing/types.go

@@ -28,16 +28,12 @@ type State struct {
 	Type              StateType `json:"type"`
 	
 	// Task fields
-	Activity          string                 `json:"activity,omitempty"`
 	Resource          string                 `json:"resource,omitempty"`
 	Parameters        map[string]interface{} `json:"parameters,omitempty"`
 	Timeout           string                 `json:"timeout,omitempty"`
 	Retry             *RetryPolicy           `json:"retry,omitempty"`
 	Catch             []CatchClause          `json:"catch,omitempty"`
 	
-	// Parallel fields
-	Branches          []interface{}          `json:"branches,omitempty"`
-	
 	// Pass fields
 	Result            interface{} `json:"result,omitempty"`
 	
@@ -54,18 +50,14 @@ type State struct {
 type StateType string
 
 const (
-	StateTypeTask     StateType = "Task"
-	StateTypePass     StateType = "Pass"
-	StateTypeFail     StateType = "Fail"
-	StateTypeParallel StateType = "Parallel"
-
-	TaskActivity = "Task"
+	StateTypeTask StateType = "Task"
+	StateTypePass StateType = "Pass"
+	StateTypeFail StateType = "Fail"
 )
 
 // RetryPolicy defines retry behavior for activities
 type RetryPolicy struct {
 	MaxAttempts     int32   `json:"maxAttempts"`
-	BackoffSeconds  int32   `json:"backoffSeconds,omitempty"`
 	BackoffRate     float64 `json:"backoffRate"`
 	InitialInterval string  `json:"initialInterval"`
 	MaxInterval     string  `json:"maxInterval,omitempty"`

k8s/configmap.enc.yaml

@@ -1,27 +0,0 @@
-apiVersion: ENC[AES256_GCM,data:IxE=,iv:nh5IQck87AsRYnvxMLxn2rZFBUTHcc9obvsYoHnvC6g=,tag:ink4m5kv1NqX2TeRrpCaqg==,type:str]
-kind: ENC[AES256_GCM,data:2U5oBCqywkR0,iv:EVBlp6G1SlznoP7Zx9Y0mQOnxbzcosP28+UbJUWFjHM=,tag:v1lot4q06QBqU2M5q2RJ/g==,type:str]
-metadata:
-    name: ENC[AES256_GCM,data:pFOLyxGAW7/KqUqijkPM1mGYaKo=,iv:jttNW2Ef0itpuWupOQkGEiFaanpcd/HuvfLg9l2tSEg=,tag:6ijOdYdjnUaCOf29z8vkxg==,type:str]
-    namespace: ENC[AES256_GCM,data:FSZFmVEp,iv:liL0yGuEbjZjf6egh8KS9zi6H3AEF9MvynGBO+49GW0=,tag:4iJNECe8vsXqzQBV/Q7c0g==,type:str]
-data:
-    #ENC[AES256_GCM,data:d+Z/w/n69JI5,iv:/w2xIrmNXg53kz+tfebT0P6rD3TC909YTDFXXUc/yHo=,tag:ao4yA2lW2a1/ELT8ZIBOgw==,type:comment]
-    TEMPORAL_NAMESPACE: ENC[AES256_GCM,data:X0D/DVfVgzhFh+oB8nU=,iv:biiLtXRALsUMNIW4qGj8JOc9C1RJK93pOw74ZmymM/g=,tag:tW5+AwUGj8jzvxPSZAf41A==,type:str]
-    TEMPORAL_HOSTPORT: ENC[AES256_GCM,data:OK8Cu7nA2fKhXmBko8MORoD6hvJBMFfE/dTBZ99ac420xjPBWjhrz2XR3WRiWhUIIQ==,iv:HKDNRT/30HNzN3pX91e63i/ZeIUUYkycknjjKTFlVwo=,tag:OjPle4p+nV4ZQYN/yuUFEw==,type:str]
-    #ENC[AES256_GCM,data:1jnEt80ScfSqnCv9Rf/XLAlYTfuZVxoRdV7Sx9EgIGmzvAM=,iv:3NKsB9ZytMxuMipSLApKX7hj2jQG+2e2Hp8+ofFin9k=,tag:s4eRatHTQJa37xBO5UYI5Q==,type:comment]
-    LOCAL_LLM_BASE_URL: ENC[AES256_GCM,data:rTWBhNovjp5WdFI2jMaMRRg41uhh1dOFqVnplmMADiubu/FGXCvUPD2d9IuU,iv:1RdS5UfD6aOXnaEfddRd4PxJ1+K7AwKlMUYxzUQxVJ4=,tag:K0g2KDCFLOtxDFkHxoRbMw==,type:str]
-    MEMORY_SERVICE_URL: ENC[AES256_GCM,data:CarE5ENFtDvBWVzB7qTZQqoL0OsN2uAO9aoPXu1LPWtdoSQ2rQsVs2scTA08jaWLD+o=,iv:e7M/aibmYdSk4NbY6fYw0UPjd/VQvWJ1tC8n0r2y6r0=,tag:bWuCTBRfKkaB2fauPR51Fg==,type:str]
-sops:
-    age:
-        - enc: |
-            -----BEGIN AGE ENCRYPTED FILE-----
-            YWdlLWVuY3J5cHRpb24ub3JnL3YxCi0+IFgyNTUxOSBBV0RJU2wzN1Bib0lZRlFl
-            K2UzSVBNUjNHUVVFeUFEWTJBUHNZT0l3RWprCmp2cE1HM2xOWnVjVGlqdWI0SndG
-            Q1VOaklVdlg4eGp6V29uVDVJRUtmKzgKLS0tIGp3MWJMdmdyZmRnWmxOWEhVZUhm
-            Rm5DNzBhUWt0SVBDK09mSWw2SHVRTWMKLS6Cz2nhz1RSoV+VUvFw9EsjlWbE2nEk
-            4P1FdNgr+v1MTNczKZGzh9HbTAmYxCRgVBszXR5ov2JfukaZAWb9mg==
-            -----END AGE ENCRYPTED FILE-----
-          recipient: age1e5fq3hwxy78psus2nfvmtmua36g0u3suk78ephw6246l974d2utsvn0hla
-    lastmodified: "2026-09-06T12:56:33Z"
-    mac: ENC[AES256_GCM,data:Ra89XdYHoJV+uFBlUMdw+I36UQfhM+r0Q4WjI3V/A7jBY2t00I0xSP+KRKPXlAA18HXpLCc5VkWklPMU4gfyYied1bPuFw6yLvvuup/ev4UAg/LofThKjtuRfXPy82Ltvro6R+ilWzJ8pRtgtk+jC4xkkMsUOACJMVwgn979D7M=,iv:UWR+CqXAreW/fx3tt1kJsOKYZkrAkxLXlU5QVy5p3gE=,tag:3t9c6631D4XrDmkR6Iv/4Q==,type:str]
-    unencrypted_suffix: _unencrypted
-    version: 3.13.2

k8s/configmap.yaml

@@ -0,0 +1,10 @@
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: poimen-config
+  namespace: poimen
+data:
+  TEMPORAL_NAMESPACE: "poimen-harness"
+  TEMPORAL_HOSTPORT: "temporal-frontend.temporal:7233"
+  LOCAL_LLM_BASE_URL: "http://api-gateway.api:8080"
+  POIMEN_MEMORY_URL: "http://poimen-memory.poimen.svc.cluster.local:8080"

k8s/poimen-application.yaml

@@ -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"

k8s/secret.yaml

@@ -0,0 +1,12 @@
+# NOTE: This file is for reference only.
+# Kustomize will auto-generate secrets from secrets.env
+# See kustomization.yaml for details
+
+apiVersion: v1
+kind: Secret
+metadata:
+  name: poimen-secrets
+  namespace: poimen
+type: Opaque
+stringData:
+  ANTHROPIC_API_KEY: "" # Generated from secrets.env by Kustomize

k8s/secrets.env

@@ -0,0 +1 @@
+ANTHROPIC_API_KEY=YOUR_ANTHROPIC_API_KEY_HERE

migrations/001_workflows_schema.sql

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

migrations/003_workflows_schema.sql

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

migrations/004_workflow_relations.sql

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

pkg/db/db.go

@@ -17,7 +17,7 @@ type DB struct {
 }
 
 // New creates a new database connection to memory-db (K8s CNPG)
-// Expected DSN format: postgresql://app:password@host:5432/dbname?sslmode=disable
+// Expected DSN format: postgresql://app:password@memory-db-rw.poimen.svc.cluster.local:5432/memory?sslmode=disable
 func New(dsn string) (*DB, error) {
 	if dsn == "" {
 		// Fallback: try to construct from K8s env vars

pkg/db/models.go

@@ -23,7 +23,6 @@ type WorkflowEdge struct {
 
 // Canvas represents the full React Flow canvas (nodes + edges)
 type Canvas struct {
-	Name  string         `json:"name,omitempty"`
 	Nodes []WorkflowNode `json:"nodes"`
 	Edges []WorkflowEdge `json:"edges"`
 }

pkg/types/synthesis.go

@@ -1,59 +0,0 @@
-package types
-
-import "time"
-
-// SynthesisInput contains the input for the synthesis workflow.
-type SynthesisInput struct {
-	Project string   `json:"project"`
-	Source  string   `json:"source"`
-	Text    string   `json:"text"`
-	Kind    string   `json:"kind"` // L1, L2, reference
-	Tags    []string `json:"tags,omitempty"`
-}
-
-// SynthesisResult contains the output of the synthesis workflow.
-type SynthesisResult struct {
-	ChunkID            string               `json:"chunk_id"`
-	EntitiesExtracted  int                  `json:"entities_extracted"`
-	FactsExtracted     int                  `json:"facts_extracted"`
-	Contradictions     int                  `json:"contradictions"`
-	ReviewQueued       int                  `json:"review_queued"`
-	Entities           []ExtractedEntity    `json:"entities"`
-	Facts              []ExtractedFact      `json:"facts"`
-	Duration           time.Duration        `json:"duration"`
-}
-
-// ExtractedEntity represents an entity found during synthesis.
-type ExtractedEntity struct {
-	Name       string  `json:"name"`
-	EntityType string  `json:"entity_type"`
-	Confidence float64 `json:"confidence"`
-}
-
-// ExtractedFact represents a fact extracted during synthesis.
-type ExtractedFact struct {
-	Subject    string  `json:"subject"`
-	Predicate  string  `json:"predicate"`
-	Object     string  `json:"object"`
-	Confidence float64 `json:"confidence"`
-}
-
-// ContradictionResult represents a contradiction detection result.
-type ContradictionResult struct {
-	FactA        ExtractedFact `json:"fact_a"`
-	FactB        ExtractedFact `json:"fact_b"`
-	Severity     string        `json:"severity"` // low, medium, high
-	AutoResolved bool          `json:"auto_resolved"`
-	QueuedReview bool          `json:"queued_review"`
-}
-
-// PersistInput groups all synthesis results for persistence.
-type PersistInput struct {
-	ChunkID        string                `json:"chunk_id"`
-	Project        string                `json:"project"`
-	Source         string                `json:"source"`
-	Kind           string                `json:"kind"`
-	Entities       []ExtractedEntity     `json:"entities"`
-	Facts          []ExtractedFact       `json:"facts"`
-	Contradictions []ContradictionResult `json:"contradictions"`
-}

pkg/types/types.go

@@ -1,223 +0,0 @@
-// Package types defines the shared domain model for Poimen workflows.
-// Both workflow/ (orchestration) and activity/ (execution) import from here.
-package types
-
-import (
-	"time"
-
-	"github.com/rockliang/poimen/workflows/pkg/db"
-)
-
-// ===== LLM Configuration =====
-
-type ModelSpec struct {
-	ModelID  string
-	Thinking string // "adaptive" or ""
-	Effort   string // "low", "medium", "high", "xhigh", "max"
-}
-
-type PromptSpec struct {
-	TemplateRef string
-	RawTemplate string
-	Variables   map[string]any
-	Model       ModelSpec
-	LessonsRef  string
-}
-
-type SkillRef struct {
-	Name string
-	URL  string
-}
-
-// ===== Retry & Tuning =====
-
-type PiRetryPolicy struct {
-	ScheduleToCloseTimeout time.Duration
-	InitialInterval        time.Duration
-	MaximumInterval        time.Duration
-	BackoffCoefficient     float64
-	StreamTimeout          time.Duration
-	StreamTimeoutMax       time.Duration
-}
-
-type ActivityTuning struct {
-	ImplementerBaseTimeout  time.Duration
-	ImplementerMaxRetries   int
-	JudgeTimeout            time.Duration
-	PiRetry                 PiRetryPolicy
-	InitialRetryInterval    time.Duration
-	MaxRetryInterval        time.Duration
-	RetryBackoffCoefficient float64
-}
-
-// ===== Orchestrator =====
-
-type OrchestratorConfig struct {
-	SystemPrompt string
-	Skills       []SkillRef
-	RolePrompts  map[string]PromptSpec
-	Tuning       ActivityTuning
-}
-
-type OrchestratorInput struct {
-	TargetRepoPath     string
-	RemoteURL          string
-	Milestone          string
-	Config             OrchestratorConfig
-	DryRun             bool
-	CycleCount         int
-	MaxCyclesBeforeCAN int
-	PiProvider         string
-}
-
-type OrchestratorOutput struct {
-	MilestoneComplete bool
-	Done              bool
-	LastError         string
-}
-
-// ===== TaskUnit =====
-
-type TaskUnitInput struct {
-	TaskID         string
-	RemoteURL      string
-	TargetRepoPath string
-	Milestone      string
-	Config         OrchestratorConfig
-	DryRun         bool
-}
-
-type TaskUnitOutput struct {
-	TaskID   string
-	Status   string
-	Verdict  string
-	Critique string
-	Branch   string
-	Reason   string
-	Changes  string
-}
-
-// ===== Canvas & Relations =====
-
-type RelationWording struct {
-	Verb              string  `json:"verb"`
-	SourceOutput      string  `json:"source_output"`
-	TargetInput       string  `json:"target_input"`
-	ConnectionType    string  `json:"connection_type"`
-	Confidence        float64 `json:"confidence"`
-	SemanticMatch     string  `json:"semantic_match"`
-	TransformerNeeded string  `json:"transformer_needed,omitempty"`
-}
-
-type EdgeWithWording struct {
-	ID              string          `json:"id,omitempty"`
-	Source          string          `json:"source"`
-	Target          string          `json:"target"`
-	RelationType    string          `json:"relation_type"`
-	RelationLabel   string          `json:"relation_label"`
-	RelationWording RelationWording `json:"relation_wording"`
-	CreatedAt       string          `json:"created_at,omitempty"`
-}
-
-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"`
-}
-
-// ===== Activity I/O =====
-
-type FetchCanvasRelationsInput struct {
-	WorkflowID string `json:"workflow_id"`
-	Version    int    `json:"version"`
-}
-
-type CanvasReasonerInput struct {
-	Nodes            []db.WorkflowNode `json:"nodes"`
-	Edges            []db.WorkflowEdge `json:"edges"`
-	PreserveExisting bool              `json:"preserve_existing,omitempty"`
-	AuthToken        string            `json:"auth_token,omitempty"`
-}
-
-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"`
-}
-
-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"`
-}
-
-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 CanvasCompatibilityInput struct {
-	Nodes []db.WorkflowNode `json:"nodes"`
-	Edges []db.WorkflowEdge `json:"edges"`
-	Query string            `json:"query,omitempty"`
-}
-
-type IndexGraphRAGInput struct {
-	WorkflowID string            `json:"workflow_id"`
-	Version    int               `json:"version"`
-	Nodes      []db.WorkflowNode `json:"nodes"`
-	Relations  []EdgeWithWording `json:"relations"`
-}
-
-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"`
-}
-
-type PromptUpdate struct {
-	Role string
-	Spec PromptSpec
-}
-
-// ===== Defaults =====
-
-func NewPiRetryPolicy() PiRetryPolicy {
-	return PiRetryPolicy{
-		ScheduleToCloseTimeout: 5 * time.Minute,
-		InitialInterval:        2 * time.Second,
-		MaximumInterval:        30 * time.Second,
-		BackoffCoefficient:     2.0,
-		StreamTimeout:          30 * time.Second,
-		StreamTimeoutMax:       2 * time.Minute,
-	}
-}
-
-func NewActivityTuning() ActivityTuning {
-	return ActivityTuning{
-		ImplementerBaseTimeout: 10 * time.Minute,
-		ImplementerMaxRetries:  3,
-		JudgeTimeout:           5 * time.Minute,
-		PiRetry:                NewPiRetryPolicy(),
-	}
-}

statemachine/orchestrator.go

@@ -1,4 +1,4 @@
-package workflow
+package statemachine
 
 import (
 	"fmt"

statemachine/orchestrator_recovery.go

@@ -1,4 +1,4 @@
-package workflow
+package statemachine
 
 import (
 	"fmt"

statemachine/routing_workflow.go

@@ -1,4 +1,4 @@
-package workflow
+package statemachine
 
 import (
 	"fmt"

statemachine/signals.go

@@ -1,3 +1,3 @@
-package workflow
+package statemachine
 
 // Empty stub - will be filled in T0.7

statemachine/taskunit.go

@@ -1,4 +1,4 @@
-package workflow
+package statemachine
 
 import (
 	"fmt"

statemachine/test_workflow.go

@@ -1,4 +1,4 @@
-package workflow
+package statemachine
 
 import (
 	"go.temporal.io/sdk/workflow"

statemachine/types.go

@@ -0,0 +1,136 @@
+package statemachine
+
+import "time"
+
+// ModelSpec defines LLM model configuration.
+type ModelSpec struct {
+	ModelID  string // e.g. "claude-opus-5", "claude-sonnet-5"
+	Thinking string // "adaptive" or ""
+	Effort   string // "low", "medium", "high", "xhigh", "max"
+}
+
+// PromptSpec defines a prompt template with variables and model.
+type PromptSpec struct {
+	TemplateRef string            // e.g. "planner/default.tmpl"
+	RawTemplate string            // overrides TemplateRef if non-empty
+	Variables   map[string]any    // template variables
+	Model       ModelSpec         // which LLM to use
+	LessonsRef  string            // key into lessons store
+}
+
+// PiRetryPolicy defines retry and timeout settings for Pi command execution.
+type PiRetryPolicy struct {
+	ScheduleToCloseTimeout time.Duration // default: 5m
+	InitialInterval        time.Duration // default: 2s
+	MaximumInterval        time.Duration // default: 30s
+	BackoffCoefficient     float64       // default: 2.0
+	StreamTimeout          time.Duration // default: 30s
+	StreamTimeoutMax       time.Duration // default: 2m
+}
+
+// ActivityTuning defines timeouts and retry counts for activities.
+type ActivityTuning struct {
+	ImplementerBaseTimeout time.Duration // default: 10m
+	ImplementerMaxRetries  int           // default: 3
+	JudgeTimeout           time.Duration // default: 5m
+	PiRetry                PiRetryPolicy
+	// Retry policy settings
+	InitialRetryInterval   time.Duration // default: 2s
+	MaxRetryInterval       time.Duration // default: 5m
+	RetryBackoffCoefficient float64      // default: 2.0
+}
+
+// OrchestratorConfig holds all runtime configuration for the orchestrator.
+type OrchestratorConfig struct {
+	SystemPrompt string                    // shared prompt prefix
+	Skills       []SkillRef                // required skill sources
+	RolePrompts  map[string]PromptSpec     // per-role: "planner", "judge", "implementer"
+	Tuning       ActivityTuning
+}
+
+// OrchestratorInput is the input to the Orchestrator workflow.
+type OrchestratorInput struct {
+	TargetRepoPath      string
+	RemoteURL           string
+	Milestone           string // e.g. "T0"
+	Config              OrchestratorConfig
+	DryRun              bool
+	CycleCount          int
+	MaxCyclesBeforeCAN  int // default: 100
+	PiProvider          string // pi provider name (e.g., "local-llm"); required for skill preparation
+}
+
+// OrchestratorOutput is the output of the Orchestrator workflow.
+type OrchestratorOutput struct {
+	MilestoneComplete bool
+	Done              bool
+	LastError         string
+}
+
+// TaskUnitInput is the input to the TaskUnit workflow.
+type TaskUnitInput struct {
+	TaskID         string
+	RemoteURL      string
+	TargetRepoPath string
+	Milestone      string
+	Config         OrchestratorConfig
+	DryRun         bool
+}
+
+// TaskUnitOutput is the output of the TaskUnit workflow.
+type TaskUnitOutput struct {
+	TaskID   string
+	Status   string // "success" or "failed"
+	Verdict  string // "pass" or "fail" from judge
+	Critique string // feedback from judge
+	Branch   string
+	Reason   string // error reason if failed
+	Changes  string // summary of changes
+}
+
+// SkillRef references a skill source.
+type SkillRef struct {
+	Name string // skill identifier
+	URL  string // source to clone
+}
+
+// Default values for types.
+const (
+	defaultScheduleToCloseTimeout = 5 * time.Minute
+	defaultInitialInterval         = 2 * time.Second
+	defaultMaximumInterval         = 30 * time.Second
+	defaultBackoffCoefficient      = 2.0
+	defaultStreamTimeout           = 30 * time.Second
+	defaultStreamTimeoutMax        = 2 * time.Minute
+	defaultImplementerBaseTimeout  = 10 * time.Minute
+	defaultImplementerMaxRetries   = 3
+	defaultJudgeTimeout            = 5 * time.Minute
+)
+
+// NewPiRetryPolicy returns a PiRetryPolicy with defaults.
+func NewPiRetryPolicy() PiRetryPolicy {
+	return PiRetryPolicy{
+		ScheduleToCloseTimeout: defaultScheduleToCloseTimeout,
+		InitialInterval:        defaultInitialInterval,
+		MaximumInterval:        defaultMaximumInterval,
+		BackoffCoefficient:     defaultBackoffCoefficient,
+		StreamTimeout:          defaultStreamTimeout,
+		StreamTimeoutMax:       defaultStreamTimeoutMax,
+	}
+}
+
+// NewActivityTuning returns an ActivityTuning with defaults.
+func NewActivityTuning() ActivityTuning {
+	return ActivityTuning{
+		ImplementerBaseTimeout: defaultImplementerBaseTimeout,
+		ImplementerMaxRetries:  defaultImplementerMaxRetries,
+		JudgeTimeout:           defaultJudgeTimeout,
+		PiRetry:                NewPiRetryPolicy(),
+	}
+}
+
+// PromptUpdate represents an update to a role prompt.
+type PromptUpdate struct {
+	Role string
+	Spec PromptSpec
+}

statemachine/workflow_graph_query.go

@@ -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
+}

tests/git_test.go

@@ -9,7 +9,7 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
-	"github.com/rockliang/poimen/workflows/activity"
+	"github.com/rockliang/poimen/workflows/action"
 )
 
 func TestGitCloneAndFetch(t *testing.T) {
@@ -56,7 +56,7 @@ func TestGitCloneAndFetch(t *testing.T) {
 
 	// Test clone into empty path
 	ctx := context.Background()
-	err := activity.CloneRepoActivity(ctx, activity.CloneRepoInput{
+	err := action.CloneRepoActivity(ctx, action.CloneRepoInput{
 		RemoteURL:      sourceDir,
 		TargetRepoPath: targetDir,
 	})
@@ -89,7 +89,7 @@ func TestGitCloneAndFetch(t *testing.T) {
 	}
 
 	// Test fetch on existing repo
-	err = activity.CloneRepoActivity(ctx, activity.CloneRepoInput{
+	err = action.CloneRepoActivity(ctx, action.CloneRepoInput{
 		RemoteURL:      sourceDir,
 		TargetRepoPath: targetDir,
 	})
@@ -139,14 +139,14 @@ func TestGitWorktreeAdd(t *testing.T) {
 
 	// Clone the repo
 	ctx := context.Background()
-	err := activity.CloneRepoActivity(ctx, activity.CloneRepoInput{
+	err := action.CloneRepoActivity(ctx, action.CloneRepoInput{
 		RemoteURL:      sourceDir,
 		TargetRepoPath: repoDir,
 	})
 	assert.NoError(t, err, "clone should succeed")
 
 	// Test worktree add
-	worktreePath, err := activity.GitWorktreeAddActivity(ctx, activity.GitWorktreeAddInput{
+	worktreePath, err := action.GitWorktreeAddActivity(ctx, action.GitWorktreeAddInput{
 		RepoPath: repoDir,
 		TaskID:   "T0.1",
 	})
@@ -207,14 +207,14 @@ func TestGitCommit(t *testing.T) {
 
 	// Clone the repo
 	ctx := context.Background()
-	err := activity.CloneRepoActivity(ctx, activity.CloneRepoInput{
+	err := action.CloneRepoActivity(ctx, action.CloneRepoInput{
 		RemoteURL:      sourceDir,
 		TargetRepoPath: repoDir,
 	})
 	assert.NoError(t, err, "clone should succeed")
 
 	// Create a worktree
-	worktreePath, err := activity.GitWorktreeAddActivity(ctx, activity.GitWorktreeAddInput{
+	worktreePath, err := action.GitWorktreeAddActivity(ctx, action.GitWorktreeAddInput{
 		RepoPath: repoDir,
 		TaskID:   "T0.1",
 	})
@@ -227,7 +227,7 @@ func TestGitCommit(t *testing.T) {
 	}
 
 	// Commit changes
-	err = activity.GitCommitActivity(ctx, activity.GitCommitInput{
+	err = action.GitCommitActivity(ctx, action.GitCommitInput{
 		WorktreePath: worktreePath,
 		Message:      "Add new file",
 	})
@@ -283,14 +283,14 @@ func TestGitDiff(t *testing.T) {
 
 	// Clone the repo
 	ctx := context.Background()
-	err := activity.CloneRepoActivity(ctx, activity.CloneRepoInput{
+	err := action.CloneRepoActivity(ctx, action.CloneRepoInput{
 		RemoteURL:      sourceDir,
 		TargetRepoPath: repoDir,
 	})
 	assert.NoError(t, err, "clone should succeed")
 
 	// Create a worktree
-	worktreePath, err := activity.GitWorktreeAddActivity(ctx, activity.GitWorktreeAddInput{
+	worktreePath, err := action.GitWorktreeAddActivity(ctx, action.GitWorktreeAddInput{
 		RepoPath: repoDir,
 		TaskID:   "T0.1",
 	})
@@ -309,7 +309,7 @@ func TestGitDiff(t *testing.T) {
 	}
 
 	// Get diff (should show the staged change)
-	diffOutput, err := activity.GitDiffActivity(ctx, activity.GitDiffInput{
+	diffOutput, err := action.GitDiffActivity(ctx, action.GitDiffInput{
 		WorktreePath: worktreePath,
 	})
 	assert.NoError(t, err, "diff should succeed")
@@ -378,7 +378,7 @@ func TestGitSquashMerge(t *testing.T) {
 
 	// Clone for the orchestrator to use
 	ctx := context.Background()
-	err := activity.CloneRepoActivity(ctx, activity.CloneRepoInput{
+	err := action.CloneRepoActivity(ctx, action.CloneRepoInput{
 		RemoteURL:      sourceDir,
 		TargetRepoPath: repoDir,
 	})
@@ -387,7 +387,7 @@ func TestGitSquashMerge(t *testing.T) {
 	// Create multiple worktrees with changes
 	for i := 1; i <= 2; i++ {
 		taskID := fmt.Sprintf("T0.%d", i)
-		worktreePath, err := activity.GitWorktreeAddActivity(ctx, activity.GitWorktreeAddInput{
+		worktreePath, err := action.GitWorktreeAddActivity(ctx, action.GitWorktreeAddInput{
 			RepoPath: repoDir,
 			TaskID:   taskID,
 		})
@@ -400,7 +400,7 @@ func TestGitSquashMerge(t *testing.T) {
 		}
 
 		// Commit changes
-		err = activity.GitCommitActivity(ctx, activity.GitCommitInput{
+		err = action.GitCommitActivity(ctx, action.GitCommitInput{
 			WorktreePath: worktreePath,
 			Message:      fmt.Sprintf("Task %s implementation", taskID),
 		})
@@ -408,7 +408,7 @@ func TestGitSquashMerge(t *testing.T) {
 	}
 
 	// Perform squash merge
-	err = activity.GitSquashMergeActivity(ctx, activity.GitSquashMergeInput{
+	err = action.GitSquashMergeActivity(ctx, action.GitSquashMergeInput{
 		RepoPath: repoDir,
 		Branches: []string{"task/T0.1", "task/T0.2"},
 		Message:  "Milestone T0: completed all tasks",

tests/routing_workflow_test.go

@@ -6,7 +6,7 @@ import (
 	"testing"
 
 	"github.com/rockliang/poimen/workflows/internal/routing"
-	"github.com/rockliang/poimen/workflows/workflow"
+	"github.com/rockliang/poimen/workflows/statemachine"
 	"github.com/stretchr/testify/require"
 	"go.temporal.io/sdk/testsuite"
 )
@@ -45,14 +45,14 @@ func TestRoutingWorkflow_SimpleWorkflow(t *testing.T) {
 		},
 	}
 
-	input := workflow.RoutingWorkflowInput{Spec: spec}
+	input := statemachine.RoutingWorkflowInput{Spec: spec}
 
-	env.ExecuteWorkflow(workflow.RoutingWorkflow, input)
+	env.ExecuteWorkflow(statemachine.RoutingWorkflow, input)
 
 	require.True(t, env.IsWorkflowCompleted())
 	require.NoError(t, env.GetWorkflowError())
 
-	var output workflow.RoutingWorkflowOutput
+	var output statemachine.RoutingWorkflowOutput
 	require.NoError(t, env.GetWorkflowResult(&output))
 	require.Equal(t, "COMPLETED", output.Status)
 	require.NotNil(t, output.FinalOutput)
@@ -96,14 +96,14 @@ func TestRoutingWorkflow_MultiStepWorkflow(t *testing.T) {
 		},
 	}
 
-	input := workflow.RoutingWorkflowInput{Spec: spec}
+	input := statemachine.RoutingWorkflowInput{Spec: spec}
 
-	env.ExecuteWorkflow(workflow.RoutingWorkflow, input)
+	env.ExecuteWorkflow(statemachine.RoutingWorkflow, input)
 
 	require.True(t, env.IsWorkflowCompleted())
 	require.NoError(t, env.GetWorkflowError())
 
-	var output workflow.RoutingWorkflowOutput
+	var output statemachine.RoutingWorkflowOutput
 	require.NoError(t, env.GetWorkflowResult(&output))
 	t.Logf("Output: %+v", output)
 	t.Logf("Error: %s", output.Error)
@@ -130,14 +130,14 @@ func TestRoutingWorkflow_PassState(t *testing.T) {
 		},
 	}
 
-	input := workflow.RoutingWorkflowInput{Spec: spec}
+	input := statemachine.RoutingWorkflowInput{Spec: spec}
 
-	env.ExecuteWorkflow(workflow.RoutingWorkflow, input)
+	env.ExecuteWorkflow(statemachine.RoutingWorkflow, input)
 
 	require.True(t, env.IsWorkflowCompleted())
 	require.NoError(t, env.GetWorkflowError())
 
-	var output workflow.RoutingWorkflowOutput
+	var output statemachine.RoutingWorkflowOutput
 	require.NoError(t, env.GetWorkflowResult(&output))
 	require.Equal(t, "COMPLETED", output.Status)
 }
@@ -160,14 +160,14 @@ func TestRoutingWorkflow_FailState(t *testing.T) {
 		},
 	}
 
-	input := workflow.RoutingWorkflowInput{Spec: spec}
+	input := statemachine.RoutingWorkflowInput{Spec: spec}
 
-	env.ExecuteWorkflow(workflow.RoutingWorkflow, input)
+	env.ExecuteWorkflow(statemachine.RoutingWorkflow, input)
 
 	require.True(t, env.IsWorkflowCompleted())
 	require.NoError(t, env.GetWorkflowError())
 
-	var output workflow.RoutingWorkflowOutput
+	var output statemachine.RoutingWorkflowOutput
 	require.NoError(t, env.GetWorkflowResult(&output))
 	require.Equal(t, "FAILED", output.Status)
 	require.Contains(t, output.Error, "WorkflowError")
@@ -219,14 +219,14 @@ func TestRoutingWorkflow_ErrorCatch(t *testing.T) {
 		},
 	}
 
-	input := workflow.RoutingWorkflowInput{Spec: spec}
+	input := statemachine.RoutingWorkflowInput{Spec: spec}
 
-	env.ExecuteWorkflow(workflow.RoutingWorkflow, input)
+	env.ExecuteWorkflow(statemachine.RoutingWorkflow, input)
 
 	require.True(t, env.IsWorkflowCompleted())
 	require.NoError(t, env.GetWorkflowError())
 
-	var output workflow.RoutingWorkflowOutput
+	var output statemachine.RoutingWorkflowOutput
 	require.NoError(t, env.GetWorkflowResult(&output))
 	require.Equal(t, "FAILED", output.Status)
 	require.Contains(t, output.Error, "CaughtError")
@@ -237,14 +237,14 @@ func TestRoutingWorkflow_EmptySpec(t *testing.T) {
 	env := testSuite.NewTestWorkflowEnvironment()
 
 	// Empty spec
-	input := workflow.RoutingWorkflowInput{Spec: nil}
+	input := statemachine.RoutingWorkflowInput{Spec: nil}
 
-	env.ExecuteWorkflow(workflow.RoutingWorkflow, input)
+	env.ExecuteWorkflow(statemachine.RoutingWorkflow, input)
 
 	require.True(t, env.IsWorkflowCompleted())
 	require.NoError(t, env.GetWorkflowError())
 
-	var output workflow.RoutingWorkflowOutput
+	var output statemachine.RoutingWorkflowOutput
 	require.NoError(t, env.GetWorkflowResult(&output))
 	require.Equal(t, "FAILED", output.Status)
 	require.Contains(t, output.Error, "empty")

tests/temporal_integration_test.go

@@ -9,7 +9,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"go.temporal.io/sdk/client"
 	"github.com/rockliang/poimen/workflows/internal/config"
-	"github.com/rockliang/poimen/workflows/workflow"
+	"github.com/rockliang/poimen/workflows/statemachine"
 )
 
 // TestTemporalConnection verifies the worker is connected and healthy
@@ -67,7 +67,7 @@ func TestActivityExecution(t *testing.T) {
 	runResp, err := c.ExecuteWorkflow(ctx, client.StartWorkflowOptions{
 		ID:        workflowID,
 		TaskQueue: "poimen-taskqueue",
-	}, workflow.TestWorkflow)
+	}, statemachine.TestWorkflow)
 
 	assert.NoError(t, err, "failed to execute test workflow")
 	assert.NotNil(t, runResp, "workflow response should not be nil")
@@ -136,28 +136,28 @@ func TestOrchestratorWorkflowIntegration(t *testing.T) {
 	defer cancel()
 
 	// Create minimal orchestrator input
-	input := workflow.OrchestratorInput{
+	input := statemachine.OrchestratorInput{
 		RemoteURL:      "https://forgejo.riotpiao.com/rock/poimen",
 		TargetRepoPath: "/tmp/test-poimen-integration",
 		Milestone:      "T0",
-		Config: workflow.OrchestratorConfig{
+		Config: statemachine.OrchestratorConfig{
 			SystemPrompt: "You are a code generation assistant. Generate simple test code.",
-			RolePrompts: map[string]workflow.PromptSpec{
+			RolePrompts: map[string]statemachine.PromptSpec{
 				"planner": {
 					TemplateRef: "planner/default.tmpl",
-					Model: workflow.ModelSpec{
+					Model: statemachine.ModelSpec{
 						ModelID: "ornith",
 					},
 				},
 				"judge": {
 					TemplateRef: "judge/default.tmpl",
-					Model: workflow.ModelSpec{
+					Model: statemachine.ModelSpec{
 						ModelID: "ornith",
 					},
 				},
 				"implementer": {
 					TemplateRef: "implementer/default.tmpl",
-					Model: workflow.ModelSpec{
+					Model: statemachine.ModelSpec{
 						ModelID: "claude-sonnet-5",
 					},
 				},
@@ -170,7 +170,7 @@ func TestOrchestratorWorkflowIntegration(t *testing.T) {
 	runResp, err := c.ExecuteWorkflow(ctx, client.StartWorkflowOptions{
 		ID:        workflowID,
 		TaskQueue: "poimen-taskqueue",
-	}, workflow.OrchestratorWorkflow, input)
+	}, statemachine.OrchestratorWorkflow, input)
 
 	assert.NoError(t, err, "failed to execute orchestrator workflow")
 	t.Logf("✅ Orchestrator workflow started: %s", workflowID)

tests/temporal_routing_test.go

@@ -9,7 +9,7 @@ import (
 	"time"
 
 	"github.com/rockliang/poimen/workflows/internal/routing"
-	"github.com/rockliang/poimen/workflows/workflow"
+	"github.com/rockliang/poimen/workflows/statemachine"
 	"github.com/stretchr/testify/require"
 	"go.temporal.io/sdk/client"
 )
@@ -69,12 +69,12 @@ func TestTemporalRoutingWorkflow(t *testing.T) {
 
 		// Submit to Temporal
 		workflowID := "test-routing-" + time.Now().Format("20060102-150405")
-		input := workflow.RoutingWorkflowInput{Spec: output.Spec}
+		input := statemachine.RoutingWorkflowInput{Spec: output.Spec}
 
 		run, err := c.ExecuteWorkflow(ctx, client.StartWorkflowOptions{
 			ID:        workflowID,
 			TaskQueue: "poimen-taskqueue",
-		}, workflow.RoutingWorkflow, input)
+		}, statemachine.RoutingWorkflow, input)
 		require.NoError(t, err)
 
 		t.Logf("Workflow submitted: ID=%s, RunID=%s", run.GetID(), run.GetRunID())
@@ -118,18 +118,18 @@ func TestTemporalRoutingWorkflow(t *testing.T) {
 		}
 
 		workflowID := "test-pass-only-" + time.Now().Format("20060102-150405")
-		input := workflow.RoutingWorkflowInput{Spec: spec}
+		input := statemachine.RoutingWorkflowInput{Spec: spec}
 
 		run, err := c.ExecuteWorkflow(ctx, client.StartWorkflowOptions{
 			ID:        workflowID,
 			TaskQueue: "poimen-taskqueue",
-		}, workflow.RoutingWorkflow, input)
+		}, statemachine.RoutingWorkflow, input)
 		require.NoError(t, err)
 
 		t.Logf("Pass-only workflow submitted: ID=%s", run.GetID())
 
 		// Wait for result (Pass states don't need workers)
-		var result workflow.RoutingWorkflowOutput
+		var result statemachine.RoutingWorkflowOutput
 		err = run.Get(ctx, &result)
 		require.NoError(t, err)
 

tests/types_test.go

@@ -5,12 +5,12 @@ import (
 	"time"
 
 	"github.com/stretchr/testify/assert"
-	"github.com/rockliang/poimen/workflows/workflow"
+	"github.com/rockliang/poimen/workflows/statemachine"
 )
 
 func TestTypesDefaults(t *testing.T) {
 	// Test PiRetryPolicy defaults
-	pr := workflow.NewPiRetryPolicy()
+	pr := statemachine.NewPiRetryPolicy()
 	assert.Equal(t, 5*time.Minute, pr.ScheduleToCloseTimeout, "ScheduleToCloseTimeout should be 5m")
 	assert.Equal(t, 2*time.Second, pr.InitialInterval, "InitialInterval should be 2s")
 	assert.Equal(t, 30*time.Second, pr.MaximumInterval, "MaximumInterval should be 30s")
@@ -19,7 +19,7 @@ func TestTypesDefaults(t *testing.T) {
 	assert.Equal(t, 2*time.Minute, pr.StreamTimeoutMax, "StreamTimeoutMax should be 2m")
 
 	// Test ActivityTuning defaults
-	at := workflow.NewActivityTuning()
+	at := statemachine.NewActivityTuning()
 	assert.Equal(t, 10*time.Minute, at.ImplementerBaseTimeout, "ImplementerBaseTimeout should be 10m")
 	assert.Equal(t, 3, at.ImplementerMaxRetries, "ImplementerMaxRetries should be 3")
 	assert.Equal(t, 5*time.Minute, at.JudgeTimeout, "JudgeTimeout should be 5m")
@@ -31,7 +31,7 @@ func TestTypesDefaults(t *testing.T) {
 }
 
 func TestModelSpec(t *testing.T) {
-	spec := workflow.ModelSpec{
+	spec := statemachine.ModelSpec{
 		ModelID:  "claude-opus-5",
 		Thinking: "adaptive",
 		Effort:   "high",
@@ -42,13 +42,13 @@ func TestModelSpec(t *testing.T) {
 }
 
 func TestPromptSpec(t *testing.T) {
-	spec := workflow.PromptSpec{
+	spec := statemachine.PromptSpec{
 		TemplateRef: "planner/default.tmpl",
 		RawTemplate: "",
 		Variables: map[string]any{
 			"key": "value",
 		},
-		Model: workflow.ModelSpec{
+		Model: statemachine.ModelSpec{
 			ModelID: "claude-opus-5",
 		},
 		LessonsRef: "T0.1",
@@ -61,18 +61,18 @@ func TestPromptSpec(t *testing.T) {
 }
 
 func TestOrchestratorConfig(t *testing.T) {
-	cfg := workflow.OrchestratorConfig{
+	cfg := statemachine.OrchestratorConfig{
 		SystemPrompt: "You are an expert",
-		Skills: []workflow.SkillRef{
+		Skills: []statemachine.SkillRef{
 			{Name: "golang-skills", URL: "https://example.com/skill1"},
 		},
-		RolePrompts: map[string]workflow.PromptSpec{
+		RolePrompts: map[string]statemachine.PromptSpec{
 			"planner": {
 				TemplateRef: "planner/default.tmpl",
-				Model: workflow.ModelSpec{ModelID: "claude-opus-5"},
+				Model: statemachine.ModelSpec{ModelID: "claude-opus-5"},
 			},
 		},
-		Tuning: workflow.NewActivityTuning(),
+		Tuning: statemachine.NewActivityTuning(),
 	}
 	assert.Equal(t, "You are an expert", cfg.SystemPrompt)
 	assert.Len(t, cfg.Skills, 1)
@@ -81,7 +81,7 @@ func TestOrchestratorConfig(t *testing.T) {
 }
 
 func TestTaskUnitInput(t *testing.T) {
-	input := workflow.TaskUnitInput{
+	input := statemachine.TaskUnitInput{
 		TaskID:         "T0.1",
 		RemoteURL:      "https://github.com/example/repo",
 		TargetRepoPath: "/tmp/repo",

workflow/synthesis.go

@@ -1,125 +0,0 @@
-package workflow
-
-import (
-	"fmt"
-	"time"
-
-	"go.temporal.io/sdk/temporal"
-	"go.temporal.io/sdk/workflow"
-	"github.com/rockliang/poimen/workflows/pkg/types"
-)
-
-// Re-export shared types from pkg/types for backward compatibility
-type SynthesisInput = types.SynthesisInput
-type SynthesisResult = types.SynthesisResult
-type ExtractedEntity = types.ExtractedEntity
-type ExtractedFact = types.ExtractedFact
-type ContradictionResult = types.ContradictionResult
-type PersistInput = types.PersistInput
-
-var synthesisActivityOptions = workflow.ActivityOptions{
-	StartToCloseTimeout: 60 * time.Second,
-	RetryPolicy: &temporal.RetryPolicy{
-		InitialInterval:    time.Second,
-		BackoffCoefficient: 2.0,
-		MaximumInterval:    30 * time.Second,
-		MaximumAttempts:    3,
-	},
-}
-
-// SynthesisWorkflow orchestrates the 4-stage memory synthesis pipeline.
-//
-// Stage 1: Chunk + embed text
-// Stage 2: Extract entities (LLM + reflection)
-// Stage 3: Extract facts (pattern + LLM)
-// Stage 4: Detect contradictions (pre-filter + LLM)
-//
-// Each stage is an activity with independent retry policy.
-func SynthesisWorkflow(ctx workflow.Context, input SynthesisInput) (*SynthesisResult, error) {
-	logger := workflow.GetLogger(ctx)
-	startTime := workflow.Now(ctx)
-
-	logger.Info("synthesis started",
-		"project", input.Project,
-		"source", input.Source,
-		"kind", input.Kind,
-	)
-
-	actCtx := workflow.WithActivityOptions(ctx, synthesisActivityOptions)
-
-	// Stage 1: Chunk + Embed
-	var chunkID string
-	err := workflow.ExecuteActivity(actCtx, "ChunkAndEmbedActivity", input).Get(ctx, &chunkID)
-	if err != nil {
-		return nil, fmt.Errorf("stage 1 chunk+embed: %w", err)
-	}
-	logger.Info("stage 1 complete", "chunk_id", chunkID)
-
-	// Stage 2: Entity Extraction
-	var entities []ExtractedEntity
-	err = workflow.ExecuteActivity(actCtx, "ExtractEntitiesActivity", chunkID, input.Text).Get(ctx, &entities)
-	if err != nil {
-		return nil, fmt.Errorf("stage 2 entity extraction: %w", err)
-	}
-	logger.Info("stage 2 complete", "entities", len(entities))
-
-	// Stage 3: Fact Extraction
-	var facts []ExtractedFact
-	err = workflow.ExecuteActivity(actCtx, "ExtractFactsActivity", chunkID, input.Text, entities).Get(ctx, &facts)
-	if err != nil {
-		return nil, fmt.Errorf("stage 3 fact extraction: %w", err)
-	}
-	logger.Info("stage 3 complete", "facts", len(facts))
-
-	// Stage 4: Contradiction Detection
-	var contradictions []ContradictionResult
-	err = workflow.ExecuteActivity(actCtx, "DetectContradictionsActivity", input.Project, facts).Get(ctx, &contradictions)
-	if err != nil {
-		return nil, fmt.Errorf("stage 4 contradiction detection: %w", err)
-	}
-
-	reviewQueued := 0
-	for _, c := range contradictions {
-		if c.QueuedReview {
-			reviewQueued++
-		}
-	}
-	logger.Info("stage 4 complete", "contradictions", len(contradictions), "review_queued", reviewQueued)
-
-	// Stage 5: Persist results
-	persistInput := PersistInput{
-		ChunkID:        chunkID,
-		Project:        input.Project,
-		Source:         input.Source,
-		Kind:           input.Kind,
-		Entities:       entities,
-		Facts:          facts,
-		Contradictions: contradictions,
-	}
-	err = workflow.ExecuteActivity(actCtx, "PersistSynthesisActivity", persistInput).Get(ctx, nil)
-	if err != nil {
-		return nil, fmt.Errorf("stage 5 persist: %w", err)
-	}
-
-	duration := workflow.Now(ctx).Sub(startTime)
-	result := &SynthesisResult{
-		ChunkID:           chunkID,
-		EntitiesExtracted: len(entities),
-		FactsExtracted:    len(facts),
-		Contradictions:    len(contradictions),
-		ReviewQueued:      reviewQueued,
-		Entities:          entities,
-		Facts:             facts,
-		Duration:          duration,
-	}
-
-	logger.Info("synthesis complete",
-		"chunk_id", chunkID,
-		"entities", len(entities),
-		"facts", len(facts),
-		"contradictions", len(contradictions),
-		"duration", duration,
-	)
-
-	return result, nil
-}

workflow/synthesis_test.go

@@ -1,179 +0,0 @@
-package workflow
-
-import (
-	"context"
-	"testing"
-
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/mock"
-	"go.temporal.io/sdk/testsuite"
-)
-
-// Stub activity functions for test registration
-func ChunkAndEmbedActivity(_ context.Context, _ SynthesisInput) (string, error) { return "", nil }
-func ExtractEntitiesActivity(_ context.Context, _ string, _ string) ([]ExtractedEntity, error) { return nil, nil }
-func ExtractFactsActivity(_ context.Context, _ string, _ string, _ []ExtractedEntity) ([]ExtractedFact, error) { return nil, nil }
-func DetectContradictionsActivity(_ context.Context, _ string, _ []ExtractedFact) ([]ContradictionResult, error) { return nil, nil }
-func PersistSynthesisActivity(_ context.Context, _ PersistInput) error { return nil }
-
-func registerSynthesisActivities(env *testsuite.TestWorkflowEnvironment) {
-	env.RegisterActivity(ChunkAndEmbedActivity)
-	env.RegisterActivity(ExtractEntitiesActivity)
-	env.RegisterActivity(ExtractFactsActivity)
-	env.RegisterActivity(DetectContradictionsActivity)
-	env.RegisterActivity(PersistSynthesisActivity)
-}
-
-func TestSynthesisWorkflow_Success(t *testing.T) {
-	ts := &testsuite.WorkflowTestSuite{}
-	env := ts.NewTestWorkflowEnvironment()
-
-	input := SynthesisInput{
-		Project: "poimen",
-		Source:  "transcript://test-123",
-		Text:    "Kubernetes uses port 8080 for the API server",
-		Kind:    "L1",
-	}
-
-	registerSynthesisActivities(env)
-
-	// Stage 1: Chunk + Embed
-	env.OnActivity(ChunkAndEmbedActivity, mock.Anything, input).Return("chunk-abc123", nil)
-
-	// Stage 2: Entity Extraction
-	entities := []ExtractedEntity{
-		{Name: "Kubernetes", EntityType: "tool", Confidence: 0.95},
-		{Name: "API server", EntityType: "component", Confidence: 0.90},
-	}
-	env.OnActivity(ExtractEntitiesActivity, mock.Anything, "chunk-abc123", input.Text).Return(entities, nil)
-
-	// Stage 3: Fact Extraction
-	facts := []ExtractedFact{
-		{Subject: "Kubernetes", Predicate: "uses_port", Object: "8080", Confidence: 0.85},
-	}
-	env.OnActivity(ExtractFactsActivity, mock.Anything, "chunk-abc123", input.Text, entities).Return(facts, nil)
-
-	// Stage 4: Contradiction Detection
-	contradictions := []ContradictionResult{}
-	env.OnActivity(DetectContradictionsActivity, mock.Anything, "poimen", facts).Return(contradictions, nil)
-
-	// Stage 5: Persist
-	env.OnActivity(PersistSynthesisActivity, mock.Anything, mock.Anything).Return(nil)
-
-	env.ExecuteWorkflow(SynthesisWorkflow, input)
-
-	assert.True(t, env.IsWorkflowCompleted())
-	assert.NoError(t, env.GetWorkflowError())
-
-	var result SynthesisResult
-	assert.NoError(t, env.GetWorkflowResult(&result))
-	assert.Equal(t, "chunk-abc123", result.ChunkID)
-	assert.Equal(t, 2, result.EntitiesExtracted)
-	assert.Equal(t, 1, result.FactsExtracted)
-	assert.Equal(t, 0, result.Contradictions)
-	assert.Equal(t, 0, result.ReviewQueued)
-}
-
-func TestSynthesisWorkflow_WithContradictions(t *testing.T) {
-	ts := &testsuite.WorkflowTestSuite{}
-	env := ts.NewTestWorkflowEnvironment()
-	registerSynthesisActivities(env)
-
-	input := SynthesisInput{
-		Project: "poimen",
-		Source:  "transcript://test-456",
-		Text:    "Port 8080 is used by nginx",
-		Kind:    "L1",
-	}
-
-	env.OnActivity(ChunkAndEmbedActivity, mock.Anything, input).Return("chunk-def456", nil)
-
-	entities := []ExtractedEntity{
-		{Name: "nginx", EntityType: "tool", Confidence: 0.92},
-	}
-	env.OnActivity(ExtractEntitiesActivity, mock.Anything, "chunk-def456", input.Text).Return(entities, nil)
-
-	facts := []ExtractedFact{
-		{Subject: "nginx", Predicate: "uses_port", Object: "8080", Confidence: 0.88},
-	}
-	env.OnActivity(ExtractFactsActivity, mock.Anything, "chunk-def456", input.Text, entities).Return(facts, nil)
-
-	contradictions := []ContradictionResult{
-		{
-			FactA:        ExtractedFact{Subject: "Kubernetes", Predicate: "uses_port", Object: "8080"},
-			FactB:        ExtractedFact{Subject: "nginx", Predicate: "uses_port", Object: "8080"},
-			Severity:     "medium",
-			AutoResolved: false,
-			QueuedReview: true,
-		},
-	}
-	env.OnActivity(DetectContradictionsActivity, mock.Anything, "poimen", facts).Return(contradictions, nil)
-	env.OnActivity(PersistSynthesisActivity, mock.Anything, mock.Anything).Return(nil)
-
-	env.ExecuteWorkflow(SynthesisWorkflow, input)
-
-	assert.True(t, env.IsWorkflowCompleted())
-	assert.NoError(t, env.GetWorkflowError())
-
-	var result SynthesisResult
-	assert.NoError(t, env.GetWorkflowResult(&result))
-	assert.Equal(t, 1, result.Contradictions)
-	assert.Equal(t, 1, result.ReviewQueued)
-}
-
-func TestSynthesisWorkflow_EntityExtractionFails(t *testing.T) {
-	ts := &testsuite.WorkflowTestSuite{}
-	env := ts.NewTestWorkflowEnvironment()
-	registerSynthesisActivities(env)
-
-	input := SynthesisInput{
-		Project: "poimen",
-		Source:  "transcript://test-789",
-		Text:    "Some text",
-		Kind:    "L1",
-	}
-
-	env.OnActivity(ChunkAndEmbedActivity, mock.Anything, input).Return("chunk-xyz", nil)
-	env.OnActivity(ExtractEntitiesActivity, mock.Anything, "chunk-xyz", input.Text).
-		Return(nil, assert.AnError)
-
-	env.ExecuteWorkflow(SynthesisWorkflow, input)
-
-	assert.True(t, env.IsWorkflowCompleted())
-	assert.Error(t, env.GetWorkflowError())
-	assert.Contains(t, env.GetWorkflowError().Error(), "stage 2 entity extraction")
-}
-
-func TestSynthesisWorkflow_ChunkFails(t *testing.T) {
-	ts := &testsuite.WorkflowTestSuite{}
-	env := ts.NewTestWorkflowEnvironment()
-	registerSynthesisActivities(env)
-
-	input := SynthesisInput{
-		Project: "poimen",
-		Source:  "transcript://test-fail",
-		Text:    "Bad text",
-		Kind:    "L1",
-	}
-
-	env.OnActivity(ChunkAndEmbedActivity, mock.Anything, input).Return("", assert.AnError)
-
-	env.ExecuteWorkflow(SynthesisWorkflow, input)
-
-	assert.True(t, env.IsWorkflowCompleted())
-	assert.Error(t, env.GetWorkflowError())
-	assert.Contains(t, env.GetWorkflowError().Error(), "stage 1 chunk+embed")
-}
-
-func TestSynthesisInput_Fields(t *testing.T) {
-	input := SynthesisInput{
-		Project: "test",
-		Source:  "source://1",
-		Text:    "hello",
-		Kind:    "L2",
-		Tags:    []string{"tag1", "tag2"},
-	}
-	assert.Equal(t, "test", input.Project)
-	assert.Equal(t, "L2", input.Kind)
-	assert.Len(t, input.Tags, 2)
-}

workflow/types.go

@@ -1,20 +0,0 @@
-package workflow
-
-import "github.com/rockliang/poimen/workflows/pkg/types"
-
-// Re-export from pkg/types — single source of truth.
-type ModelSpec = types.ModelSpec
-type PromptSpec = types.PromptSpec
-type SkillRef = types.SkillRef
-type PiRetryPolicy = types.PiRetryPolicy
-type ActivityTuning = types.ActivityTuning
-type OrchestratorConfig = types.OrchestratorConfig
-type OrchestratorInput = types.OrchestratorInput
-type OrchestratorOutput = types.OrchestratorOutput
-type TaskUnitInput = types.TaskUnitInput
-type TaskUnitOutput = types.TaskUnitOutput
-type PromptUpdate = types.PromptUpdate
-type EdgeWithWording = types.EdgeWithWording
-
-var NewPiRetryPolicy = types.NewPiRetryPolicy
-var NewActivityTuning = types.NewActivityTuning

workflow/workflow_graph_query.go

@@ -1,103 +0,0 @@
-package workflow
-
-import (
-	"time"
-
-	"go.temporal.io/sdk/temporal"
-	"go.temporal.io/sdk/workflow"
-
-	"github.com/rockliang/poimen/workflows/pkg/types"
-)
-
-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         []types.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:        []types.EdgeWithWording{},
-		Paths:          []QueryPath{},
-	}
-
-	opts := workflow.ActivityOptions{
-		StartToCloseTimeout: 120 * time.Second,
-		RetryPolicy: &temporal.RetryPolicy{
-			InitialInterval:    2 * time.Second,
-			BackoffCoefficient: 2.0,
-			MaximumInterval:    10 * time.Second,
-			MaximumAttempts:    3,
-		},
-	}
-	ctx = workflow.WithActivityOptions(ctx, opts)
-
-	var canvasData types.CanvasWithRelationsData
-	err := workflow.ExecuteActivity(ctx, "FetchCanvasRelationsActivity",
-		types.FetchCanvasRelationsInput{
-			WorkflowID: input.WorkflowID,
-			Version:    input.Version,
-		},
-	).Get(ctx, &canvasData)
-	if err != nil {
-		return output, err
-	}
-
-	var graphResults types.GraphRAGQueryOutput
-	err = workflow.ExecuteActivity(ctx, "QueryGraphRAGActivity",
-		types.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
-}