diff options
Diffstat (limited to 'internal/executor/executor.go')
| -rw-r--r-- | internal/executor/executor.go | 38 |
1 files changed, 20 insertions, 18 deletions
diff --git a/internal/executor/executor.go b/internal/executor/executor.go index 2505280..88a5783 100644 --- a/internal/executor/executor.go +++ b/internal/executor/executor.go @@ -11,6 +11,7 @@ import ( "sync" "time" + "github.com/google/uuid" "github.com/thepeterstone/claudomator/internal/event" "github.com/thepeterstone/claudomator/internal/llm" "github.com/thepeterstone/claudomator/internal/retry" @@ -18,7 +19,6 @@ import ( "github.com/thepeterstone/claudomator/internal/storage" "github.com/thepeterstone/claudomator/internal/story" "github.com/thepeterstone/claudomator/internal/task" - "github.com/google/uuid" ) // Store is the subset of storage.DB methods used by the Pool. @@ -88,8 +88,8 @@ type Pool struct { runners map[string]Runner store Store logger *slog.Logger - depPollInterval time.Duration // how often waitForDependencies polls; defaults to 5s - requeueDelay time.Duration // how long to wait before requeuing a blocked-per-agent task; defaults to 30s + depPollInterval time.Duration // how often waitForDependencies polls; defaults to 5s + requeueDelay time.Duration // how long to wait before requeuing a blocked-per-agent task; defaults to 30s mu sync.Mutex active int @@ -101,15 +101,15 @@ type Pool struct { // so repeated round_robin tier resolutions (see selectRung) actually // rotate across candidates rather than always picking the first one. roleTierIndex map[string]int - closed bool // set to true when Shutdown has been called - resultCh chan *Result - startedCh chan string // task IDs that just transitioned to RUNNING - workCh chan workItem // internal bounded queue; Submit enqueues here - doneCh chan struct{} // signals when a worker slot is freed - workerWg sync.WaitGroup // tracks in-flight execute/executeResume goroutines - dispatchDone chan struct{} // closed when the dispatch goroutine exits - Classifier *Classifier - LLM *llm.Client + closed bool // set to true when Shutdown has been called + resultCh chan *Result + startedCh chan string // task IDs that just transitioned to RUNNING + workCh chan workItem // internal bounded queue; Submit enqueues here + doneCh chan struct{} // signals when a worker slot is freed + workerWg sync.WaitGroup // tracks in-flight execute/executeResume goroutines + dispatchDone chan struct{} // closed when the dispatch goroutine exits + Classifier *Classifier + LLM *llm.Client // Budget gates paid providers against a rolling window; nil disables gating. Budget BudgetGate } @@ -166,9 +166,11 @@ func (p *Pool) dispatch() { p.active++ p.mu.Unlock() if item.exec != nil { - p.workerWg.Add(1); go func(i workItem) { defer p.workerWg.Done(); p.executeResume(i.ctx, i.task, i.exec) }(item) + p.workerWg.Add(1) + go func(i workItem) { defer p.workerWg.Done(); p.executeResume(i.ctx, i.task, i.exec) }(item) } else { - p.workerWg.Add(1); go func(i workItem) { defer p.workerWg.Done(); p.execute(i.ctx, i.task) }(item) + p.workerWg.Add(1) + go func(i workItem) { defer p.workerWg.Done(); p.execute(i.ctx, i.task) }(item) } break } @@ -255,10 +257,10 @@ func (p *Pool) Cancel(taskID string) bool { // resumablePoolStates are the task states that may be submitted for session resume. var resumablePoolStates = map[task.State]bool{ - task.StateBlocked: true, - task.StateTimedOut: true, - task.StateCancelled: true, - task.StateFailed: true, + task.StateBlocked: true, + task.StateTimedOut: true, + task.StateCancelled: true, + task.StateFailed: true, task.StateBudgetExceeded: true, } |