feat(role): add versioned role configs + escalation ladder + scheduler (Phase 5)

Two parts:

Part A (fixes a gap from Phase 1): Groq/OpenRouter/OpenAI were documented in
docs/api-keys-setup.md as usable once configured, but nothing actually
constructed runners for them. internal/cli/cloudrunners.go consolidates
anthropic/google/groq/openrouter/openai NativeRunner construction into one
table-driven registerCloudRunners() helper, replacing the two hand-written
per-provider blocks in serve.go/run.go. Groq/OpenRouter/OpenAI reuse
openaicompat (no new adapter code) at SandboxKind: "docker".

Part B: the token-husbanding harness's core routing mechanism.

- internal/role: RoleConfig/Tier/Rung -- a role's system prompt and a
  multi-tier (provider, model) escalation ladder, versioned via config_json.
- storage: new role_configs table (draft/active/retired, UNIQUE(role,
  version)) with transactional activate-retires-prior-active semantics;
  new executions.escalation_rung column.
- task.AgentConfig.Role string -- purely additive; every existing task shape
  (Agent.Role == "") is unaffected, proven by TestPool_Execute_NonRoleTask_Unaffected
  plus the full pre-existing suite passing unchanged.
- executor.Pool.execute(): role-typed tasks with no Agent.Type yet resolve
  tier 0 of their active ladder (round-robin across multi-candidate tiers,
  skipping rate-limited providers, falling back to soonest-clearing) before
  the existing pickAgent/Classifier path runs; SystemPrompt applies to
  Agent.SystemPromptAppend. Already-resolved role tasks (scheduler resubmits)
  get their escalation_rung re-derived read-only via findTierIndex.
- internal/scheduler: polls role-typed FAILED tasks, retries at the same
  rung under MaxRetries or escalates to the next tier's first candidate when
  budget.Accountant.Allow() permits (emitting event.KindEscalated), else
  leaves the task FAILED with a final:true KindEscalated event. An
  in-memory per-execution-ID "handled" set keeps the poll loop convergent.
  Started by `serve` only, config knob [scheduler].poll_interval_seconds.
- internal/api: POST/GET /api/roles/{role}/versions, POST
  /api/roles/{role}/activate -- unauthenticated, matching the existing
  projects/tasks REST endpoints' auth posture (only chatbot MCP, agent MCP,
  and WebSocket are api_token-gated in this codebase today).

Documented as stored-but-not-yet-enforced (CLAUDE.md Design Debt, matching
how task.Priority/RetryConfig are already documented): RoleConfig.Tools/
SandboxKind don't affect dispatch yet; DefaultBudgetUSD is read narrowly as
the scheduler's escalation cost estimate, not enforced at initial dispatch;
scheduler escalation always targets Candidates[0] (no round-robin, unlike
initial-dispatch tier-0 resolution); the scheduler's dedupe is per-process
and resets on restart (idempotent, harmless).

go build/vet/test -race -count=1 all pass, 21 packages.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01V1moSNCJRcP6kykA4tyUSs

2 files changed, 646 insertions, 0 deletions

