av_loopdetect/
vector_sink.rs

1//! Off-path vector persistence for semantic-loop observability.
2
3use std::future::Future;
4use std::pin::Pin;
5
6/// Future returned by vector sinks.
7pub type VectorSinkFuture<'a> = Pin<Box<dyn Future<Output = Result<(), String>> + Send + 'a>>;
8
9/// Future returned by vector similarity lookups.
10pub type VectorSearchFuture<'a> = Pin<Box<dyn Future<Output = Result<Option<f32>, String>> + Send + 'a>>;
11
12/// Storage boundary for reasoning-step embeddings.
13pub trait VectorSink: Send + Sync {
14    /// Return the highest cosine similarity among prior vectors for a session.
15    fn nearest_similarity<'a>(&'a self, _session_id: &'a str, _vector: &'a [f32]) -> VectorSearchFuture<'a> {
16        Box::pin(async { Ok(None) })
17    }
18
19    /// Persist one session vector without participating in the hot path.
20    fn record<'a>(&'a self, session_id: &'a str, vector: &'a [f32]) -> VectorSinkFuture<'a>;
21}
22
23/// Sink used when external vector persistence is disabled.
24#[derive(Debug, Default)]
25pub struct NoopVectorSink;
26
27impl VectorSink for NoopVectorSink {
28    fn record<'a>(&'a self, _session_id: &'a str, _vector: &'a [f32]) -> VectorSinkFuture<'a> {
29        Box::pin(async { Ok(()) })
30    }
31}
32
33/// Qdrant HTTP sink for local, edge, or customer-managed deployments.
34#[cfg(feature = "qdrant")]
35pub struct QdrantVectorSink {
36    client: reqwest::Client,
37    base_url: String,
38    collection: String,
39}
40
41#[cfg(feature = "qdrant")]
42impl QdrantVectorSink {
43    /// Create a Qdrant sink. Collection creation remains an operator decision
44    /// because distance metric and replication are deployment policy.
45    pub fn new(base_url: impl Into<String>, collection: impl Into<String>) -> Result<Self, String> {
46        Ok(Self {
47            // No redirect following: a hostile or misconfigured Qdrant host
48            // returning a 3xx would let it pivot the harness into an SSRF
49            // probe against private services on the harness's network.
50            client: reqwest::Client::builder()
51                .connect_timeout(std::time::Duration::from_secs(1))
52                .timeout(std::time::Duration::from_secs(2))
53                .redirect(reqwest::redirect::Policy::none())
54                .build()
55                .map_err(|error| error.to_string())?,
56            base_url: base_url.into().trim_end_matches('/').to_owned(),
57            collection: collection.into(),
58        })
59    }
60
61    /// Create or update the collection with cosine distance and the configured
62    /// embedding width.
63    pub async fn ensure_collection(&self, dimension: usize) -> Result<(), String> {
64        let url = format!("{}/collections/{}", self.base_url, self.collection);
65        self.client
66            .put(url)
67            .json(&serde_json::json!({
68                "vectors": {
69                    "size": dimension,
70                    "distance": "Cosine"
71                }
72            }))
73            .send()
74            .await
75            .map_err(|error| error.to_string())?
76            .error_for_status()
77            .map_err(|error| error.to_string())?;
78        Ok(())
79    }
80}
81
82#[cfg(feature = "qdrant")]
83impl VectorSink for QdrantVectorSink {
84    fn nearest_similarity<'a>(&'a self, session_id: &'a str, vector: &'a [f32]) -> VectorSearchFuture<'a> {
85        Box::pin(async move {
86            let url = format!("{}/collections/{}/points/search", self.base_url, self.collection);
87            let response: serde_json::Value = self
88                .client
89                .post(url)
90                .json(&serde_json::json!({
91                    "vector": vector,
92                    "filter": {
93                        "must": [{
94                            "key": "session_id",
95                            "match": { "value": session_id }
96                        }]
97                    },
98                    "limit": 1,
99                    "with_payload": false,
100                    "with_vector": false
101                }))
102                .send()
103                .await
104                .map_err(|error| error.to_string())?
105                .error_for_status()
106                .map_err(|error| error.to_string())?
107                .json()
108                .await
109                .map_err(|error| error.to_string())?;
110            let Some(score) = response
111                .pointer("/result/0/score")
112                .and_then(serde_json::Value::as_f64)
113            else {
114                return Ok(None);
115            };
116            if !score.is_finite() || !(-1.0..=1.000_001).contains(&score) {
117                return Err(format!("Qdrant returned invalid cosine score {score}"));
118            }
119            #[allow(clippy::cast_possible_truncation)]
120            let score = score.clamp(-1.0, 1.0) as f32;
121            Ok(Some(score))
122        })
123    }
124
125    fn record<'a>(&'a self, session_id: &'a str, vector: &'a [f32]) -> VectorSinkFuture<'a> {
126        Box::pin(async move {
127            let recorded_at = av_core::time::now_ms();
128            let id = av_core::new_event_uid();
129            let url = format!(
130                "{}/collections/{}/points?wait=true",
131                self.base_url, self.collection
132            );
133            self.client
134                .put(url)
135                .json(&serde_json::json!({
136                    "points": [{
137                        "id": id,
138                        "vector": vector,
139                        "payload": {
140                            "session_id": session_id,
141                            "recorded_at": recorded_at,
142                        }
143                    }]
144                }))
145                .send()
146                .await
147                .map_err(|error| error.to_string())?
148                .error_for_status()
149                .map_err(|error| error.to_string())?;
150            Ok(())
151        })
152    }
153}
154
155#[cfg(test)]
156mod tests {
157    #![allow(clippy::unwrap_used)]
158
159    use super::*;
160
161    #[tokio::test]
162    async fn noop_sink_is_total() {
163        NoopVectorSink.record("session", &[0.0, 1.0]).await.unwrap();
164    }
165}
av_loopdetect/vector_sink.rs

av_loopdetect/
vector_sink.rs
