systemprompt_analytics/repository/agents/
detail_queries.rs

1//! Single-agent analytics detail queries. The existence check, summary,
2//! status breakdown and hour-of-day distribution read the hourly task rollup
3//! (`report_agent_task_hourly`), widened to whole UTC hours at the window's
4//! start; top errors group by message text and stay on `report_agent_tasks`
5//! for their window.
6//!
7//! Copyright (c) systemprompt.io — Business Source License 1.1.
8//! See <https://systemprompt.io> for licensing details.
9
10use crate::Result;
11use chrono::{DateTime, Utc};
12
13use super::AgentAnalyticsRepository;
14use crate::models::reporting::{
15    AgentErrorRow, AgentHourlyRow, AgentStatusBreakdownRow, AgentSummaryRow,
16};
17
18impl AgentAnalyticsRepository {
19    pub async fn agent_exists(
20        &self,
21        agent_filter: &str,
22        start: DateTime<Utc>,
23        end: DateTime<Utc>,
24    ) -> Result<i64> {
25        let pattern = format!("%{}%", agent_filter);
26        let count = sqlx::query_scalar!(
27            r#"SELECT COALESCE(SUM(task_count), 0)::bigint as "count!" FROM report_agent_task_hourly WHERE bucket >= date_trunc('hour', $1::timestamptz, 'UTC') AND bucket < $2 AND agent_name <> '' AND agent_name ILIKE $3"#,
28            start,
29            end,
30            pattern
31        )
32        .fetch_one(&*self.pool)
33        .await?;
34        Ok(count)
35    }
36
37    pub async fn get_agent_summary(
38        &self,
39        agent_filter: &str,
40        start: DateTime<Utc>,
41        end: DateTime<Utc>,
42    ) -> Result<AgentSummaryRow> {
43        let pattern = format!("%{}%", agent_filter);
44        sqlx::query_as!(
45            AgentSummaryRow,
46            r#"
47            SELECT
48                COALESCE(SUM(task_count), 0)::bigint as "total_tasks!",
49                COALESCE(SUM(task_count) FILTER (WHERE status = 'TASK_STATE_COMPLETED'), 0)::bigint as "completed!",
50                COALESCE(SUM(task_count) FILTER (WHERE status = 'TASK_STATE_FAILED'), 0)::bigint as "failed!",
51                COALESCE((SUM(exec_sum_ms)::numeric / NULLIF(SUM(exec_n), 0))::float8, 0) as "avg_time!"
52            FROM report_agent_task_hourly
53            WHERE bucket >= date_trunc('hour', $1::timestamptz, 'UTC') AND bucket < $2
54              AND agent_name <> '' AND agent_name ILIKE $3
55            "#,
56            start,
57            end,
58            pattern
59        )
60        .fetch_one(&*self.pool)
61        .await
62        .map_err(Into::into)
63    }
64
65    pub async fn get_status_breakdown(
66        &self,
67        agent_filter: &str,
68        start: DateTime<Utc>,
69        end: DateTime<Utc>,
70    ) -> Result<Vec<AgentStatusBreakdownRow>> {
71        let pattern = format!("%{}%", agent_filter);
72        sqlx::query_as!(
73            AgentStatusBreakdownRow,
74            r#"
75            SELECT status as "status!", SUM(task_count)::bigint as "status_count!"
76            FROM report_agent_task_hourly
77            WHERE bucket >= date_trunc('hour', $1::timestamptz, 'UTC') AND bucket < $2
78              AND agent_name <> '' AND agent_name ILIKE $3
79            GROUP BY status
80            ORDER BY 2 DESC
81            "#,
82            start,
83            end,
84            pattern
85        )
86        .fetch_all(&*self.pool)
87        .await
88        .map_err(Into::into)
89    }
90
91    pub async fn get_top_errors(
92        &self,
93        agent_filter: &str,
94        start: DateTime<Utc>,
95        end: DateTime<Utc>,
96    ) -> Result<Vec<AgentErrorRow>> {
97        let pattern = format!("%{}%", agent_filter);
98        sqlx::query_as!(
99            AgentErrorRow,
100            r#"
101            SELECT
102                COALESCE(SUBSTRING(error_message FROM 1 FOR 100), 'Unknown error') as "error_type",
103                COUNT(*)::bigint as "error_count!"
104            FROM report_agent_tasks
105            WHERE agent_name ILIKE $1
106              AND started_at >= $2 AND started_at < $3
107              AND status = 'TASK_STATE_FAILED'
108            GROUP BY SUBSTRING(error_message FROM 1 FOR 100)
109            ORDER BY 2 DESC
110            LIMIT 10
111            "#,
112            pattern,
113            start,
114            end
115        )
116        .fetch_all(&*self.pool)
117        .await
118        .map_err(Into::into)
119    }
120
121    pub async fn get_hourly_distribution(
122        &self,
123        agent_filter: &str,
124        start: DateTime<Utc>,
125        end: DateTime<Utc>,
126    ) -> Result<Vec<AgentHourlyRow>> {
127        let pattern = format!("%{}%", agent_filter);
128        sqlx::query_as!(
129            AgentHourlyRow,
130            r#"
131            SELECT
132                EXTRACT(HOUR FROM bucket)::INTEGER as "task_hour!",
133                SUM(task_count)::bigint as "task_count!"
134            FROM report_agent_task_hourly
135            WHERE bucket >= date_trunc('hour', $1::timestamptz, 'UTC') AND bucket < $2
136              AND agent_name <> '' AND agent_name ILIKE $3
137            GROUP BY EXTRACT(HOUR FROM bucket)
138            ORDER BY 1
139            "#,
140            start,
141            end,
142            pattern
143        )
144        .fetch_all(&*self.pool)
145        .await
146        .map_err(Into::into)
147    }
148}
systemprompt_analytics/repository/agents/detail_queries.rs

systemprompt_analytics/repository/agents/
detail_queries.rs
