package executor

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log/slog"
	"path/filepath"
	"strings"
	"sync"
	"time"

	"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"
)

// Store is the subset of storage.DB methods used by the Pool.
// Defining it as an interface allows test doubles to be injected.
type Store interface {
	GetTask(id string) (*task.Task, error)
	ListTasks(filter storage.TaskFilter) ([]*task.Task, error)
	ListSubtasks(parentID string) ([]*task.Task, error)
	ListExecutions(taskID string) ([]*storage.Execution, error)
	CreateExecution(e *storage.Execution) error
	CreateExecutionAndSetRunning(e *storage.Execution) error
	UpdateExecution(e *storage.Execution) error
	UpdateTaskState(id string, newState task.State) error
	UpdateTaskQuestion(taskID, questionJSON string) error
	UpdateTaskSummary(taskID, summary string) error
	AppendTaskInteraction(taskID string, interaction task.Interaction) error
	UpdateTaskAgent(id string, agent task.AgentConfig) error
	UpdateExecutionChangestats(execID string, stats *task.Changestats) error
	RecordAgentEvent(e storage.AgentEvent) error
	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)
	// ListDependents returns tasks that directly depend on taskID. Used by
	// cascadeFail to proactively cancel a failed task's downstream subtree.
	ListDependents(taskID string) ([]*task.Task, error)
}

// LogPather is an optional interface runners can implement to provide the log
// directory for an execution before it starts. The pool uses this to persist
// log paths at CreateExecution time rather than waiting until execution ends.
type LogPather interface {
	ExecLogDir(execID string) string
}

// Runner executes a single task and returns the result. The AgentChannel is how
// the runner reports agent-originated signals (summary, questions, subtasks,
// progress) back to the system.
type Runner interface {
	Run(ctx context.Context, t *task.Task, exec *storage.Execution, ch AgentChannel) error
}

// workItem is an entry in the pool's internal work queue.
type workItem struct {
	ctx  context.Context
	task *task.Task
	exec *storage.Execution // non-nil for resume submissions
}

// Pool manages a bounded set of concurrent task workers.
type Pool struct {
	maxConcurrent   int
	maxPerAgent     int
	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

	mu                  sync.Mutex
	active              int
	activePerAgent      map[string]int
	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
	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
}

// BudgetGate reports whether a task estimated at estCost may run on a provider
// without breaching its rolling spend window. Satisfied by *budget.Accountant.
type BudgetGate interface {
	Allow(provider string, estCost float64) (bool, error)
}

// Result is emitted when a task execution completes.
type Result struct {
	TaskID    string
	Execution *storage.Execution
	Err       error
}

func NewPool(maxConcurrent int, runners map[string]Runner, store Store, logger *slog.Logger) *Pool {
	if maxConcurrent < 1 {
		maxConcurrent = 1
	}
	p := &Pool{
		maxConcurrent:       maxConcurrent,
		maxPerAgent:         1,
		runners:             runners,
		store:               store,
		logger:              logger,
		depPollInterval:     5 * time.Second,
		requeueDelay:        30 * time.Second,
		activePerAgent:      make(map[string]int),
		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),
		doneCh:              make(chan struct{}, maxConcurrent),
		dispatchDone:        make(chan struct{}),
	}
	go p.dispatch()
	return p
}

// dispatch is a long-running goroutine that reads from the internal work queue
// and launches goroutines as soon as a pool slot is available. This prevents
// tasks from being rejected when the pool is temporarily at capacity.
func (p *Pool) dispatch() {
	defer close(p.dispatchDone)
	for item := range p.workCh {
		for {
			p.mu.Lock()
			if p.active < p.maxConcurrent {
				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)
				} else {
					p.workerWg.Add(1); go func(i workItem) { defer p.workerWg.Done(); p.execute(i.ctx, i.task) }(item)
				}
				break
			}
			p.mu.Unlock()
			<-p.doneCh // wait for a worker to finish
		}
	}
}

// Submit enqueues a task for execution. Returns an error only if the internal
// work queue is full. When the pool is at capacity the task is buffered and
// dispatched as soon as a slot becomes available.
func (p *Pool) Submit(ctx context.Context, t *task.Task) error {
	p.mu.Lock()
	if p.closed {
		p.mu.Unlock()
		return fmt.Errorf("executor pool is shut down")
	}
	// Send while holding the lock so that Shutdown cannot close workCh between
	// the closed-check above and the send below. The dispatch goroutine never
	// holds p.mu while receiving from workCh, so this cannot deadlock.
	select {
	case p.workCh <- workItem{ctx: ctx, task: t}:
		p.mu.Unlock()
		return nil
	default:
		p.mu.Unlock()
		return fmt.Errorf("executor work queue full (capacity %d)", cap(p.workCh))
	}
}

// Started returns a channel that emits task IDs when they transition to RUNNING.
func (p *Pool) Started() <-chan string {
	return p.startedCh
}

// Results returns the channel for reading execution results.
func (p *Pool) Results() <-chan *Result {
	return p.resultCh
}

