feat: daily active agents

Author: think-in-universe

crates/api/src/openapi.rs

@@ -100,6 +100,7 @@ use utoipa::OpenApi;
         crate::routes::admin::bi_deployment_summary,
         crate::routes::admin::bi_status_history,
         crate::routes::admin::bi_usage,
+        crate::routes::admin::bi_daily_active_agents,
         crate::routes::admin::bi_top_consumers,
         crate::routes::agents::start_instance,
         crate::routes::agents::stop_instance,
@@ -213,6 +214,8 @@ use utoipa::OpenApi;
         services::bi_metrics::DeploymentStatusCount,
         services::bi_metrics::DeploymentSummary,
         services::bi_metrics::StatusChangeRecord,
+        services::bi_metrics::DailyActiveAgentsByType,
+        services::bi_metrics::DailyActiveAgentsPoint,
         services::bi_metrics::UsageAggregation,
         services::bi_metrics::UsageGroupBy,
         services::bi_metrics::UsageRankBy,

crates/api/src/routes/admin.rs

@@ -16,9 +16,9 @@ use chrono::{DateTime, Utc};
 use serde::{Deserialize, Serialize};
 use services::analytics::{ActivityLogEntry, AnalyticsSummary, TopActiveUsersResponse};
 use services::bi_metrics::{
-    DeploymentFilter, DeploymentRecord, DeploymentSummary, StatusChangeRecord, TopConsumer,
-    TopConsumerFilter, TopConsumerGroupBy, UsageAggregation, UsageFilter, UsageGroupBy,
-    UsageRankBy as BiUsageRankBy, UserSummary,
+    DailyActiveAgentsPoint, DeploymentFilter, DeploymentRecord, DeploymentSummary,
+    StatusChangeRecord, TopConsumer, TopConsumerFilter, TopConsumerGroupBy, UsageAggregation,
+    UsageFilter, UsageGroupBy, UsageRankBy as BiUsageRankBy, UserSummary,
 };
 
 /// Maximum rows for BI usage aggregation queries.
@@ -2376,6 +2376,42 @@ pub async fn bi_top_consumers(
     }))
 }
 
