diff options
Diffstat (limited to 'internal/executor/executor.go')
| -rw-r--r-- | internal/executor/executor.go | 353 |
1 files changed, 3 insertions, 350 deletions
diff --git a/internal/executor/executor.go b/internal/executor/executor.go index 507c65b..51cf5d9 100644 --- a/internal/executor/executor.go +++ b/internal/executor/executor.go @@ -2,11 +2,9 @@ package executor import ( "context" - "encoding/json" "errors" "fmt" "log/slog" - "os/exec" "path/filepath" "strings" "sync" @@ -38,13 +36,8 @@ type Store interface { UpdateExecutionChangestats(execID string, stats *task.Changestats) error RecordAgentEvent(e storage.AgentEvent) error GetProject(id string) (*task.Project, error) - GetStory(id string) (*task.Story, error) - ListTasksByStory(storyID string) ([]*task.Task, error) - UpdateStoryStatus(id string, status task.StoryState) error CreateTask(t *task.Task) error CreateEvent(e *event.Event) error - UpdateTaskCheckerReport(id, report string) error - GetCheckerTask(checkedTaskID string) (*task.Task, error) } // LogPather is an optional interface runners can implement to provide the log @@ -400,12 +393,6 @@ func (p *Pool) executeResume(ctx context.Context, t *task.Task, exec *storage.Ex t.RepositoryURL = proj.RemoteURL } } - // Populate BranchName from Story if missing (ADR-007). - if t.BranchName == "" && t.StoryID != "" { - if story, err := p.store.GetStory(t.StoryID); err == nil && story.BranchName != "" { - t.BranchName = story.BranchName - } - } sc := newStoreChannel(p.store, t.ID) err = runner.Run(ctx, t, exec, sc) @@ -496,47 +483,11 @@ func (p *Pool) handleRunResult(ctx context.Context, t *task.Task, exec *storage. p.consecutiveFailures[agentType]++ p.mu.Unlock() } - // If this is a checker task, attach the failure report for any terminal - // failure state (FAILED, TIMED_OUT, CANCELLED, BUDGET_EXCEEDED). - if t.CheckerForTaskID != "" && exec.ErrorMsg != "" { - if reportErr := p.store.UpdateTaskCheckerReport(t.CheckerForTaskID, exec.ErrorMsg); reportErr != nil { - p.logger.Error("handleRunResult: failed to set checker report", "taskID", t.CheckerForTaskID, "error", reportErr) - } - } - if t.StoryID != "" && exec.Status == "FAILED" { - storyID := t.StoryID - errMsg := exec.ErrorMsg - go func() { - story, getErr := p.store.GetStory(storyID) - if getErr != nil { - return - } - if story.Status == task.StoryValidating { - p.checkValidationResult(ctx, storyID, task.StateFailed, errMsg) - } - }() - } } else { p.mu.Lock() p.consecutiveFailures[agentType] = 0 p.mu.Unlock() - if t.CheckerForTaskID != "" { - // Checker task succeeded — auto-accept the checked task. - exec.Status = "COMPLETED" - if err := p.store.UpdateTaskState(t.ID, task.StateCompleted); err != nil { - p.logger.Error("handleRunResult: failed to complete checker task", "taskID", t.ID, "error", err) - } - checkedTask, getErr := p.store.GetTask(t.CheckerForTaskID) - if getErr == nil { - if acceptErr := p.store.UpdateTaskState(t.CheckerForTaskID, task.StateCompleted); acceptErr != nil { - p.logger.Error("handleRunResult: failed to auto-accept checked task", "taskID", t.CheckerForTaskID, "error", acceptErr) - } else if checkedTask.StoryID != "" { - go p.checkStoryCompletion(context.Background(), checkedTask.StoryID) - } - } else { - p.logger.Error("handleRunResult: failed to get checked task", "taskID", t.CheckerForTaskID, "error", getErr) - } - } else if t.ParentTaskID == "" { + if t.ParentTaskID == "" { subtasks, subErr := p.store.ListSubtasks(t.ID) if subErr != nil { p.logger.Error("failed to list subtasks", "taskID", t.ID, "error", subErr) @@ -551,9 +502,6 @@ func (p *Pool) handleRunResult(ctx context.Context, t *task.Task, exec *storage. if err := p.store.UpdateTaskState(t.ID, task.StateReady); err != nil { p.logger.Error("failed to update task state", "taskID", t.ID, "state", task.StateReady, "error", err) } - if agentType != "local" { - go p.spawnCheckerTask(context.Background(), t, exec) - } } } else { exec.Status = "COMPLETED" @@ -562,21 +510,6 @@ func (p *Pool) handleRunResult(ctx context.Context, t *task.Task, exec *storage. } p.maybeUnblockParent(t.ParentTaskID) } - if t.StoryID != "" { - storyID := t.StoryID - go func() { - story, getErr := p.store.GetStory(storyID) - if getErr != nil { - p.logger.Error("handleRunResult: failed to get story", "storyID", storyID, "error", getErr) - return - } - if story.Status == task.StoryValidating { - p.checkValidationResult(ctx, storyID, task.StateCompleted, "") - } else { - p.checkStoryCompletion(ctx, storyID) - } - }() - } } summary := exec.Summary @@ -591,13 +524,6 @@ func (p *Pool) handleRunResult(ctx context.Context, t *task.Task, exec *storage. p.logger.Error("failed to update task summary", "taskID", t.ID, "error", summaryErr) } } - terminalFailure := exec.Status == "FAILED" || exec.Status == "TIMED_OUT" || exec.Status == "CANCELLED" || exec.Status == "BUDGET_EXCEEDED" - if t.CheckerForTaskID != "" && terminalFailure && summary != "" { - // Overwrite the initial error-message report with the richer summary. - if reportErr := p.store.UpdateTaskCheckerReport(t.CheckerForTaskID, summary); reportErr != nil { - p.logger.Error("handleRunResult: failed to update checker report with summary", "taskID", t.CheckerForTaskID, "error", reportErr) - } - } if exec.StdoutPath != "" { if cs := task.ParseChangestatFromFile(exec.StdoutPath); cs != nil { exec.Changestats = cs @@ -612,266 +538,6 @@ func (p *Pool) handleRunResult(ctx context.Context, t *task.Task, exec *storage. p.resultCh <- &Result{TaskID: t.ID, Execution: exec, Err: err} } -// checkStoryCompletion checks whether all top-level tasks in a story have reached -// a terminal success state and transitions the story to SHIPPABLE if so. -// Subtasks are intentionally excluded — a parent task reaching READY/COMPLETED -// already accounts for its subtasks. -// CheckStoryCompletion is the exported entry point for story completion checks -// called from outside the package (e.g. the API accept handler). -func (p *Pool) CheckStoryCompletion(ctx context.Context, storyID string) { - p.checkStoryCompletion(ctx, storyID) -} - -func (p *Pool) checkStoryCompletion(ctx context.Context, storyID string) { - story, err := p.store.GetStory(storyID) - if err != nil { - p.logger.Error("checkStoryCompletion: failed to get story", "storyID", storyID, "error", err) - return - } - if story.Status != task.StoryInProgress { - return // already SHIPPABLE or beyond — nothing to do - } - tasks, err := p.store.ListTasksByStory(storyID) - if err != nil { - p.logger.Error("checkStoryCompletion: failed to list tasks", "storyID", storyID, "error", err) - return - } - if len(tasks) == 0 { - return - } - topLevelCount := 0 - for _, t := range tasks { - if t.ParentTaskID != "" { - continue // subtasks are covered by their parent - } - topLevelCount++ - if t.State != task.StateCompleted { - return // not all top-level tasks done; READY alone is not sufficient (checker may be pending) - } - } - if topLevelCount == 0 { - return // no top-level tasks — don't auto-complete - } - if err := p.store.UpdateStoryStatus(storyID, task.StoryShippable); err != nil { - p.logger.Error("checkStoryCompletion: failed to update story status", "storyID", storyID, "error", err) - return - } - p.logger.Info("story transitioned to SHIPPABLE", "storyID", storyID) - // Deploy is now triggered explicitly by the human via POST /api/stories/{id}/ship. -} - -// ShipStory merges the story branch and runs the deploy script. -// Returns an error if the story is not in SHIPPABLE state. -func (p *Pool) ShipStory(ctx context.Context, storyID string) error { - story, err := p.store.GetStory(storyID) - if err != nil { - return fmt.Errorf("story not found: %w", err) - } - if story.Status != task.StoryShippable { - return fmt.Errorf("story is not SHIPPABLE (current status: %s)", story.Status) - } - go p.triggerStoryDeploy(context.Background(), storyID) - return nil -} - -// spawnCheckerTask creates and submits a checker task for the given completed task. -// Guards: not called for subtasks, checker tasks, tasks without a repository URL, -// or tasks that already have a checker. -func (p *Pool) spawnCheckerTask(ctx context.Context, checked *task.Task, exec *storage.Execution) { - // Never spawn a checker for subtasks, checker tasks, or tasks without a repository. - if checked.ParentTaskID != "" || checked.CheckerForTaskID != "" || checked.RepositoryURL == "" { - return - } - // Idempotent: don't create a second checker if one already exists. - existing, err := p.store.GetCheckerTask(checked.ID) - if err != nil { - p.logger.Error("spawnCheckerTask: GetCheckerTask failed", "taskID", checked.ID, "error", err) - return - } - if existing != nil { - return - } - - criteria := checked.AcceptanceCriteria - if criteria == "" { - criteria = checked.Agent.Instructions - } - - // Embed the execution log so the checker can verify output without needing - // Docker, API access, or repository cloning. - logSnippet := "" - if exec != nil && exec.StdoutPath != "" { - logSnippet = "\n\nExecution log (last 200 lines):\n<execution_log>\n" + - tailFile(exec.StdoutPath, 200) + - "\n</execution_log>" - } - - instructions := fmt.Sprintf(`You are validating a completed task. Do not make any changes to the code or repository. - -Task: %s -Instructions given to the implementor: -%s - -Acceptance criteria: -%s%s - -Steps: -1. Review the execution log above to determine what the agent did. -2. If the task involved code changes, use Read/Grep/Glob to inspect the repository at %s. -3. Verify each acceptance criterion is met. Run tests or make HTTP requests as needed. -4. If all criteria are satisfied, exit normally (success). -5. If any criterion is not met, use the Bash tool to exit with a non-zero code: - bash -c "exit 1" - Before exiting, write a brief summary of what failed.`, - checked.Name, checked.Agent.Instructions, criteria, logSnippet, checked.RepositoryURL) - - now := time.Now().UTC() - checker := &task.Task{ - ID: uuid.New().String(), - Name: "Check: " + checked.Name, - CheckerForTaskID: checked.ID, - RepositoryURL: checked.RepositoryURL, - Agent: task.AgentConfig{ - Type: "claude", - Instructions: instructions, - MaxBudgetUSD: 0.50, - AllowedTools: []string{"Bash", "Read", "Glob", "Grep"}, - }, - Timeout: task.Duration{Duration: 10 * time.Minute}, - Priority: task.PriorityNormal, - Tags: []string{}, - DependsOn: []string{}, - Retry: task.RetryConfig{MaxAttempts: 1, Backoff: "linear"}, - State: task.StatePending, - CreatedAt: now, - UpdatedAt: now, - } - - if err := p.store.CreateTask(checker); err != nil { - p.logger.Error("spawnCheckerTask: CreateTask failed", "error", err) - return - } - checker.State = task.StateQueued - if err := p.store.UpdateTaskState(checker.ID, task.StateQueued); err != nil { - p.logger.Error("spawnCheckerTask: UpdateTaskState failed", "error", err) - return - } - if err := p.Submit(ctx, checker); err != nil { - p.logger.Error("spawnCheckerTask: Submit failed", "error", err) - } -} - -// triggerStoryDeploy runs the project deploy script for a SHIPPABLE story -// and advances it to DEPLOYED on success. -func (p *Pool) triggerStoryDeploy(ctx context.Context, storyID string) { - story, err := p.store.GetStory(storyID) - if err != nil { - p.logger.Error("triggerStoryDeploy: failed to get story", "storyID", storyID, "error", err) - return - } - if story.ProjectID == "" { - return - } - proj, err := p.store.GetProject(story.ProjectID) - if err != nil { - p.logger.Error("triggerStoryDeploy: failed to get project", "storyID", storyID, "projectID", story.ProjectID, "error", err) - return - } - if proj.DeployScript == "" { - return - } - // Merge story branch to main before deploying (ADR-007). - if story.BranchName != "" && proj.LocalPath != "" { - mergeSteps := [][]string{ - {"git", "-C", proj.LocalPath, "fetch", "origin"}, - {"git", "-C", proj.LocalPath, "checkout", "main"}, - {"git", "-C", proj.LocalPath, "merge", "--no-ff", story.BranchName, "-m", "Merge " + story.BranchName}, - {"git", "-C", proj.LocalPath, "push", "origin", "main"}, - } - for _, args := range mergeSteps { - if mergeOut, mergeErr := exec.CommandContext(ctx, args[0], args[1:]...).CombinedOutput(); mergeErr != nil { - p.logger.Error("triggerStoryDeploy: merge failed", "cmd", args, "output", string(mergeOut), "error", mergeErr) - return - } - } - p.logger.Info("story branch merged to main", "storyID", storyID, "branch", story.BranchName) - } - out, err := exec.CommandContext(ctx, proj.DeployScript).CombinedOutput() - if err != nil { - p.logger.Error("triggerStoryDeploy: deploy script failed", "storyID", storyID, "script", proj.DeployScript, "output", string(out), "error", err) - return - } - if err := p.store.UpdateStoryStatus(storyID, task.StoryDeployed); err != nil { - p.logger.Error("triggerStoryDeploy: failed to update story status", "storyID", storyID, "error", err) - return - } - p.logger.Info("story transitioned to DEPLOYED", "storyID", storyID) - go p.createValidationTask(ctx, storyID) -} - -// createValidationTask creates a validation subtask from the story's ValidationJSON -// and transitions the story to VALIDATING. -func (p *Pool) createValidationTask(ctx context.Context, storyID string) { - story, err := p.store.GetStory(storyID) - if err != nil { - p.logger.Error("createValidationTask: failed to get story", "storyID", storyID, "error", err) - return - } - if story.ValidationJSON == "" { - p.logger.Warn("createValidationTask: story has no ValidationJSON, skipping", "storyID", storyID) - return - } - - var spec map[string]interface{} - if err := json.Unmarshal([]byte(story.ValidationJSON), &spec); err != nil { - p.logger.Error("createValidationTask: failed to parse ValidationJSON", "storyID", storyID, "error", err) - return - } - - instructions := fmt.Sprintf("Validate the deployment for story %q.\n\nValidation spec:\n%s", story.Name, story.ValidationJSON) - - now := time.Now().UTC() - vtask := &task.Task{ - ID: uuid.New().String(), - Name: fmt.Sprintf("validation: %s", story.Name), - StoryID: storyID, - State: task.StateQueued, - Agent: task.AgentConfig{Type: "claude", Instructions: instructions}, - Tags: []string{}, - DependsOn: []string{}, - CreatedAt: now, - UpdatedAt: now, - } - - if err := p.store.CreateTask(vtask); err != nil { - p.logger.Error("createValidationTask: failed to create task", "storyID", storyID, "error", err) - return - } - if err := p.store.UpdateStoryStatus(storyID, task.StoryValidating); err != nil { - p.logger.Error("createValidationTask: failed to update story status", "storyID", storyID, "error", err) - return - } - p.logger.Info("validation task created and story transitioned to VALIDATING", "storyID", storyID, "taskID", vtask.ID) - p.Submit(ctx, vtask) //nolint:errcheck -} - -// checkValidationResult inspects a completed validation task and transitions -// the story to REVIEW_READY or NEEDS_FIX accordingly. -func (p *Pool) checkValidationResult(ctx context.Context, storyID string, taskState task.State, errorMsg string) { - if taskState == task.StateCompleted { - if err := p.store.UpdateStoryStatus(storyID, task.StoryReviewReady); err != nil { - p.logger.Error("checkValidationResult: failed to update story status", "storyID", storyID, "error", err) - return - } - p.logger.Info("story transitioned to REVIEW_READY", "storyID", storyID) - } else { - if err := p.store.UpdateStoryStatus(storyID, task.StoryNeedsFix); err != nil { - p.logger.Error("checkValidationResult: failed to update story status", "storyID", storyID, "error", err) - return - } - p.logger.Info("story transitioned to NEEDS_FIX", "storyID", storyID, "error", errorMsg) - } -} // ActiveCount returns the number of currently running tasks. func (p *Pool) ActiveCount() int { @@ -1137,12 +803,6 @@ func (p *Pool) execute(ctx context.Context, t *task.Task) { t.RepositoryURL = proj.RemoteURL } } - // Populate BranchName from Story if missing (ADR-007). - if t.BranchName == "" && t.StoryID != "" { - if story, err := p.store.GetStory(t.StoryID); err == nil && story.BranchName != "" { - t.BranchName = story.BranchName - } - } // Run the task. sc := newStoreChannel(p.store, t.ID) @@ -1220,11 +880,9 @@ func (p *Pool) RecoverStaleQueued(ctx context.Context) { // RecoverStaleBlocked promotes any BLOCKED or QUEUED parent task to READY when // all of its subtasks are already COMPLETED. This handles the case where the // server was restarted after subtasks finished but before maybeUnblockParent -// could fire, and also the case where story approval pre-created subtasks -// without ever running the parent task. +// could fire. // Call this once on server startup, after RecoverStaleRunning and RecoverStaleQueued. func (p *Pool) RecoverStaleBlocked() { - ctx := context.Background() for _, state := range []task.State{task.StateBlocked, task.StateQueued} { tasks, err := p.store.ListTasks(storage.TaskFilter{State: state}) if err != nil { @@ -1235,12 +893,7 @@ func (p *Pool) RecoverStaleBlocked() { if t.ParentTaskID != "" { continue // only promote actual parents } - before := t.State p.maybeUnblockParent(t.ID) - // If the parent was promoted, check story completion. - if after, err := p.store.GetTask(t.ID); err == nil && after.State != before && t.StoryID != "" { - p.checkStoryCompletion(ctx, t.StoryID) - } } } } @@ -1318,7 +971,7 @@ func withFailureHistory(t *task.Task, execs []*storage.Execution, err error) *ta // maybeUnblockParent transitions the parent task to READY if all of its subtasks // are in the COMPLETED state. Handles both BLOCKED parents (ran, created subtasks, -// paused) and QUEUED parents (story approval created subtasks without running parent). +// paused) and QUEUED parents (subtasks created before parent ran). func (p *Pool) maybeUnblockParent(parentID string) { parent, err := p.store.GetTask(parentID) if err != nil { |