From d9b3bb4d94a9d331ccaa89e3b4aae584a0d005aa Mon Sep 17 00:00:00 2001 From: Tracewebstudio Dev Date: Sat, 15 Aug 2026 14:20:44 +0200 Subject: [PATCH] feat(ai-guard): add semantic moderation layer + guard violations DB logging MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add llm_moderation_check() — async call to LiteLLM /moderations endpoint (routes to OpenAI Moderation API, free) for hate/sexual/violence detection. Fails open: if LiteLLM is unreachable, message is allowed through with a warning log so core chat is never broken by a network hiccup. - Add spawn_guard_log() — fire-and-forget tokio task that persists every guard rejection (keyword AND moderation) to ai_guard_violations table. - Wire both guards into ai_chat_message, ai_chat_ask, and ai_chat_stream. - Add migration: ai_guard_violations(id, user_id, message_excerpt, guard_type, reason, categories JSONB, created_at). - Add GET /api/admin/ai/guard-events with filter by guard_type, pagination. Co-Authored-By: Claude Sonnet 4.6 --- apps/users/src/handlers/admin_ai.rs | 108 ++++++++++++- apps/users/src/handlers/ai.rs | 149 +++++++++++++++++- ...000001_create_ai_guard_violations.down.sql | 1 + ...15000001_create_ai_guard_violations.up.sql | 16 ++ job_seekers.pid | 2 +- 5 files changed, 272 insertions(+), 4 deletions(-) create mode 100644 crates/db/migrations_new/20260815000001_create_ai_guard_violations.down.sql create mode 100644 crates/db/migrations_new/20260815000001_create_ai_guard_violations.up.sql diff --git a/apps/users/src/handlers/admin_ai.rs b/apps/users/src/handlers/admin_ai.rs index 5b7e8f7..e31198c 100644 --- a/apps/users/src/handlers/admin_ai.rs +++ b/apps/users/src/handlers/admin_ai.rs @@ -12,7 +12,7 @@ use db::models::ai::{ AiCreditTransactionRepository, AiFeatureCostRepository, AiPlanRepository, AiUsageLogRepository, UserAiSubscriptionRepository, }; -use serde::Deserialize; +use serde::{Deserialize, Serialize}; use uuid::Uuid; #[derive(Debug, Deserialize)] @@ -72,6 +72,112 @@ pub fn admin_ai_router() -> Router { .route("/users/{user_id}/transactions", get(user_credit_transactions)) .route("/plans", get(list_plans)) .route("/features", get(list_features)) + .route("/guard-events", get(guard_events)) +} + +// ── GET /api/admin/ai/guard-events ──────────────────────────────────────────── +// Paginated list of messages blocked by the AI content guard. +// Query params: limit (default 50, max 200), offset (default 0), guard_type filter. + +#[derive(Debug, Serialize)] +struct GuardEventRow { + id: uuid::Uuid, + user_id: Option, + message_excerpt: String, + guard_type: String, + reason: String, + categories: Option, + created_at: chrono::DateTime, +} + +#[derive(Debug, sqlx::FromRow)] +struct GuardEventDb { + id: uuid::Uuid, + user_id: Option, + message_excerpt: String, + guard_type: String, + reason: String, + categories: Option, + created_at: chrono::DateTime, +} + +#[derive(Debug, serde::Deserialize)] +struct GuardEventsQuery { + #[serde(default = "default_limit")] + limit: i64, + #[serde(default)] + offset: i64, + guard_type: Option, +} + +async fn guard_events( + auth: AuthUser, + State(state): State, + Query(q): Query, +) -> impl IntoResponse { + if let Err(e) = require_admin(&auth) { + return e.into_response(); + } + + let limit = q.limit.clamp(1, 200); + let offset = q.offset.max(0); + + let rows: Vec = if let Some(ref gt) = q.guard_type { + sqlx::query_as::<_, GuardEventDb>( + "SELECT id, user_id, message_excerpt, guard_type, reason, categories, created_at \ + FROM ai_guard_violations \ + WHERE guard_type = $1 \ + ORDER BY created_at DESC \ + LIMIT $2 OFFSET $3", + ) + .bind(gt) + .bind(limit) + .bind(offset) + .fetch_all(&state.pool) + .await + .unwrap_or_default() + } else { + sqlx::query_as::<_, GuardEventDb>( + "SELECT id, user_id, message_excerpt, guard_type, reason, categories, created_at \ + FROM ai_guard_violations \ + ORDER BY created_at DESC \ + LIMIT $1 OFFSET $2", + ) + .bind(limit) + .bind(offset) + .fetch_all(&state.pool) + .await + .unwrap_or_default() + }; + + let total: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM ai_guard_violations") + .fetch_one(&state.pool) + .await + .unwrap_or(0); + + let events: Vec = rows + .into_iter() + .map(|r| GuardEventRow { + id: r.id, + user_id: r.user_id, + message_excerpt: r.message_excerpt, + guard_type: r.guard_type, + reason: r.reason, + categories: r.categories, + created_at: r.created_at, + }) + .collect(); + + ( + StatusCode::OK, + Json(serde_json::json!({ + "events": events, + "total": total, + "limit": limit, + "offset": offset, + })), + ) + .into_response() } async fn admin_support_reply( diff --git a/apps/users/src/handlers/ai.rs b/apps/users/src/handlers/ai.rs index d8f2140..83aa5e4 100644 --- a/apps/users/src/handlers/ai.rs +++ b/apps/users/src/handlers/ai.rs @@ -294,6 +294,121 @@ fn llm_guard_check(message: &str) -> Option<(StatusCode, serde_json::Value)> { None } +/// Persist a guard violation to the DB in a fire-and-forget task. +/// Never blocks the request — if the insert fails, we just log a warning. +fn spawn_guard_log( + pool: sqlx::PgPool, + user_id: Option, + message: &str, + guard_type: &'static str, + reason: &str, + categories: Option, +) { + let excerpt: String = message.chars().take(200).collect(); + let reason = reason.to_owned(); + tokio::spawn(async move { + let result = sqlx::query( + "INSERT INTO ai_guard_violations \ + (user_id, message_excerpt, guard_type, reason, categories) \ + VALUES ($1, $2, $3, $4, $5)" + ) + .bind(user_id) + .bind(&excerpt) + .bind(guard_type) + .bind(&reason) + .bind(&categories) + .execute(&pool) + .await; + if let Err(e) = result { + tracing::warn!("Failed to persist guard violation: {}", e); + } + }); +} + +/// Call the OpenAI Moderation API via LiteLLM for semantic harm detection +/// (hate speech, sexual content, violence, self-harm, etc.). +/// +/// Fails **open** — if LiteLLM is unreachable or returns an unexpected shape, +/// we log a warning and allow the message through so the core chat experience +/// is never broken by a network hiccup. +async fn llm_moderation_check( + message: &str, +) -> Option<(StatusCode, serde_json::Value, serde_json::Value)> { + let litellm_base = std::env::var("LITELLM_BASE_URL") + .unwrap_or_else(|_| "http://litellm.nxtgauge-ai.svc.cluster.local:4000".to_string()); + let api_key = std::env::var("LITELLM_API_KEY").unwrap_or_default(); + let url = format!("{}/moderations", litellm_base.trim_end_matches('/')); + + let client = reqwest::Client::builder() + .timeout(std::time::Duration::from_secs(3)) + .build() + .unwrap_or_default(); + + let body = serde_json::json!({ "input": message }); + + let resp = match client + .post(&url) + .header("Authorization", format!("Bearer {}", api_key)) + .header("Content-Type", "application/json") + .json(&body) + .send() + .await + { + Ok(r) => r, + Err(e) => { + tracing::warn!("Moderation API unreachable, failing open: {}", e); + return None; + } + }; + + let json: serde_json::Value = match resp.json().await { + Ok(j) => j, + Err(e) => { + tracing::warn!("Moderation API bad JSON, failing open: {}", e); + return None; + } + }; + + let result = &json["results"][0]; + let flagged = result["flagged"].as_bool().unwrap_or(false); + if !flagged { + return None; + } + + // Collect which categories fired + let categories = result["categories"].clone(); + let fired: Vec<&str> = categories + .as_object() + .map(|m| { + m.iter() + .filter(|(_, v)| v.as_bool().unwrap_or(false)) + .map(|(k, _)| k.as_str()) + .collect() + }) + .unwrap_or_default(); + + let reason = if fired.is_empty() { + "Message flagged by content moderation.".to_owned() + } else { + format!( + "Message flagged by content moderation: {}.", + fired.join(", ") + ) + }; + + tracing::warn!( + "Moderation API flagged message — categories: {:?}, excerpt: {}", + fired, + message.chars().take(120).collect::() + ); + + Some(( + StatusCode::UNPROCESSABLE_ENTITY, + serde_json::json!({ "error": reason }), + categories, + )) +} + async fn classify_intent(message: &str, ollama_base: &str, model: &str, user_id: Option<&str>) -> (String, f32) { let prompt = format!( "Classify this user message into one intent category. Categories: \ @@ -435,8 +550,16 @@ async fn ai_chat_message( State(state): State, Json(body): Json, ) -> impl IntoResponse { - // ── Phase 1: LLM Guard — reject prompt-injection / abuse before any work ── + // ── Guard 1: Sync keyword / heuristic filter ────────────────────────────── if let Some((status, payload)) = llm_guard_check(&body.message) { + let reason = payload["error"].as_str().unwrap_or("rejected").to_owned(); + spawn_guard_log(state.pool.clone(), None, &body.message, "keyword", &reason, None); + return (status, Json(payload)).into_response(); + } + // ── Guard 2: Async semantic moderation (hate, sexual, violence, etc.) ───── + if let Some((status, payload, cats)) = llm_moderation_check(&body.message).await { + let reason = payload["error"].as_str().unwrap_or("flagged").to_owned(); + spawn_guard_log(state.pool.clone(), None, &body.message, "moderation_api", &reason, Some(cats)); return (status, Json(payload)).into_response(); } @@ -2314,7 +2437,16 @@ async fn ai_chat_ask( auth: AuthUser, Json(body): Json, ) -> impl IntoResponse { + // ── Guard 1: Sync keyword / heuristic filter ────────────────────────────── if let Some((status, payload)) = llm_guard_check(&body.message) { + let reason = payload["error"].as_str().unwrap_or("rejected").to_owned(); + spawn_guard_log(state.pool.clone(), Some(auth.user_id), &body.message, "keyword", &reason, None); + return (status, Json(payload)).into_response(); + } + // ── Guard 2: Async semantic moderation (hate, sexual, violence, etc.) ───── + if let Some((status, payload, cats)) = llm_moderation_check(&body.message).await { + let reason = payload["error"].as_str().unwrap_or("flagged").to_owned(); + spawn_guard_log(state.pool.clone(), Some(auth.user_id), &body.message, "moderation_api", &reason, Some(cats)); return (status, Json(payload)).into_response(); } @@ -3818,8 +3950,21 @@ pub mod phase3 { req: StreamRequest, ) -> impl Stream> { async_stream::stream! { - // Guard: LLM guard for injection + // ── Guard 1: Sync keyword / heuristic filter ───────────────────── if let Some((status, payload)) = super::llm_guard_check(&req.message) { + let reason = payload["error"].as_str().unwrap_or("rejected").to_owned(); + super::spawn_guard_log(state.pool.clone(), Some(auth_user_id), &req.message, "keyword", &reason, None); + let event = Event::default() + .event("error") + .id("0") + .data(format!("{}: {}", status.as_u16(), payload)); + yield Ok(event); + return; + } + // ── Guard 2: Async semantic moderation ─────────────────────────── + if let Some((status, payload, cats)) = super::llm_moderation_check(&req.message).await { + let reason = payload["error"].as_str().unwrap_or("flagged").to_owned(); + super::spawn_guard_log(state.pool.clone(), Some(auth_user_id), &req.message, "moderation_api", &reason, Some(cats)); let event = Event::default() .event("error") .id("0") diff --git a/crates/db/migrations_new/20260815000001_create_ai_guard_violations.down.sql b/crates/db/migrations_new/20260815000001_create_ai_guard_violations.down.sql new file mode 100644 index 0000000..6320de0 --- /dev/null +++ b/crates/db/migrations_new/20260815000001_create_ai_guard_violations.down.sql @@ -0,0 +1 @@ +DROP TABLE IF EXISTS ai_guard_violations; diff --git a/crates/db/migrations_new/20260815000001_create_ai_guard_violations.up.sql b/crates/db/migrations_new/20260815000001_create_ai_guard_violations.up.sql new file mode 100644 index 0000000..d5ffba7 --- /dev/null +++ b/crates/db/migrations_new/20260815000001_create_ai_guard_violations.up.sql @@ -0,0 +1,16 @@ +-- AI Guard Violations — records every message blocked by the content guard. +-- Migration: 20260815000001 + +CREATE TABLE IF NOT EXISTS ai_guard_violations ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + user_id UUID REFERENCES users(id) ON DELETE SET NULL, + message_excerpt TEXT NOT NULL, -- first 200 chars, never the full text + guard_type VARCHAR(40) NOT NULL, -- 'keyword' | 'length' | 'flood' | 'moderation_api' + reason TEXT NOT NULL, -- human-readable rejection reason + categories JSONB, -- OpenAI moderation categories (moderation_api only) + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); + +CREATE INDEX IF NOT EXISTS idx_ai_guard_violations_user_id ON ai_guard_violations(user_id); +CREATE INDEX IF NOT EXISTS idx_ai_guard_violations_created_at ON ai_guard_violations(created_at DESC); +CREATE INDEX IF NOT EXISTS idx_ai_guard_violations_guard_type ON ai_guard_violations(guard_type); diff --git a/job_seekers.pid b/job_seekers.pid index a1d3082..19f86b8 100644 --- a/job_seekers.pid +++ b/job_seekers.pid @@ -1 +1 @@ -9693 +15122