feat(executor): budget-gate the dispatcher — reroute to local or block (Phase 6)

The pool now consults a BudgetGate before taking an agent slot. If running on
the chosen paid provider could breach its rolling window (estimated by the
task's max_budget_usd), it reroutes to the free local runner when one is
registered; otherwise it stops before running and marks the task
BUDGET_EXCEEDED. serve.go builds a budget.Accountant from [budget] config and
wires it into the pool (only when limits are configured).

Allows the QUEUED → BUDGET_EXCEEDED transition, since the gate blocks a queued
task before it ever reaches RUNNING. Covered by pool tests for both the block
and reroute paths against a fake gate.

https://claude.ai/code/session_01SESwn7kQ7oP62trWw6pc39

4 files changed, 122 insertions, 1 deletions

diff --git a/internal/cli/serve.go b/internal/cli/serve.go
index e545763..55fdaf5 100644
--- a/internal/cli/serve.go
+++ b/internal/cli/serve.go
@@ -11,6 +11,7 @@ import (
 	"time"
 
 	"github.com/thepeterstone/claudomator/internal/api"
+	"github.com/thepeterstone/claudomator/internal/budget"
 	"github.com/thepeterstone/claudomator/internal/executor"
 	"github.com/thepeterstone/claudomator/internal/notify"
 	"github.com/thepeterstone/claudomator/internal/storage"
@@ -144,6 +145,23 @@ func serve(addr string) error {
 		pool.LLM = localClient
 	}
 
+	// Budget accountant: gate paid providers against rolling per-provider caps.
+	// With no limits configured this is created but allows everything.
+	var accountant *budget.Accountant
+	if len(cfg.Budget.Provider5hUSD) > 0 {
+		window := budget.DefaultWindow
+		if cfg.Budget.Window != "" {
+			if d, derr := time.ParseDuration(cfg.Budget.Window); derr == nil {
+				window = d
+			} else {
+				logger.Warn("invalid budget.window; using default", "value", cfg.Budget.Window, "default", window)
+			}
+		}
+		accountant = budget.New(store, budget.Limits{Window: window, PerProvider: cfg.Budget.Provider5hUSD})
+		pool.Budget = accountant
+		logger.Info("budget gating enabled", "window", window, "limits", cfg.Budget.Provider5hUSD)
+	}
+
 	if err := store.SeedProjects(); err != nil {
 		logger.Error("failed to seed projects", "error", err)
 	}
diff --git a/internal/executor/executor.go b/internal/executor/executor.go
index 4333f51..8614481 100644
--- a/internal/executor/executor.go
+++ b/internal/executor/executor.go
@@ -93,6 +93,14 @@ type Pool struct {
 	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.
@@ -937,6 +945,44 @@ func (p *Pool) execute(ctx context.Context, t *task.Task) {
 		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.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).
diff --git a/internal/executor/executor_test.go b/internal/executor/executor_test.go
index d98efbf..b737e22 100644
--- a/internal/executor/executor_test.go
+++ b/internal/executor/executor_test.go
@@ -217,6 +217,63 @@ func TestPool_Submit_Subtask_GoesToCompleted(t *testing.T) {
 	}
 }
 
+type fakeGate struct{ deny map[string]bool }
+
+func (g fakeGate) Allow(provider string, _ float64) (bool, error) { return !g.deny[provider], nil }
+
+func TestPool_BudgetGate_BlocksWhenNoLocalFallback(t *testing.T) {
+	store := testStore(t)
+	runner := &mockRunner{}
+	pool := NewPool(2, map[string]Runner{"claude": runner}, store, slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelError})))
+	pool.Budget = fakeGate{deny: map[string]bool{"claude": true}}
+
+	tk := makeTask("bg-1")
+	store.CreateTask(tk)
+	if err := pool.Submit(context.Background(), tk); err != nil {
+		t.Fatalf("submit: %v", err)
+	}
+
+	result := <-pool.Results()
+	if result.Execution.Status != "BUDGET_EXCEEDED" {
+		t.Errorf("status: want BUDGET_EXCEEDED, got %q", result.Execution.Status)
+	}
+	if runner.callCount() != 0 {
+		t.Errorf("claude runner should not have run, got %d calls", runner.callCount())
+	}
+	got, _ := store.GetTask("bg-1")
+	if got.State != task.StateBudgetExceeded {
+		t.Errorf("task state: want BUDGET_EXCEEDED, got %v", got.State)
+	}
+}
+
+func TestPool_BudgetGate_ReroutesToLocal(t *testing.T) {
+	store := testStore(t)
+	claudeRunner := &mockRunner{}
+	localRunner := &mockRunner{}
+	pool := NewPool(2, map[string]Runner{"claude": claudeRunner, "local": localRunner}, store, slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelError})))
+	pool.Budget = fakeGate{deny: map[string]bool{"claude": true}} // local is allowed
+
+	tk := makeTask("bg-2")
+	store.CreateTask(tk)
+	if err := pool.Submit(context.Background(), tk); err != nil {
+		t.Fatalf("submit: %v", err)
+	}
+
+	result := <-pool.Results()
+	if result.Err != nil {
+		t.Errorf("expected success after reroute, got %v", result.Err)
+	}
+	if localRunner.callCount() != 1 {
+		t.Errorf("local runner should have run once, got %d", localRunner.callCount())
+	}
+	if claudeRunner.callCount() != 0 {
+		t.Errorf("claude runner should not have run, got %d", claudeRunner.callCount())
+	}
+	if result.Execution.Agent != "local" {
+		t.Errorf("execution agent: want local, got %q", result.Execution.Agent)
+	}
+}
+
 func TestPool_Submit_Failure(t *testing.T) {
 	store := testStore(t)
 	runner := &mockRunner{err: fmt.Errorf("boom"), exitCode: 1}
diff --git a/internal/task/task.go b/internal/task/task.go
index 935a238..eeac49e 100644
--- a/internal/task/task.go
+++ b/internal/task/task.go
@@ -138,7 +138,7 @@ type BatchFile struct {
 // BLOCKED may advance to READY when all subtasks complete, or back to QUEUED on user answer.
 var validTransitions = map[State]map[State]bool{
 	StatePending:        {StateQueued: true, StateCancelled: true},
-	StateQueued:         {StateRunning: true, StateCancelled: true, StateReady: true}, // READY: parent task completed via subtask delegation
+	StateQueued:         {StateRunning: true, StateCancelled: true, StateReady: true, StateBudgetExceeded: true}, // READY: subtask delegation; BUDGET_EXCEEDED: dispatcher budget gate blocks before running
 	StateRunning:        {StateReady: true, StateCompleted: true, StateFailed: true, StateTimedOut: true, StateCancelled: true, StateBudgetExceeded: true, StateBlocked: true},
 	StateReady:          {StateCompleted: true, StatePending: true},
 	StateFailed:         {StateQueued: true}, // retry