feat: refresh AI services with routing, billing and usage insights

backend/src/api_docs.rs

@@ -52,6 +52,7 @@
         // AI Services (unified key management)
         crate::handlers::keys::create_key,
         crate::handlers::keys::list_keys,
+        crate::handlers::service_insights::get_insights,
         crate::handlers::keys::get_key,
         crate::handlers::keys::get_key_authorization,
         crate::handlers::keys::update_key,

backend/src/billing_integration_tests/usage.rs

@@ -19,6 +19,7 @@ fn meter(owner: &str, quantity: i64) -> UsageMeterRow {
         wallet_id: Some("wallet".into()),
         actor_user_id: owner.to_string(),
         api_key_id: None,
+        user_service_id: None,
         service_id: Some("service".into()),
         service_slug: Some("llm-test".into()),
         metric: BillingMetric::Tokens,
@@ -52,7 +53,10 @@ async fn read_usage(state: &crate::AppState, actor: &str) -> BillingUsageRespons
     billing::get_usage(
         State(state.clone()),
         test_auth_user(actor),
-        Query(UsageQuery { period: None }),
+        Query(UsageQuery {
+            period: None,
+            bucket: None,
+        }),
     )
     .await
     .expect("usage read")
@@ -968,3 +972,63 @@ async fn execution_owner_preserves_org_acl_for_non_master_credentials() {
     ));
     db.drop().await.unwrap();
 }
+
+#[tokio::test]
+async fn day_buckets_split_usage_by_utc_day_without_changing_totals() {
+    let Some(db) = connect_test_database("billing_usage_day_buckets").await else {
+        return;
+    };
+    let owner = insert_owner(&db).await;
+    let state = billing_route_state(db.clone(), Arc::new(FakeLago::default()), 0);
+    let today = Utc::now();
+    let yesterday = today - chrono::Duration::days(1);
+    for (at, quantity) in [(yesterday, 4), (today, 5), (today, 6)] {
+        let mut row = meter(&owner, quantity);
+        row.wallet_id = None;
+        row.created_at = at;
+        db.collection::<UsageMeterRow>(USAGE_METER)
+            .insert_one(row)
+            .await
+            .unwrap();
+    }
+    let flat = read_usage(&state, &owner).await;
+    assert_eq!(flat.rows.len(), 1);
+    assert!(flat.rows[0].day.is_none());
+    let daily = billing::get_usage(
+        State(state.clone()),
+        test_auth_user(&owner),
+        Query(UsageQuery {
+            period: Some("7d".into()),
+            bucket: Some("day".into()),
+        }),
+    )
+    .await
+    .expect("daily usage read")
+    .0;
+    let days: Vec<_> = daily
+        .rows
+        .iter()
+        .map(|row| {
+            (
+                row.day.expect("bucketed day").date_naive(),
+                row.quantity,
+                row.events,
+            )
+        })
+        .collect();
+    assert_eq!(
+        days,
+        vec![(yesterday.date_naive(), 4, 1), (today.date_naive(), 11, 2),]
+    );
+    assert_eq!(daily.totals.quantity, flat.totals.quantity);
+    let invalid = billing::get_usage(
+        State(state),
+        test_auth_user(&owner),
+        Query(UsageQuery {
+            period: None,
+            bucket: Some("hour".into()),
+        }),
+    )
+    .await;
+    assert!(matches!(invalid, Err(AppError::ValidationError(_))));
+}

backend/src/db.rs

@@ -513,6 +513,14 @@ async fn ensure_core_indexes(db: &Database) -> Result<(), mongodb::error::Error>
     }
     // ── audit_log ──
     let audit = db.collection::<mongodb::bson::Document>("audit_log");
