summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-rw-r--r--internal/executor/executor.go38
-rw-r--r--internal/executor/executor_test.go32
-rw-r--r--internal/task/task.go65
3 files changed, 70 insertions, 65 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
}
-
diff --git a/internal/task/task.go b/internal/task/task.go
index a95107a..3745e94 100644
--- a/internal/task/task.go
+++ b/internal/task/task.go
@@ -7,14 +7,14 @@ import (
type State string
const (
- StatePending State = "PENDING"
- StateQueued State = "QUEUED"
- StateRunning State = "RUNNING"
- StateReady State = "READY"
- StateCompleted State = "COMPLETED"
- StateFailed State = "FAILED"
- StateTimedOut State = "TIMED_OUT"
- StateCancelled State = "CANCELLED"
+ StatePending State = "PENDING"
+ StateQueued State = "QUEUED"
+ StateRunning State = "RUNNING"
+ StateReady State = "READY"
+ StateCompleted State = "COMPLETED"
+ StateFailed State = "FAILED"
+ StateTimedOut State = "TIMED_OUT"
+ StateCancelled State = "CANCELLED"
StateBudgetExceeded State = "BUDGET_EXCEEDED"
StateBlocked State = "BLOCKED"
)
@@ -28,8 +28,8 @@ const (
)
type AgentConfig struct {
- Type string `yaml:"type" json:"type"`
- Model string `yaml:"model" json:"model"`
+ Type string `yaml:"type" json:"type"`
+ Model string `yaml:"model" json:"model"`
// Role, when non-empty, names an internal/role.RoleConfig to dispatch
// through: internal/executor.Pool.execute() resolves tier 0 of the
// active role_configs row's EscalationLadder for the initial dispatch
@@ -37,7 +37,7 @@ type AgentConfig struct {
// walks the ladder on failure (retry-then-escalate). Tasks with
// Role == "" (every existing YAML/chatbot task shape) are completely
// unaffected — this is purely additive.
- Role string `yaml:"role,omitempty" json:"role,omitempty"`
+ Role string `yaml:"role,omitempty" json:"role,omitempty"`
ContextFiles []string `yaml:"context_files" json:"context_files"`
Instructions string `yaml:"instructions" json:"instructions"`
ContainerImage string `yaml:"container_image" json:"container_image"`
@@ -56,7 +56,6 @@ type AgentConfig struct {
MaxTokens int `yaml:"max_tokens,omitempty" json:"max_tokens,omitempty"`
}
-
type RetryConfig struct {
MaxAttempts int `yaml:"max_attempts" json:"max_attempts"`
Backoff string `yaml:"backoff" json:"backoff"` // "linear", "exponential"
@@ -84,34 +83,34 @@ type Interaction struct {
}
type Task struct {
- ID string `yaml:"id" json:"id"`
- ParentTaskID string `yaml:"parent_task_id" json:"parent_task_id"`
- Name string `yaml:"name" json:"name"`
- Description string `yaml:"description" json:"description"`
- Project string `yaml:"project" json:"project"` // Human-readable project name
+ ID string `yaml:"id" json:"id"`
+ ParentTaskID string `yaml:"parent_task_id" json:"parent_task_id"`
+ Name string `yaml:"name" json:"name"`
+ Description string `yaml:"description" json:"description"`
+ Project string `yaml:"project" json:"project"` // Human-readable project name
RepositoryURL string `yaml:"repository_url" json:"repository_url"`
- Agent AgentConfig `yaml:"agent" json:"agent"`
- Timeout Duration `yaml:"timeout" json:"timeout"`
- Retry RetryConfig `yaml:"retry" json:"retry"`
- Priority Priority `yaml:"priority" json:"priority"`
- Tags []string `yaml:"tags" json:"tags"`
- DependsOn []string `yaml:"depends_on" json:"depends_on"`
+ Agent AgentConfig `yaml:"agent" json:"agent"`
+ Timeout Duration `yaml:"timeout" json:"timeout"`
+ Retry RetryConfig `yaml:"retry" json:"retry"`
+ Priority Priority `yaml:"priority" json:"priority"`
+ Tags []string `yaml:"tags" json:"tags"`
+ DependsOn []string `yaml:"depends_on" json:"depends_on"`
// AcceptanceCriteria, when non-empty, states what this task's work must
// satisfy — set by a decomposing parent via spawn_subtask's
// acceptance_criteria parameter (mirrors story.Story.AcceptanceCriteria
// exactly). Unused by execution itself; a later phase's arbitrated-review
// generalization reads it when evaluating this task's work, falling back
// to the enclosing story's own AcceptanceCriteria when empty.
- AcceptanceCriteria []string `yaml:"acceptance_criteria" json:"acceptance_criteria"`
- BranchName string `yaml:"-" json:"branch_name,omitempty"`
- State State `yaml:"-" json:"state"`
- RejectionComment string `yaml:"-" json:"rejection_comment,omitempty"`
- QuestionJSON string `yaml:"-" json:"question,omitempty"`
- ElaborationInput string `yaml:"-" json:"elaboration_input,omitempty"`
- Summary string `yaml:"-" json:"summary,omitempty"`
- Interactions []Interaction `yaml:"-" json:"interactions,omitempty"`
- CreatedAt time.Time `yaml:"-" json:"created_at"`
- UpdatedAt time.Time `yaml:"-" json:"updated_at"`
+ AcceptanceCriteria []string `yaml:"acceptance_criteria" json:"acceptance_criteria"`
+ BranchName string `yaml:"-" json:"branch_name,omitempty"`
+ State State `yaml:"-" json:"state"`
+ RejectionComment string `yaml:"-" json:"rejection_comment,omitempty"`
+ QuestionJSON string `yaml:"-" json:"question,omitempty"`
+ ElaborationInput string `yaml:"-" json:"elaboration_input,omitempty"`
+ Summary string `yaml:"-" json:"summary,omitempty"`
+ Interactions []Interaction `yaml:"-" json:"interactions,omitempty"`
+ CreatedAt time.Time `yaml:"-" json:"created_at"`
+ UpdatedAt time.Time `yaml:"-" json:"updated_at"`
// NeedsReview is set by internal/scheduler.Scheduler when it resumes a
// BLOCKED role-typed task past its ask_user-timeout with a system-authored
// fallback answer rather than a real human answer — a flag for a human to
generated by cgit v1.2.3 (git 2.39.1) at 2026-07-16 07:05:33 +0000