// Shutdown stops accepting new work and waits for all in-flight workers to
// finish. Returns ctx.Err() if the context deadline is exceeded before all
// workers complete.
func (p *Pool) Shutdown(ctx context.Context) error {
	// Stop the dispatch goroutine. We must wait for it to exit before calling
	// workerWg.Wait() to avoid a race between dispatch's Add(1) and Wait().
	p.mu.Lock()
	p.closed = true
	p.mu.Unlock()
	close(p.workCh)
	select {
	case <-p.dispatchDone:
	case <-ctx.Done():
		return ctx.Err()
	}

	done := make(chan struct{})
	go func() {
		p.workerWg.Wait()
		close(done)
	}()

	select {
	case <-done:
		return nil
	case <-ctx.Done():
		return ctx.Err()
	}
}

// Cancel requests cancellation of a running task. Returns false if the task
// is not currently running in this pool.
func (p *Pool) Cancel(taskID string) bool {
	p.mu.Lock()
	cancel, ok := p.cancels[taskID]
	p.mu.Unlock()
	if !ok {
		return false
	}
	cancel()
	return true
}

// 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.StateBudgetExceeded: true,
}

// SubmitResume re-queues a blocked or interrupted task using the provided resume execution.
// The execution must have ResumeSessionID and ResumeAnswer set.
func (p *Pool) SubmitResume(ctx context.Context, t *task.Task, exec *storage.Execution) error {
	if !resumablePoolStates[t.State] {
		return fmt.Errorf("task %s must be in a resumable state to resume (current: %s)", t.ID, t.State)
	}
	if exec.ResumeSessionID == "" {
		return fmt.Errorf("resume execution for task %s must have a ResumeSessionID", t.ID)
	}
	select {
	case p.workCh <- workItem{ctx: ctx, task: t, exec: exec}:
		return nil
	default:
		return fmt.Errorf("executor work queue full (capacity %d)", cap(p.workCh))
	}
}

func (p *Pool) getRunner(t *task.Task) (Runner, error) {
	agentType := t.Agent.Type
	if agentType == "" {
		agentType = "claude" // Default for backward compatibility
	}
	runner, ok := p.runners[agentType]
	if !ok {
		return nil, fmt.Errorf("unsupported agent type: %q", agentType)
	}
	return runner, nil
}

// releaseDispatchSlot decrements the pool-level active counter and unblocks
// the dispatch goroutine. It must be deferred at the top of execute() and
// executeResume() so that every exit path — including early returns before
// the per-agent slot is taken — releases the slot acquired by dispatch().
func (p *Pool) releaseDispatchSlot() {
	p.mu.Lock()
	p.active--
	p.mu.Unlock()
	select {
	case p.doneCh <- struct{}{}:
	default:
	}
}

// decActiveAgent decrements the per-agent active counter for a finished task.
// Safe to call multiple times — subsequent calls are no-ops via the cleaned
// flag. Always call this before sending on resultCh so consumers observing a
// result see the accounting already settled (no zero-count map entries
// lingering). The pool-level p.active slot is handled separately by
// releaseDispatchSlot, which is deferred at the top of execute/executeResume.
func (p *Pool) decActiveAgent(agentType string, cleaned *bool) {
	if *cleaned {
		return
	}
	*cleaned = true
	p.mu.Lock()
	p.activePerAgent[agentType]--
	if p.activePerAgent[agentType] == 0 {
		delete(p.activePerAgent, agentType)
	}
	p.mu.Unlock()
}

// withWorkerSlot manages the lifecycle of a worker: per-agent
// slots, context timeout/cancellation, and cancellation registration. It
// provides a finish callback to settle per-agent accounting early (e.g. before
// reporting results).
func (p *Pool) withWorkerSlot(ctx context.Context, t *task.Task, agentType string, enforceCapacity bool, fn func(ctx context.Context, finish func()) error) {
	p.mu.Lock()
	if enforceCapacity && p.activePerAgent[agentType] >= p.maxPerAgent {
		p.mu.Unlock()
		time.AfterFunc(p.requeueDelay, func() { p.workCh <- workItem{ctx: ctx, task: t} })
		return
	}
	if deadline, ok := p.rateLimited[agentType]; ok && time.Now().After(deadline) {
		delete(p.rateLimited, agentType)
		agentName := agentType
		go func() {
			ev := storage.AgentEvent{
				ID:        uuid.New().String(),
				Agent:     agentName,
				Event:     "available",
				Timestamp: time.Now(),
			}
			if recErr := p.store.RecordAgentEvent(ev); recErr != nil {
				p.logger.Warn("failed to record agent available event", "error", recErr)
			}
		}()
	}
	p.activePerAgent[agentType]++
	p.mu.Unlock()

	var cleaned bool
	finish := func() {
		p.decActiveAgent(agentType, &cleaned)
	}
	defer finish()

	var cancel context.CancelFunc
	if t.Timeout.Duration > 0 {
		ctx, cancel = context.WithTimeout(ctx, t.Timeout.Duration)
	} else {
		ctx, cancel = context.WithCancel(ctx)
	}
	p.mu.Lock()
	p.cancels[t.ID] = cancel
	p.mu.Unlock()
	defer func() {
		cancel()
		p.mu.Lock()
		delete(p.cancels, t.ID)
		p.mu.Unlock()
	}()

	_ = fn(ctx, finish)
}

