1 files changed, 91 insertions, 0 deletions
diff --git a/internal/storage/dashboard.go b/internal/storage/dashboard.go
new file mode 100644
index 0000000..e432af1
--- /dev/null
+++ b/internal/storage/dashboard.go
@@ -0,0 +1,91 @@
+package storage
+
+import "time"
+
+// EscalationBucket is one (escalation_rung, provider) aggregate over
+// executions in a window — the raw material for the budget dashboard's
+// escalation funnel (GET /api/escalation-funnel), the harness's
+// "how much falls through each rung" story made visible.
+type EscalationBucket struct {
+	Rung    int     `json:"rung"`
+	Agent   string  `json:"agent"`
+	Count   int     `json:"count"`
+	CostUSD float64 `json:"cost_usd"`
+}
+
+// QueryEscalationFunnel aggregates executions started at or after `since` by
+// (escalation_rung, agent), counting executions and summing cost. Excludes
+// executions still in flight (RUNNING/QUEUED/PENDING have no terminal
+// outcome yet to attribute to a rung).
+//
+// Rung 0 is not exclusively "resolved locally": executions.escalation_rung
+// defaults to 0 uniformly, so every non-role-typed execution (which never
+// climbs a ladder at all) also lands there alongside role-typed tasks
+// genuinely resolved at their ladder's first tier. This mirrors the
+// aggregate shape suggested for this phase and keeps the query a plain
+// GROUP BY with no join — a role-only variant would need to join tasks and
+// inspect config_json (no queryable role column on executions), which is
+// intentionally left out of scope here.
+func (s *DB) QueryEscalationFunnel(since time.Time) ([]EscalationBucket, error) {
+	rows, err := s.db.Query(`
+		SELECT escalation_rung, COALESCE(agent, ''), COUNT(*), COALESCE(SUM(cost_usd), 0)
+		FROM executions
+		WHERE start_time >= ? AND status NOT IN ('RUNNING', 'QUEUED', 'PENDING')
+		GROUP BY escalation_rung, agent
+		ORDER BY escalation_rung ASC, agent ASC`, since.UTC())
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	out := []EscalationBucket{}
+	for rows.Next() {
+		var b EscalationBucket
+		if err := rows.Scan(&b.Rung, &b.Agent, &b.Count, &b.CostUSD); err != nil {
+			return nil, err
+		}
+		out = append(out, b)
+	}
+	return out, rows.Err()
+}
+
+// SpendPoint is one (time bucket, provider) aggregate of cost — the raw
+// material for the budget dashboard's spend-over-time chart
+// (GET /api/spend-timeseries).
+type SpendPoint struct {
+	Bucket  string  `json:"bucket"` // RFC3339 truncated to the hour, or YYYY-MM-DD
+	Agent   string  `json:"agent"`
+	CostUSD float64 `json:"cost_usd"`
+}
+
+// QuerySpendTimeseries aggregates cost per provider bucketed by hour
+// (hourly=true) or by calendar day (hourly=false), for executions started at
+// or after since. Mirrors QueryDashboardStats' own Throughput (hourly) /
+// Billing (daily) bucketing expressions, just grouped additionally by
+// provider so per-provider trend lines can be drawn.
+func (s *DB) QuerySpendTimeseries(since time.Time, hourly bool) ([]SpendPoint, error) {
+	bucketExpr := "date(start_time)"
+	if hourly {
+		bucketExpr = "strftime('%Y-%m-%dT%H:00:00Z', start_time)"
+	}
+	rows, err := s.db.Query(`
+		SELECT `+bucketExpr+` as bucket, COALESCE(agent, ''), COALESCE(SUM(cost_usd), 0)
+		FROM executions
+		WHERE start_time >= ?
+		GROUP BY bucket, agent
+		ORDER BY bucket ASC, agent ASC`, since.UTC())
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	out := []SpendPoint{}
+	for rows.Next() {
+		var p SpendPoint
+		if err := rows.Scan(&p.Bucket, &p.Agent, &p.CostUSD); err != nil {
+			return nil, err
+		}
+		out = append(out, p)
+	}
+	return out, rows.Err()
+}