From a461e9799a5cb517655d192f5f8becdffd9c00ab Mon Sep 17 00:00:00 2001 From: Test Date: Sat, 5 Sep 2026 05:45:17 -0700 Subject: [PATCH] docs: temporal + graph RAG integration with unified query --- docs/TEMPORAL_GRAPH_RAG_WIRING.md | 1013 +++++++++++++++++++++++ docs/WORKFLOWS_GRAPH_RAG_INTEGRATION.md | 712 ++++++++++++++++ k8s/git-commit.yaml | 2 +- k8s/worker-deployment.yaml | 2 +- 4 files changed, 1727 insertions(+), 2 deletions(-) create mode 100644 docs/TEMPORAL_GRAPH_RAG_WIRING.md create mode 100644 docs/WORKFLOWS_GRAPH_RAG_INTEGRATION.md diff --git a/docs/TEMPORAL_GRAPH_RAG_WIRING.md b/docs/TEMPORAL_GRAPH_RAG_WIRING.md new file mode 100644 index 0000000..fed6341 --- /dev/null +++ b/docs/TEMPORAL_GRAPH_RAG_WIRING.md @@ -0,0 +1,1013 @@ +# Temporal.io + Graph RAG Integration for Workflows + +## Architecture + +``` +┌─────────────────────────────────┐ +│ Workflow API Request │ +│ POST /workflows/{id}/query │ +└──────────────┬──────────────────┘ + │ {query, search_type, relation_type, version} + ▼ +┌─────────────────────────────────┐ +│ Temporal Workflow Orchestrator│ +│ - Start execution │ +│ - Route to activities │ +│ - Aggregate results │ +└──────────────┬──────────────────┘ + │ + ┌──────┴──────┐ + ▼ ▼ + ┌────────────┐ ┌─────────────────┐ + │ Activity 1 │ │ Activity 2 │ + │ Retrieve │ │ Query Graph RAG │ + │ Entities │ │ for Relations │ + └────────────┘ └─────────────────┘ + │ │ + ▼ ▼ + ┌────────────────────────────────┐ + │ Memory API │ + │ /memory/query/semantic │ + │ /memory/entities/{id}/v.. │ + └────────────────────────────────┘ + │ + ▼ + ┌────────────────────────────────┐ + │ Graph RAG (PostgreSQL) │ + │ - workflow_relations │ + │ - workflow_versions │ + │ - relation_changes │ + └────────────────────────────────┘ +``` + +--- + +## 1. Temporal Workflow Definition + +**File:** `statemachine/workflow_graph_query.go` + +```go +package statemachine + +import ( + "context" + "time" + "go.temporal.io/sdk/workflow" + "github.com/rockliang/poimen/workflows/action" + "github.com/rockliang/poimen/workflows/pkg/db" +) + +// WorkflowGraphQueryInput defines query parameters +type WorkflowGraphQueryInput struct { + WorkflowID string `json:"workflow_id"` + Query string `json:"query"` + SearchType string `json:"search_type"` // entities, edges, all + RelationType string `json:"relation_type"` // data-flow, dependency, etc + Version int `json:"version"` // canvas version + ConfidenceFloor float64 `json:"confidence_floor"` + TopK int `json:"top_k"` + FindPaths bool `json:"find_paths"` + TargetNodeID string `json:"target_node_id"` + MaxPathDepth int `json:"max_path_depth"` + RankingProfile string `json:"ranking_profile"` // default, recency, accuracy + IncludeReasoning bool `json:"include_reasoning"` +} + +// WorkflowGraphQueryOutput aggregates results +type WorkflowGraphQueryOutput struct { + WorkflowID string `json:"workflow_id"` + Query string `json:"query"` + Version int `json:"version"` + ExecutionTimeMs int64 `json:"execution_time_ms"` + Results []EdgeWithWording `json:"results"` + Paths []QueryPath `json:"paths"` + TotalCount int `json:"total_count"` + HasMore bool `json:"has_more"` + RankingProfile string `json:"ranking_profile"` +} + +type QueryPath struct { + SourceID string `json:"source_id"` + TargetID string `json:"target_id"` + PathCount int `json:"path_count"` + Distance int `json:"distance"` + Confidence float64 `json:"total_confidence"` + PathNodes []string `json:"node_ids"` +} + +// WorkflowGraphQuery is the main Temporal workflow +func WorkflowGraphQuery(ctx workflow.Context, input WorkflowGraphQueryInput) (WorkflowGraphQueryOutput, error) { + startTime := time.Now() + output := WorkflowGraphQueryOutput{ + WorkflowID: input.WorkflowID, + Query: input.Query, + Version: input.Version, + RankingProfile: input.RankingProfile, + Results: []EdgeWithWording{}, + } + + // Activity options + opts := workflow.ActivityOptions{ + StartToCloseTimeout: 120 * time.Second, + RetryPolicy: &temporal.RetryPolicy{ + InitialInterval: 2 * time.Second, + BackoffCoefficient: 2.0, + MaxInterval: 10 * time.Second, + MaxAttempts: 3, + }, + } + ctx = workflow.WithActivityOptions(ctx, opts) + + // Step 1: Fetch Canvas + Relations from Database + var canvasData action.CanvasWithRelationsData + err := workflow.ExecuteActivity( + ctx, + action.FetchCanvasRelationsActivity, + action.FetchCanvasRelationsInput{ + WorkflowID: input.WorkflowID, + Version: input.Version, + }, + ).Get(ctx, &canvasData) + if err != nil { + return output, err + } + + // Step 2: Query Graph RAG based on search_type + var graphResults action.GraphRAGQueryOutput + graphQuery := action.GraphRAGQueryInput{ + WorkflowID: input.WorkflowID, + Query: input.Query, + SearchType: input.SearchType, + RelationType: input.RelationType, + ConfidenceFloor: input.ConfidenceFloor, + TopK: input.TopK, + RankingProfile: input.RankingProfile, + Canvas: canvasData, + } + + err = workflow.ExecuteActivity( + ctx, + action.QueryGraphRAGActivity, + graphQuery, + ).Get(ctx, &graphResults) + if err != nil { + return output, err + } + + output.Results = graphResults.Edges + output.TotalCount = graphResults.TotalCount + output.HasMore = graphResults.HasMore + + // Step 3: If find_paths requested, calculate paths + if input.FindPaths && input.TargetNodeID != "" { + var pathResults action.PathfindingOutput + err = workflow.ExecuteActivity( + ctx, + action.FindRelationPathsActivity, + action.FindRelationPathsInput{ + WorkflowID: input.WorkflowID, + Version: input.Version, + SourceNodes: extractSourceNodes(graphResults.Edges), + TargetNodeID: input.TargetNodeID, + MaxDepth: input.MaxPathDepth, + Canvas: canvasData, + Relations: graphResults.Edges, + }, + ).Get(ctx, &pathResults) + if err != nil { + return output, err + } + output.Paths = pathResults.Paths + } + + // Step 4: If reasoning requested, explain results + if input.IncludeReasoning { + var reasoning action.ReasoningOutput + err = workflow.ExecuteActivity( + ctx, + action.ExplainGraphResultsActivity, + action.ExplainGraphResultsInput{ + Query: input.Query, + Results: output.Results, + Paths: output.Paths, + Confidence: calculateAverageConfidence(output.Results), + }, + ).Get(ctx, &reasoning) + if err != nil { + // Log but don't fail if reasoning fails + workflow.GetLogger(ctx).Warn("Reasoning failed", "error", err) + } + } + + output.ExecutionTimeMs = time.Since(startTime).Milliseconds() + return output, nil +} + +func extractSourceNodes(edges []EdgeWithWording) []string { + nodeSet := make(map[string]bool) + for _, edge := range edges { + nodeSet[edge.Source] = true + } + var nodes []string + for node := range nodeSet { + nodes = append(nodes, node) + } + return nodes +} + +func calculateAverageConfidence(edges []EdgeWithWording) float64 { + if len(edges) == 0 { + return 0.0 + } + sum := 0.0 + for _, edge := range edges { + sum += edge.RelationWording.Confidence + } + return sum / float64(len(edges)) +} +``` + +--- + +## 2. Activity: Fetch Canvas + Relations + +**File:** `action/fetch_canvas_relations.go` + +```go +package action + +import ( + "context" + "fmt" + "github.com/rockliang/poimen/workflows/pkg/db" +) + +type CanvasWithRelationsData struct { + WorkflowID string `json:"workflow_id"` + Version int `json:"version"` + Nodes []db.WorkflowNode `json:"nodes"` + Edges []EdgeWithWording `json:"edges"` + CreatedAt string `json:"created_at"` + Metadata map[string]interface{} `json:"metadata"` +} + +type FetchCanvasRelationsInput struct { + WorkflowID string `json:"workflow_id"` + Version int `json:"version"` +} + +// FetchCanvasRelationsActivity retrieves canvas + versioned relations +func FetchCanvasRelationsActivity(ctx context.Context, input FetchCanvasRelationsInput) (CanvasWithRelationsData, error) { + logger := newActivityLogger(ctx) + output := CanvasWithRelationsData{ + WorkflowID: input.WorkflowID, + Version: input.Version, + Nodes: []db.WorkflowNode{}, + Edges: []EdgeWithWording{}, + } + + logger.logf("info", "Fetching canvas relations for workflow %s version %d", input.WorkflowID, input.Version) + + // Get database connection (injected via Temporal context) + dbConn, err := getDBFromContext(ctx) + if err != nil { + return output, fmt.Errorf("failed to get DB connection: %w", err) + } + + // Fetch workflow version + canvas + var canvas db.Canvas + query := ` + SELECT workflow_id, version, nodes, edges, created_at, metadata + FROM workflow_versions + WHERE workflow_id = $1 AND version = $2 + ` + rows, err := dbConn.Query(ctx, query, input.WorkflowID, input.Version) + if err != nil { + return output, fmt.Errorf("failed to query workflow_versions: %w", err) + } + defer rows.Close() + + if !rows.Next() { + return output, fmt.Errorf("workflow version not found: %s v%d", input.WorkflowID, input.Version) + } + + // Parse canvas JSONB + if err := rows.Scan(&canvas.WorkflowID, &canvas.Version, &canvas.Nodes, &canvas.Edges, &canvas.CreatedAt, &canvas.Metadata); err != nil { + return output, fmt.Errorf("failed to scan canvas: %w", err) + } + + output.Nodes = canvas.Nodes + output.CreatedAt = canvas.CreatedAt.String() + + // Fetch relations (edges with wording) + relQuery := ` + SELECT id, source_node_id, target_node_id, relation_type, relation_label, + relation_wording, metadata, created_at + FROM workflow_relations + WHERE workflow_id = $1 AND version = $2 + ORDER BY created_at + ` + + relRows, err := dbConn.Query(ctx, relQuery, input.WorkflowID, input.Version) + if err != nil { + return output, fmt.Errorf("failed to query workflow_relations: %w", err) + } + defer relRows.Close() + + for relRows.Next() { + var edge EdgeWithWording + if err := relRows.Scan( + &edge.ID, + &edge.Source, + &edge.Target, + &edge.RelationType, + &edge.RelationLabel, + &edge.RelationWording, + &edge.Metadata, + &edge.CreatedAt, + ); err != nil { + logger.logf("warn", "Failed to scan relation: %v", err) + continue + } + output.Edges = append(output.Edges, edge) + } + + logger.logf("info", "Fetched %d nodes, %d relations", len(output.Nodes), len(output.Edges)) + return output, nil +} +``` + +--- + +## 3. Activity: Query Graph RAG + +**File:** `action/query_graph_rag.go` + +```go +package action + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" +) + +type GraphRAGQueryInput struct { + WorkflowID string `json:"workflow_id"` + Query string `json:"query"` + SearchType string `json:"search_type"` // entities, edges, all + RelationType string `json:"relation_type"` + ConfidenceFloor float64 `json:"confidence_floor"` + TopK int `json:"top_k"` + RankingProfile string `json:"ranking_profile"` + Canvas CanvasWithRelationsData `json:"canvas"` +} + +type GraphRAGQueryOutput struct { + WorkflowID string `json:"workflow_id"` + Query string `json:"query"` + Edges []EdgeWithWording `json:"results"` + TotalCount int `json:"total_count"` + HasMore bool `json:"has_more"` + ExecutionMs int64 `json:"execution_time_ms"` +} + +// QueryGraphRAGActivity calls Memory System unified query +func QueryGraphRAGActivity(ctx context.Context, input GraphRAGQueryInput) (GraphRAGQueryOutput, error) { + logger := newActivityLogger(ctx) + output := GraphRAGQueryOutput{ + WorkflowID: input.WorkflowID, + Query: input.Query, + Edges: []EdgeWithWording{}, + } + + logger.logf("info", "Querying Graph RAG: %s (type: %s, confidence: %.2f)", input.Query, input.SearchType, input.ConfidenceFloor) + + // Build unified query for Memory System + memoryQuery := buildMemoryUnifiedQuery(input) + + // Call Memory System /memory/query endpoint + payload, err := json.Marshal(memoryQuery) + if err != nil { + return output, fmt.Errorf("failed to marshal query: %w", err) + } + + // Get JWT token from context (set by worker) + token := ctx.Value("jwt_token").(string) + + req, err := http.NewRequestWithContext(ctx, "POST", "http://localhost:8080/memory/query", bytes.NewReader(payload)) + if err != nil { + return output, fmt.Errorf("failed to create request: %w", err) + } + + req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", token)) + req.Header.Set("Content-Type", "application/json") + + client := &http.Client{Timeout: 30 * time.Second} + resp, err := client.Do(req) + if err != nil { + return output, fmt.Errorf("failed to query Memory System: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode != 200 { + body, _ := io.ReadAll(resp.Body) + return output, fmt.Errorf("Memory System returned %d: %s", resp.StatusCode, string(body)) + } + + // Parse response + var memoryResp struct { + Results []map[string]interface{} `json:"results"` + TotalCount int `json:"total_count"` + HasMore bool `json:"has_more"` + ExecutionMs int64 `json:"execution_time_ms"` + } + + if err := json.NewDecoder(resp.Body).Decode(&memoryResp); err != nil { + return output, fmt.Errorf("failed to decode Memory response: %w", err) + } + + // Map results to edges with wording + for _, result := range memoryResp.Results { + edge := EdgeWithWording{} + + // Extract edge data from Memory result + if sourceID, ok := result["source_entity_id"].(string); ok { + edge.Source = sourceID + } + if targetID, ok := result["target_entity_id"].(string); ok { + edge.Target = targetID + } + if relType, ok := result["relation_type"].(string); ok { + edge.RelationType = relType + } + if fact, ok := result["fact"].(string); ok { + edge.RelationLabel = fact + } + if conf, ok := result["confidence"].(float64); ok { + edge.RelationWording.Confidence = conf + } + + output.Edges = append(output.Edges, edge) + } + + output.TotalCount = memoryResp.TotalCount + output.HasMore = memoryResp.HasMore + output.ExecutionMs = memoryResp.ExecutionMs + + logger.logf("info", "Graph RAG returned %d edges", len(output.Edges)) + return output, nil +} + +// buildMemoryUnifiedQuery constructs Memory System unified query format +func buildMemoryUnifiedQuery(input GraphRAGQueryInput) map[string]interface{} { + query := map[string]interface{}{ + "query": input.Query, + "search_type": "edges", // Default to edges for workflow relations + "confidence_floor": input.ConfidenceFloor, + "top_k": input.TopK, + "find_paths": input.RelationType != "", + "ranking_profile": input.RankingProfile, + "detect_communities": false, + "discover_facets": false, + } + + // Add relation type filter if specified + if input.RelationType != "" { + if query["facet_filters"] == nil { + query["facet_filters"] = map[string]interface{}{} + } + filters := query["facet_filters"].(map[string]interface{}) + filters["relation_type"] = input.RelationType + } + + return query +} +``` + +--- + +## 4. Activity: Find Relation Paths + +**File:** `action/find_relation_paths.go` + +```go +package action + +import ( + "context" + "fmt" +) + +type FindRelationPathsInput struct { + WorkflowID string `json:"workflow_id"` + Version int `json:"version"` + SourceNodes []string `json:"source_nodes"` + TargetNodeID string `json:"target_node_id"` + MaxDepth int `json:"max_depth"` + Canvas CanvasWithRelationsData `json:"canvas"` + Relations []EdgeWithWording `json:"relations"` +} + +type PathfindingOutput struct { + Paths []QueryPath `json:"paths"` +} + +// FindRelationPathsActivity computes shortest paths in relation graph +func FindRelationPathsActivity(ctx context.Context, input FindRelationPathsInput) (PathfindingOutput, error) { + logger := newActivityLogger(ctx) + output := PathfindingOutput{Paths: []QueryPath{}} + + if input.TargetNodeID == "" { + return output, fmt.Errorf("target_node_id required for pathfinding") + } + + logger.logf("info", "Finding paths to %s (max depth: %d)", input.TargetNodeID, input.MaxDepth) + + // Build adjacency list from relations + graph := buildRelationGraph(input.Relations) + + // BFS from each source to target + for _, sourceNode := range input.SourceNodes { + if sourceNode == input.TargetNodeID { + continue + } + + paths := bfsShortestPaths(graph, sourceNode, input.TargetNodeID, input.MaxDepth) + for _, path := range paths { + output.Paths = append(output.Paths, path) + } + } + + logger.logf("info", "Found %d paths", len(output.Paths)) + return output, nil +} + +func buildRelationGraph(relations []EdgeWithWording) map[string][]string { + graph := make(map[string][]string) + for _, edge := range relations { + graph[edge.Source] = append(graph[edge.Source], edge.Target) + } + return graph +} + +func bfsShortestPaths(graph map[string][]string, source, target string, maxDepth int) []QueryPath { + var paths []QueryPath + + queue := [][]string{{source}} + visited := make(map[string]bool) + visited[source] = true + + for len(queue) > 0 { + path := queue[0] + queue = queue[1:] + + if len(path) > maxDepth { + continue + } + + current := path[len(path)-1] + if current == target { + paths = append(paths, QueryPath{ + SourceID: source, + TargetID: target, + Distance: len(path) - 1, + PathCount: len(paths) + 1, + PathNodes: path, + }) + continue + } + + for _, neighbor := range graph[current] { + newPath := append([]string{}, path...) + newPath = append(newPath, neighbor) + queue = append(queue, newPath) + } + } + + return paths +} +``` + +--- + +## 5. API Handler: Wire Everything Together + +**File:** `internal/api/workflows_graph_query.go` + +```go +package api + +import ( + "context" + "encoding/json" + "net/http" + "time" + + "go.temporal.io/sdk/client" + "github.com/rockliang/poimen/workflows/statemachine" +) + +// QueryWorkflowGraph handles POST /workflows/{id}/query +func (s *Server) QueryWorkflowGraph(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) + return + } + + workflowID := r.PathValue("id") + if workflowID == "" { + http.Error(w, "Missing workflow ID", http.StatusBadRequest) + return + } + + // Parse request + var input statemachine.WorkflowGraphQueryInput + if err := json.NewDecoder(r.Body).Decode(&input); err != nil { + http.Error(w, "Invalid request body", http.StatusBadRequest) + return + } + + input.WorkflowID = workflowID + + // Set defaults + if input.SearchType == "" { + input.SearchType = "edges" + } + if input.TopK == 0 { + input.TopK = 10 + } + if input.ConfidenceFloor == 0 { + input.ConfidenceFloor = 0.5 + } + if input.MaxPathDepth == 0 { + input.MaxPathDepth = 3 + } + if input.RankingProfile == "" { + input.RankingProfile = "default" + } + + // Start Temporal workflow + temporalClient := s.TemporalClient // injected in NewServer + + ctx, cancel := context.WithTimeout(r.Context(), 30*time.Second) + defer cancel() + + we, err := temporalClient.ExecuteWorkflow( + ctx, + client.StartWorkflowOptions{ + ID: workflowID + "-query-" + time.Now().Format("20060102150405"), + TaskQueue: "workflow-tasks", + }, + statemachine.WorkflowGraphQuery, + input, + ) + if err != nil { + http.Error(w, "Failed to start workflow", http.StatusInternalServerError) + return + } + + // Wait for result + var result statemachine.WorkflowGraphQueryOutput + if err := we.Get(ctx, &result); err != nil { + http.Error(w, "Workflow failed", http.StatusInternalServerError) + return + } + + // Return result + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + json.NewEncoder(w).Encode(result) +} + +// GetWorkflowRelationVersions handles GET /workflows/{id}/relations/{edge_id}/versions +func (s *Server) GetWorkflowRelationVersions(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) + return + } + + workflowID := r.PathValue("id") + edgeID := r.PathValue("edge_id") + + if workflowID == "" || edgeID == "" { + http.Error(w, "Missing workflow or edge ID", http.StatusBadRequest) + return + } + + // Query database for edge versions + query := ` + SELECT version_num, operation, snapshot, changed_at, changed_by, fields_changed + FROM workflow_relation_versions + WHERE workflow_id = $1 AND edge_id = $2 + ORDER BY version_num + ` + + rows, err := s.DB.Query(r.Context(), query, workflowID, edgeID) + if err != nil { + http.Error(w, "Failed to query versions", http.StatusInternalServerError) + return + } + defer rows.Close() + + type Version struct { + VersionNum int `json:"version_num"` + Operation string `json:"operation"` + Snapshot interface{} `json:"snapshot"` + ChangedAt string `json:"changed_at"` + ChangedBy string `json:"changed_by"` + FieldsChanged []string `json:"fields_changed"` + } + + versions := []Version{} + for rows.Next() { + v := Version{} + if err := rows.Scan(&v.VersionNum, &v.Operation, &v.Snapshot, &v.ChangedAt, &v.ChangedBy, &v.FieldsChanged); err != nil { + continue + } + versions = append(versions, v) + } + + response := map[string]interface{}{ + "edge_id": edgeID, + "workflow_id": workflowID, + "versions": versions, + "total_versions": len(versions), + "current_version": len(versions), + } + + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(response) +} +``` + +--- + +## 6. Database Schema for Graph RAG + +**File:** `migrations/004_graph_rag_relations.sql` + +```sql +-- Versioned relations with wording +CREATE TABLE workflow_relations ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + workflow_id UUID NOT NULL, + version INT NOT NULL, + source_node_id VARCHAR(255) NOT NULL, + target_node_id VARCHAR(255) NOT NULL, + relation_type VARCHAR(100), -- data-flow, dependency, conditional, parallel + relation_label TEXT, + relation_wording JSONB, -- {verb, source_output, target_input, confidence, etc} + metadata JSONB, + created_at TIMESTAMP DEFAULT NOW(), + + FOREIGN KEY (workflow_id, version) REFERENCES workflow_versions(workflow_id, version), + INDEX (workflow_id, version), + INDEX (relation_type), + UNIQUE (workflow_id, version, source_node_id, target_node_id) +); + +-- Relation change history for versioning +CREATE TABLE workflow_relation_versions ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + workflow_id UUID NOT NULL, + edge_id UUID NOT NULL REFERENCES workflow_relations(id), + version_num INT NOT NULL, + operation VARCHAR(50), -- CREATE, UPDATE, DELETE + snapshot JSONB, -- Full relation state at this version + changed_at TIMESTAMP DEFAULT NOW(), + changed_by VARCHAR(255), + fields_changed TEXT[], + + INDEX (workflow_id, version_num), + UNIQUE (workflow_id, edge_id, version_num) +); + +-- Graph RAG indexing metadata +CREATE TABLE workflow_rag_index ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + workflow_id UUID NOT NULL, + version INT NOT NULL, + indexed_entities JSONB, -- {node_ids, count} + indexed_edges JSONB, -- {edge_ids, count} + index_status VARCHAR(50), -- indexed, pending, failed + last_indexed_at TIMESTAMP, + embedding_model VARCHAR(100), + + INDEX (workflow_id, version) +); +``` + +--- + +## 7. Worker Configuration + +**File:** `cmd/server/main.go` (additions) + +```go +import ( + "context" + "go.temporal.io/sdk/client" + "go.temporal.io/sdk/worker" + "github.com/rockliang/poimen/workflows/statemachine" + "github.com/rockliang/poimen/workflows/action" +) + +func main() { + // ... existing setup ... + + // Temporal client + c, err := client.Dial(client.Options{ + HostPort: os.Getenv("TEMPORAL_HOST_PORT"), // localhost:7233 + }) + if err != nil { + log.Fatal(err) + } + defer c.Close() + + // Worker + w := worker.New(c, "workflow-tasks", worker.Options{}) + + // Register workflows + w.RegisterWorkflow(statemachine.WorkflowGraphQuery) + + // Register activities with DB connection in context + w.RegisterActivityWithOptions( + func(ctx context.Context, input interface{}) (interface{}, error) { + // Inject DB into context + ctx = context.WithValue(ctx, "db", dbConn) + ctx = context.WithValue(ctx, "jwt_token", os.Getenv("JWT_TOKEN")) + return nil, nil + }, + activity.RegisterOptions{Name: "setup"}, + ) + + w.RegisterActivity(action.FetchCanvasRelationsActivity) + w.RegisterActivity(action.QueryGraphRAGActivity) + w.RegisterActivity(action.FindRelationPathsActivity) + w.RegisterActivity(action.ExplainGraphResultsActivity) + + // Start worker + if err := w.Start(); err != nil { + log.Fatal(err) + } + defer w.Stop() + + // Register API handler + server := api.NewServer(dbConn, c) + http.HandleFunc("POST /workflows/{id}/query", server.QueryWorkflowGraph) + http.HandleFunc("GET /workflows/{id}/relations/{edge_id}/versions", server.GetWorkflowRelationVersions) + + log.Fatal(http.ListenAndServe(":8081", nil)) +} +``` + +--- + +## 8. Request/Response Flow Example + +### Request +```bash +curl -X POST http://localhost:8081/workflows/workflow-1/query \ + -H "Content-Type: application/json" \ + -d '{ + "query": "how does code analysis flow into security scanning", + "search_type": "edges", + "relation_type": "data-flow", + "version": 3, + "confidence_floor": 0.7, + "top_k": 10, + "find_paths": true, + "target_node_id": "security-scan-1", + "max_path_depth": 3, + "ranking_profile": "default", + "include_reasoning": true + }' +``` + +### Execution Flow (Temporal) + +1. **API Handler** → Starts `WorkflowGraphQuery` workflow +2. **Workflow** → Calls `FetchCanvasRelationsActivity` + - Query DB: `SELECT * FROM workflow_versions WHERE workflow_id=? AND version=?` + - Query DB: `SELECT * FROM workflow_relations WHERE workflow_id=? AND version=?` + - Return: Canvas + Relations data + +3. **Workflow** → Calls `QueryGraphRAGActivity` + - Build unified query for Memory System + - POST to `/memory/query` with edges search + - Return: Ranked edges with wording + +4. **Workflow** → If find_paths, calls `FindRelationPathsActivity` + - BFS pathfinding in relation graph + - Return: Query paths from sources to target + +5. **Workflow** → If reasoning, calls `ExplainGraphResultsActivity` + - LLM explains why results match query + - Return: Reasoning narrative + +6. **Workflow** → Aggregates results, returns to API +7. **API Handler** → Returns 200 + JSON + +### Response +```json +{ + "workflow_id": "workflow-1", + "query": "how does code analysis flow into security scanning", + "version": 3, + "execution_time_ms": 245, + "results": [ + { + "id": "edge_analyze_scan", + "source": "analyze-code-1", + "target": "security-scan-1", + "relation_type": "data-flow", + "relation_label": "AnalyzeCode outputs metrics → SecurityScan requires code structure", + "relation_wording": { + "verb": "provides-input-for", + "source_output": "metrics (object)", + "target_input": "path (string)", + "confidence": 0.85 + } + } + ], + "paths": [ + { + "source_id": "analyze-code-1", + "target_id": "security-scan-1", + "distance": 1, + "node_ids": ["analyze-code-1", "security-scan-1"] + } + ], + "total_count": 1, + "has_more": false, + "ranking_profile": "default" +} +``` + +--- + +## 9. Error Handling + +```go +// Temporal retry policy +RetryPolicy: &temporal.RetryPolicy{ + InitialInterval: 2 * time.Second, + BackoffCoefficient: 2.0, + MaxInterval: 10 * time.Second, + MaxAttempts: 3, +} + +// Memory System errors propagate +if resp.StatusCode != 200 { + return fmt.Errorf("Memory System %d: %s", resp.StatusCode, body) +} + +// Activity timeouts +StartToCloseTimeout: 120 * time.Second +``` + +--- + +## 10. Testing + +**File:** `tests/workflow_graph_query_test.go` + +```go +func TestWorkflowGraphQuery(t *testing.T) { + // Setup Temporal test + testSuite := &testsuite.WorkflowTestSuite{} + env := testSuite.NewTestWorkflowEnvironment() + + // Mock activities + env.OnActivity(action.FetchCanvasRelationsActivity, mock.MatchedBy(func(input interface{}) bool { + return true + })).Return(CanvasWithRelationsData{ + Nodes: []db.WorkflowNode{...}, + Edges: []EdgeWithWording{...}, + }, nil) + + env.OnActivity(action.QueryGraphRAGActivity, mock.MatchedBy(func(input interface{}) bool { + return true + })).Return(GraphRAGQueryOutput{ + Edges: []EdgeWithWording{...}, + TotalCount: 1, + }, nil) + + // Execute + env.ExecuteWorkflow(statemachine.WorkflowGraphQuery, input) + + // Assert + require.True(t, env.IsWorkflowCompleted()) + require.NoError(t, env.GetWorkflowError()) + + var result statemachine.WorkflowGraphQueryOutput + err := env.GetWorkflowResult(&result) + require.NoError(t, err) + require.Equal(t, 1, result.TotalCount) +} +``` + diff --git a/docs/WORKFLOWS_GRAPH_RAG_INTEGRATION.md b/docs/WORKFLOWS_GRAPH_RAG_INTEGRATION.md new file mode 100644 index 0000000..5ff8dd6 --- /dev/null +++ b/docs/WORKFLOWS_GRAPH_RAG_INTEGRATION.md @@ -0,0 +1,712 @@ +# Workflows + Graph RAG Integration + +Align Poimen Workflows with Poimen Memory System API for versioned relations, semantic queries, and intelligent canvas reasoning. + +**Key Features:** +- Versioned workflow relations following `/memory/entities/{id}/versions` pattern +- Semantic edge queries via `/workflows/{id}/query/semantic/edges` +- Relation wording as Facts (matching Memory System edges schema) +- Point-in-time canvas reconstruction via `?as_of=timestamp` +- Ranking profiles for relation importance +- Automatic Graph RAG indexing + +--- + +## Architecture + +``` +┌─────────────────────────────────┐ +│ Workflow Canvas (React Flow) │ +│ - Nodes (activities) │ +│ - Edges (connections) │ +└──────────────┬──────────────────┘ + │ PUT /workflows/{id} + ▼ +┌─────────────────────────────────┐ +│ CanvasReasonerActivity │ +│ - Suggest edges │ +│ - Validate compatibility │ +│ - Generate change reasoning │ +└──────────────┬──────────────────┘ + │ suggested_edges + reasoning + ▼ +┌─────────────────────────────────┐ +│ Workflows API Server │ +│ - Update canvas in DB │ +│ - Create version entry │ +│ - Store change metadata │ +└──────────────┬──────────────────┘ + │ canvas_version, relations + ▼ +┌─────────────────────────────────┐ +│ Graph RAG Backend │ +│ - Store versioned relations │ +│ - Index relation wording │ +│ - Enable semantic queries │ +└─────────────────────────────────┘ +``` + +--- + +## Data Model + +### Workflow Canvas Version + +```sql +-- In memory.workflows_versions (new table) +CREATE TABLE workflow_versions ( + id UUID PRIMARY KEY, + workflow_id UUID NOT NULL REFERENCES workflows(id), + customer_id UUID NOT NULL, + version INT NOT NULL, + canvas JSONB NOT NULL, -- {nodes, edges} + changed_by UUID, + change_reason TEXT, + change_type VARCHAR(50), -- 'manual', 'auto_reasoned', 'import' + reasoner_confidence FLOAT, + reasoning_metadata JSONB, -- LLM reasoning output + created_at TIMESTAMP DEFAULT NOW(), + + UNIQUE(workflow_id, version), + FOREIGN KEY(workflow_id, customer_id) + REFERENCES workflows(id, customer_id) +); + +-- In memory.workflow_relations (replaces simple edges) +CREATE TABLE workflow_relations ( + id UUID PRIMARY KEY, + workflow_id UUID NOT NULL, + version INT NOT NULL, + source_node_id VARCHAR(255), + target_node_id VARCHAR(255), + relation_type VARCHAR(100), -- 'data-flow', 'dependency', 'conditional', 'parallel' + relation_label TEXT, -- Human-readable: "SecurityScan outputs issues → Report inputs requirements" + relation_wording JSONB, -- {verb, object, context} + metadata JSONB, -- {source_output_type, target_input_type, compatibility_score} + created_at TIMESTAMP, + + FOREIGN KEY(workflow_id, version) + REFERENCES workflow_versions(id, version), + INDEX (workflow_id, version) +); + +-- In memory.relation_changes (for Graph RAG indexing) +CREATE TABLE relation_changes ( + id UUID PRIMARY KEY, + workflow_id UUID, + version_from INT, + version_to INT, + change_type VARCHAR(50), -- 'added', 'removed', 'modified' + relation_id UUID REFERENCES workflow_relations(id), + source_node_id VARCHAR(255), + target_node_id VARCHAR(255), + old_wording JSONB, + new_wording JSONB, + change_reason TEXT, + change_timestamp TIMESTAMP, + reasoner_confidence FLOAT, + + INDEX (workflow_id, version_to) +); +``` + +### Relation Wording Schema + +```json +{ + "id": "edge-1", + "source": "clone-repo-1", + "target": "analyze-code-1", + "relation_type": "data-flow", + + "relation_label": "CloneRepo outputs path → AnalyzeCode requires path", + + "relation_wording": { + "verb": "outputs", + "source_output": "path (string): Local filesystem path where repo was cloned", + "target_input": "path (string, required): Local filesystem path to analyze", + "connection_type": "direct-map", + "confidence": 0.98, + "notes": "Perfect type match between CloneRepo.path and AnalyzeCode.path" + }, + + "metadata": { + "source_activity": "CloneRepoActivity", + "target_activity": "AnalyzeCodeActivity", + "output_type": "string", + "input_type": "string", + "compatibility_score": 0.98, + "requires_transformation": false, + "semantic_match": "File path passes directly" + }, + + "change_history": [ + { + "version": 2, + "action": "added", + "reason": "LLM reasoner suggested data-flow connection", + "confidence": 0.98, + "timestamp": "2025-09-05T10:00:00Z" + } + ] +} +``` + +--- + +## Unified API Endpoints + +All workflow queries follow the unified endpoint pattern from Memory System. + +### 1. Unified Workflow Query + +**Endpoint:** `POST /workflows/{id}/query` + +This is the primary endpoint for all workflow canvas queries (replaces separate search endpoints). + +**Request:** +```json +{ + "query": "how does code analysis flow into security scanning", + "search_type": "edges|entities|all", + "version": 3, + "relation_type": "data-flow", + "confidence_floor": 0.7, + "top_k": 10, + "find_paths": true, + "target_node_id": "security-scan-1", + "max_path_depth": 3, + "ranking_profile": "default", + "include_reasoning": true +} +``` + +**Response (200 OK):** +```json +{ + "workflow_id": "workflow-1", + "query": "how does code analysis flow into security scanning", + "search_type": "edges", + "version": 3, + "execution_time_ms": 145, + "results": [ + { + "id": "edge_analyze_scan", + "source_node_id": "analyze-code-1", + "source_name": "AnalyzeCodeActivity", + "target_node_id": "security-scan-1", + "target_name": "SecurityScanActivity", + "relation_type": "data-flow", + "relation_label": "AnalyzeCode outputs metrics → SecurityScan requires code structure", + "relation_wording": { + "verb": "provides-input-for", + "source_output": "metrics (object): Code quality and structural metrics", + "target_input": "path (string): Directory to scan", + "connection_type": "requires-transformer", + "confidence": 0.85, + "semantic_match": "Analysis metrics can guide security scan prioritization" + }, + "similarity_score": 0.92, + "confidence": 0.85, + "created_at": "2025-09-05T10:00:00Z", + "metadata": { + "source": "canvas://workflow-1:v3", + "tags": ["code-review", "security"] + } + } + ], + "paths": [ + { + "source_id": "analyze-code-1", + "target_id": "security-scan-1", + "path_count": 1, + "shortest_distance": 1, + "paths_found": [ + { + "node_ids": ["analyze-code-1", "security-scan-1"], + "relation_types": ["data-flow"], + "distance": 1, + "total_confidence": 0.85 + } + ] + } + ], + "total_count": 1, + "has_more": false, + "ranking_profile": "default" +} +``` + +--- + +### 2. Update Workflow Canvas (with Versioning) + +**Endpoint:** `PUT /workflows/{id}` + +**Request:** +```json +{ + "nodes": [...], + "edges": [...], + "auto_reason": true, + "change_reason": "User connected CloneRepo to AnalyzeCode", + "user_id": "uuid" +} +``` + +**Response:** +```json +{ + "id": "workflow-1", + "version": 3, + "canvas": { + "nodes": [...], + "edges": [...] + }, + "version_info": { + "version_number": 3, + "created_at": "2025-09-05T10:05:00Z", + "created_by": "user-uuid", + "change_reason": "User connected CloneRepo to AnalyzeCode", + "change_type": "manual" + }, + "relation_updates": { + "added": [ + { + "source": "clone-repo-1", + "target": "analyze-code-1", + "relation_wording": { + "verb": "outputs", + "source_output": "path: Local filesystem path where repo was cloned", + "target_input": "path (required): Local filesystem path to analyze", + "confidence": 0.98 + } + } + ], + "removed": [], + "modified": [] + } +} +``` + +--- + +### 2. Get Relation Version History (Memory System Pattern) + +**Endpoint:** `GET /workflows/{id}/relations/{edge_id}/versions` + +Follows `/memory/entities/{id}/versions` pattern from Memory System. + +**Response:** +```json +{ + "edge_id": "edge_1", + "workflow_id": "workflow-1", + "source": "clone-repo-1", + "target": "analyze-code-1", + "versions": [ + { + "version_num": 1, + "operation": "CREATE", + "snapshot": { + "relation_type": "data-flow", + "relation_label": "CloneRepo → AnalyzeCode", + "relation_wording": { + "verb": "connects-to", + "confidence": 0.75 + } + }, + "changed_at": "2025-09-04T12:00:00Z", + "changed_by": "system", + "fields_changed": ["relation_type", "relation_wording"] + }, + { + "version_num": 2, + "operation": "UPDATE", + "snapshot": { + "relation_type": "data-flow", + "relation_label": "CloneRepo outputs path → AnalyzeCode requires path", + "relation_wording": { + "verb": "outputs", + "source_output": "path (string)", + "target_input": "path (string, required)", + "confidence": 0.98 + } + }, + "changed_at": "2025-09-05T10:00:00Z", + "changed_by": "reasoner-activity", + "fields_changed": ["relation_wording", "relation_label"] + } + ], + "total_versions": 2, + "current_version": 2 +} +``` + +--- + +### 3. Edge Diff (Versioning Pattern) + +**Endpoint:** `POST /workflows/{id}/relations/diff` + +Follows `/memory/entities/diff` pattern. + +**Request:** +```json +{ + "edge_id": "edge_1", + "from_version": 1, + "to_version": 2 +} +``` + +**Response:** +```json +{ + "edge_id": "edge_1", + "from_version": 1, + "to_version": 2, + "source": "clone-repo-1", + "target": "analyze-code-1", + "diff": { + "added_fields": {}, + "removed_fields": {}, + "modified_fields": { + "relation_wording": { + "old": { + "verb": "connects-to", + "confidence": 0.75 + }, + "new": { + "verb": "outputs", + "source_output": "path (string)", + "target_input": "path (string, required)", + "confidence": 0.98 + } + } + } + }, + "change_timeline": [ + { + "version": 1, + "confidence": 0.75, + "changed_at": "2025-09-04T12:00:00Z" + }, + { + "version": 2, + "confidence": 0.98, + "changed_at": "2025-09-05T10:00:00Z" + } + ], + "editors_involved": ["system", "reasoner-activity"] +} +``` + +--- + +### 4. Point-in-Time Canvas (Versioning Pattern) + +**Endpoint:** `GET /workflows/{id}?at_version={v}` or `?as_of=2025-09-05T10:00:00Z` + +Follows `/memory/entities/at` pattern. + +**Response:** +```json +{ + "workflow_id": "workflow-1", + "version": 2, + "as_of_timestamp": "2025-09-05T10:00:00Z", + "canvas": { + "nodes": [...], + "edges": [...] + }, + "relations": [ + { + "id": "edge_1", + "source": "clone-repo-1", + "target": "analyze-code-1", + "relation_type": "data-flow", + "relation_label": "CloneRepo outputs path → AnalyzeCode requires path", + "relation_wording": { + "verb": "outputs", + "source_output": "path (string)", + "target_input": "path (string, required)", + "confidence": 0.98 + }, + "version": 2 + } + ], + "metadata": { + "version_number": 2, + "created_at": "2025-09-05T10:00:00Z", + "changed_by": "reasoner-activity", + "change_reason": "LLM reasoner refined relation wording" + } +} +``` + +--- + +### 5. Bulk Import Canvas (with Relations) + +**Endpoint:** `POST /workflows/{id}/import` + +**Request:** +```json +{ + "canvas": { + "nodes": [...], + "edges": [...] + }, + "relations": [ + { + "source": "n1", + "target": "n2", + "relation_type": "data-flow", + "relation_wording": { + "verb": "outputs", + "source_output": "result (string)", + "target_input": "input (string, required)" + } + } + ], + "change_reason": "Imported from external workflow system" +} +``` + +**Response:** +```json +{ + "workflow_id": "workflow-1", + "version": 4, + "canvas": {...}, + "relations": {...}, + "import_metadata": { + "imported_nodes": 5, + "imported_edges": 4, + "validation_status": "success", + "indexing_status": "queued_for_graph_rag" + } +} +``` + +--- + +## Integration with CanvasReasonerActivity + +### Flow + +``` +User edits canvas + ↓ +PUT /workflows/{id} with auto_reason=true + ↓ +Backend calls CanvasReasonerActivity + ↓ +LLM suggests edges + reasoning + ↓ +Generate relation_wording from suggestions + ↓ +Create workflow_version entry + ↓ +Create workflow_relations entries + ↓ +Index relations in Graph RAG + ↓ +Response includes suggested_edges + relation_wording +``` + +### Response Structure + +```json +{ + "version": 3, + "suggested_edges": [ + { + "source": "clone-1", + "target": "analyze-1" + } + ], + "reasoning": "Standard code review workflow", + "confidence": 0.92, + "relation_wordings": [ + { + "source": "clone-1", + "target": "analyze-1", + "relation_wording": { + "verb": "outputs", + "source_output": "path (string): Cloned repository path", + "target_input": "path (string, required): Directory to analyze", + "connection_type": "direct-map", + "confidence": 0.98, + "semantic_description": "Repository path flows from clone operation to analysis" + } + } + ] +} +``` + +--- + +## Graph RAG Indexing + +### Relations Indexed + +Each workflow relation creates Graph RAG entities: + +``` +Node: { + id: "clone-repo-1", + type: "activity", + name: "CloneRepoActivity", + workflow_id: "workflow-1", + version: 3 +} + +Node: { + id: "analyze-code-1", + type: "activity", + ... +} + +Edge: { + id: "rel-1", + source: "clone-repo-1", + target: "analyze-code-1", + type: "data-flow", + label: "outputs path", + wording: {...}, + version: 3, + created_at: "2025-09-05T10:00:00Z" +} +``` + +### Semantic Queries + +Users can query like: + +- *"Which activities receive data from CloneRepo?"* +- *"What's the data flow from Analyze to Report?"* +- *"Which relations were added in version 3?"* +- *"Show me all type-mismatched connections"* +- *"Find workflows with Security Scan that require approval"* + +--- + +## Versioning Strategy + +### Version Numbering + +- Increment on every canvas change +- Track change_type: `manual`, `auto_reasoned`, `import`, `rag_query_applied` +- Store reasoner_confidence for auto changes + +### Relation Wording Versions + +- Each relation has independent wording history +- Confidence scores tracked per version +- Sources: user input, LLM reasoner, import, RAG query + +### Changelog + +```json +{ + "workflow_id": "workflow-1", + "total_versions": 5, + "changes": [ + { + "version": 1, + "type": "created", + "timestamp": "2025-09-04T10:00:00Z", + "user": "system", + "reason": "Workflow initialized" + }, + { + "version": 2, + "type": "auto_reasoned", + "timestamp": "2025-09-04T12:00:00Z", + "reasoner_confidence": 0.89, + "changes": { + "edges_added": 4, + "edges_modified": 0 + } + }, + { + "version": 3, + "type": "manual", + "timestamp": "2025-09-05T10:05:00Z", + "user": "user-uuid", + "reason": "Connected SecurityScan to Report manually" + } + ] +} +``` + +--- + +## Data Flow Example + +### Scenario: User drops SecurityScan node, system suggests connection + +1. **User Action:** Drops SecurityScan node onto existing workflow +2. **Frontend Call:** `PUT /workflows/{id}` with new nodes + `auto_reason=true` +3. **Backend:** + - Saves canvas to `workflow_versions` v.3 + - Calls CanvasReasonerActivity +4. **LLM Reasoning:** + - Analyzes: AnalyzeCode (outputs: quality, metrics) → SecurityScan (inputs: path, depth) + - Suggests: Add transformer node OR use metrics for decision + - Confidence: 0.85 (type mismatch, requires transformation) +5. **Relation Wording:** + ```json + { + "source": "analyze-code-1", + "target": "security-scan-1", + "relation_wording": { + "verb": "provides-context-for", + "source_output": "quality (object): Code quality metrics", + "target_input": "path (string): Directory to scan", + "connection_type": "requires-transformer", + "reasoning": "Metrics inform which files to prioritize in scanning" + } + } + ``` +6. **Graph RAG:** Relations indexed automatically +7. **Response:** Frontend shows suggestions with natural language descriptions + +--- + +## Implementation Checklist + +- [ ] Create `workflow_versions` table in memory DB +- [ ] Create `workflow_relations` table with wording schema +- [ ] Create `relation_changes` table for tracking modifications +- [ ] Update CanvasReasonerActivity to generate relation wordings +- [ ] Implement versioned CRUD endpoints (PUT, GET, DIFF) +- [ ] Add Graph RAG indexing on relation creation +- [ ] Implement semantic query endpoint +- [ ] Add changelog view +- [ ] Test point-in-time reconstruction +- [ ] Document API in homelab-frontend/API.md + +--- + +## Relation Wording Language + +### Verbs (relation_type → verb mapping) + +| Type | Verbs | Example | +|------|-------|---------| +| `data-flow` | outputs, inputs, receives, provides | "CloneRepo outputs path → AnalyzeCode inputs path" | +| `dependency` | must-complete-before, depends-on, requires | "SecurityScan depends-on AnalyzeCode completion" | +| `conditional` | triggers-if, branches-on, routes-to | "ApproveWorkflow branches-on result approval" | +| `parallel` | runs-alongside, concurrent-with, independent-of | "Notify runs-alongside Report generation" | +| `transformation` | transforms, converts, maps, adapts | "LLMTransform converts metrics → deployment plan" | + +### Confidence Scoring + +- **0.9-1.0:** Perfect match (type-compatible, direct data flow) +- **0.7-0.9:** Good match (semantic fit, minor transformation needed) +- **0.5-0.7:** Possible match (requires user confirmation) +- **<0.5:** Poor match (suggest removal or transformer) + diff --git a/k8s/git-commit.yaml b/k8s/git-commit.yaml index 4ab17a9..a0eb94d 100644 --- a/k8s/git-commit.yaml +++ b/k8s/git-commit.yaml @@ -9,6 +9,6 @@ metadata: app.kubernetes.io/name: poimen app.kubernetes.io/component: orchestrator data: - GIT_COMMIT: "8cfa23e5d" # Updated automatically by CI/CD + GIT_COMMIT: "64fa03cae" # Updated automatically by CI/CD GIT_BRANCH: "main" DEPLOYMENT_DATE: "2026-09-05" diff --git a/k8s/worker-deployment.yaml b/k8s/worker-deployment.yaml index 2ef32ca..128ffad 100644 --- a/k8s/worker-deployment.yaml +++ b/k8s/worker-deployment.yaml @@ -13,7 +13,7 @@ spec: labels: app: poimen-worker annotations: - git-commit: "8cfa23e5d" # ✅ Updated on each push, triggers rolling restart + git-commit: "64fa03cae" # ✅ Updated on each push, triggers rolling restart deployment-date: "2026-09-05" spec: containers: