feat(ai-guard): add semantic moderation layer + guard violations DB logging

- 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 <noreply@anthropic.com>

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<AppState> {
         .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<uuid::Uuid>,
+    message_excerpt: String,
+    guard_type: String,
+    reason: String,
+    categories: Option<serde_json::Value>,
+    created_at: chrono::DateTime<chrono::Utc>,
+}
+
+#[derive(Debug, sqlx::FromRow)]
+struct GuardEventDb {
+    id: uuid::Uuid,
+    user_id: Option<uuid::Uuid>,
+    message_excerpt: String,
+    guard_type: String,
+    reason: String,
+    categories: Option<serde_json::Value>,
+    created_at: chrono::DateTime<chrono::Utc>,
+}
+
+#[derive(Debug, serde::Deserialize)]
+struct GuardEventsQuery {
+    #[serde(default = "default_limit")]
+    limit: i64,
+    #[serde(default)]
+    offset: i64,
+    guard_type: Option<String>,
+}
+
+async fn guard_events(
+    auth: AuthUser,
+    State(state): State<AppState>,
+    Query(q): Query<GuardEventsQuery>,
+) -> 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<GuardEventDb> = 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<GuardEventRow> = 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(

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<Uuid>,
+    message: &str,
+    guard_type: &'static str,
+    reason: &str,
+    categories: Option<serde_json::Value>,
+) {
+    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::<String>()
+    );
+
+    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<AppState>,
     Json(body): Json<OllamaChatRequest>,
 ) -> 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<AskAshRequest>,
 ) -> 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<Item = Result<Event, Infallible>> {
         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")

crates/db/migrations_new/20260815000001_create_ai_guard_violations.down.sql

@@ -0,0 +1 @@
+DROP TABLE IF EXISTS ai_guard_violations;

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);

job_seekers.pid

@@ -1 +1 @@
-9693
+15122