feat: implement proper orchestrator workflow with reconciliation loop
ci / test (push) Failing after 1m2s
ci / test (push) Failing after 1m2s
Rewrite OrchestratorWorkflow as true reconciliation loop: - PlanningActivity decides what tasks to dispatch - Fan-out TaskUnit workflows for parallel execution - Each TaskUnit runs Implementer → Test → Judge → Commit - Judge reviews code quality, retries on failure with lessons - Fan-in waits for all TaskUnits - Board update and squash merge on success - continue-as-new for long-running workflows - Proper error handling and signal support Key changes: - statemachine/orchestrator.go: Reconciliation loop (Plan → Dispatch → Review → Repeat) - statemachine/taskunit.go: Task execution with retry loop & judge review - statemachine/types.go: Updated TaskUnitInput/Output for new workflow - cmd/worker/main.go: Register RunIntegrationTestActivity - action/integration.go: Renamed from integration_test.go (fix Go build issue) Models: - Planner: reasoning (OpenAI-compatible from local LLM API) - Judge: reasoning (reviews diff + tests, gates success) - Implementer: ornith:35b (executes tasks) Verification: go build ./cmd/worker ./cmd/starter ✓
This commit is contained in:
Test
+135
-126
@@ -11,151 +11,168 @@ import (
|
||||
)
|
||||
|
||||
// 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: "",
|
||||
}
|
||||
|
||||
// Step 1: Clone the repository
|
||||
activityOptions := workflow.ActivityOptions{
|
||||
StartToCloseTimeout: 10 * time.Minute,
|
||||
// Clone repo once at start
|
||||
activityOpts := workflow.ActivityOptions{
|
||||
StartToCloseTimeout: 10 * time.Minute,
|
||||
ScheduleToCloseTimeout: 15 * time.Minute,
|
||||
}
|
||||
ctxWithOptions := workflow.WithActivityOptions(ctx, activityOptions)
|
||||
|
||||
cloneErr := workflow.ExecuteActivity(
|
||||
ctxWithOptions,
|
||||
"CloneRepoActivity",
|
||||
map[string]interface{}{
|
||||
"RemoteURL": in.RemoteURL,
|
||||
"TargetRepoPath": in.TargetRepoPath,
|
||||
},
|
||||
).Get(ctx, nil)
|
||||
ctxWithOpts := workflow.WithActivityOptions(ctx, activityOpts)
|
||||
|
||||
if cloneErr != nil {
|
||||
output.LastError = fmt.Sprintf("Clone failed: %v", cloneErr)
|
||||
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
|
||||
}
|
||||
|
||||
// Step 2: Read tasks from board.md
|
||||
tasksToRun, err := readTasksFromBoard(in.TargetRepoPath)
|
||||
if err != nil {
|
||||
output.LastError = fmt.Sprintf("Failed to read tasks: %v", err)
|
||||
return output, nil
|
||||
}
|
||||
|
||||
if len(tasksToRun) == 0 {
|
||||
output.LastError = "No tasks found in board.md"
|
||||
return output, nil
|
||||
}
|
||||
|
||||
// Step 3: Process each task
|
||||
completedTasks := 0
|
||||
for _, task := range tasksToRun {
|
||||
taskID := task["id"].(string)
|
||||
taskDesc := task["description"].(string)
|
||||
|
||||
// taskID will be used for worktree and branch
|
||||
|
||||
// Add worktree
|
||||
var worktreePath string
|
||||
wtErr := workflow.ExecuteActivity(
|
||||
ctxWithOptions,
|
||||
"GitWorktreeAddActivity",
|
||||
map[string]interface{}{
|
||||
"RepoPath": in.TargetRepoPath,
|
||||
"TaskID": taskID,
|
||||
},
|
||||
).Get(ctx, &worktreePath)
|
||||
|
||||
if wtErr != nil {
|
||||
continue // Skip this task on error
|
||||
// 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)
|
||||
}
|
||||
}
|
||||
|
||||
// Call implementer to generate code (longer timeout for LLM calls)
|
||||
implOptions := workflow.ActivityOptions{
|
||||
StartToCloseTimeout: 30 * time.Minute,
|
||||
// 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, implOptions)
|
||||
|
||||
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)
|
||||
implCtx := workflow.WithActivityOptions(ctx, implOpts)
|
||||
|
||||
if implErr != nil {
|
||||
continue
|
||||
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
|
||||
}
|
||||
|
||||
// Commit changes
|
||||
commitErr := workflow.ExecuteActivity(
|
||||
ctxWithOptions,
|
||||
"GitCommitActivity",
|
||||
map[string]interface{}{
|
||||
"WorktreePath": worktreePath,
|
||||
"Message": fmt.Sprintf("%s: implementation", taskID),
|
||||
},
|
||||
).Get(ctx, nil)
|
||||
// Extract tasks to dispatch
|
||||
var tasksToDispatch []interface{}
|
||||
if tasks, ok := planOutput["TasksToDispatch"]; ok {
|
||||
tasksToDispatch = tasks.([]interface{})
|
||||
}
|
||||
|
||||
if commitErr == nil {
|
||||
completedTasks++
|
||||
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)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Step 4: Push to remote
|
||||
pushErr := workflow.ExecuteActivity(
|
||||
ctxWithOptions,
|
||||
"GitPushActivity",
|
||||
map[string]interface{}{
|
||||
"RepoPath": in.TargetRepoPath,
|
||||
},
|
||||
).Get(ctx, nil)
|
||||
|
||||
if pushErr != nil {
|
||||
output.LastError = fmt.Sprintf("Push failed: %v", pushErr)
|
||||
return output, nil
|
||||
// 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)
|
||||
}
|
||||
|
||||
// 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))
|
||||
}
|
||||
|
||||
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 {
|
||||
output.LastError = fmt.Sprintf("Merge failed: %v", mergeErr)
|
||||
return output, nil
|
||||
}
|
||||
|
||||
// Success!
|
||||
output.MilestoneComplete = true
|
||||
// Success
|
||||
output.Done = true
|
||||
output.LastError = fmt.Sprintf("Completed %d tasks successfully", completedTasks)
|
||||
output.LastError = fmt.Sprintf("Completed %d tasks in %d cycles", completedTasks, cycleCount)
|
||||
|
||||
return output, nil
|
||||
}
|
||||
@@ -192,11 +209,3 @@ func readTasksFromBoard(repoPath string) ([]map[string]interface{}, error) {
|
||||
|
||||
return tasks, nil
|
||||
}
|
||||
|
||||
// isPiStreamTimeout checks if an error is a 504 stream timeout from Pi command
|
||||
func isPiStreamTimeout(err error) bool {
|
||||
if err == nil {
|
||||
return false
|
||||
}
|
||||
return strings.Contains(err.Error(), "PiStreamTimeout")
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user