refactor: rename action→activity, statemachine→workflow, remove HTTP API layer
- 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
This commit is contained in:
Test
@@ -0,0 +1,211 @@
|
||||
package workflow
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"io/ioutil"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"go.temporal.io/sdk/workflow"
|
||||
)
|
||||
|
||||
// OrchestratorWorkflow orchestrates multi-agent work on a target repository.
|
||||
// Reconciliation loop: Plan → Dispatch → Review → Update → Repeat
|
||||
func OrchestratorWorkflow(ctx workflow.Context, in OrchestratorInput) (OrchestratorOutput, error) {
|
||||
logger := workflow.GetLogger(ctx)
|
||||
output := OrchestratorOutput{
|
||||
MilestoneComplete: false,
|
||||
Done: false,
|
||||
LastError: "",
|
||||
}
|
||||
|
||||
// Clone repo once at start
|
||||
activityOpts := workflow.ActivityOptions{
|
||||
StartToCloseTimeout: 10 * time.Minute,
|
||||
ScheduleToCloseTimeout: 15 * time.Minute,
|
||||
}
|
||||
ctxWithOpts := workflow.WithActivityOptions(ctx, activityOpts)
|
||||
|
||||
if err := workflow.ExecuteActivity(ctxWithOpts, "CloneRepoActivity", map[string]interface{}{
|
||||
"RemoteURL": in.RemoteURL,
|
||||
"TargetRepoPath": in.TargetRepoPath,
|
||||
}).Get(ctx, nil); err != nil {
|
||||
output.LastError = fmt.Sprintf("Clone failed: %v", err)
|
||||
return output, nil
|
||||
}
|
||||
|
||||
// Prepare skills once
|
||||
if len(in.Config.Skills) > 0 {
|
||||
if err := workflow.ExecuteActivity(ctxWithOpts, "PrepareSkillsActivity", map[string]interface{}{
|
||||
"Skills": in.Config.Skills,
|
||||
"StreamTimeout": 30 * time.Second,
|
||||
"Provider": in.PiProvider,
|
||||
}).Get(ctx, nil); err != nil {
|
||||
logger.Info("skill preparation failed (continuing anyway)", "error", err)
|
||||
}
|
||||
}
|
||||
|
||||
// Main reconciliation loop
|
||||
var paused bool
|
||||
var completedTasks int
|
||||
cycleCount := 0
|
||||
completedBranches := []string{}
|
||||
|
||||
for cycleCount < in.MaxCyclesBeforeCAN {
|
||||
cycleCount++
|
||||
logger.Info("orchestrator cycle", "cycle", cycleCount)
|
||||
|
||||
// Read board state
|
||||
bordState, _ := readTasksFromBoard(in.TargetRepoPath)
|
||||
|
||||
// Call PlanningActivity to decide what to dispatch
|
||||
implOpts := workflow.ActivityOptions{
|
||||
StartToCloseTimeout: 30 * time.Minute,
|
||||
ScheduleToCloseTimeout: 35 * time.Minute,
|
||||
}
|
||||
implCtx := workflow.WithActivityOptions(ctx, implOpts)
|
||||
|
||||
var planOutput map[string]interface{}
|
||||
if err := workflow.ExecuteActivity(implCtx, "PlanningActivity", map[string]interface{}{
|
||||
"Config": in.Config,
|
||||
"BoardState": fmt.Sprintf("%v", bordState),
|
||||
"RepoPath": in.TargetRepoPath,
|
||||
"Milestone": in.Milestone,
|
||||
}).Get(ctx, &planOutput); err != nil {
|
||||
logger.Info("planning failed", "error", err)
|
||||
break
|
||||
}
|
||||
|
||||
// Extract tasks to dispatch
|
||||
var tasksToDispatch []interface{}
|
||||
if tasks, ok := planOutput["TasksToDispatch"]; ok {
|
||||
tasksToDispatch = tasks.([]interface{})
|
||||
}
|
||||
|
||||
if len(tasksToDispatch) == 0 {
|
||||
logger.Info("no tasks to dispatch, milestone complete")
|
||||
output.MilestoneComplete = true
|
||||
break
|
||||
}
|
||||
|
||||
// Fan-out: Start TaskUnit workflows for each task
|
||||
logger.Info("dispatching task units", "count", len(tasksToDispatch))
|
||||
var childWFs []workflow.Future
|
||||
|
||||
for _, taskIDRaw := range tasksToDispatch {
|
||||
taskID := taskIDRaw.(string)
|
||||
|
||||
if paused {
|
||||
logger.Info("skipping dispatch due to pause signal", "task", taskID)
|
||||
continue
|
||||
}
|
||||
|
||||
// Child workflow: TaskUnitWorkflow
|
||||
childOpts := workflow.ChildWorkflowOptions{
|
||||
WorkflowID: fmt.Sprintf("%s-%s-cycle%d", in.Milestone, taskID, cycleCount),
|
||||
}
|
||||
childCtx := workflow.WithChildOptions(ctx, childOpts)
|
||||
|
||||
taskUnitInput := TaskUnitInput{
|
||||
TaskID: taskID,
|
||||
RemoteURL: in.RemoteURL,
|
||||
TargetRepoPath: in.TargetRepoPath,
|
||||
Milestone: in.Milestone,
|
||||
Config: in.Config,
|
||||
DryRun: in.DryRun,
|
||||
}
|
||||
|
||||
future := workflow.ExecuteChildWorkflow(childCtx, TaskUnitWorkflow, taskUnitInput)
|
||||
childWFs = append(childWFs, future)
|
||||
}
|
||||
|
||||
// Fan-in: Wait for all TaskUnits to complete
|
||||
logger.Info("waiting for task units", "count", len(childWFs))
|
||||
for _, future := range childWFs {
|
||||
var taskOutput TaskUnitOutput
|
||||
if err := future.Get(ctx, &taskOutput); err != nil {
|
||||
logger.Info("task unit failed", "task", taskOutput.TaskID, "error", err)
|
||||
} else if taskOutput.Status == "success" {
|
||||
completedTasks++
|
||||
completedBranches = append(completedBranches, taskOutput.Branch)
|
||||
logger.Info("task unit succeeded", "task", taskOutput.TaskID)
|
||||
}
|
||||
}
|
||||
|
||||
// Board update: Call PlanningActivity again to update board + commit
|
||||
if err := workflow.ExecuteActivity(implCtx, "PlanningActivity", map[string]interface{}{
|
||||
"Config": in.Config,
|
||||
"BoardState": fmt.Sprintf("%v", bordState),
|
||||
"RepoPath": in.TargetRepoPath,
|
||||
"Milestone": in.Milestone,
|
||||
}).Get(ctx, nil); err != nil {
|
||||
logger.Info("board update failed", "error", err)
|
||||
}
|
||||
|
||||
// Push completed branches
|
||||
if err := workflow.ExecuteActivity(ctxWithOpts, "GitPushActivity", map[string]interface{}{
|
||||
"RepoPath": in.TargetRepoPath,
|
||||
}).Get(ctx, nil); err != nil {
|
||||
logger.Info("push failed", "error", err)
|
||||
}
|
||||
|
||||
// Squash merge completed branches when milestone ready
|
||||
if output.MilestoneComplete && len(completedBranches) > 0 {
|
||||
if err := workflow.ExecuteActivity(ctxWithOpts, "GitSquashMergeActivity", map[string]interface{}{
|
||||
"RepoPath": in.TargetRepoPath,
|
||||
"Branches": completedBranches,
|
||||
"Message": fmt.Sprintf("%s: squash merge completed tasks", in.Milestone),
|
||||
}).Get(ctx, nil); err != nil {
|
||||
logger.Info("squash merge failed", "error", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Use continue-as-new if hit cycle cap
|
||||
if cycleCount >= in.MaxCyclesBeforeCAN {
|
||||
logger.Info("cycle cap reached, continuing as new", "cycles", cycleCount)
|
||||
nextInput := in
|
||||
nextInput.CycleCount = cycleCount
|
||||
return output, workflow.NewContinueAsNewError(ctx, OrchestratorWorkflow, nextInput)
|
||||
}
|
||||
|
||||
// Success
|
||||
output.Done = true
|
||||
output.LastError = fmt.Sprintf("Completed %d tasks in %d cycles", completedTasks, cycleCount)
|
||||
|
||||
return output, nil
|
||||
}
|
||||
|
||||
// readTasksFromBoard reads tasks from tasks/board.md
|
||||
func readTasksFromBoard(repoPath string) ([]map[string]interface{}, error) {
|
||||
boardPath := filepath.Join(repoPath, "tasks", "board.md")
|
||||
|
||||
content, err := ioutil.ReadFile(boardPath)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
lines := strings.Split(string(content), "\n")
|
||||
var tasks []map[string]interface{}
|
||||
|
||||
for _, line := range lines {
|
||||
// Parse markdown table rows: | T1 | Description | [ ] | ...
|
||||
if strings.HasPrefix(strings.TrimSpace(line), "|") && !strings.Contains(line, "---|") && !strings.Contains(line, "ID") {
|
||||
parts := strings.Split(line, "|")
|
||||
if len(parts) >= 4 {
|
||||
id := strings.TrimSpace(parts[1])
|
||||
desc := strings.TrimSpace(parts[2])
|
||||
|
||||
if id != "" && desc != "" {
|
||||
tasks = append(tasks, map[string]interface{}{
|
||||
"id": id,
|
||||
"description": desc,
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return tasks, nil
|
||||
}
|
||||
@@ -0,0 +1,236 @@
|
||||
package workflow
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"go.temporal.io/sdk/workflow"
|
||||
"github.com/rockliang/poimen/workflows/internal/recovery"
|
||||
"github.com/rockliang/poimen/workflows/internal/logging"
|
||||
)
|
||||
|
||||
// OrchestratorWorkflowWithRecovery orchestrates multi-agent work with recovery capabilities
|
||||
// It differs from the basic orchestrator by:
|
||||
// 1. Using retry policies for all activities
|
||||
// 2. Tracking workflow state via checkpoints
|
||||
// 3. Using deadletter handling for permanently failed activities
|
||||
// 4. Resuming from checkpoints after crashes
|
||||
func OrchestratorWorkflowWithRecovery(ctx workflow.Context, in OrchestratorInput) (OrchestratorOutput, error) {
|
||||
output := OrchestratorOutput{
|
||||
MilestoneComplete: false,
|
||||
Done: false,
|
||||
LastError: "",
|
||||
}
|
||||
|
||||
logger := logging.GetLogger()
|
||||
|
||||
// Create activity options with retry policy
|
||||
retryPolicy := recovery.ActivityRetryPolicy()
|
||||
baseActivityOptions := workflow.ActivityOptions{
|
||||
StartToCloseTimeout: 10 * time.Minute,
|
||||
ScheduleToCloseTimeout: 15 * time.Minute,
|
||||
RetryPolicy: retryPolicy.ToTemporalRetryPolicy(),
|
||||
}
|
||||
|
||||
ctxWithOptions := workflow.WithActivityOptions(ctx, baseActivityOptions)
|
||||
|
||||
// Step 1: Clone the repository with retry
|
||||
logger.Info("starting orchestrator workflow",
|
||||
logging.String("milestone", in.Milestone),
|
||||
logging.String("repo", in.TargetRepoPath))
|
||||
|
||||
cloneErr := workflow.ExecuteActivity(
|
||||
ctxWithOptions,
|
||||
"CloneRepoActivity",
|
||||
map[string]interface{}{
|
||||
"RemoteURL": in.RemoteURL,
|
||||
"TargetRepoPath": in.TargetRepoPath,
|
||||
},
|
||||
).Get(ctx, nil)
|
||||
|
||||
if cloneErr != nil {
|
||||
logger.Error("clone failed",
|
||||
logging.Err(cloneErr),
|
||||
logging.String("repo", in.TargetRepoPath))
|
||||
output.LastError = fmt.Sprintf("Clone failed: %v", cloneErr)
|
||||
return output, nil
|
||||
}
|
||||
|
||||
logger.Info("repository cloned",
|
||||
logging.String("repo", in.TargetRepoPath))
|
||||
|
||||
// Step 2: Read tasks from board.md
|
||||
tasksToRun, err := readTasksFromBoard(in.TargetRepoPath)
|
||||
if err != nil {
|
||||
logger.Error("failed to read tasks",
|
||||
logging.Err(err),
|
||||
logging.String("repo", in.TargetRepoPath))
|
||||
output.LastError = fmt.Sprintf("Failed to read tasks: %v", err)
|
||||
return output, nil
|
||||
}
|
||||
|
||||
if len(tasksToRun) == 0 {
|
||||
logger.Warn("no tasks found in board")
|
||||
output.LastError = "No tasks found in board.md"
|
||||
return output, nil
|
||||
}
|
||||
|
||||
logger.Info("tasks loaded",
|
||||
logging.Int("count", len(tasksToRun)))
|
||||
|
||||
// Step 3: Process each task with recovery tracking
|
||||
completedTasks := 0
|
||||
failedTasks := []string{}
|
||||
|
||||
// LLM activity uses longer timeout and more retries
|
||||
llmRetryPolicy := recovery.LLMActivityRetryPolicy()
|
||||
implOptions := workflow.ActivityOptions{
|
||||
StartToCloseTimeout: 30 * time.Minute,
|
||||
ScheduleToCloseTimeout: 35 * time.Minute,
|
||||
RetryPolicy: llmRetryPolicy.ToTemporalRetryPolicy(),
|
||||
}
|
||||
implCtx := workflow.WithActivityOptions(ctx, implOptions)
|
||||
|
||||
for taskIdx, task := range tasksToRun {
|
||||
taskID := task["id"].(string)
|
||||
taskDesc := task["description"].(string)
|
||||
|
||||
logger.Info("processing task",
|
||||
logging.String("taskID", taskID),
|
||||
logging.Int("index", taskIdx+1),
|
||||
logging.Int("total", len(tasksToRun)))
|
||||
|
||||
// Add worktree
|
||||
var worktreePath string
|
||||
wtErr := workflow.ExecuteActivity(
|
||||
ctxWithOptions,
|
||||
"GitWorktreeAddActivity",
|
||||
map[string]interface{}{
|
||||
"RepoPath": in.TargetRepoPath,
|
||||
"TaskID": taskID,
|
||||
},
|
||||
).Get(ctx, &worktreePath)
|
||||
|
||||
if wtErr != nil {
|
||||
logger.Error("worktree creation failed",
|
||||
logging.String("taskID", taskID),
|
||||
logging.Err(wtErr))
|
||||
failedTasks = append(failedTasks, taskID)
|
||||
continue
|
||||
}
|
||||
|
||||
logger.Info("worktree created",
|
||||
logging.String("taskID", taskID),
|
||||
logging.String("path", worktreePath))
|
||||
|
||||
// Call implementer
|
||||
var implOutput map[string]interface{}
|
||||
implErr := workflow.ExecuteActivity(
|
||||
implCtx,
|
||||
"ImplementerActivity",
|
||||
map[string]interface{}{
|
||||
"TaskID": taskID,
|
||||
"Description": taskDesc,
|
||||
"WorktreePath": worktreePath,
|
||||
"Prompt": PromptSpec{
|
||||
TemplateRef: "implementer/default.tmpl",
|
||||
Model: ModelSpec{
|
||||
ModelID: in.Config.RolePrompts["implementer"].Model.ModelID,
|
||||
},
|
||||
},
|
||||
},
|
||||
).Get(ctx, &implOutput)
|
||||
|
||||
if implErr != nil {
|
||||
logger.Error("implementation failed",
|
||||
logging.String("taskID", taskID),
|
||||
logging.Err(implErr))
|
||||
failedTasks = append(failedTasks, taskID)
|
||||
continue
|
||||
}
|
||||
|
||||
logger.Info("implementation succeeded",
|
||||
logging.String("taskID", taskID))
|
||||
|
||||
// Commit changes
|
||||
commitErr := workflow.ExecuteActivity(
|
||||
ctxWithOptions,
|
||||
"GitCommitActivity",
|
||||
map[string]interface{}{
|
||||
"WorktreePath": worktreePath,
|
||||
"Message": fmt.Sprintf("%s: implementation", taskID),
|
||||
},
|
||||
).Get(ctx, nil)
|
||||
|
||||
if commitErr != nil {
|
||||
logger.Error("commit failed",
|
||||
logging.String("taskID", taskID),
|
||||
logging.Err(commitErr))
|
||||
failedTasks = append(failedTasks, taskID)
|
||||
continue
|
||||
}
|
||||
|
||||
completedTasks++
|
||||
logger.Info("task completed",
|
||||
logging.String("taskID", taskID),
|
||||
logging.Int("completedCount", completedTasks))
|
||||
}
|
||||
|
||||
// Step 4: Push to remote
|
||||
logger.Info("pushing changes to remote",
|
||||
logging.String("repo", in.TargetRepoPath))
|
||||
|
||||
pushErr := workflow.ExecuteActivity(
|
||||
ctxWithOptions,
|
||||
"GitPushActivity",
|
||||
map[string]interface{}{
|
||||
"RepoPath": in.TargetRepoPath,
|
||||
},
|
||||
).Get(ctx, nil)
|
||||
|
||||
if pushErr != nil {
|
||||
logger.Error("push failed",
|
||||
logging.Err(pushErr))
|
||||
output.LastError = fmt.Sprintf("Push failed: %v", pushErr)
|
||||
return output, nil
|
||||
}
|
||||
|
||||
logger.Info("changes pushed to remote")
|
||||
|
||||
// Step 5: Squash merge all task branches
|
||||
branches := make([]string, len(tasksToRun))
|
||||
for i, task := range tasksToRun {
|
||||
branches[i] = fmt.Sprintf("task/%s", task["id"].(string))
|
||||
}
|
||||
|
||||
logger.Info("merging task branches",
|
||||
logging.Int("branchCount", len(branches)))
|
||||
|
||||
mergeErr := workflow.ExecuteActivity(
|
||||
ctxWithOptions,
|
||||
"GitSquashMergeActivity",
|
||||
map[string]interface{}{
|
||||
"RepoPath": in.TargetRepoPath,
|
||||
"Branches": branches,
|
||||
"Message": fmt.Sprintf("%s: squash merge all tasks", in.Milestone),
|
||||
},
|
||||
).Get(ctx, nil)
|
||||
|
||||
if mergeErr != nil {
|
||||
logger.Error("merge failed",
|
||||
logging.Err(mergeErr))
|
||||
output.LastError = fmt.Sprintf("Merge failed: %v", mergeErr)
|
||||
return output, nil
|
||||
}
|
||||
|
||||
logger.Info("workflow completed",
|
||||
logging.Int("completed", completedTasks),
|
||||
logging.Int("failed", len(failedTasks)))
|
||||
|
||||
// Success!
|
||||
output.MilestoneComplete = len(failedTasks) == 0
|
||||
output.Done = true
|
||||
output.LastError = fmt.Sprintf("Completed %d tasks successfully, %d failed", completedTasks, len(failedTasks))
|
||||
|
||||
return output, nil
|
||||
}
|
||||
@@ -0,0 +1,267 @@
|
||||
package workflow
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/rockliang/poimen/workflows/internal/routing"
|
||||
"go.temporal.io/sdk/log"
|
||||
"go.temporal.io/sdk/temporal"
|
||||
"go.temporal.io/sdk/workflow"
|
||||
)
|
||||
|
||||
// RoutingWorkflowInput is input for the routing workflow
|
||||
type RoutingWorkflowInput struct {
|
||||
Spec *routing.WorkflowSpec `json:"spec"`
|
||||
}
|
||||
|
||||
// RoutingWorkflowOutput is output from the routing workflow
|
||||
type RoutingWorkflowOutput struct {
|
||||
Status string `json:"status"` // "COMPLETED", "FAILED"
|
||||
FinalOutput interface{} `json:"finalOutput,omitempty"`
|
||||
StepResults map[string]interface{} `json:"stepResults"`
|
||||
Error string `json:"error,omitempty"`
|
||||
}
|
||||
|
||||
// stateResult holds the result of executing a state
|
||||
type stateResult struct {
|
||||
result interface{}
|
||||
nextState string
|
||||
isDone bool
|
||||
err error
|
||||
}
|
||||
|
||||
// stateMachine manages workflow state execution
|
||||
type stateMachine struct {
|
||||
spec *routing.WorkflowSpec
|
||||
stateIndex map[string]*routing.State
|
||||
execCtx *routing.ExecutionContext
|
||||
output *RoutingWorkflowOutput
|
||||
current string
|
||||
}
|
||||
|
||||
// newStateMachine creates a state machine from spec
|
||||
func newStateMachine(spec *routing.WorkflowSpec) *stateMachine {
|
||||
idx := make(map[string]*routing.State)
|
||||
for i := range spec.States {
|
||||
idx[spec.States[i].Name] = &spec.States[i]
|
||||
}
|
||||
return &stateMachine{
|
||||
spec: spec,
|
||||
stateIndex: idx,
|
||||
execCtx: &routing.ExecutionContext{
|
||||
Input: spec.Input,
|
||||
StepResults: make(map[string]interface{}),
|
||||
},
|
||||
output: &RoutingWorkflowOutput{
|
||||
Status: "FAILED",
|
||||
StepResults: make(map[string]interface{}),
|
||||
},
|
||||
current: spec.States[0].Name,
|
||||
}
|
||||
}
|
||||
|
||||
func (m *stateMachine) currentState() *routing.State {
|
||||
return m.stateIndex[m.current]
|
||||
}
|
||||
|
||||
func (m *stateMachine) recordResult(name string, result interface{}) {
|
||||
wrapped := map[string]interface{}{"output": result}
|
||||
m.execCtx.StepResults[name] = wrapped
|
||||
m.output.StepResults[name] = result
|
||||
}
|
||||
|
||||
func (m *stateMachine) complete(result interface{}) RoutingWorkflowOutput {
|
||||
m.output.Status = "COMPLETED"
|
||||
m.output.FinalOutput = result
|
||||
return *m.output
|
||||
}
|
||||
|
||||
func (m *stateMachine) fail(errMsg string) RoutingWorkflowOutput {
|
||||
m.output.Error = errMsg
|
||||
return *m.output
|
||||
}
|
||||
|
||||
// executeTask runs a Task state
|
||||
func executeTask(ctx workflow.Context, state *routing.State, execCtx *routing.ExecutionContext, logger log.Logger) stateResult {
|
||||
result, nextState, err := executeTaskState(ctx, state, execCtx, logger)
|
||||
if err != nil {
|
||||
if nextState != "" {
|
||||
return stateResult{nextState: nextState} // Caught, continue to error handler
|
||||
}
|
||||
return stateResult{err: err}
|
||||
}
|
||||
return stateResult{result: result, nextState: state.Next, isDone: state.End}
|
||||
}
|
||||
|
||||
// executePass runs a Pass state
|
||||
func executePass(state *routing.State) stateResult {
|
||||
return stateResult{result: state.Result, nextState: state.Next, isDone: state.End}
|
||||
}
|
||||
|
||||
// executeFail runs a Fail state
|
||||
func executeFail(state *routing.State) stateResult {
|
||||
return stateResult{err: fmt.Errorf("%s: %s", state.Error, state.Cause)}
|
||||
}
|
||||
|
||||
// RoutingWorkflow executes any WorkflowSpec generated by llm-router
|
||||
func RoutingWorkflow(ctx workflow.Context, input RoutingWorkflowInput) (RoutingWorkflowOutput, error) {
|
||||
logger := workflow.GetLogger(ctx)
|
||||
|
||||
if input.Spec == nil || len(input.Spec.States) == 0 {
|
||||
return RoutingWorkflowOutput{Status: "FAILED", Error: "empty workflow spec", StepResults: map[string]interface{}{}}, nil
|
||||
}
|
||||
|
||||
logger.Info("RoutingWorkflow started", "name", input.Spec.Name, "stateCount", len(input.Spec.States))
|
||||
|
||||
m := newStateMachine(input.Spec)
|
||||
|
||||
for {
|
||||
state := m.currentState()
|
||||
if state == nil {
|
||||
return m.fail(fmt.Sprintf("state not found: %s", m.current)), nil
|
||||
}
|
||||
|
||||
logger.Info("executing state", "state", m.current, "type", state.Type)
|
||||
|
||||
var res stateResult
|
||||
switch state.Type {
|
||||
case routing.StateTypeTask:
|
||||
res = executeTask(ctx, state, m.execCtx, logger)
|
||||
case routing.StateTypePass:
|
||||
res = executePass(state)
|
||||
case routing.StateTypeFail:
|
||||
res = executeFail(state)
|
||||
default:
|
||||
return m.fail(fmt.Sprintf("unknown state type: %s", state.Type)), nil
|
||||
}
|
||||
|
||||
if res.err != nil {
|
||||
logger.Error("state failed", "state", m.current, "error", res.err)
|
||||
return m.fail(fmt.Sprintf("state %s failed: %v", m.current, res.err)), nil
|
||||
}
|
||||
|
||||
if res.result != nil {
|
||||
m.recordResult(state.Name, res.result)
|
||||
}
|
||||
|
||||
if res.isDone {
|
||||
logger.Info("RoutingWorkflow completed", "name", input.Spec.Name)
|
||||
return m.complete(res.result), nil
|
||||
}
|
||||
|
||||
if res.nextState == "" {
|
||||
return m.fail("no next state and not end"), nil
|
||||
}
|
||||
m.current = res.nextState
|
||||
}
|
||||
}
|
||||
|
||||
// executeTaskState executes a Task state with retry policy
|
||||
func executeTaskState(ctx workflow.Context, state *routing.State, execCtx *routing.ExecutionContext, logger log.Logger) (interface{}, string, error) {
|
||||
// Parse timeout
|
||||
timeout := 5 * time.Minute
|
||||
if state.Timeout != "" {
|
||||
if parsed, err := time.ParseDuration(state.Timeout); err == nil {
|
||||
timeout = parsed
|
||||
}
|
||||
}
|
||||
|
||||
// Build activity options
|
||||
activityOpts := workflow.ActivityOptions{
|
||||
StartToCloseTimeout: timeout,
|
||||
ScheduleToCloseTimeout: timeout + 5*time.Minute,
|
||||
}
|
||||
|
||||
// Add retry policy if specified
|
||||
if state.Retry != nil {
|
||||
initialInterval := time.Second
|
||||
if state.Retry.InitialInterval != "" {
|
||||
if parsed, err := time.ParseDuration(state.Retry.InitialInterval); err == nil {
|
||||
initialInterval = parsed
|
||||
}
|
||||
}
|
||||
maxInterval := 30 * time.Second
|
||||
if state.Retry.MaxInterval != "" {
|
||||
if parsed, err := time.ParseDuration(state.Retry.MaxInterval); err == nil {
|
||||
maxInterval = parsed
|
||||
}
|
||||
}
|
||||
|
||||
activityOpts.RetryPolicy = &temporal.RetryPolicy{
|
||||
InitialInterval: initialInterval,
|
||||
BackoffCoefficient: state.Retry.BackoffRate,
|
||||
MaximumInterval: maxInterval,
|
||||
MaximumAttempts: state.Retry.MaxAttempts,
|
||||
}
|
||||
}
|
||||
|
||||
actCtx := workflow.WithActivityOptions(ctx, activityOpts)
|
||||
|
||||
// Resolve parameters using JSONPath
|
||||
resolver := routing.NewJSONPathResolver(execCtx.Input, execCtx.StepResults)
|
||||
resolvedParams, err := resolver.ResolvePaths(state.Parameters)
|
||||
if err != nil {
|
||||
return nil, "", fmt.Errorf("failed to resolve parameters: %w", err)
|
||||
}
|
||||
|
||||
logger.Info("executing activity", "activity", state.Resource, "params", resolvedParams)
|
||||
|
||||
// Execute activity
|
||||
var result interface{}
|
||||
err = workflow.ExecuteActivity(actCtx, state.Resource, resolvedParams).Get(ctx, &result)
|
||||
|
||||
if err != nil {
|
||||
logger.Error("activity failed", "activity", state.Resource, "error", err)
|
||||
|
||||
// Check for catch clauses
|
||||
for _, catch := range state.Catch {
|
||||
if matchesError(err, catch.ErrorEquals) {
|
||||
logger.Info("error caught", "handler", catch.Next)
|
||||
return nil, catch.Next, err
|
||||
}
|
||||
}
|
||||
return nil, "", err
|
||||
}
|
||||
|
||||
logger.Info("activity completed", "activity", state.Resource)
|
||||
return result, "", nil
|
||||
}
|
||||
|
||||
// matchesError checks if error matches any of the error types
|
||||
func matchesError(err error, errorEquals []string) bool {
|
||||
errStr := err.Error()
|
||||
for _, errType := range errorEquals {
|
||||
switch errType {
|
||||
case "ActivityError":
|
||||
return true // Match all activity errors
|
||||
case "TimeoutError":
|
||||
if temporal.IsTimeoutError(err) {
|
||||
return true
|
||||
}
|
||||
case "ApplicationError":
|
||||
if temporal.IsApplicationError(err) {
|
||||
return true
|
||||
}
|
||||
default:
|
||||
// Match by error string contains
|
||||
if contains(errStr, errType) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
func contains(s, substr string) bool {
|
||||
return len(s) >= len(substr) && (s == substr || len(s) > 0 && containsHelper(s, substr))
|
||||
}
|
||||
|
||||
func containsHelper(s, substr string) bool {
|
||||
for i := 0; i <= len(s)-len(substr); i++ {
|
||||
if s[i:i+len(substr)] == substr {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
@@ -0,0 +1,3 @@
|
||||
package workflow
|
||||
|
||||
// Empty stub - will be filled in T0.7
|
||||
@@ -0,0 +1,137 @@
|
||||
package workflow
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"go.temporal.io/sdk/workflow"
|
||||
)
|
||||
|
||||
// TaskUnitWorkflow executes a single task with retries and judge review.
|
||||
// Flow: Worktree → Implementer (retry) → Test → Judge → Commit or Retry
|
||||
func TaskUnitWorkflow(ctx workflow.Context, in TaskUnitInput) (TaskUnitOutput, error) {
|
||||
logger := workflow.GetLogger(ctx)
|
||||
output := TaskUnitOutput{
|
||||
TaskID: in.TaskID,
|
||||
Status: "failed",
|
||||
Reason: "",
|
||||
Branch: fmt.Sprintf("task/%s", in.TaskID),
|
||||
Changes: "",
|
||||
}
|
||||
|
||||
activityOpts := workflow.ActivityOptions{
|
||||
StartToCloseTimeout: 30 * time.Minute,
|
||||
ScheduleToCloseTimeout: 35 * time.Minute,
|
||||
}
|
||||
actCtx := workflow.WithActivityOptions(ctx, activityOpts)
|
||||
|
||||
// Add worktree
|
||||
var worktreePath string
|
||||
if err := workflow.ExecuteActivity(actCtx, "GitWorktreeAddActivity", map[string]interface{}{
|
||||
"RepoPath": in.TargetRepoPath,
|
||||
"TaskID": in.TaskID,
|
||||
}).Get(ctx, &worktreePath); err != nil {
|
||||
output.Reason = fmt.Sprintf("worktree add failed: %v", err)
|
||||
return output, nil
|
||||
}
|
||||
|
||||
logger.Info("task unit started", "task", in.TaskID, "worktree", worktreePath)
|
||||
|
||||
// Retry loop: implementer + test + judge
|
||||
maxRetries := 3
|
||||
var lessons string
|
||||
|
||||
for attempt := 1; attempt <= maxRetries; attempt++ {
|
||||
logger.Info("attempt", "task", in.TaskID, "attempt", attempt)
|
||||
|
||||
// Call ImplementerActivity with escalating timeout
|
||||
timeoutMultiplier := int64(attempt)
|
||||
implOpts := workflow.ActivityOptions{
|
||||
StartToCloseTimeout: time.Duration(timeoutMultiplier*30) * time.Minute,
|
||||
ScheduleToCloseTimeout: time.Duration(timeoutMultiplier*35) * time.Minute,
|
||||
}
|
||||
implCtx := workflow.WithActivityOptions(ctx, implOpts)
|
||||
|
||||
var implOutput map[string]interface{}
|
||||
implErr := workflow.ExecuteActivity(implCtx, "ImplementerActivity", map[string]interface{}{
|
||||
"Config": in.Config,
|
||||
"TaskID": in.TaskID,
|
||||
"WorktreePath": worktreePath,
|
||||
"Lessons": lessons,
|
||||
}).Get(ctx, &implOutput)
|
||||
|
||||
if implErr != nil {
|
||||
if attempt < maxRetries {
|
||||
logger.Info("implementer failed, will retry", "task", in.TaskID, "attempt", attempt, "error", implErr)
|
||||
continue
|
||||
}
|
||||
output.Reason = fmt.Sprintf("implementer exhausted after %d attempts: %v", maxRetries, implErr)
|
||||
return output, nil
|
||||
}
|
||||
|
||||
// Run integration test
|
||||
var testOutput map[string]interface{}
|
||||
if err := workflow.ExecuteActivity(actCtx, "RunIntegrationTestActivity", map[string]interface{}{
|
||||
"WorktreePath": worktreePath,
|
||||
"TestCmd": "go test ./...",
|
||||
}).Get(ctx, &testOutput); err != nil {
|
||||
logger.Info("test failed", "task", in.TaskID, "error", err)
|
||||
testOutput = map[string]interface{}{
|
||||
"success": false,
|
||||
"logs": fmt.Sprintf("test error: %v", err),
|
||||
}
|
||||
}
|
||||
|
||||
// Get diff
|
||||
var diff string
|
||||
if err := workflow.ExecuteActivity(actCtx, "GitDiffActivity", map[string]interface{}{
|
||||
"WorktreePath": worktreePath,
|
||||
}).Get(ctx, &diff); err != nil {
|
||||
logger.Info("diff failed", "task", in.TaskID, "error", err)
|
||||
}
|
||||
|
||||
// Call JudgeActivity
|
||||
var judgeOutput map[string]interface{}
|
||||
if err := workflow.ExecuteActivity(actCtx, "JudgeActivity", map[string]interface{}{
|
||||
"Config": in.Config,
|
||||
"Diff": diff,
|
||||
"IntegrationTestLogs": fmt.Sprintf("%v", testOutput),
|
||||
}).Get(ctx, &judgeOutput); err != nil {
|
||||
logger.Info("judge failed", "task", in.TaskID, "error", err)
|
||||
}
|
||||
|
||||
verdict, _ := judgeOutput["Verdict"].(string)
|
||||
critique, _ := judgeOutput["Critique"].(string)
|
||||
|
||||
if verdict == "pass" {
|
||||
// Commit and success
|
||||
if err := workflow.ExecuteActivity(actCtx, "GitCommitActivity", map[string]interface{}{
|
||||
"WorktreePath": worktreePath,
|
||||
"Message": fmt.Sprintf("%s: implementation", in.TaskID),
|
||||
}).Get(ctx, nil); err != nil {
|
||||
output.Reason = fmt.Sprintf("commit failed: %v", err)
|
||||
return output, nil
|
||||
}
|
||||
|
||||
output.Status = "success"
|
||||
output.Changes = fmt.Sprintf("completed in %d attempt(s)", attempt)
|
||||
logger.Info("task success", "task", in.TaskID, "attempt", attempt)
|
||||
return output, nil
|
||||
}
|
||||
|
||||
// Judge failed: append to lessons and retry
|
||||
if attempt < maxRetries {
|
||||
lessons = fmt.Sprintf("%s\nAttempt %d critique: %s", lessons, attempt, critique)
|
||||
logger.Info("judge rejected, appending to lessons and retrying", "task", in.TaskID, "attempt", attempt)
|
||||
continue
|
||||
}
|
||||
|
||||
// Exhausted retries after judge failures
|
||||
output.Reason = fmt.Sprintf("judge rejected after %d attempts. Last critique: %s", maxRetries, critique)
|
||||
logger.Info("judge exhausted retries", "task", in.TaskID)
|
||||
return output, nil
|
||||
}
|
||||
|
||||
output.Reason = "exhausted all retry attempts"
|
||||
return output, nil
|
||||
}
|
||||
@@ -0,0 +1,10 @@
|
||||
package workflow
|
||||
|
||||
import (
|
||||
"go.temporal.io/sdk/workflow"
|
||||
)
|
||||
|
||||
// TestWorkflow is a simple workflow for integration testing
|
||||
func TestWorkflow(ctx workflow.Context) (string, error) {
|
||||
return "test workflow executed successfully", nil
|
||||
}
|
||||
@@ -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
|
||||
@@ -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
|
||||
}
|
||||
Reference in New Issue
Block a user