reddb_server/runtime/ai/
moderation.rs

1//! Local content-moderation backend (#1274, PRD #1267, ADR 0057).
2//!
3//! Moderation is the third AI modality on the write/CDC lane (after
4//! embeddings #1272 and vision #1275). A collection that declares a
5//! `MODERATE (...)` policy names text/image-reference fields that are
6//! screened by a moderation provider. Unlike embed/vision, moderation has
7//! a **synchronous pre-commit gate**: when `sync = true`, the write path
8//! screens the declared fields *before the durable commit* and refuses the
9//! write on a reject (ADR 0057). Provider-down behaviour and re-moderation
10//! of quarantined rows ride the existing CDC enrichment lane.
11//!
12//! Mirroring [`super::vision`] and [`super::local_embedding`], the engine
13//! is a swappable, process-global [`LocalModerationBackend`]. There is no
14//! built-in default — a real engine or a mock is installed via
15//! [`install_local_moderation_backend`]. A backend may also report itself
16//! *down* (returning [`ModerationOutcome::ProviderDown`]) so tests can
17//! exercise the fail-open-quarantine and fail-closed paths deterministically.
18
19use std::sync::{Arc, OnceLock, RwLock};
20
21use crate::{RedDBError, RedDBResult};
22
23pub use reddb_engine::storage::visibility::{
24    entity_moderation_hidden, MODERATION_STATUS_FIELD, MODERATION_STATUS_PENDING,
25    MODERATION_STATUS_REJECTED,
26};
27
28/// Verdict for one moderation pass over a row's declared fields.
29#[derive(Debug, Clone, PartialEq, Eq)]
30pub enum ModerationOutcome {
31    /// Content passed — the write/row is allowed to be visible.
32    Allow,
33    /// Content failed moderation — the write is rejected. `categories`
34    /// carries the flagged reasons for audit/appeal.
35    Reject { categories: Vec<String> },
36    /// The provider could not be reached. The caller applies the policy's
37    /// degraded-mode behaviour (fail-open + quarantine, or fail-closed).
38    ProviderDown { reason: String },
39}
40
41impl ModerationOutcome {
42    /// True when the content was rejected by moderation.
43    pub fn is_reject(&self) -> bool {
44        matches!(self, Self::Reject { .. })
45    }
46
47    /// True when the provider was unreachable.
48    pub fn is_provider_down(&self) -> bool {
49        matches!(self, Self::ProviderDown { .. })
50    }
51}
52
53/// A materialised moderation request handed to a backend. The declared
54/// text fields are concatenated by the caller into `text`; an empty
55/// `text` means there was nothing to screen.
56#[derive(Debug, Clone)]
57pub struct ModerationRequest {
58    /// Model name as written in the collection's MODERATE policy.
59    pub model: String,
60    /// The concatenated text of the row's declared moderated fields.
61    pub text: String,
62}
63
64/// Backend abstraction so the gate/enrichment lanes do not depend on a
65/// specific moderation engine. Tests install a mock; production installs a
66/// real engine via [`install_local_moderation_backend`].
67pub trait LocalModerationBackend: Send + Sync {
68    fn moderate(&self, request: &ModerationRequest) -> RedDBResult<ModerationOutcome>;
69}
70
71const LOCAL_MODERATION_DISABLED_MESSAGE: &str =
72    "local moderation requires a backend installed via \
73     runtime::ai::moderation::install_local_moderation_backend. Alternatively, \
74     declare a moderation-capable remote provider in the collection's MODERATE \
75     policy.";
76
77type BackendSlot = Arc<dyn LocalModerationBackend>;
78
79fn backend_slot() -> &'static RwLock<Option<BackendSlot>> {
80    static SLOT: OnceLock<RwLock<Option<BackendSlot>>> = OnceLock::new();
81    SLOT.get_or_init(|| RwLock::new(None))
82}
83
84/// Install (or replace) the process-global local moderation backend.
85///
86/// Production servers call this once at boot with their real engine. Tests
87/// use it to swap in a mock moderation provider. Safe to call from any
88/// thread; the most recent install wins.
89pub fn install_local_moderation_backend(backend: Arc<dyn LocalModerationBackend>) {
90    let mut guard = backend_slot()
91        .write()
92        .expect("moderation backend slot poisoned");
93    *guard = Some(backend);
94}
95
96/// Test-only: clear the installed backend so a subsequent call exercises
97/// the feature-disabled path again.
98#[doc(hidden)]
99pub fn clear_local_moderation_backend_for_tests() {
100    let mut guard = backend_slot()
101        .write()
102        .expect("moderation backend slot poisoned");
103    *guard = None;
104}
105
106fn current_backend() -> Option<BackendSlot> {
107    backend_slot()
108        .read()
109        .expect("moderation backend slot poisoned")
110        .as_ref()
111        .map(Arc::clone)
112}
113
114/// Resolve and run a local moderation request end-to-end. Errors with a
115/// clear message when no backend is installed; a *down provider* is NOT an
116/// error — the backend signals it via [`ModerationOutcome::ProviderDown`]
117/// so the caller can apply the policy's degraded-mode behaviour.
118pub fn moderate_local(model: &str, text: String) -> RedDBResult<ModerationOutcome> {
119    let backend = current_backend().ok_or_else(|| {
120        RedDBError::FeatureNotEnabled(LOCAL_MODERATION_DISABLED_MESSAGE.to_string())
121    })?;
122    backend.moderate(&ModerationRequest {
123        model: model.to_string(),
124        text,
125    })
126}
127
128#[cfg(test)]
129mod tests {
130    use super::*;
131
132    struct AllowAll;
133    impl LocalModerationBackend for AllowAll {
134        fn moderate(&self, _request: &ModerationRequest) -> RedDBResult<ModerationOutcome> {
135            Ok(ModerationOutcome::Allow)
136        }
137    }
138
139    #[test]
140    fn missing_backend_is_feature_disabled_error() {
141        clear_local_moderation_backend_for_tests();
142        let err = moderate_local("m", "hello".to_string()).expect_err("no backend");
143        assert!(matches!(err, RedDBError::FeatureNotEnabled(_)));
144    }
145
146    #[test]
147    fn installed_backend_drives_outcome() {
148        install_local_moderation_backend(Arc::new(AllowAll));
149        let outcome = moderate_local("m", "hello".to_string()).expect("allowed");
150        assert_eq!(outcome, ModerationOutcome::Allow);
151        clear_local_moderation_backend_for_tests();
152    }
153
154    #[test]
155    fn outcome_predicates() {
156        assert!(ModerationOutcome::Reject {
157            categories: vec!["hate".to_string()],
158        }
159        .is_reject());
160        assert!(ModerationOutcome::ProviderDown {
161            reason: "timeout".to_string(),
162        }
163        .is_provider_down());
164        assert!(!ModerationOutcome::Allow.is_reject());
165    }
166}
reddb_server/runtime/ai/moderation.rs

reddb_server/runtime/ai/
moderation.rs