func (p *Pool) executeResume(ctx context.Context, t *task.Task, exec *storage.Execution) {
	agentType := t.Agent.Type
	if agentType == "" {
		agentType = "claude"
	}

	p.withWorkerSlot(ctx, t, agentType, false, func(ctx context.Context, finish func()) error {
		runner, err := p.getRunner(t)
		if err != nil {
			p.logger.Error("failed to get runner for resume", "error", err, "taskID", t.ID)
			finish()
			p.resultCh <- &Result{TaskID: t.ID, Execution: exec, Err: err}
			return nil
		}

		// Pre-populate log paths.
		if lp, ok := runner.(LogPather); ok {
			if logDir := lp.ExecLogDir(exec.ID); logDir != "" {
				exec.StdoutPath = filepath.Join(logDir, "stdout.log")
				exec.StderrPath = filepath.Join(logDir, "stderr.log")
				exec.ArtifactDir = logDir
			}
		}
		exec.StartTime = time.Now().UTC()
		exec.Status = "RUNNING"

		if err := p.store.CreateExecutionAndSetRunning(exec); err != nil {
			p.logger.Error("failed to create resume execution record", "error", err)
		}
		select {
		case p.startedCh <- t.ID:
		default:
		}

		// Populate RepositoryURL from Project registry if missing (ADR-007).
		if t.RepositoryURL == "" && t.Project != "" {
			if proj, err := p.store.GetProject(t.Project); err == nil && proj.RemoteURL != "" {
				t.RepositoryURL = proj.RemoteURL
			}
		}

		sc := newStoreChannel(p.store, t.ID)
		err = runner.Run(ctx, t, exec, sc)
		exec.EndTime = time.Now().UTC()
		if sum, ok := sc.ReportedSummary(); ok {
			exec.Summary = sum
		}

		finish()
		p.handleRunResult(ctx, t, exec, err, agentType)
		return nil
	})

}

