From 787b7fb1aed92c2b701724a7741576053b93cccb Mon Sep 17 00:00:00 2001 From: Claude Sonnet 5 Date: Fri, 3 Jul 2026 23:01:50 +0000 Subject: 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 Claude-Session: https://claude.ai/code/session_01V1moSNCJRcP6kykA4tyUSs --- internal/scheduler/scheduler.go | 277 ++++++++++++++++++++++++++ internal/scheduler/scheduler_test.go | 369 +++++++++++++++++++++++++++++++++++ 2 files changed, 646 insertions(+) create mode 100644 internal/scheduler/scheduler.go create mode 100644 internal/scheduler/scheduler_test.go (limited to 'internal/scheduler') 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()) + } +} -- cgit v1.2.3