diff --git a/src/agent/dispatcher.rs b/src/agent/dispatcher.rs index eabf5dc4..ca5c5a86 100644 --- a/src/agent/dispatcher.rs +++ b/src/agent/dispatcher.rs @@ -1286,6 +1286,7 @@ mod tests { max_tool_iterations: 50, auto_approve_tools: false, default_timezone: "UTC".to_string(), + max_jobs_per_user: None, max_tokens_per_job: 0, multi_tenant: false, }, @@ -2156,6 +2157,7 @@ mod tests { max_tool_iterations, auto_approve_tools: true, default_timezone: "UTC".to_string(), + max_jobs_per_user: None, max_tokens_per_job: 0, multi_tenant: false, }, @@ -2279,6 +2281,7 @@ mod tests { max_tool_iterations: max_iter, auto_approve_tools: true, default_timezone: "UTC".to_string(), + max_jobs_per_user: None, max_tokens_per_job: 0, multi_tenant: false, }, diff --git a/src/agent/scheduler.rs b/src/agent/scheduler.rs index a082fe23..d14bd672 100644 --- a/src/agent/scheduler.rs +++ b/src/agent/scheduler.rs @@ -267,6 +267,20 @@ impl Scheduler { }); } + // Per-user concurrency check + if let Some(max_per_user) = self.config.max_jobs_per_user + && let Ok(ctx) = self.context_manager.get_context(job_id).await + { + let user_active = self + .context_manager + .active_jobs_for(&ctx.user_id) + .await + .len(); + if user_active >= max_per_user { + return Err(JobError::MaxJobsExceeded { max: max_per_user }); + } + } + // Transition job to in_progress self.context_manager .update_context(job_id, |ctx| { @@ -784,6 +798,7 @@ mod tests { max_tool_iterations: 10, auto_approve_tools: true, default_timezone: "UTC".to_string(), + max_jobs_per_user: None, max_tokens_per_job, multi_tenant: false, }; diff --git a/src/channels/web/handlers/users.rs b/src/channels/web/handlers/users.rs index 185d0bc1..01c92408 100644 --- a/src/channels/web/handlers/users.rs +++ b/src/channels/web/handlers/users.rs @@ -268,3 +268,136 @@ pub async fn users_activate_handler( "status": "active", }))) } + +/// DELETE /api/admin/users/{id} — delete a user and all their data. +pub async fn users_delete_handler( + State(state): State>, + AdminUser(_user): AdminUser, + Path(id): Path, +) -> Result, (StatusCode, String)> { + let store = state.store.as_ref().ok_or(( + StatusCode::SERVICE_UNAVAILABLE, + "Database not available".to_string(), + ))?; + + let deleted = store + .delete_user(&id) + .await + .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?; + + if !deleted { + return Err((StatusCode::NOT_FOUND, "User not found".to_string())); + } + + Ok(Json(serde_json::json!({ + "id": id, + "deleted": true, + }))) +} + +/// GET /api/profile — get the authenticated user's own profile. +pub async fn profile_get_handler( + State(state): State>, + AuthenticatedUser(user): AuthenticatedUser, +) -> Result, (StatusCode, String)> { + let store = state.store.as_ref().ok_or(( + StatusCode::SERVICE_UNAVAILABLE, + "Database not available".to_string(), + ))?; + + let record = store + .get_user(&user.user_id) + .await + .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))? + .ok_or((StatusCode::NOT_FOUND, "User not found".to_string()))?; + + Ok(Json(serde_json::json!({ + "id": record.id, + "email": record.email, + "display_name": record.display_name, + "status": record.status, + "role": record.role, + "created_at": record.created_at.to_rfc3339(), + "last_login_at": record.last_login_at.map(|dt| dt.to_rfc3339()), + }))) +} + +/// PATCH /api/profile — update the authenticated user's own profile. +pub async fn profile_update_handler( + State(state): State>, + AuthenticatedUser(user): AuthenticatedUser, + Json(body): Json, +) -> Result, (StatusCode, String)> { + let store = state.store.as_ref().ok_or(( + StatusCode::SERVICE_UNAVAILABLE, + "Database not available".to_string(), + ))?; + + let current = store + .get_user(&user.user_id) + .await + .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))? + .ok_or((StatusCode::NOT_FOUND, "User not found".to_string()))?; + + let display_name = body + .get("display_name") + .and_then(|v| v.as_str()) + .unwrap_or(¤t.display_name); + let metadata = body.get("metadata").unwrap_or(¤t.metadata); + + store + .update_user_profile(&user.user_id, display_name, metadata) + .await + .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?; + + Ok(Json(serde_json::json!({ + "id": user.user_id, + "display_name": display_name, + "updated": true, + }))) +} + +/// GET /api/admin/usage — per-user LLM usage stats. +pub async fn usage_stats_handler( + State(state): State>, + AdminUser(_user): AdminUser, + axum::extract::Query(params): axum::extract::Query>, +) -> Result, (StatusCode, String)> { + let store = state.store.as_ref().ok_or(( + StatusCode::SERVICE_UNAVAILABLE, + "Database not available".to_string(), + ))?; + + let user_id = params.get("user_id").map(|s| s.as_str()); + let period = params.get("period").map(|s| s.as_str()).unwrap_or("day"); + let since = match period { + "week" => chrono::Utc::now() - chrono::Duration::days(7), + "month" => chrono::Utc::now() - chrono::Duration::days(30), + _ => chrono::Utc::now() - chrono::Duration::days(1), + }; + + let stats = store + .user_usage_stats(user_id, since) + .await + .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?; + + let entries: Vec = stats + .iter() + .map(|s| { + serde_json::json!({ + "user_id": s.user_id, + "model": s.model, + "call_count": s.call_count, + "input_tokens": s.input_tokens, + "output_tokens": s.output_tokens, + "total_cost": s.total_cost.to_string(), + }) + }) + .collect(); + + Ok(Json(serde_json::json!({ + "period": period, + "since": since.to_rfc3339(), + "usage": entries, + }))) +} diff --git a/src/channels/web/server.rs b/src/channels/web/server.rs index ab36fd05..0016d15e 100644 --- a/src/channels/web/server.rs +++ b/src/channels/web/server.rs @@ -519,7 +519,8 @@ pub async fn start_server( .route( "/api/admin/users/{id}", get(super::handlers::users::users_detail_handler) - .patch(super::handlers::users::users_update_handler), + .patch(super::handlers::users::users_update_handler) + .delete(super::handlers::users::users_delete_handler), ) .route( "/api/admin/users/{id}/suspend", @@ -529,6 +530,17 @@ pub async fn start_server( "/api/admin/users/{id}/activate", post(super::handlers::users::users_activate_handler), ) + // Usage reporting (admin) + .route( + "/api/admin/usage", + get(super::handlers::users::usage_stats_handler), + ) + // User self-service profile + .route( + "/api/profile", + get(super::handlers::users::profile_get_handler) + .patch(super::handlers::users::profile_update_handler), + ) // Token management .route( "/api/tokens", diff --git a/src/config/agent.rs b/src/config/agent.rs index 06629d3f..15a8a22d 100644 --- a/src/config/agent.rs +++ b/src/config/agent.rs @@ -31,6 +31,8 @@ pub struct AgentConfig { pub auto_approve_tools: bool, /// Default timezone for new sessions (IANA name, e.g. "America/New_York"). pub default_timezone: String, + /// Maximum concurrent jobs per user. None = use global max_parallel_jobs. + pub max_jobs_per_user: Option, /// Maximum tokens per job (0 = unlimited). pub max_tokens_per_job: u64, /// Whether the deployment is multi-tenant (multiple users sharing one @@ -59,6 +61,7 @@ impl AgentConfig { max_tool_iterations: 10, auto_approve_tools: true, default_timezone: "UTC".to_string(), + max_jobs_per_user: None, max_tokens_per_job: 0, multi_tenant: false, } @@ -117,6 +120,7 @@ impl AgentConfig { } tz }, + max_jobs_per_user: parse_option_env("MAX_JOBS_PER_USER")?, max_tokens_per_job: parse_optional_env( "AGENT_MAX_TOKENS_PER_JOB", settings.agent.max_tokens_per_job, diff --git a/src/db/libsql/users.rs b/src/db/libsql/users.rs index 34928723..85e2a287 100644 --- a/src/db/libsql/users.rs +++ b/src/db/libsql/users.rs @@ -385,6 +385,117 @@ impl UserStore for LibSqlBackend { .is_some(); Ok(has_users) } + + async fn delete_user(&self, id: &str) -> Result { + let conn = self.connect().await?; + // Delete from child tables first to avoid FK violations. + // agent_jobs cascades to job_actions, llm_calls, estimation_snapshots + // conversations cascades to conversation_messages + // memory_documents cascades to memory_chunks + // routines cascades to routine_runs + for table in &[ + "settings", + "heartbeat_state", + "tool_rate_limit_state", + "secret_usage_log", + "leak_detection_events", + "secrets", + "wasm_tools", + "routines", + "memory_documents", + "agent_jobs", + "conversations", + ] { + conn.execute( + &format!("DELETE FROM {} WHERE user_id = ?1", table), + params![id], + ) + .await + .map_err(|e| DatabaseError::Query(e.to_string()))?; + } + // Nullify self-referencing created_by before deleting the user + conn.execute( + "UPDATE users SET created_by = NULL WHERE created_by = ?1", + params![id], + ) + .await + .map_err(|e| DatabaseError::Query(e.to_string()))?; + // api_tokens cascade automatically via FK + let rows = conn + .execute("DELETE FROM users WHERE id = ?1", params![id]) + .await + .map_err(|e| DatabaseError::Query(e.to_string()))?; + Ok(rows > 0) + } + + async fn user_usage_stats( + &self, + user_id: Option<&str>, + since: DateTime, + ) -> Result, DatabaseError> { + let conn = self.connect().await?; + let since_str = fmt_ts(&since); + let mut rows = if let Some(uid) = user_id { + conn.query( + r#" + SELECT j.user_id, l.model, COUNT(*) as call_count, + COALESCE(SUM(l.input_tokens), 0) as input_tokens, + COALESCE(SUM(l.output_tokens), 0) as output_tokens, + COALESCE(SUM(l.cost), 0) as total_cost + FROM llm_calls l + JOIN agent_jobs j ON l.job_id = j.id + WHERE l.created_at >= ?1 + AND j.user_id = ?2 + GROUP BY j.user_id, l.model + ORDER BY total_cost DESC + "#, + params![since_str, uid], + ) + .await + .map_err(|e| DatabaseError::Query(e.to_string()))? + } else { + conn.query( + r#" + SELECT j.user_id, l.model, COUNT(*) as call_count, + COALESCE(SUM(l.input_tokens), 0) as input_tokens, + COALESCE(SUM(l.output_tokens), 0) as output_tokens, + COALESCE(SUM(l.cost), 0) as total_cost + FROM llm_calls l + JOIN agent_jobs j ON l.job_id = j.id + WHERE l.created_at >= ?1 + GROUP BY j.user_id, l.model + ORDER BY total_cost DESC + "#, + params![since_str], + ) + .await + .map_err(|e| DatabaseError::Query(e.to_string()))? + }; + let mut stats = Vec::new(); + while let Some(row) = rows + .next() + .await + .map_err(|e| DatabaseError::Query(e.to_string()))? + { + let cost_str = get_text(&row, 5); + let total_cost = rust_decimal::Decimal::from_str_exact(&cost_str).unwrap_or_default(); + stats.push(crate::db::UserUsageStats { + user_id: get_text(&row, 0), + model: get_text(&row, 1), + call_count: row + .get::(2) + .map_err(|e| DatabaseError::Query(e.to_string()))?, + input_tokens: row + .get::(3) + .map_err(|e| DatabaseError::Query(e.to_string()))?, + output_tokens: row + .get::(4) + .map_err(|e| DatabaseError::Query(e.to_string()))?, + total_cost, + }); + } + Ok(stats) + } } #[cfg(test)] diff --git a/src/db/mod.rs b/src/db/mod.rs index aad26c0e..1e5f36cf 100644 --- a/src/db/mod.rs +++ b/src/db/mod.rs @@ -848,6 +848,29 @@ pub trait UserStore: Send + Sync { /// Check whether any user records exist (for first-run bootstrap detection). async fn has_any_users(&self) -> Result; + + /// Delete a user and all their data across all user-scoped tables. + /// Returns false if the user doesn't exist. + async fn delete_user(&self, id: &str) -> Result; + + /// Get per-user LLM usage stats for a time period. + /// Aggregates from llm_calls via agent_jobs.user_id. + async fn user_usage_stats( + &self, + user_id: Option<&str>, + since: DateTime, + ) -> Result, DatabaseError>; +} + +/// Per-user LLM usage statistics. +#[derive(Debug, Clone)] +pub struct UserUsageStats { + pub user_id: String, + pub model: String, + pub call_count: i64, + pub input_tokens: i64, + pub output_tokens: i64, + pub total_cost: Decimal, } /// Backend-agnostic database supertrait. diff --git a/src/db/postgres.rs b/src/db/postgres.rs index 31ef0c8e..14f38700 100644 --- a/src/db/postgres.rs +++ b/src/db/postgres.rs @@ -861,4 +861,16 @@ impl UserStore for PgBackend { async fn has_any_users(&self) -> Result { self.store.has_any_users().await } + + async fn delete_user(&self, id: &str) -> Result { + self.store.delete_user(id).await + } + + async fn user_usage_stats( + &self, + user_id: Option<&str>, + since: DateTime, + ) -> Result, DatabaseError> { + self.store.user_usage_stats(user_id, since).await + } } diff --git a/src/history/store.rs b/src/history/store.rs index 46a18d70..17ebc9f6 100644 --- a/src/history/store.rs +++ b/src/history/store.rs @@ -2536,6 +2536,100 @@ impl Store { .await?; Ok(row.get("has_users")) } + + /// Delete a user and all their data across all user-scoped tables. + /// Returns false if the user doesn't exist. + pub async fn delete_user(&self, id: &str) -> Result { + let conn = self.conn().await?; + // Delete from child tables first to avoid FK violations. + // agent_jobs cascades to job_actions, llm_calls, estimation_snapshots + // conversations cascades to conversation_messages + // memory_documents cascades to memory_chunks + // routines cascades to routine_runs + for table in &[ + "settings", + "heartbeat_state", + "tool_rate_limit_state", + "secret_usage_log", + "leak_detection_events", + "secrets", + "wasm_tools", + "routines", + "memory_documents", + "agent_jobs", + "conversations", + ] { + conn.execute(&format!("DELETE FROM {} WHERE user_id = $1", table), &[&id]) + .await?; + } + // Nullify self-referencing created_by before deleting the user + conn.execute( + "UPDATE users SET created_by = NULL WHERE created_by = $1", + &[&id], + ) + .await?; + // api_tokens cascade automatically via FK + let result = conn + .execute("DELETE FROM users WHERE id = $1", &[&id]) + .await?; + Ok(result > 0) + } + + /// Get per-user LLM usage stats for a time period. + /// Aggregates from llm_calls via agent_jobs.user_id. + pub async fn user_usage_stats( + &self, + user_id: Option<&str>, + since: DateTime, + ) -> Result, DatabaseError> { + let conn = self.conn().await?; + let rows = if let Some(uid) = user_id { + conn.query( + r#" + SELECT j.user_id, l.model, COUNT(*) as call_count, + COALESCE(SUM(l.input_tokens), 0) as input_tokens, + COALESCE(SUM(l.output_tokens), 0) as output_tokens, + COALESCE(SUM(l.cost), 0) as total_cost + FROM llm_calls l + JOIN agent_jobs j ON l.job_id = j.id + WHERE l.created_at >= $1 + AND j.user_id = $2 + GROUP BY j.user_id, l.model + ORDER BY total_cost DESC + "#, + &[&since, &uid], + ) + .await? + } else { + conn.query( + r#" + SELECT j.user_id, l.model, COUNT(*) as call_count, + COALESCE(SUM(l.input_tokens), 0) as input_tokens, + COALESCE(SUM(l.output_tokens), 0) as output_tokens, + COALESCE(SUM(l.cost), 0) as total_cost + FROM llm_calls l + JOIN agent_jobs j ON l.job_id = j.id + WHERE l.created_at >= $1 + GROUP BY j.user_id, l.model + ORDER BY total_cost DESC + "#, + &[&since], + ) + .await? + }; + let mut stats = Vec::with_capacity(rows.len()); + for row in &rows { + stats.push(crate::db::UserUsageStats { + user_id: row.get("user_id"), + model: row.get("model"), + call_count: row.get("call_count"), + input_tokens: row.get("input_tokens"), + output_tokens: row.get("output_tokens"), + total_cost: row.get("total_cost"), + }); + } + Ok(stats) + } } #[cfg(feature = "postgres")]