1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
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()
}
|
