2 files changed, 103 insertions, 0 deletions

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}