package db import ( "context" "errors" "fmt" "math" "slices" "sort" "strings" "time" "unicode" "go.kenn.io/agentsview/internal/signals" ) // maxSQLVars is the maximum bind variables per IN clause to stay // within SQLite's default SQLITE_MAX_VARIABLE_NUMBER (999). const maxSQLVars = 500 var ErrUnsupportedAnalyticsSignal = errors.New( "unsupported analytics signal", ) var supportedAnalyticsSignals = map[string]struct{}{ "outcome_errored": {}, "outcome_abandoned": {}, "outcome_completed": {}, "tool_failure_signals": {}, "tool_retries": {}, "edit_churn": {}, "sessions_with_compaction": {}, "mid_task_compaction_count": {}, "high_pressure_sessions": {}, "short_prompt_count": {}, "unstructured_start": {}, "missing_success_criteria_count": {}, "missing_verification_count": {}, "duplicate_prompt_count": {}, "no_code_context_count": {}, "runaway_tool_loop_count": {}, "frustration_marker_count": {}, } func IsSupportedAnalyticsSignal(signal string) bool { _, ok := supportedAnalyticsSignals[signal] return ok } // inPlaceholders returns a "(?,?,...)" string and []any args for // a slice of string IDs. func inPlaceholders(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 } // queryChunked executes a callback for each chunk of IDs, // splitting at maxSQLVars to avoid SQLite bind-variable limits. func queryChunked( ids []string, fn func(chunk []string) error, ) error { return queryChunkedSize(ids, maxSQLVars, fn) } // queryChunkedSize is queryChunked with an explicit per-chunk size, for // queries that bind each ID more than once (and so need a smaller chunk to // keep the total bind count within SQLite's variable limit). func queryChunkedSize( ids []string, size int, fn func(chunk []string) error, ) error { for i := 0; i < len(ids); i += size { end := min(i+size, len(ids)) if err := fn(ids[i:end]); err != nil { return err } } return nil } // AnalyticsFilter is the shared filter for all analytics queries. type AnalyticsFilter struct { From string // ISO date YYYY-MM-DD, inclusive To string // ISO date YYYY-MM-DD, inclusive Machine string // optional machine filter Project string // optional project filter // GitBranch is a branchListSep-joined list of opaque (project, branch) tokens (EncodeBranchFilterToken). GitBranch string Agent string // optional agent filter Model string // optional model filter Timezone string // IANA timezone for day bucketing DayOfWeek *int // nil = all, 0=Mon, 6=Sun (ISO) Hour *int // nil = all, 0-23 MinUserMessages int // user_message_count >= N ExcludeOneShot bool // exclude sessions with user_message_count <= 1 ExcludeAutomated bool // exclude automated (roborev) sessions // ExcludeInteractive is the mirror of ExcludeAutomated: it keeps only // automated sessions. The two are never set together (that would match // nothing); the activity report uses them for its automation filter. ExcludeInteractive bool AutomatedScope string // "", "human", "all", or "automated" ActiveSince string // ISO timestamp cutoff Termination string // "", "clean", or "unclean" // IncludeSubagents counts subagent sessions (including workflow // subagents) in token/session aggregates. It is opt-in and set only // on the sum/count surfaces GetAnalyticsSummary and // GetAnalyticsProjects. Distribution surfaces (session-shape, // velocity, timing) leave it false so short subagent sessions do not // skew them. IncludeSubagents bool // IncludeForks counts fork sessions (rewound conversation branches // split out by the claude and piebald parsers). Fork sessions hold // only their own branch's messages, never replayed copies, so their // usage is real spend; GetDailyUsage counts it via per-row dedup. // Set only by GetActivityReport so its cost totals match daily // usage. Aggregate analytics surfaces leave forks excluded so a // rewound branch does not count as an extra session. IncludeForks bool } // RelationshipExclusionSQL returns the relationship_type predicate for // analytics aggregation. The default excludes subagent and fork rows // (matching the session list); IncludeSubagents and IncludeForks each // lift one exclusion. Exported so the PostgreSQL and DuckDB analytics // builders apply the same rule. func (f AnalyticsFilter) RelationshipExclusionSQL() string { return RelationshipExclusionSQL(f.IncludeSubagents, f.IncludeForks, "") } // RelationshipExclusionSQL is the single source of truth for the // relationship_type analytics predicate, shared by the analytics // builders (AnalyticsFilter) and the stats pipeline (StatsFilter). // colPrefix qualifies the column for callers that alias the sessions // table (e.g. "s."); pass "" for an unqualified column. Subagent and // fork rows are excluded unless the corresponding include flag is set; // with both set the predicate is a no-op so the clause stays composable. func RelationshipExclusionSQL(includeSubagents, includeForks bool, colPrefix string) string { col := colPrefix + "relationship_type" switch { case includeSubagents && includeForks: return "1=1" case includeSubagents: return col + " NOT IN ('fork')" case includeForks: return col + " NOT IN ('subagent')" default: return col + " NOT IN ('subagent', 'fork')" } } // OneShotExclusionSQL wraps the one-shot exclusion predicate so it does // not drop subagent rows when subagents are being counted. Workflow // subagents are inherently one-shot (a single orchestrator prompt // yields one result) but represent real work, so the one-shot filter // would otherwise re-hide exactly the sessions IncludeSubagents is // meant to surface. Exported so the PostgreSQL and DuckDB builders // apply the same rule. base must be a self-contained boolean clause. func (f AnalyticsFilter) OneShotExclusionSQL(base string) string { if f.IncludeSubagents { return "(" + base + " OR relationship_type = 'subagent')" } return base } // location loads the timezone or returns UTC on error. func (f AnalyticsFilter) location() *time.Location { if f.Timezone == "" { return time.UTC } loc, err := time.LoadLocation(f.Timezone) if err != nil { return time.UTC } return loc } // utcRange returns UTC time bounds padded by ±14h to cover // all possible timezone offsets. The WHERE clause uses these // to leverage the started_at index. Empty From/To inputs // collapse to wide-open sentinels so a zero AnalyticsFilter // matches every session (mirrors the PG store). func (f AnalyticsFilter) utcRange() (string, string) { const ( unboundedFrom = "0001-01-01T00:00:00Z" unboundedTo = "9999-12-31T23:59:59Z" ) from := unboundedFrom if f.From != "" { from = f.From + "T00:00:00Z" } to := unboundedTo if f.To != "" { to = f.To + "T23:59:59Z" } tFrom, err := time.Parse(time.RFC3339, from) if err != nil { return unboundedFrom, unboundedTo } tTo, err := time.Parse(time.RFC3339, to) if err != nil { return unboundedFrom, unboundedTo } // Skip ±14h padding on the sentinels to avoid pushing the // lower bound below year 1. if f.From == "" { from = unboundedFrom } else { from = tFrom.Add(-14 * time.Hour).Format(time.RFC3339) } if f.To == "" { to = unboundedTo } else { to = tTo.Add(14 * time.Hour).Format(time.RFC3339) } return from, to } // buildWhere returns a WHERE clause and args for common // analytics filters. func (f AnalyticsFilter) buildWhere( dateCol string, ) (string, []any) { return f.buildWhereWithDate(dateCol, true, "sessions.id") } // buildWhereWithoutDate returns common analytics predicates // without adding session date bounds. Callers that evaluate // date windows against message timestamps should use this to // avoid pre-filtering by the parent session timestamp. func (f AnalyticsFilter) buildWhereWithoutDate() (string, []any) { return f.buildWhereWithDate("", false, "sessions.id") } func csvFilterValues(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 sqliteAnalyticsCSVPredicate( col string, raw string, ) (string, []any) { values := csvFilterValues(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 (db *DB) getAnalyticsFilteredMessageCounts( ctx context.Context, sessionIDs []string, f AnalyticsFilter, ) (map[string]int, error) { stats, err := db.getAnalyticsFilteredMessageStats( ctx, sessionIDs, f, ) if err != nil { return nil, err } counts := make(map[string]int, len(stats)) for sessionID, stat := range stats { counts[sessionID] = stat.Messages } return counts, nil } func (db *DB) getAnalyticsModelScopedMessages( ctx context.Context, sessionIDs []string, f AnalyticsFilter, ) (map[string][]ScopedMessage, error) { scope, err := db.resolveAnalyticsMessageScope(ctx, sessionIDs, f, true) if err != nil { return nil, err } if scope == nil { return map[string][]ScopedMessage{}, nil } return scope.MessagesBySession(), nil } func (db *DB) getAnalyticsFilteredMessageStats( ctx context.Context, sessionIDs []string, f AnalyticsFilter, ) (map[string]MessageStats, error) { scope, err := db.resolveAnalyticsMessageScope(ctx, sessionIDs, f, false) if err != nil { return nil, err } if scope == nil { return map[string]MessageStats{}, nil } return scope.StatsBySession(), nil } func (f AnalyticsFilter) buildWhereWithDate( dateCol string, includeDate bool, sessionIDExpr string, ) (string, []any) { if sessionIDExpr == "" { sessionIDExpr = "sessions.id" } preds := []string{ "message_count > 0", f.RelationshipExclusionSQL(), "deleted_at IS NULL", } var args []any if includeDate { utcFrom, utcTo := f.utcRange() preds = append(preds, dateCol+" >= ?") args = append(args, utcFrom) preds = append(preds, dateCol+" <= ?") args = append(args, utcTo) } if f.Machine != "" { machines := csvFilterValues(f.Machine) if len(machines) == 1 { preds = append(preds, "machine = ?") args = append(args, machines[0]) } else if len(machines) > 1 { placeholders := make( []string, len(machines), ) for i, machine := range machines { placeholders[i] = "?" args = append(args, machine) } preds = append(preds, "machine IN ("+ strings.Join(placeholders, ",")+ ")", ) } } if f.Project != "" { preds = append(preds, "project = ?") args = append(args, f.Project) } if f.GitBranch != "" { var clause string clause, args = BranchPairClauseArgs("project", "git_branch", f.GitBranch, args) preds = append(preds, clause) } if f.Agent != "" { agents := csvFilterValues(f.Agent) if len(agents) == 1 { preds = append(preds, "agent = ?") args = append(args, agents[0]) } else if len(agents) > 1 { placeholders := make( []string, len(agents), ) for i, a := range agents { placeholders[i] = "?" args = append(args, a) } preds = append(preds, "agent IN ("+ strings.Join(placeholders, ",")+ ")", ) } } if f.Model != "" { models := csvFilterValues(f.Model) if len(models) == 1 { preds = append(preds, "EXISTS (SELECT 1 FROM messages m WHERE "+ "m.session_id = "+sessionIDExpr+" AND "+ "m.model = ?)") args = append(args, models[0]) } else if len(models) > 1 { placeholders := make( []string, len(models), ) for i, m := range models { placeholders[i] = "?" args = append(args, m) } preds = append(preds, "EXISTS (SELECT 1 FROM messages m WHERE "+ "m.session_id = "+sessionIDExpr+" AND "+ "m.model IN ("+ strings.Join(placeholders, ",")+ "))") } } if f.MinUserMessages > 0 { preds = append(preds, "user_message_count >= ?") args = append(args, f.MinUserMessages) } scope := normalizeAutomatedScope(f.AutomatedScope, f.ExcludeAutomated) if f.ExcludeOneShot { if scope != "human" { preds = append(preds, f.OneShotExclusionSQL( "(user_message_count > 1 OR is_automated = 1)")) } else { preds = append(preds, f.OneShotExclusionSQL("user_message_count > 1")) } } if pred := automatedScopePredicate(scope, "is_automated"); pred != "" { preds = append(preds, pred) } if f.ExcludeInteractive { preds = append(preds, "is_automated = 1") } if f.ActiveSince != "" { preds = append(preds, "COALESCE(NULLIF(ended_at, ''), NULLIF(started_at, ''), created_at) >= ?") args = append(args, f.ActiveSince) } if pred, pargs := buildTerminationPredSQLite(f.Termination); pred != "" { preds = append(preds, pred) args = append(args, pargs...) } return strings.Join(preds, " AND "), args } func normalizeAutomatedScope(scope string, excludeAutomated bool) string { switch strings.TrimSpace(scope) { case "human", "all", "automated": return strings.TrimSpace(scope) } if excludeAutomated { return "human" } return "all" } func automatedScopePredicate(scope, col string) string { switch scope { case "human": return col + " = 0" case "automated": return col + " = 1" default: return "" } } func (db *DB) queryAnalyticsModels( ctx context.Context, query string, args []any, ) ([]string, error) { rows, err := db.getReader().QueryContext(ctx, ` `+query, args...) if err != nil { return nil, fmt.Errorf("querying analytics models: %w", err) } defer rows.Close() models := make([]string, 0) for rows.Next() { var model string if err := rows.Scan(&model); err != nil { return nil, fmt.Errorf("scanning analytics model: %w", err) } models = append(models, model) } if err := rows.Err(); err != nil { return nil, fmt.Errorf("iterating analytics models: %w", err) } return models, nil } func (db *DB) getAnalyticsModelsSQLiteSummary( ctx context.Context, f AnalyticsFilter, dateCol string, ) ([]string, error) { if f.HasTimeFilter() { ids, err := db.filteredSessionIDs(ctx, f) if err != nil { return nil, err } sessionIDs := make([]string, 0, len(ids)) for id := range ids { sessionIDs = append(sessionIDs, id) } return db.getAnalyticsModelsForSessionIDsFiltered( ctx, sessionIDs, f, ) } where, args := sqliteAnalyticsWhereSQL(f, dateCol, "s.id", true) return db.queryAnalyticsModels(ctx, ` SELECT DISTINCT m.model FROM sessions s JOIN messages m ON m.session_id = s.id WHERE `+where+` AND COALESCE(m.model, '') <> '' ORDER BY m.model`, args, ) } func (db *DB) getAnalyticsModelsForSessionIDs( ctx context.Context, sessionIDs []string, ) ([]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) } modelSet := make(map[string]struct{}) models := make([]string, 0) if err := queryChunked(unique, func(chunk []string) error { ph, args := inPlaceholders(chunk) found, err := db.queryAnalyticsModels(ctx, ` SELECT DISTINCT model FROM messages WHERE session_id IN `+ph+` AND COALESCE(model, '') <> '' ORDER BY model`, args, ) if err != nil { return err } for _, model := range found { if _, ok := modelSet[model]; ok { continue } modelSet[model] = struct{}{} models = append(models, model) } return nil }); err != nil { return nil, err } sort.Strings(models) return models, nil } func (db *DB) getAnalyticsModelsForSessionIDsFiltered( ctx context.Context, sessionIDs []string, f 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 := csvFilterValues(f.Model) allowedModels := make(map[string]struct{}, len(filterModels)) for _, model := range filterModels { allowedModels[model] = struct{}{} } loc := f.location() modelSet := make(map[string]struct{}) models := make([]string, 0) if err := queryChunked(unique, func(chunk []string) error { ph, args := inPlaceholders(chunk) rows, err := db.getReader().QueryContext(ctx, ` SELECT model, COALESCE(timestamp, '') FROM messages WHERE session_id IN `+ph+` AND COALESCE(model, '') <> ''`, args..., ) if err != nil { return fmt.Errorf("querying filtered analytics models: %w", err) } defer rows.Close() for rows.Next() { var model, ts string if err := rows.Scan(&model, &ts); err != nil { return fmt.Errorf("scanning filtered analytics model: %w", err) } if len(allowedModels) > 0 { if _, ok := allowedModels[model]; !ok { continue } } if f.HasTimeFilter() { t, ok := localTime(ts, loc) if !ok || !f.matchesTimeFilter(t) { continue } } if _, ok := modelSet[model]; ok { continue } modelSet[model] = struct{}{} models = append(models, model) } if err := rows.Err(); err != nil { return fmt.Errorf("iterating filtered analytics models: %w", err) } return nil }); err != nil { return nil, err } sort.Strings(models) return models, nil } // HasTimeFilter returns true when hour-of-day or day-of-week // filtering is active. func (f AnalyticsFilter) HasTimeFilter() bool { return f.DayOfWeek != nil || f.Hour != nil } // matchesTimeFilter checks whether a local time matches the // active hour and/or day-of-week filter. func (f AnalyticsFilter) matchesTimeFilter( t time.Time, ) bool { if f.DayOfWeek != nil { dow := (int(t.Weekday()) + 6) % 7 // ISO Mon=0 if dow != *f.DayOfWeek { return false } } if f.Hour != nil { if t.Hour() != *f.Hour { return false } } return true } func (f AnalyticsFilter) canUseSQLiteTimeSQL() bool { _, ok := f.sqliteTimeModifier() return ok } func (f AnalyticsFilter) sqliteTimeModifier() (string, bool) { if f.Timezone == "" || f.Timezone == "UTC" { return "", true } if f.From == "" || f.To == "" { return "", false } loc, err := time.LoadLocation(f.Timezone) if err != nil { return "", false } start, err := time.Parse("2006-01-02", f.From) if err != nil { return "", false } end, err := time.Parse("2006-01-02", f.To) if err != nil { return "", false } var offset *int for d := start; !d.After(end); d = d.AddDate(0, 0, 1) { checks := []time.Time{ time.Date(d.Year(), d.Month(), d.Day(), 0, 0, 0, 0, loc), time.Date(d.Year(), d.Month(), d.Day(), 23, 59, 59, 0, loc), } for _, local := range checks { _, current := local.Zone() if current%60 != 0 { return "", false } if offset == nil { v := current offset = &v continue } if *offset != current { return "", false } } } if offset == nil { return "", false } sign := "+" value := *offset if value < 0 { sign = "-" value = -value } return fmt.Sprintf("%s%02d:%02d", sign, value/3600, (value%3600)/60), true } func sqliteDateExpr(dateCol string, modifier string) string { if modifier == "" { return "strftime('%Y-%m-%d', " + dateCol + ")" } return "strftime('%Y-%m-%d', " + dateCol + ", '" + modifier + "')" } func sqliteAnalyticsWhereSQL( f AnalyticsFilter, dateCol string, sessionIDExpr string, includeTime bool, ) (string, []any) { where, args := f.buildWhereWithDate( dateCol, true, sessionIDExpr, ) modifier, _ := f.sqliteTimeModifier() dateExpr := sqliteDateExpr(dateCol, modifier) if f.From != "" { where += " AND " + dateExpr + " >= ?" args = append(args, f.From) } if f.To != "" { where += " AND " + dateExpr + " <= ?" args = append(args, f.To) } if includeTime && f.HasTimeFilter() { preds := []string{ "m.session_id = " + sessionIDExpr, "m.timestamp != ''", } if modelPred, modelArgs := sqliteAnalyticsCSVPredicate( "m.model", f.Model, ); modelPred != "" { preds = append(preds, modelPred) args = append(args, modelArgs...) } if f.DayOfWeek != nil { dowExpr := "strftime('%w', m.timestamp)" if modifier != "" { dowExpr = "strftime('%w', m.timestamp, '" + modifier + "')" } preds = append(preds, "((CAST("+dowExpr+" AS INTEGER) + 6) % 7) = ?") args = append(args, *f.DayOfWeek) } if f.Hour != nil { hourExpr := "strftime('%H', m.timestamp)" if modifier != "" { hourExpr = "strftime('%H', m.timestamp, '" + modifier + "')" } preds = append(preds, "CAST("+hourExpr+" AS INTEGER) = ?") args = append(args, *f.Hour) } where += " AND EXISTS (SELECT 1 FROM messages m WHERE " + strings.Join(preds, " AND ") + ")" } return where, args } // filteredSessionIDs returns the set of session IDs that have // at least one message matching the hour/dow filter. Used by // session-level queries to restrict results when time filters // are active. With a model filter active it pairs through the // shared scope reducer (see filteredSessionIDsModel) so an // empty-model user turn at the selected hour keeps its session, // matching how the model-scoped panels count it. func (db *DB) filteredSessionIDs( ctx context.Context, f AnalyticsFilter, ) (map[string]bool, error) { if strings.TrimSpace(f.Model) != "" { return db.filteredSessionIDsModel(ctx, f) } loc := f.location() dateCol := "COALESCE(NULLIF(s.started_at, ''), s.created_at)" where, args := f.buildWhereWithDate( dateCol, true, "s.id", ) query := `SELECT s.id, m.timestamp FROM sessions s JOIN messages m ON m.session_id = s.id WHERE ` + where + ` AND m.timestamp != ''` rows, err := db.getReader().QueryContext(ctx, query, args...) if err != nil { return nil, fmt.Errorf( "querying filtered session IDs: %w", err, ) } defer rows.Close() ids := make(map[string]bool) for rows.Next() { var sid, msgTS string if err := rows.Scan(&sid, &msgTS); err != nil { return nil, fmt.Errorf( "scanning filtered session ID: %w", err, ) } if ids[sid] { continue // already matched } t, ok := localTime(msgTS, loc) if !ok { continue } if f.matchesTimeFilter(t) { ids[sid] = true } } if err := rows.Err(); err != nil { return nil, fmt.Errorf( "iterating filtered session IDs: %w", err, ) } return ids, nil } // filteredSessionIDsModel returns the sessions that have at least one // model-scoped message matching the hour/dow filter. Unlike a direct m.model // predicate, it runs the shared scope reducer (with the day/hour filter), so an // empty-model user turn paired with a selected-model assistant keeps its // session when the user turn falls in the selected hour. func (db *DB) filteredSessionIDsModel( ctx context.Context, f AnalyticsFilter, ) (map[string]bool, error) { sessionIDs, err := db.analyticsModelCandidateSessionIDs(ctx, f) if err != nil { return nil, err } scope, err := db.resolveAnalyticsMessageScope(ctx, sessionIDs, f, false) if err != nil { return nil, err } ids := make(map[string]bool) if scope != nil { for id := range scope.MessagesBySession() { ids[id] = true } } return ids, nil } // localTime parses a UTC timestamp string and converts it to the // given location. Returns the local time and true on success. func localTime( ts string, loc *time.Location, ) (time.Time, bool) { t, err := time.Parse(time.RFC3339Nano, ts) if err != nil { t, err = time.Parse("2006-01-02T15:04:05Z", ts) if err != nil { return time.Time{}, false } } return t.In(loc), true } // localDate converts a UTC timestamp string to a local date // string (YYYY-MM-DD) in the given location. func localDate(ts string, loc *time.Location) string { t, ok := localTime(ts, loc) if !ok { if len(ts) >= 10 { return ts[:10] } return "" } return t.Format("2006-01-02") } // percentileFloat returns the value at the given percentile // from a pre-sorted float64 slice. func percentileFloat(sorted []float64, pct float64) float64 { n := len(sorted) if n == 0 { return 0 } idx := int(float64(n) * pct) if idx >= n { idx = n - 1 } return sorted[idx] } // inDateRange checks if a local date falls within [from, to]. // Empty bounds are treated as unbounded so callers can pass a // zero AnalyticsFilter to get every session. func inDateRange(date, from, to string) bool { if from != "" && date < from { return false } if to != "" && date > to { return false } return true } // medianInt returns the median of a sorted int slice of // length n. For even n, returns the average of the two // middle elements. func medianInt(sorted []int, n int) int { if n == 0 { return 0 } if n%2 == 0 { return (sorted[n/2-1] + sorted[n/2]) / 2 } return sorted[n/2] } // --- Summary --- // AgentSummary holds per-agent counts for the summary. type AgentSummary struct { Sessions int `json:"sessions"` Messages int `json:"messages"` } // AnalyticsSummary is the response for the summary endpoint. type AnalyticsSummary struct { TotalSessions int `json:"total_sessions"` TotalMessages int `json:"total_messages"` TotalOutputTokens int `json:"total_output_tokens"` TokenReportingSessions int `json:"token_reporting_sessions"` Models []string `json:"models"` ActiveProjects int `json:"active_projects"` ActiveDays int `json:"active_days"` AvgMessages float64 `json:"avg_messages"` MedianMessages int `json:"median_messages"` P90Messages int `json:"p90_messages"` MostActive string `json:"most_active_project"` Concentration float64 `json:"concentration"` Agents map[string]*AgentSummary `json:"agents"` } // GetAnalyticsSummary returns aggregate statistics. func (db *DB) GetAnalyticsSummary( ctx context.Context, f AnalyticsFilter, ) (AnalyticsSummary, error) { // The summary is a token/session aggregate, so subagent sessions // (including workflow subagents) are counted here. f.IncludeSubagents = true if !f.canUseSQLiteTimeSQL() || strings.TrimSpace(f.Model) != "" { return db.getAnalyticsSummaryGo(ctx, f) } dateCol := "COALESCE(NULLIF(started_at, ''), created_at)" where, args := sqliteAnalyticsWhereSQL(f, dateCol, "sessions.id", true) modifier, _ := f.sqliteTimeModifier() dateExpr := sqliteDateExpr(dateCol, modifier) query := ` WITH filtered AS ( SELECT id, project, agent, message_count, total_output_tokens, has_total_output_tokens, ` + dateExpr + ` AS local_date FROM sessions 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(AVG(message_count) AS INTEGER) FROM ranked WHERE rn IN ( CAST(((n + 1) / 2) AS INTEGER), CAST(((n + 2) / 2) AS INTEGER) ) ), 0) AS median_messages, COALESCE(( SELECT message_count FROM ranked WHERE rn = MIN(CAST(n * 0.9 AS INTEGER) + 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 ) ) * 1.0 / NULLIF(SUM(message_count), 0), 3), 0) AS concentration FROM filtered` rows, err := db.getReader().QueryContext(ctx, query, args...) if err != nil { return AnalyticsSummary{}, fmt.Errorf("querying analytics summary: %w", err) } s := AnalyticsSummary{ Agents: make(map[string]*AgentSummary), Models: []string{}, } if !rows.Next() { rows.Close() return s, nil } if err := rows.Scan( &s.TotalSessions, &s.TotalMessages, &s.TotalOutputTokens, &s.TokenReportingSessions, &s.ActiveProjects, &s.ActiveDays, &s.AvgMessages, &s.MedianMessages, &s.P90Messages, &s.MostActive, &s.Concentration, ); err != nil { rows.Close() return AnalyticsSummary{}, fmt.Errorf("scanning summary row: %w", err) } if err := rows.Err(); err != nil { rows.Close() return AnalyticsSummary{}, fmt.Errorf("iterating summary rows: %w", err) } if err := rows.Close(); err != nil { return AnalyticsSummary{}, fmt.Errorf("closing summary rows: %w", err) } models, err := db.getAnalyticsModelsSQLiteSummary(ctx, f, dateCol) if err != nil { return AnalyticsSummary{}, err } s.Models = models agentRows, err := db.getReader().QueryContext(ctx, ` WITH filtered AS ( SELECT agent, message_count FROM sessions WHERE `+where+` ) SELECT agent, COUNT(*), COALESCE(SUM(message_count), 0) FROM filtered GROUP BY agent`, args..., ) if err != nil { return AnalyticsSummary{}, fmt.Errorf("querying analytics summary agents: %w", err) } defer agentRows.Close() for agentRows.Next() { var agent string var summary AgentSummary if err := agentRows.Scan( &agent, &summary.Sessions, &summary.Messages, ); err != nil { return AnalyticsSummary{}, fmt.Errorf("scanning summary agent: %w", err) } s.Agents[agent] = &summary } if err := agentRows.Err(); err != nil { return AnalyticsSummary{}, fmt.Errorf("iterating summary agents: %w", err) } return s, nil } func (db *DB) getAnalyticsSummaryGo( ctx context.Context, f AnalyticsFilter, ) (AnalyticsSummary, error) { loc := f.location() dateCol := "COALESCE(NULLIF(started_at, ''), created_at)" where, args := f.buildWhere(dateCol) var timeIDs map[string]bool if f.HasTimeFilter() { var err error timeIDs, err = db.filteredSessionIDs(ctx, f) if err != nil { return AnalyticsSummary{}, err } } // Fetch sessions with their message counts and agents query := `SELECT id, ` + dateCol + `, message_count, agent, project, total_output_tokens, has_total_output_tokens FROM sessions WHERE ` + where + ` ORDER BY message_count ASC` rows, err := db.getReader().QueryContext(ctx, query, args...) if err != nil { return AnalyticsSummary{}, fmt.Errorf("querying analytics summary: %w", err) } defer rows.Close() type sessionRow struct { id string date string messages int agent string project string outputTokens int hasTokens bool } var all []sessionRow for rows.Next() { var id, ts string var mc int var agent, project string var outputTokens int var hasTokens bool if err := rows.Scan( &id, &ts, &mc, &agent, &project, &outputTokens, &hasTokens, ); err != nil { return AnalyticsSummary{}, fmt.Errorf("scanning summary row: %w", err) } date := localDate(ts, loc) if !inDateRange(date, f.From, f.To) { continue } if timeIDs != nil && !timeIDs[id] { continue } all = append(all, sessionRow{ id: id, date: date, messages: mc, agent: agent, project: project, outputTokens: outputTokens, hasTokens: hasTokens, }) } if err := rows.Err(); err != nil { return AnalyticsSummary{}, fmt.Errorf("iterating summary rows: %w", err) } var s AnalyticsSummary s.Agents = make(map[string]*AgentSummary) s.Models = []string{} if len(all) == 0 { return s, nil } if f.Model != "" { ids := make([]string, 0, len(all)) for _, r := range all { ids = append(ids, r.id) } stats, err := db.getAnalyticsFilteredMessageStats(ctx, ids, f) if err != nil { return AnalyticsSummary{}, err } for i := range all { stat := stats[all[i].id] all[i].messages = stat.Messages all[i].outputTokens = stat.OutputTokens all[i].hasTokens = stat.HasOutputTokens } } days := make(map[string]bool) projects := make(map[string]int) // project -> message count msgCounts := make([]int, 0, len(all)) sessionIDs := make([]string, 0, len(all)) for _, r := range all { s.TotalSessions++ s.TotalMessages += r.messages if r.hasTokens { s.TotalOutputTokens += r.outputTokens s.TokenReportingSessions++ } days[r.date] = true projects[r.project] += r.messages msgCounts = append(msgCounts, r.messages) sessionIDs = append(sessionIDs, r.id) if s.Agents[r.agent] == nil { s.Agents[r.agent] = &AgentSummary{} } s.Agents[r.agent].Sessions++ s.Agents[r.agent].Messages += r.messages } var models []string var modelErr error if strings.TrimSpace(f.Model) != "" || f.HasTimeFilter() { models, modelErr = db.getAnalyticsModelsForSessionIDsFiltered( ctx, sessionIDs, f, ) } else { models, modelErr = db.getAnalyticsModelsForSessionIDs( ctx, sessionIDs, ) } if modelErr != nil { return s, modelErr } s.Models = models s.ActiveProjects = len(projects) s.ActiveDays = len(days) s.AvgMessages = math.Round( float64(s.TotalMessages)/float64(s.TotalSessions)*10, ) / 10 sort.Ints(msgCounts) n := len(msgCounts) if n%2 == 0 { s.MedianMessages = (msgCounts[n/2-1] + msgCounts[n/2]) / 2 } else { s.MedianMessages = msgCounts[n/2] } p90Idx := int(float64(n) * 0.9) if p90Idx >= n { p90Idx = n - 1 } s.P90Messages = msgCounts[p90Idx] // Most active project by message count (deterministic tie-break) maxMsgs := 0 for name, count := range projects { if count > maxMsgs || (count == maxMsgs && name < s.MostActive) { maxMsgs = count s.MostActive = name } } // Concentration: fraction of messages in top 3 projects if s.TotalMessages > 0 { counts := make([]int, 0, len(projects)) for _, c := range projects { counts = append(counts, c) } sort.Sort(sort.Reverse(sort.IntSlice(counts))) top := min(3, len(counts)) topSum := 0 for _, c := range counts[:top] { topSum += c } s.Concentration = math.Round( float64(topSum)/float64(s.TotalMessages)*1000, ) / 1000 } return s, nil } // --- Activity --- // ActivityEntry is one time bucket in the activity timeline. type ActivityEntry struct { Date string `json:"date"` Sessions int `json:"sessions"` Messages int `json:"messages"` UserMessages int `json:"user_messages"` AssistantMessages int `json:"assistant_messages"` ToolCalls int `json:"tool_calls"` ThinkingMessages int `json:"thinking_messages"` ByAgent map[string]int `json:"by_agent"` } // ActivityResponse wraps the activity series. type ActivityResponse struct { Granularity string `json:"granularity"` Series []ActivityEntry `json:"series"` } // bucketDate truncates a date to the start of its bucket. func bucketDate(date string, granularity string) string { t, err := time.Parse("2006-01-02", date) if err != nil { return date } switch granularity { case "week": // ISO week: Monday start weekday := int(t.Weekday()) if weekday == 0 { weekday = 7 } t = t.AddDate(0, 0, -(weekday - 1)) return t.Format("2006-01-02") case "month": return t.Format("2006-01") + "-01" default: return date } } func (db *DB) getModelScopedToolCallCounts( ctx context.Context, sessionIDs []string, f AnalyticsFilter, ) (map[string]int, error) { counts := make(map[string]int, len(sessionIDs)) if len(sessionIDs) == 0 || strings.TrimSpace(f.Model) == "" { return counts, nil } flt := f.messageScopeFilter() loc := f.location() if err := queryChunked(sessionIDs, func(chunk []string) error { ph, args := inPlaceholders(chunk) rows, err := db.getReader().QueryContext(ctx, ` SELECT tc.session_id, m.model, COALESCE(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, COALESCE(m.timestamp, '')`, args..., ) if err != nil { return fmt.Errorf("querying model-scoped analytics tool calls: %w", err) } defer rows.Close() for rows.Next() { var sessionID, model, ts string var count int if err := rows.Scan(&sessionID, &model, &ts, &count); err != nil { return fmt.Errorf("scanning model-scoped analytics tool calls: %w", err) } if _, ok := flt.Models[model]; !ok { continue } parsed, has := localTime(ts, loc) if !flt.MatchesDayHour(parsed, has) { continue } counts[sessionID] += count } return rows.Err() }); err != nil { return nil, err } return counts, nil } func (db *DB) getAnalyticsActivityFilteredByModelTime( ctx context.Context, f AnalyticsFilter, granularity string, ) (ActivityResponse, error) { loc := f.location() dateCol := "COALESCE(NULLIF(started_at, ''), created_at)" where, args := f.buildWhereWithDate(dateCol, true, "sessions.id") var timeIDs map[string]bool if f.HasTimeFilter() { var err error timeIDs, err = db.filteredSessionIDs(ctx, f) if err != nil { return ActivityResponse{}, err } } rows, err := db.getReader().QueryContext(ctx, `SELECT id, `+dateCol+`, agent FROM sessions WHERE `+where, args...) if err != nil { return ActivityResponse{}, fmt.Errorf("querying analytics activity sessions: %w", err) } defer rows.Close() type sessionRow struct { id, date, agent string } sessions := make([]sessionRow, 0) sessionIDs := make([]string, 0) for rows.Next() { var id, ts, agent string if err := rows.Scan(&id, &ts, &agent); err != nil { return ActivityResponse{}, fmt.Errorf("scanning analytics activity session: %w", err) } date := localDate(ts, loc) if !inDateRange(date, f.From, f.To) { continue } if timeIDs != nil && !timeIDs[id] { continue } sessions = append(sessions, sessionRow{id: id, date: date, agent: agent}) sessionIDs = append(sessionIDs, id) } if err := rows.Err(); err != nil { return ActivityResponse{}, fmt.Errorf("iterating analytics activity sessions: %w", err) } messageStats, err := db.getAnalyticsFilteredMessageStats( ctx, sessionIDs, f, ) if err != nil { return ActivityResponse{}, err } toolCalls, err := db.getModelScopedToolCallCounts( ctx, sessionIDs, f, ) if err != nil { return ActivityResponse{}, err } buckets := make(map[string]*ActivityEntry) for _, session := range sessions { bucket := bucketDate(session.date, granularity) entry := buckets[bucket] if entry == nil { entry = &ActivityEntry{ Date: bucket, ByAgent: make(map[string]int), } buckets[bucket] = 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 += toolCalls[session.id] entry.ByAgent[session.agent] += stat.Messages } series := make([]ActivityEntry, 0, len(buckets)) for _, entry := range buckets { series = append(series, *entry) } sort.Slice(series, func(i, j int) bool { return series[i].Date < series[j].Date }) return ActivityResponse{ Granularity: granularity, Series: series, }, nil } // GetAnalyticsActivity returns session/message counts grouped // by time bucket. func (db *DB) GetAnalyticsActivity( ctx context.Context, f AnalyticsFilter, granularity string, ) (ActivityResponse, error) { if granularity == "" { granularity = "day" } if strings.TrimSpace(f.Model) != "" { return db.getAnalyticsActivityFilteredByModelTime( ctx, f, granularity, ) } loc := f.location() dateCol := "COALESCE(NULLIF(s.started_at, ''), s.created_at)" where, args := f.buildWhereWithDate(dateCol, true, "s.id") if modelPred, modelArgs := sqliteAnalyticsCSVPredicate( "m.model", f.Model, ); modelPred != "" { where += " AND " + modelPred args = append(args, modelArgs...) } var timeIDs map[string]bool if f.HasTimeFilter() { var err error timeIDs, err = db.filteredSessionIDs(ctx, f) if err != nil { return ActivityResponse{}, err } } query := `SELECT ` + dateCol + `, s.agent, s.id, m.role, m.has_thinking, m.is_system, COUNT(*) FROM sessions s LEFT JOIN messages m ON m.session_id = s.id WHERE ` + where + ` GROUP BY s.id, m.role, m.has_thinking, m.is_system` rows, err := db.getReader().QueryContext(ctx, query, args...) if err != nil { return ActivityResponse{}, fmt.Errorf("querying analytics activity: %w", err) } defer rows.Close() buckets := make(map[string]*ActivityEntry) sessionSeen := make(map[string]string) // session_id -> bucket var sessionIDs []string for rows.Next() { var ts, agent, sid string var role *string var hasThinking, isSystem *bool var count int if err := rows.Scan( &ts, &agent, &sid, &role, &hasThinking, &isSystem, &count, ); err != nil { return ActivityResponse{}, fmt.Errorf("scanning activity row: %w", err) } date := localDate(ts, loc) if !inDateRange(date, f.From, f.To) { continue } if timeIDs != nil && !timeIDs[sid] { continue } bucket := bucketDate(date, granularity) entry, ok := buckets[bucket] if !ok { entry = &ActivityEntry{ Date: bucket, ByAgent: make(map[string]int), } buckets[bucket] = entry } // Count this session once per bucket if _, seen := sessionSeen[sid]; !seen { sessionSeen[sid] = bucket sessionIDs = append(sessionIDs, sid) entry.Sessions++ } sys := isSystem != nil && *isSystem if role != nil { entry.Messages += count entry.ByAgent[agent] += count switch *role { case "user": if !sys { entry.UserMessages += count } case "assistant": entry.AssistantMessages += count } if hasThinking != nil && *hasThinking { entry.ThinkingMessages += count } } } if err := rows.Err(); err != nil { return ActivityResponse{}, fmt.Errorf("iterating activity rows: %w", err) } // Merge tool_call counts per session into buckets. if len(sessionIDs) > 0 { err = queryChunked(sessionIDs, func(chunk []string) error { return db.mergeActivityToolCalls( ctx, chunk, sessionSeen, buckets, f.Model, ) }) if err != nil { return ActivityResponse{}, err } } // Sort by date series := make([]ActivityEntry, 0, len(buckets)) for _, e := range buckets { series = append(series, *e) } sort.Slice(series, func(i, j int) bool { return series[i].Date < series[j].Date }) return ActivityResponse{ Granularity: granularity, Series: series, }, nil } // mergeActivityToolCalls queries tool_calls for a chunk of // session IDs and adds counts to the matching activity buckets. func (db *DB) mergeActivityToolCalls( ctx context.Context, chunk []string, sessionBucket map[string]string, buckets map[string]*ActivityEntry, model string, ) error { ph, args := inPlaceholders(chunk) q := `SELECT tc.session_id, COUNT(*) FROM tool_calls tc` if model != "" { q += ` JOIN messages m ON m.session_id = tc.session_id AND m.id = tc.message_id` } q += ` WHERE tc.session_id IN ` + ph if modelPred, modelArgs := sqliteAnalyticsCSVPredicate( "m.model", model, ); modelPred != "" { q += ` AND ` + modelPred args = append(args, modelArgs...) } q += ` GROUP BY tc.session_id` rows, err := db.getReader().QueryContext(ctx, q, args...) if err != nil { return fmt.Errorf( "querying activity 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 fmt.Errorf( "scanning activity tool_call: %w", err, ) } bucket := sessionBucket[sid] if entry, ok := buckets[bucket]; ok { entry.ToolCalls += count } } return rows.Err() } // --- Heatmap --- // HeatmapEntry is one day in the heatmap calendar. type HeatmapEntry struct { Date string `json:"date"` Value int `json:"value"` Level int `json:"level"` } // HeatmapLevels defines the quartile thresholds for levels 1-4. type HeatmapLevels struct { L1 int `json:"l1"` L2 int `json:"l2"` L3 int `json:"l3"` L4 int `json:"l4"` } // HeatmapResponse wraps the heatmap data. type HeatmapResponse struct { Metric string `json:"metric"` Entries []HeatmapEntry `json:"entries"` Levels HeatmapLevels `json:"levels"` EntriesFrom string `json:"entries_from"` } // GetAnalyticsHeatmap returns daily counts with intensity levels. func (db *DB) GetAnalyticsHeatmap( ctx context.Context, f AnalyticsFilter, metric string, ) (HeatmapResponse, error) { if metric == "" { metric = "messages" } loc := f.location() dateCol := "COALESCE(NULLIF(started_at, ''), created_at)" where, args := f.buildWhere(dateCol) var timeIDs map[string]bool if f.HasTimeFilter() { var err error timeIDs, err = db.filteredSessionIDs(ctx, f) if err != nil { return HeatmapResponse{}, err } } query := `SELECT id, ` + dateCol + `, message_count, total_output_tokens, has_total_output_tokens FROM sessions WHERE ` + where rows, err := db.getReader().QueryContext(ctx, query, args...) if err != nil { return HeatmapResponse{}, fmt.Errorf("querying analytics heatmap: %w", err) } defer rows.Close() type heatmapRow struct { id string date string messages int outputTokens int hasTokens bool } var heatmapRows []heatmapRow dayCounts := make(map[string]int) daySessions := make(map[string]int) dayOutputTokens := make(map[string]int) for rows.Next() { var id, ts string var mc, outputTokens int var hasTokens bool if err := rows.Scan( &id, &ts, &mc, &outputTokens, &hasTokens, ); err != nil { return HeatmapResponse{}, fmt.Errorf("scanning heatmap row: %w", err) } date := localDate(ts, loc) if !inDateRange(date, f.From, f.To) { continue } if timeIDs != nil && !timeIDs[id] { continue } heatmapRows = append(heatmapRows, heatmapRow{ id: id, date: date, messages: mc, outputTokens: outputTokens, hasTokens: hasTokens, }) } if err := rows.Err(); err != nil { return HeatmapResponse{}, fmt.Errorf("iterating heatmap rows: %w", err) } if f.Model != "" && (metric == "messages" || metric == "output_tokens") { ids := make([]string, 0, len(heatmapRows)) for _, row := range heatmapRows { ids = append(ids, row.id) } stats, err := db.getAnalyticsFilteredMessageStats(ctx, ids, f) if err != nil { return HeatmapResponse{}, err } for i := range heatmapRows { stat := stats[heatmapRows[i].id] heatmapRows[i].messages = stat.Messages heatmapRows[i].outputTokens = stat.OutputTokens heatmapRows[i].hasTokens = stat.HasOutputTokens } } for _, row := range heatmapRows { dayCounts[row.date] += row.messages daySessions[row.date]++ if row.hasTokens { dayOutputTokens[row.date] += row.outputTokens } } // Choose which map to use based on metric source := dayCounts switch metric { case "sessions": source = daySessions case "output_tokens": source = dayOutputTokens } // For output_tokens, an empty source means no sessions // reported token coverage. Return an empty heatmap so the // UI can show "no data" instead of a misleading zero grid. if metric == "output_tokens" && len(source) == 0 { return HeatmapResponse{ Metric: metric, EntriesFrom: clampFrom(f.From, f.To), }, nil } // Determine effective date range (clamped to MaxHeatmapDays) entriesFrom := clampFrom(f.From, f.To) // Collect non-zero values from the displayed range only, // so outliers outside the window don't skew intensity. var values []int for date, v := range source { if v > 0 && date >= entriesFrom && date <= f.To { values = append(values, v) } } sort.Ints(values) levels := computeQuartileLevels(values) // Build entries for each day in the clamped range entries := buildDateEntries( entriesFrom, f.To, source, levels, ) return HeatmapResponse{ Metric: metric, Entries: entries, Levels: levels, EntriesFrom: entriesFrom, }, nil } // computeQuartileLevels computes thresholds from sorted values. func computeQuartileLevels(sorted []int) HeatmapLevels { if len(sorted) == 0 { return HeatmapLevels{L1: 1, L2: 2, L3: 3, L4: 4} } n := len(sorted) return HeatmapLevels{ L1: sorted[0], L2: sorted[n/4], L3: sorted[n/2], L4: sorted[n*3/4], } } // assignLevel determines the heatmap level (0-4) for a value. func assignLevel(value int, levels 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 } // MaxHeatmapDays is the maximum number of day entries the // heatmap will return. Ranges exceeding this are clamped to // the most recent MaxHeatmapDays from the end date. const MaxHeatmapDays = 366 // clampFrom returns from clamped so that [from, to] spans at // most MaxHeatmapDays. If the range is already within bounds, // from is returned unchanged. func clampFrom(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, -(MaxHeatmapDays - 1)) if start.Before(earliest) { return earliest.Format("2006-01-02") } return from } // buildDateEntries creates a HeatmapEntry for each day in // [from, to]. The caller is responsible for clamping the // range via clampFrom before calling this function. func buildDateEntries( from, to string, values map[string]int, levels HeatmapLevels, ) []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 } var entries []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, HeatmapEntry{ Date: date, Value: v, Level: assignLevel(v, levels), }) } return entries } // --- Projects --- // ProjectAnalytics holds analytics for a single project. type ProjectAnalytics struct { Name string `json:"name"` Sessions int `json:"sessions"` Messages int `json:"messages"` FirstSession string `json:"first_session"` LastSession string `json:"last_session"` AvgMessages float64 `json:"avg_messages"` MedianMessages int `json:"median_messages"` Agents map[string]int `json:"agents"` DailyTrend float64 `json:"daily_trend"` } // ProjectsAnalyticsResponse wraps the projects list. type ProjectsAnalyticsResponse struct { Projects []ProjectAnalytics `json:"projects"` } // GetAnalyticsProjects returns per-project analytics. func (db *DB) GetAnalyticsProjects( ctx context.Context, f AnalyticsFilter, ) (ProjectsAnalyticsResponse, error) { // Per-project session/token breakdown is an aggregate, so subagent // sessions (including workflow subagents) are counted here. f.IncludeSubagents = true loc := f.location() dateCol := "COALESCE(NULLIF(started_at, ''), created_at)" where, args := f.buildWhere(dateCol) var timeIDs map[string]bool if f.HasTimeFilter() { var err error timeIDs, err = db.filteredSessionIDs(ctx, f) if err != nil { return ProjectsAnalyticsResponse{}, err } } query := `SELECT id, project, ` + dateCol + `, message_count, agent FROM sessions WHERE ` + where + ` ORDER BY project, ` + dateCol rows, err := db.getReader().QueryContext(ctx, query, args...) if err != nil { return ProjectsAnalyticsResponse{}, fmt.Errorf("querying analytics projects: %w", err) } defer rows.Close() type projectData struct { name string sessions int messages int first string last string counts []int agents map[string]int days map[string]int } projectMap := make(map[string]*projectData) var projectOrder []string type projectRow struct { id string project string date string messages int agent string } var projectRows []projectRow for rows.Next() { var id, project, ts, agent string var mc int if err := rows.Scan( &id, &project, &ts, &mc, &agent, ); err != nil { return ProjectsAnalyticsResponse{}, fmt.Errorf("scanning project row: %w", err) } date := localDate(ts, loc) if !inDateRange(date, f.From, f.To) { continue } if timeIDs != nil && !timeIDs[id] { continue } projectRows = append(projectRows, projectRow{ id: id, project: project, date: date, messages: mc, agent: agent, }) } if err := rows.Err(); err != nil { return ProjectsAnalyticsResponse{}, fmt.Errorf("iterating project rows: %w", err) } if f.Model != "" { ids := make([]string, 0, len(projectRows)) for _, row := range projectRows { ids = append(ids, row.id) } counts, err := db.getAnalyticsFilteredMessageCounts(ctx, ids, f) if err != nil { return ProjectsAnalyticsResponse{}, err } for i := range projectRows { projectRows[i].messages = counts[projectRows[i].id] } } for _, row := range projectRows { pd, ok := projectMap[row.project] if !ok { pd = &projectData{ name: row.project, agents: make(map[string]int), days: make(map[string]int), } projectMap[row.project] = pd projectOrder = append(projectOrder, row.project) } pd.sessions++ pd.messages += row.messages pd.counts = append(pd.counts, row.messages) pd.agents[row.agent]++ pd.days[row.date] += row.messages if pd.first == "" || row.date < pd.first { pd.first = row.date } if row.date > pd.last { pd.last = row.date } } projects := make([]ProjectAnalytics, 0, len(projectMap)) for _, name := range projectOrder { pd, ok := projectMap[name] if !ok || pd == nil { continue } sort.Ints(pd.counts) n := len(pd.counts) avg := 0.0 if n > 0 { avg = math.Round( float64(pd.messages)/float64(n)*10, ) / 10 } // Daily trend: messages per active day trend := 0.0 if len(pd.days) > 0 { trend = math.Round( float64(pd.messages)/float64(len(pd.days))*10, ) / 10 } projects = append(projects, ProjectAnalytics{ Name: pd.name, Sessions: pd.sessions, Messages: pd.messages, FirstSession: pd.first, LastSession: pd.last, AvgMessages: avg, MedianMessages: medianInt(pd.counts, n), Agents: pd.agents, DailyTrend: trend, }) } // Sort by message count descending sort.Slice(projects, func(i, j int) bool { return projects[i].Messages > projects[j].Messages }) return ProjectsAnalyticsResponse{Projects: projects}, nil } // --- Hour-of-Week --- // HourOfWeekCell is one cell in the 7x24 hour-of-week grid. type HourOfWeekCell struct { DayOfWeek int `json:"day_of_week"` // 0=Mon, 6=Sun Hour int `json:"hour"` // 0-23 Messages int `json:"messages"` } // HourOfWeekResponse wraps the hour-of-week heatmap data. type HourOfWeekResponse struct { Cells []HourOfWeekCell `json:"cells"` } // GetAnalyticsHourOfWeek returns message counts bucketed by // day-of-week and hour-of-day in the user's timezone. func (db *DB) GetAnalyticsHourOfWeek( ctx context.Context, f AnalyticsFilter, ) (HourOfWeekResponse, error) { if strings.TrimSpace(f.Model) != "" { return db.getAnalyticsHourOfWeekFilteredByModel(ctx, f) } loc := f.location() dateCol := "COALESCE(NULLIF(s.started_at, ''), s.created_at)" where, args := f.buildWhereWithDate(dateCol, true, "s.id") query := `SELECT ` + dateCol + `, m.timestamp FROM sessions s JOIN messages m ON m.session_id = s.id WHERE ` + where + ` AND m.timestamp != ''` rows, err := db.getReader().QueryContext(ctx, query, args...) if err != nil { return HourOfWeekResponse{}, fmt.Errorf("querying hour-of-week: %w", err) } defer rows.Close() var grid [7][24]int for rows.Next() { var sessTS, msgTS string if err := rows.Scan(&sessTS, &msgTS); err != nil { return HourOfWeekResponse{}, fmt.Errorf("scanning hour-of-week row: %w", err) } sessDate := localDate(sessTS, loc) if !inDateRange(sessDate, f.From, f.To) { continue } t, ok := localTime(msgTS, loc) if !ok { continue } // Go Sunday=0, convert to ISO Monday=0 dow := (int(t.Weekday()) + 6) % 7 grid[dow][t.Hour()]++ } if err := rows.Err(); err != nil { return HourOfWeekResponse{}, fmt.Errorf("iterating hour-of-week rows: %w", err) } return HourOfWeekResponseFromGrid(grid), nil } // getAnalyticsHourOfWeekFilteredByModel buckets model-scoped messages by // day-of-week and hour. Like every model-filtered panel it pairs empty-model // user turns with their selected-model assistant via the shared scope reducer, // so those user 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. // analyticsModelCandidateSessionIDs returns the date-filtered, model-scoped // session IDs that feed the shared message-scope reducer. The day/hour filter // is intentionally not applied here: callers that need it let the reducer // apply it (so paired empty-model user turns are kept), while the hour-of-week // heatmap clears it to show the full grid. func (db *DB) analyticsModelCandidateSessionIDs( ctx context.Context, f AnalyticsFilter, ) ([]string, error) { loc := f.location() dateCol := "COALESCE(NULLIF(started_at, ''), created_at)" where, args := f.buildWhereWithDate(dateCol, true, "sessions.id") rows, err := db.getReader().QueryContext(ctx, `SELECT id, `+dateCol+` FROM sessions WHERE `+where, args...) if err != nil { return nil, fmt.Errorf("querying model candidate sessions: %w", err) } defer rows.Close() ids := make([]string, 0) for rows.Next() { var id, ts string if err := rows.Scan(&id, &ts); err != nil { return nil, fmt.Errorf("scanning model candidate session: %w", err) } if !inDateRange(localDate(ts, loc), f.From, f.To) { continue } ids = append(ids, id) } if err := rows.Err(); err != nil { return nil, fmt.Errorf("iterating model candidate sessions: %w", err) } return ids, nil } func (db *DB) getAnalyticsHourOfWeekFilteredByModel( ctx context.Context, f AnalyticsFilter, ) (HourOfWeekResponse, error) { sessionIDs, err := db.analyticsModelCandidateSessionIDs(ctx, f) if err != nil { return HourOfWeekResponse{}, err } scopeFilter := f scopeFilter.DayOfWeek = nil scopeFilter.Hour = nil scope, err := db.resolveAnalyticsMessageScope( ctx, sessionIDs, scopeFilter, false, ) if err != nil { return HourOfWeekResponse{}, err } var grid [7][24]int if scope != nil { for _, msgs := range scope.MessagesBySession() { for _, m := range msgs { if !m.HasLocalTime { continue } // Go Sunday=0, convert to ISO Monday=0 dow := (int(m.LocalTime.Weekday()) + 6) % 7 grid[dow][m.LocalTime.Hour()]++ } } } return HourOfWeekResponseFromGrid(grid), nil } // HourOfWeekResponseFromGrid flattens the 7x24 grid into the dense 168-cell // response. func HourOfWeekResponseFromGrid(grid [7][24]int) HourOfWeekResponse { cells := make([]HourOfWeekCell, 0, 168) for d := range 7 { for h := range 24 { cells = append(cells, HourOfWeekCell{ DayOfWeek: d, Hour: h, Messages: grid[d][h], }) } } return HourOfWeekResponse{Cells: cells} } // --- Session Shape --- // DistributionBucket is a labeled count for histogram display. type DistributionBucket struct { Label string `json:"label"` Count int `json:"count"` } // SessionShapeResponse holds distribution histograms for session // characteristics. type SessionShapeResponse struct { Count int `json:"count"` LengthDistribution []DistributionBucket `json:"length_distribution"` DurationDistribution []DistributionBucket `json:"duration_distribution"` AutonomyDistribution []DistributionBucket `json:"autonomy_distribution"` } // lengthBucket returns the bucket label for a message count. 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+" } } // durationBucket returns the bucket label for a duration in // minutes. 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+" } } // autonomyBucket returns the bucket label for an autonomy ratio. 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+" } } // bucketOrder maps label → order index for consistent output. var ( lengthOrder = map[string]int{ "1-5": 0, "6-15": 1, "16-30": 2, "31-60": 3, "61-120": 4, "121+": 5, } durationOrder = map[string]int{ "<5m": 0, "5-15m": 1, "15-30m": 2, "30-60m": 3, "1-2h": 4, "2h+": 5, } autonomyOrder = map[string]int{ "<0.5": 0, "0.5-1": 1, "1-2": 2, "2-5": 3, "5-10": 4, "10+": 5, } ) // sortBuckets sorts distribution buckets by their defined order. func sortBuckets( buckets []DistributionBucket, order map[string]int, ) { sort.Slice(buckets, func(i, j int) bool { return order[buckets[i].Label] < order[buckets[j].Label] }) } // mapToBuckets converts a label→count map to sorted buckets. func mapToBuckets( m map[string]int, order map[string]int, ) []DistributionBucket { buckets := make([]DistributionBucket, 0, len(m)) for label, count := range m { buckets = append(buckets, DistributionBucket{ Label: label, Count: count, }) } sortBuckets(buckets, order) return buckets } // GetAnalyticsSessionShape returns distribution histograms for // session length, duration, and autonomy ratio. func (db *DB) GetAnalyticsSessionShape( ctx context.Context, f AnalyticsFilter, ) (SessionShapeResponse, error) { loc := f.location() modelFilter := strings.TrimSpace(f.Model) != "" dateCol := "COALESCE(NULLIF(started_at, ''), created_at)" where, args := f.buildWhere(dateCol) var timeIDs map[string]bool if f.HasTimeFilter() { var err error timeIDs, err = db.filteredSessionIDs(ctx, f) if err != nil { return SessionShapeResponse{}, err } } query := `SELECT ` + dateCol + `, started_at, ended_at, message_count, id FROM sessions WHERE ` + where rows, err := db.getReader().QueryContext(ctx, query, args...) if err != nil { return SessionShapeResponse{}, fmt.Errorf("querying session shape: %w", err) } defer rows.Close() lengthCounts := make(map[string]int) durationCounts := make(map[string]int) var sessionIDs []string totalCount := 0 for rows.Next() { var ts string var startedAt, endedAt *string var mc int var id string if err := rows.Scan( &ts, &startedAt, &endedAt, &mc, &id, ); err != nil { return SessionShapeResponse{}, fmt.Errorf("scanning session shape row: %w", err) } date := localDate(ts, loc) if !inDateRange(date, f.From, f.To) { continue } if timeIDs != nil && !timeIDs[id] { continue } totalCount++ if !modelFilter { lengthCounts[lengthBucket(mc)]++ } sessionIDs = append(sessionIDs, id) if startedAt != nil && endedAt != nil && *startedAt != "" && *endedAt != "" { tStart, okS := localTime(*startedAt, loc) tEnd, okE := localTime(*endedAt, loc) if okS && okE { mins := tEnd.Sub(tStart).Minutes() if mins >= 0 { durationCounts[durationBucket(mins)]++ } } } } if err := rows.Err(); err != nil { return SessionShapeResponse{}, fmt.Errorf("iterating session shape rows: %w", err) } autonomyCounts := make(map[string]int) if modelFilter && len(sessionIDs) > 0 { stats, err := db.getAnalyticsFilteredMessageStats( ctx, sessionIDs, f, ) if err != nil { return SessionShapeResponse{}, err } lengthCounts = make(map[string]int) seen := make(map[string]struct{}, len(sessionIDs)) for _, sessionID := range sessionIDs { if _, ok := seen[sessionID]; ok { continue } seen[sessionID] = struct{}{} stat := stats[sessionID] lengthCounts[lengthBucket(stat.Messages)]++ if stat.UserMessages > 0 { ratio := float64(stat.ToolUseMessages) / float64(stat.UserMessages) autonomyCounts[autonomyBucket(ratio)]++ } } } else if len(sessionIDs) > 0 { err := queryChunked(sessionIDs, func(chunk []string) error { return db.queryAutonomyChunk( ctx, chunk, autonomyCounts, ) }) if err != nil { return SessionShapeResponse{}, err } } return SessionShapeResponse{ Count: totalCount, LengthDistribution: mapToBuckets(lengthCounts, lengthOrder), DurationDistribution: mapToBuckets(durationCounts, durationOrder), AutonomyDistribution: mapToBuckets(autonomyCounts, autonomyOrder), }, nil } // queryAutonomyChunk queries autonomy stats for a chunk of // session IDs and accumulates results into counts. func (db *DB) queryAutonomyChunk( ctx context.Context, chunk []string, counts map[string]int, ) error { ph, args := inPlaceholders(chunk) q := `SELECT session_id, SUM(CASE WHEN role='user' AND is_system=0 THEN 1 ELSE 0 END), SUM(CASE WHEN role='assistant' AND has_tool_use=1 THEN 1 ELSE 0 END) FROM messages WHERE session_id IN ` + ph + ` GROUP BY session_id` rows, err := db.getReader().QueryContext(ctx, q, args...) if err != nil { return fmt.Errorf("querying autonomy: %w", err) } defer rows.Close() for rows.Next() { var sid string var userCount, toolCount int if err := rows.Scan( &sid, &userCount, &toolCount, ); err != nil { return fmt.Errorf("scanning autonomy row: %w", err) } if userCount > 0 { ratio := float64(toolCount) / float64(userCount) counts[autonomyBucket(ratio)]++ } } return rows.Err() } // --- Tools --- // ToolCategoryCount holds a count and percentage for one tool // category. type ToolCategoryCount struct { Category string `json:"category"` Count int `json:"count"` Pct float64 `json:"pct"` } // ToolAgentBreakdown holds tool usage breakdown for one agent. type ToolAgentBreakdown struct { Agent string `json:"agent"` Total int `json:"total"` Categories []ToolCategoryCount `json:"categories"` } // ToolUsageAnalysis holds ranked usage for one concrete tool name. type ToolUsageAnalysis struct { ToolName string `json:"tool_name"` Category string `json:"category"` CallCount int `json:"call_count"` SessionCount int `json:"session_count"` Pct float64 `json:"pct"` } // ToolTrendEntry holds tool call counts for one time bucket. type ToolTrendEntry struct { Date string `json:"date"` ByCat map[string]int `json:"by_category"` } // ToolsAnalyticsResponse wraps tool usage analytics. type ToolsAnalyticsResponse struct { TotalCalls int `json:"total_calls"` ByCategory []ToolCategoryCount `json:"by_category"` ByAgent []ToolAgentBreakdown `json:"by_agent"` ByTool []ToolUsageAnalysis `json:"by_tool"` Trend []ToolTrendEntry `json:"trend"` } // ToolAnalyticsRow is a backend-neutral intermediate row used to // aggregate concrete tool usage after native stores apply their filters. type ToolAnalyticsRow struct { SessionID string ToolName string Category string Agent string Date string Count int } // SkillAgentBreakdown holds skill usage for one agent. type SkillAgentBreakdown struct { Agent string `json:"agent"` Count int `json:"count"` } // SkillProjectBreakdown holds skill usage for one project. type SkillProjectBreakdown struct { Project string `json:"project"` Count int `json:"count"` } // SkillUsage holds usage metrics for one skill name. type SkillUsage struct { SkillName string `json:"skill_name"` CallCount int `json:"call_count"` SessionCount int `json:"session_count"` AgentBreakdown []SkillAgentBreakdown `json:"agent_breakdown"` ProjectBreakdown []SkillProjectBreakdown `json:"project_breakdown"` LastUsedAt string `json:"last_used_at"` Pct float64 `json:"pct"` } // SkillTrendEntry holds skill call counts for one time bucket. type SkillTrendEntry struct { Date string `json:"date"` BySkill map[string]int `json:"by_skill"` } // SkillsAnalyticsResponse wraps skill usage analytics. type SkillsAnalyticsResponse struct { TotalSkillCalls int `json:"total_skill_calls"` DistinctSkills int `json:"distinct_skills"` BySkill []SkillUsage `json:"by_skill"` Trend []SkillTrendEntry `json:"trend"` } // SkillAnalyticsRow is a backend-neutral intermediate row used to // aggregate skill usage after each store applies its native filters. type SkillAnalyticsRow struct { SessionID string SkillName string Agent string Project string Date string LastUsedAt string Count int } type toolUsageAccumulator struct { toolName string category string callCount int sessionIDs map[string]struct{} } // BuildToolsAnalytics folds backend-neutral tool rows into the public // response shape. Tool names are trimmed and blank names are reported as // Unknown so legacy rows remain visible instead of disappearing. func BuildToolsAnalytics(rows []ToolAnalyticsRow) ToolsAnalyticsResponse { resp := ToolsAnalyticsResponse{ ByCategory: []ToolCategoryCount{}, ByAgent: []ToolAgentBreakdown{}, ByTool: []ToolUsageAnalysis{}, Trend: []ToolTrendEntry{}, } if len(rows) == 0 { return resp } catCounts := make(map[string]int) agentCats := make(map[string]map[string]int) trendBuckets := make(map[string]map[string]int) toolCounts := make(map[string]*toolUsageAccumulator) for _, row := range rows { if row.Count <= 0 { continue } catCounts[row.Category] += row.Count resp.TotalCalls += row.Count if agentCats[row.Agent] == nil { agentCats[row.Agent] = make(map[string]int) } agentCats[row.Agent][row.Category] += row.Count week := bucketDate(row.Date, "week") if trendBuckets[week] == nil { trendBuckets[week] = make(map[string]int) } trendBuckets[week][row.Category] += row.Count name := strings.TrimSpace(row.ToolName) if name == "" { name = "Unknown" } key := row.Category + "\x00" + name acc := toolCounts[key] if acc == nil { acc = &toolUsageAccumulator{ toolName: name, category: row.Category, sessionIDs: map[string]struct{}{}, } toolCounts[key] = acc } acc.callCount += row.Count if row.SessionID != "" { acc.sessionIDs[row.SessionID] = struct{}{} } } if resp.TotalCalls == 0 { return resp } resp.ByCategory = make([]ToolCategoryCount, 0, len(catCounts)) for cat, count := range catCounts { pct := math.Round( float64(count)/float64(resp.TotalCalls)*1000, ) / 10 resp.ByCategory = append(resp.ByCategory, ToolCategoryCount{ Category: cat, Count: count, Pct: pct, }) } sort.Slice(resp.ByCategory, func(i, j int) bool { if resp.ByCategory[i].Count != resp.ByCategory[j].Count { return resp.ByCategory[i].Count > resp.ByCategory[j].Count } return resp.ByCategory[i].Category < resp.ByCategory[j].Category }) agentKeys := make([]string, 0, len(agentCats)) for agent := range agentCats { agentKeys = append(agentKeys, agent) } sort.Strings(agentKeys) resp.ByAgent = make([]ToolAgentBreakdown, 0, len(agentKeys)) for _, agent := range agentKeys { cats := agentCats[agent] total := 0 for _, c := range cats { total += c } catList := make([]ToolCategoryCount, 0, len(cats)) for cat, count := range cats { pct := math.Round( float64(count)/float64(total)*1000, ) / 10 catList = append(catList, ToolCategoryCount{ Category: cat, Count: count, Pct: pct, }) } sort.Slice(catList, func(i, j int) bool { if catList[i].Count != catList[j].Count { return catList[i].Count > catList[j].Count } return catList[i].Category < catList[j].Category }) resp.ByAgent = append(resp.ByAgent, ToolAgentBreakdown{ Agent: agent, Total: total, Categories: catList, }) } resp.ByTool = make([]ToolUsageAnalysis, 0, len(toolCounts)) for _, acc := range toolCounts { pct := math.Round( float64(acc.callCount)/float64(resp.TotalCalls)*1000, ) / 10 resp.ByTool = append(resp.ByTool, ToolUsageAnalysis{ ToolName: acc.toolName, Category: acc.category, CallCount: acc.callCount, SessionCount: len(acc.sessionIDs), Pct: pct, }) } sort.Slice(resp.ByTool, func(i, j int) bool { if resp.ByTool[i].CallCount != resp.ByTool[j].CallCount { return resp.ByTool[i].CallCount > resp.ByTool[j].CallCount } if resp.ByTool[i].ToolName != resp.ByTool[j].ToolName { return resp.ByTool[i].ToolName < resp.ByTool[j].ToolName } return resp.ByTool[i].Category < resp.ByTool[j].Category }) resp.Trend = make([]ToolTrendEntry, 0, len(trendBuckets)) for week, cats := range trendBuckets { resp.Trend = append(resp.Trend, ToolTrendEntry{ Date: week, ByCat: cats, }) } sort.Slice(resp.Trend, func(i, j int) bool { return resp.Trend[i].Date < resp.Trend[j].Date }) return resp } type skillUsageAccumulator struct { callCount int sessionIDs map[string]struct{} agentCounts map[string]int projectCounts map[string]int lastUsedAt string } // timestampAfter reports whether timestamp a is chronologically later // than b. Both are parsed as UTC timestamps so callers stay correct when // stores feed differing precisions (for example fractional seconds). It // falls back to lexical comparison only when a value cannot be parsed. func timestampAfter(a, b string) bool { if b == "" { return a != "" } if a == "" { return false } ta, aok := localTime(a, time.UTC) tb, bok := localTime(b, time.UTC) if aok && bok { return ta.After(tb) } return a > b } // skillsTrendGranularity normalizes a skills trend granularity value. // Empty defaults to week, the endpoint's historical bucket size. func skillsTrendGranularity(granularity string) string { switch granularity { case "day", "week", "month": return granularity default: return "week" } } // BuildSkillsAnalytics folds backend-neutral skill rows into the public // response shape. Skill names are trimmed and empty names are ignored. // granularity picks the trend bucket size (day, week, or month); empty // or unknown values fall back to week. When from and to are present, the // trend includes every bucket in that range, including zero-usage buckets. func BuildSkillsAnalytics( rows []SkillAnalyticsRow, from, to, granularity string, ) SkillsAnalyticsResponse { resp := SkillsAnalyticsResponse{ BySkill: []SkillUsage{}, Trend: []SkillTrendEntry{}, } bucket := skillsTrendGranularity(granularity) bySkill := map[string]*skillUsageAccumulator{} trendBuckets := map[string]map[string]int{} for _, row := range rows { name := strings.TrimSpace(row.SkillName) if name == "" || row.Count <= 0 { continue } acc := bySkill[name] if acc == nil { acc = &skillUsageAccumulator{ sessionIDs: map[string]struct{}{}, agentCounts: map[string]int{}, projectCounts: map[string]int{}, } bySkill[name] = acc } acc.callCount += row.Count resp.TotalSkillCalls += row.Count if row.SessionID != "" { acc.sessionIDs[row.SessionID] = struct{}{} } if row.Agent != "" { acc.agentCounts[row.Agent] += row.Count } if row.Project != "" { acc.projectCounts[row.Project] += row.Count } if timestampAfter(row.LastUsedAt, acc.lastUsedAt) { acc.lastUsedAt = row.LastUsedAt } if row.Date != "" { date := bucketDate(row.Date, bucket) if trendBuckets[date] == nil { trendBuckets[date] = map[string]int{} } trendBuckets[date][name] += row.Count } } resp.DistinctSkills = len(bySkill) if resp.DistinctSkills == 0 { return resp } for _, entry := range TrendBucketRange(from, to, bucket) { if trendBuckets[entry.Date] == nil { trendBuckets[entry.Date] = map[string]int{} } } for name, acc := range bySkill { usage := SkillUsage{ SkillName: name, CallCount: acc.callCount, SessionCount: len(acc.sessionIDs), AgentBreakdown: skillAgentBreakdowns(acc.agentCounts), ProjectBreakdown: skillProjectBreakdowns(acc.projectCounts), LastUsedAt: acc.lastUsedAt, Pct: math.Round( float64(acc.callCount)/ float64(resp.TotalSkillCalls)*1000, ) / 10, } resp.BySkill = append(resp.BySkill, usage) } sort.Slice(resp.BySkill, func(i, j int) bool { if resp.BySkill[i].CallCount != resp.BySkill[j].CallCount { return resp.BySkill[i].CallCount > resp.BySkill[j].CallCount } return resp.BySkill[i].SkillName < resp.BySkill[j].SkillName }) for date, skills := range trendBuckets { resp.Trend = append(resp.Trend, SkillTrendEntry{ Date: date, BySkill: skills, }) } sort.Slice(resp.Trend, func(i, j int) bool { return resp.Trend[i].Date < resp.Trend[j].Date }) return resp } func skillAgentBreakdowns( counts map[string]int, ) []SkillAgentBreakdown { out := make([]SkillAgentBreakdown, 0, len(counts)) for agent, count := range counts { out = append(out, SkillAgentBreakdown{ Agent: agent, Count: count, }) } sort.Slice(out, func(i, j int) bool { if out[i].Count != out[j].Count { return out[i].Count > out[j].Count } return out[i].Agent < out[j].Agent }) return out } func skillProjectBreakdowns( counts map[string]int, ) []SkillProjectBreakdown { out := make([]SkillProjectBreakdown, 0, len(counts)) for project, count := range counts { out = append(out, SkillProjectBreakdown{ Project: project, Count: count, }) } sort.Slice(out, func(i, j int) bool { if out[i].Count != out[j].Count { return out[i].Count > out[j].Count } return out[i].Project < out[j].Project }) return out } func analyticsToolsQuery( placeholders string, modelPred string, includeMessageMeta bool, ) string { query := `SELECT tc.session_id, tc.category, TRIM(COALESCE(tc.tool_name, '')), COUNT(*)` if includeMessageMeta { query += `, COALESCE(m.timestamp, '')` } query += ` FROM tool_calls tc` if includeMessageMeta { query += ` LEFT JOIN messages m ON m.session_id = tc.session_id AND m.id = tc.message_id` } query += ` WHERE tc.session_id IN ` + placeholders if modelPred != "" { query += ` AND ` + modelPred } query += ` GROUP BY tc.session_id, tc.category, TRIM(COALESCE(tc.tool_name, ''))` if includeMessageMeta { query += `, COALESCE(m.timestamp, '')` } return query } func analyticsSkillsQuery( placeholders string, modelPred string, ) string { query := `SELECT tc.session_id, TRIM(tc.skill_name), COUNT(*), COALESCE(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 ` + placeholders + ` AND TRIM(COALESCE(tc.skill_name, '')) != ''` if modelPred != "" { query += ` AND ` + modelPred } query += ` GROUP BY tc.session_id, TRIM(tc.skill_name), COALESCE(m.timestamp, '')` return query } // GetAnalyticsTools returns tool usage analytics aggregated // from the tool_calls table. func (db *DB) GetAnalyticsTools( ctx context.Context, f AnalyticsFilter, ) (ToolsAnalyticsResponse, error) { dateCol := "COALESCE(NULLIF(started_at, ''), created_at)" where, args := f.buildWhereWithoutDate() // Fetch filtered session IDs and their metadata. sessQ := `SELECT id, ` + dateCol + `, agent FROM sessions WHERE ` + where sessRows, err := db.getReader().QueryContext(ctx, sessQ, args...) if err != nil { return ToolsAnalyticsResponse{}, fmt.Errorf("querying tool sessions: %w", err) } defer sessRows.Close() type sessInfo struct { ts string agent string } sessionMap := make(map[string]sessInfo) var sessionIDs []string for sessRows.Next() { var id, ts, agent string if err := sessRows.Scan(&id, &ts, &agent); err != nil { return ToolsAnalyticsResponse{}, fmt.Errorf("scanning tool session: %w", err) } sessionMap[id] = sessInfo{ts: ts, agent: agent} sessionIDs = append(sessionIDs, id) } if err := sessRows.Err(); err != nil { return ToolsAnalyticsResponse{}, fmt.Errorf("iterating tool sessions: %w", err) } resp := ToolsAnalyticsResponse{ ByCategory: []ToolCategoryCount{}, ByAgent: []ToolAgentBreakdown{}, ByTool: []ToolUsageAnalysis{}, Trend: []ToolTrendEntry{}, } if len(sessionIDs) == 0 { return resp, nil } // Query tool_calls for filtered sessions (chunked). var toolRows []ToolAnalyticsRow err = queryChunked(sessionIDs, func(chunk []string) error { ph, chunkArgs := inPlaceholders(chunk) modelPred, modelArgs := sqliteAnalyticsCSVPredicate( "m.model", f.Model, ) chunkArgs = append(chunkArgs, modelArgs...) q := analyticsToolsQuery(ph, modelPred, true) rows, qErr := db.getReader().QueryContext( ctx, q, chunkArgs..., ) if qErr != nil { return fmt.Errorf( "querying tool_calls: %w", qErr, ) } defer rows.Close() for rows.Next() { var sid, cat, toolName, ts string var count int if err := rows.Scan( &sid, &cat, &toolName, &count, &ts, ); err != nil { return fmt.Errorf( "scanning tool_call: %w", err, ) } info, ok := sessionMap[sid] if !ok { continue } _, date, keep := f.ResolveSkillRowTime( ts, info.ts, ) if !keep { continue } toolRows = append(toolRows, ToolAnalyticsRow{ SessionID: sid, Category: cat, ToolName: toolName, Agent: info.agent, Count: count, Date: date, }) } return rows.Err() }) if err != nil { return ToolsAnalyticsResponse{}, err } if len(toolRows) == 0 { return resp, nil } return BuildToolsAnalytics(toolRows), nil } // ResolveSkillRowTime resolves the timestamp for a single skill call and // applies the date and hour/day-of-week filters to it. The message // timestamp is authoritative; the session timestamp is used only when the // message has none. Because skill rows are bucketed by the call's own // timestamp, the date/time filters must be applied here rather than to the // owning session, so a session that started outside the range still // contributes its in-range calls and drops its out-of-range ones. // // It returns the resolved timestamp, its local date, and whether the call // passes the filters. func (f AnalyticsFilter) ResolveSkillRowTime( messageTS, sessionTS string, ) (usedTS, date string, keep bool) { loc := f.location() usedTS = messageTS if strings.TrimSpace(usedTS) == "" { usedTS = sessionTS } date = localDate(usedTS, loc) if !inDateRange(date, f.From, f.To) { return usedTS, date, false } if f.HasTimeFilter() { t, ok := localTime(usedTS, loc) if !ok || !f.matchesTimeFilter(t) { return usedTS, date, false } } return usedTS, date, true } // GetAnalyticsSkills returns skill usage analytics aggregated // from non-empty tool_calls.skill_name values. granularity picks the // trend bucket size (day, week, or month); empty defaults to week. func (db *DB) GetAnalyticsSkills( ctx context.Context, f AnalyticsFilter, granularity string, ) (SkillsAnalyticsResponse, error) { dateCol := "COALESCE(NULLIF(started_at, ''), created_at)" where, args := f.buildWhereWithoutDate() sessQ := `SELECT id, ` + dateCol + `, agent, project FROM sessions WHERE ` + where sessRows, err := db.getReader().QueryContext(ctx, sessQ, args...) if err != nil { return SkillsAnalyticsResponse{}, fmt.Errorf("querying skill sessions: %w", err) } defer sessRows.Close() type sessInfo struct { ts string agent string project string } sessionMap := make(map[string]sessInfo) var sessionIDs []string for sessRows.Next() { var id, ts, agent, project string if err := sessRows.Scan( &id, &ts, &agent, &project, ); err != nil { return SkillsAnalyticsResponse{}, fmt.Errorf("scanning skill session: %w", err) } sessionMap[id] = sessInfo{ ts: ts, agent: agent, project: project, } sessionIDs = append(sessionIDs, id) } if err := sessRows.Err(); err != nil { return SkillsAnalyticsResponse{}, fmt.Errorf("iterating skill sessions: %w", err) } if len(sessionIDs) == 0 { return BuildSkillsAnalytics(nil, f.From, f.To, granularity), nil } var skillRows []SkillAnalyticsRow err = queryChunked(sessionIDs, func(chunk []string) error { ph, chunkArgs := inPlaceholders(chunk) modelPred, modelArgs := sqliteAnalyticsCSVPredicate( "m.model", f.Model, ) chunkArgs = append(chunkArgs, modelArgs...) q := analyticsSkillsQuery(ph, modelPred) rows, qErr := db.getReader().QueryContext( ctx, q, chunkArgs..., ) if qErr != nil { return fmt.Errorf( "querying skill tool_calls: %w", qErr, ) } defer rows.Close() for rows.Next() { var sid, skill, lastTS string var count int if err := rows.Scan( &sid, &skill, &count, &lastTS, ); err != nil { return fmt.Errorf( "scanning skill tool_call: %w", err, ) } info := sessionMap[sid] usedTS, date, keep := f.ResolveSkillRowTime( lastTS, info.ts, ) if !keep { continue } skillRows = append(skillRows, SkillAnalyticsRow{ SessionID: sid, SkillName: skill, Agent: info.agent, Project: info.project, Date: date, LastUsedAt: usedTS, Count: count, }) } return rows.Err() }) if err != nil { return SkillsAnalyticsResponse{}, err } return BuildSkillsAnalytics( skillRows, f.From, f.To, granularity, ), nil } // --- Velocity --- // velocityMsg holds per-message data needed for velocity // calculations. type velocityMsg struct { role string ts time.Time valid bool contentLength int } // queryVelocityMsgs fetches messages for a chunk of session IDs // and appends them to sessionMsgs, keyed by session ID. func (db *DB) queryVelocityMsgs( ctx context.Context, chunk []string, loc *time.Location, sessionMsgs map[string][]velocityMsg, ) error { ph, args := inPlaceholders(chunk) // COALESCE the nullable timestamp column to '' so a NULL (only present // on imported/migrated archives) does not fail rows.Scan with // "converting NULL to string is unsupported". localTime treats "" as // invalid, so the row is excluded from velocity stats rather than // crashing the analytics endpoint. This matches the NULL-safe // PostgreSQL and DuckDB velocity twins. q := `SELECT session_id, ordinal, role, COALESCE(timestamp, ''), content_length FROM messages WHERE session_id IN ` + ph + ` ORDER BY session_id, ordinal` rows, err := db.getReader().QueryContext(ctx, q, args...) if err != nil { return fmt.Errorf( "querying velocity messages: %w", err, ) } defer rows.Close() for rows.Next() { var sid string var ordinal int var role, ts string var cl int if err := rows.Scan( &sid, &ordinal, &role, &ts, &cl, ); err != nil { return fmt.Errorf( "scanning velocity msg: %w", err, ) } t, ok := localTime(ts, loc) sessionMsgs[sid] = append(sessionMsgs[sid], velocityMsg{ role: role, ts: t, valid: ok, contentLength: cl, }) } return rows.Err() } func (db *DB) getAnalyticsVelocityMessages( ctx context.Context, sessionIDs []string, f AnalyticsFilter, ) (map[string][]velocityMsg, error) { sessionMsgs := make(map[string][]velocityMsg, len(sessionIDs)) if len(sessionIDs) == 0 { return sessionMsgs, nil } loc := f.location() if strings.TrimSpace(f.Model) == "" { err := queryChunked(sessionIDs, func(chunk []string) error { return db.queryVelocityMsgs(ctx, chunk, loc, sessionMsgs) }) return sessionMsgs, err } scope, err := db.resolveAnalyticsMessageScope(ctx, sessionIDs, f, false) if err != nil { return nil, err } if scope == nil { return sessionMsgs, nil } for sessionID, rows := range scope.TimingBySession() { for _, row := range rows { sessionMsgs[sessionID] = append(sessionMsgs[sessionID], velocityMsg{ role: row.Role, ts: row.Time, valid: row.Valid, contentLength: row.ContentLength, }) } } return sessionMsgs, nil } // Percentiles holds p50 and p90 values. type Percentiles struct { P50 float64 `json:"p50"` P90 float64 `json:"p90"` } // VelocityOverview holds aggregate velocity metrics. type VelocityOverview struct { TurnCycleSec Percentiles `json:"turn_cycle_sec"` FirstResponseSec Percentiles `json:"first_response_sec"` MsgsPerActiveMin float64 `json:"msgs_per_active_min"` CharsPerActiveMin float64 `json:"chars_per_active_min"` ToolCallsPerActiveMin float64 `json:"tool_calls_per_active_min"` } // VelocityBreakdown is velocity metrics for a subgroup. type VelocityBreakdown struct { Label string `json:"label"` Sessions int `json:"sessions"` Overview VelocityOverview `json:"overview"` } // VelocityResponse wraps overall and grouped velocity metrics. type VelocityResponse struct { Overall VelocityOverview `json:"overall"` ByAgent []VelocityBreakdown `json:"by_agent"` ByComplexity []VelocityBreakdown `json:"by_complexity"` } // complexityBucket returns the complexity label based on // message count. func complexityBucket(mc int) string { switch { case mc <= 15: return "1-15" case mc <= 60: return "16-60" default: return "61+" } } // velocityAccumulator collects raw values for a velocity group. type velocityAccumulator struct { turnCycles []float64 firstResponses []float64 totalMsgs int totalChars int totalToolCalls int activeMinutes float64 sessions int } // populateVelocityAccumulator fetches per-message timestamps and tool // counts for the given sessions and feeds them through // processSessionVelocity into a single accumulator. Used by // GetSessionStats, which already has its filtered session list and // only needs the overall velocity slice — no agent/complexity // breakdowns. Sessions with fewer than two messages are silently // skipped, matching GetAnalyticsVelocity. func populateVelocityAccumulator( ctx context.Context, db *DB, sessionIDs []string, loc *time.Location, ) (*velocityAccumulator, error) { accum := &velocityAccumulator{} if len(sessionIDs) == 0 { return accum, nil } sessionMsgs := make(map[string][]velocityMsg) if err := queryChunked(sessionIDs, func(chunk []string) error { return db.queryVelocityMsgs( ctx, chunk, loc, sessionMsgs, ) }); err != nil { return nil, err } toolCountMap := make(map[string]int) err := queryChunked(sessionIDs, func(chunk []string) error { ph, chunkArgs := inPlaceholders(chunk) q := `SELECT session_id, COUNT(*) FROM tool_calls WHERE session_id IN ` + ph + ` GROUP BY session_id` rows, qErr := db.getReader().QueryContext( ctx, q, chunkArgs..., ) if qErr != nil { return fmt.Errorf( "querying velocity tool_calls: %w", qErr, ) } defer rows.Close() for rows.Next() { var sid string var count int if err := rows.Scan(&sid, &count); err != nil { return fmt.Errorf( "scanning velocity tool_call: %w", err, ) } toolCountMap[sid] = count } return rows.Err() }) if err != nil { return nil, err } for _, sid := range sessionIDs { msgs := sessionMsgs[sid] if len(msgs) < 2 { continue } processSessionVelocity( []*velocityAccumulator{accum}, msgs, toolCountMap[sid], ) } return accum, nil } // processSessionVelocity updates every accumulator in accums with one // session's turn cycles, first response, and throughput contribution. // Shared by GetAnalyticsVelocity (which tracks overall/byAgent/ // byComplexity) and GetSessionStats (which tracks a single overall). // // Caller must pass len(msgs) >= 2 in ordinal order. The function // itself bumps each accumulator's sessions counter. func processSessionVelocity( accums []*velocityAccumulator, msgs []velocityMsg, toolCount int, ) { const maxCycleSec = 1800.0 // Shared with the Top Sessions "active duration" SQL so the two // "active" definitions stay in lockstep. const maxGapSec = ActiveGapCapSec for _, a := range accums { a.sessions++ } // Turn cycles: user→assistant transitions 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 _, a := range accums { a.turnCycles = append(a.turnCycles, delta) } } } } // First response: first user → first assistant after it. // Scan by ordinal (conversation order), not timestamp. var firstUser, firstAsst *velocityMsg 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() // Clamp negative deltas to 0: ordinal order is // authoritative, so a negative delta means clock skew, // not a missing response. if delta < 0 { delta = 0 } for _, a := range accums { a.firstResponses = append(a.firstResponses, delta) } } // Active minutes and throughput activeSec := 0.0 asstChars := 0 for i, m := range msgs { if m.role == "assistant" { asstChars += m.contentLength } if i > 0 && msgs[i-1].valid && m.valid { gap := m.ts.Sub(msgs[i-1].ts).Seconds() if gap > 0 { if gap > maxGapSec { gap = maxGapSec } activeSec += gap } } } activeMins := activeSec / 60.0 if activeMins > 0 { for _, a := range accums { a.totalMsgs += len(msgs) a.totalChars += asstChars a.totalToolCalls += toolCount a.activeMinutes += activeMins } } } // turnCycleMean returns the arithmetic mean of turnCycles, or 0 when // empty. Session stats reports mean alongside p50/p90 — the retained // slice lets us compute both from the same sample. func (a *velocityAccumulator) turnCycleMean() float64 { return meanFloats(a.turnCycles) } // firstResponseMean returns the arithmetic mean of firstResponses, or // 0 when empty. See turnCycleMean for rationale. func (a *velocityAccumulator) firstResponseMean() float64 { return meanFloats(a.firstResponses) } func meanFloats(xs []float64) float64 { if len(xs) == 0 { return 0 } var sum float64 for _, x := range xs { sum += x } return sum / float64(len(xs)) } func (a *velocityAccumulator) computeOverview() VelocityOverview { sort.Float64s(a.turnCycles) sort.Float64s(a.firstResponses) var v VelocityOverview v.TurnCycleSec = Percentiles{ P50: math.Round( percentileFloat(a.turnCycles, 0.5)*10) / 10, P90: math.Round( percentileFloat(a.turnCycles, 0.9)*10) / 10, } v.FirstResponseSec = Percentiles{ P50: math.Round( percentileFloat(a.firstResponses, 0.5)*10) / 10, P90: math.Round( percentileFloat(a.firstResponses, 0.9)*10) / 10, } if a.activeMinutes > 0 { v.MsgsPerActiveMin = math.Round( float64(a.totalMsgs)/a.activeMinutes*10) / 10 v.CharsPerActiveMin = math.Round( float64(a.totalChars)/a.activeMinutes*10) / 10 v.ToolCallsPerActiveMin = math.Round( float64(a.totalToolCalls)/a.activeMinutes*10) / 10 } return v } // GetAnalyticsVelocity computes turn cycle, first response, and // throughput metrics with breakdowns by agent and complexity. func (db *DB) GetAnalyticsVelocity( ctx context.Context, f AnalyticsFilter, ) (VelocityResponse, error) { loc := f.location() dateCol := "COALESCE(NULLIF(started_at, ''), created_at)" where, args := f.buildWhere(dateCol) var timeIDs map[string]bool if f.HasTimeFilter() { var err error timeIDs, err = db.filteredSessionIDs(ctx, f) if err != nil { return VelocityResponse{}, err } } // Phase 1: Get filtered session metadata sessQuery := `SELECT id, ` + dateCol + `, agent, message_count FROM sessions WHERE ` + where sessRows, err := db.getReader().QueryContext( ctx, sessQuery, args..., ) if err != nil { return VelocityResponse{}, fmt.Errorf("querying velocity sessions: %w", err) } defer sessRows.Close() type sessInfo struct { agent string mc int } sessionMap := make(map[string]sessInfo) var sessionIDs []string for sessRows.Next() { var id, ts, agent string var mc int if err := sessRows.Scan( &id, &ts, &agent, &mc, ); err != nil { return VelocityResponse{}, fmt.Errorf("scanning velocity session: %w", err) } date := localDate(ts, loc) if !inDateRange(date, f.From, f.To) { continue } if timeIDs != nil && !timeIDs[id] { continue } sessionMap[id] = sessInfo{agent: agent, mc: mc} sessionIDs = append(sessionIDs, id) } if err := sessRows.Err(); err != nil { return VelocityResponse{}, fmt.Errorf("iterating velocity sessions: %w", err) } if len(sessionIDs) == 0 { return VelocityResponse{ ByAgent: []VelocityBreakdown{}, ByComplexity: []VelocityBreakdown{}, }, nil } if strings.TrimSpace(f.Model) != "" { stats, err := db.getAnalyticsFilteredMessageStats( ctx, sessionIDs, f, ) if err != nil { return VelocityResponse{}, err } for _, sid := range sessionIDs { info := sessionMap[sid] info.mc = stats[sid].Messages sessionMap[sid] = info } } sessionMsgs, err := db.getAnalyticsVelocityMessages( ctx, sessionIDs, f, ) if err != nil { return VelocityResponse{}, err } var toolCountMap map[string]int if strings.TrimSpace(f.Model) != "" { toolCountMap, err = db.getModelScopedToolCallCounts( ctx, sessionIDs, f, ) if err != nil { return VelocityResponse{}, err } } else { toolCountMap = make(map[string]int) err = queryChunked(sessionIDs, func(chunk []string) error { ph, chunkArgs := inPlaceholders(chunk) q := `SELECT session_id, COUNT(*) FROM tool_calls WHERE session_id IN ` + ph + ` GROUP BY session_id` rows, qErr := db.getReader().QueryContext( ctx, q, chunkArgs..., ) if qErr != nil { return fmt.Errorf( "querying velocity tool_calls: %w", qErr, ) } defer rows.Close() for rows.Next() { var sid string var count int if err := rows.Scan(&sid, &count); err != nil { return fmt.Errorf( "scanning velocity tool_call: %w", err, ) } toolCountMap[sid] = count } return rows.Err() }) if err != nil { return VelocityResponse{}, err } } // Process per-session metrics overall := &velocityAccumulator{} byAgent := make(map[string]*velocityAccumulator) byComplexity := make(map[string]*velocityAccumulator) for _, sid := range sessionIDs { info := sessionMap[sid] msgs := sessionMsgs[sid] if len(msgs) < 2 { continue } agentKey := info.agent compKey := complexityBucket(info.mc) if byAgent[agentKey] == nil { byAgent[agentKey] = &velocityAccumulator{} } if byComplexity[compKey] == nil { byComplexity[compKey] = &velocityAccumulator{} } processSessionVelocity( []*velocityAccumulator{ overall, byAgent[agentKey], byComplexity[compKey], }, msgs, toolCountMap[sid], ) } resp := VelocityResponse{ Overall: overall.computeOverview(), } // Build by-agent breakdowns agentKeys := make([]string, 0, len(byAgent)) for k := range byAgent { agentKeys = append(agentKeys, k) } sort.Strings(agentKeys) resp.ByAgent = make([]VelocityBreakdown, 0, len(agentKeys)) for _, k := range agentKeys { a, ok := byAgent[k] if !ok || a == nil { continue } resp.ByAgent = append(resp.ByAgent, VelocityBreakdown{ Label: k, Sessions: a.sessions, Overview: a.computeOverview(), }) } // Build by-complexity breakdowns compOrder := map[string]int{ "1-15": 0, "16-60": 1, "61+": 2, } compKeys := make([]string, 0, len(byComplexity)) for k := range byComplexity { compKeys = append(compKeys, k) } sort.Slice(compKeys, func(i, j int) bool { return compOrder[compKeys[i]] < compOrder[compKeys[j]] }) resp.ByComplexity = make( []VelocityBreakdown, 0, len(compKeys), ) for _, k := range compKeys { a, ok := byComplexity[k] if !ok || a == nil { continue } resp.ByComplexity = append(resp.ByComplexity, VelocityBreakdown{ Label: k, Sessions: a.sessions, Overview: a.computeOverview(), }) } return resp, nil } // --- Signals --- // SignalsAnalyticsResponse holds aggregated session signal data. type SignalsAnalyticsResponse struct { ScoredSessions int `json:"scored_sessions"` UnscoredSessions int `json:"unscored_sessions"` GradeDistribution map[string]int `json:"grade_distribution"` AvgHealthScore *float64 `json:"avg_health_score"` OutcomeDistribution map[string]int `json:"outcome_distribution"` OutcomeConfidenceDistribution map[string]int `json:"outcome_confidence_distribution"` ToolHealth SignalsToolHealth `json:"tool_health"` ContextHealth SignalsContextHealth `json:"context_health"` QualityHealth SignalsQualityHealth `json:"quality_health"` Trend []SignalsTrendBucket `json:"trend"` ByAgent []SignalsAgentRow `json:"by_agent"` ByProject []SignalsProjectRow `json:"by_project"` Calibration map[string]SignalCalibration `json:"calibration"` } // SignalsToolHealth holds aggregate tool failure metrics. type SignalsToolHealth struct { TotalFailureSignals int `json:"total_failure_signals"` TotalRetries int `json:"total_retries"` TotalEditChurn int `json:"total_edit_churn"` SessionsWithFailures int `json:"sessions_with_failures"` FailureRate float64 `json:"failure_rate"` } // SignalsContextHealth holds aggregate context pressure metrics. type SignalsContextHealth struct { AvgCompactionCount float64 `json:"avg_compaction_count"` SessionsWithCompaction int `json:"sessions_with_compaction"` MidTaskCompactionCount int `json:"mid_task_compaction_count"` SessionsWithMidTaskCompac int `json:"sessions_with_mid_task_compaction"` SessionsWithContextData int `json:"sessions_with_context_data"` AvgContextPressure *float64 `json:"avg_context_pressure"` HighPressureSessions int `json:"high_pressure_sessions"` } // SignalsQualityHealth holds aggregate deterministic quality-signal // metrics. Totals are raw signal sums; SessionsWithSignal counts // sessions where each signal was non-zero. type SignalsQualityHealth struct { ComputedSessions int `json:"computed_sessions"` Totals QualitySignalTotals `json:"totals"` SessionsWithSignal QualitySignalTotals `json:"sessions_with_signal"` } // QualitySignalTotals is shared by aggregate quality-signal totals. type QualitySignalTotals struct { ShortPromptCount int `json:"short_prompt_count"` UnstructuredStart int `json:"unstructured_start"` MissingSuccessCriteriaCount int `json:"missing_success_criteria_count"` MissingVerificationCount int `json:"missing_verification_count"` DuplicatePromptCount int `json:"duplicate_prompt_count"` NoCodeContextCount int `json:"no_code_context_count"` RunawayToolLoopCount int `json:"runaway_tool_loop_count"` FrustrationMarkerCount int `json:"frustration_marker_count"` } // SignalCalibration compares sessions with a signal to sessions // without it for the active filter slice. type SignalCalibration struct { Signal string `json:"signal"` AffectedSessions int `json:"affected_sessions"` BaselineSessions int `json:"baseline_sessions"` AffectedIncompleteRate float64 `json:"affected_incomplete_rate"` BaselineIncompleteRate float64 `json:"baseline_incomplete_rate"` IncompleteLift *float64 `json:"incomplete_lift"` AvgScoreDelta *float64 `json:"avg_score_delta"` } // SignalSessionsResponse returns concrete sessions that triggered // an aggregate signal, including the best available message excerpt. type SignalSessionsResponse struct { Signal string `json:"signal"` Sessions []SignalSessionExample `json:"sessions" nullable:"false"` } type SignalSessionExample struct { SessionID string `json:"session_id"` Project string `json:"project"` Agent string `json:"agent"` Date string `json:"date"` IsAutomated bool `json:"is_automated"` Outcome string `json:"outcome"` HealthScore *int `json:"health_score"` HealthGrade *string `json:"health_grade"` SignalTotal int `json:"signal_total"` ReasonCode string `json:"reason_code"` Excerpt string `json:"excerpt"` MessageOrdinal *int `json:"message_ordinal,omitempty"` FailureSignals int `json:"failure_signals"` Retries int `json:"retries"` EditChurn int `json:"edit_churn"` } type SignalMessage struct { SessionID string Ordinal int Role string Content string Timestamp string IsSystem bool HasToolUse bool } // SignalsTrendBucket holds signal data for one date bucket. type SignalsTrendBucket struct { Date string `json:"date"` SessionCount int `json:"session_count"` AvgHealthScore *float64 `json:"avg_health_score"` Completed int `json:"completed"` Errored int `json:"errored"` Abandoned int `json:"abandoned"` AvgFailureSignals float64 `json:"avg_failure_signals"` } // SignalsAgentRow holds signal data grouped by agent. type SignalsAgentRow struct { Agent string `json:"agent"` SessionCount int `json:"session_count"` AvgHealthScore *float64 `json:"avg_health_score"` CompletedRate float64 `json:"completed_rate"` AvgFailureSignals float64 `json:"avg_failure_signals"` } // SignalsProjectRow holds signal data grouped by project. type SignalsProjectRow struct { Project string `json:"project"` SessionCount int `json:"session_count"` AvgHealthScore *float64 `json:"avg_health_score"` CompletedRate float64 `json:"completed_rate"` AvgFailureSignals float64 `json:"avg_failure_signals"` } // SignalRow holds per-session signal data from the query. // Exported so the PostgreSQL store can build the same rows // from its own SELECT and feed them into AggregateSignals // without duplicating the aggregation logic. type SignalRow struct { ID string Agent string Project string Date string FirstMessage *string IsAutomated bool HealthScore *int HealthGrade *string Outcome string OutcomeConfidence string ToolFailureSignalCount int ToolRetryCount int EditChurnCount int CompactionCount int MidTaskCompactionCount int ContextPressureMax *float64 QualitySignalVersion int ShortPromptCount int UnstructuredStart bool MissingSuccessCriteriaCount int MissingVerificationCount int DuplicatePromptCount int NoCodeContextCount int RunawayToolLoopCount int FrustrationMarkerCount int } // GetAnalyticsSignals returns aggregated session signal data. // // Signals are session-scoped: the counts come from persisted per-session // columns computed at parse time (health, context pressure, compaction, // quality markers) that describe the whole session, not one model's turns. // When a model filter is active the session set is scoped to sessions that // used the model, but the totals stay session-level aggregates and are not // re-attributed per model -- per-model signal attribution is not well defined // for most signals and is intentionally not computed here. The PostgreSQL and // DuckDB stores mirror this. func (db *DB) GetAnalyticsSignals( ctx context.Context, f AnalyticsFilter, ) (SignalsAnalyticsResponse, error) { loc := f.location() dateCol := "COALESCE(NULLIF(started_at, ''), created_at)" where, args := f.buildWhere(dateCol) var timeIDs map[string]bool if f.HasTimeFilter() { var err error timeIDs, err = db.filteredSessionIDs(ctx, f) if err != nil { return SignalsAnalyticsResponse{}, err } } query := `SELECT id, agent, project, first_message, is_automated, ` + dateCol + `, 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 WHERE ` + where rows, err := db.getReader().QueryContext( ctx, query, args..., ) if err != nil { return SignalsAnalyticsResponse{}, fmt.Errorf( "querying analytics signals: %w", err, ) } defer rows.Close() var all []SignalRow for rows.Next() { var r SignalRow var ts string if err := rows.Scan( &r.ID, &r.Agent, &r.Project, &r.FirstMessage, &r.IsAutomated, &ts, &r.HealthScore, &r.HealthGrade, &r.Outcome, &r.OutcomeConfidence, &r.ToolFailureSignalCount, &r.ToolRetryCount, &r.EditChurnCount, &r.CompactionCount, &r.MidTaskCompactionCount, &r.ContextPressureMax, &r.QualitySignalVersion, &r.ShortPromptCount, &r.UnstructuredStart, &r.MissingSuccessCriteriaCount, &r.MissingVerificationCount, &r.DuplicatePromptCount, &r.NoCodeContextCount, &r.RunawayToolLoopCount, ); err != nil { return SignalsAnalyticsResponse{}, fmt.Errorf( "scanning signals row: %w", err, ) } r.Date = localDate(ts, loc) if !inDateRange(r.Date, f.From, f.To) { continue } if timeIDs != nil && !timeIDs[r.ID] { continue } all = append(all, r) } if err := rows.Err(); err != nil { return SignalsAnalyticsResponse{}, fmt.Errorf( "iterating signals rows: %w", err, ) } if err := db.populateFrustrationMarkers(ctx, all); err != nil { return SignalsAnalyticsResponse{}, err } return AggregateSignals(all), nil } // GetAnalyticsSignalSessions returns concrete examples for a // signal within the current analytics filter. Candidates are ranked by the // session-level signal counts (see GetAnalyticsSignals); under a model filter // the drill-down evidence is model-scoped while the ranking stays session-level. func (db *DB) GetAnalyticsSignalSessions( ctx context.Context, f AnalyticsFilter, signal string, limit int, ) (SignalSessionsResponse, error) { if !IsSupportedAnalyticsSignal(signal) { return SignalSessionsResponse{}, ErrUnsupportedAnalyticsSignal } if limit <= 0 || limit > 20 { limit = 10 } rows, err := db.signalRows(ctx, f) if err != nil { return SignalSessionsResponse{}, err } if err := db.populateFrustrationMarkers(ctx, rows); err != nil { return SignalSessionsResponse{}, err } candidates := SignalCandidates(rows, signal, limit) messages, err := db.signalMessages(ctx, candidates, f) if err != nil { return SignalSessionsResponse{}, err } return SignalSessionsResponse{ Signal: signal, Sessions: BuildSignalExamples(candidates, messages, signal), }, nil } func (db *DB) signalRows( ctx context.Context, f AnalyticsFilter, ) ([]SignalRow, error) { loc := f.location() dateCol := "COALESCE(NULLIF(started_at, ''), created_at)" where, args := f.buildWhere(dateCol) var timeIDs map[string]bool if f.HasTimeFilter() { var err error timeIDs, err = db.filteredSessionIDs(ctx, f) if err != nil { return nil, err } } query := `SELECT id, agent, project, first_message, is_automated, ` + dateCol + `, 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 WHERE ` + where rows, err := db.getReader().QueryContext(ctx, query, args...) if err != nil { return nil, fmt.Errorf( "querying analytics signal rows: %w", err, ) } defer rows.Close() var all []SignalRow for rows.Next() { r, err := scanSignalRow(rows, loc) if err != nil { return nil, err } if !inDateRange(r.Date, f.From, f.To) { continue } if timeIDs != nil && !timeIDs[r.ID] { continue } all = append(all, r) } if err := rows.Err(); err != nil { return nil, fmt.Errorf( "iterating analytics signal rows: %w", err, ) } return all, nil } func (db *DB) populateFrustrationMarkers( ctx context.Context, rows []SignalRow, ) error { if len(rows) == 0 { return nil } idx := make(map[string]int, len(rows)) ids := make([]string, 0, len(rows)) for i := range rows { idx[rows[i].ID] = i ids = append(ids, rows[i].ID) } return queryChunked(ids, func(chunk []string) error { ph, args := inPlaceholders(chunk) q := `SELECT session_id, ordinal, content, is_system FROM messages WHERE role = 'user' AND session_id IN ` + ph msgRows, err := db.getReader().QueryContext(ctx, q, args...) if err != nil { return fmt.Errorf( "querying frustration markers: %w", err, ) } defer msgRows.Close() for msgRows.Next() { var sessionID, content string var ordinal int var isSystem bool if err := msgRows.Scan( &sessionID, &ordinal, &content, &isSystem, ); err != nil { return fmt.Errorf( "scanning 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 frustration markers: %w", err, ) } return nil }) } func scanSignalRow(rs rowScanner, loc *time.Location) (SignalRow, error) { var r SignalRow var ts string if err := rs.Scan( &r.ID, &r.Agent, &r.Project, &r.FirstMessage, &r.IsAutomated, &ts, &r.HealthScore, &r.HealthGrade, &r.Outcome, &r.OutcomeConfidence, &r.ToolFailureSignalCount, &r.ToolRetryCount, &r.EditChurnCount, &r.CompactionCount, &r.MidTaskCompactionCount, &r.ContextPressureMax, &r.QualitySignalVersion, &r.ShortPromptCount, &r.UnstructuredStart, &r.MissingSuccessCriteriaCount, &r.MissingVerificationCount, &r.DuplicatePromptCount, &r.NoCodeContextCount, &r.RunawayToolLoopCount, ); err != nil { return SignalRow{}, fmt.Errorf( "scanning signal row: %w", err, ) } r.Date = localDate(ts, loc) return r, nil } func SignalCandidates( rows []SignalRow, signal string, limit int, ) []SignalRow { candidates := make([]SignalRow, 0) for _, r := range rows { if signalValue(r, signal) > 0 { candidates = append(candidates, r) } } sort.SliceStable(candidates, func(i, j int) bool { iv := signalValue(candidates[i], signal) jv := signalValue(candidates[j], signal) if iv != jv { return iv > jv } ib := isIncompleteOrLowQuality(candidates[i]) jb := isIncompleteOrLowQuality(candidates[j]) if ib != jb { return ib } if candidates[i].HealthScore != nil && candidates[j].HealthScore != nil && *candidates[i].HealthScore != *candidates[j].HealthScore { return *candidates[i].HealthScore < *candidates[j].HealthScore } return candidates[i].Date > candidates[j].Date }) if len(candidates) > limit { candidates = candidates[:limit] } return candidates } func (db *DB) signalMessages( ctx context.Context, rows []SignalRow, f AnalyticsFilter, ) (map[string][]SignalMessage, error) { out := make(map[string][]SignalMessage, len(rows)) if len(rows) == 0 { return out, nil } ids := make([]string, 0, len(rows)) for _, r := range rows { ids = append(ids, r.ID) } if strings.TrimSpace(f.Model) != "" { rowsBySession, err := db.getAnalyticsModelScopedMessages(ctx, ids, f) if err != nil { return nil, err } for sessionID, scopedRows := range rowsBySession { for _, row := range scopedRows { out[sessionID] = append(out[sessionID], 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 } filterModels := csvFilterValues(f.Model) err := queryChunked(ids, func(chunk []string) error { ph, args := inPlaceholders(chunk) q := `SELECT session_id, ordinal, role, content, COALESCE(timestamp, ''), is_system, has_tool_use FROM messages WHERE session_id IN ` + ph if len(filterModels) == 1 { q += ` AND model = ?` args = append(args, filterModels[0]) } else if len(filterModels) > 1 { modelPH, modelArgs := inPlaceholders(filterModels) q += ` AND model IN ` + modelPH args = append(args, modelArgs...) } q += ` ORDER BY session_id, ordinal` msgRows, err := db.getReader().QueryContext(ctx, q, args...) if err != nil { return fmt.Errorf( "querying signal messages: %w", err, ) } defer msgRows.Close() for msgRows.Next() { var m SignalMessage if err := msgRows.Scan( &m.SessionID, &m.Ordinal, &m.Role, &m.Content, &m.Timestamp, &m.IsSystem, &m.HasToolUse, ); err != nil { return fmt.Errorf( "scanning signal message: %w", err, ) } out[m.SessionID] = append(out[m.SessionID], m) } if err := msgRows.Err(); err != nil { return fmt.Errorf( "iterating signal messages: %w", err, ) } return nil }) return out, err } func BuildSignalExamples( rows []SignalRow, messages map[string][]SignalMessage, signal string, ) []SignalSessionExample { examples := make([]SignalSessionExample, 0, len(rows)) for _, r := range rows { excerpt, ordinal := signalExcerpt( signal, r, messages[r.ID], ) examples = append(examples, SignalSessionExample{ SessionID: r.ID, Project: r.Project, Agent: r.Agent, Date: r.Date, IsAutomated: r.IsAutomated, Outcome: r.Outcome, HealthScore: r.HealthScore, HealthGrade: r.HealthGrade, SignalTotal: signalValue(r, signal), ReasonCode: signalReason(signal), Excerpt: truncateExcerpt(excerpt, 180), MessageOrdinal: ordinal, FailureSignals: r.ToolFailureSignalCount, Retries: r.ToolRetryCount, EditChurn: r.EditChurnCount, }) } return examples } func signalExcerpt( signal string, r SignalRow, messages []SignalMessage, ) (string, *int) { switch signal { case "frustration_marker_count": if content, ordinal, ok := firstFrustrationPrompt(messages); ok { return content, ordinal } case "short_prompt_count": if content, ordinal, ok := firstShortPrompt(messages); ok { return content, ordinal } case "duplicate_prompt_count": if content, ordinal, ok := firstRepeatedPrompt(messages); ok { return content, ordinal } case "tool_failure_signals", "tool_retries", "edit_churn", "runaway_tool_loop_count": if content, ordinal, ok := firstToolUseMessage(messages); ok { return content, ordinal } case "outcome_errored", "outcome_abandoned", "outcome_completed", "sessions_with_compaction", "mid_task_compaction_count", "high_pressure_sessions": if content, ordinal, ok := lastSessionMessage(messages); ok { return content, ordinal } } if content, ordinal, ok := firstSubstantiveUserMessage(messages); ok { return content, ordinal } if r.FirstMessage != nil { return *r.FirstMessage, nil } return "", nil } func firstFrustrationPrompt( messages []SignalMessage, ) (string, *int, bool) { for _, m := range messages { if !isUserEvidenceMessage(m) { continue } if signals.IsFrustrationMarker(m.Content) { content, ordinal := messageEvidence(m) return content, ordinal, true } } return "", nil, false } func firstShortPrompt( messages []SignalMessage, ) (string, *int, bool) { firstUserOrdinal, ok := firstSubstantiveUserOrdinal(messages) if !ok { return "", nil, false } var previousAssistantTimestamp string hasPreviousAssistant := false userSinceLastAssistant := false for _, m := range messages { if m.IsSystem { continue } if m.Role == "assistant" { previousAssistantTimestamp = m.Timestamp hasPreviousAssistant = true userSinceLastAssistant = false continue } if !isUserEvidenceMessage(m) { continue } firstAfterAssistant := !userSinceLastAssistant normalized := normalizeEvidenceText(m.Content) if isControlEvidencePrompt(normalized) { continue } userSinceLastAssistant = true if len(normalized) >= 30 { continue } if m.Ordinal == firstUserOrdinal || (firstAfterAssistant && hasStaleEvidenceAssistantBefore( m.Timestamp, previousAssistantTimestamp, hasPreviousAssistant, )) { content, ordinal := messageEvidence(m) return content, ordinal, true } } return "", nil, false } func firstSubstantiveUserOrdinal( messages []SignalMessage, ) (int, bool) { for _, m := range messages { if !isUserEvidenceMessage(m) { continue } if isControlEvidencePrompt(normalizeEvidenceText(m.Content)) { continue } return m.Ordinal, true } return 0, false } func hasStaleEvidenceAssistantBefore( userTimestamp string, assistantTimestamp string, hasPreviousAssistant bool, ) bool { if !hasPreviousAssistant { return false } userTime, ok := parseEvidenceTime(userTimestamp) if !ok { return false } assistantTime, ok := parseEvidenceTime(assistantTimestamp) if !ok { return false } return userTime.Sub(assistantTime) > 30*time.Minute } func parseEvidenceTime(raw string) (time.Time, bool) { if raw == "" { return time.Time{}, false } for _, layout := range []string{ time.RFC3339Nano, time.RFC3339, "2006-01-02 15:04:05.999999999-07:00", "2006-01-02 15:04:05-07:00", } { t, err := time.Parse(layout, raw) if err == nil { return t, true } } return time.Time{}, false } func firstRepeatedPrompt( messages []SignalMessage, ) (string, *int, bool) { type prompt struct { normalized string tokens []string } seen := make([]prompt, 0, len(messages)) for _, m := range messages { if !isUserEvidenceMessage(m) { continue } key := evidencePromptKey(m.Content) if key == "" { continue } tokens := evidencePromptTokens(key) if len(tokens) < 4 { continue } for _, prev := range seen { if key == prev.normalized || evidenceJaccard(tokens, prev.tokens) >= 0.85 { content, ordinal := messageEvidence(m) return content, ordinal, true } } seen = append(seen, prompt{ normalized: key, tokens: tokens, }) } return "", nil, false } func firstToolUseMessage( messages []SignalMessage, ) (string, *int, bool) { for _, m := range messages { if m.IsSystem || !m.HasToolUse { continue } content, ordinal := messageEvidence(m) return content, ordinal, true } return "", nil, false } func lastSessionMessage( messages []SignalMessage, ) (string, *int, bool) { for _, v := range slices.Backward(messages) { m := v if m.IsSystem { continue } if !isSubstantiveEvidence(m.Content) && !m.HasToolUse { continue } content, ordinal := messageEvidence(m) return content, ordinal, true } return "", nil, false } func firstSubstantiveUserMessage( messages []SignalMessage, ) (string, *int, bool) { for _, m := range messages { if !isUserEvidenceMessage(m) { continue } content, ordinal := messageEvidence(m) return content, ordinal, true } return "", nil, false } func isUserEvidenceMessage(m SignalMessage) bool { return m.Role == "user" && !m.IsSystem && isSubstantiveEvidence(m.Content) } func isSubstantiveEvidence(content string) bool { return normalizeEvidenceText(content) != "" } func evidencePromptKey(content string) string { key := normalizeEvidenceText(content) if len(key) < 20 || isControlEvidencePrompt(key) { return "" } return key } func evidencePromptTokens(normalized string) []string { return strings.FieldsFunc(normalized, func(r rune) bool { return !unicode.IsLetter(r) && !unicode.IsDigit(r) }) } func evidenceJaccard(a, b []string) float64 { if len(a) == 0 || len(b) == 0 { return 0 } seen := map[string]struct{}{} for _, token := range a { seen[token] = struct{}{} } intersections := 0 union := len(seen) for _, token := range b { if _, ok := seen[token]; ok { intersections++ } else { union++ } } if union == 0 { return 0 } return float64(intersections) / float64(union) } func isControlEvidencePrompt(normalized string) bool { switch normalized { case "yes", "y", "no", "n", "ok", "okay", "continue", "go ahead", "proceed", "do it", "done", "thanks", "thank you", "please continue", "keep going": return true default: return false } } func messageEvidence(m SignalMessage) (string, *int) { ordinal := m.Ordinal content := strings.TrimSpace(m.Content) if content == "" && m.HasToolUse { content = "Tool-use turn" } return content, &ordinal } func signalReason(signal string) string { switch signal { case "short_prompt_count": return "short-start-contextual" case "unstructured_start": return "unstructured-task-start" case "missing_success_criteria_count": return "missing-observable-acceptance" case "missing_verification_count": return "missing-targeted-verification-path" case "duplicate_prompt_count": return "possible-stuck-reask" case "no_code_context_count": return "code-task-without-context" case "runaway_tool_loop_count": return "repeated-failing-tool-cycle" case "frustration_marker_count": return "frustration-marker" case "outcome_errored": return "errored-outcome" case "outcome_abandoned": return "abandoned-outcome" case "outcome_completed": return "completed-outcome" case "tool_failure_signals": return "tool-failure-signal" case "tool_retries": return "tool-retry" case "edit_churn": return "edit-churn" case "sessions_with_compaction": return "context-compaction" case "mid_task_compaction_count": return "mid-task-compaction" case "high_pressure_sessions": return "high-context-pressure" default: return signal } } func normalizeEvidenceText(content string) string { lower := strings.ToLower(strings.TrimSpace(content)) return spaceReplacer(lower) } func truncateExcerpt(s string, max int) string { s = strings.TrimSpace(spaceReplacer(s)) if len(s) <= max { return s } if max <= 3 { return s[:max] } return s[:max-3] + "..." } func spaceReplacer(s string) string { return strings.Join(strings.Fields(s), " ") } // AggregateSignals builds the response from collected rows. // Exported so the PostgreSQL store can reuse the same // aggregation logic instead of re-implementing it. func AggregateSignals( all []SignalRow, ) SignalsAnalyticsResponse { resp := SignalsAnalyticsResponse{ GradeDistribution: make(map[string]int), OutcomeDistribution: make(map[string]int), OutcomeConfidenceDistribution: make(map[string]int), Calibration: make(map[string]SignalCalibration), } if len(all) == 0 { resp.Trend = []SignalsTrendBucket{} resp.ByAgent = []SignalsAgentRow{} resp.ByProject = []SignalsProjectRow{} return resp } type groupAccum struct { count int healthScoreSum int healthScoreCount int completed int failureSignalSum int } totalCount := len(all) var healthScoreSum int var healthScoreCount int agentMap := make(map[string]*groupAccum) projectMap := make(map[string]*groupAccum) trendMap := make(map[string]*groupAccum) // Also track trend-specific outcome counts. type trendExtra struct { errored int abandoned int } trendExtras := make(map[string]*trendExtra) for _, r := range all { // Scored vs unscored if r.HealthScore != nil { resp.ScoredSessions++ healthScoreSum += *r.HealthScore healthScoreCount++ } else { resp.UnscoredSessions++ } // Grade distribution if r.HealthGrade != nil && *r.HealthGrade != "" { resp.GradeDistribution[*r.HealthGrade]++ } // Outcome distribution if r.Outcome != "" { resp.OutcomeDistribution[r.Outcome]++ } if r.OutcomeConfidence != "" { resp.OutcomeConfidenceDistribution[r.OutcomeConfidence]++ } // Tool health resp.ToolHealth.TotalFailureSignals += r.ToolFailureSignalCount resp.ToolHealth.TotalRetries += r.ToolRetryCount resp.ToolHealth.TotalEditChurn += r.EditChurnCount if r.ToolFailureSignalCount > 0 { resp.ToolHealth.SessionsWithFailures++ } // Context health if r.CompactionCount > 0 { resp.ContextHealth.SessionsWithCompaction++ } resp.ContextHealth.AvgCompactionCount += float64( r.CompactionCount, ) resp.ContextHealth.MidTaskCompactionCount += r.MidTaskCompactionCount if r.MidTaskCompactionCount > 0 { resp.ContextHealth.SessionsWithMidTaskCompac++ } if r.ContextPressureMax != nil { resp.ContextHealth.SessionsWithContextData++ if *r.ContextPressureMax >= 0.8 { resp.ContextHealth.HighPressureSessions++ } } accumulateQualityHealth(&resp.QualityHealth, r) // Accumulate by agent ga := agentMap[r.Agent] if ga == nil { ga = &groupAccum{} agentMap[r.Agent] = ga } ga.count++ ga.failureSignalSum += r.ToolFailureSignalCount if r.HealthScore != nil { ga.healthScoreSum += *r.HealthScore ga.healthScoreCount++ } if r.Outcome == "completed" { ga.completed++ } // Accumulate by project gp := projectMap[r.Project] if gp == nil { gp = &groupAccum{} projectMap[r.Project] = gp } gp.count++ gp.failureSignalSum += r.ToolFailureSignalCount if r.HealthScore != nil { gp.healthScoreSum += *r.HealthScore gp.healthScoreCount++ } if r.Outcome == "completed" { gp.completed++ } // Accumulate by date (trend) gt := trendMap[r.Date] if gt == nil { gt = &groupAccum{} trendMap[r.Date] = gt } gt.count++ gt.failureSignalSum += r.ToolFailureSignalCount if r.HealthScore != nil { gt.healthScoreSum += *r.HealthScore gt.healthScoreCount++ } if r.Outcome == "completed" { gt.completed++ } te := trendExtras[r.Date] if te == nil { te = &trendExtra{} trendExtras[r.Date] = te } if r.Outcome == "errored" { te.errored++ } if r.Outcome == "abandoned" { te.abandoned++ } } // Average health score if healthScoreCount > 0 { avg := math.Round( float64(healthScoreSum)/ float64(healthScoreCount)*10, ) / 10 resp.AvgHealthScore = &avg } // Tool health failure rate if totalCount > 0 { resp.ToolHealth.FailureRate = math.Round( float64(resp.ToolHealth.SessionsWithFailures)/ float64(totalCount)*1000, ) / 10 } // Context health averages if totalCount > 0 { resp.ContextHealth.AvgCompactionCount = math.Round( resp.ContextHealth.AvgCompactionCount/ float64(totalCount)*10, ) / 10 } if resp.ContextHealth.SessionsWithContextData > 0 { var pressureSum float64 for _, r := range all { if r.ContextPressureMax != nil { pressureSum += *r.ContextPressureMax } } avg := math.Round( pressureSum/ float64( resp.ContextHealth.SessionsWithContextData, )*1000, ) / 1000 resp.ContextHealth.AvgContextPressure = &avg } // Build trend (sorted by date) resp.Trend = make( []SignalsTrendBucket, 0, len(trendMap), ) for date, g := range trendMap { bucket := SignalsTrendBucket{ Date: date, SessionCount: g.count, Completed: g.completed, } if te := trendExtras[date]; te != nil { bucket.Errored = te.errored bucket.Abandoned = te.abandoned } if g.healthScoreCount > 0 { avg := math.Round( float64(g.healthScoreSum)/ float64(g.healthScoreCount)*10, ) / 10 bucket.AvgHealthScore = &avg } if g.count > 0 { bucket.AvgFailureSignals = math.Round( float64(g.failureSignalSum)/ float64(g.count)*10, ) / 10 } resp.Trend = append(resp.Trend, bucket) } sort.Slice(resp.Trend, func(i, j int) bool { return resp.Trend[i].Date < resp.Trend[j].Date }) // Build by-agent (sorted alphabetically) agentKeys := make([]string, 0, len(agentMap)) for k := range agentMap { agentKeys = append(agentKeys, k) } sort.Strings(agentKeys) resp.ByAgent = make( []SignalsAgentRow, 0, len(agentKeys), ) for _, agent := range agentKeys { g, ok := agentMap[agent] if !ok || g == nil { continue } row := SignalsAgentRow{ Agent: agent, SessionCount: g.count, } if g.healthScoreCount > 0 { avg := math.Round( float64(g.healthScoreSum)/ float64(g.healthScoreCount)*10, ) / 10 row.AvgHealthScore = &avg } if g.count > 0 { row.CompletedRate = math.Round( float64(g.completed)/ float64(g.count)*1000, ) / 10 row.AvgFailureSignals = math.Round( float64(g.failureSignalSum)/ float64(g.count)*10, ) / 10 } resp.ByAgent = append(resp.ByAgent, row) } // Build by-project (sorted by session count desc) resp.ByProject = make( []SignalsProjectRow, 0, len(projectMap), ) for project, g := range projectMap { row := SignalsProjectRow{ Project: project, SessionCount: g.count, } if g.healthScoreCount > 0 { avg := math.Round( float64(g.healthScoreSum)/ float64(g.healthScoreCount)*10, ) / 10 row.AvgHealthScore = &avg } if g.count > 0 { row.CompletedRate = math.Round( float64(g.completed)/ float64(g.count)*1000, ) / 10 row.AvgFailureSignals = math.Round( float64(g.failureSignalSum)/ float64(g.count)*10, ) / 10 } resp.ByProject = append(resp.ByProject, row) } sort.Slice(resp.ByProject, func(i, j int) bool { if resp.ByProject[i].SessionCount != resp.ByProject[j].SessionCount { return resp.ByProject[i].SessionCount > resp.ByProject[j].SessionCount } return resp.ByProject[i].Project < resp.ByProject[j].Project }) resp.Calibration = buildSignalCalibrations(all) return resp } func accumulateQualityHealth( q *SignalsQualityHealth, r SignalRow, ) { if r.QualitySignalVersion <= 0 { return } q.ComputedSessions++ q.Totals.ShortPromptCount += r.ShortPromptCount if r.ShortPromptCount > 0 { q.SessionsWithSignal.ShortPromptCount++ } if r.UnstructuredStart { q.Totals.UnstructuredStart++ q.SessionsWithSignal.UnstructuredStart++ } q.Totals.MissingSuccessCriteriaCount += r.MissingSuccessCriteriaCount if r.MissingSuccessCriteriaCount > 0 { q.SessionsWithSignal.MissingSuccessCriteriaCount++ } q.Totals.MissingVerificationCount += r.MissingVerificationCount if r.MissingVerificationCount > 0 { q.SessionsWithSignal.MissingVerificationCount++ } q.Totals.DuplicatePromptCount += r.DuplicatePromptCount if r.DuplicatePromptCount > 0 { q.SessionsWithSignal.DuplicatePromptCount++ } q.Totals.NoCodeContextCount += r.NoCodeContextCount if r.NoCodeContextCount > 0 { q.SessionsWithSignal.NoCodeContextCount++ } q.Totals.RunawayToolLoopCount += r.RunawayToolLoopCount if r.RunawayToolLoopCount > 0 { q.SessionsWithSignal.RunawayToolLoopCount++ } q.Totals.FrustrationMarkerCount += r.FrustrationMarkerCount if r.FrustrationMarkerCount > 0 { q.SessionsWithSignal.FrustrationMarkerCount++ } } func buildSignalCalibrations( rows []SignalRow, ) map[string]SignalCalibration { signals := []string{ "tool_failure_signals", "tool_retries", "edit_churn", "sessions_with_compaction", "mid_task_compaction_count", "high_pressure_sessions", "short_prompt_count", "unstructured_start", "missing_success_criteria_count", "missing_verification_count", "duplicate_prompt_count", "no_code_context_count", "runaway_tool_loop_count", "frustration_marker_count", } out := make(map[string]SignalCalibration, len(signals)) for _, signal := range signals { out[signal] = calibrateSignal(rows, signal) } return out } func calibrateSignal(rows []SignalRow, signal string) SignalCalibration { type side struct { count int incomplete int scoreSum int scoreCount int } var affected, baseline side for _, r := range rows { target := &baseline if signalValue(r, signal) > 0 { target = &affected } target.count++ if isIncompleteOrLowQuality(r) { target.incomplete++ } if r.HealthScore != nil { target.scoreSum += *r.HealthScore target.scoreCount++ } } result := SignalCalibration{ Signal: signal, AffectedSessions: affected.count, BaselineSessions: baseline.count, } if affected.count > 0 { result.AffectedIncompleteRate = round1( float64(affected.incomplete) / float64(affected.count) * 100, ) } if baseline.count > 0 { result.BaselineIncompleteRate = round1( float64(baseline.incomplete) / float64(baseline.count) * 100, ) } if baseline.count > 0 && result.BaselineIncompleteRate > 0 && affected.count > 0 { lift := round1( result.AffectedIncompleteRate / result.BaselineIncompleteRate, ) result.IncompleteLift = &lift } if affected.scoreCount > 0 && baseline.scoreCount > 0 { delta := round1( float64(affected.scoreSum)/ float64(affected.scoreCount) - float64(baseline.scoreSum)/ float64(baseline.scoreCount), ) result.AvgScoreDelta = &delta } return result } func signalValue(r SignalRow, signal string) int { switch signal { case "outcome_errored": if r.Outcome == "errored" { return 1 } case "outcome_abandoned": if r.Outcome == "abandoned" { return 1 } case "outcome_completed": if r.Outcome == "completed" { return 1 } case "tool_failure_signals": return r.ToolFailureSignalCount case "tool_retries": return r.ToolRetryCount case "edit_churn": return r.EditChurnCount case "sessions_with_compaction": if r.CompactionCount > 0 { return 1 } case "mid_task_compaction_count": return r.MidTaskCompactionCount case "high_pressure_sessions": if r.ContextPressureMax != nil && *r.ContextPressureMax >= 0.8 { return 1 } case "short_prompt_count": return r.ShortPromptCount case "unstructured_start": if r.UnstructuredStart { return 1 } case "missing_success_criteria_count": return r.MissingSuccessCriteriaCount case "missing_verification_count": return r.MissingVerificationCount case "duplicate_prompt_count": return r.DuplicatePromptCount case "no_code_context_count": return r.NoCodeContextCount case "runaway_tool_loop_count": return r.RunawayToolLoopCount case "frustration_marker_count": return r.FrustrationMarkerCount } return 0 } func SignalValue(r SignalRow, signal string) int { return signalValue(r, signal) } func isIncompleteOrLowQuality(r SignalRow) bool { if r.Outcome == "errored" || r.Outcome == "abandoned" { return true } if r.HealthGrade == nil { return false } return *r.HealthGrade == "D" || *r.HealthGrade == "F" } func round1(v float64) float64 { return math.Round(v*10) / 10 } // --- Top Sessions --- // TopSession holds summary info for a ranked session. type TopSession struct { ID string `json:"id"` Project string `json:"project"` FirstMessage *string `json:"first_message"` DisplayName *string `json:"display_name,omitempty"` MessageCount int `json:"message_count"` OutputTokens int `json:"output_tokens"` DurationMin float64 `json:"duration_min"` ActiveDurationMin float64 `json:"active_duration_min"` // StartedAt and EndedAt are included so the frontend can // derive a recency-based status tier — the StatusDot in the // Top Sessions column needs the same time window inputs as // the sidebar's session list. StartedAt *string `json:"started_at,omitempty"` EndedAt *string `json:"ended_at,omitempty"` TerminationStatus *string `json:"termination_status,omitempty"` } // TopSessionsResponse wraps the top sessions list. type TopSessionsResponse struct { Metric string `json:"metric"` Sessions []TopSession `json:"sessions"` } // ActiveGapCapSec and ActiveGapCapMs bound how much a single // inter-message gap can contribute to "active" time: a gap below the cap // (5 minutes) counts in full -- model generation, tool execution, or a // quick human turnaround -- while anything longer is treated as idle // beyond the cap. Defined once and shared by the velocity "active // minutes" metric and the Top Sessions "active duration" SQL so the two // definitions cannot drift. timing_test.go asserts the two stay equal. const ( ActiveGapCapSec = 300.0 ActiveGapCapMs = 300_000 ) // sqliteActiveDurationExpr builds the correlated-subquery SQL that computes a // session's active duration in minutes: the sum of consecutive inter-message // gaps with each gap capped at ActiveGapCapMs. sessionIDCol is the outer // reference to correlate on (for example "sessions.id"). Shared by the in-SQL // and timezone-aware Go fallback Top Sessions paths so the two cannot drift. func sqliteActiveDurationExpr(sessionIDCol string) string { return 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( (julianday(LEAD(m2.timestamp) OVER (ORDER BY m2.ordinal)) - julianday(m2.timestamp)) * 86400000 ) AS INTEGER ) AS delta_ms FROM messages m2 WHERE m2.session_id = %[2]s ) inner2)`, ActiveGapCapMs, sessionIDCol) } // GetAnalyticsTopSessions returns the top 10 sessions by the // given metric ("messages", "duration", or "output_tokens") // within the filter. func (db *DB) GetAnalyticsTopSessions( ctx context.Context, f AnalyticsFilter, metric string, ) (TopSessionsResponse, error) { if metric == "" { metric = "messages" } if !f.canUseSQLiteTimeSQL() || strings.TrimSpace(f.Model) != "" { return db.getAnalyticsTopSessionsGo(ctx, f, metric) } dateCol := "COALESCE(NULLIF(started_at, ''), created_at)" where, args := sqliteAnalyticsWhereSQL(f, dateCol, "sessions.id", true) durationExpr := `ROUND((julianday(ended_at) - julianday(started_at)) * 1440, 1)` durationSelectExpr := "COALESCE(" + durationExpr + ", 0)" activeDurationSelectExpr := "COALESCE(" + sqliteActiveDurationExpr("sessions.id") + ", 0)" needsGoSort := metric == "duration" var orderExpr string switch metric { case "output_tokens": if strings.TrimSpace(f.Model) == "" { where += " AND has_total_output_tokens = TRUE" } orderExpr = "total_output_tokens DESC, id ASC" case "duration": orderExpr = activeDurationSelectExpr + " DESC, id ASC" where += " AND NULLIF(started_at, '') IS NOT NULL" + " AND NULLIF(ended_at, '') IS NOT NULL" + " AND julianday(ended_at) >= julianday(started_at)" default: metric = "messages" orderExpr = "message_count DESC, id ASC" } query := `SELECT id, project, first_message, COALESCE(display_name, session_name) AS display_name, message_count, total_output_tokens, ` + durationSelectExpr + `, ` + activeDurationSelectExpr + `, started_at, ended_at, termination_status FROM sessions WHERE ` + where + ` ORDER BY ` + orderExpr + ` LIMIT 10` rows, err := db.getReader().QueryContext(ctx, query, args...) if err != nil { return TopSessionsResponse{}, fmt.Errorf("querying top sessions: %w", err) } defer rows.Close() resp := TopSessionsResponse{Metric: metric} for rows.Next() { var row TopSession if err := rows.Scan( &row.ID, &row.Project, &row.FirstMessage, &row.DisplayName, &row.MessageCount, &row.OutputTokens, &row.DurationMin, &row.ActiveDurationMin, &row.StartedAt, &row.EndedAt, &row.TerminationStatus, ); err != nil { return TopSessionsResponse{}, fmt.Errorf("scanning top session: %w", err) } resp.Sessions = append(resp.Sessions, row) } if err := rows.Err(); err != nil { return TopSessionsResponse{}, fmt.Errorf("iterating top sessions: %w", err) } sessions := rankTopSessions(resp.Sessions, needsGoSort) resp.Sessions = sessions return resp, nil } func (db *DB) getAnalyticsTopSessionsGo( ctx context.Context, f AnalyticsFilter, metric string, ) (TopSessionsResponse, error) { loc := f.location() dateCol := "COALESCE(NULLIF(started_at, ''), created_at)" where, args := f.buildWhere(dateCol) var timeIDs map[string]bool if f.HasTimeFilter() { var err error timeIDs, err = db.filteredSessionIDs(ctx, f) if err != nil { return TopSessionsResponse{}, err } } var orderExpr string needsGoSort := metric == "duration" switch metric { case "output_tokens": where += " AND has_total_output_tokens = TRUE" orderExpr = "total_output_tokens DESC, id ASC" case "duration": orderExpr = `(julianday(ended_at) - julianday(started_at)) * 1440 DESC, id ASC` where += " AND NULLIF(started_at, '') IS NOT NULL" + " AND NULLIF(ended_at, '') IS NOT NULL" + " AND julianday(ended_at) >= julianday(started_at)" default: metric = "messages" orderExpr = "message_count DESC, id ASC" } limitClause := " LIMIT 200" if f.HasTimeFilter() || needsGoSort || (strings.TrimSpace(f.Model) != "" && (metric == "messages" || metric == "output_tokens")) { limitClause = "" } // Active duration is only consumed for the duration metric. Compute it // in the outer query (matching the in-SQL path) rather than issuing a // per-row follow-up query: the per-row form held the outer rows iterator // open while acquiring a second reader connection for each row, an // unbounded N+1 that could exhaust the capped reader pool and deadlock // under concurrent requests. activeDurationSelectExpr := "0.0" if needsGoSort { activeDurationSelectExpr = "COALESCE(" + sqliteActiveDurationExpr("sessions.id") + ", 0)" } query := `SELECT id, ` + dateCol + `, project, first_message, COALESCE(display_name, session_name) AS display_name, message_count, total_output_tokens, ` + activeDurationSelectExpr + ` AS active_duration_min, started_at, ended_at, termination_status FROM sessions WHERE ` + where + ` ORDER BY ` + orderExpr + limitClause rows, err := db.getReader().QueryContext(ctx, query, args...) if err != nil { return TopSessionsResponse{}, fmt.Errorf("querying top sessions: %w", err) } defer rows.Close() var sessions []TopSession for rows.Next() { var id, ts, project string var firstMsg, displayName, startedAt, endedAt *string var termStatus *string var mc, outputTokens int var activeDurationMin float64 if err := rows.Scan( &id, &ts, &project, &firstMsg, &displayName, &mc, &outputTokens, &activeDurationMin, &startedAt, &endedAt, &termStatus, ); err != nil { return TopSessionsResponse{}, fmt.Errorf("scanning top session: %w", err) } date := localDate(ts, loc) if !inDateRange(date, f.From, f.To) { continue } if timeIDs != nil && !timeIDs[id] { continue } durMin := 0.0 if startedAt != nil && endedAt != nil { tS, okS := localTime(*startedAt, loc) tE, okE := localTime(*endedAt, loc) if okS && okE { durMin = math.Round( tE.Sub(tS).Minutes()*10) / 10 } } sessions = append(sessions, TopSession{ ID: id, Project: project, FirstMessage: firstMsg, DisplayName: displayName, MessageCount: mc, OutputTokens: outputTokens, DurationMin: durMin, ActiveDurationMin: activeDurationMin, StartedAt: startedAt, EndedAt: endedAt, TerminationStatus: termStatus, }) } if err := rows.Err(); err != nil { return TopSessionsResponse{}, fmt.Errorf("iterating top sessions: %w", err) } if strings.TrimSpace(f.Model) != "" && (metric == "messages" || metric == "output_tokens") { sessionIDs := make([]string, 0, len(sessions)) for _, session := range sessions { sessionIDs = append(sessionIDs, session.ID) } stats, err := db.getAnalyticsFilteredMessageStats(ctx, sessionIDs, f) if err != nil { return TopSessionsResponse{}, err } filtered := sessions[:0] for i := range sessions { stat := stats[sessions[i].ID] sessions[i].MessageCount = stat.Messages sessions[i].OutputTokens = stat.OutputTokens if metric == "output_tokens" && !stat.HasOutputTokens { continue } filtered = append(filtered, sessions[i]) } sessions = filtered sort.SliceStable(sessions, func(i, j int) bool { if metric == "output_tokens" { if sessions[i].OutputTokens != sessions[j].OutputTokens { return sessions[i].OutputTokens > sessions[j].OutputTokens } } else if sessions[i].MessageCount != sessions[j].MessageCount { return sessions[i].MessageCount > sessions[j].MessageCount } return sessions[i].ID < sessions[j].ID }) } if sessions == nil { sessions = []TopSession{} } sessions = rankTopSessions(sessions, needsGoSort) if len(sessions) > 10 { sessions = sessions[:10] } return TopSessionsResponse{ Metric: metric, Sessions: sessions, }, nil } func rankTopSessions(sessions []TopSession, needsGoSort bool) []TopSession { if sessions == nil { return []TopSession{} } if needsGoSort && len(sessions) > 1 { sort.SliceStable(sessions, func(i, j int) bool { if sessions[i].ActiveDurationMin != sessions[j].ActiveDurationMin { return sessions[i].ActiveDurationMin > sessions[j].ActiveDurationMin } return sessions[i].ID < sessions[j].ID }) } for i := range sessions { sessions[i].DurationMin = round1(sessions[i].DurationMin) sessions[i].ActiveDurationMin = round1(sessions[i].ActiveDurationMin) } return sessions }