+/// Get daily active agents (with usage > 0) time series (BI). Requires admin authentication.
+/// Dates are UTC. Maximum range 365 days.
+#[utoipa::path(
+    get,
+    path = "/v1/admin/bi/usage/daily-active-agents",
+    tag = "Admin",
+    params(BiSummaryQuery),
+    responses(
+        (status = 200, description = "Daily active agents time series", body = [DailyActiveAgentsPoint]),
+        (status = 400, description = "Bad request", body = crate::error::ApiErrorResponse),
+        (status = 401, description = "Unauthorized", body = crate::error::ApiErrorResponse),
+        (status = 403, description = "Forbidden - Admin access required", body = crate::error::ApiErrorResponse),
+        (status = 500, description = "Internal server error", body = crate::error::ApiErrorResponse)
+    ),
+    security(
+        ("session_token" = [])
+    )
+)]
+pub async fn bi_daily_active_agents(
+    State(app_state): State<AppState>,
+    Query(params): Query<BiSummaryQuery>,
+) -> Result<Json<Vec<DailyActiveAgentsPoint>>, ApiError> {
+    validate_date_range(params.start_date, params.end_date)?;
+
+    let points = app_state
+        .bi_metrics_service
+        .get_daily_active_agents(params.start_date, params.end_date)
+        .await
+        .map_err(|e| {
+            tracing::error!("Failed to get daily active agents: {}", e);
+            ApiError::internal_server_error("Failed to get daily active agents")
+        })?;
+
+    Ok(Json(points))
+}
+
 /// Create admin router with all admin routes (requires admin authentication)
 pub fn create_admin_router() -> Router<AppState> {
     Router::new()
@@ -2424,6 +2460,7 @@ pub fn create_admin_router() -> Router<AppState> {
                 .route("/deployments", get(bi_list_deployments))
                 .route("/deployments/summary", get(bi_deployment_summary))
                 .route("/deployments/{id}/status-history", get(bi_status_history))
+                .route("/usage/daily-active-agents", get(bi_daily_active_agents))
                 .route("/usage", get(bi_usage))
                 .route("/usage/top", get(bi_top_consumers)),
         )

crates/api/tests/bi_metrics_tests.rs

@@ -1,5 +1,6 @@
 mod common;
 
+use chrono::{Duration, Utc};
 use common::{create_test_server_and_db, mock_login, TestServerConfig};
 use services::user::ports::UserRepository;
 use uuid::Uuid;
@@ -724,6 +725,128 @@ async fn test_bi_usage_with_type_filter() {
     cleanup(&db, inst1_id, inst2_id).await;
 }
 
+// =============================================================================
+// Daily active agents endpoint
+// =============================================================================
+
+#[tokio::test]
+async fn test_bi_daily_active_agents_with_data() {
+    let (server, db) = server_and_db().await;
+    let admin_token = mock_login(&server, "bi_admin_daily_agents@admin.org").await;
+
+    let user = db
+        .user_repository()
+        .get_user_by_email("bi_admin_daily_agents@admin.org")
+        .await
+        .unwrap()
+        .unwrap();
+
+    let (inst1_id, inst2_id) = seed_bi_test_data(&db, user.id.0).await;
+
+    // Date range that includes today (UTC) so the seeded usage events are counted
+    let now = Utc::now();
+    let start = (now - Duration::days(1))
+        .format("%Y-%m-%dT00:00:00Z")
+        .to_string();
+    let end = (now + Duration::days(1))
+        .format("%Y-%m-%dT00:00:00Z")
+        .to_string();
+    let url = format!(
+        "/v1/admin/bi/usage/daily-active-agents?start_date={}&end_date={}",
+        start, end,
+    );
+
+    let response = server
+        .get(&url)
+        .add_header(AUTH, auth_header(&admin_token))
+        .await;
+
+    assert_eq!(
+        response.status_code(),
+        200,
+        "daily-active-agents should return 200"
+    );
+
+    let body: serde_json::Value = response.json();
+    let points = body.as_array().expect("response should be array of points");
+    assert!(
+        !points.is_empty(),
+        "Should have at least one day in the range"
+    );
+
+    // Today (UTC) should have 2 active agents (both instances have usage > 0)
+    let today_str = now.format("%Y-%m-%d").to_string();
+    let today_point = points
+        .iter()
+        .find(|p| p.get("date").and_then(|v| v.as_str()).unwrap_or("") == today_str);
+    assert!(
+        today_point.is_some(),
+        "Should have a point for today {} in {:?}",
+        today_str,
+        points
+            .iter()
+            .map(|p| p.get("date").and_then(|v| v.as_str()).unwrap_or("?"))
+            .collect::<Vec<_>>()
+    );
+    let count = today_point
+        .unwrap()
+        .get("active_agents_count")
+        .unwrap()
+        .as_i64()
+        .unwrap();
+    assert!(
+        count >= 2,
+        "Today should have at least 2 active agents (we seeded 2 instances with usage), got {}",
+        count
+    );
+
+    // Breakdown by instance type: we seeded one openclaw and one ironclaw with usage
+    let by_type = today_point
+        .unwrap()
+        .get("by_instance_type")
+        .and_then(|v| v.as_array())
+        .unwrap_or(&vec![]);
+    let openclaw_count = by_type
+        .iter()
+        .find(|e| e.get("instance_type").and_then(|v| v.as_str()) == Some("openclaw"))
+        .and_then(|e| e.get("active_agents_count").and_then(|v| v.as_i64()))
+        .unwrap_or(0);
+    let ironclaw_count = by_type
+        .iter()
+        .find(|e| e.get("instance_type").and_then(|v| v.as_str()) == Some("ironclaw"))
+        .and_then(|e| e.get("active_agents_count").and_then(|v| v.as_i64()))
+        .unwrap_or(0);
+    assert!(
+        openclaw_count >= 1,
+        "Today should have at least 1 openclaw active agent, by_instance_type={:?}",
+        by_type
+    );
+    assert!(
+        ironclaw_count >= 1,
+        "Today should have at least 1 ironclaw active agent, by_instance_type={:?}",
+        by_type
+    );
+
+    cleanup(&db, inst1_id, inst2_id).await;
+}
+
+#[tokio::test]
+async fn test_bi_daily_active_agents_requires_date_range() {
+    let (server, _db) = server_and_db().await;
+    let admin_token = mock_login(&server, "bi_admin_daily_no_range@admin.org").await;
+
+    // No query params: validate_date_range passes (both None), repo returns []
+    let response = server
+        .get("/v1/admin/bi/usage/daily-active-agents")
+        .add_header(AUTH, auth_header(&admin_token))
+        .await;
+
+    assert_eq!(response.status_code(), 200);
+    let body: serde_json::Value = response.json();
+    let points = body.as_array().unwrap();
+    assert!(points.is_empty(), "No date range should return empty array");
+}
+
 // =============================================================================
 // Top consumers endpoint with seeded data
 // =============================================================================

crates/database/src/repositories/bi_metrics_repository.rs

@@ -1,11 +1,11 @@
 use crate::pool::DbPool;
 use async_trait::async_trait;
-use chrono::{DateTime, Utc};
+use chrono::{DateTime, Duration, Utc};
 use services::bi_metrics::{
-    BiMetricsRepository, DeploymentFilter, DeploymentRecord, DeploymentStatusCount,
-    DeploymentSummary, ListUsersFilter, ListUsersSort, StatusChangeRecord, TopConsumer,
-    TopConsumerFilter, TopConsumerGroupBy, UsageAggregation, UsageFilter, UsageGroupBy,
-    UsageRankBy, UserWithStats, UsersSortBy, UsersSortOrder,
+    BiMetricsRepository, DailyActiveAgentsByType, DailyActiveAgentsPoint, DeploymentFilter,
+    DeploymentRecord, DeploymentStatusCount, DeploymentSummary, ListUsersFilter, ListUsersSort,
+    StatusChangeRecord, TopConsumer, TopConsumerFilter, TopConsumerGroupBy, UsageAggregation,
+    UsageFilter, UsageGroupBy, UsageRankBy, UserWithStats, UsersSortBy, UsersSortOrder,
 };
 use services::user::ports::User;
 use services::UserId;
@@ -17,6 +17,9 @@ const AGGREGATION_TIMEOUT_MS: u32 = 30_000;
 /// Hard limit for unbounded queries (status history, usage aggregation).
 const MAX_UNBOUNDED_ROWS: i64 = 1000;
 
+/// Max date range (days) for daily active agents time series.
+const MAX_DAILY_ACTIVE_AGENTS_RANGE_DAYS: i64 = 365;
+
 pub struct PostgresBiMetricsRepository {
     pool: DbPool,
 }
@@ -744,4 +747,116 @@ WHERE 1=1
 
         Ok((users, total_count as u64))
     }
