reddb_server/runtime/ai/
vision.rs

1//! Local computer-vision backend + image fetch (#1275, PRD #1267).
2//!
3//! Vision is the second async AI modality wired over the CDC enrichment
4//! lane (the first being embeddings, #1272). A collection that declares a
5//! `VISION (...)` policy names an *image-reference field* whose value is a
6//! URL/URI. After commit, the CDC enrichment consumer fetches the
7//! referenced image and runs the policy's vision provider, producing:
8//!
9//!   * a structured **component-detections** array
10//!     (`[{label, confidence, bbox:[x,y,w,h]}]`) written to a derived
11//!     field that RQL can filter, and
12//!   * an optional **image-embedding** vector reusing the existing vector
13//!     pipeline for image similarity search.
14//!
15//! Mirroring [`super::local_embedding`], the engine is a swappable,
16//! process-global [`LocalVisionBackend`]. The default
17//! [`DeterministicFakeVisionBackend`] derives stable detections from
18//! `SHA-256(model || image-bytes)` so the end-to-end contract (fetch →
19//! analyze → attach) can be exercised without a real model. Tests install
20//! their own mock provider via [`install_local_vision_backend`]; a real
21//! candle/onnx engine slots in the same way at server boot.
22
23use std::sync::{Arc, OnceLock, RwLock};
24use std::time::Duration;
25
26use crate::crypto::sha256::Sha256;
27use crate::{RedDBError, RedDBResult};
28
29/// One structured component detection: a labelled, scored bounding box.
30/// `bbox` is `[x, y, w, h]` in image-relative units, matching the
31/// canonical output shape recorded in the issue.
32#[derive(Debug, Clone, PartialEq)]
33pub struct VisionDetection {
34    pub label: String,
35    pub confidence: f32,
36    pub bbox: [f32; 4],
37}
38
39/// Output of a single vision analysis pass.
40#[derive(Debug, Clone, Default, PartialEq)]
41pub struct VisionResult {
42    /// Structured component detections (present when detections were
43    /// requested by the policy).
44    pub detections: Vec<VisionDetection>,
45    /// Image embedding (present only when the policy requested an
46    /// image-embedding output kind).
47    pub embedding: Option<Vec<f32>>,
48}
49
50/// A materialised vision request handed to a backend.
51#[derive(Debug, Clone)]
52pub struct VisionRequest {
53    /// Model name as written in the collection's VISION policy.
54    pub model: String,
55    /// Fetched image bytes (already resolved from the row's reference).
56    pub image_bytes: Vec<u8>,
57    /// Whether the policy asked for structured detections.
58    pub want_detections: bool,
59    /// Whether the policy asked for an image-embedding output.
60    pub want_embedding: bool,
61}
62
63/// Backend abstraction so the enrichment lane does not depend on a
64/// specific vision engine. Tests install a mock; production installs a
65/// real engine via [`install_local_vision_backend`].
66pub trait LocalVisionBackend: Send + Sync {
67    fn analyze(&self, request: &VisionRequest) -> RedDBResult<VisionResult>;
68}
69
70const LOCAL_VISION_DISABLED_MESSAGE: &str =
71    "local vision requires the `local-models` feature flag at engine build time, \
72     or a backend installed via \
73     runtime::ai::vision::install_local_vision_backend. Alternatively, declare a \
74     vision-capable remote provider in the collection's VISION policy.";
75
76/// Width (in f32 lanes) of the deterministic fake image embedding.
77const FAKE_EMBEDDING_DIM: usize = 16;
78
79/// Deterministic, dependency-free vision backend used to prove the
80/// fetch → analyze → attach contract end-to-end. Output is a pure
81/// function of `(model, image-bytes)` — no I/O, no clocks, no RNGs — so
82/// tests get byte-identical results across runs.
83#[derive(Debug, Default, Clone, Copy)]
84pub struct DeterministicFakeVisionBackend;
85
86impl LocalVisionBackend for DeterministicFakeVisionBackend {
87    fn analyze(&self, request: &VisionRequest) -> RedDBResult<VisionResult> {
88        let digest = {
89            let mut hasher = Sha256::new();
90            hasher.update(request.model.as_bytes());
91            hasher.update(&[0u8]);
92            hasher.update(&request.image_bytes);
93            hasher.finalize()
94        };
95
96        let detections = if request.want_detections {
97            // One detection per non-overlapping label, chosen
98            // deterministically from a small fixed vocabulary by the
99            // digest. Two distinct, stable detections so containment
100            // filters have something to match.
101            const VOCAB: [&str; 4] = ["person", "car", "dog", "bicycle"];
102            let pick = |byte: u8| VOCAB[(byte as usize) % VOCAB.len()].to_string();
103            let conf = |byte: u8| (byte as f32) / 255.0;
104            let coord = |byte: u8| (byte as f32) / 255.0;
105            vec![
106                VisionDetection {
107                    label: pick(digest[0]),
108                    confidence: conf(digest[1]),
109                    bbox: [
110                        coord(digest[2]),
111                        coord(digest[3]),
112                        coord(digest[4]),
113                        coord(digest[5]),
114                    ],
115                },
116                VisionDetection {
117                    label: pick(digest[6]),
118                    confidence: conf(digest[7]),
119                    bbox: [
120                        coord(digest[8]),
121                        coord(digest[9]),
122                        coord(digest[10]),
123                        coord(digest[11]),
124                    ],
125                },
126            ]
127        } else {
128            Vec::new()
129        };
130
131        let embedding = if request.want_embedding {
132            let mut out = Vec::with_capacity(FAKE_EMBEDDING_DIM);
133            let mut counter: u32 = 0;
134            while out.len() < FAKE_EMBEDDING_DIM {
135                let mut hasher = Sha256::new();
136                hasher.update(&digest);
137                hasher.update(&counter.to_le_bytes());
138                let chunk_digest = hasher.finalize();
139                for chunk in chunk_digest.chunks(4) {
140                    if out.len() >= FAKE_EMBEDDING_DIM {
141                        break;
142                    }
143                    let mut bytes = [0u8; 4];
144                    bytes.copy_from_slice(chunk);
145                    let raw = u32::from_le_bytes(bytes) as f32 / u32::MAX as f32;
146                    out.push(raw * 2.0 - 1.0);
147                }
148                counter = counter.wrapping_add(1);
149            }
150            Some(out)
151        } else {
152            None
153        };
154
155        Ok(VisionResult {
156            detections,
157            embedding,
158        })
159    }
160}
161
162type BackendSlot = Arc<dyn LocalVisionBackend>;
163
164fn backend_slot() -> &'static RwLock<Option<BackendSlot>> {
165    static SLOT: OnceLock<RwLock<Option<BackendSlot>>> = OnceLock::new();
166    SLOT.get_or_init(|| RwLock::new(None))
167}
168
169/// Install (or replace) the process-global local vision backend.
170///
171/// Production servers built with `--features local-models` call this once
172/// at boot with their real engine. Tests use it to swap in a mock vision
173/// provider. Safe to call from any thread; the most recent install wins.
174pub fn install_local_vision_backend(backend: Arc<dyn LocalVisionBackend>) {
175    let mut guard = backend_slot()
176        .write()
177        .expect("vision backend slot poisoned");
178    *guard = Some(backend);
179}
180
181/// Test-only: clear the installed backend so a subsequent call exercises
182/// the feature-disabled path again.
183#[doc(hidden)]
184pub fn clear_local_vision_backend_for_tests() {
185    let mut guard = backend_slot()
186        .write()
187        .expect("vision backend slot poisoned");
188    *guard = None;
189}
190
191fn current_backend() -> Option<BackendSlot> {
192    backend_slot()
193        .read()
194        .expect("vision backend slot poisoned")
195        .as_ref()
196        .map(Arc::clone)
197}
198
199/// Resolve and run a local vision request end-to-end. Falls back to the
200/// deterministic fake when the `local-models` feature is on but no engine
201/// was installed; errors with a clear message when neither is available.
202pub fn analyze_local(
203    model: &str,
204    image_bytes: Vec<u8>,
205    want_detections: bool,
206    want_embedding: bool,
207) -> RedDBResult<VisionResult> {
208    let backend = match current_backend() {
209        Some(b) => b,
210        None => {
211            if cfg!(feature = "local-models") {
212                let fake: Arc<dyn LocalVisionBackend> = Arc::new(DeterministicFakeVisionBackend);
213                install_local_vision_backend(Arc::clone(&fake));
214                fake
215            } else {
216                return Err(RedDBError::FeatureNotEnabled(
217                    LOCAL_VISION_DISABLED_MESSAGE.to_string(),
218                ));
219            }
220        }
221    };
222
223    backend.analyze(&VisionRequest {
224        model: model.to_string(),
225        image_bytes,
226        want_detections,
227        want_embedding,
228    })
229}
230
231/// Fetch the bytes of an image referenced by a row field.
232///
233/// The reference is a URL/URI (ADR 0057 stores the reference, never the
234/// bytes). Supported schemes:
235///   * `file://<path>` and bare filesystem paths — read from disk;
236///   * `http://` / `https://` — fetched via `ureq` (rustls).
237///
238/// Network/IO failures surface as [`RedDBError`] so the enrichment lane's
239/// retry-with-backoff and dead-letter machinery handles them like any
240/// other provider failure.
241pub fn fetch_image_bytes(reference: &str) -> RedDBResult<Vec<u8>> {
242    let reference = reference.trim();
243    if reference.is_empty() {
244        return Err(RedDBError::Query(
245            "vision image reference is empty".to_string(),
246        ));
247    }
248
249    // The reference is row data: whoever can write the image column
250    // chooses what the server fetches. Local paths are therefore off
251    // unless the operator opts in, and HTTP targets follow the same
252    // egress rule as AI providers (https off loopback, no private or
253    // link-local hosts).
254    if reference.starts_with("http://") || reference.starts_with("https://") {
255        crate::ai::validate_custom_provider_url(reference)?;
256        return fetch_http_image(reference);
257    }
258
259    if !local_files_allowed() {
260        return Err(RedDBError::Query(format!(
261            "vision image reference '{reference}' names a local file; local image \
262             paths are disabled unless {LOCAL_FILES_ENV}=1"
263        )));
264    }
265
266    if let Some(rest) = reference.strip_prefix("file://") {
267        // Tolerate the `file://localhost/path` authority form.
268        let path = rest.strip_prefix("localhost").unwrap_or(rest);
269        return read_local_image(path);
270    }
271
272    // Bare filesystem path.
273    read_local_image(reference)
274}
275
276/// Operator opt-in for `file://` and bare-path image references.
277const LOCAL_FILES_ENV: &str = "REDDB_AI_VISION_ALLOW_LOCAL_FILES";
278
279/// Largest image the enrichment lane will read, in bytes; override with
280/// `REDDB_AI_VISION_MAX_IMAGE_BYTES`.
281const DEFAULT_MAX_IMAGE_BYTES: u64 = 32 * 1024 * 1024;
282
283fn local_files_allowed() -> bool {
284    std::env::var(LOCAL_FILES_ENV)
285        .map(|value| matches!(value.trim(), "1" | "true" | "yes"))
286        .unwrap_or(false)
287}
288
289fn max_image_bytes() -> u64 {
290    std::env::var("REDDB_AI_VISION_MAX_IMAGE_BYTES")
291        .ok()
292        .and_then(|raw| raw.trim().parse::<u64>().ok())
293        .filter(|value| *value > 0)
294        .unwrap_or(DEFAULT_MAX_IMAGE_BYTES)
295}
296
297fn read_local_image(path: &str) -> RedDBResult<Vec<u8>> {
298    use std::io::Read;
299    let limit = max_image_bytes();
300    let file = std::fs::File::open(path)
301        .map_err(|err| RedDBError::Internal(format!("read image '{path}': {err}")))?;
302    let mut bytes = Vec::new();
303    file.take(limit + 1)
304        .read_to_end(&mut bytes)
305        .map_err(|err| RedDBError::Internal(format!("read image '{path}': {err}")))?;
306    if bytes.len() as u64 > limit {
307        return Err(RedDBError::Query(format!(
308            "image '{path}' exceeds the {limit}-byte vision limit"
309        )));
310    }
311    Ok(bytes)
312}
313
314fn fetch_http_image(url: &str) -> RedDBResult<Vec<u8>> {
315    let agent: ureq::Agent = ureq::Agent::config_builder()
316        .timeout_connect(Some(Duration::from_secs(15)))
317        .timeout_send_request(Some(Duration::from_secs(30)))
318        .timeout_recv_response(Some(Duration::from_secs(30)))
319        .timeout_recv_body(Some(Duration::from_secs(120)))
320        .build()
321        .into();
322
323    let mut resp = agent
324        .get(url)
325        .call()
326        .map_err(|err| RedDBError::Internal(format!("HTTP GET image '{url}': {err}")))?;
327
328    let status = resp.status().as_u16();
329    if status != 200 {
330        return Err(RedDBError::Internal(format!(
331            "HTTP GET image '{url}': status {status}"
332        )));
333    }
334
335    let limit = max_image_bytes();
336    let bytes = resp
337        .body_mut()
338        .with_config()
339        .limit(limit + 1)
340        .read_to_vec()
341        .map_err(|err| RedDBError::Internal(format!("read image body from '{url}': {err}")))?;
342    if bytes.len() as u64 > limit {
343        return Err(RedDBError::Query(format!(
344            "image at '{url}' exceeds the {limit}-byte vision limit"
345        )));
346    }
347    Ok(bytes)
348}
349
350#[cfg(test)]
351mod tests {
352    use super::*;
353
354    #[test]
355    fn image_references_from_row_data_cannot_name_local_files_or_private_hosts() {
356        if std::env::var_os(LOCAL_FILES_ENV).is_some()
357            || std::env::var_os("REDDB_AI_ALLOW_PRIVATE_PROVIDERS").is_some()
358        {
359            return; // an operator opt-in is active in this environment
360        }
361        for reference in [
362            "file:///etc/hostname",
363            "/etc/hostname",
364            "relative/image.png",
365        ] {
366            let err = fetch_image_bytes(reference).expect_err(reference);
367            assert!(
368                err.to_string().contains(LOCAL_FILES_ENV),
369                "{reference}: {err}"
370            );
371        }
372        for reference in [
373            "http://169.254.169.254/latest/meta-data/",
374            "http://10.0.0.8/internal.png",
375            "http://example.com/plaintext.png",
376        ] {
377            assert!(
378                fetch_image_bytes(reference).is_err(),
379                "{reference} must be refused"
380            );
381        }
382    }
383
384    #[test]
385    fn deterministic_fake_is_pure() {
386        let req = VisionRequest {
387            model: "fake-vision".to_string(),
388            image_bytes: b"some image bytes".to_vec(),
389            want_detections: true,
390            want_embedding: true,
391        };
392        let a = DeterministicFakeVisionBackend.analyze(&req).expect("a");
393        let b = DeterministicFakeVisionBackend.analyze(&req).expect("b");
394        assert_eq!(a, b, "fake vision backend must be pure");
395        assert_eq!(a.detections.len(), 2);
396        assert_eq!(a.embedding.as_ref().map(Vec::len), Some(FAKE_EMBEDDING_DIM));
397    }
398
399    #[test]
400    fn detections_and_embedding_are_gated_by_request() {
401        let base = VisionRequest {
402            model: "m".to_string(),
403            image_bytes: b"img".to_vec(),
404            want_detections: false,
405            want_embedding: false,
406        };
407        let none = DeterministicFakeVisionBackend.analyze(&base).expect("none");
408        assert!(none.detections.is_empty());
409        assert!(none.embedding.is_none());
410
411        let detect_only = DeterministicFakeVisionBackend
412            .analyze(&VisionRequest {
413                want_detections: true,
414                ..base.clone()
415            })
416            .expect("detect");
417        assert!(!detect_only.detections.is_empty());
418        assert!(detect_only.embedding.is_none());
419    }
420
421    #[test]
422    fn local_reads_honour_the_size_limit() {
423        // The policy gate in `fetch_image_bytes` is exercised separately;
424        // this covers the reader an operator opts into.
425        let dir = std::env::temp_dir();
426        let path = dir.join("reddb_vision_fetch_fixture.bin");
427        std::fs::write(&path, b"\x89PNG fixture").expect("write fixture");
428
429        let bytes = read_local_image(path.to_str().expect("utf8 path")).expect("read");
430        assert_eq!(bytes, b"\x89PNG fixture");
431        assert!(read_local_image("/definitely/missing/image.png").is_err());
432
433        let _ = std::fs::remove_file(&path);
434    }
435
436    #[test]
437    fn fetch_rejects_empty_reference() {
438        assert!(fetch_image_bytes("   ").is_err());
439    }
440}
reddb_server/runtime/ai/vision.rs

reddb_server/runtime/ai/
vision.rs
