From ad4fcd22ec6f7408d1efc6a2bb0009c2377f57a6 Mon Sep 17 00:00:00 2001 From: Peter Stone Date: Wed, 3 Jun 2026 23:31:20 +0000 Subject: fix: release dispatch slot on all early-return paths in execute() p.active was incremented by dispatch() before launching execute(), but execute() had four early-return paths (budget gate, dep terminal failure, deps not ready, per-agent limit) that returned before the defer decActiveAgent was registered. Each such return leaked a slot, eventually saturating maxConcurrent and permanently blocking dispatch() on doneCh. Introduce releaseDispatchSlot() (decrements p.active + signals doneCh) and defer it at the very top of execute() and executeResume(), so every exit path releases the dispatch slot. Remove p.active and doneCh management from decActiveAgent so it only handles the per-agent counter. Co-Authored-By: Claude Sonnet 4.6 --- internal/executor/executor.go | 33 ++++++++++++++++++++++++--------- 1 file changed, 24 insertions(+), 9 deletions(-) (limited to 'internal/executor') 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) -- cgit v1.2.3