// handleRunResult applies the shared post-run error-classification and
// state-update logic used by both execute() and executeResume(). It sets
// exec.Status and exec.ErrorMsg, updates storage, and emits the result to
// resultCh. The caller must set exec.EndTime before calling.
func (p *Pool) handleRunResult(ctx context.Context, t *task.Task, exec *storage.Execution, err error, agentType string) {
	if err != nil {
		if retry.IsRateLimitError(err) || isQuotaExhausted(err) {
			p.mu.Lock()
			retryAfter := retry.ParseRetryAfter(err.Error())
			reason := "transient"
			if isQuotaExhausted(err) {
				reason = "quota"
				if retryAfter == 0 {
					retryAfter = 5 * time.Hour
				}
			} else if retryAfter == 0 {
				retryAfter = 1 * time.Minute
			}
			until := time.Now().Add(retryAfter)
			p.rateLimited[agentType] = until
			p.logger.Info("agent rate limited", "agent", agentType, "retryAfter", retryAfter, "quotaExhausted", isQuotaExhausted(err))
			p.mu.Unlock()
			go func() {
				ev := storage.AgentEvent{
					ID:        uuid.New().String(),
					Agent:     agentType,
					Event:     "rate_limited",
					Timestamp: time.Now(),
					Until:     &until,
					Reason:    reason,
				}
				if recErr := p.store.RecordAgentEvent(ev); recErr != nil {
					p.logger.Warn("failed to record agent event", "error", recErr)
				}
			}()
		}

		var blockedErr *BlockedError
		if errors.As(err, &blockedErr) {
			exec.Status = "BLOCKED"
			exec.SandboxDir = blockedErr.SandboxDir // preserve so resume runs in same dir
			if err := p.store.UpdateTaskState(t.ID, task.StateBlocked); err != nil {
				p.logger.Error("failed to update task state", "taskID", t.ID, "state", task.StateBlocked, "error", err)
			}
			if err := p.store.UpdateTaskQuestion(t.ID, blockedErr.QuestionJSON); err != nil {
				p.logger.Error("failed to update task question", "taskID", t.ID, "error", err)
			}
		} else if ctx.Err() == context.DeadlineExceeded {
			exec.Status = "TIMED_OUT"
			exec.ErrorMsg = "execution timed out"
			if err := p.store.UpdateTaskState(t.ID, task.StateTimedOut); err != nil {
				p.logger.Error("failed to update task state", "taskID", t.ID, "state", task.StateTimedOut, "error", err)
			}
		} else if ctx.Err() == context.Canceled {
			exec.Status = "CANCELLED"
			exec.ErrorMsg = "execution cancelled"
			if err := p.store.UpdateTaskState(t.ID, task.StateCancelled); err != nil {
				p.logger.Error("failed to update task state", "taskID", t.ID, "state", task.StateCancelled, "error", err)
			}
		} else if isQuotaExhausted(err) {
			exec.Status = "BUDGET_EXCEEDED"
			exec.ErrorMsg = err.Error()
			if err := p.store.UpdateTaskState(t.ID, task.StateBudgetExceeded); err != nil {
				p.logger.Error("failed to update task state", "taskID", t.ID, "state", task.StateBudgetExceeded, "error", err)
			}
		} else {
			exec.Status = "FAILED"
			exec.ErrorMsg = err.Error()
			if err := p.store.UpdateTaskState(t.ID, task.StateFailed); err != nil {
				p.logger.Error("failed to update task state", "taskID", t.ID, "state", task.StateFailed, "error", err)
			}
			p.mu.Lock()
			p.consecutiveFailures[agentType]++
			p.mu.Unlock()
		}
	} else {
		p.mu.Lock()
		p.consecutiveFailures[agentType] = 0
		p.mu.Unlock()
		if t.ParentTaskID == "" {
			subtasks, subErr := p.store.ListSubtasks(t.ID)
			if subErr != nil {
				p.logger.Error("failed to list subtasks", "taskID", t.ID, "error", subErr)
			}
			if subErr == nil && len(subtasks) > 0 {
				exec.Status = "BLOCKED"
				if err := p.store.UpdateTaskState(t.ID, task.StateBlocked); err != nil {
					p.logger.Error("failed to update task state", "taskID", t.ID, "state", task.StateBlocked, "error", err)
				}
			} else {
				exec.Status = "READY"
				if err := p.store.UpdateTaskState(t.ID, task.StateReady); err != nil {
					p.logger.Error("failed to update task state", "taskID", t.ID, "state", task.StateReady, "error", err)
				}
			}
		} else {
			exec.Status = "COMPLETED"
			if err := p.store.UpdateTaskState(t.ID, task.StateCompleted); err != nil {
				p.logger.Error("failed to update task state", "taskID", t.ID, "state", task.StateCompleted, "error", err)
			}
			p.maybeUnblockParent(t.ParentTaskID)
		}
	}

	// Proactively cancel the downstream dependency subtree the moment this
	// task lands in a terminal failure state, so tasks depending on it (e.g.
	// evaluator subtasks fanned out from a Builder) don't sit around waiting
	// on a dependency that will never complete. BLOCKED/READY/COMPLETED are
	// intentionally excluded — those aren't terminal failures.
	switch exec.Status {
	case "FAILED", "TIMED_OUT", "CANCELLED", "BUDGET_EXCEEDED":
		p.cascadeFail(t.ID, exec.ErrorMsg)
	}

	summary := exec.Summary
	if summary == "" && exec.StdoutPath != "" {
		summary = extractSummary(exec.StdoutPath)
	}
	if summary == "" && p.LLM != nil && exec.StdoutPath != "" {
		summary = synthesizeSummary(ctx, p.LLM, exec.StdoutPath)
	}
	if summary != "" {
		if summaryErr := p.store.UpdateTaskSummary(t.ID, summary); summaryErr != nil {
			p.logger.Error("failed to update task summary", "taskID", t.ID, "error", summaryErr)
		}
	}
	if exec.StdoutPath != "" {
		if cs := task.ParseChangestatFromFile(exec.StdoutPath); cs != nil {
			exec.Changestats = cs
			if csErr := p.store.UpdateExecutionChangestats(exec.ID, cs); csErr != nil {
				p.logger.Error("failed to store changestats", "execID", exec.ID, "error", csErr)
			}
		}
	}
	if updateErr := p.store.UpdateExecution(exec); updateErr != nil {
		p.logger.Error("failed to update execution", "error", updateErr)
	}
	p.resultCh <- &Result{TaskID: t.ID, Execution: exec, Err: err}
}

// cascadeCancellableStates are the task states a dependent task can safely be
// preempted from: it hasn't started running yet. RUNNING, terminal states,
// and BLOCKED (subtask delegation) are all left untouched — cascadeFail only
// preempts work that hasn't started.
var cascadeCancellableStates = map[task.State]bool{
	task.StatePending: true,
	task.StateQueued:  true,
}

// cascadeFail proactively cancels the downstream dependency subtree rooted at
// taskID. It is called right after a task is marked into a terminal failure
// state (FAILED/TIMED_OUT/CANCELLED/BUDGET_EXCEEDED): every direct dependent
// still waiting to run (PENDING or QUEUED) is recorded as CANCELLED with a
// message referencing the upstream failure, and the same cancellation is
// applied recursively to that dependent's own dependents — so a multi-level
// chain (A fails, B depends on A, C depends on B) cascades all the way down,
// not just one hop. Dependents that are already RUNNING or in a terminal
// state are left alone.
//
// This exists so a fan-out pattern (e.g. a Builder task with several
// Evaluator subtasks depending on it) doesn't leave those subtasks sitting
// around forever — or worse, waiting on a dependency that will never
// complete — when the upstream task fails. Without this, a dependent only
// discovers its dependency failed the next time something tries to dispatch
// it (see checkDepsReady), which may be never for a task nobody has
// submitted to the pool yet.
func (p *Pool) cascadeFail(taskID string, reason string) {
	p.cascadeFailFrom(taskID, reason, make(map[string]bool))
}

