path: root/internal/scheduler/scheduler.go

// Package scheduler implements a first-pass retry-then-escalate loop for
// role-typed tasks (task.AgentConfig.Role != ""). On a poll interval it looks
// for tasks whose most recent execution ended FAILED, and either resubmits
// them at the same escalation-ladder tier (if under that tier's MaxRetries)
// or escalates them to the next tier (if the budget allows) — recording an
// event.KindEscalated event either way. If the ladder is exhausted or the
// budget denies the escalation, the task is left FAILED for human attention.
//
// Explicit non-goals for this phase (see the Phase 5 task description):
// no AskUser-timeout escalation, no DAG/cascade-fail logic. Handling for
// TIMED_OUT/CANCELLED/BUDGET_EXCEEDED tasks follows the same shape as FAILED
// but isn't implemented yet — only FAILED is polled.
package scheduler

import (
	"context"
	"encoding/json"
	"log/slog"
	"sync"
	"time"

	"github.com/thepeterstone/claudomator/internal/event"
	"github.com/thepeterstone/claudomator/internal/role"
	"github.com/thepeterstone/claudomator/internal/storage"
	"github.com/thepeterstone/claudomator/internal/task"
)

// Store is the subset of storage.DB methods the Scheduler needs.
type Store interface {
	ListTasks(filter storage.TaskFilter) ([]*task.Task, error)
	ListExecutions(taskID string) ([]*storage.Execution, error)
	GetActiveRoleConfig(role string) (*storage.RoleConfigRow, error)
	UpdateTaskAgent(id string, agent task.AgentConfig) error
	UpdateTaskState(id string, newState task.State) error
	CreateEvent(e *event.Event) error
}

// Pool is the subset of *executor.Pool the Scheduler needs. Satisfied by
// *executor.Pool directly (see internal/cli/serve.go); declared as an
// interface here purely so tests can supply a fake without dragging in the
// executor package's runner/sandbox machinery.
type Pool interface {
	Submit(ctx context.Context, t *task.Task) error
}

// BudgetGate reports whether an escalation to provider estimated at estCost
// is allowed. Satisfied by *budget.Accountant.
type BudgetGate interface {
	Allow(provider string, estCost float64) (bool, error)
}

// Scheduler polls for role-typed FAILED tasks and retries or escalates them
// per their active role_configs escalation ladder.
type Scheduler struct {
	Store  Store
	Pool   Pool
	Budget BudgetGate // nil means "no budget gating" (always allow)
	Logger *slog.Logger

	// handled dedupes processing within a single running process: once a
	// decision (retry/escalate/decline) has been made for a given
	// execution ID, it is never reconsidered again by this Scheduler
	// instance. This is what keeps Run's poll loop convergent — a task left
	// FAILED after its ladder is exhausted (or an escalation is budget-
	// denied) has the same "latest execution" on every subsequent tick, so
	// without this it would emit a fresh "final" KindEscalated event, and
	// re-run the same decision, every single poll forever. A restart clears
	// this map, so a task can be reconsidered once more after a restart —
	// intentional: it's an idempotent bookkeeping decision, not orchestration
	// state, so re-deriving it once is harmless.
	mu      sync.Mutex
	handled map[string]bool
}

// DefaultPollInterval is used by Run when pollInterval <= 0.
const DefaultPollInterval = 30 * time.Second

// Run polls for role-typed FAILED tasks every pollInterval until ctx is
// cancelled.
func (s *Scheduler) Run(ctx context.Context, pollInterval time.Duration) {
	if pollInterval <= 0 {
		pollInterval = DefaultPollInterval
	}
	ticker := time.NewTicker(pollInterval)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			s.Tick(ctx)
		}
	}
}

// Tick runs a single poll pass. Exported so tests can drive it directly
// without waiting on a ticker.
func (s *Scheduler) Tick(ctx context.Context) {
	tasks, err := s.Store.ListTasks(storage.TaskFilter{State: task.StateFailed})
	if err != nil {
		s.logf("scheduler: list failed tasks", "error", err)
		return
	}
	for _, t := range tasks {
		if t.Agent.Role == "" {
			continue
		}
		s.processTask(ctx, t)
	}
}

func (s *Scheduler) logf(msg string, args ...any) {
	if s.Logger != nil {
		s.Logger.Warn(msg, args...)
	}
}

func (s *Scheduler) markHandled(execID string) bool {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.handled == nil {
		s.handled = make(map[string]bool)
	}
	if s.handled[execID] {
		return false
	}
	s.handled[execID] = true
	return true
}

