From d00acaa57ed3e09147a3b137c0c634335e972f05 Mon Sep 17 00:00:00 2001 From: Eureka Date: Wed, 11 Nov 2026 19:19:51 +0000 Subject: [PATCH] sources: SourceIndex chat writer reads and delete_chat, Qdrant and fake impls Co-Authored-By: Claude Sonnet 5.5 --- crates/huginn-daemon/src/index/fakes.rs | 59 +++++++++++++++- crates/huginn-daemon/src/index/qdrant.rs | 89 +++++++++++++++++++++++- crates/huginn-daemon/src/sources/mod.rs | 18 +++++ 3 files changed, 164 insertions(+), 2 deletions(-) diff --git a/crates/huginn-daemon/src/index/fakes.rs b/crates/huginn-daemon/src/index/fakes.rs index 7a905fb..8af75dc 100644 --- a/crates/huginn-daemon/src/index/fakes.rs +++ b/crates/huginn-daemon/src/index/fakes.rs @@ -8,7 +8,7 @@ use std::time::Duration; use anyhow::{Result, bail, ensure}; -use crate::sources::{ChatFilter, ChatPayload, ChatPoint, ChatRow, DescribedSources, SourceIndex, SourcePoint, SourceRow, SourcesMeta}; +use crate::sources::{ChatFilter, ChatPayload, ChatPoint, ChatRow, DescribedSources, SourceIndex, SourcePoint, SourceRow, SourcesMeta, StoredChat}; use super::{CollectionMeta, Described, Embedder, Hit, ModelIdentity, Point, VectorIndex, point_id}; @@ -264,6 +264,12 @@ pub(crate) struct SourcesState { pub(crate) deletes: Vec<(String, Option>)>, /// A delete naming this repo refuses. pub(crate) fail_deletes_of: Option, + /// Every chat upsert refuses while this is set. + pub(crate) fail_chat_upserts: bool, + /// Every chat delete refuses while this is set. + pub(crate) fail_chat_deletes: bool, + /// Every chat delete that went through: the room and the ids (`None` is all of the room). + pub(crate) chat_deletes: Vec<(String, Option>)>, /// The repo filter and limit of every search, in order. pub(crate) searches: Vec<(Option, u32)>, } @@ -375,6 +381,7 @@ impl SourceIndex for FakeSources { fn upsert_chat(&mut self, points: &[ChatPoint]) -> Result<()> { let mut state = self.state.lock().unwrap(); + ensure!(!state.fail_chat_upserts, "the vector store refused the chat upsert"); for point in points { ensure!(!point.payload.messages.is_empty(), "a chat chunk carries no messages"); state.chat.insert(point.id.clone(), point.clone()); @@ -401,6 +408,44 @@ impl SourceIndex for FakeSources { kept.sort_by_key(|payload| payload.messages.iter().map(|message| message.ts_ms).min()); Ok(kept.into_iter().take(limit as usize).cloned().collect()) } + + fn chat_tail(&mut self, room_id: &str) -> Result> { + let state = self.state.lock().unwrap(); + Ok(state.chat.iter().filter(|(_, point)| point.payload.room_id == room_id).max_by_key(|(id, point)| (span(&point.payload).1, (*id).clone())).map(stored)) + } + + fn chat_chunks_holding(&mut self, room_id: &str, event_ids: &[String]) -> Result> { + let state = self.state.lock().unwrap(); + Ok(state + .chat + .iter() + .filter(|(_, point)| point.payload.room_id == room_id && point.payload.messages.iter().any(|message| event_ids.contains(&message.event_id))) + .map(stored) + .collect()) + } + + fn chat_chunks_overlapping(&mut self, room_id: &str, from_ms: u64, to_ms: u64) -> Result> { + let state = self.state.lock().unwrap(); + Ok(state.chat.iter().filter(|(_, point)| point.payload.room_id == room_id && span(&point.payload).1 >= from_ms && span(&point.payload).0 < to_ms).map(stored).collect()) + } + + fn delete_chat(&mut self, room_id: &str, ids: Option<&[String]>) -> Result<()> { + let mut state = self.state.lock().unwrap(); + ensure!(!state.fail_chat_deletes, "the vector store refused the chat delete"); + state.chat_deletes.push((room_id.to_owned(), ids.map(<[String]>::to_vec))); + state.chat.retain(|id, point| !(point.payload.room_id == room_id && ids.is_none_or(|ids| ids.contains(id)))); + Ok(()) + } +} + +fn stored((id, point): (&String, &ChatPoint)) -> StoredChat { + StoredChat { id: id.clone(), payload: point.payload.clone() } +} + +/// (start, end) of the times a chunk's messages make. +fn span(payload: &ChatPayload) -> (u64, u64) { + let times = || payload.messages.iter().map(|message| message.ts_ms); + (times().min().unwrap_or(0), times().max().unwrap_or(0)) } /// A chat chunk under a filter, by the span its messages make. @@ -440,4 +485,16 @@ impl SourceIndex for FakeIndex { fn chat_scroll(&mut self, filter: &ChatFilter, limit: u32, within: Duration) -> Result> { self.sources.chat_scroll(filter, limit, within) } + fn chat_tail(&mut self, room_id: &str) -> Result> { + self.sources.chat_tail(room_id) + } + fn chat_chunks_holding(&mut self, room_id: &str, event_ids: &[String]) -> Result> { + self.sources.chat_chunks_holding(room_id, event_ids) + } + fn chat_chunks_overlapping(&mut self, room_id: &str, from_ms: u64, to_ms: u64) -> Result> { + self.sources.chat_chunks_overlapping(room_id, from_ms, to_ms) + } + fn delete_chat(&mut self, room_id: &str, ids: Option<&[String]>) -> Result<()> { + self.sources.delete_chat(room_id, ids) + } } diff --git a/crates/huginn-daemon/src/index/qdrant.rs b/crates/huginn-daemon/src/index/qdrant.rs index 66d08e4..13ae48e 100644 --- a/crates/huginn-daemon/src/index/qdrant.rs +++ b/crates/huginn-daemon/src/index/qdrant.rs @@ -11,7 +11,7 @@ use huginn_mind::ChatMessage; use crate::sources::{ COLLECTION, CORPUS_CHAT, CORPUS_FORGE, ChatFilter, ChatPayload, ChatPoint, ChatRow, DescribedSources, FRONT_PAGE_FILES, SourceIndex, SourcePayload, SourcePoint, SourceRow, - SourcesMeta, + SourcesMeta, StoredChat, }; use super::{CollectionMeta, INDEX_CALL_TIMEOUT, Described, Hit, Http, Method, Point, VectorIndex}; @@ -455,6 +455,70 @@ impl SourceIndex for QdrantIndex { let found = answer["result"]["points"].as_array().context("Qdrant's scroll has no points")?; found.iter().map(|point| chat_payload_of(&point["payload"])).collect() } + + fn chat_tail(&mut self, room_id: &str) -> Result> { + let order = json!({ "key": "ts_end_ms", "direction": "desc" }); + Ok(self.scroll_chat(chat_room_filter(room_id, vec![]), Some((order, 1)))?.into_iter().next()) + } + + fn chat_chunks_holding(&mut self, room_id: &str, event_ids: &[String]) -> Result> { + if event_ids.is_empty() { + return Ok(Vec::new()); + } + self.scroll_chat(chat_room_filter(room_id, vec![json!({ "key": "event_ids", "match": { "any": event_ids } })]), None) + } + + fn chat_chunks_overlapping(&mut self, room_id: &str, from_ms: u64, to_ms: u64) -> Result> { + let span = vec![json!({ "key": "ts_end_ms", "range": { "gte": from_ms } }), json!({ "key": "ts_start_ms", "range": { "lt": to_ms } })]; + self.scroll_chat(chat_room_filter(room_id, span), None) + } + + fn delete_chat(&mut self, room_id: &str, ids: Option<&[String]>) -> Result<()> { + let collection = self.sources.clone(); + let body = json!({ "filter": chat_delete_filter(room_id, ids) }); + self.ask(Method::Post, &format!("/collections/{collection}/points/delete?wait=true"), Some(&body), &[200])?; + Ok(()) + } +} + +impl QdrantIndex { + /// Every chat chunk `filter` keeps, with its point id, paged; or, given an + /// order and a limit, the first of them in that order. + fn scroll_chat(&mut self, filter: Value, ordered: Option<(Value, u64)>) -> Result> { + let collection = self.sources.clone(); + let mut out = Vec::new(); + let mut offset = Value::Null; + loop { + let mut body = json!({ "filter": filter, "limit": SCROLL_PAGE, "with_payload": true, "with_vector": false }); + if let Some((order, limit)) = &ordered { + body["order_by"] = order.clone(); + body["limit"] = json!(limit); + } else if !offset.is_null() { + body["offset"] = offset; + } + let (_, answer) = self.ask(Method::Post, &format!("/collections/{collection}/points/scroll"), Some(&body), &[200])?; + for point in answer["result"]["points"].as_array().context("Qdrant's scroll has no points")? { + out.push(StoredChat { id: point["id"].as_str().context("a point id is not a string")?.to_owned(), payload: chat_payload_of(&point["payload"])? }); + } + offset = answer["result"]["next_page_offset"].clone(); + if ordered.is_some() || offset.is_null() { + return Ok(out); + } + } + } +} + +/// What a chat delete removes: the room's chat points, all of them or those with `ids`. The corpus +/// and room conditions hold either way, so a forge point or another room's point cannot match. +fn chat_delete_filter(room_id: &str, ids: Option<&[String]>) -> Value { + chat_room_filter(room_id, ids.map(|ids| json!({ "has_id": ids })).into_iter().collect()) +} + +/// The chat points of one room, narrowed by `more` conditions. +fn chat_room_filter(room_id: &str, more: Vec) -> Value { + let mut must = vec![json!({ "key": "corpus", "match": { "value": CORPUS_CHAT } }), json!({ "key": "room_id", "match": { "value": room_id } })]; + must.extend(more); + json!({ "must": must }) } #[cfg(test)] @@ -517,6 +581,29 @@ mod source_filter_tests { assert!(chat_payload_of(&forge).is_err()); } + /// The writer-side filters name the chat corpus and the room, so a forge point never matches. + #[test] + fn the_writer_filters_are_always_chat_corpus_and_the_room() { + let filter = chat_room_filter("!r:x", vec![json!({ "key": "event_ids", "match": { "any": ["$a"] } })]); + assert_eq!( + filter, + json!({ "must": [ + { "key": "corpus", "match": { "value": "chat" } }, + { "key": "room_id", "match": { "value": "!r:x" } }, + { "key": "event_ids", "match": { "any": ["$a"] } }, + ] }) + ); + } + + #[test] + fn a_chat_delete_names_the_corpus_and_room_with_or_without_ids() { + let all = chat_delete_filter("!r:x", None); + assert_eq!(all["must"].as_array().unwrap().len(), 2); + let some = chat_delete_filter("!r:x", Some(&["p1".to_owned()])); + assert_eq!(some["must"][2], json!({ "has_id": ["p1"] })); + assert_eq!(some["must"][0], json!({ "key": "corpus", "match": { "value": "chat" } })); + } + /// The chat filter is the chat corpus, the rooms as an any-of `should` split /// by id and alias, and the two ranges that keep an overlapping span. #[test] diff --git a/crates/huginn-daemon/src/sources/mod.rs b/crates/huginn-daemon/src/sources/mod.rs index 471bce2..3b237e3 100644 --- a/crates/huginn-daemon/src/sources/mod.rs +++ b/crates/huginn-daemon/src/sources/mod.rs @@ -179,6 +179,14 @@ impl ChatFilter { } } +/// A chat chunk as the index holds it, with its point id: what the ingest job +/// reads back to rewrite a chunk. +#[derive(Clone, Debug, PartialEq)] +pub struct StoredChat { + pub id: String, + pub payload: ChatPayload, +} + /// One chunk a vector search returned and how near it scored. #[derive(Clone, Debug, PartialEq)] pub struct SourceRow { @@ -216,6 +224,16 @@ pub trait SourceIndex { /// Up to `limit` chat chunks `filter` keeps, oldest first by span start. /// Reads only `corpus=chat` points; the daemon's. fn chat_scroll(&mut self, filter: &ChatFilter, limit: u32, within: Duration) -> Result>; + /// The room's chunk with the greatest `ts_end_ms`, if it has any. The ingest job's alone. + fn chat_tail(&mut self, room_id: &str) -> Result>; + /// The room's chunks holding any of `event_ids`. The ingest job's alone. + fn chat_chunks_holding(&mut self, room_id: &str, event_ids: &[String]) -> Result>; + /// The room's chunks whose span reaches into [`from_ms`, `to_ms`): `ts_end_ms >= from_ms` + /// and `ts_start_ms < to_ms`. The ingest job's alone. + fn chat_chunks_overlapping(&mut self, room_id: &str, from_ms: u64, to_ms: u64) -> Result>; + /// Deletes a room's chat chunks: those with the point ids `ids`, or all of them when `None`. + /// Acknowledged before it returns. Never touches a forge point. The ingest job's alone. + fn delete_chat(&mut self, room_id: &str, ids: Option<&[String]>) -> Result<()>; } /// A repo the forge lists as indexable.