diff options
Diffstat (limited to 'internal/executor/executor.go')
| -rw-r--r-- | internal/executor/executor.go | 150 |
1 files changed, 145 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 — |