package duckdb import ( "context" "database/sql" "fmt" "math" "slices" "sort" "strings" "time" "go.kenn.io/agentsview/internal/db" "go.kenn.io/agentsview/internal/export" pricingpkg "go.kenn.io/agentsview/internal/pricing" "go.kenn.io/agentsview/internal/signals" ) const ( duckActiveWindow = 10 * time.Minute duckStaleWindow = 60 * time.Minute ) type duckAnalyticsSession struct { id string project string machine string agent string firstMessage *string displayName *string startedAt string endedAt string createdAt string messageCount int userMessageCount int totalOutputTokens int hasTotalOutputTokens bool isAutomated bool terminationStatus *string healthScore *int healthGrade *string outcome string outcomeConfidence string toolFailures int toolRetries int editChurn int compactions int midTaskCompactions int contextPressureMax *float64 qualitySignalVersion int shortPromptCount int unstructuredStart bool missingSuccessCriteriaCount int missingVerificationCount int duplicatePromptCount int noCodeContextCount int runawayToolLoopCount int frustrationMarkerCount int } func (s *Store) analyticsSessions( ctx context.Context, f db.AnalyticsFilter, ) ([]duckAnalyticsSession, error) { return s.analyticsSessionsFiltered(ctx, f, true, true) } // analyticsSessionsFiltered loads candidate sessions, optionally applying // the date and hour/day-of-week predicates at the session level. Skill // analytics passes false for both so those filters can be applied to each // call's own message timestamp instead. With a model filter and an active // hour/dow filter it pairs through the shared scope reducer (see // analyticsSessionsModelTimeFiltered) so an empty-model user turn at the // selected hour keeps its session, matching how the model-scoped panels count. func (s *Store) analyticsSessionsFiltered( ctx context.Context, f db.AnalyticsFilter, includeDate, includeTime bool, ) ([]duckAnalyticsSession, error) { if includeTime && f.HasTimeFilter() && strings.TrimSpace(f.Model) != "" { return s.analyticsSessionsModelTimeFiltered(ctx, f, includeDate) } where, args := duckBuildAnalyticsWhere( f, "COALESCE(s.started_at, s.created_at)", "s.", includeDate, includeTime) rows, err := s.queryContext(ctx, ` SELECT id, project, machine, agent, first_message, COALESCE(display_name, session_name) AS display_name, started_at, ended_at, created_at, message_count, user_message_count, total_output_tokens, has_total_output_tokens, is_automated, termination_status, health_score, health_grade, outcome, outcome_confidence, tool_failure_signal_count, tool_retry_count, edit_churn_count, compaction_count, mid_task_compaction_count, context_pressure_max, quality_signal_version, short_prompt_count, unstructured_start, missing_success_criteria_count, missing_verification_count, duplicate_prompt_count, no_code_context_count, runaway_tool_loop_count FROM sessions s WHERE `+where, args...) if err != nil { return nil, fmt.Errorf("querying duckdb analytics sessions: %w", err) } defer rows.Close() var out []duckAnalyticsSession for rows.Next() { var r duckAnalyticsSession var startedAt, endedAt, createdAt any if err := rows.Scan( &r.id, &r.project, &r.machine, &r.agent, &r.firstMessage, &r.displayName, &startedAt, &endedAt, &createdAt, &r.messageCount, &r.userMessageCount, &r.totalOutputTokens, &r.hasTotalOutputTokens, &r.isAutomated, &r.terminationStatus, &r.healthScore, &r.healthGrade, &r.outcome, &r.outcomeConfidence, &r.toolFailures, &r.toolRetries, &r.editChurn, &r.compactions, &r.midTaskCompactions, &r.contextPressureMax, &r.qualitySignalVersion, &r.shortPromptCount, &r.unstructuredStart, &r.missingSuccessCriteriaCount, &r.missingVerificationCount, &r.duplicatePromptCount, &r.noCodeContextCount, &r.runawayToolLoopCount, ); err != nil { return nil, fmt.Errorf("scanning duckdb analytics session: %w", err) } r.startedAt = formatDBTime(startedAt) r.endedAt = formatDBTime(endedAt) r.createdAt = formatDBTime(createdAt) out = append(out, r) } return out, rows.Err() } // analyticsSessionsModelTimeFiltered loads the date- and model-scoped sessions // (without the in-SQL day/hour predicate) and keeps only those with at least // one scoped message matching the hour/dow filter. Running the shared reducer // instead of the direct m.model time predicate keeps sessions whose matching // message is an empty-model user turn paired with the selected-model assistant. func (s *Store) analyticsSessionsModelTimeFiltered( ctx context.Context, f db.AnalyticsFilter, includeDate bool, ) ([]duckAnalyticsSession, error) { sessions, err := s.analyticsSessionsFiltered(ctx, f, includeDate, false) if err != nil { return nil, err } candidateIDs := make([]string, 0, len(sessions)) for _, session := range sessions { candidateIDs = append(candidateIDs, session.id) } scope, err := s.resolveAnalyticsMessageScope(ctx, candidateIDs, f, false) if err != nil { return nil, err } matched := make(map[string]struct{}) if scope != nil { for id := range scope.MessagesBySession() { matched[id] = struct{}{} } } out := make([]duckAnalyticsSession, 0, len(sessions)) for _, session := range sessions { if _, ok := matched[session.id]; ok { out = append(out, session) } } return out, nil } func duckBuildAnalyticsWhere( f db.AnalyticsFilter, dateCol string, tablePrefix string, includeDate bool, includeTime bool, ) (string, []any) { q := func(col string) string { return tablePrefix + col } preds := []string{ q("message_count") + " > 0", // Mirror the SQLite analytics filter: subagent and fork rows are // excluded unless the filter opts in (sum/count surfaces for // subagents, the activity report for both). The shared helper // qualifies the column with tablePrefix directly. db.RelationshipExclusionSQL(f.IncludeSubagents, f.IncludeForks, tablePrefix), q("deleted_at") + " IS NULL", } var args []any if includeDate { if f.From != "" { preds = append(preds, dateCol+" >= CAST(? AS TIMESTAMP)") args = append(args, duckUsagePaddedUTCBound(f.From+"T00:00:00Z", -14)) } if f.To != "" { preds = append(preds, dateCol+" <= CAST(? AS TIMESTAMP)") args = append(args, duckUsagePaddedUTCBound(f.To+"T23:59:59Z", 14)) } localDate, localDateArgs := duckAnalyticsLocalDateExpr(dateCol, f) if f.From != "" { preds = append(preds, localDate+" >= ?") args = append(args, append(localDateArgs, f.From)...) } if f.To != "" { preds = append(preds, localDate+" <= ?") args = append(args, append(localDateArgs, f.To)...) } } if f.Machine != "" { preds, args = appendDuckAnalyticsCSVFilter(preds, args, q("machine"), f.Machine) } if f.Project != "" { preds = append(preds, q("project")+" = ?") args = append(args, f.Project) } if f.GitBranch != "" { var clause string clause, args = db.BranchPairClauseArgs(q("project"), q("git_branch"), f.GitBranch, args) preds = append(preds, clause) } if f.Agent != "" { preds, args = appendDuckAnalyticsCSVFilter(preds, args, q("agent"), f.Agent) } if modelPred, modelArgs := duckAnalyticsCSVPredicate("m.model", f.Model); modelPred != "" { preds = append(preds, "EXISTS (SELECT 1 FROM messages m WHERE m.session_id = "+q("id")+" AND "+modelPred+")") args = append(args, modelArgs...) } if f.MinUserMessages > 0 { preds = append(preds, q("user_message_count")+" >= ?") args = append(args, f.MinUserMessages) } scope := duckNormalizeAutomatedScope( f.AutomatedScope, f.ExcludeAutomated) if f.ExcludeOneShot { // Exempt subagents from one-shot exclusion when counting them, // mirroring db.AnalyticsFilter.OneShotExclusionSQL. Workflow // subagents are inherently one-shot but represent real work. oneShot := func(base string) string { if f.IncludeSubagents { return "(" + base + " OR " + q("relationship_type") + " = 'subagent')" } return base } if scope != "human" { preds = append(preds, oneShot("("+q("user_message_count")+" > 1 OR "+q("is_automated")+" = TRUE)")) } else { preds = append(preds, oneShot(q("user_message_count")+" > 1")) } } if pred := duckAutomatedScopePredicate( scope, q("is_automated")); pred != "" { preds = append(preds, pred) } if f.ExcludeInteractive { preds = append(preds, q("is_automated")+" = TRUE") } if f.ActiveSince != "" { activeSince := f.ActiveSince if parsed, ok := parseAnalyticsTime(f.ActiveSince); ok { activeSince = parsed.Format(time.RFC3339) } preds = append(preds, "COALESCE("+q("ended_at")+", "+q("started_at")+", "+q("created_at")+") >= CAST(? AS TIMESTAMP)") args = append(args, activeSince) } if pred, predArgs := duckAnalyticsTerminationPred( f.Termination, "COALESCE("+q("ended_at")+", "+q("started_at")+", "+q("created_at")+")", q("termination_status"), ); pred != "" { preds = append(preds, pred) args = append(args, predArgs...) } if includeTime && (f.DayOfWeek != nil || f.Hour != nil) { pred, predArgs := duckAnalyticsMessageTimeExists(f, q("id")) preds = append(preds, pred) args = append(args, predArgs...) } return strings.Join(preds, " AND "), args } func duckNormalizeAutomatedScope( scope string, excludeAutomated bool, ) string { switch strings.TrimSpace(scope) { case "human", "all", "automated": return strings.TrimSpace(scope) } if excludeAutomated { return "human" } return "all" } func duckAutomatedScopePredicate(scope, col string) string { switch scope { case "human": return col + " = FALSE" case "automated": return col + " = TRUE" default: return "" } } func appendDuckAnalyticsCSVFilter( preds []string, args []any, col string, raw string, ) ([]string, []any) { pred, predArgs := duckAnalyticsCSVPredicate(col, raw) if pred != "" { preds = append(preds, pred) args = append(args, predArgs...) } return preds, args } func duckAnalyticsCSVValues(raw string) []string { values := strings.Split(raw, ",") out := values[:0] for _, value := range values { trimmed := strings.TrimSpace(value) if trimmed != "" { out = append(out, trimmed) } } return out } func duckAnalyticsCSVPredicate( col string, raw string, ) (string, []any) { values := duckAnalyticsCSVValues(raw) if len(values) == 0 { return "", nil } if len(values) == 1 { return col + " = ?", []any{values[0]} } placeholders := make([]string, len(values)) args := make([]any, 0, len(values)) for i, value := range values { placeholders[i] = "?" args = append(args, value) } return col + " IN (" + strings.Join(placeholders, ",") + ")", args } func duckAnalyticsLocalDateExpr( tsExpr string, f db.AnalyticsFilter, ) (string, []any) { if f.Timezone != "" { return "strftime(timezone(?, timezone('UTC', " + tsExpr + ")), '%Y-%m-%d')", []any{f.Timezone} } return "strftime(" + tsExpr + ", '%Y-%m-%d')", nil } func duckAnalyticsLocalTimeExpr( tsExpr string, f db.AnalyticsFilter, ) (string, []any) { if f.Timezone != "" { return "timezone(?, timezone('UTC', " + tsExpr + "))", []any{f.Timezone} } return tsExpr, nil } func duckAnalyticsMessageTimeExists( f db.AnalyticsFilter, sessionIDExpr string, ) (string, []any) { preds := []string{ "m.session_id = " + sessionIDExpr, "m.timestamp IS NOT NULL", } var args []any if modelPred, modelArgs := duckAnalyticsCSVPredicate("m.model", f.Model); modelPred != "" { preds = append(preds, modelPred) args = append(args, modelArgs...) } if f.DayOfWeek != nil { local, localArgs := duckAnalyticsLocalTimeExpr("m.timestamp", f) preds = append(preds, "((CAST(strftime("+local+", '%w') AS INTEGER) + 6) % 7) = ?") args = append(args, append(localArgs, *f.DayOfWeek)...) } if f.Hour != nil { local, localArgs := duckAnalyticsLocalTimeExpr("m.timestamp", f) preds = append(preds, "CAST(strftime("+local+", '%H') AS INTEGER) = ?") args = append(args, append(localArgs, *f.Hour)...) } return "EXISTS (SELECT 1 FROM messages m WHERE " + strings.Join(preds, " AND ") + ")", args } func duckAnalyticsTerminationPred( status string, activityExpr string, statusExpr string, ) (string, []any) { if status == "" || status == "all" { return "", nil } now := time.Now().UTC() activeCutoff := now.Add(-duckActiveWindow) staleCutoff := now.Add(-duckStaleWindow) flagged := statusExpr + " IN ('tool_call_pending', 'truncated')" var parts []string var args []any for part := range strings.SplitSeq(status, ",") { switch strings.TrimSpace(part) { case "active": parts = append(parts, activityExpr+" > CAST(? AS TIMESTAMP)") args = append(args, activeCutoff.Format(time.RFC3339)) case "stale": parts = append(parts, "("+flagged+ " AND "+activityExpr+" > CAST(? AS TIMESTAMP)"+ " AND "+activityExpr+" <= CAST(? AS TIMESTAMP))") args = append(args, staleCutoff.Format(time.RFC3339), activeCutoff.Format(time.RFC3339), ) case "unclean": parts = append(parts, "("+flagged+ " AND "+activityExpr+" <= CAST(? AS TIMESTAMP))") args = append(args, staleCutoff.Format(time.RFC3339)) case "clean": parts = append(parts, statusExpr+" = 'clean'") case "awaiting_user": parts = append(parts, statusExpr+" = 'awaiting_user'") } } if len(parts) == 0 { return "", nil } return "(" + strings.Join(parts, " OR ") + ")", args } func duckAnalyticsTimeMatches(t time.Time, f db.AnalyticsFilter) bool { if f.DayOfWeek != nil { dow := (int(t.Weekday()) + 6) % 7 if dow != *f.DayOfWeek { return false } } if f.Hour != nil && t.Hour() != *f.Hour { return false } return true } func analyticsDateTime(r duckAnalyticsSession) string { if r.startedAt != "" { return r.startedAt } return r.createdAt } func analyticsLocalDate(ts, tz string) string { t, ok := parseAnalyticsTime(ts) if !ok { return "" } return t.In(analyticsLocation(tz)).Format("2006-01-02") } func analyticsLocation(tz string) *time.Location { if tz == "" { return time.UTC } loc, err := time.LoadLocation(tz) if err != nil { return time.UTC } return loc } func parseAnalyticsTime(ts string) (time.Time, bool) { if t, ok := parseTimestamp(ts); ok { return t, true } layouts := []string{ "2006-01-02 15:04:05.999999-07", "2006-01-02 15:04:05.999999", "2006-01-02 15:04:05", } for _, layout := range layouts { if t, err := time.Parse(layout, ts); err == nil { return t.UTC(), true } } return time.Time{}, false } func median(values []int) int { if len(values) == 0 { return 0 } n := len(values) if n%2 == 0 { return (values[n/2-1] + values[n/2]) / 2 } return values[n/2] } func firstNonEmpty(values ...string) string { for _, v := range values { if v != "" { return v } } return "" } func round1(v float64) float64 { return math.Round(v*10) / 10 } func (s *Store) getAnalyticsModelsForSessionIDs( ctx context.Context, sessionIDs []string, ) ([]string, error) { if len(sessionIDs) == 0 { return []string{}, nil } models := map[string]bool{} err := duckQueryChunked(sessionIDs, func(chunk []string) error { ph, args := duckInPlaceholders(chunk) rows, err := s.queryContext(ctx, ` SELECT DISTINCT model FROM messages WHERE session_id IN `+ph+` AND COALESCE(model, '') <> '' ORDER BY model`, args...) if err != nil { return fmt.Errorf("querying duckdb analytics models: %w", err) } defer rows.Close() for rows.Next() { var model string if err := rows.Scan(&model); err != nil { return fmt.Errorf("scanning duckdb analytics model: %w", err) } models[model] = true } return rows.Err() }) if err != nil { return nil, err } return sortedBoolKeys(models), nil } func (s *Store) getAnalyticsModelsForSessionIDsFiltered( ctx context.Context, sessionIDs []string, f db.AnalyticsFilter, ) ([]string, error) { if len(sessionIDs) == 0 { return []string{}, nil } seen := make(map[string]struct{}, len(sessionIDs)) unique := make([]string, 0, len(sessionIDs)) for _, sessionID := range sessionIDs { if _, ok := seen[sessionID]; ok { continue } seen[sessionID] = struct{}{} unique = append(unique, sessionID) } filterModels := duckAnalyticsCSVValues(f.Model) allowedModels := make(map[string]struct{}, len(filterModels)) for _, model := range filterModels { allowedModels[model] = struct{}{} } loc := analyticsLocation(f.Timezone) models := map[string]bool{} err := duckQueryChunked(unique, func(chunk []string) error { ph, args := duckInPlaceholders(chunk) rows, err := s.queryContext(ctx, ` SELECT model, timestamp FROM messages WHERE session_id IN `+ph+` AND COALESCE(model, '') <> ''`, args...) if err != nil { return fmt.Errorf("querying duckdb filtered analytics models: %w", err) } defer rows.Close() for rows.Next() { var model string var ts any if err := rows.Scan(&model, &ts); err != nil { return fmt.Errorf("scanning duckdb filtered analytics model: %w", err) } if len(allowedModels) > 0 { if _, ok := allowedModels[model]; !ok { continue } } if f.HasTimeFilter() { t, ok := parseAnalyticsTime(formatDBTime(ts)) if !ok || !duckAnalyticsTimeMatches(t.In(loc), f) { continue } } models[model] = true } return rows.Err() }) if err != nil { return nil, err } return sortedBoolKeys(models), nil } func (s *Store) getAnalyticsFilteredMessageStats( ctx context.Context, sessionIDs []string, f db.AnalyticsFilter, ) (map[string]db.MessageStats, error) { scope, err := s.resolveAnalyticsMessageScope(ctx, sessionIDs, f, false) if err != nil { return nil, err } if scope == nil { return map[string]db.MessageStats{}, nil } return scope.StatsBySession(), nil } func (s *Store) analyticsSessionsWithModelMessageCounts( ctx context.Context, f db.AnalyticsFilter, ) ([]duckAnalyticsSession, error) { sessions, err := s.analyticsSessions(ctx, f) if err != nil || strings.TrimSpace(f.Model) == "" || len(sessions) == 0 { return sessions, err } sessionIDs := make([]string, 0, len(sessions)) for _, session := range sessions { sessionIDs = append(sessionIDs, session.id) } stats, err := s.getAnalyticsFilteredMessageStats( ctx, sessionIDs, f, ) if err != nil { return nil, err } for i := range sessions { stat := stats[sessions[i].id] sessions[i].messageCount = stat.Messages sessions[i].totalOutputTokens = stat.OutputTokens sessions[i].hasTotalOutputTokens = stat.HasOutputTokens } return sessions, nil } func (s *Store) getAnalyticsSummaryWithModelCounts( ctx context.Context, f db.AnalyticsFilter, ) (db.AnalyticsSummary, error) { sessions, err := s.analyticsSessionsWithModelMessageCounts(ctx, f) if err != nil { return db.AnalyticsSummary{}, err } resp := db.AnalyticsSummary{ Agents: map[string]*db.AgentSummary{}, Models: []string{}, } if len(sessions) == 0 { return resp, nil } days := map[string]bool{} projects := map[string]int{} msgCounts := make([]int, 0, len(sessions)) sessionIDs := make([]string, 0, len(sessions)) for _, session := range sessions { date := analyticsLocalDate(analyticsDateTime(session), f.Timezone) resp.TotalSessions++ resp.TotalMessages += session.messageCount if session.hasTotalOutputTokens { resp.TotalOutputTokens += session.totalOutputTokens resp.TokenReportingSessions++ } days[date] = true projects[session.project] += session.messageCount msgCounts = append(msgCounts, session.messageCount) sessionIDs = append(sessionIDs, session.id) if resp.Agents[session.agent] == nil { resp.Agents[session.agent] = &db.AgentSummary{} } resp.Agents[session.agent].Sessions++ resp.Agents[session.agent].Messages += session.messageCount } var models []string if strings.TrimSpace(f.Model) != "" { models, err = s.getAnalyticsModelsForSessionIDsFiltered( ctx, sessionIDs, f, ) } else { models, err = s.getAnalyticsModelsForSessionIDs(ctx, sessionIDs) } if err != nil { return db.AnalyticsSummary{}, err } resp.Models = models resp.ActiveProjects = len(projects) resp.ActiveDays = len(days) resp.AvgMessages = round1(float64(resp.TotalMessages) / float64(resp.TotalSessions)) sort.Ints(msgCounts) resp.MedianMessages = median(msgCounts) if n := len(msgCounts); n > 0 { resp.P90Messages = msgCounts[min(int(math.Floor(float64(n)*0.9))+1, n)-1] } maxMsgs := -1 for _, name := range sortedKeys(projects) { if projects[name] > maxMsgs { maxMsgs = projects[name] resp.MostActive = name } } if resp.TotalMessages > 0 { counts := make([]int, 0, len(projects)) for _, count := range projects { counts = append(counts, count) } sort.Sort(sort.Reverse(sort.IntSlice(counts))) topSum := 0 for _, count := range counts[:min(3, len(counts))] { topSum += count } resp.Concentration = math.Round( float64(topSum)/float64(resp.TotalMessages)*1000, ) / 1000 } return resp, nil } func (s *Store) GetAnalyticsSummary( ctx context.Context, f db.AnalyticsFilter, ) (db.AnalyticsSummary, error) { // Sum/count aggregate: count subagent sessions (mirrors SQLite). f.IncludeSubagents = true if strings.TrimSpace(f.Model) != "" { return s.getAnalyticsSummaryWithModelCounts(ctx, f) } where, args := duckBuildAnalyticsWhere( f, "COALESCE(s.started_at, s.created_at)", "s.", true, true) localDate, localDateArgs := duckAnalyticsLocalDateExpr( "COALESCE(s.started_at, s.created_at)", f) queryArgs := append([]any{}, localDateArgs...) queryArgs = append(queryArgs, args...) query := ` WITH filtered AS ( SELECT s.id, s.project, s.agent, s.message_count, s.total_output_tokens, s.has_total_output_tokens, ` + localDate + ` AS local_date FROM sessions s WHERE ` + where + ` ), ranked AS ( SELECT message_count, ROW_NUMBER() OVER (ORDER BY message_count ASC) AS rn, COUNT(*) OVER () AS n FROM filtered ), project_totals AS ( SELECT project, SUM(message_count) AS messages FROM filtered GROUP BY project ) SELECT COUNT(*) AS total_sessions, COALESCE(SUM(message_count), 0) AS total_messages, COALESCE(SUM(CASE WHEN has_total_output_tokens THEN total_output_tokens ELSE 0 END), 0) AS total_output_tokens, COALESCE(SUM(CASE WHEN has_total_output_tokens THEN 1 ELSE 0 END), 0) AS token_reporting_sessions, COUNT(DISTINCT project) AS active_projects, COUNT(DISTINCT local_date) AS active_days, COALESCE(ROUND(AVG(message_count), 1), 0) AS avg_messages, COALESCE(( SELECT CAST(FLOOR(AVG(message_count)) AS INTEGER) FROM ranked WHERE rn IN ( CAST(FLOOR((n + 1) / 2.0) AS BIGINT), CAST(FLOOR((n + 2) / 2.0) AS BIGINT) ) ), 0) AS median_messages, COALESCE(( SELECT message_count FROM ranked WHERE rn = LEAST(CAST(FLOOR(n * 0.9) AS BIGINT) + 1, n) LIMIT 1 ), 0) AS p90_messages, COALESCE(( SELECT project FROM project_totals ORDER BY messages DESC, project ASC LIMIT 1 ), '') AS most_active, COALESCE(ROUND(( SELECT SUM(messages) FROM ( SELECT messages FROM project_totals ORDER BY messages DESC LIMIT 3 ) top_projects )::DOUBLE / NULLIF(SUM(message_count), 0), 3), 0) AS concentration FROM filtered` rows, err := s.queryContext(ctx, query, queryArgs...) if err != nil { return db.AnalyticsSummary{}, fmt.Errorf("querying duckdb analytics summary: %w", err) } resp := db.AnalyticsSummary{Agents: map[string]*db.AgentSummary{}} if !rows.Next() { rows.Close() return resp, nil } if err := rows.Scan( &resp.TotalSessions, &resp.TotalMessages, &resp.TotalOutputTokens, &resp.TokenReportingSessions, &resp.ActiveProjects, &resp.ActiveDays, &resp.AvgMessages, &resp.MedianMessages, &resp.P90Messages, &resp.MostActive, &resp.Concentration, ); err != nil { rows.Close() return db.AnalyticsSummary{}, fmt.Errorf("scanning duckdb analytics summary: %w", err) } if err := rows.Err(); err != nil { rows.Close() return db.AnalyticsSummary{}, fmt.Errorf("iterating duckdb analytics summary: %w", err) } if err := rows.Close(); err != nil { return db.AnalyticsSummary{}, fmt.Errorf("closing duckdb analytics summary rows: %w", err) } agentRows, err := s.queryContext(ctx, ` WITH filtered AS ( SELECT s.agent, s.message_count FROM sessions s WHERE `+where+` ) SELECT agent, COUNT(*), COALESCE(SUM(message_count), 0) FROM filtered GROUP BY agent`, args..., ) if err != nil { return db.AnalyticsSummary{}, fmt.Errorf("querying duckdb analytics summary agents: %w", err) } defer agentRows.Close() for agentRows.Next() { var agent string var summary db.AgentSummary if err := agentRows.Scan(&agent, &summary.Sessions, &summary.Messages); err != nil { return db.AnalyticsSummary{}, fmt.Errorf("scanning duckdb analytics summary agent: %w", err) } resp.Agents[agent] = &summary } if err := agentRows.Err(); err != nil { return db.AnalyticsSummary{}, fmt.Errorf("iterating duckdb analytics summary agents: %w", err) } sessions, err := s.analyticsSessions(ctx, f) if err != nil { return db.AnalyticsSummary{}, err } sessionIDs := make([]string, 0, len(sessions)) for _, sess := range sessions { sessionIDs = append(sessionIDs, sess.id) } var models []string if f.HasTimeFilter() { models, err = s.getAnalyticsModelsForSessionIDsFiltered( ctx, sessionIDs, f, ) } else { models, err = s.getAnalyticsModelsForSessionIDs( ctx, sessionIDs, ) } if err != nil { return db.AnalyticsSummary{}, err } resp.Models = models return resp, nil } func (s *Store) getAnalyticsFilteredToolCallCounts( ctx context.Context, sessionIDs []string, f db.AnalyticsFilter, ) (map[string]int, error) { counts := make(map[string]int, len(sessionIDs)) if len(sessionIDs) == 0 || strings.TrimSpace(f.Model) == "" { return counts, nil } allowedModels := make(map[string]struct{}) for _, model := range duckAnalyticsCSVValues(f.Model) { allowedModels[model] = struct{}{} } loc := analyticsLocation(f.Timezone) err := duckQueryChunked(sessionIDs, func(chunk []string) error { ph, args := duckInPlaceholders(chunk) rows, err := s.queryContext(ctx, ` SELECT tc.session_id, m.model, m.timestamp, COUNT(*) FROM tool_calls tc JOIN messages m ON m.session_id = tc.session_id AND m.id = tc.message_id WHERE tc.session_id IN `+ph+` GROUP BY tc.session_id, m.model, m.timestamp`, args...) if err != nil { return fmt.Errorf( "querying duckdb filtered analytics tool calls: %w", err, ) } defer rows.Close() for rows.Next() { var sessionID, model string var ts any var count int if err := rows.Scan(&sessionID, &model, &ts, &count); err != nil { return fmt.Errorf( "scanning duckdb filtered analytics tool calls: %w", err, ) } if _, ok := allowedModels[model]; !ok { continue } if f.HasTimeFilter() { t, ok := parseAnalyticsTime(formatDBTime(ts)) if !ok || !duckAnalyticsTimeMatches(t.In(loc), f) { continue } } counts[sessionID] += count } return rows.Err() }) if err != nil { return nil, err } return counts, nil } func (s *Store) getAnalyticsActivityFilteredByModelTime( ctx context.Context, f db.AnalyticsFilter, granularity string, ) (db.ActivityResponse, error) { sessions, err := s.analyticsSessions(ctx, f) if err != nil { return db.ActivityResponse{}, err } sessionIDs := make([]string, 0, len(sessions)) for _, session := range sessions { sessionIDs = append(sessionIDs, session.id) } messageStats, err := s.getAnalyticsFilteredMessageStats( ctx, sessionIDs, f, ) if err != nil { return db.ActivityResponse{}, err } toolCounts, err := s.getAnalyticsFilteredToolCallCounts( ctx, sessionIDs, f, ) if err != nil { return db.ActivityResponse{}, err } out := db.ActivityResponse{Granularity: granularity} buckets := map[string]*db.ActivityEntry{} for _, session := range sessions { date := bucketAnalyticsDate( analyticsLocalDate(analyticsDateTime(session), f.Timezone), granularity, ) entry := buckets[date] if entry == nil { entry = &db.ActivityEntry{ Date: date, ByAgent: map[string]int{}, } buckets[date] = entry } entry.Sessions++ stat := messageStats[session.id] entry.Messages += stat.Messages entry.UserMessages += stat.UserMessages entry.AssistantMessages += stat.AssistantMessages entry.ThinkingMessages += stat.ThinkingMessages entry.ToolCalls += toolCounts[session.id] entry.ByAgent[session.agent] += stat.Messages } for _, key := range sortedKeys(buckets) { entry := buckets[key] if entry == nil { continue } out.Series = append(out.Series, *entry) } return out, nil } func (s *Store) GetAnalyticsActivity( ctx context.Context, f db.AnalyticsFilter, granularity string, ) (db.ActivityResponse, error) { if granularity == "" { granularity = "day" } if strings.TrimSpace(f.Model) != "" { return s.getAnalyticsActivityFilteredByModelTime( ctx, f, granularity, ) } buckets, err := s.queryActivityBuckets(ctx, f, granularity) if err != nil { return db.ActivityResponse{}, err } if err := s.addActivityAgentCounts(ctx, f, granularity, buckets); err != nil { return db.ActivityResponse{}, err } out := db.ActivityResponse{Granularity: granularity} keys := sortedKeys(buckets) for _, key := range keys { entry, ok := buckets[key] if !ok || entry == nil { continue } out.Series = append(out.Series, *entry) } return out, nil } func (s *Store) queryActivityBuckets( ctx context.Context, f db.AnalyticsFilter, granularity string, ) (map[string]*db.ActivityEntry, error) { where, args := duckBuildAnalyticsWhere( f, "COALESCE(s.started_at, s.created_at)", "s.", true, true) localDate, localDateArgs := duckAnalyticsLocalDateExpr( "COALESCE(s.started_at, s.created_at)", f) bucketExpr := duckAnalyticsBucketExpr("local_date", granularity) queryArgs := append([]any{}, localDateArgs...) queryArgs = append(queryArgs, args...) if _, modelArgs := duckAnalyticsCSVPredicate("m.model", f.Model); len(modelArgs) > 0 { queryArgs = append(queryArgs, modelArgs...) queryArgs = append(queryArgs, modelArgs...) } rows, err := s.queryContext(ctx, ` WITH filtered_sessions AS ( SELECT s.id, s.message_count, `+localDate+` AS local_date FROM sessions s WHERE `+where+` ), session_rows AS ( SELECT `+bucketExpr+` AS bucket, COUNT(*) AS sessions FROM filtered_sessions GROUP BY bucket ), message_rows AS ( SELECT `+bucketExpr+` AS bucket, COUNT(*) AS messages, COUNT(*) FILTER (WHERE m.role = 'user' AND m.is_system = FALSE) AS user_messages, COUNT(*) FILTER (WHERE m.role = 'assistant') AS assistant_messages, COUNT(*) FILTER (WHERE m.has_thinking = TRUE) AS thinking_messages FROM filtered_sessions fs JOIN messages m ON m.session_id = fs.id `+duckAnalyticsMessageFilterClause("m.model", f.Model)+` GROUP BY bucket ), tool_rows AS ( SELECT `+bucketExpr+` AS bucket, COUNT(*) AS tool_calls FROM filtered_sessions fs JOIN tool_calls tc ON tc.session_id = fs.id `+duckAnalyticsToolMessageJoin("tc", f.Model)+` `+duckAnalyticsMessageFilterClause("m.model", f.Model)+` GROUP BY bucket ) SELECT COALESCE(sr.bucket, mr.bucket, tr.bucket) AS bucket, COALESCE(sr.sessions, 0) AS sessions, COALESCE(mr.messages, 0) AS messages, COALESCE(mr.user_messages, 0) AS user_messages, COALESCE(mr.assistant_messages, 0) AS assistant_messages, COALESCE(mr.thinking_messages, 0) AS thinking_messages, COALESCE(tr.tool_calls, 0) AS tool_calls FROM session_rows sr FULL OUTER JOIN message_rows mr USING (bucket) FULL OUTER JOIN tool_rows tr ON tr.bucket = COALESCE(sr.bucket, mr.bucket) ORDER BY bucket`, queryArgs..., ) if err != nil { return nil, fmt.Errorf("querying duckdb analytics activity buckets: %w", err) } defer rows.Close() buckets := map[string]*db.ActivityEntry{} for rows.Next() { entry := db.ActivityEntry{ByAgent: map[string]int{}} if err := rows.Scan( &entry.Date, &entry.Sessions, &entry.Messages, &entry.UserMessages, &entry.AssistantMessages, &entry.ThinkingMessages, &entry.ToolCalls, ); err != nil { return nil, fmt.Errorf("scanning duckdb analytics activity bucket: %w", err) } buckets[entry.Date] = &entry } if err := rows.Err(); err != nil { return nil, fmt.Errorf("iterating duckdb analytics activity buckets: %w", err) } return buckets, nil } func (s *Store) addActivityAgentCounts( ctx context.Context, f db.AnalyticsFilter, granularity string, buckets map[string]*db.ActivityEntry, ) error { where, args := duckBuildAnalyticsWhere( f, "COALESCE(s.started_at, s.created_at)", "s.", true, true) localDate, localDateArgs := duckAnalyticsLocalDateExpr( "COALESCE(s.started_at, s.created_at)", f) bucketExpr := duckAnalyticsBucketExpr("local_date", granularity) queryArgs := append([]any{}, localDateArgs...) queryArgs = append(queryArgs, args...) if _, modelArgs := duckAnalyticsCSVPredicate("m.model", f.Model); len(modelArgs) > 0 { queryArgs = append(queryArgs, modelArgs...) } rows, err := s.queryContext(ctx, ` WITH filtered_sessions AS ( SELECT s.id, s.agent, `+localDate+` AS local_date FROM sessions s WHERE `+where+` ) SELECT `+bucketExpr+` AS bucket, fs.agent, COUNT(*) AS messages FROM filtered_sessions fs JOIN messages m ON m.session_id = fs.id `+duckAnalyticsMessageFilterClause("m.model", f.Model)+` GROUP BY bucket, fs.agent ORDER BY bucket, fs.agent`, queryArgs..., ) if err != nil { return fmt.Errorf("querying duckdb analytics activity agents: %w", err) } defer rows.Close() for rows.Next() { var bucket, agent string var count int if err := rows.Scan(&bucket, &agent, &count); err != nil { return fmt.Errorf("scanning duckdb analytics activity agent: %w", err) } if entry, ok := buckets[bucket]; ok { entry.ByAgent[agent] = count } } if err := rows.Err(); err != nil { return fmt.Errorf("iterating duckdb analytics activity agents: %w", err) } return nil } func bucketAnalyticsDate(date, granularity string) string { t, err := time.Parse("2006-01-02", date) if err != nil { return date } switch granularity { case "week": dow := int(t.Weekday()) if dow == 0 { dow = 7 } return t.AddDate(0, 0, -(dow - 1)).Format("2006-01-02") case "month": return t.Format("2006-01") + "-01" default: return date } } func duckAnalyticsBucketExpr(dateExpr, granularity string) string { switch granularity { case "week": return "strftime(date_trunc('week', CAST(" + dateExpr + " AS DATE)), '%Y-%m-%d')" case "month": return "strftime(date_trunc('month', CAST(" + dateExpr + " AS DATE)), '%Y-%m-%d')" default: return dateExpr } } func sortedKeys[V any](m map[string]V) []string { keys := make([]string, 0, len(m)) for k := range m { keys = append(keys, k) } sort.Strings(keys) return keys } func duckAnalyticsMessageFilterClause(col, raw string) string { pred, _ := duckAnalyticsCSVPredicate(col, raw) if pred == "" { return "" } return "WHERE " + pred } func duckAnalyticsAndClause(pred string) string { if pred == "" { return "" } return " AND " + pred } func duckAnalyticsToolMessageJoin( toolAlias string, model string, ) string { if model == "" { return "" } return ` JOIN messages m ON m.session_id = ` + toolAlias + `.session_id AND m.id = ` + toolAlias + `.message_id` } func (s *Store) GetAnalyticsHeatmap( ctx context.Context, f db.AnalyticsFilter, metric string, ) (db.HeatmapResponse, error) { if metric == "" { metric = "messages" } if strings.TrimSpace(f.Model) != "" && (metric == "messages" || metric == "output_tokens" || metric == "sessions") { sessions, err := s.analyticsSessionsWithModelMessageCounts(ctx, f) if err != nil { return db.HeatmapResponse{}, err } counts := map[string]int{} for _, session := range sessions { date := analyticsLocalDate(analyticsDateTime(session), f.Timezone) switch metric { case "sessions": counts[date]++ case "output_tokens": if session.hasTotalOutputTokens { counts[date] += session.totalOutputTokens } default: counts[date] += session.messageCount } } entriesFrom := duckClampHeatmapFrom(f.From, f.To) values := []int{} for date, v := range counts { if v > 0 && date >= entriesFrom && date <= f.To { values = append(values, v) } } sort.Ints(values) levels := duckComputeHeatmapLevels(values) entries := duckBuildHeatmapEntries(entriesFrom, f.To, counts, levels) if metric == "output_tokens" && len(counts) == 0 { return db.HeatmapResponse{ Metric: metric, EntriesFrom: entriesFrom, }, nil } return db.HeatmapResponse{ Metric: metric, Entries: entries, Levels: levels, EntriesFrom: entriesFrom, }, nil } where, args := duckBuildAnalyticsWhere( f, "COALESCE(s.started_at, s.created_at)", "s.", true, true) localDate, localDateArgs := duckAnalyticsLocalDateExpr( "COALESCE(s.started_at, s.created_at)", f) valueExpr := "COALESCE(SUM(s.message_count), 0)" switch metric { case "sessions": valueExpr = "COUNT(*)" case "output_tokens": where += " AND s.has_total_output_tokens = TRUE" valueExpr = "COALESCE(SUM(s.total_output_tokens), 0)" } queryArgs := append([]any{}, localDateArgs...) queryArgs = append(queryArgs, args...) rows, err := s.queryContext(ctx, ` SELECT `+localDate+` AS local_date, `+valueExpr+` AS value FROM sessions s WHERE `+where+` GROUP BY local_date ORDER BY local_date`, queryArgs..., ) if err != nil { return db.HeatmapResponse{}, fmt.Errorf("querying duckdb analytics heatmap: %w", err) } defer rows.Close() counts := map[string]int{} for rows.Next() { var date string var value int if err := rows.Scan(&date, &value); err != nil { return db.HeatmapResponse{}, fmt.Errorf("scanning duckdb analytics heatmap: %w", err) } counts[date] = value } if err := rows.Err(); err != nil { return db.HeatmapResponse{}, fmt.Errorf("iterating duckdb analytics heatmap: %w", err) } if metric == "output_tokens" && len(counts) == 0 { return db.HeatmapResponse{ Metric: metric, EntriesFrom: duckClampHeatmapFrom(f.From, f.To), }, nil } entriesFrom := duckClampHeatmapFrom(f.From, f.To) values := []int{} for date, v := range counts { if v > 0 && date >= entriesFrom && date <= f.To { values = append(values, v) } } sort.Ints(values) levels := duckComputeHeatmapLevels(values) entries := duckBuildHeatmapEntries(entriesFrom, f.To, counts, levels) return db.HeatmapResponse{ Metric: metric, Entries: entries, Levels: levels, EntriesFrom: entriesFrom, }, nil } const duckMaxHeatmapDays = 366 func duckClampHeatmapFrom(from, to string) string { start, err := time.Parse("2006-01-02", from) if err != nil { return from } end, err := time.Parse("2006-01-02", to) if err != nil { return from } earliest := end.AddDate(0, 0, -(duckMaxHeatmapDays - 1)) if start.Before(earliest) { return earliest.Format("2006-01-02") } return from } func duckComputeHeatmapLevels(sorted []int) db.HeatmapLevels { if len(sorted) == 0 { return db.HeatmapLevels{L1: 1, L2: 2, L3: 3, L4: 4} } n := len(sorted) return db.HeatmapLevels{ L1: sorted[0], L2: sorted[n/4], L3: sorted[n/2], L4: sorted[n*3/4], } } func duckHeatmapLevel(value int, levels db.HeatmapLevels) int { if value <= 0 { return 0 } if value <= levels.L2 { return 1 } if value <= levels.L3 { return 2 } if value <= levels.L4 { return 3 } return 4 } func duckBuildHeatmapEntries( from, to string, values map[string]int, levels db.HeatmapLevels, ) []db.HeatmapEntry { start, err := time.Parse("2006-01-02", from) if err != nil { return nil } end, err := time.Parse("2006-01-02", to) if err != nil { return nil } entries := []db.HeatmapEntry{} for d := start; !d.After(end); d = d.AddDate(0, 0, 1) { date := d.Format("2006-01-02") v := values[date] entries = append(entries, db.HeatmapEntry{ Date: date, Value: v, Level: duckHeatmapLevel(v, levels), }) } return entries } func (s *Store) GetAnalyticsProjects( ctx context.Context, f db.AnalyticsFilter, ) (db.ProjectsAnalyticsResponse, error) { // Per-project aggregate: count subagent sessions (mirrors SQLite). f.IncludeSubagents = true sessions, err := s.analyticsSessionsWithModelMessageCounts(ctx, f) if err != nil { return db.ProjectsAnalyticsResponse{}, err } type acc struct { row db.ProjectAnalytics counts []int days map[string]int } byProject := map[string]*acc{} for _, r := range sessions { a := byProject[r.project] if a == nil { a = &acc{ row: db.ProjectAnalytics{Name: r.project, Agents: map[string]int{}}, days: map[string]int{}, } byProject[r.project] = a } date := analyticsLocalDate(analyticsDateTime(r), f.Timezone) if a.row.FirstSession == "" || date < a.row.FirstSession { a.row.FirstSession = date } if date > a.row.LastSession { a.row.LastSession = date } a.row.Sessions++ a.row.Messages += r.messageCount a.row.Agents[r.agent]++ a.counts = append(a.counts, r.messageCount) a.days[date] += r.messageCount } resp := db.ProjectsAnalyticsResponse{} for _, name := range sortedKeys(byProject) { a, ok := byProject[name] if !ok || a == nil { continue } sort.Ints(a.counts) a.row.AvgMessages = round1(float64(a.row.Messages) / float64(a.row.Sessions)) a.row.MedianMessages = median(a.counts) if len(a.days) > 0 { a.row.DailyTrend = round1(float64(a.row.Messages) / float64(len(a.days))) } resp.Projects = append(resp.Projects, a.row) } sort.Slice(resp.Projects, func(i, j int) bool { if resp.Projects[i].Messages != resp.Projects[j].Messages { return resp.Projects[i].Messages > resp.Projects[j].Messages } return resp.Projects[i].Name < resp.Projects[j].Name }) return resp, nil } func (s *Store) GetAnalyticsHourOfWeek( ctx context.Context, f db.AnalyticsFilter, ) (db.HourOfWeekResponse, error) { if strings.TrimSpace(f.Model) != "" { return s.getAnalyticsHourOfWeekFilteredByModel(ctx, f) } sessionFilter := f sessionFilter.DayOfWeek = nil sessionFilter.Hour = nil where, args := duckBuildAnalyticsWhere( sessionFilter, "COALESCE(s.started_at, s.created_at)", "s.", true, false) localTime, localTimeArgs := duckAnalyticsLocalTimeExpr("m.timestamp", f) queryArgs := append([]any{}, args...) queryArgs = append(queryArgs, localTimeArgs...) rows, err := s.queryContext(ctx, ` WITH filtered_sessions AS ( SELECT s.id FROM sessions s WHERE `+where+` ), message_times AS ( SELECT `+localTime+` AS local_ts FROM messages m JOIN filtered_sessions fs ON fs.id = m.session_id WHERE m.timestamp IS NOT NULL ), message_buckets AS ( SELECT ((CAST(strftime(local_ts, '%w') AS INTEGER) + 6) % 7) AS day_of_week, CAST(strftime(local_ts, '%H') AS INTEGER) AS hour FROM message_times WHERE local_ts IS NOT NULL ) SELECT day_of_week, hour, COUNT(*) FROM message_buckets GROUP BY day_of_week, hour ORDER BY day_of_week, hour`, queryArgs..., ) if err != nil { return db.HourOfWeekResponse{}, fmt.Errorf("querying duckdb analytics hour-of-week: %w", err) } defer rows.Close() var grid [7][24]int for rows.Next() { var day, hour, messages int if err := rows.Scan(&day, &hour, &messages); err != nil { return db.HourOfWeekResponse{}, fmt.Errorf("scanning duckdb analytics hour-of-week: %w", err) } grid[day][hour] = messages } if err := rows.Err(); err != nil { return db.HourOfWeekResponse{}, fmt.Errorf("iterating duckdb analytics hour-of-week: %w", err) } return db.HourOfWeekResponseFromGrid(grid), nil } // getAnalyticsHourOfWeekFilteredByModel buckets model-scoped messages by // day-of-week and hour. It pairs empty-model user turns with their // selected-model assistant via the shared scope reducer, so those turns appear // in the heatmap consistently with the summary, activity, velocity, and trends // panels. The heatmap is the control that sets the day/hour filter, so it // clears DayOfWeek/Hour before scoping to keep showing the full grid, matching // the no-model path. func (s *Store) getAnalyticsHourOfWeekFilteredByModel( ctx context.Context, f db.AnalyticsFilter, ) (db.HourOfWeekResponse, error) { sessions, err := s.analyticsSessionsFiltered(ctx, f, true, false) if err != nil { return db.HourOfWeekResponse{}, err } sessionIDs := make([]string, 0, len(sessions)) for _, session := range sessions { sessionIDs = append(sessionIDs, session.id) } scopeFilter := f scopeFilter.DayOfWeek = nil scopeFilter.Hour = nil scope, err := s.resolveAnalyticsMessageScope( ctx, sessionIDs, scopeFilter, false, ) if err != nil { return db.HourOfWeekResponse{}, err } var grid [7][24]int if scope != nil { for _, msgs := range scope.MessagesBySession() { for _, m := range msgs { if !m.HasLocalTime { continue } dow := (int(m.LocalTime.Weekday()) + 6) % 7 grid[dow][m.LocalTime.Hour()]++ } } } return db.HourOfWeekResponseFromGrid(grid), nil } func (s *Store) GetAnalyticsSessionShape( ctx context.Context, f db.AnalyticsFilter, ) (db.SessionShapeResponse, error) { sessions, err := s.analyticsSessions(ctx, f) if err != nil { return db.SessionShapeResponse{}, err } modelFilter := strings.TrimSpace(f.Model) != "" lengths := map[string]int{} durations := map[string]int{} ids := []string{} for _, r := range sessions { ids = append(ids, r.id) if !modelFilter { lengths[lengthBucket(r.messageCount)]++ } if start, okS := parseAnalyticsTime(r.startedAt); okS { if end, okE := parseAnalyticsTime(r.endedAt); okE && !end.Before(start) { durations[durationBucket(end.Sub(start).Minutes())]++ } } } autonomy := map[string]int{} if modelFilter && len(ids) > 0 { stats, err := s.getAnalyticsFilteredMessageStats(ctx, ids, f) if err != nil { return db.SessionShapeResponse{}, err } lengths = map[string]int{} for _, r := range sessions { stat := stats[r.id] lengths[lengthBucket(stat.Messages)]++ if stat.UserMessages > 0 { ratio := float64(stat.ToolUseMessages) / float64(stat.UserMessages) autonomy[autonomyBucket(ratio)]++ } } } else { autonomy, err = s.analyticsAutonomyBuckets(ctx, ids) if err != nil { return db.SessionShapeResponse{}, err } } return db.SessionShapeResponse{ Count: len(sessions), LengthDistribution: mapBuckets(lengths, lengthOrder()), DurationDistribution: mapBuckets(durations, durationOrder()), AutonomyDistribution: mapBuckets(autonomy, autonomyOrder()), }, nil } func lengthBucket(mc int) string { switch { case mc <= 5: return "1-5" case mc <= 15: return "6-15" case mc <= 30: return "16-30" case mc <= 60: return "31-60" case mc <= 120: return "61-120" default: return "121+" } } func durationBucket(mins float64) string { switch { case mins < 5: return "<5m" case mins < 15: return "5-15m" case mins < 30: return "15-30m" case mins < 60: return "30-60m" case mins < 120: return "1-2h" default: return "2h+" } } func lengthOrder() map[string]int { return map[string]int{"1-5": 0, "6-15": 1, "16-30": 2, "31-60": 3, "61-120": 4, "121+": 5} } func durationOrder() map[string]int { return map[string]int{"<5m": 0, "5-15m": 1, "15-30m": 2, "30-60m": 3, "1-2h": 4, "2h+": 5} } func autonomyBucket(ratio float64) string { switch { case ratio < 0.5: return "<0.5" case ratio < 1: return "0.5-1" case ratio < 2: return "1-2" case ratio < 5: return "2-5" case ratio < 10: return "5-10" default: return "10+" } } func autonomyOrder() map[string]int { return map[string]int{"<0.5": 0, "0.5-1": 1, "1-2": 2, "2-5": 3, "5-10": 4, "10+": 5} } func mapBuckets(values map[string]int, order map[string]int) []db.DistributionBucket { out := make([]db.DistributionBucket, 0, len(values)) for label, count := range values { out = append(out, db.DistributionBucket{Label: label, Count: count}) } sort.Slice(out, func(i, j int) bool { return order[out[i].Label] < order[out[j].Label] }) return out } func (s *Store) analyticsAutonomyBuckets( ctx context.Context, sessionIDs []string, ) (map[string]int, error) { counts := map[string]int{} if len(sessionIDs) == 0 { return counts, nil } args := make([]any, len(sessionIDs)) placeholders := make([]string, len(sessionIDs)) for i, id := range sessionIDs { args[i] = id placeholders[i] = "?" } rows, err := s.queryContext(ctx, ` SELECT session_id, SUM(CASE WHEN role = 'user' AND is_system = FALSE THEN 1 ELSE 0 END) AS user_count, SUM(CASE WHEN role = 'assistant' AND has_tool_use = TRUE THEN 1 ELSE 0 END) AS tool_count FROM messages WHERE session_id IN (`+strings.Join(placeholders, ",")+`) GROUP BY session_id`, args..., ) if err != nil { return nil, fmt.Errorf("querying duckdb autonomy: %w", err) } defer rows.Close() for rows.Next() { var sessionID string var userCount, toolCount int if err := rows.Scan(&sessionID, &userCount, &toolCount); err != nil { return nil, fmt.Errorf("scanning duckdb autonomy: %w", err) } if userCount > 0 { counts[autonomyBucket(float64(toolCount)/float64(userCount))]++ } } return counts, rows.Err() } // duckMaxSQLVars bounds the IN-list size per query to stay well under // driver bind-variable limits; larger ID sets are split into chunks. const duckMaxSQLVars = 900 func duckInPlaceholders(ids []string) (string, []any) { ph := make([]string, len(ids)) args := make([]any, len(ids)) for i, id := range ids { ph[i] = "?" args[i] = id } return "(" + strings.Join(ph, ",") + ")", args } func duckQueryChunked(ids []string, fn func(chunk []string) error) error { for i := 0; i < len(ids); i += duckMaxSQLVars { end := min(i+duckMaxSQLVars, len(ids)) if err := fn(ids[i:end]); err != nil { return err } } return nil } func (s *Store) GetAnalyticsTools( ctx context.Context, f db.AnalyticsFilter, ) (db.ToolsAnalyticsResponse, error) { sessions, err := s.analyticsSessionsFiltered( ctx, f, false, false, ) if err != nil { return db.ToolsAnalyticsResponse{}, err } meta := map[string]duckAnalyticsSession{} var ids []string for _, r := range sessions { meta[r.id] = r ids = append(ids, r.id) } if len(ids) == 0 { return db.BuildToolsAnalytics(nil), nil } var toolRows []db.ToolAnalyticsRow err = duckQueryChunked(ids, func(chunk []string) error { ph, args := duckInPlaceholders(chunk) modelPred, modelArgs := duckAnalyticsCSVPredicate("m.model", f.Model) args = append(args, modelArgs...) query := `SELECT tc.session_id, tc.category, TRIM(COALESCE(tc.tool_name, '')), COUNT(*), m.timestamp FROM tool_calls tc LEFT JOIN messages m ON m.session_id = tc.session_id AND m.id = tc.message_id WHERE tc.session_id IN ` + ph if modelPred != "" { query += ` AND ` + modelPred } query += ` GROUP BY tc.session_id, tc.category, TRIM(COALESCE(tc.tool_name, '')), m.timestamp` rows, qErr := s.queryContext(ctx, query, args...) if qErr != nil { return qErr } defer rows.Close() for rows.Next() { var sid, cat, toolName string var ts any var count int if err := rows.Scan(&sid, &cat, &toolName, &count, &ts); err != nil { return err } r, ok := meta[sid] if !ok { continue } _, date, keep := f.ResolveSkillRowTime( formatDBTime(ts), analyticsDateTime(r), ) if !keep { continue } toolRows = append(toolRows, db.ToolAnalyticsRow{ SessionID: sid, Category: cat, ToolName: toolName, Agent: r.agent, Count: count, Date: date, }) } return rows.Err() }) if err != nil { return db.ToolsAnalyticsResponse{}, err } return db.BuildToolsAnalytics(toolRows), nil } // GetAnalyticsSkills returns skill usage analytics. granularity picks // the trend bucket size (day, week, or month); empty defaults to week. func (s *Store) GetAnalyticsSkills( ctx context.Context, f db.AnalyticsFilter, granularity string, ) (db.SkillsAnalyticsResponse, error) { sessions, err := s.analyticsSessionsFiltered(ctx, f, false, false) if err != nil { return db.SkillsAnalyticsResponse{}, err } meta := map[string]duckAnalyticsSession{} var ids []string for _, r := range sessions { meta[r.id] = r ids = append(ids, r.id) } if len(ids) == 0 { return db.BuildSkillsAnalytics( nil, f.From, f.To, granularity, ), nil } var skillRows []db.SkillAnalyticsRow err = duckQueryChunked(ids, func(chunk []string) error { ph, args := duckInPlaceholders(chunk) modelPred, modelArgs := duckAnalyticsCSVPredicate("m.model", f.Model) args = append(args, modelArgs...) rows, qErr := s.queryContext(ctx, `SELECT tc.session_id, TRIM(COALESCE(tc.skill_name, '')), COUNT(*), m.timestamp FROM tool_calls tc LEFT JOIN messages m ON m.session_id = tc.session_id AND m.id = tc.message_id WHERE tc.session_id IN `+ph+` AND TRIM(COALESCE(tc.skill_name, '')) != '' `+duckAnalyticsAndClause(modelPred)+` GROUP BY tc.session_id, TRIM(COALESCE(tc.skill_name, '')), m.timestamp`, args...) if qErr != nil { return qErr } defer rows.Close() for rows.Next() { var sid, skill string var count int var msgTS any if err := rows.Scan(&sid, &skill, &count, &msgTS); err != nil { return err } r, ok := meta[sid] if !ok { continue } usedTS, date, keep := f.ResolveSkillRowTime( formatDBTime(msgTS), analyticsDateTime(r), ) if !keep { continue } skillRows = append(skillRows, db.SkillAnalyticsRow{ SessionID: sid, SkillName: skill, Agent: r.agent, Project: r.project, Date: date, LastUsedAt: usedTS, Count: count, }) } return rows.Err() }) if err != nil { return db.SkillsAnalyticsResponse{}, err } return db.BuildSkillsAnalytics( skillRows, f.From, f.To, granularity, ), nil } func (s *Store) GetAnalyticsVelocity( ctx context.Context, f db.AnalyticsFilter, ) (db.VelocityResponse, error) { sessions, err := s.analyticsSessions(ctx, f) if err != nil { return db.VelocityResponse{}, err } if len(sessions) == 0 { return db.VelocityResponse{ ByAgent: []db.VelocityBreakdown{}, ByComplexity: []db.VelocityBreakdown{}, }, nil } sessionIDs := make([]string, 0, len(sessions)) sessionInfo := make(map[string]duckVelocitySession, len(sessions)) for _, sess := range sessions { sessionIDs = append(sessionIDs, sess.id) sessionInfo[sess.id] = duckVelocitySession{ agent: sess.agent, mc: sess.messageCount, } } if strings.TrimSpace(f.Model) != "" { stats, err := s.getAnalyticsFilteredMessageStats( ctx, sessionIDs, f, ) if err != nil { return db.VelocityResponse{}, err } for _, sid := range sessionIDs { info := sessionInfo[sid] info.mc = stats[sid].Messages sessionInfo[sid] = info } } var sessionMsgs map[string][]duckVelocityMsg if strings.TrimSpace(f.Model) != "" { sessionMsgs, err = s.filteredVelocityMessages( ctx, sessionIDs, f, ) } else { sessionMsgs, err = s.velocityMessages( ctx, sessionIDs, analyticsLocation(f.Timezone), ) } if err != nil { return db.VelocityResponse{}, err } var toolCounts map[string]int if strings.TrimSpace(f.Model) != "" { toolCounts, err = s.getAnalyticsFilteredToolCallCounts( ctx, sessionIDs, f, ) } else { toolCounts, err = s.velocityToolCounts(ctx, sessionIDs) } if err != nil { return db.VelocityResponse{}, err } overall := &duckVelocityAccumulator{} byAgent := make(map[string]*duckVelocityAccumulator) byComplexity := make(map[string]*duckVelocityAccumulator) for _, sid := range sessionIDs { msgs := sessionMsgs[sid] if len(msgs) < 2 { continue } info := sessionInfo[sid] agentKey := info.agent compKey := duckComplexityBucket(info.mc) if byAgent[agentKey] == nil { byAgent[agentKey] = &duckVelocityAccumulator{} } if byComplexity[compKey] == nil { byComplexity[compKey] = &duckVelocityAccumulator{} } processDuckSessionVelocity( []*duckVelocityAccumulator{overall, byAgent[agentKey], byComplexity[compKey]}, msgs, toolCounts[sid], ) } resp := db.VelocityResponse{ Overall: overall.computeOverview(), ByAgent: []db.VelocityBreakdown{}, ByComplexity: []db.VelocityBreakdown{}, } for _, key := range sortedKeys(byAgent) { acc := byAgent[key] if acc == nil { continue } resp.ByAgent = append(resp.ByAgent, db.VelocityBreakdown{ Label: key, Sessions: acc.sessions, Overview: acc.computeOverview(), }) } compOrder := map[string]int{"1-15": 0, "16-60": 1, "61+": 2} compKeys := sortedKeys(byComplexity) sort.Slice(compKeys, func(i, j int) bool { return compOrder[compKeys[i]] < compOrder[compKeys[j]] }) for _, key := range compKeys { acc := byComplexity[key] if acc == nil { continue } resp.ByComplexity = append(resp.ByComplexity, db.VelocityBreakdown{ Label: key, Sessions: acc.sessions, Overview: acc.computeOverview(), }) } return resp, nil } type duckVelocitySession struct { agent string mc int } type duckVelocityMsg struct { role string ts time.Time valid bool contentLength int } type duckVelocityAccumulator struct { turnCycles []float64 firstResponses []float64 totalMsgs int totalChars int totalToolCalls int activeMinutes float64 sessions int } func (s *Store) velocityMessages( ctx context.Context, sessionIDs []string, loc *time.Location, ) (map[string][]duckVelocityMsg, error) { out := make(map[string][]duckVelocityMsg, len(sessionIDs)) if len(sessionIDs) == 0 { return out, nil } args, placeholders := stringInArgs(sessionIDs) rows, err := s.queryContext(ctx, ` SELECT session_id, ordinal, role, timestamp, content_length FROM messages WHERE session_id IN (`+strings.Join(placeholders, ",")+`) ORDER BY session_id, ordinal`, args..., ) if err != nil { return nil, fmt.Errorf("querying duckdb velocity messages: %w", err) } defer rows.Close() for rows.Next() { var sid, role string var ordinal int var ts any var contentLength int if err := rows.Scan(&sid, &ordinal, &role, &ts, &contentLength); err != nil { return nil, fmt.Errorf("scanning duckdb velocity message: %w", err) } parsed, ok := duckLocalTime(formatDBTime(ts), loc) out[sid] = append(out[sid], duckVelocityMsg{ role: role, ts: parsed, valid: ok, contentLength: contentLength, }) } return out, rows.Err() } func (s *Store) filteredVelocityMessages( ctx context.Context, sessionIDs []string, f db.AnalyticsFilter, ) (map[string][]duckVelocityMsg, error) { out := make(map[string][]duckVelocityMsg, len(sessionIDs)) if len(sessionIDs) == 0 { return out, nil } scope, err := s.resolveAnalyticsMessageScope(ctx, sessionIDs, f, false) if err != nil { return nil, err } if scope == nil { return out, nil } for sessionID, rows := range scope.TimingBySession() { for _, row := range rows { out[sessionID] = append(out[sessionID], duckVelocityMsg{ role: row.Role, ts: row.Time, valid: row.Valid, contentLength: row.ContentLength, }) } } return out, nil } func (s *Store) velocityToolCounts( ctx context.Context, sessionIDs []string, ) (map[string]int, error) { out := make(map[string]int, len(sessionIDs)) if len(sessionIDs) == 0 { return out, nil } args, placeholders := stringInArgs(sessionIDs) rows, err := s.queryContext(ctx, ` SELECT session_id, COUNT(*) FROM tool_calls WHERE session_id IN (`+strings.Join(placeholders, ",")+`) GROUP BY session_id`, args..., ) if err != nil { return nil, fmt.Errorf("querying duckdb velocity tool calls: %w", err) } defer rows.Close() for rows.Next() { var sid string var count int if err := rows.Scan(&sid, &count); err != nil { return nil, fmt.Errorf("scanning duckdb velocity tool call count: %w", err) } out[sid] = count } return out, rows.Err() } func stringInArgs(values []string) ([]any, []string) { args := make([]any, len(values)) placeholders := make([]string, len(values)) for i, value := range values { args[i] = value placeholders[i] = "?" } return args, placeholders } func duckLocalTime(ts string, loc *time.Location) (time.Time, bool) { t, ok := parseAnalyticsTime(ts) if !ok { return time.Time{}, false } return t.In(loc), true } func duckComplexityBucket(mc int) string { switch { case mc <= 15: return "1-15" case mc <= 60: return "16-60" default: return "61+" } } func processDuckSessionVelocity( accums []*duckVelocityAccumulator, msgs []duckVelocityMsg, toolCount int, ) { const maxCycleSec = 1800.0 // Shared with the Top Sessions "active duration" SQL so the two // "active" definitions stay in lockstep. const maxGapSec = db.ActiveGapCapSec for _, acc := range accums { acc.sessions++ } for i := 1; i < len(msgs); i++ { prev := msgs[i-1] cur := msgs[i] if !prev.valid || !cur.valid { continue } if prev.role == "user" && cur.role == "assistant" { delta := cur.ts.Sub(prev.ts).Seconds() if delta > 0 && delta <= maxCycleSec { for _, acc := range accums { acc.turnCycles = append(acc.turnCycles, delta) } } } } var firstUser, firstAsst *duckVelocityMsg firstUserIdx := -1 for i := range msgs { if msgs[i].role == "user" && msgs[i].valid { firstUser = &msgs[i] firstUserIdx = i break } } if firstUserIdx >= 0 { for i := firstUserIdx + 1; i < len(msgs); i++ { if msgs[i].role == "assistant" && msgs[i].valid { firstAsst = &msgs[i] break } } } if firstUser != nil && firstAsst != nil { delta := firstAsst.ts.Sub(firstUser.ts).Seconds() if delta < 0 { delta = 0 } for _, acc := range accums { acc.firstResponses = append(acc.firstResponses, delta) } } activeSec := 0.0 assistantChars := 0 for i, msg := range msgs { if msg.role == "assistant" { assistantChars += msg.contentLength } if i > 0 && msgs[i-1].valid && msg.valid { gap := msg.ts.Sub(msgs[i-1].ts).Seconds() if gap > 0 { if gap > maxGapSec { gap = maxGapSec } activeSec += gap } } } activeMinutes := activeSec / 60 if activeMinutes > 0 { for _, acc := range accums { acc.totalMsgs += len(msgs) acc.totalChars += assistantChars acc.totalToolCalls += toolCount acc.activeMinutes += activeMinutes } } } func (a *duckVelocityAccumulator) computeOverview() db.VelocityOverview { sort.Float64s(a.turnCycles) sort.Float64s(a.firstResponses) out := db.VelocityOverview{} out.TurnCycleSec = db.Percentiles{ P50: round1(percentileFloat(a.turnCycles, 0.5)), P90: round1(percentileFloat(a.turnCycles, 0.9)), } out.FirstResponseSec = db.Percentiles{ P50: round1(percentileFloat(a.firstResponses, 0.5)), P90: round1(percentileFloat(a.firstResponses, 0.9)), } if a.activeMinutes > 0 { out.MsgsPerActiveMin = round1(float64(a.totalMsgs) / a.activeMinutes) out.CharsPerActiveMin = round1(float64(a.totalChars) / a.activeMinutes) out.ToolCallsPerActiveMin = round1(float64(a.totalToolCalls) / a.activeMinutes) } return out } func percentileFloat(values []float64, p float64) float64 { if len(values) == 0 { return 0 } idx := int(float64(len(values)) * p) if idx >= len(values) { idx = len(values) - 1 } return round1(values[idx]) } func (s *Store) GetAnalyticsTopSessions( ctx context.Context, f db.AnalyticsFilter, metric string, ) (db.TopSessionsResponse, error) { switch metric { case "", "messages": metric = "messages" case "duration", "output_tokens": default: metric = "messages" } if strings.TrimSpace(f.Model) != "" && (metric == "messages" || metric == "output_tokens") { sessions, err := s.analyticsSessionsWithModelMessageCounts(ctx, f) if err != nil { return db.TopSessionsResponse{}, err } sort.SliceStable(sessions, func(i, j int) bool { if metric == "output_tokens" { if sessions[i].totalOutputTokens != sessions[j].totalOutputTokens { return sessions[i].totalOutputTokens > sessions[j].totalOutputTokens } } else if sessions[i].messageCount != sessions[j].messageCount { return sessions[i].messageCount > sessions[j].messageCount } return sessions[i].id < sessions[j].id }) out := db.TopSessionsResponse{Metric: metric} for i := range sessions { if metric == "output_tokens" && !sessions[i].hasTotalOutputTokens { continue } if len(out.Sessions) >= 10 { break } startedAt := sessions[i].startedAt endedAt := sessions[i].endedAt out.Sessions = append(out.Sessions, db.TopSession{ ID: sessions[i].id, Project: sessions[i].project, FirstMessage: sessions[i].firstMessage, DisplayName: sessions[i].displayName, MessageCount: sessions[i].messageCount, OutputTokens: sessions[i].totalOutputTokens, DurationMin: duckSessionDurationMinutes(sessions[i]), StartedAt: &startedAt, EndedAt: &endedAt, TerminationStatus: sessions[i].terminationStatus, }) } return out, nil } includeTime := true var pairedSet map[string]bool if f.HasTimeFilter() && strings.TrimSpace(f.Model) != "" { // Filter the scoped session set in Go rather than binding every // paired ID into one IN (...) predicate, which would exceed the // driver bind-variable cap for large result sets. Mirrors the // SQLite/PostgreSQL top-sessions Go path under a model filter: load // the model+date candidates, then keep only the paired sessions and // limit in Go. The in-SQL ORDER BY still ranks them by the metric. paired, err := s.analyticsSessionsModelTimeFiltered(ctx, f, true) if err != nil { return db.TopSessionsResponse{}, err } if len(paired) == 0 { return db.TopSessionsResponse{Metric: metric}, nil } pairedSet = make(map[string]bool, len(paired)) for _, session := range paired { pairedSet[session.id] = true } includeTime = false } where, args := duckBuildAnalyticsWhere( f, "COALESCE(s.started_at, s.created_at)", "s.", true, includeTime) durationExpr := "(epoch(s.ended_at) - epoch(s.started_at)) / 60.0" durationSelectExpr := "COALESCE(" + durationExpr + ", 0)" activeDurationExpr := fmt.Sprintf(` ( SELECT COALESCE(SUM( CASE WHEN inner2.delta_ms <= 0 THEN 0 WHEN inner2.delta_ms > %[1]d THEN %[1]d ELSE inner2.delta_ms END), 0) / 60000.0 FROM ( SELECT CAST( ROUND(epoch( LEAD(m2.timestamp) OVER (ORDER BY m2.ordinal) - m2.timestamp ) * 1000) AS BIGINT ) AS delta_ms FROM messages m2 WHERE m2.session_id = s.id ) inner2 )`, db.ActiveGapCapMs) activeDurationSelectExpr := "COALESCE(" + activeDurationExpr + ", 0)" orderExpr := "s.message_count DESC, s.id ASC" switch metric { case "duration": where += " AND s.started_at IS NOT NULL AND s.ended_at IS NOT NULL AND s.ended_at >= s.started_at" orderExpr = activeDurationSelectExpr + " DESC, s.id ASC" case "output_tokens": where += " AND s.has_total_output_tokens = TRUE" orderExpr = "s.total_output_tokens DESC, s.id ASC" } // When filtering the scoped set in Go (model+time), drop the SQL LIMIT so // the paired sessions aren't truncated before the Go filter; the in-SQL // ORDER BY keeps them ranked and the top 10 is taken after filtering. limitClause := "\n\t\tLIMIT 10" if pairedSet != nil { limitClause = "" } query := ` SELECT s.id, s.project, s.first_message, s.message_count, s.total_output_tokens, ` + durationSelectExpr + ` AS duration_min, ` + activeDurationSelectExpr + ` AS active_duration_min, s.started_at, s.ended_at, s.termination_status FROM sessions s WHERE ` + where + ` ORDER BY ` + orderExpr + limitClause rows, err := s.queryContext(ctx, query, args...) if err != nil { return db.TopSessionsResponse{}, fmt.Errorf("querying duckdb analytics top sessions: %w", err) } defer rows.Close() out := db.TopSessionsResponse{Metric: metric} for rows.Next() { var row db.TopSession var startedRaw, endedRaw any if err := rows.Scan( &row.ID, &row.Project, &row.FirstMessage, &row.MessageCount, &row.OutputTokens, &row.DurationMin, &row.ActiveDurationMin, &startedRaw, &endedRaw, &row.TerminationStatus, ); err != nil { return db.TopSessionsResponse{}, fmt.Errorf("scanning duckdb analytics top session: %w", err) } if pairedSet != nil && !pairedSet[row.ID] { continue } startedAt := formatDBTime(startedRaw) endedAt := formatDBTime(endedRaw) row.StartedAt = &startedAt row.EndedAt = &endedAt row.DurationMin = round1(row.DurationMin) row.ActiveDurationMin = round1(row.ActiveDurationMin) out.Sessions = append(out.Sessions, row) if pairedSet != nil && len(out.Sessions) >= 10 { break } } if err := rows.Err(); err != nil { return db.TopSessionsResponse{}, fmt.Errorf("iterating duckdb analytics top sessions: %w", err) } return out, nil } func duckSessionDurationMinutes(session duckAnalyticsSession) float64 { startedAt, okStart := parseAnalyticsTime(session.startedAt) endedAt, okEnd := parseAnalyticsTime(session.endedAt) if !okStart || !okEnd || endedAt.Before(startedAt) { return 0 } return round1(endedAt.Sub(startedAt).Minutes()) } // GetAnalyticsSignals returns aggregated session signal data. Signals stay // session-scoped under a model filter (totals are session-level aggregates // over sessions that used the model, not re-attributed per model); see the // SQLite GetAnalyticsSignals for the rationale. func (s *Store) GetAnalyticsSignals( ctx context.Context, f db.AnalyticsFilter, ) (db.SignalsAnalyticsResponse, error) { sessions, err := s.analyticsSessions(ctx, f) if err != nil { return db.SignalsAnalyticsResponse{}, err } rows := duckSignalRowsFromSessions(sessions, f) if err := s.duckPopulateFrustrationMarkers(ctx, rows); err != nil { return db.SignalsAnalyticsResponse{}, err } return db.AggregateSignals(rows), nil } func (s *Store) GetAnalyticsSignalSessions( ctx context.Context, f db.AnalyticsFilter, signal string, limit int, ) (db.SignalSessionsResponse, error) { if !db.IsSupportedAnalyticsSignal(signal) { return db.SignalSessionsResponse{}, db.ErrUnsupportedAnalyticsSignal } if limit <= 0 || limit > 20 { limit = 10 } sessions, err := s.analyticsSessions(ctx, f) if err != nil { return db.SignalSessionsResponse{}, err } rows := duckSignalRowsFromSessions(sessions, f) if err := s.duckPopulateFrustrationMarkers(ctx, rows); err != nil { return db.SignalSessionsResponse{}, err } candidates := db.SignalCandidates(rows, signal, limit) messages, err := s.duckSignalMessages(ctx, candidates, f) if err != nil { return db.SignalSessionsResponse{}, err } return db.SignalSessionsResponse{ Signal: signal, Sessions: db.BuildSignalExamples(candidates, messages, signal), }, nil } func duckSignalRowsFromSessions( sessions []duckAnalyticsSession, f db.AnalyticsFilter, ) []db.SignalRow { rows := make([]db.SignalRow, 0, len(sessions)) for _, r := range sessions { rows = append(rows, db.SignalRow{ ID: r.id, Agent: r.agent, Project: r.project, FirstMessage: r.firstMessage, IsAutomated: r.isAutomated, Date: analyticsLocalDate(analyticsDateTime(r), f.Timezone), HealthScore: r.healthScore, HealthGrade: r.healthGrade, Outcome: r.outcome, OutcomeConfidence: r.outcomeConfidence, ToolFailureSignalCount: r.toolFailures, ToolRetryCount: r.toolRetries, EditChurnCount: r.editChurn, CompactionCount: r.compactions, MidTaskCompactionCount: r.midTaskCompactions, ContextPressureMax: r.contextPressureMax, QualitySignalVersion: r.qualitySignalVersion, ShortPromptCount: r.shortPromptCount, UnstructuredStart: r.unstructuredStart, MissingSuccessCriteriaCount: r.missingSuccessCriteriaCount, MissingVerificationCount: r.missingVerificationCount, DuplicatePromptCount: r.duplicatePromptCount, NoCodeContextCount: r.noCodeContextCount, RunawayToolLoopCount: r.runawayToolLoopCount, FrustrationMarkerCount: r.frustrationMarkerCount, }) } return rows } func (s *Store) duckPopulateFrustrationMarkers( ctx context.Context, rows []db.SignalRow, ) error { if len(rows) == 0 { return nil } idx := make(map[string]int, len(rows)) placeholders := make([]string, len(rows)) args := make([]any, len(rows)) for i := range rows { idx[rows[i].ID] = i placeholders[i] = "?" args[i] = rows[i].ID } q := `SELECT session_id, content, is_system FROM messages WHERE role = 'user' AND session_id IN (` + strings.Join(placeholders, ",") + `)` msgRows, err := s.queryContext(ctx, q, args...) if err != nil { return fmt.Errorf("querying duckdb frustration markers: %w", err) } defer msgRows.Close() for msgRows.Next() { var sessionID, content string var isSystem bool if err := msgRows.Scan( &sessionID, &content, &isSystem, ); err != nil { return fmt.Errorf("scanning duckdb frustration marker: %w", err) } i, ok := idx[sessionID] if !ok || isSystem { continue } if signals.IsFrustrationMarker(content) { rows[i].FrustrationMarkerCount++ } } if err := msgRows.Err(); err != nil { return fmt.Errorf("iterating duckdb frustration markers: %w", err) } return nil } func (s *Store) duckSignalMessages( ctx context.Context, rows []db.SignalRow, f db.AnalyticsFilter, ) (map[string][]db.SignalMessage, error) { out := make(map[string][]db.SignalMessage, len(rows)) if len(rows) == 0 { return out, nil } if strings.TrimSpace(f.Model) != "" { ids := make([]string, 0, len(rows)) for _, r := range rows { ids = append(ids, r.ID) } scope, err := s.resolveAnalyticsMessageScope(ctx, ids, f, true) if err != nil { return nil, err } if scope != nil { for sessionID, scopedRows := range scope.MessagesBySession() { for _, row := range scopedRows { out[sessionID] = append(out[sessionID], db.SignalMessage{ SessionID: row.SessionID, Ordinal: row.Ordinal, Role: row.Role, Content: row.Content, Timestamp: row.Timestamp, IsSystem: row.IsSystem, HasToolUse: row.HasToolUse, }) } } } return out, nil } placeholders := make([]string, len(rows)) args := make([]any, len(rows)) for i, r := range rows { placeholders[i] = "?" args[i] = r.ID } filterModels := duckAnalyticsCSVValues(f.Model) q := `SELECT session_id, ordinal, role, content, timestamp, is_system, has_tool_use FROM messages WHERE session_id IN (` + strings.Join(placeholders, ",") + `)` if len(filterModels) == 1 { q += ` AND model = ?` args = append(args, filterModels[0]) } else if len(filterModels) > 1 { modelPlaceholders := make([]string, len(filterModels)) for i, model := range filterModels { modelPlaceholders[i] = "?" args = append(args, model) } q += ` AND model IN (` + strings.Join(modelPlaceholders, ",") + `)` } q += ` ORDER BY session_id, ordinal` msgRows, err := s.queryContext(ctx, q, args...) if err != nil { return nil, fmt.Errorf("querying duckdb signal messages: %w", err) } defer msgRows.Close() for msgRows.Next() { var m db.SignalMessage var ts any if err := msgRows.Scan( &m.SessionID, &m.Ordinal, &m.Role, &m.Content, &ts, &m.IsSystem, &m.HasToolUse, ); err != nil { return nil, fmt.Errorf("scanning duckdb signal message: %w", err) } m.Timestamp = formatDBTime(ts) out[m.SessionID] = append(out[m.SessionID], m) } if err := msgRows.Err(); err != nil { return nil, fmt.Errorf("iterating duckdb signal messages: %w", err) } return out, nil } func (s *Store) GetTrendsTerms( ctx context.Context, f db.AnalyticsFilter, terms []db.TrendTermInput, granularity string, ) (db.TrendsTermsResponse, error) { if granularity == "" { granularity = "week" } buckets := db.TrendBucketRange(f.From, f.To, granularity) index := map[string]int{} for i, bucket := range buckets { index[bucket.Date] = i } counts := make([][]int, len(terms)) for i := range counts { counts[i] = make([]int, len(buckets)) } messageCounts := make([]int, len(buckets)) sessionFilter := f sessionFilter.From = "" sessionFilter.To = "" sessionFilter.Model = "" sessionFilter.DayOfWeek = nil sessionFilter.Hour = nil sessions, err := s.analyticsSessions(ctx, sessionFilter) if err != nil { return db.TrendsTermsResponse{}, err } allowedSessions := make(map[string]bool, len(sessions)) for _, sess := range sessions { allowedSessions[sess.id] = true } if len(allowedSessions) == 0 { return db.BuildTrendsTermsResponse( f.From, f.To, granularity, buckets, terms, counts, messageCounts, ), nil } loc := analyticsLocation(f.Timezone) flt := messageScopeFilter(f) modelFiltering := len(flt.Models) > 0 trendLocal := func(msgTS, startedAt, createdAt any) (time.Time, bool) { ts := firstNonEmpty(formatDBTime(msgTS), formatDBTime(startedAt), formatDBTime(createdAt)) t, ok := parseAnalyticsTime(ts) if !ok { return time.Time{}, false } return t.In(loc), true } rows, err := s.queryContext(ctx, ` SELECT m.session_id, m.ordinal, m.role, m.is_system, COALESCE(m.model, ''), m.content, m.timestamp, s.started_at, s.created_at FROM messages m JOIN sessions s ON s.id = m.session_id WHERE s.deleted_at IS NULL AND m.role IN ('user', 'assistant') AND m.is_system = FALSE AND `+db.DuckDBSystemPrefixSQL("m.content", "m.role")+` ORDER BY m.session_id, m.ordinal`) if err != nil { return db.TrendsTermsResponse{}, err } defer rows.Close() type trendRow struct { sessionID string role string isSystem bool model string content string msgTS any startedAt any createdAt any } processRow := func(sessionID, content string, local time.Time) { if !allowedSessions[sessionID] { return } date := local.Format("2006-01-02") if f.From != "" && date < f.From { return } if f.To != "" && date > f.To { return } bucket := bucketAnalyticsDate(date, granularity) pos, ok := index[bucket] if !ok { return } messageCounts[pos]++ for i, term := range terms { counts[i][pos] += db.CountTrendOccurrences(content, term) } } emit := func(m db.ScopedMessage) { if !m.HasLocalTime { return } processRow(m.SessionID, m.Content, m.LocalTime) } reducer := db.NewScopeReducer(flt, emit) for rows.Next() { var row trendRow var ordinal int if err := rows.Scan(&row.sessionID, &ordinal, &row.role, &row.isSystem, &row.model, &row.content, &row.msgTS, &row.startedAt, &row.createdAt); err != nil { return db.TrendsTermsResponse{}, err } local, has := trendLocal(row.msgTS, row.startedAt, row.createdAt) if !modelFiltering { if has && flt.MatchesDayHour(local, true) { processRow(row.sessionID, row.content, local) } continue } if err := reducer.Push(db.MessageInput{ SessionID: row.sessionID, Ordinal: ordinal, Role: row.role, Model: row.model, IsSystem: row.isSystem, LocalTime: local, HasLocalTime: has, Content: row.content, }); err != nil { return db.TrendsTermsResponse{}, err } } if err := rows.Err(); err != nil { return db.TrendsTermsResponse{}, err } return db.BuildTrendsTermsResponse( f.From, f.To, granularity, buckets, terms, counts, messageCounts, ), nil } type duckRates struct { input float64 output float64 cacheCreation float64 cacheRead float64 updatedAt *time.Time source export.PricingRowSource } func (s *Store) loadPricing(ctx context.Context) (map[string]duckRates, error) { rows, err := s.queryContext(ctx, ` SELECT model_pattern, input_per_mtok, output_per_mtok, cache_creation_per_mtok, cache_read_per_mtok, updated_at FROM model_pricing`) if err != nil { return nil, err } defer rows.Close() out := map[string]duckRates{} count := 0 for rows.Next() { var model string var rates duckRates var updatedAt string if err := rows.Scan( &model, &rates.input, &rates.output, &rates.cacheCreation, &rates.cacheRead, &updatedAt, ); err != nil { return nil, err } if strings.HasPrefix(model, "_") { continue } if parsed, err := time.Parse(time.RFC3339Nano, updatedAt); err == nil { t := parsed.UTC() rates.updatedAt = &t } rates.source = duckPricingSource(model, rates) out[model] = rates count++ } if err := rows.Err(); err != nil { return nil, err } if count == 0 { out = duckFallbackPricingMap() } for model, custom := range s.customPricing { rates := duckRates{ input: custom.Input, output: custom.Output, cacheCreation: custom.CacheCreation, cacheRead: custom.CacheRead, } rates.source = duckCustomPricingSource() out[model] = rates } return out, nil } func duckCustomPricingSource() export.PricingRowSource { return export.PricingRowSourceCustom } func duckFallbackPricingMap() map[string]duckRates { prices := pricingpkg.FallbackPricing() out := make(map[string]duckRates, len(prices)) for _, p := range prices { if strings.HasPrefix(p.ModelPattern, "_") { continue } out[p.ModelPattern] = duckRates{ input: p.InputPerMTok, output: p.OutputPerMTok, cacheCreation: p.CacheCreationPerMTok, cacheRead: p.CacheReadPerMTok, source: export.PricingRowSourceEmbedded, } } return out } func duckPricingSource(model string, rates duckRates) export.PricingRowSource { fallback := duckFallbackPricingMap() if f, ok := fallback[model]; ok && f.input == rates.input && f.output == rates.output && f.cacheCreation == rates.cacheCreation && f.cacheRead == rates.cacheRead { return export.PricingRowSourceEmbedded } return export.PricingRowSourceFetched } func duckPricingRows( in map[string]duckRates, ) []export.EffectivePricingRow { out := make([]export.EffectivePricingRow, 0, len(in)) for pattern, rates := range in { source := rates.source if source == "" { source = duckPricingSource(pattern, rates) } out = append(out, export.EffectivePricingRow{ ModelPattern: pattern, Rates: export.ModelRates{ InputPerMTok: rates.input, OutputPerMTok: rates.output, CacheWritePerMTok: rates.cacheCreation, CacheReadPerMTok: rates.cacheRead, UpdatedAt: rates.updatedAt, Source: source, }, }) } return out } type duckUsageBounds struct { from string to string } func duckUsagePaddedUTCBound(ts string, hours int) string { t, err := time.Parse(time.RFC3339, ts) if err != nil { return ts } return t.Add(time.Duration(hours) * time.Hour).Format(time.RFC3339) } func duckUsageBoundsForFilter(f db.UsageFilter) duckUsageBounds { var b duckUsageBounds if f.From != "" { b.from = duckUsagePaddedUTCBound(f.From+"T00:00:00Z", -14) } if f.To != "" { b.to = duckUsagePaddedUTCBound(f.To+"T23:59:59Z", 14) } return b } func appendDuckUsageColumnBounds( where, col string, b duckUsageBounds, args []any, ) (string, []any) { if b.from != "" { where += "\n\t\t\tAND " + col + " >= CAST(? AS TIMESTAMP)" args = append(args, b.from) } if b.to != "" { where += "\n\t\t\tAND " + col + " <= CAST(? AS TIMESTAMP)" args = append(args, b.to) } return where, args } func appendDuckUsageCSVFilter( where string, args []any, col, csv string, include bool, ) (string, []any) { if csv == "" { return where, args } parts := strings.Split(csv, ",") vals := make([]string, 0, len(parts)) for _, value := range parts { trimmed := strings.TrimSpace(value) if trimmed != "" { vals = append(vals, trimmed) } } return appendDuckUsageValuesFilter(where, args, col, vals, include) } func appendDuckUsageValuesFilter( where string, args []any, col string, vals []string, include bool, ) (string, []any) { if len(vals) == 0 { return where, args } op := "IN" if !include { op = "NOT IN" } if len(vals) == 1 { if include { where += "\n\t\t\tAND " + col + " = ?" } else { where += "\n\t\t\tAND " + col + " != ?" } args = append(args, vals[0]) return where, args } ph := make([]string, len(vals)) for i, value := range vals { ph[i] = "?" args = append(args, value) } where += "\n\t\t\tAND " + col + " " + op + " (" + strings.Join(ph, ",") + ")" return where, args } func appendDuckUsageSourceFilterClauses( where string, args []any, modelCol string, f db.UsageFilter, ) (string, []any) { where, args = appendDuckUsageCSVFilter(where, args, modelCol, f.Model, true) return appendDuckUsageCSVFilter(where, args, modelCol, f.ExcludeModel, false) } func appendDuckUsageSessionFilterClauses( where string, args []any, f db.UsageFilter, sessionID string, ) (string, []any) { where, args = appendDuckUsageCSVFilter(where, args, "s.agent", f.Agent, true) where, args = appendDuckUsageValuesFilter( where, args, "s.project", f.ProjectFilterLabels(), true, ) where, args = appendDuckUsageCSVFilter(where, args, "s.machine", f.Machine, true) if f.GitBranch != "" { var clause string clause, args = db.BranchPairClauseArgs("s.project", "s.git_branch", f.GitBranch, args) where += "\n\t\t\tAND " + clause } where, args = appendDuckUsageValuesFilter( where, args, "s.project", f.ExcludedProjectFilterLabels(), false, ) where, args = appendDuckUsageCSVFilter(where, args, "s.agent", f.ExcludeAgent, false) if sessionID != "" { where += "\n\t\t\tAND s.id = ?" args = append(args, sessionID) } if f.MinUserMessages > 0 { where += "\n\t\t\tAND s.user_message_count >= ?" args = append(args, f.MinUserMessages) } scope := duckNormalizeAutomatedScope( f.AutomatedScope, f.ExcludeAutomated) if f.ExcludeOneShot { if scope == "human" { where += "\n\t\t\tAND s.user_message_count > 1" } else { where += "\n\t\t\tAND (s.user_message_count > 1 OR COALESCE(s.is_automated, FALSE) = TRUE)" } } if pred := duckAutomatedScopePredicate( scope, "COALESCE(s.is_automated, FALSE)"); pred != "" { where += "\n\t\t\tAND " + pred } if f.ActiveSince != "" { where += "\n\t\t\tAND COALESCE(s.ended_at, s.started_at, s.created_at) >= CAST(? AS TIMESTAMP)" args = append(args, f.ActiveSince) } if pred, predArgs := duckUsageTerminationPred(f.Termination); pred != "" { where += "\n\t\t\tAND " + pred args = append(args, predArgs...) } return where, args } func duckUsageTerminationPred(status string) (string, []any) { return duckAnalyticsTerminationPred( status, "COALESCE(s.ended_at, s.started_at, s.created_at)", "s.termination_status", ) } const duckDailyCursorUsageRowsSQLTemplate = ` SELECT '' AS session_id, NULL AS message_ordinal, 'cursor' AS source, cu.occurred_at AS ts, cu.model AS model, '' AS token_json, '' AS claude_message_id, '' AS claude_request_id, '' AS source_uuid, cu.dedup_key AS usage_dedup_key, cu.input_tokens AS input_tokens, cu.output_tokens AS output_tokens, cu.cache_write_tokens AS cache_create, cu.cache_read_tokens AS cache_read, 0 AS reasoning_tokens, cu.charged_cents / 100.0 AS cost_usd, '' AS project, 'cursor' AS agent, '' AS machine, 0 AS user_message_count, cu.is_headless AS is_automated, '' AS display_name, NULL AS started_at, cu.occurred_at AS activity_at FROM cursor_usage_events cu WHERE %s` const duckUsageMessageEligibility = ` m.token_usage != '' AND m.model != '' AND m.model != '' AND s.deleted_at IS NULL` // duckUsageMatchingMessageSourceEligibility is the message-only half of // duckUsageMessageEligibility with the token-presence requirement removed // and the model-presence requirement relaxed to a role check, for // GetUsageMatchingSessionCount. See the usageMatchingMessageEligibility // doc comment in internal/db. const duckUsageMatchingMessageSourceEligibility = ` m.role = 'assistant' AND m.model != ''` const duckUsageMatchingMessageEligibility = duckUsageMatchingMessageSourceEligibility + ` AND s.deleted_at IS NULL` const duckUsageEventSourceEligibility = ` ue.model != ''` const duckUsageEventEligibility = duckUsageEventSourceEligibility + ` AND s.deleted_at IS NULL` // duckUsageSourceWheres builds the message/event WHERE clauses shared by // duckUsageRawSQL and duckMatchingUsageRawSQL; the two callers differ only // in the message eligibility predicate. func duckUsageSourceWheres( f db.UsageFilter, sessionID, messageEligibility string, b duckUsageBounds, ) (string, []any, string, []any) { messageWhere := messageEligibility var messageArgs []any messageWhere, messageArgs = appendDuckUsageSourceFilterClauses( messageWhere, messageArgs, "m.model", f) messageWhere, messageArgs = appendDuckUsageSessionFilterClauses( messageWhere, messageArgs, f, sessionID) messageWhere, messageArgs = appendDuckUsageColumnBounds( messageWhere, "COALESCE(m.timestamp, s.started_at)", b, messageArgs) eventWhere := duckUsageEventEligibility var eventArgs []any eventWhere, eventArgs = appendDuckUsageSourceFilterClauses( eventWhere, eventArgs, "ue.model", f) eventWhere, eventArgs = appendDuckUsageSessionFilterClauses( eventWhere, eventArgs, f, sessionID) eventWhere, eventArgs = appendDuckUsageColumnBounds( eventWhere, "COALESCE(ue.occurred_at, s.started_at)", b, eventArgs) return messageWhere, messageArgs, eventWhere, eventArgs } func duckUsageRawSQL(f db.UsageFilter, sessionID string) (string, []any) { messageWhere, messageArgs, eventWhere, eventArgs := duckUsageSourceWheres( f, sessionID, duckUsageMessageEligibility, duckUsageBoundsForFilter(f)) query := fmt.Sprintf(` SELECT m.session_id AS session_id, m.ordinal AS message_ordinal, 'message' AS source, COALESCE(m.timestamp, s.started_at) AS ts, m.model AS model, m.token_usage AS token_json, m.claude_message_id AS claude_message_id, m.claude_request_id AS claude_request_id, m.source_uuid AS source_uuid, '' AS usage_dedup_key, 0 AS input_tokens, 0 AS output_tokens, 0 AS cache_create, 0 AS cache_read, COALESCE(TRY_CAST(json_extract_string(m.token_usage, '$.reasoning_tokens') AS BIGINT), 0) AS reasoning_tokens, NULL AS cost_usd, s.project AS project, s.agent AS agent, s.machine AS machine, s.user_message_count AS user_message_count, s.is_automated AS is_automated, COALESCE(s.display_name, s.session_name, s.first_message, s.project, s.id) AS display_name, s.started_at AS started_at, COALESCE(s.ended_at, s.started_at, s.created_at) AS activity_at FROM messages m JOIN sessions s ON s.id = m.session_id WHERE %s UNION ALL SELECT ue.session_id AS session_id, ue.message_ordinal AS message_ordinal, ue.source AS source, COALESCE(ue.occurred_at, s.started_at) AS ts, ue.model AS model, '' AS token_json, '' AS claude_message_id, '' AS claude_request_id, '' AS source_uuid, CASE WHEN ue.dedup_key != '' THEN ue.session_id || ':' || ue.source || ':' || ue.dedup_key ELSE ue.session_id || ':' || ue.source || ':id:' || CAST(ue.id AS VARCHAR) END AS usage_dedup_key, ue.input_tokens AS input_tokens, ue.output_tokens AS output_tokens, ue.cache_creation_input_tokens AS cache_create, ue.cache_read_input_tokens AS cache_read, ue.reasoning_tokens AS reasoning_tokens, ue.cost_usd AS cost_usd, s.project AS project, s.agent AS agent, s.machine AS machine, s.user_message_count AS user_message_count, s.is_automated AS is_automated, COALESCE(s.display_name, s.session_name, s.first_message, s.project, s.id) AS display_name, s.started_at AS started_at, COALESCE(s.ended_at, s.started_at, s.created_at) AS activity_at FROM usage_events ue JOIN sessions s ON s.id = ue.session_id WHERE %s`, messageWhere, eventWhere) args := make([]any, 0, len(messageArgs)+len(eventArgs)) args = append(args, messageArgs...) args = append(args, eventArgs...) return query, args } // duckMatchingUsageRawSQL builds the bounded-range row source for // GetUsageMatchingSessionCount. It shares duckUsageRawSQL's WHERE // assembly (via duckUsageSourceWheres) but relaxes the message predicate: // no token_usage requirement (Copilot messages never populate it) and no // model-presence requirement (some Copilot assistant messages parse // before a model name is known), scoping to assistant rows via m.role // instead. Model/ExcludeModel filters are still applied per-row, same as // duckUsageRawSQL. func duckMatchingUsageRawSQL(f db.UsageFilter) (string, []any) { messageWhere, messageArgs, eventWhere, eventArgs := duckUsageSourceWheres( f, "", duckUsageMatchingMessageEligibility, duckUsageBoundsForFilter(f)) query := fmt.Sprintf(` SELECT m.session_id AS session_id, COALESCE(m.timestamp, s.started_at) AS ts FROM messages m JOIN sessions s ON s.id = m.session_id WHERE %s UNION ALL SELECT ue.session_id AS session_id, COALESCE(ue.occurred_at, s.started_at) AS ts FROM usage_events ue JOIN sessions s ON s.id = ue.session_id WHERE %s`, messageWhere, eventWhere) args := make([]any, 0, len(messageArgs)+len(eventArgs)) args = append(args, messageArgs...) args = append(args, eventArgs...) return query, args } func duckCursorUsageRowsSQLForBounds( f db.UsageFilter, b duckUsageBounds, ) (string, []any, bool) { hasTermFilter := f.Termination != "" && f.Termination != "all" // Cursor usage rows carry no project or git branch and bypass the session // filter, so any filter they cannot satisfy (project, machine, branch) // must exclude them entirely rather than let them leak into totals. if len(f.ProjectFilterLabels()) > 0 || len(f.ExcludedProjectFilterLabels()) > 0 || f.Machine != "" || f.GitBranch != "" || f.MinUserMessages > 0 || f.ExcludeOneShot || hasTermFilter || f.ActiveSince != "" { return "", nil, false } if f.Agent != "" { vals := strings.Split(f.Agent, ",") for i := range vals { vals[i] = strings.TrimSpace(vals[i]) } if !slices.Contains(vals, "cursor") { return "", nil, false } } if f.ExcludeAgent != "" { vals := strings.Split(f.ExcludeAgent, ",") for i := range vals { vals[i] = strings.TrimSpace(vals[i]) } if slices.Contains(vals, "cursor") { return "", nil, false } } where := "cu.model != ''" var args []any scope := duckNormalizeAutomatedScope(f.AutomatedScope, f.ExcludeAutomated) if pred := duckAutomatedScopePredicate(scope, "cu.is_headless"); pred != "" { where += "\n\tAND " + pred } where, args = appendDuckUsageSourceFilterClauses( where, args, "cu.model", f, ) where, args = appendDuckUsageColumnBounds( where, "cu.occurred_at", b, args, ) return fmt.Sprintf(duckDailyCursorUsageRowsSQLTemplate, where), args, true } func duckDailyUsageRawSQL(f db.UsageFilter) (string, []any) { bounds := duckUsageBoundsForFilter(f) sessionRowsSQL, sessionArgs := duckUsageRawSQL(f, "") cursorRowsSQL, cursorArgs, ok := duckCursorUsageRowsSQLForBounds(f, bounds) if !ok { return sessionRowsSQL, sessionArgs } rowsSQL := sessionRowsSQL + "\n\t\tUNION ALL\n" + cursorRowsSQL args := make([]any, 0, len(sessionArgs)+len(cursorArgs)) args = append(args, sessionArgs...) args = append(args, cursorArgs...) return rowsSQL, args } func duckUsageLocalDateSQL(f db.UsageFilter) (string, any) { if f.Timezone != "" { return "COALESCE(strftime(timezone(?, timezone('UTC', ts)), '%Y-%m-%d'), '')", f.Timezone } ref := time.Now().UTC() if f.From != "" { if t, err := time.Parse(time.RFC3339, f.From+"T12:00:00Z"); err == nil { ref = t } } _, offset := ref.In(time.Local).Zone() return "COALESCE(strftime(ts + (? * INTERVAL 1 SECOND), '%Y-%m-%d'), '')", offset } func duckUsageCTE(f db.UsageFilter, sessionID string) (string, []any) { rawSQL, args := duckUsageRawSQL(f, sessionID) return duckUsageCTEFromRaw(f, rawSQL, args) } func duckDailyUsageCTE(f db.UsageFilter) (string, []any) { rawSQL, args := duckDailyUsageRawSQL(f) return duckUsageCTEFromRaw(f, rawSQL, args) } func duckUsageCTEFromRaw( f db.UsageFilter, rawSQL string, args []any, ) (string, []any) { localDateSQL, localDateArg := duckUsageLocalDateSQL(f) // Apply the local-date window BEFORE deduping so an out-of-range // duplicate (pulled in by the padded UTC bounds) cannot win // dedup_rank = 1 and suppress the in-range row. Mirrors the // dedup-after-date-filter order in internal/db/usage.go. datePred := "TRUE" var dateArgs []any if f.From != "" { datePred += " AND local_date >= ?" dateArgs = append(dateArgs, f.From) } if f.To != "" { datePred += " AND local_date <= ?" dateArgs = append(dateArgs, f.To) } query := fmt.Sprintf(` WITH usage_raw AS ( %[1]s ), usage_normalized AS ( SELECT *, CASE WHEN source = 'message' THEN LEAST(GREATEST(COALESCE(TRY_CAST(json_extract_string(token_json, '$.input_tokens') AS BIGINT), 0), 0), %[4]d) WHEN source = 'session' THEN GREATEST(input_tokens, 0) ELSE LEAST(GREATEST(input_tokens, 0), %[4]d) END AS input_tokens_norm, CASE WHEN source = 'message' THEN LEAST(GREATEST(COALESCE(TRY_CAST(json_extract_string(token_json, '$.output_tokens') AS BIGINT), 0), 0), %[4]d) WHEN source = 'session' THEN GREATEST(output_tokens, 0) ELSE LEAST(GREATEST(output_tokens, 0), %[4]d) END AS output_tokens_norm, CASE WHEN source = 'message' THEN LEAST(GREATEST(COALESCE(TRY_CAST(json_extract_string(token_json, '$.cache_creation_input_tokens') AS BIGINT), 0), 0), %[4]d) WHEN source = 'session' THEN GREATEST(cache_create, 0) ELSE LEAST(GREATEST(cache_create, 0), %[4]d) END AS cache_create_norm, CASE WHEN source = 'message' THEN LEAST(GREATEST(COALESCE(TRY_CAST(json_extract_string(token_json, '$.cache_read_input_tokens') AS BIGINT), 0), 0), %[4]d) WHEN source = 'session' THEN GREATEST(cache_read, 0) ELSE LEAST(GREATEST(cache_read, 0), %[4]d) END AS cache_read_norm, CASE WHEN source = 'message' THEN LEAST(GREATEST(COALESCE(TRY_CAST(json_extract_string(token_json, '$.reasoning_tokens') AS BIGINT), 0), 0), %[4]d) WHEN source = 'session' THEN GREATEST(reasoning_tokens, 0) ELSE LEAST(GREATEST(reasoning_tokens, 0), %[4]d) END AS reasoning_tokens_norm, CASE WHEN claude_message_id != '' AND claude_request_id != '' THEN 'claude:' || claude_message_id || ':' || claude_request_id WHEN source = 'message' AND agent != '' AND source_uuid != '' THEN 'source:' || agent || ':' || source_uuid WHEN usage_dedup_key != '' THEN 'usage:' || usage_dedup_key ELSE 'row:' || session_id || ':' || source || ':' || COALESCE(CAST(message_ordinal AS VARCHAR), '') || ':' || COALESCE(CAST(ts AS VARCHAR), '') || ':' || model END AS dedup_group, %[2]s AS local_date FROM usage_raw ), usage_windowed AS ( SELECT * FROM usage_normalized WHERE %[3]s ), usage_ranked AS ( SELECT *, ROW_NUMBER() OVER ( PARTITION BY dedup_group ORDER BY ts ASC, session_id ASC, COALESCE(message_ordinal, -1) ASC ) AS dedup_rank FROM usage_windowed ), usage_localized AS ( SELECT * FROM usage_ranked WHERE dedup_rank = 1 )`, rawSQL, localDateSQL, datePred, db.MaxPlausibleTokens) args = append(args, localDateArg) args = append(args, dateArgs...) return query, args } type duckUsageBucket struct { inputTok int outputTok int cacheCr int cacheRd int cost float64 } type duckUsageAggregateRow struct { date string sessionID string project string agent string model string displayName string startedAt string inputTok int outputTok int cacheCr int cacheRd int billableInput int // Output-rate billable tokens. SQL folds reasoning-only rows into this // value before grouping because reasoning is otherwise a row-level choice. billableOutput int billableReason int billableCacheCr int billableCacheRd int explicitCost float64 reportedCostRows int } type duckSessionUsageRow struct { sessionID string messageOrdinal sql.NullInt64 source string ts string model string inputTok int outputTok int cacheCr int cacheRd int reasoningTok int costUSD sql.NullFloat64 } func duckUsageAggregateCost( model string, inputTok, outputTok, cacheCr, cacheRd int, billableInput, billableOutput, billableReasoning, billableCacheCr, billableCacheRd int, explicitCost float64, hasReportedCost bool, pricing *export.PricingResolver, ) (float64, float64, bool, bool) { hasBillableTokens := billableInput != 0 || billableOutput != 0 || billableReasoning != 0 || billableCacheCr != 0 || billableCacheRd != 0 if !hasReportedCost && explicitCost == 0 && inputTok == 0 && outputTok == 0 && cacheCr == 0 && cacheRd == 0 && !hasBillableTokens { pricing.RecordComputed(model, pricing.Lookup(model)) return 0, 0, true, false } lookup := pricing.Lookup(model) rates := lookup.Rates cost := explicitCost + rates.CostForTokens( billableInput, billableOutput, billableReasoning, billableCacheCr, billableCacheRd) if hasReportedCost { pricing.RecordReported(model, lookup) } if hasBillableTokens { pricing.RecordComputed(model, lookup) } readDelta := float64(cacheRd) * (rates.InputPerMTok - rates.CacheReadPerMTok) / 1_000_000 createDelta := float64(cacheCr) * (rates.InputPerMTok - rates.CacheWritePerMTok) / 1_000_000 priced := lookup.OK if !hasBillableTokens && hasReportedCost { priced = true } return cost, readDelta + createDelta, priced, true } func duckSessionUsageRowCost( r duckSessionUsageRow, pricing map[string]duckRates, ) (float64, bool, bool) { if r.costUSD.Valid { return r.costUSD.Float64, true, true } if r.inputTok == 0 && r.outputTok == 0 && r.reasoningTok == 0 && r.cacheCr == 0 && r.cacheRd == 0 { return 0, true, false } rates, priced := pricingpkg.Resolve(pricing, r.model) if !priced { return 0, false, true } // Reasoning is a breakdown of output, not additional billable // output; reasoning-only rows bill at the output rate. Mirrors // export.ModelRates.CostForTokens and the aggregate SQL fold. billableOutput := r.outputTok if billableOutput == 0 { billableOutput = r.reasoningTok } cost := (float64(r.inputTok)*rates.input + float64(billableOutput)*rates.output + float64(r.cacheCr)*rates.cacheCreation + float64(r.cacheRd)*rates.cacheRead) / 1_000_000 return cost, true, true } func duckSessionUsageBreakdownEntry( r duckSessionUsageRow, ordinal int, cost float64, priced bool, ) db.SessionUsageBreakdownEntry { entry := db.SessionUsageBreakdownEntry{ Ordinal: ordinal, Source: r.source, Label: duckSessionUsageBreakdownLabel(r), Timestamp: r.ts, Model: r.model, InputTokens: r.inputTok, OutputTokens: r.outputTok, CacheCreationInputTokens: r.cacheCr, CacheReadInputTokens: r.cacheRd, CostUSD: cost, HasCost: priced, } if r.messageOrdinal.Valid { messageOrdinal := int(r.messageOrdinal.Int64) entry.MessageOrdinal = &messageOrdinal } return entry } func duckSessionUsageBreakdownLabel(r duckSessionUsageRow) string { if r.messageOrdinal.Valid { if r.source == "message" { return fmt.Sprintf("Prompt %d", r.messageOrdinal.Int64+1) } return fmt.Sprintf("Step %d", r.messageOrdinal.Int64+1) } if r.source != "" { return r.source } return "usage" } func (s *Store) dailyUsageAggregateRows( ctx context.Context, f db.UsageFilter, ) ([]duckUsageAggregateRow, error) { cte, args := duckDailyUsageCTE(f) query := cte + ` SELECT local_date, project, agent, model, SUM(input_tokens_norm) AS input_tokens, SUM(output_tokens_norm) AS output_tokens, SUM(cache_create_norm) AS cache_creation_tokens, SUM(cache_read_norm) AS cache_read_tokens, SUM(CASE WHEN cost_usd IS NULL THEN input_tokens_norm ELSE 0 END) AS billable_input_tokens, SUM(CASE WHEN cost_usd IS NOT NULL THEN 0 WHEN output_tokens_norm = 0 THEN reasoning_tokens_norm ELSE output_tokens_norm END) AS billable_output_tokens, CAST(0 AS BIGINT) AS billable_reasoning_tokens, SUM(CASE WHEN cost_usd IS NULL THEN cache_create_norm ELSE 0 END) AS billable_cache_creation_tokens, SUM(CASE WHEN cost_usd IS NULL THEN cache_read_norm ELSE 0 END) AS billable_cache_read_tokens, COALESCE(SUM(cost_usd), 0) AS explicit_cost, COUNT(cost_usd) AS reported_cost_rows FROM usage_localized GROUP BY local_date, project, agent, model ORDER BY local_date ASC, project ASC, agent ASC, model ASC` rows, err := s.queryContext(ctx, query, args...) if err != nil { return nil, fmt.Errorf("querying duckdb daily usage aggregates: %w", err) } defer rows.Close() var out []duckUsageAggregateRow for rows.Next() { var r duckUsageAggregateRow if err := rows.Scan( &r.date, &r.project, &r.agent, &r.model, &r.inputTok, &r.outputTok, &r.cacheCr, &r.cacheRd, &r.billableInput, &r.billableOutput, &r.billableReason, &r.billableCacheCr, &r.billableCacheRd, &r.explicitCost, &r.reportedCostRows, ); err != nil { return nil, fmt.Errorf("scanning duckdb daily usage aggregate: %w", err) } out = append(out, r) } return out, rows.Err() } func (s *Store) GetDailyUsage( ctx context.Context, f db.UsageFilter, ) (db.DailyUsageResult, error) { pricing, err := s.loadPricing(ctx) if err != nil { return db.DailyUsageResult{}, err } rateResolver := export.NewPricingResolver(duckPricingRows(pricing)) rows, err := s.dailyUsageAggregateRows(ctx, f) if err != nil { return db.DailyUsageResult{}, err } type usageAccumKey struct { date string project string agent string model string } accum := map[usageAccumKey]*duckUsageBucket{} projectLabels := map[string]bool{} totalSavings := 0.0 for _, r := range rows { key := usageAccumKey{date: r.date, project: r.project, agent: r.agent, model: r.model} if r.project != "" { projectLabels[r.project] = true } b := accum[key] if b == nil { b = &duckUsageBucket{} accum[key] = b } cost, savings, _, _ := duckUsageAggregateCost( r.model, r.inputTok, r.outputTok, r.cacheCr, r.cacheRd, r.billableInput, r.billableOutput, r.billableReason, r.billableCacheCr, r.billableCacheRd, r.explicitCost, r.reportedCostRows > 0, rateResolver, ) totalSavings += savings b.inputTok += r.inputTok b.outputTok += r.outputTok b.cacheCr += r.cacheCr b.cacheRd += r.cacheRd b.cost += cost } type dayMaps struct { models map[string]duckUsageBucket projects map[string]duckUsageBucket agents map[string]duckUsageBucket } days := map[string]*dayMaps{} for key, b := range accum { day := days[key.date] if day == nil { day = &dayMaps{ models: map[string]duckUsageBucket{}, projects: map[string]duckUsageBucket{}, agents: map[string]duckUsageBucket{}, } days[key.date] = day } addUsageBucket(day.models, key.model, *b) if f.Breakdowns { addUsageBucket(day.projects, key.project, *b) addUsageBucket(day.agents, key.agent, *b) } } var result db.DailyUsageResult for _, date := range sortedKeys(days) { day := days[date] if day == nil { continue } entry := db.DailyUsageEntry{Date: date} modelNames := sortedUsageBucketKeys(day.models) entry.ModelsUsed = modelNames for _, model := range modelNames { b := day.models[model] entry.InputTokens += b.inputTok entry.OutputTokens += b.outputTok entry.CacheCreationTokens += b.cacheCr entry.CacheReadTokens += b.cacheRd entry.TotalCost += b.cost entry.ModelBreakdowns = append(entry.ModelBreakdowns, db.ModelBreakdown{ ModelName: model, InputTokens: b.inputTok, OutputTokens: b.outputTok, CacheCreationTokens: b.cacheCr, CacheReadTokens: b.cacheRd, Cost: roundCost(b.cost), }) } if f.Breakdowns { for _, project := range sortedUsageBucketKeys(day.projects) { b := day.projects[project] entry.ProjectBreakdowns = append(entry.ProjectBreakdowns, db.ProjectBreakdown{ Project: project, InputTokens: b.inputTok, OutputTokens: b.outputTok, CacheCreationTokens: b.cacheCr, CacheReadTokens: b.cacheRd, Cost: roundCost(b.cost), }) } for _, agent := range sortedUsageBucketKeys(day.agents) { b := day.agents[agent] entry.AgentBreakdowns = append(entry.AgentBreakdowns, db.AgentBreakdown{ Agent: agent, InputTokens: b.inputTok, OutputTokens: b.outputTok, CacheCreationTokens: b.cacheCr, CacheReadTokens: b.cacheRd, Cost: roundCost(b.cost), }) } } entry.TotalCost = roundCost(entry.TotalCost) result.Daily = append(result.Daily, entry) result.Totals.InputTokens += entry.InputTokens result.Totals.OutputTokens += entry.OutputTokens result.Totals.CacheCreationTokens += entry.CacheCreationTokens result.Totals.CacheReadTokens += entry.CacheReadTokens result.Totals.TotalCost += entry.TotalCost } result.Totals.CacheSavings = roundCost(totalSavings) result.Totals.TotalCost = roundCost(result.Totals.TotalCost) var aiCredits float64 for key, b := range accum { aiCredits += db.AICreditsFromCost(key.agent, b.cost) } if aiCredits > 0 { result.Totals.CopilotAICredits = aiCredits } if result.Daily == nil { result.Daily = []db.DailyUsageEntry{} } result.SchemaVersion = export.UsageDailySchemaVersion pricingBlock, err := rateResolver.BuildBlock() if err != nil { return db.DailyUsageResult{}, fmt.Errorf( "building pricing block: %w", err) } result.Pricing = &pricingBlock projects, err := s.BuildProjectIdentityMap(ctx, sortedBoolKeys(projectLabels)) if err != nil { return db.DailyUsageResult{}, err } result.Projects = export.ProjectMapForWire(projects) if !f.SkipSessionCounts { counts, err := s.GetUsageSessionCounts(ctx, f) if err != nil { return db.DailyUsageResult{}, err } result.SessionCounts = counts } db.SanitizeDailyUsageProjectLabelsWithCatalog(&result, projects) return result, nil } func addUsageBucket(m map[string]duckUsageBucket, key string, b duckUsageBucket) { cur := m[key] cur.inputTok += b.inputTok cur.outputTok += b.outputTok cur.cacheCr += b.cacheCr cur.cacheRd += b.cacheRd cur.cost += b.cost m[key] = cur } func sortedUsageBucketKeys(m map[string]duckUsageBucket) []string { out := make([]string, 0, len(m)) for key := range m { out = append(out, key) } sort.Slice(out, func(i, j int) bool { left := m[out[i]] right := m[out[j]] if left.cost != right.cost { return left.cost > right.cost } return out[i] < out[j] }) return out } func sortedBoolKeys(m map[string]bool) []string { out := make([]string, 0, len(m)) for k := range m { out = append(out, k) } sort.Strings(out) return out } func roundCost(v float64) float64 { return math.Round(v*1_000_000) / 1_000_000 } func (s *Store) sessionUsageAggregateRows( ctx context.Context, f db.UsageFilter, sessionID string, ) ([]duckUsageAggregateRow, error) { cte, args := duckUsageCTE(f, sessionID) query := cte + ` SELECT session_id, project, agent, model, ANY_VALUE(display_name) AS display_name, ANY_VALUE(started_at) AS started_at, SUM(input_tokens_norm) AS input_tokens, SUM(output_tokens_norm) AS output_tokens, SUM(cache_create_norm) AS cache_creation_tokens, SUM(cache_read_norm) AS cache_read_tokens, SUM(CASE WHEN cost_usd IS NULL THEN input_tokens_norm ELSE 0 END) AS billable_input_tokens, SUM(CASE WHEN cost_usd IS NOT NULL THEN 0 WHEN output_tokens_norm = 0 THEN reasoning_tokens_norm ELSE output_tokens_norm END) AS billable_output_tokens, CAST(0 AS BIGINT) AS billable_reasoning_tokens, SUM(CASE WHEN cost_usd IS NULL THEN cache_create_norm ELSE 0 END) AS billable_cache_creation_tokens, SUM(CASE WHEN cost_usd IS NULL THEN cache_read_norm ELSE 0 END) AS billable_cache_read_tokens, COALESCE(SUM(cost_usd), 0) AS explicit_cost, COUNT(cost_usd) AS reported_cost_rows FROM usage_localized GROUP BY session_id, project, agent, model ORDER BY session_id ASC, model ASC` rows, err := s.queryContext(ctx, query, args...) if err != nil { return nil, fmt.Errorf("querying duckdb session usage aggregates: %w", err) } defer rows.Close() var out []duckUsageAggregateRow for rows.Next() { var r duckUsageAggregateRow var startedAt any if err := rows.Scan( &r.sessionID, &r.project, &r.agent, &r.model, &r.displayName, &startedAt, &r.inputTok, &r.outputTok, &r.cacheCr, &r.cacheRd, &r.billableInput, &r.billableOutput, &r.billableReason, &r.billableCacheCr, &r.billableCacheRd, &r.explicitCost, &r.reportedCostRows, ); err != nil { return nil, fmt.Errorf("scanning duckdb session usage aggregate: %w", err) } r.startedAt = formatDBTime(startedAt) out = append(out, r) } return out, rows.Err() } // sessionUsageRowCount counts the deduped usage rows that would // contribute breakdown entries, mirroring duckSessionUsageRowCost's // contributes rule (an explicit cost or any nonzero token counter) // without shipping the rows. func (s *Store) sessionUsageRowCount( ctx context.Context, sessionID string, ) (int, error) { cte, args := duckUsageCTE(db.UsageFilter{}, sessionID) query := cte + ` SELECT COUNT(*) FROM usage_localized WHERE cost_usd IS NOT NULL OR input_tokens_norm != 0 OR output_tokens_norm != 0 OR cache_create_norm != 0 OR cache_read_norm != 0 OR reasoning_tokens_norm != 0` var count int if err := s.queryRowContext(ctx, query, args...). Scan(&count); err != nil { return 0, fmt.Errorf( "counting duckdb session usage rows: %w", err) } return count, nil } func (s *Store) sessionUsageRows( ctx context.Context, sessionID string, ) ([]duckSessionUsageRow, error) { cte, args := duckUsageCTE(db.UsageFilter{}, sessionID) query := cte + ` SELECT session_id, message_ordinal, source, ts, model, input_tokens_norm, output_tokens_norm, cache_create_norm, cache_read_norm, reasoning_tokens_norm, cost_usd FROM usage_localized ORDER BY ts ASC, session_id ASC, COALESCE(message_ordinal, -1) ASC, source ASC, usage_dedup_key ASC` rows, err := s.queryContext(ctx, query, args...) if err != nil { return nil, fmt.Errorf("querying duckdb session usage rows: %w", err) } defer rows.Close() var out []duckSessionUsageRow for rows.Next() { var r duckSessionUsageRow var ts any if err := rows.Scan( &r.sessionID, &r.messageOrdinal, &r.source, &ts, &r.model, &r.inputTok, &r.outputTok, &r.cacheCr, &r.cacheRd, &r.reasoningTok, &r.costUSD, ); err != nil { return nil, fmt.Errorf("scanning duckdb session usage row: %w", err) } r.ts = formatDBTime(ts) out = append(out, r) } return out, rows.Err() } func (s *Store) GetTopSessionsByCost( ctx context.Context, f db.UsageFilter, limit int, ) ([]db.TopSessionEntry, error) { if limit <= 0 { limit = 20 } pricing, err := s.loadPricing(ctx) if err != nil { return nil, err } rateResolver := export.NewPricingResolver(duckPricingRows(pricing)) rows, err := s.sessionUsageAggregateRows(ctx, f, "") if err != nil { return nil, err } type acc struct { row db.TopSessionEntry tokens int cost float64 } bySession := map[string]*acc{} for _, r := range rows { a := bySession[r.sessionID] if a == nil { a = &acc{row: db.TopSessionEntry{ SessionID: r.sessionID, DisplayName: r.displayName, Agent: r.agent, Project: r.project, StartedAt: r.startedAt, }} bySession[r.sessionID] = a } cost, _, _, _ := duckUsageAggregateCost( r.model, r.inputTok, r.outputTok, r.cacheCr, r.cacheRd, r.billableInput, r.billableOutput, r.billableReason, r.billableCacheCr, r.billableCacheRd, r.explicitCost, r.reportedCostRows > 0, rateResolver, ) a.tokens += r.inputTok + r.outputTok + r.cacheCr + r.cacheRd a.cost += cost } out := make([]db.TopSessionEntry, 0, len(bySession)) for _, a := range bySession { a.row.TotalTokens = a.tokens a.row.Cost = roundCost(a.cost) out = append(out, a.row) } sort.Slice(out, func(i, j int) bool { if out[i].Cost != out[j].Cost { return out[i].Cost > out[j].Cost } return out[i].SessionID < out[j].SessionID }) if len(out) > limit { out = out[:limit] } return out, nil } func (s *Store) GetUsageSessionCounts( ctx context.Context, f db.UsageFilter, ) (db.UsageSessionCounts, error) { rows, err := s.sessionUsageAggregateRows(ctx, f, "") if err != nil { return db.UsageSessionCounts{}, err } type sessionInfo struct { project string agent string } seen := map[string]sessionInfo{} for _, r := range rows { seen[r.sessionID] = sessionInfo{ project: r.project, agent: r.agent, } } out := db.UsageSessionCounts{ByProject: map[string]int{}, ByAgent: map[string]int{}} for _, r := range seen { out.Total++ out.ByProject[r.project]++ out.ByAgent[r.agent]++ } return out, nil } // appendDuckUsageMatchingActivityClauses requires the session to have at // least one row that GetUsageMatchingSessionCount's bounded branch would // count, mirroring appendUsageMatchingActivityClauses in internal/db so // bounded and unbounded requests agree on which sessions match. func appendDuckUsageMatchingActivityClauses( where string, args []any, f db.UsageFilter, ) (string, []any) { var messageArgs []any messageWhere, messageArgs := appendDuckUsageSourceFilterClauses( duckUsageMatchingMessageSourceEligibility, messageArgs, "m.model", f, ) var eventArgs []any eventWhere, eventArgs := appendDuckUsageSourceFilterClauses( duckUsageEventSourceEligibility, eventArgs, "ue.model", f, ) where += ` AND ( EXISTS ( SELECT 1 FROM messages m WHERE m.session_id = s.id AND ` + messageWhere + ` ) OR EXISTS ( SELECT 1 FROM usage_events ue WHERE ue.session_id = s.id AND ` + eventWhere + ` ) )` args = append(args, messageArgs...) args = append(args, eventArgs...) return where, args } // GetUsageMatchingSessionCount counts sessions that match the usage filter // even when they have no token-bearing usage rows. Bounded ranges are // resolved against message/usage_events timestamps (falling back to // s.started_at), the same shape duckUsageRawSQL already uses for the // normal usage query, so a session whose activity falls outside the // window but whose message timestamp falls inside it is still counted. func (s *Store) GetUsageMatchingSessionCount( ctx context.Context, f db.UsageFilter, ) (int, error) { if f.From == "" && f.To == "" { where, args := appendDuckUsageSessionFilterClauses( "s.deleted_at IS NULL", nil, f, "") where, args = appendDuckUsageMatchingActivityClauses(where, args, f) var count int err := s.duck.QueryRowContext(ctx, ` SELECT COUNT(*) FROM sessions s WHERE `+where, args...).Scan(&count) if err != nil { return 0, fmt.Errorf("querying matching usage sessions: %w", err) } return count, nil } query, args := duckMatchingUsageRawSQL(f) rows, err := s.duck.QueryContext(ctx, query, args...) if err != nil { return 0, fmt.Errorf("querying matching usage sessions: %w", err) } defer rows.Close() seen := make(map[string]struct{}) for rows.Next() { var ( id string ts any ) if err := rows.Scan(&id, &ts); err != nil { return 0, fmt.Errorf("scanning matching usage session: %w", err) } date := analyticsLocalDate(formatDBTime(ts), f.Timezone) if date == "" { continue } if f.From != "" && date < f.From { continue } if f.To != "" && date > f.To { continue } seen[id] = struct{}{} } if err := rows.Err(); err != nil { return 0, fmt.Errorf("iterating matching usage sessions: %w", err) } return len(seen), nil } func (s *Store) GetSessionUsage( ctx context.Context, sessionID string, includeBreakdown bool, ) (*db.SessionUsage, error) { sess, err := s.GetSession(ctx, sessionID) if err != nil || sess == nil { return nil, err } pricing, err := s.loadPricing(ctx) if err != nil { return nil, err } rateResolver := export.NewPricingResolver(duckPricingRows(pricing)) rows, err := s.sessionUsageAggregateRows(ctx, db.UsageFilter{}, sessionID) if err != nil { return nil, err } var breakdownRows []duckSessionUsageRow breakdownCount := 0 if includeBreakdown { breakdownRows, err = s.sessionUsageRows(ctx, sessionID) } else { breakdownCount, err = s.sessionUsageRowCount(ctx, sessionID) } if err != nil { return nil, err } models := map[string]bool{} unpriced := map[string]bool{} totalCost := 0.0 hasRows := false for _, r := range rows { hasRows = true models[r.model] = true cost, _, priced, contributes := duckUsageAggregateCost( r.model, r.inputTok, r.outputTok, r.cacheCr, r.cacheRd, r.billableInput, r.billableOutput, r.billableReason, r.billableCacheCr, r.billableCacheRd, r.explicitCost, r.reportedCostRows > 0, rateResolver, ) if !contributes { continue } totalCost += cost if !priced { unpriced[r.model] = true } } breakdown := make([]db.SessionUsageBreakdownEntry, 0, len(breakdownRows)) for _, r := range breakdownRows { cost, priced, contributes := duckSessionUsageRowCost(r, pricing) if !contributes { continue } breakdown = append(breakdown, duckSessionUsageBreakdownEntry( r, len(breakdown)+1, cost, priced)) } if includeBreakdown { breakdownCount = len(breakdown) } out := &db.SessionUsage{ SessionID: sessionID, Agent: sess.Agent, Project: sess.Project, TotalOutputTokens: sess.TotalOutputTokens, PeakContextTokens: sess.PeakContextTokens, HasTokenData: hasRows || sess.HasTotalOutputTokens || sess.HasPeakContextTokens, Models: sortedBoolKeys(models), UnpricedModels: sortedBoolKeys(unpriced), BreakdownCount: breakdownCount, Breakdown: breakdown, } if len(unpriced) == 0 && hasRows { out.HasCost = true out.CostUSD = roundCost(totalCost) out.AICredits = db.AICreditsFromCost(sess.Agent, out.CostUSD) } return out, nil }