From 39f74dbbc7adca0e2058112408bb99694deefcaf Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 26 May 2026 20:27:36 +0000 Subject: feat(budget,storage,config): per-provider spend accounting substrate (Phase 6) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- internal/storage/db.go | 50 ++++++++++++++++++++++++++++++++++++--------- internal/storage/db_test.go | 38 ++++++++++++++++++++++++++++++++++ 2 files changed, 78 insertions(+), 10 deletions(-) (limited to 'internal/storage') 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) -- cgit v1.2.3