// cascadeFailFrom does the actual work for cascadeFail, threading a visited
// set through the recursion so a pathological dependency cycle can't cause
// infinite recursion.
func (p *Pool) cascadeFailFrom(taskID string, reason string, visited map[string]bool) {
	if visited[taskID] {
		return
	}
	visited[taskID] = true

	dependents, err := p.store.ListDependents(taskID)
	if err != nil {
		p.logger.Error("cascadeFail: list dependents", "taskID", taskID, "error", err)
		return
	}
	for _, dep := range dependents {
		if !cascadeCancellableStates[dep.State] {
			continue // already running, already terminal, or blocked — leave it alone
		}
		msg := fmt.Sprintf("upstream dependency %s failed: %s", taskID, reason)
		now := time.Now().UTC()
		exec := &storage.Execution{
			ID:        uuid.New().String(),
			TaskID:    dep.ID,
			StartTime: now,
			EndTime:   now,
			Status:    "CANCELLED",
			ErrorMsg:  msg,
		}
		if createErr := p.store.CreateExecution(exec); createErr != nil {
			p.logger.Error("cascadeFail: create execution", "taskID", dep.ID, "error", createErr)
		}
		if stateErr := p.store.UpdateTaskState(dep.ID, task.StateCancelled); stateErr != nil {
			p.logger.Error("cascadeFail: update task state", "taskID", dep.ID, "error", stateErr)
			continue // couldn't actually cancel it — don't cascade further from here
		}
		p.logger.Info("cascade-cancelled dependent task", "taskID", dep.ID, "upstream", taskID)
		p.cascadeFailFrom(dep.ID, msg, visited)
	}
}

// ActiveCount returns the number of currently running tasks.
func (p *Pool) ActiveCount() int {
	p.mu.Lock()
	defer p.mu.Unlock()
	return p.active
}

// AgentStatusInfo holds the current state of a single agent.
type AgentStatusInfo struct {
	Agent       string     `json:"agent"`
	ActiveTasks int        `json:"active_tasks"`
	RateLimited bool       `json:"rate_limited"`
	Until       *time.Time `json:"until,omitempty"`
}