+
+    async fn get_daily_active_agents(
+        &self,
+        start_date: Option<DateTime<Utc>>,
+        end_date: Option<DateTime<Utc>>,
+    ) -> anyhow::Result<Vec<DailyActiveAgentsPoint>> {
+        let (start, end) = match (start_date, end_date) {
+            (Some(s), Some(e)) => {
+                if s >= e {
+                    return Ok(Vec::new());
+                }
+                let cap_end = if (e - s).num_days() > MAX_DAILY_ACTIVE_AGENTS_RANGE_DAYS {
+                    s + Duration::days(MAX_DAILY_ACTIVE_AGENTS_RANGE_DAYS)
+                } else {
+                    e
+                };
+                (s, cap_end)
+            }
+            _ => return Ok(Vec::new()),
+        };
+
+        let mut client = self.pool.get().await?;
+        let tx = client.build_transaction().start().await?;
+        tx.execute(
+            &format!("SET LOCAL statement_timeout = '{AGGREGATION_TIMEOUT_MS}'"),
+            &[],
+        )
+        .await?;
+
+        // Total per day: generate all days, left join to count of distinct instance_id
+        let rows = tx
+            .query(
+                r#"
+                WITH days AS (
+                    SELECT d::date AS day
+                    FROM generate_series($1::timestamptz, $2::timestamptz, '1 day'::interval) AS d
+                ),
+                daily_counts AS (
+                    SELECT (created_at AT TIME ZONE 'UTC')::date AS day,
+                           COUNT(DISTINCT instance_id)::bigint AS cnt
+                    FROM user_usage_event
+                    WHERE instance_id IS NOT NULL
+                      AND (quantity > 0 OR COALESCE(cost_nano_usd, 0) > 0)
+                      AND created_at >= $1
+                      AND created_at < $2 + interval '1 day'
+                    GROUP BY (created_at AT TIME ZONE 'UTC')::date
+                )
+                SELECT days.day, COALESCE(daily_counts.cnt, 0)::bigint
+                FROM days
+                LEFT JOIN daily_counts ON days.day = daily_counts.day
+                ORDER BY days.day
+                "#,
+                &[&start, &end],
+            )
+            .await?;
+
+        // Per day per instance_type: join to agent_instances to get type
+        let by_type_rows = tx
+            .query(
+                r#"
+                SELECT (u.created_at AT TIME ZONE 'UTC')::date AS day,
+                       ai.type AS instance_type,
+                       COUNT(DISTINCT u.instance_id)::bigint AS cnt
+                FROM user_usage_event u
+                JOIN agent_instances ai ON u.instance_id = ai.id
+                WHERE u.instance_id IS NOT NULL
+                  AND (u.quantity > 0 OR COALESCE(u.cost_nano_usd, 0) > 0)
+                  AND u.created_at >= $1
+                  AND u.created_at < $2 + interval '1 day'
+                GROUP BY (u.created_at AT TIME ZONE 'UTC')::date, ai.type
+                ORDER BY day, instance_type
+                "#,
+                &[&start, &end],
+            )
+            .await?;
+
+        tx.commit().await?;
+
+        // Index by day: day -> Vec<(instance_type, count)>
+        let mut by_type_per_day: std::collections::HashMap<
+            chrono::NaiveDate,
+            Vec<DailyActiveAgentsByType>,
+        > = std::collections::HashMap::new();
+        for r in by_type_rows {
+            let day: chrono::NaiveDate = r.get(0);
+            let instance_type: String = r.get(1);
+            let count: i64 = r.get(2);
+            by_type_per_day
+                .entry(day)
+                .or_default()
+                .push(DailyActiveAgentsByType {
+                    instance_type,
+                    active_agents_count: count,
+                });
+        }
+
+        let points = rows
+            .into_iter()
+            .map(|r| {
+                let day: chrono::NaiveDate = r.get(0);
+                let count: i64 = r.get(1);
+                let by_instance_type = by_type_per_day.remove(&day).unwrap_or_default();
+                DailyActiveAgentsPoint {
+                    date: day.format("%Y-%m-%d").to_string(),
+                    active_agents_count: count,
+                    by_instance_type,
+                }
+            })
+            .collect();
+
+        Ok(points)
+    }
 }