summaryrefslogtreecommitdiff
path: root/internal/executor/executor.go
diff options
context:
space:
mode:
Diffstat (limited to 'internal/executor/executor.go')
-rw-r--r--internal/executor/executor.go33
1 files changed, 24 insertions, 9 deletions
diff --git a/internal/executor/executor.go b/internal/executor/executor.go
index 2f01b93..0ad9da9 100644
--- a/internal/executor/executor.go
+++ b/internal/executor/executor.go
@@ -273,29 +273,42 @@ func (p *Pool) getRunner(t *task.Task) (Runner, error) {
return runner, nil
}
-// decActiveAgent decrements the active counters for a finished task. Safe to
-// call multiple times — subsequent calls are no-ops via the cleaned flag.
-// Always call this before sending on resultCh so consumers observing a result
-// see the accounting already settled (no zero-count map entries lingering).
+// releaseDispatchSlot decrements the pool-level active counter and unblocks
+// the dispatch goroutine. It must be deferred at the top of execute() and
+// executeResume() so that every exit path — including early returns before
+// the per-agent slot is taken — releases the slot acquired by dispatch().
+func (p *Pool) releaseDispatchSlot() {
+ p.mu.Lock()
+ p.active--
+ p.mu.Unlock()
+ select {
+ case p.doneCh <- struct{}{}:
+ default:
+ }
+}
+
+// decActiveAgent decrements the per-agent active counter for a finished task.
+// Safe to call multiple times — subsequent calls are no-ops via the cleaned
+// flag. Always call this before sending on resultCh so consumers observing a
+// result see the accounting already settled (no zero-count map entries
+// lingering). The pool-level p.active slot is handled separately by
+// releaseDispatchSlot, which is deferred at the top of execute/executeResume.
func (p *Pool) decActiveAgent(agentType string, cleaned *bool) {
if *cleaned {
return
}
*cleaned = true
p.mu.Lock()
- p.active--
p.activePerAgent[agentType]--
if p.activePerAgent[agentType] == 0 {
delete(p.activePerAgent, agentType)
}
p.mu.Unlock()
- select {
- case p.doneCh <- struct{}{}:
- default:
- }
}
func (p *Pool) executeResume(ctx context.Context, t *task.Task, exec *storage.Execution) {
+ defer p.releaseDispatchSlot()
+
agentType := t.Agent.Type
if agentType == "" {
agentType = "claude"
@@ -892,6 +905,8 @@ func pickAgent(status SystemStatus) string {
}
func (p *Pool) execute(ctx context.Context, t *task.Task) {
+ defer p.releaseDispatchSlot()
+
// 1. Load-balanced agent selection + model classification.
p.mu.Lock()
activeTasks := make(map[string]int)