diff --git a/internal/scheduler/scheduler.go b/internal/scheduler/scheduler.go
new file mode 100644
index 0000000..b3756fc
--- /dev/null
+++ b/internal/scheduler/scheduler.go
@@ -0,0 +1,277 @@
+// 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)
+	}
+}
diff --git a/internal/scheduler/scheduler_test.go b/internal/scheduler/scheduler_test.go
new file mode 100644
index 0000000..18c9e42
--- /dev/null
+++ b/internal/scheduler/scheduler_test.go
@@ -0,0 +1,369 @@
+package scheduler
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/thepeterstone/claudomator/internal/event"
+	"github.com/thepeterstone/claudomator/internal/role"
+	"github.com/thepeterstone/claudomator/internal/storage"
+	"github.com/thepeterstone/claudomator/internal/task"
+)
+
+// fakeStore is a minimal, in-memory implementation of Store for unit-testing
+// the Scheduler without a real SQLite database.
+type fakeStore struct {
+	mu           sync.Mutex
+	tasks        map[string]*task.Task
+	execsByTask  map[string][]*storage.Execution // index 0 = most recent (matches ListExecutions' DESC order)
+	roleConfigs  map[string]*storage.RoleConfigRow
+	agentUpdates []task.AgentConfig
+	stateUpdates []task.State
+	events       []*event.Event
+}
+
+func newFakeStore() *fakeStore {
+	return &fakeStore{
+		tasks:       make(map[string]*task.Task),
+		execsByTask: make(map[string][]*storage.Execution),
+		roleConfigs: make(map[string]*storage.RoleConfigRow),
+	}
+}
+
+func (f *fakeStore) ListTasks(filter storage.TaskFilter) ([]*task.Task, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	var out []*task.Task
+	for _, t := range f.tasks {
+		if filter.State != "" && t.State != filter.State {
+			continue
+		}
+		out = append(out, t)
+	}
+	return out, nil
+}
+
+func (f *fakeStore) ListExecutions(taskID string) ([]*storage.Execution, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return f.execsByTask[taskID], nil
+}
+
+func (f *fakeStore) GetActiveRoleConfig(roleName string) (*storage.RoleConfigRow, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	row, ok := f.roleConfigs[roleName]
+	if !ok {
+		return nil, errors.New("not found")
+	}
+	return row, nil
+}
+
+func (f *fakeStore) UpdateTaskAgent(id string, agent task.AgentConfig) error {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.agentUpdates = append(f.agentUpdates, agent)
+	if t, ok := f.tasks[id]; ok {
+		t.Agent = agent
+	}
+	return nil
+}
+
+func (f *fakeStore) UpdateTaskState(id string, newState task.State) error {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.stateUpdates = append(f.stateUpdates, newState)
+	if t, ok := f.tasks[id]; ok {
+		t.State = newState
+	}
+	return nil
+}
+
+func (f *fakeStore) CreateEvent(e *event.Event) error {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	e.ID = uuid.NewString()
+	f.events = append(f.events, e)
+	return nil
+}
+
+func (f *fakeStore) eventsOfKind(k event.Kind) []*event.Event {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	var out []*event.Event
+	for _, e := range f.events {
+		if e.Kind == k {
+			out = append(out, e)
+		}
+	}
+	return out
+}
+
+// fakePool records every task submitted to it.
+type fakePool struct {
+	mu        sync.Mutex
+	submitted []*task.Task
+	err       error
+}
+
+func (f *fakePool) Submit(_ context.Context, t *task.Task) error {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.submitted = append(f.submitted, t)
+	return f.err
+}
+
+func (f *fakePool) submitCount() int {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return len(f.submitted)
+}
+
+// fakeBudget is a configurable BudgetGate.
+type fakeBudget struct {
+	allow bool
+	err   error
+}
+
+func (f *fakeBudget) Allow(_ string, _ float64) (bool, error) { return f.allow, f.err }
+
+func twoTierLadder() role.RoleConfig {
+	return role.RoleConfig{
+		Role: "coder",
+		EscalationLadder: []role.Tier{
+			{
+				Candidates:    []role.Rung{{Provider: "local", Model: "m0"}},
+				SelectionMode: "single",
+				MaxRetries:    2,
+			},
+			{
+				Candidates: []role.Rung{{Provider: "anthropic", Model: "claude-sonnet-5"}},
+				MaxRetries: 1,
+			},
+		},
+	}
+}
+
+func seedRoleConfig(t *testing.T, f *fakeStore, rc role.RoleConfig) {
+	t.Helper()
+	b, err := json.Marshal(rc)
+	if err != nil {
+		t.Fatalf("marshal role config: %v", err)
+	}
+	f.roleConfigs[rc.Role] = &storage.RoleConfigRow{
+		ID: uuid.NewString(), Role: rc.Role, Version: 1, Status: "active", ConfigJSON: string(b),
+	}
+}
+
+func failedTask(id, roleName string, agentType string) *task.Task {
+	return &task.Task{
+		ID:    id,
+		Name:  "test",
+		Agent: task.AgentConfig{Type: agentType, Role: roleName, MaxBudgetUSD: 0.5},
+		State: task.StateFailed,
+	}
+}
+
+func failedExec(rung int) *storage.Execution {
+	return &storage.Execution{
+		ID:             uuid.NewString(),
+		Status:         "FAILED",
+		EscalationRung: rung,
+		StartTime:      time.Now(),
+	}
+}
+
+// TestScheduler_RetriesSameRung_WhileUnderMaxRetries proves that a task
+// whose current rung has fewer attempts than tier.MaxRetries is resubmitted
+// at the same rung, with no escalation event and no Agent.Type/Model change.
+func TestScheduler_RetriesSameRung_WhileUnderMaxRetries(t *testing.T) {
+	store := newFakeStore()
+	seedRoleConfig(t, store, twoTierLadder())
+
+	tk := failedTask("t1", "coder", "local")
+	store.tasks[tk.ID] = tk
+	// Only 1 attempt so far at rung 0; tier0.MaxRetries == 2, so 1 < 2 → retry.
+	store.execsByTask[tk.ID] = []*storage.Execution{failedExec(0)}
+
+	pool := &fakePool{}
+	sch := &Scheduler{Store: store, Pool: pool}
+	sch.Tick(context.Background())
+
+	if pool.submitCount() != 1 {
+		t.Fatalf("expected 1 submission, got %d", pool.submitCount())
+	}
+	resubmitted := pool.submitted[0]
+	if resubmitted.Agent.Type != "local" {
+		t.Errorf("retry should keep the same provider: got %q", resubmitted.Agent.Type)
+	}
+	if resubmitted.State != task.StateQueued {
+		t.Errorf("resubmitted task should be QUEUED, got %v", resubmitted.State)
+	}
+	if len(store.eventsOfKind(event.KindEscalated)) != 0 {
+		t.Errorf("a same-rung retry should not emit a KindEscalated event")
+	}
+	if len(store.agentUpdates) != 0 {
+		t.Errorf("a same-rung retry should not change the task's Agent config, got %d UpdateTaskAgent calls", len(store.agentUpdates))
+	}
+}
+
+// TestScheduler_EscalatesToNextRung_WhenBudgetAllows proves that once a
+// tier's MaxRetries is exhausted, the scheduler escalates to the next tier
+// (when the budget allows it), updates the task's Agent to the new rung's
+// provider/model, resubmits, and emits a non-final KindEscalated event.
+func TestScheduler_EscalatesToNextRung_WhenBudgetAllows(t *testing.T) {
+	store := newFakeStore()
+	seedRoleConfig(t, store, twoTierLadder())
+
+	tk := failedTask("t1", "coder", "local")
+	store.tasks[tk.ID] = tk
+	// 2 consecutive failed attempts at rung 0; tier0.MaxRetries == 2, so
+	// 2 < 2 is false → escalate.
+	store.execsByTask[tk.ID] = []*storage.Execution{failedExec(0), failedExec(0)}
+
+	pool := &fakePool{}
+	sch := &Scheduler{Store: store, Pool: pool, Budget: &fakeBudget{allow: true}}
+	sch.Tick(context.Background())
+
+	if pool.submitCount() != 1 {
+		t.Fatalf("expected 1 submission, got %d", pool.submitCount())
+	}
+	resubmitted := pool.submitted[0]
+	if resubmitted.Agent.Type != "anthropic" || resubmitted.Agent.Model != "claude-sonnet-5" {
+		t.Errorf("escalated task should carry tier 1's rung: got %q/%q", resubmitted.Agent.Type, resubmitted.Agent.Model)
+	}
+	if resubmitted.State != task.StateQueued {
+		t.Errorf("resubmitted task should be QUEUED, got %v", resubmitted.State)
+	}
+	if len(store.agentUpdates) != 1 {
+		t.Fatalf("expected 1 UpdateTaskAgent call, got %d", len(store.agentUpdates))
+	}
+
+	evs := store.eventsOfKind(event.KindEscalated)
+	if len(evs) != 1 {
+		t.Fatalf("expected 1 KindEscalated event, got %d", len(evs))
+	}
+	var payload struct {
+		FromRung     int    `json:"from_rung"`
+		ToRung       int    `json:"to_rung"`
+		FromProvider string `json:"from_provider"`
+		ToProvider   string `json:"to_provider"`
+		Final        bool   `json:"final"`
+	}
+	if err := json.Unmarshal(evs[0].Payload, &payload); err != nil {
+		t.Fatalf("unmarshal event payload: %v", err)
+	}
+	if payload.Final {
+		t.Errorf("escalation event should have final=false")
+	}
+	if payload.FromRung != 0 || payload.ToRung != 1 {
+		t.Errorf("expected from_rung=0 to_rung=1, got %d -> %d", payload.FromRung, payload.ToRung)
+	}
+	if payload.ToProvider != "anthropic" {
+		t.Errorf("expected to_provider=anthropic, got %q", payload.ToProvider)
+	}
+}
+
+// TestScheduler_DeclinesFinal_WhenBudgetDenies proves that when the budget
+// denies an otherwise-due escalation, the task is left FAILED (no state
+// change, no resubmission) and a final:true KindEscalated event is recorded.
+func TestScheduler_DeclinesFinal_WhenBudgetDenies(t *testing.T) {
+	store := newFakeStore()
+	seedRoleConfig(t, store, twoTierLadder())
+
+	tk := failedTask("t1", "coder", "local")
+	store.tasks[tk.ID] = tk
+	store.execsByTask[tk.ID] = []*storage.Execution{failedExec(0), failedExec(0)}
+
+	pool := &fakePool{}
+	sch := &Scheduler{Store: store, Pool: pool, Budget: &fakeBudget{allow: false}}
+	sch.Tick(context.Background())
+
+	if pool.submitCount() != 0 {
+		t.Fatalf("budget-denied escalation should not resubmit, got %d submissions", pool.submitCount())
+	}
+	if len(store.stateUpdates) != 0 {
+		t.Errorf("budget-denied escalation should not change task state, got %d state updates", len(store.stateUpdates))
+	}
+
+	evs := store.eventsOfKind(event.KindEscalated)
+	if len(evs) != 1 {
+		t.Fatalf("expected 1 KindEscalated event, got %d", len(evs))
+	}
+	var payload struct {
+		Final bool `json:"final"`
+	}
+	if err := json.Unmarshal(evs[0].Payload, &payload); err != nil {
+		t.Fatalf("unmarshal event payload: %v", err)
+	}
+	if !payload.Final {
+		t.Errorf("budget-denied escalation event should have final=true")
+	}
+}
+
+// TestScheduler_DeclinesFinal_WhenLadderExhausted proves the same
+// final:true behavior when the task is already at the ladder's last tier.
+func TestScheduler_DeclinesFinal_WhenLadderExhausted(t *testing.T) {
+	store := newFakeStore()
+	seedRoleConfig(t, store, twoTierLadder())
+
+	tk := failedTask("t1", "coder", "anthropic")
+	store.tasks[tk.ID] = tk
+	// At rung 1 (the last tier), tier1.MaxRetries == 1, so 1 attempt already
+	// made means 1 < 1 is false → would escalate, but there is no rung 2.
+	store.execsByTask[tk.ID] = []*storage.Execution{failedExec(1)}
+
+	pool := &fakePool{}
+	sch := &Scheduler{Store: store, Pool: pool, Budget: &fakeBudget{allow: true}}
+	sch.Tick(context.Background())
+
+	if pool.submitCount() != 0 {
+		t.Fatalf("exhausted ladder should not resubmit, got %d submissions", pool.submitCount())
+	}
+	evs := store.eventsOfKind(event.KindEscalated)
+	if len(evs) != 1 {
+		t.Fatalf("expected 1 KindEscalated event, got %d", len(evs))
+	}
+	var payload struct {
+		Final bool `json:"final"`
+	}
+	if err := json.Unmarshal(evs[0].Payload, &payload); err != nil {
+		t.Fatalf("unmarshal event payload: %v", err)
+	}
+	if !payload.Final {
+		t.Errorf("exhausted-ladder decline event should have final=true")
+	}
+}
+
+// TestScheduler_Convergence_DoesNotReprocessSameExecution proves the
+// double-processing guard: ticking twice against the same unchanged FAILED
+// execution only acts once (no duplicate submissions or events), which is
+// what keeps a task stuck at the end of its ladder from generating a fresh
+// "final" event (or worse, a fresh escalation) on every single poll forever.
+func TestScheduler_Convergence_DoesNotReprocessSameExecution(t *testing.T) {
+	store := newFakeStore()
+	seedRoleConfig(t, store, twoTierLadder())
+
+	tk := failedTask("t1", "coder", "anthropic")
+	store.tasks[tk.ID] = tk
+	store.execsByTask[tk.ID] = []*storage.Execution{failedExec(1)} // last tier, exhausted
+
+	pool := &fakePool{}
+	sch := &Scheduler{Store: store, Pool: pool, Budget: &fakeBudget{allow: true}}
+
+	sch.Tick(context.Background())
+	sch.Tick(context.Background())
+	sch.Tick(context.Background())
+
+	if got := len(store.eventsOfKind(event.KindEscalated)); got != 1 {
+		t.Fatalf("expected exactly 1 KindEscalated event across 3 ticks, got %d", got)
+	}
+	if pool.submitCount() != 0 {
+		t.Fatalf("expected 0 submissions, got %d", pool.submitCount())
+	}
+}