// AgentStatuses returns the current status of all registered agents.
func (p *Pool) AgentStatuses() []AgentStatusInfo {
	p.mu.Lock()
	defer p.mu.Unlock()
	now := time.Now()
	var out []AgentStatusInfo
	for agent := range p.runners {
		info := AgentStatusInfo{
			Agent:       agent,
			ActiveTasks: p.activePerAgent[agent],
		}
		if deadline, ok := p.rateLimited[agent]; ok && now.Before(deadline) {
			info.RateLimited = true
			info.Until = &deadline
		}
		out = append(out, info)
	}
	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.
func pickAgent(status SystemStatus) string {
	best := ""
	bestActive := -1

	// First pass: only consider non-rate-limited agents.
	for agent, active := range status.ActiveTasks {
		if status.RateLimited[agent] {
			continue
		}
		if bestActive == -1 || active < bestActive || (active == bestActive && agent < best) {
			best = agent
			bestActive = active
		}
	}
	if best != "" {
		return best
	}

	// Fallback: all rate-limited — pick least active anyway.
	for agent, active := range status.ActiveTasks {
		if bestActive == -1 || active < bestActive || (active == bestActive && agent < best) {
			best = agent
			bestActive = active
		}
	}
	return best
}

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)
	rateLimited := make(map[string]bool)
	now := time.Now()
	for agent := range p.runners {
		activeTasks[agent] = p.activePerAgent[agent]
		if deadline, ok := p.rateLimited[agent]; ok && now.After(deadline) {
			delete(p.rateLimited, agent)
			agentName := agent
			go func() {
				ev := storage.AgentEvent{
					ID:        uuid.New().String(),
					Agent:     agentName,
					Event:     "available",
					Timestamp: time.Now(),
				}
				if recErr := p.store.RecordAgentEvent(ev); recErr != nil {
					p.logger.Warn("failed to record agent available event", "error", recErr)
				}
			}()
		}
		rateLimited[agent] = now.Before(p.rateLimited[agent])
	}
	status := SystemStatus{
		ActiveTasks: activeTasks,
		RateLimited: rateLimited,
	}
	p.mu.Unlock()

	// If a specific agent is already requested AND we have a runner registered
	// for it, skip selection and classification. Unknown/empty types fall
	// through to the load balancer.
	_, runnerKnown := p.runners[t.Agent.Type]
	skipClassification := t.Agent.Type != "" && runnerKnown

	if !skipClassification {
		// Deterministically pick the agent with fewest active tasks.
		selectedAgent := pickAgent(status)
		if selectedAgent != "" {
			t.Agent.Type = selectedAgent
		}

		if p.Classifier != nil {
			cls, err := p.Classifier.Classify(ctx, t.Name, t.Agent.Instructions, status, t.Agent.Type)
			if err == nil {
				p.logger.Info("task classified", "taskID", t.ID, "agent", t.Agent.Type, "model", cls.Model, "reason", cls.Reason)
				t.Agent.Model = cls.Model
			} else {
				p.logger.Error("classification failed", "error", err, "taskID", t.ID)
			}
		}
	}

	// Persist the assigned agent (and model) to the database before running.
	if err := p.store.UpdateTaskAgent(t.ID, t.Agent); err != nil {
		p.logger.Error("failed to persist agent config", "error", err, "taskID", t.ID)
	}

	agentType := t.Agent.Type
	if agentType == "" {
		agentType = "claude"
	}

	// Budget gating: if running on this provider could breach its rolling window
	// (estimated by the task's max_budget_usd), reroute to the free local runner
	// when one is registered, otherwise stop and mark the task BUDGET_EXCEEDED.
	if p.Budget != nil {
		if ok, berr := p.Budget.Allow(agentType, t.Agent.MaxBudgetUSD); berr != nil {
			p.logger.Warn("budget check failed; proceeding", "error", berr, "taskID", t.ID)
		} else if !ok {
			if _, hasLocal := p.runners["local"]; hasLocal && agentType != "local" {
				p.logger.Info("budget window would be breached; rerouting to local", "taskID", t.ID, "from", agentType)
				t.Agent.Type = "local"
				t.Agent.Model = ""
				agentType = "local"
				if uerr := p.store.UpdateTaskAgent(t.ID, t.Agent); uerr != nil {
					p.logger.Error("failed to persist rerouted agent", "error", uerr, "taskID", t.ID)
				}
			} else {
				now := time.Now().UTC()
				exec := &storage.Execution{
					ID:        uuid.New().String(),
					TaskID:    t.ID,
					StartTime: now,
					EndTime:   now,
					Status:    "BUDGET_EXCEEDED",
					Agent:     agentType,
					ErrorMsg:  fmt.Sprintf("rolling budget window cap reached for provider %q", agentType),
				}
				if createErr := p.store.CreateExecution(exec); createErr != nil {
					p.logger.Error("failed to create execution record", "error", createErr)
				}
				if err := p.store.UpdateTaskState(t.ID, task.StateBudgetExceeded); err != nil {
					p.logger.Error("failed to update task state", "taskID", t.ID, "state", task.StateBudgetExceeded, "error", err)
				}
				p.cascadeFail(t.ID, exec.ErrorMsg)
				p.resultCh <- &Result{TaskID: t.ID, Execution: exec, Err: fmt.Errorf("budget cap reached for %s", agentType)}
				return
			}
		}
	}

	// Check dependencies before taking the per-agent slot to avoid deadlock:
	// if a dependent task holds the slot while waiting for its dependency to run,
	// the dependency can never start (maxPerAgent=1).
	if len(t.DependsOn) > 0 {
		// cascadeFail (triggered when some ancestor of t failed) may already
		// have cancelled t out from under it while it sat in the AfterFunc
		// requeue loop below — re-check the fresh state so we don't try (and
		// fail, since CANCELLED has no valid self-transition) to re-apply a
		// terminal transition, and — more importantly — so a cascade-cancelled
		// task never actually gets dispatched to a runner.
		if fresh, ferr := p.store.GetTask(t.ID); ferr == nil && fresh.State != task.StateQueued {
			p.logger.Info("skipping dispatch: task no longer queued (likely cascade-cancelled)", "taskID", t.ID, "state", fresh.State)
			return
		}
		ready, depErr := p.checkDepsReady(t)
		if depErr != nil {
			// A dependency hit a terminal failure — cancel this task immediately.
			now := time.Now().UTC()
			exec := &storage.Execution{
				ID:        uuid.New().String(),
				TaskID:    t.ID,
				StartTime: now,
				EndTime:   now,
				Status:    "CANCELLED",
				ErrorMsg:  depErr.Error(),
			}
			if createErr := p.store.CreateExecution(exec); createErr != nil {
				p.logger.Error("failed to create execution record", "error", createErr)
			}
			if err := p.store.UpdateTaskState(t.ID, task.StateCancelled); err != nil {
				p.logger.Error("failed to update task state", "taskID", t.ID, "state", task.StateCancelled, "error", err)
			}
			p.cascadeFail(t.ID, depErr.Error())
			p.resultCh <- &Result{TaskID: t.ID, Execution: exec, Err: depErr}
			return
		}
		if !ready {
			// Dependencies not yet done — requeue without holding the slot.
			time.AfterFunc(p.requeueDelay, func() { p.workCh <- workItem{ctx: ctx, task: t} })
			return
		}
	}
	p.withWorkerSlot(ctx, t, agentType, true, func(ctx context.Context, finish func()) error {
		runner, err := p.getRunner(t)
		if err != nil {
			p.logger.Error("failed to get runner", "error", err, "taskID", t.ID)
			now := time.Now().UTC()
			exec := &storage.Execution{
				ID:        uuid.New().String(),
				TaskID:    t.ID,
				StartTime: now,
				EndTime:   now,
				Status:    "FAILED",
				ErrorMsg:  err.Error(),
			}
			if createErr := p.store.CreateExecution(exec); createErr != nil {
				p.logger.Error("failed to create execution record", "error", createErr)
			}
			if err := p.store.UpdateTaskState(t.ID, task.StateFailed); err != nil {
				p.logger.Error("failed to update task state", "taskID", t.ID, "state", task.StateFailed, "error", err)
			}
			finish()
			p.resultCh <- &Result{TaskID: t.ID, Execution: exec, Err: err}
			return nil
		}

		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,
			EscalationRung: escalationRung,
		}

		// Pre-populate log paths so they're available in the DB immediately —
		// before the subprocess starts — enabling live tailing and debugging.
		if lp, ok := runner.(LogPather); ok {
			if logDir := lp.ExecLogDir(execID); logDir != "" {
				exec.StdoutPath = filepath.Join(logDir, "stdout.log")
				exec.StderrPath = filepath.Join(logDir, "stderr.log")
				exec.ArtifactDir = logDir
			}
		}

		// Record execution start atomically with the RUNNING state transition.
		if err := p.store.CreateExecutionAndSetRunning(exec); err != nil {
			p.logger.Error("failed to create execution record", "error", err)
		}
		select {
		case p.startedCh <- t.ID:
		default:
		}

		// Inject prior failure history so the agent knows what went wrong before.
		priorExecs, priorErr := p.store.ListExecutions(t.ID)
		t = withFailureHistory(t, priorExecs, priorErr)

		// Populate RepositoryURL from Project registry if missing (ADR-007).
		if t.RepositoryURL == "" && t.Project != "" {
			if proj, err := p.store.GetProject(t.Project); err == nil && proj.RemoteURL != "" {
				t.RepositoryURL = proj.RemoteURL
			}
		}

		// Run the task.
		sc := newStoreChannel(p.store, t.ID)
		err = runner.Run(ctx, t, exec, sc)
		exec.EndTime = time.Now().UTC()
		if sum, ok := sc.ReportedSummary(); ok {
			exec.Summary = sum
		}

		finish()
		p.handleRunResult(ctx, t, exec, err, agentType)
		return nil
	})
}

