feat(budget,storage,config): per-provider spend accounting substrate (Phase 6)

Adds the budget accountant foundation:
- storage: a per-execution `agent` column (additive migration, populated by the
  dispatcher) and SpendByProviderSince(since), summing cost_usd per provider in
  a window. Accurate attribution survives a task running on different providers
  across retries.
- internal/budget: Accountant over a SpendSource + Limits, exposing Headroom
  (remaining/fraction per provider), Allow (would estCost breach the cap), and
  All (one query, sorted). Unconfigured/local providers are unlimited.
- config: a [budget] section (window + provider_5h_usd map). No default cap —
  gating is opt-in by configuring limits.

Fully unit-tested; dispatcher gating and the API/UI surface follow.

https://claude.ai/code/session_01SESwn7kQ7oP62trWw6pc39

6 files changed, 324 insertions, 10 deletions

diff --git a/internal/budget/budget.go b/internal/budget/budget.go
new file mode 100644
index 0000000..60ce9d2
--- /dev/null
+++ b/internal/budget/budget.go
@@ -0,0 +1,122 @@
+// Package budget tracks agent spend against rolling per-provider windows so the
+// dispatcher can refuse (or reroute) paid work that would breach a cap. Local
+// runners are free and simply carry no configured limit.
+package budget
+
+import (
+	"sort"
+	"time"
+)
+
+// DefaultWindow is the rolling spend window when none is configured.
+const DefaultWindow = 5 * time.Hour
+
+// Limits configures per-provider spend caps within a rolling window.
+type Limits struct {
+	// Window is the rolling lookback; DefaultWindow when zero.
+	Window time.Duration
+	// PerProvider maps a provider (claude/gemini/…) to its USD cap within the
+	// window. A missing or non-positive entry means "unlimited" (e.g. local).
+	PerProvider map[string]float64
+}
+
+// SpendSource supplies total spend per provider since a given time.
+type SpendSource interface {
+	SpendByProviderSince(since time.Time) (map[string]float64, error)
+}
+
+// Headroom describes remaining budget for one provider in the current window.
+type Headroom struct {
+	Provider  string  `json:"provider"`
+	Limited   bool    `json:"limited"`
+	Limit     float64 `json:"limit_usd"`
+	Spent     float64 `json:"spent_usd"`
+	Remaining float64 `json:"remaining_usd"`
+	// Fraction is the share of the cap still available, 0..1 (0 when unlimited).
+	Fraction float64 `json:"fraction_remaining"`
+}
+
+// Accountant answers spend questions against a SpendSource and Limits.
+type Accountant struct {
+	src SpendSource
+	lim Limits
+	now func() time.Time
+}
+
+// New builds an Accountant. A zero Window defaults to DefaultWindow.
+func New(src SpendSource, lim Limits) *Accountant {
+	if lim.Window <= 0 {
+		lim.Window = DefaultWindow
+	}
+	return &Accountant{src: src, lim: lim, now: time.Now}
+}
+
+func (a *Accountant) since() time.Time { return a.now().Add(-a.lim.Window) }
+
+func (a *Accountant) headroomFrom(provider string, spends map[string]float64) Headroom {
+	limit := a.lim.PerProvider[provider]
+	if limit <= 0 {
+		return Headroom{Provider: provider, Limited: false}
+	}
+	spent := spends[provider]
+	remaining := limit - spent
+	if remaining < 0 {
+		remaining = 0
+	}
+	return Headroom{
+		Provider:  provider,
+		Limited:   true,
+		Limit:     limit,
+		Spent:     spent,
+		Remaining: remaining,
+		Fraction:  remaining / limit,
+	}
+}
+
+// Headroom returns the remaining budget for one provider in the current window.
+func (a *Accountant) Headroom(provider string) (Headroom, error) {
+	if a.lim.PerProvider[provider] <= 0 {
+		return Headroom{Provider: provider, Limited: false}, nil
+	}
+	spends, err := a.src.SpendByProviderSince(a.since())
+	if err != nil {
+		return Headroom{}, err
+	}
+	return a.headroomFrom(provider, spends), nil
+}
+
+// Allow reports whether starting a task on provider that may cost up to estCost
+// is permitted without breaching the window cap. Unlimited providers always
+// pass. estCost is typically the task's max_budget_usd (0 = unknown).
+func (a *Accountant) Allow(provider string, estCost float64) (bool, error) {
+	h, err := a.Headroom(provider)
+	if err != nil {
+		return false, err
+	}
+	if !h.Limited {
+		return true, nil
+	}
+	return h.Spent+estCost <= h.Limit, nil
+}
+
+// All returns headroom for every configured provider, sorted by name, in a
+// single spend query.
+func (a *Accountant) All() ([]Headroom, error) {
+	if len(a.lim.PerProvider) == 0 {
+		return []Headroom{}, nil
+	}
+	spends, err := a.src.SpendByProviderSince(a.since())
+	if err != nil {
+		return nil, err
+	}
+	providers := make([]string, 0, len(a.lim.PerProvider))
+	for p := range a.lim.PerProvider {
+		providers = append(providers, p)
+	}
+	sort.Strings(providers)
+	out := make([]Headroom, 0, len(providers))
+	for _, p := range providers {
+		out = append(out, a.headroomFrom(p, spends))
+	}
+	return out, nil
+}
diff --git a/internal/budget/budget_test.go b/internal/budget/budget_test.go
new file mode 100644
index 0000000..c257c91
--- /dev/null
+++ b/internal/budget/budget_test.go
@@ -0,0 +1,112 @@
+package budget
+
+import (
+	"testing"
+	"time"
+)
+
+type fakeSpend struct {
+	spends map[string]float64
+	since  time.Time
+}
+
+func (f *fakeSpend) SpendByProviderSince(since time.Time) (map[string]float64, error) {
+	f.since = since
+	return f.spends, nil
+}
+
+func newAcct(spends map[string]float64, lim Limits) (*Accountant, *fakeSpend) {
+	src := &fakeSpend{spends: spends}
+	a := New(src, lim)
+	a.now = func() time.Time { return time.Date(2026, 5, 26, 12, 0, 0, 0, time.UTC) }
+	return a, src
+}
+
+func TestNew_DefaultsWindow(t *testing.T) {
+	a, _ := newAcct(nil, Limits{})
+	if a.lim.Window != DefaultWindow {
+		t.Errorf("want default window %v, got %v", DefaultWindow, a.lim.Window)
+	}
+}
+
+func TestHeadroom_UnlimitedWhenNoLimit(t *testing.T) {
+	a, _ := newAcct(map[string]float64{"local": 999}, Limits{PerProvider: map[string]float64{"claude": 10}})
+	h, err := a.Headroom("local")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if h.Limited {
+		t.Errorf("local should be unlimited, got %+v", h)
+	}
+}
+
+func TestHeadroom_ComputesRemainingAndFraction(t *testing.T) {
+	a, src := newAcct(map[string]float64{"claude": 4}, Limits{Window: 5 * time.Hour, PerProvider: map[string]float64{"claude": 10}})
+	h, err := a.Headroom("claude")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !h.Limited || h.Limit != 10 || h.Spent != 4 || h.Remaining != 6 {
+		t.Errorf("unexpected headroom: %+v", h)
+	}
+	if h.Fraction != 0.6 {
+		t.Errorf("fraction: want 0.6, got %v", h.Fraction)
+	}
+	// Window lookback is now-5h.
+	want := time.Date(2026, 5, 26, 7, 0, 0, 0, time.UTC)
+	if !src.since.Equal(want) {
+		t.Errorf("since: want %v, got %v", want, src.since)
+	}
+}
+
+func TestHeadroom_ClampsNegativeRemaining(t *testing.T) {
+	a, _ := newAcct(map[string]float64{"claude": 15}, Limits{PerProvider: map[string]float64{"claude": 10}})
+	h, _ := a.Headroom("claude")
+	if h.Remaining != 0 {
+		t.Errorf("remaining should clamp to 0, got %v", h.Remaining)
+	}
+}
+
+func TestAllow(t *testing.T) {
+	a, _ := newAcct(map[string]float64{"claude": 8}, Limits{PerProvider: map[string]float64{"claude": 10}})
+
+	if ok, _ := a.Allow("claude", 1.5); !ok {
+		t.Error("8 + 1.5 <= 10 should be allowed")
+	}
+	if ok, _ := a.Allow("claude", 3); ok {
+		t.Error("8 + 3 > 10 should be denied")
+	}
+	if ok, _ := a.Allow("local", 1000); !ok {
+		t.Error("unlimited provider should always allow")
+	}
+}
+
+func TestAll_SortedAndSingleQuery(t *testing.T) {
+	a, src := newAcct(map[string]float64{"claude": 2, "gemini": 1},
+		Limits{PerProvider: map[string]float64{"gemini": 5, "claude": 10}})
+	calls := 0
+	orig := src.spends
+	a2 := New(&countingSpend{spends: orig, calls: &calls}, a.lim)
+	a2.now = a.now
+
+	hs, err := a2.All()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(hs) != 2 || hs[0].Provider != "claude" || hs[1].Provider != "gemini" {
+		t.Errorf("want [claude gemini] sorted, got %+v", hs)
+	}
+	if calls != 1 {
+		t.Errorf("All should issue exactly one spend query, got %d", calls)
+	}
+}
+
+type countingSpend struct {
+	spends map[string]float64
+	calls  *int
+}
+
+func (c *countingSpend) SpendByProviderSince(time.Time) (map[string]float64, error) {
+	*c.calls++
+	return c.spends, nil
+}
diff --git a/internal/config/config.go b/internal/config/config.go
index 94f3ec7..58de95c 100644
--- a/internal/config/config.go
+++ b/internal/config/config.go
@@ -69,6 +69,17 @@ type Config struct {
 	VAPIDEmail       string        `toml:"vapid_email"`
 	ClaudeConfigDir  string        `toml:"claude_config_dir"`
 	LocalModel       LocalModel    `toml:"local_model"`
+	Budget           BudgetConfig  `toml:"budget"`
+}
+
+// BudgetConfig configures rolling per-provider spend caps. With no providers
+// set, budget gating is disabled (nothing is blocked) — there is intentionally
+// no default cap; the user must opt in by configuring limits.
+type BudgetConfig struct {
+	// Window is the rolling lookback duration (e.g. "5h"); defaults to 5h.
+	Window string `toml:"window"`
+	// Provider5hUSD maps a provider (claude/gemini) to its USD cap within Window.
+	Provider5hUSD map[string]float64 `toml:"provider_5h_usd"`
 }
 
 func Default() (*Config, error) {
diff --git a/internal/executor/executor.go b/internal/executor/executor.go
index 6d6c528..4333f51 100644
--- a/internal/executor/executor.go
+++ b/internal/executor/executor.go
@@ -1025,6 +1025,7 @@ func (p *Pool) execute(ctx context.Context, t *task.Task) {
 		TaskID:    t.ID,
 		StartTime: time.Now().UTC(),
 		Status:    "RUNNING",
+		Agent:     agentType,
 	}
 
 	// Pre-populate log paths so they're available in the DB immediately —
diff --git a/internal/storage/db.go b/internal/storage/db.go
index adb7c02..e03a902 100644
--- a/internal/storage/db.go
+++ b/internal/storage/db.go
@@ -140,6 +140,7 @@ func (s *DB) migrate() error {
 		`ALTER TABLE tasks ADD COLUMN checker_report TEXT NOT NULL DEFAULT ''`,
 		`ALTER TABLE executions ADD COLUMN tokens_in INTEGER`,
 		`ALTER TABLE executions ADD COLUMN tokens_out INTEGER`,
+		`ALTER TABLE executions ADD COLUMN agent TEXT`,
 		`CREATE TABLE IF NOT EXISTS events (
 			id TEXT PRIMARY KEY,
 			task_id TEXT NOT NULL,
@@ -536,6 +537,7 @@ type Execution struct {
 	ErrorMsg    string
 	SessionID  string // claude --session-id; persisted for resume
 	SandboxDir string // preserved sandbox path when task is BLOCKED; resume must run here
+	Agent      string // provider that ran this execution (claude/gemini/local); for budget accounting
 
 	Changestats *task.Changestats // stored as JSON; nil if not yet recorded
 	Commits     []task.GitCommit // stored as JSON; empty if no commits
@@ -584,10 +586,10 @@ func (s *DB) CreateExecutionAndSetRunning(e *Execution) error {
 		commitsJSON = string(b)
 	}
 	if _, err := tx.Exec(`
-		INSERT INTO executions (id, task_id, start_time, end_time, exit_code, status, stdout_path, stderr_path, artifact_dir, cost_usd, error_msg, session_id, sandbox_dir, changestats_json, commits_json)
-		VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, ?)`,
+		INSERT INTO executions (id, task_id, start_time, end_time, exit_code, status, stdout_path, stderr_path, artifact_dir, cost_usd, error_msg, session_id, sandbox_dir, changestats_json, commits_json, agent)
+		VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, ?, ?)`,
 		e.ID, e.TaskID, e.StartTime.UTC(), e.EndTime.UTC(), e.ExitCode, e.Status,
-		e.StdoutPath, e.StderrPath, e.ArtifactDir, e.CostUSD, e.ErrorMsg, e.SessionID, e.SandboxDir, commitsJSON,
+		e.StdoutPath, e.StderrPath, e.ArtifactDir, e.CostUSD, e.ErrorMsg, e.SessionID, e.SandboxDir, commitsJSON, e.Agent,
 	); err != nil {
 		return err
 	}
