From 9ec7e6a344a7673b460899295d8c0e9021d694db Mon Sep 17 00:00:00 2001 From: Test Date: Sat, 5 Sep 2026 23:59:13 -0700 Subject: [PATCH] =?UTF-8?q?refactor:=20rename=20action=E2=86=92activity,?= =?UTF-8?q?=20statemachine=E2=86=92workflow,=20remove=20HTTP=20API=20layer?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - action/ → activity/ (Temporal activities) - statemachine/ → workflow/ (Temporal workflows) - Removed internal/api/ and cmd/server/ (api-gw handles HTTP, Temporal is the API) - Created pkg/types/types.go as single source of truth for all shared types - Extracted CallRoleLLM helper (DRY: implementer/planner/judge shared pattern) - Fixed circular import: workflow_graph_query uses string activity names - Fixed logger.logf → logger.Info/Warn (method didn't exist) - Fixed routing types: added Branches, Activity, BackoffSeconds, TaskActivity - Fixed db.Canvas.Name, db.Client→DB, GetWorkflow→FetchWorkflow - Removed unused imports - All tests pass, build clean, vet clean --- action/implementer.go | 93 --- action/judge.go | 78 -- action/llm_inference.go | 159 ---- action/planner.go | 91 --- {action => activity}/analysis.go | 2 +- {action => activity}/analysis_test.go | 2 +- {action => activity}/assume_role.go | 2 +- {action => activity}/canvas_compatibility.go | 43 +- {action => activity}/canvas_reasoner.go | 60 +- .../fetch_canvas_relations.go | 30 +- {action => activity}/git.go | 2 +- activity/implementer.go | 41 + {action => activity}/index_graph_rag.go | 29 +- {action => activity}/integration.go | 2 +- activity/judge.go | 32 + {action => activity}/lessons.go | 2 +- {action => activity}/llm/client.go | 4 +- {action => activity}/llm/client_test.go | 4 +- activity/llm_helper.go | 47 ++ activity/llm_inference.go | 114 +++ {action => activity}/logger.go | 2 +- {action => activity}/memory.go | 2 +- {action => activity}/memory_test.go | 2 +- {action => activity}/notification.go | 2 +- {action => activity}/notification_test.go | 2 +- activity/planner.go | 47 ++ {action => activity}/query_graph_rag.go | 38 +- {action => activity}/router.go | 2 +- {action => activity}/skills.go | 5 +- activity/types.go | 21 + cmd/server/main.go | 125 ---- cmd/starter/main.go | 30 +- cmd/worker/main.go | 74 +- internal/api/server.go | 117 --- internal/api/workflows.go | 704 ------------------ internal/routing/llm_client.go | 1 - internal/routing/types.go | 14 +- pkg/db/models.go | 1 + pkg/types/types.go | 223 ++++++ statemachine/types.go | 136 ---- statemachine/workflow_graph_query.go | 106 --- tests/git_test.go | 30 +- tests/routing_workflow_test.go | 38 +- tests/temporal_integration_test.go | 18 +- tests/temporal_routing_test.go | 12 +- tests/types_test.go | 24 +- {statemachine => workflow}/orchestrator.go | 2 +- .../orchestrator_recovery.go | 2 +- .../routing_workflow.go | 2 +- {statemachine => workflow}/signals.go | 2 +- {statemachine => workflow}/taskunit.go | 2 +- {statemachine => workflow}/test_workflow.go | 2 +- workflow/types.go | 20 + workflow/workflow_graph_query.go | 103 +++ 54 files changed, 847 insertions(+), 1901 deletions(-) delete mode 100644 action/implementer.go delete mode 100644 action/judge.go delete mode 100644 action/llm_inference.go delete mode 100644 action/planner.go rename {action => activity}/analysis.go (99%) rename {action => activity}/analysis_test.go (99%) rename {action => activity}/assume_role.go (99%) rename {action => activity}/canvas_compatibility.go (91%) rename {action => activity}/canvas_reasoner.go (73%) rename {action => activity}/fetch_canvas_relations.go (64%) rename {action => activity}/git.go (99%) create mode 100644 activity/implementer.go rename {action => activity}/index_graph_rag.go (67%) rename {action => activity}/integration.go (98%) create mode 100644 activity/judge.go rename {action => activity}/lessons.go (99%) rename {action => activity}/llm/client.go (98%) rename {action => activity}/llm/client_test.go (96%) create mode 100644 activity/llm_helper.go create mode 100644 activity/llm_inference.go rename {action => activity}/logger.go (98%) rename {action => activity}/memory.go (99%) rename {action => activity}/memory_test.go (99%) rename {action => activity}/notification.go (99%) rename {action => activity}/notification_test.go (99%) create mode 100644 activity/planner.go rename {action => activity}/query_graph_rag.go (64%) rename {action => activity}/router.go (99%) rename {action => activity}/skills.go (97%) create mode 100644 activity/types.go delete mode 100644 cmd/server/main.go delete mode 100644 internal/api/server.go delete mode 100644 internal/api/workflows.go create mode 100644 pkg/types/types.go delete mode 100644 statemachine/types.go delete mode 100644 statemachine/workflow_graph_query.go rename {statemachine => workflow}/orchestrator.go (99%) rename {statemachine => workflow}/orchestrator_recovery.go (99%) rename {statemachine => workflow}/routing_workflow.go (99%) rename {statemachine => workflow}/signals.go (65%) rename {statemachine => workflow}/taskunit.go (99%) rename {statemachine => workflow}/test_workflow.go (91%) create mode 100644 workflow/types.go create mode 100644 workflow/workflow_graph_query.go diff --git a/action/implementer.go b/action/implementer.go deleted file mode 100644 index bc16d84..0000000 --- a/action/implementer.go +++ /dev/null @@ -1,93 +0,0 @@ -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 -} diff --git a/action/judge.go b/action/judge.go deleted file mode 100644 index 9dc5c23..0000000 --- a/action/judge.go +++ /dev/null @@ -1,78 +0,0 @@ -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 -} diff --git a/action/llm_inference.go b/action/llm_inference.go deleted file mode 100644 index e6980ae..0000000 --- a/action/llm_inference.go +++ /dev/null @@ -1,159 +0,0 @@ -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 -} diff --git a/action/planner.go b/action/planner.go deleted file mode 100644 index 9682538..0000000 --- a/action/planner.go +++ /dev/null @@ -1,91 +0,0 @@ -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 -} diff --git a/action/analysis.go b/activity/analysis.go similarity index 99% rename from action/analysis.go rename to activity/analysis.go index 3f823fe..28a065a 100644 --- a/action/analysis.go +++ b/activity/analysis.go @@ -1,4 +1,4 @@ -package action +package activity import ( "context" diff --git a/action/analysis_test.go b/activity/analysis_test.go similarity index 99% rename from action/analysis_test.go rename to activity/analysis_test.go index 0914d32..3664bb3 100644 --- a/action/analysis_test.go +++ b/activity/analysis_test.go @@ -1,4 +1,4 @@ -package action +package activity import ( "context" diff --git a/action/assume_role.go b/activity/assume_role.go similarity index 99% rename from action/assume_role.go rename to activity/assume_role.go index cb9a07e..bfbaccb 100644 --- a/action/assume_role.go +++ b/activity/assume_role.go @@ -1,4 +1,4 @@ -package action +package activity import ( "bytes" diff --git a/action/canvas_compatibility.go b/activity/canvas_compatibility.go similarity index 91% rename from action/canvas_compatibility.go rename to activity/canvas_compatibility.go index 4d01ac8..ea23f0b 100644 --- a/action/canvas_compatibility.go +++ b/activity/canvas_compatibility.go @@ -1,14 +1,21 @@ -package action +package activity 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 @@ -41,7 +48,7 @@ type OutputField struct { // getActivitySchema returns schema from knowledge base func getActivitySchema(activityType string) (*ActivitySchema, error) { kb := knowledgeBaseData() - if kb == nil { + if kb == "" { return nil, fmt.Errorf("knowledge base not loaded") } @@ -189,7 +196,7 @@ func CheckConnectionCompatibility(sourceNode, targetNode db.WorkflowNode) []Inco } // CheckCanvasConnectivity analyzes all suggested edges for compatibility -func CheckCanvasConnectivity(nodes []db.WorkflowNode, suggestedEdges []db.WorkflowEdge) []IncompatibilityWarning { +func CheckCanvasConnectivity(nodes []db.WorkflowNode, suggestedEdges []EdgeWithWording) []IncompatibilityWarning { warnings := []IncompatibilityWarning{} nodeMap := make(map[string]db.WorkflowNode) for _, n := range nodes { @@ -214,7 +221,7 @@ func CheckCanvasConnectivity(nodes []db.WorkflowNode, suggestedEdges []db.Workfl } // IdentifyDisconnectedNodes finds nodes that can't connect to anything -func IdentifyDisconnectedNodes(nodes []db.WorkflowNode, suggestedEdges []db.WorkflowEdge) []string { +func IdentifyDisconnectedNodes(nodes []db.WorkflowNode, suggestedEdges []EdgeWithWording) []string { edgeMap := make(map[string]bool) for _, edge := range suggestedEdges { edgeMap[edge.Source] = true @@ -294,20 +301,6 @@ 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{ @@ -352,3 +345,15 @@ 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 +} diff --git a/action/canvas_reasoner.go b/activity/canvas_reasoner.go similarity index 73% rename from action/canvas_reasoner.go rename to activity/canvas_reasoner.go index 5a98143..09d6c9e 100644 --- a/action/canvas_reasoner.go +++ b/activity/canvas_reasoner.go @@ -1,44 +1,14 @@ -package action +package activity import ( "context" "encoding/json" "fmt" - "github.com/rockliang/poimen/workflows/action/llm" + "github.com/rockliang/poimen/workflows/activity/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 @@ -55,14 +25,14 @@ func CanvasReasonerActivity(ctx context.Context, in CanvasReasonerInput) (Canvas logger := newActivityLogger(ctx) output := CanvasReasonerOutput{ - SuggestedEdges: []db.WorkflowEdge{}, + SuggestedEdges: []EdgeWithWording{}, } if len(in.Nodes) == 0 { return output, fmt.Errorf("no nodes provided") } - logger.logf("info", "Analyzing canvas with %d nodes, %d edges", len(in.Nodes), len(in.Edges)) + logger.Info("Analyzing canvas with %d nodes, %d edges", len(in.Nodes), len(in.Edges)) // Build activity descriptions for LLM context nodeDesc := buildNodeDescriptions(in.Nodes) @@ -118,7 +88,7 @@ KEY RULES: Return ONLY valid JSON, no markdown code blocks.`, nodeDesc, edgeDesc, getReasoningTask(in.PreserveExisting)) - logger.logf("info", "Calling LLM reasoning (preserve_existing=%v)", in.PreserveExisting) + logger.Info("Calling LLM reasoning (preserve_existing=%v)", in.PreserveExisting) // Call LLM client, err := llm.NewClient() @@ -127,7 +97,7 @@ Return ONLY valid JSON, no markdown code blocks.`, nodeDesc, edgeDesc, getReason } response, err := client.CreateMessage(ctx, llm.MessageInput{ - Model: statemachine.ModelSpec{ + Model: ModelSpec{ ModelID: "reasoning", // Use reasoning model for complex analysis }, SystemPrompt: systemPrompt, @@ -146,13 +116,13 @@ Return ONLY valid JSON, no markdown code blocks.`, nodeDesc, edgeDesc, getReason // Parse LLM response var reasonerResp struct { - Edges []db.WorkflowEdge `json:"edges"` + Edges []EdgeWithWording `json:"edges"` Reasoning string `json:"reasoning"` Confidence float64 `json:"confidence"` } if err := json.Unmarshal([]byte(response), &reasonerResp); err != nil { - logger.logf("warn", "Failed to parse LLM response as JSON: %v", err) + logger.Warn("Failed to parse LLM response as JSON: %v", err) // Try to extract from response text output.Reasoning = response output.Confidence = 0.5 @@ -165,19 +135,19 @@ Return ONLY valid JSON, no markdown code blocks.`, nodeDesc, edgeDesc, getReason nodeMap[n.ID] = true } - validEdges := []db.WorkflowEdge{} + validEdges := []EdgeWithWording{} for _, edge := range reasonerResp.Edges { if !nodeMap[edge.Source] { - logger.logf("warn", "Suggested edge references unknown source: %s", edge.Source) + logger.Warn("Suggested edge references unknown source: %s", edge.Source) continue } if !nodeMap[edge.Target] { - logger.logf("warn", "Suggested edge references unknown target: %s", edge.Target) + logger.Warn("Suggested edge references unknown target: %s", edge.Target) continue } // Don't suggest self-loops if edge.Source == edge.Target { - logger.logf("warn", "Skipping self-loop: %s", edge.Source) + logger.Warn("Skipping self-loop: %s", edge.Source) continue } validEdges = append(validEdges, edge) @@ -191,7 +161,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.logf("warn", "Found %d incompatible edge connections", len(incompatibilities)) + logger.Warn("Found %d incompatible edge connections", len(incompatibilities)) // Generate user-friendly alerts for i, incompat := range incompatibilities { @@ -209,7 +179,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.logf("warn", "Found %d disconnected nodes", len(disconnected)) + logger.Warn("Found %d disconnected nodes", len(disconnected)) for _, nodeID := range disconnected { var label string @@ -227,7 +197,7 @@ Return ONLY valid JSON, no markdown code blocks.`, nodeDesc, edgeDesc, getReason } } - logger.logf("info", "LLM suggested %d edges with confidence %.2f | %d incompatibilities | %d disconnected", + logger.Info("LLM suggested %d edges with confidence %.2f | %d incompatibilities | %d disconnected", len(validEdges), output.Confidence, len(incompatibilities), len(disconnected)) return output, nil diff --git a/action/fetch_canvas_relations.go b/activity/fetch_canvas_relations.go similarity index 64% rename from action/fetch_canvas_relations.go rename to activity/fetch_canvas_relations.go index 3a79373..dacb45d 100644 --- a/action/fetch_canvas_relations.go +++ b/activity/fetch_canvas_relations.go @@ -1,4 +1,4 @@ -package action +package activity import ( "context" @@ -8,22 +8,6 @@ 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) @@ -35,16 +19,16 @@ func FetchCanvasRelationsActivity(ctx context.Context, input FetchCanvasRelation Relations: []EdgeWithWording{}, } - logger.logf("info", "Fetching canvas relations: %s v%d", input.WorkflowID, input.Version) + logger.Info("Fetching canvas relations: %s v%d", input.WorkflowID, input.Version) // Get database client from context or activity manager - dbClient, ok := ctx.Value("db_client").(*db.Client) + dbClient, ok := ctx.Value("db_client").(*db.DB) if !ok { return output, fmt.Errorf("database client not in context") } // Fetch workflow - workflow, err := dbClient.GetWorkflow(ctx, input.WorkflowID) + workflow, err := dbClient.FetchWorkflow(ctx, input.WorkflowID, "") if err != nil { return output, fmt.Errorf("failed to get workflow: %w", err) } @@ -68,7 +52,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.logf("warn", "Failed to fetch relations: %v", err) + logger.Warn("Failed to fetch relations: %v", err) return output, nil } @@ -85,12 +69,12 @@ func FetchCanvasRelationsActivity(ctx context.Context, input FetchCanvasRelation // Parse relation wording JSON if err := json.Unmarshal(rel.RelationWording, &edge.RelationWording); err != nil { - logger.logf("warn", "Failed to parse relation wording: %v", err) + logger.Warn("Failed to parse relation wording: %v", err) } output.Relations = append(output.Relations, edge) } - logger.logf("info", "Fetched %d nodes, %d edges, %d relations", len(output.Nodes), len(output.Edges), len(output.Relations)) + logger.Info("Fetched %d nodes, %d edges, %d relations", len(output.Nodes), len(output.Edges), len(output.Relations)) return output, nil } diff --git a/action/git.go b/activity/git.go similarity index 99% rename from action/git.go rename to activity/git.go index 193ce4b..fdf6d01 100644 --- a/action/git.go +++ b/activity/git.go @@ -1,4 +1,4 @@ -package action +package activity import ( "context" diff --git a/activity/implementer.go b/activity/implementer.go new file mode 100644 index 0000000..39eead5 --- /dev/null +++ b/activity/implementer.go @@ -0,0 +1,41 @@ +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 +} diff --git a/action/index_graph_rag.go b/activity/index_graph_rag.go similarity index 67% rename from action/index_graph_rag.go rename to activity/index_graph_rag.go index 07a70d8..3b09658 100644 --- a/action/index_graph_rag.go +++ b/activity/index_graph_rag.go @@ -1,37 +1,10 @@ -package action +package activity 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{ diff --git a/action/integration.go b/activity/integration.go similarity index 98% rename from action/integration.go rename to activity/integration.go index 0141829..0f2bad3 100644 --- a/action/integration.go +++ b/activity/integration.go @@ -1,4 +1,4 @@ -package action +package activity import ( "context" diff --git a/activity/judge.go b/activity/judge.go new file mode 100644 index 0000000..492599c --- /dev/null +++ b/activity/judge.go @@ -0,0 +1,32 @@ +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 +} diff --git a/action/lessons.go b/activity/lessons.go similarity index 99% rename from action/lessons.go rename to activity/lessons.go index b7ed2a0..4a71f64 100644 --- a/action/lessons.go +++ b/activity/lessons.go @@ -1,4 +1,4 @@ -package action +package activity import ( "context" diff --git a/action/llm/client.go b/activity/llm/client.go similarity index 98% rename from action/llm/client.go rename to activity/llm/client.go index 007fa21..7f203e3 100644 --- a/action/llm/client.go +++ b/activity/llm/client.go @@ -9,7 +9,7 @@ import ( "net/http" "os" - "github.com/rockliang/poimen/workflows/statemachine" + "github.com/rockliang/poimen/workflows/pkg/types" ) var ( @@ -54,7 +54,7 @@ func NewClient() (*OpenAIClient, error) { // MessageInput is the input to CreateMessage. type MessageInput struct { - Model statemachine.ModelSpec + Model types.ModelSpec SystemPrompt string Messages []MessageParam AuthToken string // Optional JWT token for authenticated endpoints diff --git a/action/llm/client_test.go b/activity/llm/client_test.go similarity index 96% rename from action/llm/client_test.go rename to activity/llm/client_test.go index 67782f2..e76a1d6 100644 --- a/action/llm/client_test.go +++ b/activity/llm/client_test.go @@ -4,7 +4,7 @@ import ( "context" "testing" - "github.com/rockliang/poimen/workflows/statemachine" + "github.com/rockliang/poimen/workflows/pkg/types" ) 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: statemachine.ModelSpec{ + Model: types.ModelSpec{ ModelID: tt.modelID, }, SystemPrompt: "test", diff --git a/activity/llm_helper.go b/activity/llm_helper.go new file mode 100644 index 0000000..6067da2 --- /dev/null +++ b/activity/llm_helper.go @@ -0,0 +1,47 @@ +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 +} diff --git a/activity/llm_inference.go b/activity/llm_inference.go new file mode 100644 index 0000000..d7ce112 --- /dev/null +++ b/activity/llm_inference.go @@ -0,0 +1,114 @@ +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 +} diff --git a/action/logger.go b/activity/logger.go similarity index 98% rename from action/logger.go rename to activity/logger.go index 9b806f2..35bb6a5 100644 --- a/action/logger.go +++ b/activity/logger.go @@ -1,4 +1,4 @@ -package action +package activity import ( "context" diff --git a/action/memory.go b/activity/memory.go similarity index 99% rename from action/memory.go rename to activity/memory.go index fafe76f..b4b6ad8 100644 --- a/action/memory.go +++ b/activity/memory.go @@ -1,4 +1,4 @@ -package action +package activity import ( "context" diff --git a/action/memory_test.go b/activity/memory_test.go similarity index 99% rename from action/memory_test.go rename to activity/memory_test.go index 072b06e..a2e3162 100644 --- a/action/memory_test.go +++ b/activity/memory_test.go @@ -1,4 +1,4 @@ -package action +package activity import ( "context" diff --git a/action/notification.go b/activity/notification.go similarity index 99% rename from action/notification.go rename to activity/notification.go index f694ffe..815aeca 100644 --- a/action/notification.go +++ b/activity/notification.go @@ -1,4 +1,4 @@ -package action +package activity import ( "bytes" diff --git a/action/notification_test.go b/activity/notification_test.go similarity index 99% rename from action/notification_test.go rename to activity/notification_test.go index 22ace49..38e6002 100644 --- a/action/notification_test.go +++ b/activity/notification_test.go @@ -1,4 +1,4 @@ -package action +package activity import ( "context" diff --git a/activity/planner.go b/activity/planner.go new file mode 100644 index 0000000..f0ff7dc --- /dev/null +++ b/activity/planner.go @@ -0,0 +1,47 @@ +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 +} diff --git a/action/query_graph_rag.go b/activity/query_graph_rag.go similarity index 64% rename from action/query_graph_rag.go rename to activity/query_graph_rag.go index e1a3bea..9ed39e1 100644 --- a/action/query_graph_rag.go +++ b/activity/query_graph_rag.go @@ -1,4 +1,4 @@ -package action +package activity import ( "bytes" @@ -11,38 +11,6 @@ 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) @@ -53,7 +21,7 @@ func QueryGraphRAGActivity(ctx context.Context, input GraphRAGQueryInput) (Graph Paths: []QueryPathData{}, } - logger.logf("info", "Querying GraphRAG: %s", input.Query) + logger.Info("Querying GraphRAG: %s", input.Query) // Get Memory Service URL from env memoryURL := os.Getenv("MEMORY_SERVICE_URL") @@ -127,7 +95,7 @@ func QueryGraphRAGActivity(ctx context.Context, input GraphRAGQueryInput) (Graph output.HasMore = graphResp.HasMore output.ExecutionMs = time.Since(startTime).Milliseconds() - logger.logf("info", "GraphRAG returned %d edges, %d paths in %dms", + logger.Info("GraphRAG returned %d edges, %d paths in %dms", len(output.Edges), len(output.Paths), output.ExecutionMs) return output, nil } diff --git a/action/router.go b/activity/router.go similarity index 99% rename from action/router.go rename to activity/router.go index 276017b..e0f4356 100644 --- a/action/router.go +++ b/activity/router.go @@ -1,4 +1,4 @@ -package action +package activity import ( "context" diff --git a/action/skills.go b/activity/skills.go similarity index 97% rename from action/skills.go rename to activity/skills.go index c624541..f5ecd64 100644 --- a/action/skills.go +++ b/activity/skills.go @@ -1,4 +1,4 @@ -package action +package activity import ( "context" @@ -10,12 +10,11 @@ import ( "time" "go.temporal.io/sdk/activity" - "github.com/rockliang/poimen/workflows/statemachine" ) // PrepareSkillsInput is input to PrepareSkillsActivity. type PrepareSkillsInput struct { - Skills []statemachine.SkillRef + Skills []SkillRef StreamTimeout time.Duration Provider string // pi provider name (e.g. "homelab-reasoning"); required, pi has no usable default provider } diff --git a/activity/types.go b/activity/types.go new file mode 100644 index 0000000..3f51ec4 --- /dev/null +++ b/activity/types.go @@ -0,0 +1,21 @@ +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 diff --git a/cmd/server/main.go b/cmd/server/main.go deleted file mode 100644 index e410360..0000000 --- a/cmd/server/main.go +++ /dev/null @@ -1,125 +0,0 @@ -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") -} diff --git a/cmd/starter/main.go b/cmd/starter/main.go index 8f47413..6ded4f5 100644 --- a/cmd/starter/main.go +++ b/cmd/starter/main.go @@ -11,12 +11,12 @@ import ( "time" "go.temporal.io/sdk/client" - "github.com/rockliang/poimen/workflows/action/llm" + "github.com/rockliang/poimen/workflows/activity/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/statemachine" + "github.com/rockliang/poimen/workflows/workflow" ) func main() { @@ -89,20 +89,20 @@ func main() { // Build OrchestratorInput - input := statemachine.OrchestratorInput{ + input := workflow.OrchestratorInput{ TargetRepoPath: *repoPath, RemoteURL: *remoteURL, Milestone: *milestone, DryRun: *dryRun, MaxCyclesBeforeCAN: 100, PiProvider: *piProvider, - Config: statemachine.OrchestratorConfig{ + Config: workflow.OrchestratorConfig{ SystemPrompt: "You are an expert software developer orchestrating multi-agent work.", - Skills: []statemachine.SkillRef{}, - RolePrompts: map[string]statemachine.PromptSpec{ + Skills: []workflow.SkillRef{}, + RolePrompts: map[string]workflow.PromptSpec{ "planner": { TemplateRef: "planner/default.tmpl", - Model: statemachine.ModelSpec{ + Model: workflow.ModelSpec{ ModelID: *plannerModel, Thinking: "adaptive", Effort: "high", @@ -110,7 +110,7 @@ func main() { }, "judge": { TemplateRef: "judge/default.tmpl", - Model: statemachine.ModelSpec{ + Model: workflow.ModelSpec{ ModelID: *judgeModel, Thinking: "adaptive", Effort: "high", @@ -118,12 +118,12 @@ func main() { }, "implementer": { TemplateRef: "implementer/default.tmpl", - Model: statemachine.ModelSpec{ + Model: workflow.ModelSpec{ ModelID: *implementerModel, }, }, }, - Tuning: statemachine.NewActivityTuning(), + Tuning: workflow.NewActivityTuning(), }, } @@ -144,7 +144,7 @@ func main() { run, err := c.ExecuteWorkflow(context.Background(), client.StartWorkflowOptions{ ID: workflowID, TaskQueue: "poimen-taskqueue", - }, statemachine.OrchestratorWorkflow, input) + }, workflow.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 statemachine.OrchestratorOutput + var result workflow.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 := statemachine.RoutingWorkflowInput{Spec: spec} + input := workflow.RoutingWorkflowInput{Spec: spec} run, err := c.ExecuteWorkflow(ctx, client.StartWorkflowOptions{ ID: workflowID, TaskQueue: "poimen-taskqueue", - }, statemachine.RoutingWorkflow, input) + }, workflow.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 statemachine.RoutingWorkflowOutput + var result workflow.RoutingWorkflowOutput if err := run.Get(waitCtx, &result); err != nil { fmt.Printf("\nWorkflow running (check Temporal UI for status)\n") } else { diff --git a/cmd/worker/main.go b/cmd/worker/main.go index f34f816..0f96bb4 100644 --- a/cmd/worker/main.go +++ b/cmd/worker/main.go @@ -11,11 +11,11 @@ import ( "go.temporal.io/sdk/client" "go.temporal.io/sdk/worker" - "github.com/rockliang/poimen/workflows/action" + "github.com/rockliang/poimen/workflows/activity" "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/statemachine" + "github.com/rockliang/poimen/workflows/workflow" ) func main() { @@ -48,56 +48,56 @@ func main() { } // Register all workflows - w.RegisterWorkflow(statemachine.OrchestratorWorkflow) - w.RegisterWorkflow(statemachine.TaskUnitWorkflow) - w.RegisterWorkflow(statemachine.TestWorkflow) - w.RegisterWorkflow(statemachine.RoutingWorkflow) - w.RegisterWorkflow(statemachine.WorkflowGraphQuery) + w.RegisterWorkflow(workflow.OrchestratorWorkflow) + w.RegisterWorkflow(workflow.TaskUnitWorkflow) + w.RegisterWorkflow(workflow.TestWorkflow) + w.RegisterWorkflow(workflow.RoutingWorkflow) + w.RegisterWorkflow(workflow.WorkflowGraphQuery) // Register all activities - 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) + 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) // Integration and lessons activities - register when fully tested - w.RegisterActivity(action.RunIntegrationTestActivity) - // w.RegisterActivity(action.UpdateLessonsActivity) - // w.RegisterActivity(action.ReadLessonsActivity) + w.RegisterActivity(activity.RunIntegrationTestActivity) + // w.RegisterActivity(activity.UpdateLessonsActivity) + // w.RegisterActivity(activity.ReadLessonsActivity) // Routing workflow activities - w.RegisterActivity(action.LLMRouterActivity) - w.RegisterActivity(action.ValidateWorkflowSpecActivity) - w.RegisterActivity(action.ValidateCronWorkflowSpecActivity) + w.RegisterActivity(activity.LLMRouterActivity) + w.RegisterActivity(activity.ValidateWorkflowSpecActivity) + w.RegisterActivity(activity.ValidateCronWorkflowSpecActivity) // Analysis activities - w.RegisterActivity(action.AnalyzeCodeActivity) - w.RegisterActivity(action.SecurityScanActivity) - w.RegisterActivity(action.GenerateReportActivity) + w.RegisterActivity(activity.AnalyzeCodeActivity) + w.RegisterActivity(activity.SecurityScanActivity) + w.RegisterActivity(activity.GenerateReportActivity) // Notification and utility activities - w.RegisterActivity(action.NotifyStatusActivity) - w.RegisterActivity(action.ArchiveResultsActivity) - w.RegisterActivity(action.DeploymentPreCheckActivity) - w.RegisterActivity(action.ApproveWorkflowActivity) + w.RegisterActivity(activity.NotifyStatusActivity) + w.RegisterActivity(activity.ArchiveResultsActivity) + w.RegisterActivity(activity.DeploymentPreCheckActivity) + w.RegisterActivity(activity.ApproveWorkflowActivity) // Authentication activities - w.RegisterActivity(action.AssumeRoleActivity) + w.RegisterActivity(activity.AssumeRoleActivity) // Memory activities - w.RegisterActivity(action.RetrieveMemoryActivity) + w.RegisterActivity(activity.RetrieveMemoryActivity) // GraphRAG activities - w.RegisterActivity(action.FetchCanvasRelationsActivity) - w.RegisterActivity(action.QueryGraphRAGActivity) - w.RegisterActivity(action.CanvasReasonerActivity) - w.RegisterActivity(action.IndexGraphRAGActivity) - w.RegisterActivity(action.CanvasCompatibilityActivity) + w.RegisterActivity(activity.FetchCanvasRelationsActivity) + w.RegisterActivity(activity.QueryGraphRAGActivity) + w.RegisterActivity(activity.CanvasReasonerActivity) + w.RegisterActivity(activity.IndexGraphRAGActivity) + w.RegisterActivity(activity.CanvasCompatibilityActivity) // Initialize health checker healthChecker := health.NewChecker(c) diff --git a/internal/api/server.go b/internal/api/server.go deleted file mode 100644 index ad7c206..0000000 --- a/internal/api/server.go +++ /dev/null @@ -1,117 +0,0 @@ -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) -} diff --git a/internal/api/workflows.go b/internal/api/workflows.go deleted file mode 100644 index c17b73b..0000000 --- a/internal/api/workflows.go +++ /dev/null @@ -1,704 +0,0 @@ -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), - }) -} diff --git a/internal/routing/llm_client.go b/internal/routing/llm_client.go index 4b55399..3e7c676 100644 --- a/internal/routing/llm_client.go +++ b/internal/routing/llm_client.go @@ -8,7 +8,6 @@ import ( "io" "net/http" "os" - "strings" ) var ( diff --git a/internal/routing/types.go b/internal/routing/types.go index b623c39..b052607 100644 --- a/internal/routing/types.go +++ b/internal/routing/types.go @@ -28,12 +28,16 @@ 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"` @@ -50,14 +54,18 @@ type State struct { type StateType string const ( - StateTypeTask StateType = "Task" - StateTypePass StateType = "Pass" - StateTypeFail StateType = "Fail" + StateTypeTask StateType = "Task" + StateTypePass StateType = "Pass" + StateTypeFail StateType = "Fail" + StateTypeParallel StateType = "Parallel" + + TaskActivity = "Task" ) // 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"` diff --git a/pkg/db/models.go b/pkg/db/models.go index c022c4f..ba352c4 100644 --- a/pkg/db/models.go +++ b/pkg/db/models.go @@ -23,6 +23,7 @@ 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"` } diff --git a/pkg/types/types.go b/pkg/types/types.go new file mode 100644 index 0000000..5318f87 --- /dev/null +++ b/pkg/types/types.go @@ -0,0 +1,223 @@ +// 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(), + } +} diff --git a/statemachine/types.go b/statemachine/types.go deleted file mode 100644 index cf8b403..0000000 --- a/statemachine/types.go +++ /dev/null @@ -1,136 +0,0 @@ -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 -} diff --git a/statemachine/workflow_graph_query.go b/statemachine/workflow_graph_query.go deleted file mode 100644 index 7066a16..0000000 --- a/statemachine/workflow_graph_query.go +++ /dev/null @@ -1,106 +0,0 @@ -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 -} diff --git a/tests/git_test.go b/tests/git_test.go index 87e2caf..fadff62 100644 --- a/tests/git_test.go +++ b/tests/git_test.go @@ -9,7 +9,7 @@ import ( "testing" "github.com/stretchr/testify/assert" - "github.com/rockliang/poimen/workflows/action" + "github.com/rockliang/poimen/workflows/activity" ) func TestGitCloneAndFetch(t *testing.T) { @@ -56,7 +56,7 @@ func TestGitCloneAndFetch(t *testing.T) { // Test clone into empty path ctx := context.Background() - err := action.CloneRepoActivity(ctx, action.CloneRepoInput{ + err := activity.CloneRepoActivity(ctx, activity.CloneRepoInput{ RemoteURL: sourceDir, TargetRepoPath: targetDir, }) @@ -89,7 +89,7 @@ func TestGitCloneAndFetch(t *testing.T) { } // Test fetch on existing repo - err = action.CloneRepoActivity(ctx, action.CloneRepoInput{ + err = activity.CloneRepoActivity(ctx, activity.CloneRepoInput{ RemoteURL: sourceDir, TargetRepoPath: targetDir, }) @@ -139,14 +139,14 @@ func TestGitWorktreeAdd(t *testing.T) { // Clone the repo ctx := context.Background() - err := action.CloneRepoActivity(ctx, action.CloneRepoInput{ + err := activity.CloneRepoActivity(ctx, activity.CloneRepoInput{ RemoteURL: sourceDir, TargetRepoPath: repoDir, }) assert.NoError(t, err, "clone should succeed") // Test worktree add - worktreePath, err := action.GitWorktreeAddActivity(ctx, action.GitWorktreeAddInput{ + worktreePath, err := activity.GitWorktreeAddActivity(ctx, activity.GitWorktreeAddInput{ RepoPath: repoDir, TaskID: "T0.1", }) @@ -207,14 +207,14 @@ func TestGitCommit(t *testing.T) { // Clone the repo ctx := context.Background() - err := action.CloneRepoActivity(ctx, action.CloneRepoInput{ + err := activity.CloneRepoActivity(ctx, activity.CloneRepoInput{ RemoteURL: sourceDir, TargetRepoPath: repoDir, }) assert.NoError(t, err, "clone should succeed") // Create a worktree - worktreePath, err := action.GitWorktreeAddActivity(ctx, action.GitWorktreeAddInput{ + worktreePath, err := activity.GitWorktreeAddActivity(ctx, activity.GitWorktreeAddInput{ RepoPath: repoDir, TaskID: "T0.1", }) @@ -227,7 +227,7 @@ func TestGitCommit(t *testing.T) { } // Commit changes - err = action.GitCommitActivity(ctx, action.GitCommitInput{ + err = activity.GitCommitActivity(ctx, activity.GitCommitInput{ WorktreePath: worktreePath, Message: "Add new file", }) @@ -283,14 +283,14 @@ func TestGitDiff(t *testing.T) { // Clone the repo ctx := context.Background() - err := action.CloneRepoActivity(ctx, action.CloneRepoInput{ + err := activity.CloneRepoActivity(ctx, activity.CloneRepoInput{ RemoteURL: sourceDir, TargetRepoPath: repoDir, }) assert.NoError(t, err, "clone should succeed") // Create a worktree - worktreePath, err := action.GitWorktreeAddActivity(ctx, action.GitWorktreeAddInput{ + worktreePath, err := activity.GitWorktreeAddActivity(ctx, activity.GitWorktreeAddInput{ RepoPath: repoDir, TaskID: "T0.1", }) @@ -309,7 +309,7 @@ func TestGitDiff(t *testing.T) { } // Get diff (should show the staged change) - diffOutput, err := action.GitDiffActivity(ctx, action.GitDiffInput{ + diffOutput, err := activity.GitDiffActivity(ctx, activity.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 := action.CloneRepoActivity(ctx, action.CloneRepoInput{ + err := activity.CloneRepoActivity(ctx, activity.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 := action.GitWorktreeAddActivity(ctx, action.GitWorktreeAddInput{ + worktreePath, err := activity.GitWorktreeAddActivity(ctx, activity.GitWorktreeAddInput{ RepoPath: repoDir, TaskID: taskID, }) @@ -400,7 +400,7 @@ func TestGitSquashMerge(t *testing.T) { } // Commit changes - err = action.GitCommitActivity(ctx, action.GitCommitInput{ + err = activity.GitCommitActivity(ctx, activity.GitCommitInput{ WorktreePath: worktreePath, Message: fmt.Sprintf("Task %s implementation", taskID), }) @@ -408,7 +408,7 @@ func TestGitSquashMerge(t *testing.T) { } // Perform squash merge - err = action.GitSquashMergeActivity(ctx, action.GitSquashMergeInput{ + err = activity.GitSquashMergeActivity(ctx, activity.GitSquashMergeInput{ RepoPath: repoDir, Branches: []string{"task/T0.1", "task/T0.2"}, Message: "Milestone T0: completed all tasks", diff --git a/tests/routing_workflow_test.go b/tests/routing_workflow_test.go index d9b3f58..8ef683d 100644 --- a/tests/routing_workflow_test.go +++ b/tests/routing_workflow_test.go @@ -6,7 +6,7 @@ import ( "testing" "github.com/rockliang/poimen/workflows/internal/routing" - "github.com/rockliang/poimen/workflows/statemachine" + "github.com/rockliang/poimen/workflows/workflow" "github.com/stretchr/testify/require" "go.temporal.io/sdk/testsuite" ) @@ -45,14 +45,14 @@ func TestRoutingWorkflow_SimpleWorkflow(t *testing.T) { }, } - input := statemachine.RoutingWorkflowInput{Spec: spec} + input := workflow.RoutingWorkflowInput{Spec: spec} - env.ExecuteWorkflow(statemachine.RoutingWorkflow, input) + env.ExecuteWorkflow(workflow.RoutingWorkflow, input) require.True(t, env.IsWorkflowCompleted()) require.NoError(t, env.GetWorkflowError()) - var output statemachine.RoutingWorkflowOutput + var output workflow.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 := statemachine.RoutingWorkflowInput{Spec: spec} + input := workflow.RoutingWorkflowInput{Spec: spec} - env.ExecuteWorkflow(statemachine.RoutingWorkflow, input) + env.ExecuteWorkflow(workflow.RoutingWorkflow, input) require.True(t, env.IsWorkflowCompleted()) require.NoError(t, env.GetWorkflowError()) - var output statemachine.RoutingWorkflowOutput + var output workflow.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 := statemachine.RoutingWorkflowInput{Spec: spec} + input := workflow.RoutingWorkflowInput{Spec: spec} - env.ExecuteWorkflow(statemachine.RoutingWorkflow, input) + env.ExecuteWorkflow(workflow.RoutingWorkflow, input) require.True(t, env.IsWorkflowCompleted()) require.NoError(t, env.GetWorkflowError()) - var output statemachine.RoutingWorkflowOutput + var output workflow.RoutingWorkflowOutput require.NoError(t, env.GetWorkflowResult(&output)) require.Equal(t, "COMPLETED", output.Status) } @@ -160,14 +160,14 @@ func TestRoutingWorkflow_FailState(t *testing.T) { }, } - input := statemachine.RoutingWorkflowInput{Spec: spec} + input := workflow.RoutingWorkflowInput{Spec: spec} - env.ExecuteWorkflow(statemachine.RoutingWorkflow, input) + env.ExecuteWorkflow(workflow.RoutingWorkflow, input) require.True(t, env.IsWorkflowCompleted()) require.NoError(t, env.GetWorkflowError()) - var output statemachine.RoutingWorkflowOutput + var output workflow.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 := statemachine.RoutingWorkflowInput{Spec: spec} + input := workflow.RoutingWorkflowInput{Spec: spec} - env.ExecuteWorkflow(statemachine.RoutingWorkflow, input) + env.ExecuteWorkflow(workflow.RoutingWorkflow, input) require.True(t, env.IsWorkflowCompleted()) require.NoError(t, env.GetWorkflowError()) - var output statemachine.RoutingWorkflowOutput + var output workflow.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 := statemachine.RoutingWorkflowInput{Spec: nil} + input := workflow.RoutingWorkflowInput{Spec: nil} - env.ExecuteWorkflow(statemachine.RoutingWorkflow, input) + env.ExecuteWorkflow(workflow.RoutingWorkflow, input) require.True(t, env.IsWorkflowCompleted()) require.NoError(t, env.GetWorkflowError()) - var output statemachine.RoutingWorkflowOutput + var output workflow.RoutingWorkflowOutput require.NoError(t, env.GetWorkflowResult(&output)) require.Equal(t, "FAILED", output.Status) require.Contains(t, output.Error, "empty") diff --git a/tests/temporal_integration_test.go b/tests/temporal_integration_test.go index 4a545c8..bf397c1 100644 --- a/tests/temporal_integration_test.go +++ b/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/statemachine" + "github.com/rockliang/poimen/workflows/workflow" ) // 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", - }, statemachine.TestWorkflow) + }, workflow.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 := statemachine.OrchestratorInput{ + input := workflow.OrchestratorInput{ RemoteURL: "https://forgejo.riotpiao.com/rock/poimen", TargetRepoPath: "/tmp/test-poimen-integration", Milestone: "T0", - Config: statemachine.OrchestratorConfig{ + Config: workflow.OrchestratorConfig{ SystemPrompt: "You are a code generation assistant. Generate simple test code.", - RolePrompts: map[string]statemachine.PromptSpec{ + RolePrompts: map[string]workflow.PromptSpec{ "planner": { TemplateRef: "planner/default.tmpl", - Model: statemachine.ModelSpec{ + Model: workflow.ModelSpec{ ModelID: "ornith", }, }, "judge": { TemplateRef: "judge/default.tmpl", - Model: statemachine.ModelSpec{ + Model: workflow.ModelSpec{ ModelID: "ornith", }, }, "implementer": { TemplateRef: "implementer/default.tmpl", - Model: statemachine.ModelSpec{ + Model: workflow.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", - }, statemachine.OrchestratorWorkflow, input) + }, workflow.OrchestratorWorkflow, input) assert.NoError(t, err, "failed to execute orchestrator workflow") t.Logf("✅ Orchestrator workflow started: %s", workflowID) diff --git a/tests/temporal_routing_test.go b/tests/temporal_routing_test.go index 1e1d99d..efdf169 100644 --- a/tests/temporal_routing_test.go +++ b/tests/temporal_routing_test.go @@ -9,7 +9,7 @@ import ( "time" "github.com/rockliang/poimen/workflows/internal/routing" - "github.com/rockliang/poimen/workflows/statemachine" + "github.com/rockliang/poimen/workflows/workflow" "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 := statemachine.RoutingWorkflowInput{Spec: output.Spec} + input := workflow.RoutingWorkflowInput{Spec: output.Spec} run, err := c.ExecuteWorkflow(ctx, client.StartWorkflowOptions{ ID: workflowID, TaskQueue: "poimen-taskqueue", - }, statemachine.RoutingWorkflow, input) + }, workflow.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 := statemachine.RoutingWorkflowInput{Spec: spec} + input := workflow.RoutingWorkflowInput{Spec: spec} run, err := c.ExecuteWorkflow(ctx, client.StartWorkflowOptions{ ID: workflowID, TaskQueue: "poimen-taskqueue", - }, statemachine.RoutingWorkflow, input) + }, workflow.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 statemachine.RoutingWorkflowOutput + var result workflow.RoutingWorkflowOutput err = run.Get(ctx, &result) require.NoError(t, err) diff --git a/tests/types_test.go b/tests/types_test.go index 79d360e..7d1c250 100644 --- a/tests/types_test.go +++ b/tests/types_test.go @@ -5,12 +5,12 @@ import ( "time" "github.com/stretchr/testify/assert" - "github.com/rockliang/poimen/workflows/statemachine" + "github.com/rockliang/poimen/workflows/workflow" ) func TestTypesDefaults(t *testing.T) { // Test PiRetryPolicy defaults - pr := statemachine.NewPiRetryPolicy() + pr := workflow.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 := statemachine.NewActivityTuning() + at := workflow.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 := statemachine.ModelSpec{ + spec := workflow.ModelSpec{ ModelID: "claude-opus-5", Thinking: "adaptive", Effort: "high", @@ -42,13 +42,13 @@ func TestModelSpec(t *testing.T) { } func TestPromptSpec(t *testing.T) { - spec := statemachine.PromptSpec{ + spec := workflow.PromptSpec{ TemplateRef: "planner/default.tmpl", RawTemplate: "", Variables: map[string]any{ "key": "value", }, - Model: statemachine.ModelSpec{ + Model: workflow.ModelSpec{ ModelID: "claude-opus-5", }, LessonsRef: "T0.1", @@ -61,18 +61,18 @@ func TestPromptSpec(t *testing.T) { } func TestOrchestratorConfig(t *testing.T) { - cfg := statemachine.OrchestratorConfig{ + cfg := workflow.OrchestratorConfig{ SystemPrompt: "You are an expert", - Skills: []statemachine.SkillRef{ + Skills: []workflow.SkillRef{ {Name: "golang-skills", URL: "https://example.com/skill1"}, }, - RolePrompts: map[string]statemachine.PromptSpec{ + RolePrompts: map[string]workflow.PromptSpec{ "planner": { TemplateRef: "planner/default.tmpl", - Model: statemachine.ModelSpec{ModelID: "claude-opus-5"}, + Model: workflow.ModelSpec{ModelID: "claude-opus-5"}, }, }, - Tuning: statemachine.NewActivityTuning(), + Tuning: workflow.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 := statemachine.TaskUnitInput{ + input := workflow.TaskUnitInput{ TaskID: "T0.1", RemoteURL: "https://github.com/example/repo", TargetRepoPath: "/tmp/repo", diff --git a/statemachine/orchestrator.go b/workflow/orchestrator.go similarity index 99% rename from statemachine/orchestrator.go rename to workflow/orchestrator.go index e63da74..e3e22d3 100644 --- a/statemachine/orchestrator.go +++ b/workflow/orchestrator.go @@ -1,4 +1,4 @@ -package statemachine +package workflow import ( "fmt" diff --git a/statemachine/orchestrator_recovery.go b/workflow/orchestrator_recovery.go similarity index 99% rename from statemachine/orchestrator_recovery.go rename to workflow/orchestrator_recovery.go index 2730f69..6d0042d 100644 --- a/statemachine/orchestrator_recovery.go +++ b/workflow/orchestrator_recovery.go @@ -1,4 +1,4 @@ -package statemachine +package workflow import ( "fmt" diff --git a/statemachine/routing_workflow.go b/workflow/routing_workflow.go similarity index 99% rename from statemachine/routing_workflow.go rename to workflow/routing_workflow.go index 05e8791..948cd63 100644 --- a/statemachine/routing_workflow.go +++ b/workflow/routing_workflow.go @@ -1,4 +1,4 @@ -package statemachine +package workflow import ( "fmt" diff --git a/statemachine/signals.go b/workflow/signals.go similarity index 65% rename from statemachine/signals.go rename to workflow/signals.go index 5e821db..e9599ee 100644 --- a/statemachine/signals.go +++ b/workflow/signals.go @@ -1,3 +1,3 @@ -package statemachine +package workflow // Empty stub - will be filled in T0.7 diff --git a/statemachine/taskunit.go b/workflow/taskunit.go similarity index 99% rename from statemachine/taskunit.go rename to workflow/taskunit.go index 309ec1a..3072efa 100644 --- a/statemachine/taskunit.go +++ b/workflow/taskunit.go @@ -1,4 +1,4 @@ -package statemachine +package workflow import ( "fmt" diff --git a/statemachine/test_workflow.go b/workflow/test_workflow.go similarity index 91% rename from statemachine/test_workflow.go rename to workflow/test_workflow.go index c8b4b06..ea6597e 100644 --- a/statemachine/test_workflow.go +++ b/workflow/test_workflow.go @@ -1,4 +1,4 @@ -package statemachine +package workflow import ( "go.temporal.io/sdk/workflow" diff --git a/workflow/types.go b/workflow/types.go new file mode 100644 index 0000000..a1e917e --- /dev/null +++ b/workflow/types.go @@ -0,0 +1,20 @@ +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 diff --git a/workflow/workflow_graph_query.go b/workflow/workflow_graph_query.go new file mode 100644 index 0000000..c049bfa --- /dev/null +++ b/workflow/workflow_graph_query.go @@ -0,0 +1,103 @@ +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 +}