// RecoverStaleRunning marks any tasks stuck in RUNNING state (from a previous
// server crash or restart) as FAILED, then immediately re-queues them for
// retry. It also closes any open RUNNING execution records for those tasks.
// Call this once on server startup.
func (p *Pool) RecoverStaleRunning(ctx context.Context) {
	tasks, err := p.store.ListTasks(storage.TaskFilter{State: task.StateRunning})
	if err != nil {
		p.logger.Error("RecoverStaleRunning: list tasks", "error", err)
		return
	}
	for _, t := range tasks {
		p.logger.Warn("recovering stale RUNNING task", "taskID", t.ID, "name", t.Name)
		// Close any open execution records.
		execs, err := p.store.ListExecutions(t.ID)
		if err == nil {
			for _, e := range execs {
				if e.Status == "RUNNING" {
					e.Status = "FAILED"
					e.ErrorMsg = "server restarted while task was running"
					e.EndTime = time.Now().UTC()
					if updateErr := p.store.UpdateExecution(e); updateErr != nil {
						p.logger.Error("RecoverStaleRunning: update execution", "error", updateErr, "execID", e.ID)
					}
				}
			}
		}
		if err := p.store.UpdateTaskState(t.ID, task.StateFailed); err != nil {
			p.logger.Error("RecoverStaleRunning: update task state", "error", err, "taskID", t.ID)
			continue
		}
		// Re-queue so the task retries automatically. Submit expects QUEUED state.
		if err := p.store.UpdateTaskState(t.ID, task.StateQueued); err != nil {
			p.logger.Error("RecoverStaleRunning: set queued", "error", err, "taskID", t.ID)
			continue
		}
		t.State = task.StateQueued
		if err := p.Submit(ctx, t); err != nil {
			p.logger.Error("RecoverStaleRunning: re-queue", "error", err, "taskID", t.ID)
		}
	}
}

// RecoverStaleQueued re-submits any tasks that are stuck in QUEUED state from
// a previous server instance. Call this once on server startup, after
// RecoverStaleRunning.
func (p *Pool) RecoverStaleQueued(ctx context.Context) {
	tasks, err := p.store.ListTasks(storage.TaskFilter{State: task.StateQueued})
	if err != nil {
		p.logger.Error("RecoverStaleQueued: list tasks", "error", err)
		return
	}
	for _, t := range tasks {
		p.logger.Info("resubmitting stale QUEUED task", "taskID", t.ID, "name", t.Name)
		if err := p.Submit(ctx, t); err != nil {
			p.logger.Error("RecoverStaleQueued: submit", "error", err, "taskID", t.ID)
		}
	}
}

