summaryrefslogtreecommitdiff
path: root/internal/executor
diff options
context:
space:
mode:
Diffstat (limited to 'internal/executor')
-rw-r--r--internal/executor/executor.go150
-rw-r--r--internal/executor/executor_test.go4
-rw-r--r--internal/executor/groq_routing_test.go98
-rw-r--r--internal/executor/role_dispatch_test.go206
4 files changed, 453 insertions, 5 deletions
diff --git a/internal/executor/executor.go b/internal/executor/executor.go
index 51cf5d9..981a3ad 100644
--- a/internal/executor/executor.go
+++ b/internal/executor/executor.go
@@ -2,6 +2,7 @@ package executor
import (
"context"
+ "encoding/json"
"errors"
"fmt"
"log/slog"
@@ -13,6 +14,7 @@ import (
"github.com/thepeterstone/claudomator/internal/event"
"github.com/thepeterstone/claudomator/internal/llm"
"github.com/thepeterstone/claudomator/internal/retry"
+ "github.com/thepeterstone/claudomator/internal/role"
"github.com/thepeterstone/claudomator/internal/storage"
"github.com/thepeterstone/claudomator/internal/task"
"github.com/google/uuid"
@@ -38,6 +40,11 @@ type Store interface {
GetProject(id string) (*task.Project, error)
CreateTask(t *task.Task) error
CreateEvent(e *event.Event) error
+ // GetActiveRoleConfig returns the active role_configs row for a role
+ // (see internal/role.RoleConfig), or an error (typically sql.ErrNoRows)
+ // if none is active. Used by execute() to resolve tier 0 of a role-typed
+ // task's escalation ladder.
+ GetActiveRoleConfig(role string) (*storage.RoleConfigRow, error)
}
// LogPather is an optional interface runners can implement to provide the log
@@ -77,6 +84,10 @@ type Pool struct {
rateLimited map[string]time.Time // agentType -> until
cancels map[string]context.CancelFunc // taskID → cancel
consecutiveFailures map[string]int // agentType -> count
+ // roleTierIndex tracks a rotating candidate index per "role:tierIndex" key
+ // 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
@@ -119,6 +130,7 @@ func NewPool(maxConcurrent int, runners map[string]Runner, store Store, logger *
rateLimited: make(map[string]time.Time),
cancels: make(map[string]context.CancelFunc),
consecutiveFailures: make(map[string]int),
+ roleTierIndex: make(map[string]int),
resultCh: make(chan *Result, maxConcurrent*2),
startedCh: make(chan string, maxConcurrent*2),
workCh: make(chan workItem, maxConcurrent*10+100),
@@ -574,6 +586,85 @@ func (p *Pool) AgentStatuses() []AgentStatusInfo {
return out
}
+// decodeRoleConfig unmarshals a storage.RoleConfigRow's raw ConfigJSON into a
+// role.RoleConfig. storage intentionally doesn't depend on internal/role, so
+// this decode step lives in each consumer package (executor, scheduler, api).
+func decodeRoleConfig(row *storage.RoleConfigRow) (role.RoleConfig, error) {
+ var rc role.RoleConfig
+ if row == nil {
+ return rc, fmt.Errorf("nil role config row")
+ }
+ if err := json.Unmarshal([]byte(row.ConfigJSON), &rc); err != nil {
+ return rc, fmt.Errorf("decoding role config: %w", err)
+ }
+ return rc, nil
+}
+
+// findTierIndex returns the index of the first tier in ladder whose
+// Candidates contains (provider, model), or -1 if none match. Used to
+// determine which tier an already-resolved Agent.Type/Model (set by
+// internal/scheduler for a retry/escalation resubmit) belongs to, so the new
+// execution's EscalationRung can be stamped correctly without re-resolving.
+func findTierIndex(ladder []role.Tier, provider, model string) int {
+ for i, tier := range ladder {
+ for _, c := range tier.Candidates {
+ if c.Provider == provider && c.Model == model {
+ return i
+ }
+ }
+ }
+ return -1
+}
+
+// selectRung picks one candidate rung from tier for roleName's tierIdx-th
+// tier. "single" selection mode (or a single-candidate tier) always returns
+// Candidates[0]. "round_robin" (the default) rotates through candidates
+// across successive calls via p.roleTierIndex, skipping any provider
+// currently in p.rateLimited; if every candidate is rate-limited it falls
+// back to the one whose rate limit clears soonest (same tie-break spirit as
+// pickAgent's rate-limited fallback).
+func (p *Pool) selectRung(roleName string, tierIdx int, tier role.Tier) role.Rung {
+ if len(tier.Candidates) == 0 {
+ return role.Rung{}
+ }
+ if tier.EffectiveSelectionMode() == "single" || len(tier.Candidates) == 1 {
+ return tier.Candidates[0]
+ }
+
+ key := fmt.Sprintf("%s:%d", roleName, tierIdx)
+ n := len(tier.Candidates)
+ now := time.Now()
+
+ p.mu.Lock()
+ defer p.mu.Unlock()
+ if p.roleTierIndex == nil {
+ p.roleTierIndex = make(map[string]int)
+ }
+ start := p.roleTierIndex[key] % n
+ p.roleTierIndex[key] = (start + 1) % n
+
+ for i := 0; i < n; i++ {
+ idx := (start + i) % n
+ cand := tier.Candidates[idx]
+ if deadline, limited := p.rateLimited[cand.Provider]; !limited || now.After(deadline) {
+ return cand
+ }
+ }
+
+ // All candidates are currently rate-limited: fall back to the one
+ // clearing soonest.
+ bestIdx := 0
+ var bestDeadline time.Time
+ for i, cand := range tier.Candidates {
+ d := p.rateLimited[cand.Provider]
+ if i == 0 || d.Before(bestDeadline) {
+ bestDeadline = d
+ bestIdx = i
+ }
+ }
+ return tier.Candidates[bestIdx]
+}
+
// pickAgent selects the best agent from the given SystemStatus using explicit
// load balancing: prefer the available (non-rate-limited) agent with the fewest
// active tasks. If all agents are rate-limited, fall back to fewest active.
@@ -608,6 +699,50 @@ func pickAgent(status SystemStatus) string {
func (p *Pool) execute(ctx context.Context, t *task.Task) {
defer p.releaseDispatchSlot()
+ // 0. Role-based dispatch (additive; every existing task shape has
+ // Agent.Role == "" and takes none of the branches below). For a
+ // role-typed task that hasn't yet been assigned a concrete Agent.Type
+ // (the initial dispatch — Type == ""), resolve tier 0 of the active
+ // role_configs row's EscalationLadder and apply it to a copy of t, the
+ // same copy-before-mutate pattern withFailureHistory uses. Escalating to
+ // later tiers on failure is internal/scheduler's job, not execute()'s —
+ // by the time a role-typed task reaches execute() a second time with
+ // Agent.Type already set (scheduler-driven retry/escalation resubmit),
+ // this block only looks up which tier that (Type, Model) pair belongs to
+ // (read-only) so the new execution's EscalationRung can be stamped
+ // correctly; it does not re-resolve or mutate the task.
+ resolvedRung := -1
+ if t.Agent.Role != "" {
+ if row, err := p.store.GetActiveRoleConfig(t.Agent.Role); err != nil {
+ p.logger.Warn("no active role config; dispatching without role resolution", "role", t.Agent.Role, "taskID", t.ID, "error", err)
+ } else if rc, err := decodeRoleConfig(row); err != nil {
+ p.logger.Error("failed to decode role config", "role", t.Agent.Role, "taskID", t.ID, "error", err)
+ } else if len(rc.EscalationLadder) > 0 {
+ if t.Agent.Type == "" {
+ tier0 := rc.EscalationLadder[0]
+ selected := p.selectRung(t.Agent.Role, 0, tier0)
+ if selected.Provider != "" {
+ nt := *t
+ nt.Agent = t.Agent
+ nt.Agent.Type = selected.Provider
+ nt.Agent.Model = selected.Model
+ if rc.SystemPrompt != "" {
+ if nt.Agent.SystemPromptAppend != "" {
+ nt.Agent.SystemPromptAppend = rc.SystemPrompt + "\n\n" + nt.Agent.SystemPromptAppend
+ } else {
+ nt.Agent.SystemPromptAppend = rc.SystemPrompt
+ }
+ }
+ t = &nt
+ resolvedRung = 0
+ p.logger.Info("role dispatch resolved tier 0", "role", t.Agent.Role, "taskID", t.ID, "provider", selected.Provider, "model", selected.Model)
+ }
+ } else {
+ resolvedRung = findTierIndex(rc.EscalationLadder, t.Agent.Type, t.Agent.Model)
+ }
+ }
+ }
+
// 1. Load-balanced agent selection + model classification.
p.mu.Lock()
activeTasks := make(map[string]int)
@@ -766,12 +901,17 @@ func (p *Pool) execute(ctx context.Context, t *task.Task) {
}
execID := uuid.New().String()
+ escalationRung := resolvedRung
+ if escalationRung < 0 {
+ escalationRung = 0
+ }
exec := &storage.Execution{
- ID: execID,
- TaskID: t.ID,
- StartTime: time.Now().UTC(),
- Status: "RUNNING",
- Agent: agentType,
+ ID: execID,
+ TaskID: t.ID,
+ StartTime: time.Now().UTC(),
+ Status: "RUNNING",
+ Agent: agentType,
+ EscalationRung: escalationRung,
}
// Pre-populate log paths so they're available in the DB immediately —
diff --git a/internal/executor/executor_test.go b/internal/executor/executor_test.go
index cb6554c..8eb7e72 100644
--- a/internal/executor/executor_test.go
+++ b/internal/executor/executor_test.go
@@ -2,6 +2,7 @@ package executor
import (
"context"
+ "database/sql"
"fmt"
"log/slog"
"os"
@@ -1162,6 +1163,9 @@ func (m *minimalMockStore) RecordAgentEvent(_ storage.AgentEvent) error {
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
+}
func (m *minimalMockStore) lastStateUpdate() (string, task.State, bool) {
m.mu.Lock()
diff --git a/internal/executor/groq_routing_test.go b/internal/executor/groq_routing_test.go
new file mode 100644
index 0000000..722bf95
--- /dev/null
+++ b/internal/executor/groq_routing_test.go
@@ -0,0 +1,98 @@
+package executor
+
+import (
+ "context"
+ "encoding/json"
+ "log/slog"
+ "net/http"
+ "net/http/httptest"
+ "os"
+ "sync/atomic"
+ "testing"
+
+ "github.com/thepeterstone/claudomator/internal/llm"
+ "github.com/thepeterstone/claudomator/internal/provider/openaicompat"
+)
+
+// TestPool_Submit_GroqAgentType_RoutesThroughOpenAICompatProvider is the
+// Part A wiring-level confirmation (verification item 5) that Groq — one of
+// the three providers that docs/api-keys-setup.md documented as "works once
+// configured" but was never actually registered by internal/cli/serve.go or
+// run.go before this phase — is a real, usable runner once wired up. Groq is
+// OpenAI-wire-compatible, so this reuses the existing Phase 1
+// internal/provider/openaicompat adapter (internal/llm.Client) exactly the
+// way internal/cli/cloudrunners.go's registerCloudRunners constructs it: no
+// new adapter package, purely config-driven construction pointed at Groq's
+// endpoint.
+func TestPool_Submit_GroqAgentType_RoutesThroughOpenAICompatProvider(t *testing.T) {
+ var calls int32
+ srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ atomic.AddInt32(&calls, 1)
+ if r.URL.Path != "/chat/completions" {
+ t.Errorf("unexpected path %q", r.URL.Path)
+ }
+ if got := r.Header.Get("Authorization"); got != "Bearer test-groq-key" {
+ t.Errorf("wrong Authorization header: %q", got)
+ }
+ w.Header().Set("Content-Type", "application/json")
+ _ = json.NewEncoder(w).Encode(map[string]any{
+ "choices": []map[string]any{
+ {
+ "message": map[string]any{"role": "assistant", "content": "## Summary\ndone"},
+ "finish_reason": "stop",
+ },
+ },
+ "model": "llama-3.3-70b-versatile",
+ "usage": map[string]any{"prompt_tokens": 10, "completion_tokens": 3},
+ })
+ }))
+ defer srv.Close()
+
+ // Mirrors registerCloudRunners' construction for groq/openrouter/openai:
+ // openaicompat.New(&llm.Client{...}) wrapped in a NativeRunner registered
+ // under the "groq" runner-map key, SandboxKind "docker" for the real
+ // wiring (irrelevant here since the task has no ProjectDir/sandbox use).
+ client := &llm.Client{
+ Endpoint: srv.URL,
+ Model: "llama-3.3-70b-versatile",
+ APIKey: "test-groq-key",
+ }
+ runners := map[string]Runner{
+ "groq": &NativeRunner{
+ Provider: openaicompat.New(client),
+ Logger: slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelError})),
+ LogDir: t.TempDir(),
+ },
+ }
+
+ store := testStore(t)
+ logger := slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelError}))
+ pool := NewPool(2, runners, store, logger)
+
+ tk := makeTask("groq-routing-1")
+ tk.Agent.Type = "groq"
+ tk.Agent.Model = "llama-3.3-70b-versatile"
+ tk.Agent.SkipPlanning = true
+ if err := store.CreateTask(tk); err != nil {
+ t.Fatalf("CreateTask: %v", err)
+ }
+
+ if err := pool.Submit(context.Background(), tk); err != nil {
+ t.Fatalf("Submit: %v", err)
+ }
+
+ result := <-pool.Results()
+ if result.Err != nil {
+ t.Fatalf("expected no error, got: %v", result.Err)
+ }
+ if got := atomic.LoadInt32(&calls); got != 1 {
+ t.Fatalf("expected exactly 1 call to the fake Groq server, got %d", got)
+ }
+ if result.Execution.Agent != "groq" {
+ t.Errorf("execution.Agent: want %q, got %q", "groq", result.Execution.Agent)
+ }
+ if result.Execution.TokensIn != 10 || result.Execution.TokensOut != 3 {
+ t.Errorf("tokens should come from the openaicompat provider's usage: got in=%d out=%d",
+ result.Execution.TokensIn, result.Execution.TokensOut)
+ }
+}
diff --git a/internal/executor/role_dispatch_test.go b/internal/executor/role_dispatch_test.go
new file mode 100644
index 0000000..5628bdb
--- /dev/null
+++ b/internal/executor/role_dispatch_test.go
@@ -0,0 +1,206 @@
+package executor
+
+import (
+ "context"
+ "encoding/json"
+ "log/slog"
+ "os"
+ "testing"
+ "time"
+
+ "github.com/thepeterstone/claudomator/internal/role"
+)
+
+// TestPool_Execute_RoleTypedTask_ResolvesTier0 is the Pool.execute()-level
+// verification (Phase 5 item 3) that a role-typed task with an empty
+// Agent.Type resolves tier 0 of the active role_configs ladder before
+// dispatch: it should run through the tier-0 provider/model, and the
+// resulting execution should be stamped with EscalationRung 0.
+func TestPool_Execute_RoleTypedTask_ResolvesTier0(t *testing.T) {
+ store := testStore(t)
+
+ rc := role.RoleConfig{
+ Role: "coder",
+ EscalationLadder: []role.Tier{
+ {
+ Candidates: []role.Rung{{Provider: "groq", Model: "llama-3.3-70b-versatile"}},
+ SelectionMode: "single",
+ MaxRetries: 1,
+ },
+ {
+ Candidates: []role.Rung{{Provider: "anthropic", Model: "claude-sonnet-5"}},
+ MaxRetries: 1,
+ },
+ },
+ }
+ configJSON, err := json.Marshal(rc)
+ if err != nil {
+ t.Fatalf("marshal role config: %v", err)
+ }
+ created, err := store.CreateRoleConfig("coder", string(configJSON), "human")
+ if err != nil {
+ t.Fatalf("CreateRoleConfig: %v", err)
+ }
+ if err := store.ActivateRoleConfigVersion("coder", created.Version); err != nil {
+ t.Fatalf("ActivateRoleConfigVersion: %v", err)
+ }
+
+ runners := map[string]Runner{
+ "groq": &mockRunner{},
+ }
+ logger := slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelError}))
+ pool := NewPool(2, runners, store, logger)
+
+ tk := makeTask("role-task-1")
+ tk.Agent.Type = ""
+ tk.Agent.Model = ""
+ tk.Agent.Role = "coder"
+ tk.Agent.SkipPlanning = true
+ if err := store.CreateTask(tk); err != nil {
+ t.Fatalf("CreateTask: %v", err)
+ }
+
+ if err := pool.Submit(context.Background(), tk); err != nil {
+ t.Fatalf("Submit: %v", err)
+ }
+
+ select {
+ case result := <-pool.Results():
+ if result.Err != nil {
+ t.Fatalf("expected no error, got: %v", result.Err)
+ }
+ if result.Execution.Agent != "groq" {
+ t.Errorf("execution.Agent: want %q, got %q", "groq", result.Execution.Agent)
+ }
+ if result.Execution.EscalationRung != 0 {
+ t.Errorf("execution.EscalationRung: want 0, got %d", result.Execution.EscalationRung)
+ }
+ case <-time.After(5 * time.Second):
+ t.Fatal("timed out waiting for result")
+ }
+
+ persisted, err := store.GetTask(tk.ID)
+ if err != nil {
+ t.Fatalf("GetTask: %v", err)
+ }
+ if persisted.Agent.Type != "groq" {
+ t.Errorf("persisted Agent.Type: want %q, got %q", "groq", persisted.Agent.Type)
+ }
+ if persisted.Agent.Model != "llama-3.3-70b-versatile" {
+ t.Errorf("persisted Agent.Model: want %q, got %q", "llama-3.3-70b-versatile", persisted.Agent.Model)
+ }
+}
+
+// TestPool_Execute_NonRoleTask_Unaffected proves that a task with
+// Agent.Role == "" (every existing YAML/chatbot task shape) takes none of
+// the role-resolution branches in execute() — it dispatches exactly as it
+// did before this phase, using its explicitly-set Agent.Type/Model verbatim.
+func TestPool_Execute_NonRoleTask_Unaffected(t *testing.T) {
+ store := testStore(t)
+ runners := map[string]Runner{
+ "claude": &mockRunner{},
+ }
+ logger := slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelError}))
+ pool := NewPool(2, runners, store, logger)
+
+ tk := makeTask("non-role-task-1")
+ tk.Agent.Type = "claude"
+ tk.Agent.Model = "sonnet"
+ // Agent.Role intentionally left "" (zero value).
+ if err := store.CreateTask(tk); err != nil {
+ t.Fatalf("CreateTask: %v", err)
+ }
+
+ if err := pool.Submit(context.Background(), tk); err != nil {
+ t.Fatalf("Submit: %v", err)
+ }
+
+ select {
+ case result := <-pool.Results():
+ if result.Err != nil {
+ t.Fatalf("expected no error, got: %v", result.Err)
+ }
+ if result.Execution.Agent != "claude" {
+ t.Errorf("execution.Agent: want %q, got %q", "claude", result.Execution.Agent)
+ }
+ if result.Execution.EscalationRung != 0 {
+ t.Errorf("non-role task execution.EscalationRung should default to 0, got %d", result.Execution.EscalationRung)
+ }
+ case <-time.After(5 * time.Second):
+ t.Fatal("timed out waiting for result")
+ }
+
+ persisted, err := store.GetTask(tk.ID)
+ if err != nil {
+ t.Fatalf("GetTask: %v", err)
+ }
+ if persisted.Agent.Type != "claude" || persisted.Agent.Model != "sonnet" {
+ t.Errorf("non-role task's Agent.Type/Model should be unchanged: got %q/%q", persisted.Agent.Type, persisted.Agent.Model)
+ }
+}
+
+// TestPool_SelectRung_RoundRobin_SkipsRateLimited proves that repeated
+// selectRung calls on a round_robin tier rotate across candidates rather
+// than always returning candidate 0, and skip a candidate whose provider is
+// currently rate-limited.
+func TestPool_SelectRung_RoundRobin_SkipsRateLimited(t *testing.T) {
+ p := &Pool{
+ rateLimited: make(map[string]time.Time),
+ roleTierIndex: make(map[string]int),
+ }
+ tier := role.Tier{
+ Candidates: []role.Rung{
+ {Provider: "local", Model: "m1"},
+ {Provider: "groq", Model: "m2"},
+ {Provider: "openrouter", Model: "m3"},
+ },
+ SelectionMode: "round_robin",
+ }
+
+ // With nothing rate-limited, repeated calls should rotate through all
+ // three candidates before repeating.
+ seen := map[string]int{}
+ for i := 0; i < 6; i++ {
+ r := p.selectRung("coder", 0, tier)
+ seen[r.Provider]++
+ }
+ for _, prov := range []string{"local", "groq", "openrouter"} {
+ if seen[prov] != 2 {
+ t.Errorf("expected provider %q to be picked exactly twice across 6 calls, got %d (seen=%v)", prov, seen[prov], seen)
+ }
+ }
+
+ // Now rate-limit "groq" and confirm it's skipped on subsequent picks.
+ p.mu.Lock()
+ p.rateLimited["groq"] = time.Now().Add(1 * time.Hour)
+ p.mu.Unlock()
+
+ for i := 0; i < 6; i++ {
+ r := p.selectRung("coder", 0, tier)
+ if r.Provider == "groq" {
+ t.Errorf("expected rate-limited provider %q to be skipped, got picked at iteration %d", "groq", i)
+ }
+ }
+}
+
+// TestPool_SelectRung_SingleMode_AlwaysFirstCandidate proves "single"
+// selection mode never rotates.
+func TestPool_SelectRung_SingleMode_AlwaysFirstCandidate(t *testing.T) {
+ p := &Pool{
+ rateLimited: make(map[string]time.Time),
+ roleTierIndex: make(map[string]int),
+ }
+ tier := role.Tier{
+ Candidates: []role.Rung{
+ {Provider: "local", Model: "m1"},
+ {Provider: "groq", Model: "m2"},
+ },
+ SelectionMode: "single",
+ }
+ for i := 0; i < 4; i++ {
+ r := p.selectRung("coder", 0, tier)
+ if r.Provider != "local" {
+ t.Errorf("iteration %d: expected always \"local\" in single mode, got %q", i, r.Provider)
+ }
+ }
+}
generated by cgit v1.2.3 (git 2.39.1) at 2026-07-16 12:47:12 +0000