func (s *Scheduler) processTask(ctx context.Context, t *task.Task) {
	execs, err := s.Store.ListExecutions(t.ID)
	if err != nil || len(execs) == 0 {
		return
	}
	latest := execs[0] // ListExecutions orders DESC by start_time.
	if latest.Status != "FAILED" {
		return
	}
	if !s.markHandled(latest.ID) {
		return // already decided for this execution — converged, nothing to do.
	}

	row, err := s.Store.GetActiveRoleConfig(t.Agent.Role)
	if err != nil {
		s.logf("scheduler: no active role config for role", "role", t.Agent.Role, "taskID", t.ID, "error", err)
		return
	}
	var rc role.RoleConfig
	if err := json.Unmarshal([]byte(row.ConfigJSON), &rc); err != nil {
		s.logf("scheduler: decode role config", "role", t.Agent.Role, "taskID", t.ID, "error", err)
		return
	}
	if len(rc.EscalationLadder) == 0 {
		return
	}

	currentRung := latest.EscalationRung
	if currentRung < 0 {
		currentRung = 0
	}
	if currentRung >= len(rc.EscalationLadder) {
		// Ladder already exhausted (e.g. the ladder was shortened after this
		// task started climbing it) — nothing more to do.
		return
	}
	tier := rc.EscalationLadder[currentRung]
	attempts := attemptsAtRung(execs, currentRung)

	if attempts < tier.MaxRetries {
		s.retrySameRung(ctx, t, currentRung)
		return
	}

	nextRung := currentRung + 1
	if nextRung >= len(rc.EscalationLadder) || len(rc.EscalationLadder[nextRung].Candidates) == 0 {
		s.decline(ctx, t, currentRung, "", "escalation ladder exhausted")
		return
	}
	nextTier := rc.EscalationLadder[nextRung]
	target := nextTier.Candidates[0]

	estCost := rc.DefaultBudgetUSD
	if estCost <= 0 {
		estCost = t.Agent.MaxBudgetUSD
	}
	allowed := true
	if s.Budget != nil {
		var berr error
		allowed, berr = s.Budget.Allow(target.Provider, estCost)
		if berr != nil {
			s.logf("scheduler: budget check failed; declining escalation", "taskID", t.ID, "error", berr)
			allowed = false
		}
	}
	if !allowed {
		s.decline(ctx, t, currentRung, target.Provider, "budget denied")
		return
	}
	s.escalate(ctx, t, currentRung, nextRung, target)
}

// attemptsAtRung counts how many executions, starting from the most recent
// (execs[0]) and moving backward, ran at rung consecutively — i.e. how many
// attempts have already been made at the task's current tier since it last
// moved to (or started at) that tier.
func attemptsAtRung(execs []*storage.Execution, rung int) int {
	n := 0
	for _, e := range execs {
		if e.EscalationRung != rung {
			break
		}
		n++
	}
	return n
}

func (s *Scheduler) retrySameRung(ctx context.Context, t *task.Task, rung int) {
	if err := s.Store.UpdateTaskState(t.ID, task.StateQueued); err != nil {
		s.logf("scheduler: retry: update task state", "taskID", t.ID, "error", err)
		return
	}
	resubmit := *t
	resubmit.State = task.StateQueued
	if err := s.Pool.Submit(ctx, &resubmit); err != nil {
		s.logf("scheduler: retry: submit", "taskID", t.ID, "error", err)
	}
}

func (s *Scheduler) escalate(ctx context.Context, t *task.Task, fromRung, toRung int, target role.Rung) {
	fromProvider := t.Agent.Type
	newAgent := t.Agent
	newAgent.Type = target.Provider
	newAgent.Model = target.Model
	if err := s.Store.UpdateTaskAgent(t.ID, newAgent); err != nil {
		s.logf("scheduler: escalate: update task agent", "taskID", t.ID, "error", err)
		return
	}
	if err := s.Store.UpdateTaskState(t.ID, task.StateQueued); err != nil {
		s.logf("scheduler: escalate: update task state", "taskID", t.ID, "error", err)
		return
	}
	s.emitEscalated(t.ID, fromRung, toRung, fromProvider, target.Provider, false, "")

	resubmit := *t
	resubmit.Agent = newAgent
	resubmit.State = task.StateQueued
	if err := s.Pool.Submit(ctx, &resubmit); err != nil {
		s.logf("scheduler: escalate: submit", "taskID", t.ID, "error", err)
	}
}

// decline records that no further escalation will happen for t right now
// (ladder exhausted or budget denied) and leaves it FAILED for human
// attention.
func (s *Scheduler) decline(ctx context.Context, t *task.Task, atRung int, consideredProvider, reason string) {
	s.emitEscalated(t.ID, atRung, atRung, t.Agent.Type, consideredProvider, true, reason)
}

func (s *Scheduler) emitEscalated(taskID string, fromRung, toRung int, fromProvider, toProvider string, final bool, reason string) {
	payload, _ := json.Marshal(struct {
		FromRung     int    `json:"from_rung"`
		ToRung       int    `json:"to_rung"`
		FromProvider string `json:"from_provider"`
		ToProvider   string `json:"to_provider,omitempty"`
		Final        bool   `json:"final"`
		Reason       string `json:"reason,omitempty"`
	}{FromRung: fromRung, ToRung: toRung, FromProvider: fromProvider, ToProvider: toProvider, Final: final, Reason: reason})
	if err := s.Store.CreateEvent(&event.Event{
		TaskID:  taskID,
		Kind:    event.KindEscalated,
		Actor:   event.ActorSystem,
		Payload: payload,
	}); err != nil {
		s.logf("scheduler: emit escalated event", "taskID", taskID, "error", err)
	}
}