executor: deduplicate worker slots and fix git command execution

1 files changed, 150 insertions, 159 deletions

diff --git a/internal/executor/executor.go b/internal/executor/executor.go
index fbe29a8..507c65b 100644
--- a/internal/executor/executor.go
+++ b/internal/executor/executor.go
@@ -306,47 +306,40 @@ func (p *Pool) decActiveAgent(agentType string, cleaned *bool) {
 	p.mu.Unlock()
 }
 
-func (p *Pool) executeResume(ctx context.Context, t *task.Task, exec *storage.Execution) {
-	defer p.releaseDispatchSlot()
-
-	agentType := t.Agent.Type
-	if agentType == "" {
-		agentType = "claude"
-	}
-
+// withWorkerSlot manages the lifecycle of a worker: per-agent
+// slots, context timeout/cancellation, and cancellation registration. It
+// provides a finish callback to settle per-agent accounting early (e.g. before
+// reporting results).
+func (p *Pool) withWorkerSlot(ctx context.Context, t *task.Task, agentType string, enforceCapacity bool, fn func(ctx context.Context, finish func()) error) {
 	p.mu.Lock()
+	if enforceCapacity && p.activePerAgent[agentType] >= p.maxPerAgent {
+		p.mu.Unlock()
+		time.AfterFunc(p.requeueDelay, func() { p.workCh <- workItem{ctx: ctx, task: t} })
+		return
+	}
+	if deadline, ok := p.rateLimited[agentType]; ok && time.Now().After(deadline) {
+		delete(p.rateLimited, agentType)
+		agentName := agentType
+		go func() {
+			ev := storage.AgentEvent{
+				ID:        uuid.New().String(),
+				Agent:     agentName,
+				Event:     "available",
+				Timestamp: time.Now(),
+			}
+			if recErr := p.store.RecordAgentEvent(ev); recErr != nil {
+				p.logger.Warn("failed to record agent available event", "error", recErr)
+			}
+		}()
+	}
 	p.activePerAgent[agentType]++
 	p.mu.Unlock()
 
 	var cleaned bool
-	defer p.decActiveAgent(agentType, &cleaned)
-
-	runner, err := p.getRunner(t)
-	if err != nil {
-		p.logger.Error("failed to get runner for resume", "error", err, "taskID", t.ID)
+	finish := func() {
 		p.decActiveAgent(agentType, &cleaned)
-		p.resultCh <- &Result{TaskID: t.ID, Execution: exec, Err: err}
-		return
-	}
-
-	// Pre-populate log paths.
-	if lp, ok := runner.(LogPather); ok {
-		if logDir := lp.ExecLogDir(exec.ID); logDir != "" {
-			exec.StdoutPath = filepath.Join(logDir, "stdout.log")
-			exec.StderrPath = filepath.Join(logDir, "stderr.log")
-			exec.ArtifactDir = logDir
-		}
-	}
-	exec.StartTime = time.Now().UTC()
-	exec.Status = "RUNNING"
-
-	if err := p.store.CreateExecutionAndSetRunning(exec); err != nil {
-		p.logger.Error("failed to create resume execution record", "error", err)
-	}
-	select {
-	case p.startedCh <- t.ID:
-	default:
 	}
+	defer finish()
 
 	var cancel context.CancelFunc
 	if t.Timeout.Duration > 0 {
@@ -364,28 +357,68 @@ func (p *Pool) executeResume(ctx context.Context, t *task.Task, exec *storage.Ex
 		p.mu.Unlock()
 	}()
 
-	// Populate RepositoryURL from Project registry if missing (ADR-007).
-	if t.RepositoryURL == "" && t.Project != "" {
-		if proj, err := p.store.GetProject(t.Project); err == nil && proj.RemoteURL != "" {
-			t.RepositoryURL = proj.RemoteURL
-		}
+	_ = fn(ctx, finish)
+}
+
+func (p *Pool) executeResume(ctx context.Context, t *task.Task, exec *storage.Execution) {
+	agentType := t.Agent.Type
+	if agentType == "" {
+		agentType = "claude"
 	}
-	// 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
+
+	p.withWorkerSlot(ctx, t, agentType, false, func(ctx context.Context, finish func()) error {
+		runner, err := p.getRunner(t)
+		if err != nil {
+			p.logger.Error("failed to get runner for resume", "error", err, "taskID", t.ID)
+			finish()
+			p.resultCh <- &Result{TaskID: t.ID, Execution: exec, Err: err}
+			return nil
 		}
-	}
 
-	sc := newStoreChannel(p.store, t.ID)
-	err = runner.Run(ctx, t, exec, sc)
-	exec.EndTime = time.Now().UTC()
-	if sum, ok := sc.ReportedSummary(); ok {
-		exec.Summary = sum
-	}
+		// Pre-populate log paths.
+		if lp, ok := runner.(LogPather); ok {
+			if logDir := lp.ExecLogDir(exec.ID); logDir != "" {
+				exec.StdoutPath = filepath.Join(logDir, "stdout.log")
+				exec.StderrPath = filepath.Join(logDir, "stderr.log")
+				exec.ArtifactDir = logDir
+			}
+		}
+		exec.StartTime = time.Now().UTC()
+		exec.Status = "RUNNING"
+
+		if err := p.store.CreateExecutionAndSetRunning(exec); err != nil {
+			p.logger.Error("failed to create resume execution record", "error", err)
+		}
+		select {
+		case p.startedCh <- t.ID:
+		default:
+		}
+
+		// Populate RepositoryURL from Project registry if missing (ADR-007).
+		if t.RepositoryURL == "" && t.Project != "" {
+			if proj, err := p.store.GetProject(t.Project); err == nil && proj.RemoteURL != "" {
+				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)
+		exec.EndTime = time.Now().UTC()
+		if sum, ok := sc.ReportedSummary(); ok {
+			exec.Summary = sum
+		}
+
+		finish()
+		p.handleRunResult(ctx, t, exec, err, agentType)
+		return nil
+	})
 
-	p.decActiveAgent(agentType, &cleaned)
-	p.handleRunResult(ctx, t, exec, err, agentType)
 }
 
 // handleRunResult applies the shared post-run error-classification and
@@ -1042,129 +1075,87 @@ func (p *Pool) execute(ctx context.Context, t *task.Task) {
 			return
 		}
 	}
-	p.mu.Lock()
-
-	if p.activePerAgent[agentType] >= p.maxPerAgent {
-		p.mu.Unlock()
-		time.AfterFunc(p.requeueDelay, func() { p.workCh <- workItem{ctx: ctx, task: t} })
-		return
-	}
-	if deadline, ok := p.rateLimited[agentType]; ok && time.Now().After(deadline) {
-		delete(p.rateLimited, agentType)
-		agentName := agentType
-		go func() {
-			ev := storage.AgentEvent{
+	p.withWorkerSlot(ctx, t, agentType, true, func(ctx context.Context, finish func()) error {
+		runner, err := p.getRunner(t)
+		if err != nil {
+			p.logger.Error("failed to get runner", "error", err, "taskID", t.ID)
+			now := time.Now().UTC()
+			exec := &storage.Execution{
 				ID:        uuid.New().String(),
-				Agent:     agentName,
-				Event:     "available",
-				Timestamp: time.Now(),
+				TaskID:    t.ID,
+				StartTime: now,
+				EndTime:   now,
+				Status:    "FAILED",
+				ErrorMsg:  err.Error(),
 			}
-			if recErr := p.store.RecordAgentEvent(ev); recErr != nil {
-				p.logger.Warn("failed to record agent available event", "error", recErr)
+			if createErr := p.store.CreateExecution(exec); createErr != nil {
+				p.logger.Error("failed to create execution record", "error", createErr)
 			}
-		}()
-	}
-	p.activePerAgent[agentType]++
-	p.mu.Unlock()
-
-	var cleaned bool
-	defer p.decActiveAgent(agentType, &cleaned)
+			if err := p.store.UpdateTaskState(t.ID, task.StateFailed); err != nil {
+				p.logger.Error("failed to update task state", "taskID", t.ID, "state", task.StateFailed, "error", err)
+			}
+			finish()
+			p.resultCh <- &Result{TaskID: t.ID, Execution: exec, Err: err}
+			return nil
+		}
 
-	runner, err := p.getRunner(t)
-	if err != nil {
-		p.logger.Error("failed to get runner", "error", err, "taskID", t.ID)
-		now := time.Now().UTC()
+		execID := uuid.New().String()
 		exec := &storage.Execution{
-			ID:        uuid.New().String(),
+			ID:        execID,
 			TaskID:    t.ID,
-			StartTime: now,
-			EndTime:   now,
-			Status:    "FAILED",
-			ErrorMsg:  err.Error(),
-		}
-		if createErr := p.store.CreateExecution(exec); createErr != nil {
-			p.logger.Error("failed to create execution record", "error", createErr)
+			StartTime: time.Now().UTC(),
+			Status:    "RUNNING",
+			Agent:     agentType,
 		}
-		if err := p.store.UpdateTaskState(t.ID, task.StateFailed); err != nil {
-			p.logger.Error("failed to update task state", "taskID", t.ID, "state", task.StateFailed, "error", err)
-		}
-		p.decActiveAgent(agentType, &cleaned)
-		p.resultCh <- &Result{TaskID: t.ID, Execution: exec, Err: err}
-		return
-	}
-
-	execID := uuid.New().String()
-	exec := &storage.Execution{
-		ID:        execID,
-		TaskID:    t.ID,
-		StartTime: time.Now().UTC(),
-		Status:    "RUNNING",
-		Agent:     agentType,
-	}
 
-	// Pre-populate log paths so they're available in the DB immediately —
-	// before the subprocess starts — enabling live tailing and debugging.
-	if lp, ok := runner.(LogPather); ok {
-		if logDir := lp.ExecLogDir(execID); logDir != "" {
-			exec.StdoutPath = filepath.Join(logDir, "stdout.log")
-			exec.StderrPath = filepath.Join(logDir, "stderr.log")
-			exec.ArtifactDir = logDir
+		// Pre-populate log paths so they're available in the DB immediately —
+		// before the subprocess starts — enabling live tailing and debugging.
+		if lp, ok := runner.(LogPather); ok {
+			if logDir := lp.ExecLogDir(execID); logDir != "" {
+				exec.StdoutPath = filepath.Join(logDir, "stdout.log")
+				exec.StderrPath = filepath.Join(logDir, "stderr.log")
+				exec.ArtifactDir = logDir
+			}
 		}
-	}
-
-	// Record execution start atomically with the RUNNING state transition.
-	if err := p.store.CreateExecutionAndSetRunning(exec); err != nil {
-		p.logger.Error("failed to create execution record", "error", err)
-	}
-	select {
-	case p.startedCh <- t.ID:
-	default:
-	}
 
-	// Apply task timeout and register cancel so callers can stop this task.
-	var cancel context.CancelFunc
-	if t.Timeout.Duration > 0 {
-		ctx, cancel = context.WithTimeout(ctx, t.Timeout.Duration)
-	} else {
-		ctx, cancel = context.WithCancel(ctx)
-	}
-	p.mu.Lock()
-	p.cancels[t.ID] = cancel
-	p.mu.Unlock()
-	defer func() {
-		cancel()
-		p.mu.Lock()
-		delete(p.cancels, t.ID)
-		p.mu.Unlock()
-	}()
+		// Record execution start atomically with the RUNNING state transition.
+		if err := p.store.CreateExecutionAndSetRunning(exec); err != nil {
+			p.logger.Error("failed to create execution record", "error", err)
+		}
+		select {
+		case p.startedCh <- t.ID:
+		default:
+		}
 
-	// Inject prior failure history so the agent knows what went wrong before.
-	priorExecs, priorErr := p.store.ListExecutions(t.ID)
-	t = withFailureHistory(t, priorExecs, priorErr)
+		// Inject prior failure history so the agent knows what went wrong before.
+		priorExecs, priorErr := p.store.ListExecutions(t.ID)
+		t = withFailureHistory(t, priorExecs, priorErr)
 
-	// Populate RepositoryURL from Project registry if missing (ADR-007).
-	if t.RepositoryURL == "" && t.Project != "" {
-		if proj, err := p.store.GetProject(t.Project); err == nil && proj.RemoteURL != "" {
-			t.RepositoryURL = proj.RemoteURL
+		// Populate RepositoryURL from Project registry if missing (ADR-007).
+		if t.RepositoryURL == "" && t.Project != "" {
+			if proj, err := p.store.GetProject(t.Project); err == nil && proj.RemoteURL != "" {
+				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
+		// 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)
-	err = runner.Run(ctx, t, exec, sc)
-	exec.EndTime = time.Now().UTC()
-	if sum, ok := sc.ReportedSummary(); ok {
-		exec.Summary = sum
-	}
+		// Run the task.
+		sc := newStoreChannel(p.store, t.ID)
+		err = runner.Run(ctx, t, exec, sc)
+		exec.EndTime = time.Now().UTC()
+		if sum, ok := sc.ReportedSummary(); ok {
+			exec.Summary = sum
+		}
 
-	p.decActiveAgent(agentType, &cleaned)
-	p.handleRunResult(ctx, t, exec, err, agentType)
+		finish()
+		p.handleRunResult(ctx, t, exec, err, agentType)
+		return nil
+	})
 }
 
 // RecoverStaleRunning marks any tasks stuck in RUNNING state (from a previous