@@ -601,6 +603,32 @@ func (s *DB) CreateExecutionAndSetRunning(e *Execution) error {
 	return tx.Commit()
 }
 
+// SpendByProviderSince returns total cost_usd per provider (executions.agent)
+// for executions started at or after `since`. Executions with no recorded agent
+// are grouped under the empty string and can be ignored by callers.
+func (s *DB) SpendByProviderSince(since time.Time) (map[string]float64, error) {
+	rows, err := s.db.Query(`
+		SELECT COALESCE(agent, ''), COALESCE(SUM(cost_usd), 0)
+		FROM executions
+		WHERE start_time >= ?
+		GROUP BY agent`, since.UTC())
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	out := make(map[string]float64)
+	for rows.Next() {
+		var agent string
+		var cost float64
+		if err := rows.Scan(&agent, &cost); err != nil {
+			return nil, err
+		}
+		out[agent] = cost
+	}
+	return out, rows.Err()
+}
+
 // CreateExecution inserts an execution record.
 func (s *DB) CreateExecution(e *Execution) error {
 	var changestatsJSON *string
@@ -621,23 +649,23 @@ func (s *DB) CreateExecution(e *Execution) error {
 		commitsJSON = string(b)
 	}
 	_, err := s.db.Exec(`
-		INSERT INTO executions (id, task_id, start_time, end_time, exit_code, status, stdout_path, stderr_path, artifact_dir, cost_usd, error_msg, session_id, sandbox_dir, changestats_json, commits_json, tokens_in, tokens_out)
-		VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
+		INSERT INTO executions (id, task_id, start_time, end_time, exit_code, status, stdout_path, stderr_path, artifact_dir, cost_usd, error_msg, session_id, sandbox_dir, changestats_json, commits_json, tokens_in, tokens_out, agent)
+		VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
 		e.ID, e.TaskID, e.StartTime.UTC(), e.EndTime.UTC(), e.ExitCode, e.Status,
-		e.StdoutPath, e.StderrPath, e.ArtifactDir, e.CostUSD, e.ErrorMsg, e.SessionID, e.SandboxDir, changestatsJSON, commitsJSON, e.TokensIn, e.TokensOut,
+		e.StdoutPath, e.StderrPath, e.ArtifactDir, e.CostUSD, e.ErrorMsg, e.SessionID, e.SandboxDir, changestatsJSON, commitsJSON, e.TokensIn, e.TokensOut, e.Agent,
 	)
 	return err
 }
 
 // GetExecution retrieves an execution by ID.
 func (s *DB) GetExecution(id string) (*Execution, error) {
-	row := s.db.QueryRow(`SELECT id, task_id, start_time, end_time, exit_code, status, stdout_path, stderr_path, artifact_dir, cost_usd, error_msg, session_id, sandbox_dir, changestats_json, commits_json, tokens_in, tokens_out FROM executions WHERE id = ?`, id)
+	row := s.db.QueryRow(`SELECT id, task_id, start_time, end_time, exit_code, status, stdout_path, stderr_path, artifact_dir, cost_usd, error_msg, session_id, sandbox_dir, changestats_json, commits_json, tokens_in, tokens_out, agent FROM executions WHERE id = ?`, id)
 	return scanExecution(row)
 }
 
 // ListExecutions returns executions for a task.
 func (s *DB) ListExecutions(taskID string) ([]*Execution, error) {
-	rows, err := s.db.Query(`SELECT id, task_id, start_time, end_time, exit_code, status, stdout_path, stderr_path, artifact_dir, cost_usd, error_msg, session_id, sandbox_dir, changestats_json, commits_json, tokens_in, tokens_out FROM executions WHERE task_id = ? ORDER BY start_time DESC`, taskID)
+	rows, err := s.db.Query(`SELECT id, task_id, start_time, end_time, exit_code, status, stdout_path, stderr_path, artifact_dir, cost_usd, error_msg, session_id, sandbox_dir, changestats_json, commits_json, tokens_in, tokens_out, agent FROM executions WHERE task_id = ? ORDER BY start_time DESC`, taskID)
 	if err != nil {
 		return nil, err
 	}
@@ -656,7 +684,7 @@ func (s *DB) ListExecutions(taskID string) ([]*Execution, error) {
 
 // GetLatestExecution returns the most recent execution for a task.
 func (s *DB) GetLatestExecution(taskID string) (*Execution, error) {
-	row := s.db.QueryRow(`SELECT id, task_id, start_time, end_time, exit_code, status, stdout_path, stderr_path, artifact_dir, cost_usd, error_msg, session_id, sandbox_dir, changestats_json, commits_json, tokens_in, tokens_out FROM executions WHERE task_id = ? ORDER BY start_time DESC LIMIT 1`, taskID)
+	row := s.db.QueryRow(`SELECT id, task_id, start_time, end_time, exit_code, status, stdout_path, stderr_path, artifact_dir, cost_usd, error_msg, session_id, sandbox_dir, changestats_json, commits_json, tokens_in, tokens_out, agent FROM executions WHERE task_id = ? ORDER BY start_time DESC LIMIT 1`, taskID)
 	return scanExecution(row)
 }
 
@@ -1137,8 +1165,9 @@ func scanExecution(row scanner) (*Execution, error) {
 	var commitsJSON sql.NullString
 	var tokensIn sql.NullInt64
 	var tokensOut sql.NullInt64
+	var agent sql.NullString
 	err := row.Scan(&e.ID, &e.TaskID, &e.StartTime, &e.EndTime, &e.ExitCode, &e.Status,
-		&e.StdoutPath, &e.StderrPath, &e.ArtifactDir, &e.CostUSD, &e.ErrorMsg, &sessionID, &sandboxDir, &changestatsJSON, &commitsJSON, &tokensIn, &tokensOut)
+		&e.StdoutPath, &e.StderrPath, &e.ArtifactDir, &e.CostUSD, &e.ErrorMsg, &sessionID, &sandboxDir, &changestatsJSON, &commitsJSON, &tokensIn, &tokensOut, &agent)
 	if err != nil {
 		return nil, err
 	}
@@ -1146,6 +1175,7 @@ func scanExecution(row scanner) (*Execution, error) {
 	e.SandboxDir = sandboxDir.String
 	e.TokensIn = tokensIn.Int64
 	e.TokensOut = tokensOut.Int64
+	e.Agent = agent.String
 	if changestatsJSON.Valid && changestatsJSON.String != "" {
 		var cs task.Changestats
 		if err := json.Unmarshal([]byte(changestatsJSON.String), &cs); err != nil {
diff --git a/internal/storage/db_test.go b/internal/storage/db_test.go
index 0e67e02..4d744a4 100644
--- a/internal/storage/db_test.go
+++ b/internal/storage/db_test.go
@@ -309,6 +309,44 @@ func TestListTasks_WithLimit(t *testing.T) {
 	}
 }
 
+func TestSpendByProviderSince(t *testing.T) {
+	db := testDB(t)
+	now := time.Now().UTC()
+
+	tk := &task.Task{
+		ID: "spend-task", Name: "S", Agent: task.AgentConfig{Type: "claude", Instructions: "x"},
+		Priority: task.PriorityNormal, Retry: task.RetryConfig{MaxAttempts: 1, Backoff: "linear"},
+		Tags: []string{}, DependsOn: []string{}, State: task.StatePending, CreatedAt: now, UpdatedAt: now,
+	}
+	if err := db.CreateTask(tk); err != nil {
+		t.Fatal(err)
+	}
+
+	// Two recent claude execs, one recent gemini, one old claude (outside window).
+	execs := []*Execution{
+		{ID: "s1", TaskID: "spend-task", StartTime: now.Add(-1 * time.Hour), Status: "COMPLETED", CostUSD: 1.5, Agent: "claude"},
+		{ID: "s2", TaskID: "spend-task", StartTime: now.Add(-2 * time.Hour), Status: "COMPLETED", CostUSD: 2.0, Agent: "claude"},
+		{ID: "s3", TaskID: "spend-task", StartTime: now.Add(-30 * time.Minute), Status: "COMPLETED", CostUSD: 0.75, Agent: "gemini"},
+		{ID: "s4", TaskID: "spend-task", StartTime: now.Add(-10 * time.Hour), Status: "COMPLETED", CostUSD: 99, Agent: "claude"},
+	}
+	for _, e := range execs {
+		if err := db.CreateExecution(e); err != nil {
+			t.Fatalf("create exec %s: %v", e.ID, err)
+		}
+	}
+
+	spends, err := db.SpendByProviderSince(now.Add(-5 * time.Hour))
+	if err != nil {
+		t.Fatalf("SpendByProviderSince: %v", err)
+	}
+	if got := spends["claude"]; got != 3.5 {
+		t.Errorf("claude spend: want 3.5 (old $99 excluded), got %v", got)
+	}
+	if got := spends["gemini"]; got != 0.75 {
+		t.Errorf("gemini spend: want 0.75, got %v", got)
+	}
+}
+
 func TestCreateExecution_AndGet(t *testing.T) {
 	db := testDB(t)
 	now := time.Now().UTC().Truncate(time.Second)