+    audit
+        .create_index(
+            IndexModel::builder()
+                .keys(doc! { "event_type": 1, "event_data.user_service_id": 1, "created_at": -1, "_id": -1 })
+                .options(IndexOptions::builder().name("audit_service_requests".to_string()).build())
+                .build(),
+        )
+        .await?;
     audit
         .create_index(
             IndexModel::builder()
@@ -4407,6 +4415,12 @@ async fn migrate_provider_tokens(db: &Database) -> Result<(), Box<dyn std::error
 
         // Create UserApiKey -- clean up endpoint on failure
         let api_key = UserApiKey {
+            oauth_app_observation: (token.token_type == "oauth2").then(|| {
+                crate::services::oauth_app_source::OAuthAppSource::from_credential_owner(
+                    token.credential_user_id.as_deref(),
+                )
+                .observation(1)
+            }),
             credential_source: None,
             id: api_key_id.clone(),
             user_id: token.user_id.clone(),
@@ -4657,6 +4671,7 @@ async fn migrate_service_connections(db: &Database) -> Result<(), Box<dyn std::e
             .or_else(|| service.auth_type.clone())
             .unwrap_or_else(|| "api_key".to_string());
         let api_key = UserApiKey {
+            oauth_app_observation: None,
             credential_source: None,
             id: api_key_id.clone(),
             user_id: conn.user_id.clone(),
@@ -4924,6 +4939,7 @@ async fn migrate_node_service_bindings(db: &Database) -> Result<(), Box<dyn std:
 
         // Create UserApiKey (placeholder -- node-managed or SSH certificate)
         let api_key = UserApiKey {
+            oauth_app_observation: None,
             credential_source: None,
             id: api_key_id.clone(),
             user_id: binding.user_id.clone(),

backend/src/handlers/agent_bindings.rs

@@ -676,6 +676,7 @@ mod tests {
 
     fn fixture_user_api_key(id: &str, user_id: &str) -> UserApiKey {
         UserApiKey {
+            oauth_app_observation: None,
             credential_source: None,
             id: id.to_string(),
             user_id: user_id.to_string(),

backend/src/handlers/assistant_action_effects_endpoints.rs

@@ -949,6 +949,7 @@ mod tests {
 
     fn fixture_external_key(key_id: &str, user_id: &str, label: &str) -> UserApiKey {
         UserApiKey {
+            oauth_app_observation: None,
             credential_epoch: 1,
             credential_source: None,
             id: key_id.to_string(),

backend/src/handlers/assistant_action_effects_keys.rs

@@ -974,6 +974,7 @@ mod tests {
 
     fn fixture_user_api_key(id: &str, user_id: &str) -> UserApiKey {
         UserApiKey {
+            oauth_app_observation: None,
             credential_epoch: 1,
             credential_source: None,
             id: id.to_string(),

backend/src/handlers/assistant_action_effects_services.rs

@@ -1131,6 +1131,7 @@ mod tests {
 
     fn fixture_api_key(id: &str, user_id: &str) -> UserApiKey {
         UserApiKey {
+            oauth_app_observation: None,
             credential_epoch: 1,
             credential_source: None,
             id: id.to_string(),

backend/src/handlers/billing.rs

@@ -35,6 +35,8 @@ const LAGO_HMAC_SHA256_ALGORITHM: &str = "hmac";
 #[derive(Debug, Deserialize)]
 pub struct UsageQuery {
     pub period: Option<String>,
+    /// `day` additionally splits rows by UTC calendar day for usage charts.
+    pub bucket: Option<String>,
 }
 
 #[derive(Debug, Serialize, ToSchema)]
@@ -81,6 +83,9 @@ pub struct BillingUsageRow {
     /// only). None when no row in the group carried a breakdown.
     #[serde(skip_serializing_if = "Option::is_none")]
     pub token_breakdown: Option<crate::models::service_billing::TokenBreakdown>,
+    /// UTC day start; present only when requested with `bucket=day`.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub day: Option<chrono::DateTime<Utc>>,
 }
 
 #[derive(Debug, Serialize, ToSchema)]
@@ -215,7 +220,8 @@ pub struct InvoiceDownloadResponse {
     path = "/api/v1/billing/usage",
     tag = "Billing",
     params(
-        ("period" = Option<String>, Query, description = "Usage period: 24h, 7d, 30d, 90d, or all")
+        ("period" = Option<String>, Query, description = "Usage period: 24h, 7d, 30d, 90d, or all"),
+        ("bucket" = Option<String>, Query, description = "`day` splits rows by UTC day")
     ),
     responses(
         (status = 200, description = "Billing usage summary", body = BillingUsageResponse)
@@ -230,6 +236,15 @@ pub async fn get_usage(
     let owner_id = auth_user.user_id.to_string();
     let period = query.period.unwrap_or_else(|| "30d".to_string());
     let since = period_start(&period);
+    let by_day = match query.bucket.as_deref() {
+        None => false,
+        Some("day") => true,
+        Some(_) => {
+            return Err(AppError::ValidationError(
+                "bucket must be `day` when supplied".to_string(),
+            ));
+        }
+    };
     let mut match_doc = doc! {
         "billing_owner_id": &owner_id,
         "quantity": { "$ne": null },
@@ -296,6 +311,13 @@ pub async fn get_usage(
                     // only (service not platform_billable); they carry no
                     // cost and are never pushed to Lago.
                     "billable": { "$ne": [{ "$ifNull": ["$wallet_id", null] }, null] },
+                    "day": if by_day {
+                        bson::Bson::Document(doc! { "$dateTrunc": {
+                            "date": "$created_at", "unit": "day", "timezone": "UTC",
+                        } })
+                    } else {
+                        bson::Bson::Null
+                    },
                 },
                 "quantity": { "$sum": "$quantity" },
                 // Keep exact settlements separate from historical estimates.
@@ -338,7 +360,7 @@ pub async fn get_usage(
                 "cache_creation_tokens": { "$sum": { "$ifNull": ["$token_breakdown.cache_creation_tokens", 0] } },
             }
         },
-        doc! { "$sort": { "_id.service_slug": 1, "_id.layer": 1, "_id.metric": 1 } },
+        doc! { "$sort": { "_id.service_slug": 1, "_id.layer": 1, "_id.metric": 1, "_id.day": 1 } },
     ];
 
     let mut cursor = state
@@ -404,6 +426,7 @@ pub async fn get_usage(
                 0
             },
             token_breakdown: usage_row_breakdown(&doc),
+            day: id_doc.get_datetime("day").ok().map(|day| day.to_chrono()),
         });
     }
 

backend/src/handlers/delegation.rs

@@ -3015,6 +3015,7 @@ mod tests {
             .db
             .collection::<UserApiKey>(USER_API_KEYS)
             .insert_one(UserApiKey {
+                oauth_app_observation: None,
                 id: credential_id.to_string(),
                 user_id: TEST_USER_ID.to_string(),
                 label: "full-router credential".to_string(),
@@ -3497,6 +3498,7 @@ mod tests {
             .db
             .collection::<UserApiKey>(USER_API_KEYS)
             .insert_one(UserApiKey {
+                oauth_app_observation: None,
                 id: credential_id.to_string(),
                 user_id: TEST_USER_ID.to_string(),
                 label: "ac5 credential".to_string(),
@@ -3662,6 +3664,7 @@ mod tests {
             .db
             .collection::<UserApiKey>(USER_API_KEYS)
             .insert_one(UserApiKey {
+                oauth_app_observation: None,
                 id: "00000000-0000-4000-8000-000000000713".to_string(),
                 user_id: TEST_USER_ID.to_string(),
                 label: "oauth refresh canary".to_string(),