// RecoverStaleBlocked promotes any BLOCKED or QUEUED parent task to READY when
// all of its subtasks are already COMPLETED. This handles the case where the
// server was restarted after subtasks finished but before maybeUnblockParent
// could fire.
// Call this once on server startup, after RecoverStaleRunning and RecoverStaleQueued.
func (p *Pool) RecoverStaleBlocked() {
	for _, state := range []task.State{task.StateBlocked, task.StateQueued} {
		tasks, err := p.store.ListTasks(storage.TaskFilter{State: state})
		if err != nil {
			p.logger.Error("RecoverStaleBlocked: list tasks", "error", err, "state", state)
			continue
		}
		for _, t := range tasks {
			if t.ParentTaskID != "" {
				continue // only promote actual parents
			}
			p.maybeUnblockParent(t.ID)
		}
	}
}

// terminalFailureStates are dependency states that cause the waiting task to fail immediately.
var terminalFailureStates = map[task.State]bool{
	task.StateFailed:         true,
	task.StateTimedOut:       true,
	task.StateCancelled:      true,
	task.StateBudgetExceeded: true,
}

// depDoneStates are task states that satisfy a DependsOn dependency.
var depDoneStates = map[task.State]bool{
	task.StateCompleted: true,
	task.StateReady:     true, // leaf tasks finish at READY
}

// checkDepsReady does a single synchronous check of t.DependsOn.
// Returns (true, nil) if all deps are done, (false, nil) if any are still pending,
// or (false, err) if a dep entered a terminal failure state.
func (p *Pool) checkDepsReady(t *task.Task) (bool, error) {
	for _, depID := range t.DependsOn {
		dep, err := p.store.GetTask(depID)
		if err != nil {
			return false, fmt.Errorf("dependency %q not found: %w", depID, err)
		}
		if depDoneStates[dep.State] {
			continue
		}
		if terminalFailureStates[dep.State] {
			return false, fmt.Errorf("dependency %q ended in state %s", depID, dep.State)
		}
		return false, nil // still pending
	}
	return true, nil
}

// withFailureHistory returns a shallow copy of t with prior failed execution
// error messages prepended to SystemPromptAppend so the agent knows what went
// wrong in previous attempts.
func withFailureHistory(t *task.Task, execs []*storage.Execution, err error) *task.Task {
	if err != nil || len(execs) == 0 {
		return t
	}

	var failures []storage.Execution
	for _, e := range execs {
		if (e.Status == "FAILED" || e.Status == "TIMED_OUT") && e.ErrorMsg != "" {
			failures = append(failures, *e)
		}
	}
	if len(failures) == 0 {
		return t
	}

	var sb strings.Builder
	sb.WriteString("## Prior Attempt History\n\n")
	sb.WriteString("This task has failed before. Do not repeat the same mistakes.\n\n")
	for i, f := range failures {
		fmt.Fprintf(&sb, "**Attempt %d** (%s) — %s:\n%s\n\n",
			i+1, f.StartTime.Format("2006-01-02 15:04 UTC"), f.Status, f.ErrorMsg)
	}
	sb.WriteString("---\n\n")

	copy := *t
	copy.Agent = t.Agent
	if copy.Agent.SystemPromptAppend != "" {
		copy.Agent.SystemPromptAppend = sb.String() + copy.Agent.SystemPromptAppend
	} else {
		copy.Agent.SystemPromptAppend = sb.String()
	}
	return &copy
}

// maybeUnblockParent transitions the parent task to READY if all of its subtasks
// are in the COMPLETED state. Handles both BLOCKED parents (ran, created subtasks,
// paused) and QUEUED parents (subtasks created before parent ran).
func (p *Pool) maybeUnblockParent(parentID string) {
	parent, err := p.store.GetTask(parentID)
	if err != nil {
		p.logger.Error("maybeUnblockParent: get parent", "parentID", parentID, "error", err)
		return
	}
	if parent.State != task.StateBlocked && parent.State != task.StateQueued {
		return
	}
	subtasks, err := p.store.ListSubtasks(parentID)
	if err != nil {
		p.logger.Error("maybeUnblockParent: list subtasks", "parentID", parentID, "error", err)
		return
	}
	// A task with no subtasks was never blocked by subtask delegation — don't promote it.
	// This prevents incorrectly promoting leaf tasks that are stuck in QUEUED to READY.
	if len(subtasks) == 0 {
		return
	}
	for _, sub := range subtasks {
		if sub.State != task.StateCompleted {
			return
		}
	}
	if err := p.store.UpdateTaskState(parentID, task.StateReady); err != nil {
		p.logger.Error("maybeUnblockParent: update parent state", "parentID", parentID, "error", err)
	}
}

// waitForDependencies polls storage until all tasks in t.DependsOn reach COMPLETED,
// or until a dependency enters a terminal failure state or the context is cancelled.
func (p *Pool) waitForDependencies(ctx context.Context, t *task.Task) error {
	for {
		allDone := true
		for _, depID := range t.DependsOn {
			dep, err := p.store.GetTask(depID)
			if err != nil {
				return fmt.Errorf("dependency %q not found: %w", depID, err)
			}
			if depDoneStates[dep.State] {
				continue
			}
			if terminalFailureStates[dep.State] {
				return fmt.Errorf("dependency %q ended in state %s", depID, dep.State)
			}
			allDone = false
		}
		if allDone {
			return nil
		}
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-time.After(p.depPollInterval):
		}
	}
}