diff options
Diffstat (limited to 'internal/executor')
| -rw-r--r-- | internal/executor/executor.go | 38 | ||||
| -rw-r--r-- | internal/executor/executor_test.go | 32 |
2 files changed, 38 insertions, 32 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, } diff --git a/internal/executor/executor_test.go b/internal/executor/executor_test.go index d89a9eb..e2b9b44 100644 --- a/internal/executor/executor_test.go +++ b/internal/executor/executor_test.go @@ -109,7 +109,7 @@ func makeTask(id string) *task.Task { now := time.Now().UTC() return &task.Task{ ID: id, Name: "Test " + id, - Agent: task.AgentConfig{Type: "claude", Instructions: "test"}, + Agent: task.AgentConfig{Type: "claude", Instructions: "test"}, Priority: task.PriorityNormal, Retry: task.RetryConfig{MaxAttempts: 1, Backoff: "linear"}, Tags: []string{}, @@ -1206,10 +1206,13 @@ func TestPool_Submit_ParentNotBlocked_NoTransition(t *testing.T) { // minimalMockStore is a standalone Store implementation for unit-testing Pool // methods that do not require a real SQLite database. type minimalMockStore struct { - mu sync.Mutex - tasks map[string]*task.Task - executions map[string]*storage.Execution - stateUpdates []struct{ id string; state task.State } + mu sync.Mutex + tasks map[string]*task.Task + executions map[string]*storage.Execution + stateUpdates []struct { + id string + state task.State + } questionUpdates []string summaryUpdates []struct{ taskID, summary string } changestatCalls []struct { @@ -1245,7 +1248,7 @@ func (m *minimalMockStore) ListSubtasks(parentID string) ([]*task.Task, error) { return nil, nil } func (m *minimalMockStore) ListExecutions(_ string) ([]*storage.Execution, error) { return nil, nil } -func (m *minimalMockStore) CreateExecution(e *storage.Execution) error { return nil } +func (m *minimalMockStore) CreateExecution(e *storage.Execution) error { return nil } func (m *minimalMockStore) CreateExecutionAndSetRunning(e *storage.Execution) error { return nil } @@ -1257,7 +1260,10 @@ func (m *minimalMockStore) UpdateTaskState(id string, newState task.State) error return m.updateStateErr } m.mu.Lock() - m.stateUpdates = append(m.stateUpdates, struct{ id string; state task.State }{id, newState}) + m.stateUpdates = append(m.stateUpdates, struct { + id string + state task.State + }{id, newState}) if t, ok := m.tasks[id]; ok { t.State = newState } @@ -1298,10 +1304,10 @@ func (m *minimalMockStore) UpdateExecutionChangestats(execID string, stats *task m.mu.Unlock() return nil } -func (m *minimalMockStore) RecordAgentEvent(_ storage.AgentEvent) error { return nil } -func (m *minimalMockStore) GetProject(_ string) (*task.Project, error) { return nil, nil } -func (m *minimalMockStore) CreateTask(_ *task.Task) error { return nil } -func (m *minimalMockStore) CreateEvent(_ *event.Event) error { return nil } +func (m *minimalMockStore) RecordAgentEvent(_ storage.AgentEvent) error { return nil } +func (m *minimalMockStore) GetProject(_ string) (*task.Project, error) { return nil, nil } +func (m *minimalMockStore) CreateTask(_ *task.Task) error { return nil } +func (m *minimalMockStore) CreateEvent(_ *event.Event) error { return nil } func (m *minimalMockStore) GetActiveRoleConfig(_ string) (*storage.RoleConfigRow, error) { return nil, sql.ErrNoRows } @@ -1709,7 +1715,7 @@ func TestPool_MaxPerAgent_BlocksSecondTask(t *testing.T) { runners := map[string]Runner{"claude": runner} logger := slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelError})) pool := NewPool(2, runners, store, logger) // pool size 2, but maxPerAgent=1 - pool.requeueDelay = 50 * time.Millisecond // speed up test + pool.requeueDelay = 50 * time.Millisecond // speed up test tk1 := makeTask("mpa-1") tk2 := makeTask("mpa-2") @@ -1838,7 +1844,6 @@ func TestPool_ConsecutiveFailures_ResetOnSuccess(t *testing.T) { } } - // TestPool_DependsOn_NoDeadlock verifies that a task waiting for a dependency // does NOT hold the per-agent slot, allowing the dependency to run first. func TestPool_DependsOn_NoDeadlock(t *testing.T) { @@ -1975,4 +1980,3 @@ func TestPool_Shutdown_TimesOut(t *testing.T) { } close(unblock) // cleanup } - |