systemprompt_analytics/repository/agents/
stats_queries.rs

1//! Per-agent aggregate statistics queries. Totals read the hourly task and
2//! AI usage rollups, widened to whole UTC hours at the window's start; the
3//! trend series is per task and stays on `report_agent_tasks`.
4//!
5//! Copyright (c) systemprompt.io — Business Source License 1.1.
6//! See <https://systemprompt.io> for licensing details.
7
8use crate::Result;
9use chrono::{DateTime, Utc};
10
11use super::AgentAnalyticsRepository;
12use crate::models::reporting::{AgentAiStatsRow, AgentStatsRow, AgentTaskRow};
13
14impl AgentAnalyticsRepository {
15    pub async fn get_stats(
16        &self,
17        start: DateTime<Utc>,
18        end: DateTime<Utc>,
19        agent_filter: Option<&str>,
20    ) -> Result<AgentStatsRow> {
21        if let Some(agent) = agent_filter {
22            let pattern = format!("%{}%", agent);
23            sqlx::query_as!(
24                AgentStatsRow,
25                r#"
26                SELECT
27                    COUNT(DISTINCT agent_name)::bigint as "total_agents!",
28                    COALESCE(SUM(task_count), 0)::bigint as "total_tasks!",
29                    -- agent_tasks.status stores the canonical A2A TaskState value
30                    -- written by task_state_to_db_string (TASK_STATE_*), which the
31                    -- agent_tasks_status_check CHECK constraint also enforces.
32                    COALESCE(SUM(task_count) FILTER (WHERE status = 'TASK_STATE_COMPLETED'), 0)::bigint as "completed_tasks!",
33                    COALESCE(SUM(task_count) FILTER (WHERE status = 'TASK_STATE_FAILED'), 0)::bigint as "failed_tasks!",
34                    COALESCE((SUM(exec_sum_ms)::numeric / NULLIF(SUM(exec_n), 0))::float8, 0) as "avg_execution_time_ms!"
35                FROM report_agent_task_hourly
36                WHERE bucket >= date_trunc('hour', $1::timestamptz, 'UTC') AND bucket < $2
37                  AND agent_name <> '' AND agent_name ILIKE $3
38                "#,
39                start,
40                end,
41                pattern
42            )
43            .fetch_one(&*self.pool)
44            .await
45            .map_err(Into::into)
46        } else {
47            sqlx::query_as!(
48                AgentStatsRow,
49                r#"
50                SELECT
51                    COUNT(DISTINCT NULLIF(agent_name, ''))::bigint as "total_agents!",
52                    COALESCE(SUM(task_count), 0)::bigint as "total_tasks!",
53                    COALESCE(SUM(task_count) FILTER (WHERE status = 'TASK_STATE_COMPLETED'), 0)::bigint as "completed_tasks!",
54                    COALESCE(SUM(task_count) FILTER (WHERE status = 'TASK_STATE_FAILED'), 0)::bigint as "failed_tasks!",
55                    COALESCE((SUM(exec_sum_ms)::numeric / NULLIF(SUM(exec_n), 0))::float8, 0) as "avg_execution_time_ms!"
56                FROM report_agent_task_hourly
57                WHERE bucket >= date_trunc('hour', $1::timestamptz, 'UTC') AND bucket < $2
58                "#,
59                start,
60                end
61            )
62            .fetch_one(&*self.pool)
63            .await
64            .map_err(Into::into)
65        }
66    }
67
68    pub async fn get_ai_stats(
69        &self,
70        start: DateTime<Utc>,
71        end: DateTime<Utc>,
72    ) -> Result<AgentAiStatsRow> {
73        sqlx::query_as!(
74            AgentAiStatsRow,
75            r#"
76            SELECT
77                COALESCE(SUM(request_count), 0)::bigint as "total_ai_requests!",
78                COALESCE(SUM(cost_microdollars), 0)::bigint as "total_cost_microdollars!"
79            FROM report_ai_usage_hourly
80            WHERE bucket >= date_trunc('hour', $1::timestamptz, 'UTC') AND bucket < $2
81            "#,
82            start,
83            end
84        )
85        .fetch_one(&*self.pool)
86        .await
87        .map_err(Into::into)
88    }
89
90    pub async fn get_tasks_for_trends(
91        &self,
92        start: DateTime<Utc>,
93        end: DateTime<Utc>,
94        agent_filter: Option<&str>,
95    ) -> Result<Vec<AgentTaskRow>> {
96        if let Some(agent) = agent_filter {
97            let pattern = format!("%{}%", agent);
98            sqlx::query_as!(
99                AgentTaskRow,
100                r#"
101                SELECT
102                    started_at as "started_at!",
103                    status,
104                    execution_time_ms
105                FROM report_agent_tasks
106                WHERE started_at >= $1 AND started_at < $2
107                  AND agent_name ILIKE $3
108                ORDER BY started_at
109                "#,
110                start,
111                end,
112                pattern
113            )
114            .fetch_all(&*self.pool)
115            .await
116            .map_err(Into::into)
117        } else {
118            sqlx::query_as!(
119                AgentTaskRow,
120                r#"
121                SELECT
122                    started_at as "started_at!",
123                    status,
124                    execution_time_ms
125                FROM report_agent_tasks
126                WHERE started_at >= $1 AND started_at < $2
127                ORDER BY started_at
128                "#,
129                start,
130                end
131            )
132            .fetch_all(&*self.pool)
133            .await
134            .map_err(Into::into)
135        }
136    }
137}
systemprompt_analytics/repository/agents/stats_queries.rs

systemprompt_analytics/repository/agents/
stats_queries.rs
