diff --git a/crates/eureka-state/Cargo.toml b/crates/eureka-state/Cargo.toml index b138f56..a84aee6 100644 --- a/crates/eureka-state/Cargo.toml +++ b/crates/eureka-state/Cargo.toml @@ -32,6 +32,7 @@ required-features = ["cli"] [dependencies] anyhow = "1" +chrono = "0.4.44" clap = { version = "4", features = ["derive"], optional = true } cultnet-rs.workspace = true huginn-mind = { path = "../huginn-mind" } @@ -48,6 +49,5 @@ unicode-general-category = "1" # Tests run a real daemon in-process against a temporary mind, and the real # binary against it. [dev-dependencies] -chrono = "0.4.44" huginn-daemon = { path = "../huginn-daemon" } tempfile = "3" diff --git a/crates/eureka-state/src/bin/huginn/chat.rs b/crates/eureka-state/src/bin/huginn/chat.rs new file mode 100644 index 0000000..d4ba4a0 --- /dev/null +++ b/crates/eureka-state/src/bin/huginn/chat.rs @@ -0,0 +1,57 @@ +//! The `huginn chat` verbs' arguments: one instant parser for `--after` and +//! `--before`. Display and the request are in `main` and `render`. + +use chrono::{DateTime, SecondsFormat, Utc}; + +use crate::trouble::Trouble; + +/// An instant as Unix milliseconds: an RFC 3339 time (`2026-11-11T10:00:00Z`) +/// or an age before `now` (`90s`, `15m`, `6h`, `2d`, `1w`). The one parse of +/// both flags. +pub fn instant_ms(flag: &str, text: &str, now: DateTime) -> Result { + let wrong = || Trouble::invalid(format!("{flag} is an RFC 3339 time or an age such as 6h or 2d, not `{text}`")); + let moment = if let Ok(time) = DateTime::parse_from_rfc3339(text) { + time.with_timezone(&Utc) + } else { + let (digits, unit) = text.split_at(text.char_indices().next_back().map_or(0, |(at, _)| at)); + if digits.is_empty() || !digits.bytes().all(|byte| byte.is_ascii_digit()) { + return Err(wrong()); + } + let count: u32 = digits.parse().map_err(|_| wrong())?; + let seconds = match unit { + "s" => 1, + "m" => 60, + "h" => 3600, + "d" => 86_400, + "w" => 604_800, + _ => return Err(wrong()), + }; + now.checked_sub_signed(chrono::Duration::try_seconds(i64::from(count) * seconds).ok_or_else(wrong)?).ok_or_else(wrong)? + }; + u64::try_from(moment.timestamp_millis()).map_err(|_| wrong()) +} + +/// An instant as the CLI prints it. +pub fn shown(ms: u64) -> String { + DateTime::from_timestamp_millis(i64::try_from(ms).unwrap_or(i64::MAX)) + .map_or_else(|| ms.to_string(), |time| time.to_rfc3339_opts(SecondsFormat::Millis, true)) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn an_instant_is_rfc3339_or_an_age_before_now() { + let now = DateTime::parse_from_rfc3339("2026-11-11T12:00:00Z").unwrap().with_timezone(&Utc); + let at = |text: &str| instant_ms("--after", text, now).map(shown); + assert_eq!(at("2026-11-11T10:00:00Z").unwrap(), "2026-11-11T10:00:00.000Z"); + assert_eq!(at("2026-11-11T12:00:00+02:00").unwrap(), "2026-11-11T10:00:00.000Z"); + assert_eq!(at("6h").unwrap(), "2026-11-11T06:00:00.000Z"); + assert_eq!(at("2d").unwrap(), "2026-11-09T12:00:00.000Z"); + assert_eq!(at("90s").unwrap(), "2026-11-11T11:58:30.000Z"); + for bad in ["", "h", "6", "6x", "-2d", "yesterday", "1e3h", "+2d", "6日", "日", "4000000000w", "99999999999999999999d"] { + assert!(matches!(at(bad), Err(Trouble::InvalidInput(ref detail)) if detail.contains("--after")), "{bad}"); + } + } +} diff --git a/crates/eureka-state/src/bin/huginn/main.rs b/crates/eureka-state/src/bin/huginn/main.rs index 9f6cb10..0b98084 100644 --- a/crates/eureka-state/src/bin/huginn/main.rs +++ b/crates/eureka-state/src/bin/huginn/main.rs @@ -17,6 +17,7 @@ //! exits 1 when the campaign is not moved to this mind or this mind holds records of its own under it, and 4 when the history cannot be replayed whole. mod admit_hold; +mod chat; mod config; mod door; mod import; @@ -40,7 +41,7 @@ use huginn_mind::eureka_pipeline::{ Line, PipelineDocument, PipelineRef, Short, Slug, pipeline_key, }; use huginn_mind::{ - BATCH_MAX, Faculty, HuginnMindRequest, HuginnMindResponse, MindRefusal, PipelineAdmissionBatch, PipelineAdmissionOutcome, + BATCH_MAX, ChatQuery, Faculty, HuginnMindRequest, HuginnMindResponse, MindRefusal, PipelineAdmissionBatch, PipelineAdmissionOutcome, PersonaPutOutcome, PipelinePageItems, PipelineProvenance, PipelineSelectionPage, SemanticQuery, SourceQuery, }; use serde_json::Value; @@ -71,6 +72,34 @@ struct Cli { command: Command, } +#[derive(Subcommand, Debug)] +enum ChatCommand { + /// Chunks of the indexed rooms, each citing its room, + /// event ids and times. With texts, ranked by meaning; with none, a listing + /// oldest first, at most --top-k, ending in the `--after` to continue from. + Search { + /// Up to eight topics to find; the ranked lists of several are fused. + #[arg(num_args = 0..=8)] + texts: Vec, + /// Only this room: an id (`!abc:gamecult.org`) or a canonical alias + /// (`#swarm:gamecult.org`). Repeat for several. + #[arg(long = "room")] + rooms: Vec, + /// From this instant: an RFC 3339 time, or an age such as 6h or 2d. + #[arg(long)] + after: Option, + /// Until this instant, as --after. + #[arg(long)] + before: Option, + /// How many chunks (default 10, at most 50). + #[arg(long)] + top_k: Option, + /// Print the answer as JSON. + #[arg(long)] + json: bool, + }, +} + #[derive(Subcommand, Debug)] enum Command { /// Who this process acts for and whether the daemon answers. @@ -118,6 +147,11 @@ enum Command { #[arg(long)] json: bool, }, + /// Search the Minds' chat rooms. + Chat { + #[command(subcommand)] + command: ChatCommand, + }, #[command(flatten)] Recipe(Recipe), /// Admit a batch of documents, whole or not at all, and read it back. @@ -284,6 +318,9 @@ fn run(settings: &Settings, command: Command) -> Result { top_k, json, } => sources(settings, texts, repo, front, top_k, json), + Command::Chat { command: ChatCommand::Search { texts, rooms, after, before, top_k, json } } => { + chat_search(settings, texts, rooms, after.as_deref(), before.as_deref(), top_k, json) + } Command::Recipe(Recipe::Blocked { spec_id, page }) => { blocked(settings, &spec_id, page.cursor, page.documents) } @@ -471,6 +508,37 @@ fn sources( } } +fn chat_search( + settings: &Settings, + texts: Vec, + rooms: Vec, + after: Option<&str>, + before: Option<&str>, + top_k: Option, + json: bool, +) -> Result { + let now = chrono::Utc::now(); + let query = ChatQuery { + texts: texts.into_iter().map(Line).collect(), + rooms: rooms.into_iter().map(Short).collect(), + after_ms: after.map(|text| chat::instant_ms("--after", text, now)).transpose()?, + before_ms: before.map(|text| chat::instant_ms("--before", text, now)).transpose()?, + top_k: top_k.unwrap_or(DEFAULT_TOP_K), + }; + match settings.ask(|instance| HuginnMindRequest::SearchChat { instance: instance.clone(), query })? { + HuginnMindResponse::Chat(found) => { + if json { + say(serde_json::to_string_pretty(&found).map_err(|error| Trouble::Internal(error.to_string()))?); + } else { + say(render::chat_hits(&found)); + } + Ok(0) + } + HuginnMindResponse::Refused(refusal) => Ok(refused(&refusal)), + other => Err(unexpected(&other)), + } +} + fn show(page: &PipelineSelectionPage) { match &page.items { PipelinePageItems::Headers(headers) => { @@ -688,6 +756,7 @@ fn unexpected(answer: &HuginnMindResponse) -> Trouble { HuginnMindResponse::Persona(_) => "a persona", HuginnMindResponse::PersonaPut(_) => "a persona put", HuginnMindResponse::Sources(_) => "source hits", + HuginnMindResponse::Chat(_) => "chat hits", HuginnMindResponse::Activity(_) => "repo activity", HuginnMindResponse::RootsTouched(_) => "roots touched", HuginnMindResponse::Refused(_) => "a refusal", diff --git a/crates/eureka-state/src/bin/huginn/render.rs b/crates/eureka-state/src/bin/huginn/render.rs index 68e5f7a..5c5db1c 100644 --- a/crates/eureka-state/src/bin/huginn/render.rs +++ b/crates/eureka-state/src/bin/huginn/render.rs @@ -3,7 +3,7 @@ use std::fmt::Write; use huginn_mind::eureka_pipeline::{PipelineResolution, ResolutionOutcome}; -use huginn_mind::{SourceHits, IndexStatus, PersonaStatus, PipelineDocumentSummary, PipelineDocumentView, PipelineStatus, PipelineStatusSummary}; +use huginn_mind::{ChatHits, SourceHits, IndexStatus, PersonaStatus, PipelineDocumentSummary, PipelineDocumentView, PipelineStatus, PipelineStatusSummary}; use serde::Serialize; use serde_json::Value; @@ -147,6 +147,40 @@ pub fn source_hits(found: &SourceHits) -> String { ") } +/// Chat hits as the CLI prints them. Per hit the organ's own header line, +/// `#alias (!room) first..last time..time score` (the alias and the score only +/// when there are some), then one `time sender event_id body` line per message, +/// the body exactly as the room held it: quoted data under our header, never a +/// message framed as if sent to the reader. A blank line between hits; a +/// truncated listing ends with the `--after` that continues it. +pub fn chat_hits(found: &ChatHits) -> String { + let mut out: Vec = found + .hits + .iter() + .map(|hit| { + let room = hit.room_alias.as_ref().map_or_else(|| hit.room_id.clone(), |alias| format!("{alias} ({})", hit.room_id)); + let score = hit.score.map_or_else(String::new, |score| format!(" {score:.3}")); + let mut text = format!( + "{room} {}..{} {}..{}{score}\n", + hit.first_event_id, + hit.last_event_id, + crate::chat::shown(hit.ts_start_ms), + crate::chat::shown(hit.ts_end_ms) + ); + for message in &hit.messages { + let _ = writeln!(text, "{} {} {} {}", crate::chat::shown(message.ts_ms), message.sender, message.event_id, message.body); + } + text + }) + .collect(); + if found.truncated + && let Some(last) = found.hits.last() + { + out.push(format!("truncated: more follow, continue with --after {}\n", crate::chat::shown(last.ts_end_ms + 1))); + } + out.join("\n") +} + /// A fact as one token: a string, a number, a flag, a reference's id, or a /// list of those joined by commas. Anything deeper is not a header fact. fn scalar(value: &Value) -> Option { diff --git a/crates/eureka-state/tests/cli.rs b/crates/eureka-state/tests/cli.rs index 89e652a..0b81629 100644 --- a/crates/eureka-state/tests/cli.rs +++ b/crates/eureka-state/tests/cli.rs @@ -1233,7 +1233,8 @@ struct Desk { embeds_under: &'static str, held: Vec, tickets: u64, - done: Vec<(SearchTicket, Result)>, + chat: Vec, + done: Vec<(SearchTicket, huginn_daemon::daemon::Found)>, } /// An embedder that sends every text to one vector, under a digest. @@ -1297,6 +1298,35 @@ impl huginn_daemon::sources::SourceIndex for Desk { .map(|payload| huginn_daemon::sources::SourceRow { score: 0.9, payload: payload.clone() }) .collect()) } + + fn upsert_chat(&mut self, _points: &[huginn_daemon::sources::ChatPoint]) -> anyhow::Result<()> { + anyhow::bail!("a daemon never writes the source index") + } + + fn chat_nearest( + &mut self, + _vector: &[f32], + filter: &huginn_daemon::sources::ChatFilter, + limit: u32, + _within: Duration, + ) -> anyhow::Result> { + Ok(self.chat_scroll(filter, limit, _within)?.into_iter().map(|payload| huginn_daemon::sources::ChatRow { score: 0.9, payload }).collect()) + } + + fn chat_scroll(&mut self, filter: &huginn_daemon::sources::ChatFilter, limit: u32, _within: Duration) -> anyhow::Result> { + let mut kept: Vec<_> = self + .chat + .iter() + .filter(|payload| { + let times = payload.messages.iter().map(|message| message.ts_ms); + filter.keeps(&payload.room_id, payload.room_alias.as_deref(), times.clone().min().unwrap(), times.max().unwrap()) + }) + .cloned() + .collect(); + kept.sort_by_key(|payload| payload.messages[0].ts_ms); + kept.truncate(limit as usize); + Ok(kept) + } } impl IndexSink for Desk { @@ -1313,11 +1343,21 @@ impl IndexSink for Desk { self.tickets += 1; let ticket = SearchTicket(self.tickets); let mut embedder = Flat(self.embeds_under); - let found = huginn_daemon::sources::search(&mut embedder, self, query, Duration::from_secs(5)).map_err(|error| format!("{error:#}")); + let found = huginn_daemon::sources::search(&mut embedder, self, query, Duration::from_secs(5)) + .map_or_else(|error| huginn_daemon::daemon::Found::Failed(format!("{error:#}")), huginn_daemon::daemon::Found::Sources); self.done.push((ticket, found)); Ok(ticket) } - fn sources_searched(&mut self) -> Vec<(SearchTicket, Result)> { + fn search_chat(&mut self, query: &huginn_mind::ChatQuery) -> Result { + self.tickets += 1; + let ticket = SearchTicket(self.tickets); + let mut embedder = Flat(self.embeds_under); + let found = huginn_daemon::sources::search_chat(&mut embedder, self, query, Duration::from_secs(5)) + .map_or_else(|error| huginn_daemon::daemon::Found::Failed(format!("{error:#}")), huginn_daemon::daemon::Found::Chat); + self.done.push((ticket, found)); + Ok(ticket) + } + fn sources_searched(&mut self) -> Vec<(SearchTicket, huginn_daemon::daemon::Found)> { std::mem::take(&mut self.done) } fn search_deadline(&self) -> Duration { @@ -1351,6 +1391,15 @@ fn source_daemon(built_under: &'static str, embeds_under: &'static str) -> Serve chunk("Demo", "README.md", 1, 2, "# Demo"), chunk("Other", "x.rs", 1, 1, "fn other() {}"), ], + chat: vec![huginn_daemon::sources::ChatPayload { + room_id: "!r:gamecult.org".into(), + room_alias: Some("#swarm:gamecult.org".into()), + messages: vec![ + huginn_mind::ChatMessage { event_id: "$e1".into(), sender: "@nibu:gamecult.org".into(), ts_ms: 1_700_000_000_000, body: "ship the cut\nthen rest".into() }, + huginn_mind::ChatMessage { event_id: "$e2".into(), sender: "@eureka:gamecult.org".into(), ts_ms: 1_700_000_060_000, body: "done".into() }, + ], + text: "swarm".into(), + }], tickets: 0, done: Vec::new(), }; @@ -2886,3 +2935,42 @@ fn a_malformed_huginn_session_is_refused_naming_the_variable_without_echoing_it( let edge = with_session(live.addr, &"E".repeat(200), &admit_without_session(&file), None); assert_eq!(edge.code, 0, "{}", edge.err); } + +/// `huginn chat search` end to end: a listing through the daemon's worker path +/// prints the organ's header line and each message verbatim with its event id, +/// a room that holds nothing prints nothing, and the flags reach the request as +/// the instants and rooms they name. +#[test] +fn chat_search_prints_cited_messages_and_sends_its_selectors() { + let server = source_daemon("d1", "d1"); + let found = at(server.addr, &["chat", "search", "--room", "#swarm:gamecult.org", "--after", "2023-11-14T22:00:00Z"], None); + assert_eq!(found.code, 0, "{}", found.err); + assert_eq!( + found.out, + "#swarm:gamecult.org (!r:gamecult.org) $e1..$e2 2023-11-14T22:13:20.000Z..2023-11-14T22:14:20.000Z\n\ + 2023-11-14T22:13:20.000Z @nibu:gamecult.org $e1 ship the cut\nthen rest\n\ + 2023-11-14T22:14:20.000Z @eureka:gamecult.org $e2 done\n\n" + ); + let other = at(server.addr, &["chat", "search", "--room", "!elsewhere:gamecult.org"], None); + assert_eq!((other.code, other.out.trim()), (0, "")); + let by_meaning = at(server.addr, &["chat", "search", "who shipped it", "--top-k", "1"], None); + assert!(by_meaning.out.starts_with("#swarm:gamecult.org (!r:gamecult.org) $e1..$e2 ") && by_meaning.out.lines().next().unwrap().ends_with(" 0.900"), "{}", by_meaning.out); + + let (spy, seen) = replying( + "search_chat", + HuginnMindResponse::Chat(huginn_mind::ChatHits { indexed_under: huginn_mind::SourceModel { name: "m".into(), digest: "d".into() }, truncated: false, hits: vec![] }), + ); + let ran = at(spy.addr, &["chat", "search", "a", "b", "--room", "!x:y", "--room", "#z:y", "--after", "2023-11-14T22:13:20Z", "--before", "2023-11-14T22:14:20Z", "--top-k", "3"], None); + assert_eq!(ran.code, 0, "{}", ran.err); + let HuginnMindRequest::SearchChat { query, .. } = seen.lock().unwrap().remove(0) else { panic!("a chat search") }; + let texts: Vec<&str> = query.texts.iter().map(|text| text.0.as_str()).collect(); + let rooms: Vec<&str> = query.rooms.iter().map(|room| room.0.as_str()).collect(); + assert_eq!((texts, rooms, query.after_ms, query.before_ms, query.top_k), (vec!["a", "b"], vec!["!x:y", "#z:y"], Some(1_700_000_000_000), Some(1_700_000_060_000), 3)); + + assert_ne!(at(spy.addr, &["chat", "search", "--after", "yesterday"], None).code, 0); + assert!(seen.lock().unwrap().is_empty(), "a bad instant never left the CLI"); + for bad in [&["chat", "search", "--room", "swarm"][..], &["chat", "search", "--top-k", "0"]] { + let ran = at(server.addr, bad, None); + assert_eq!(ran.code, 1, "{bad:?} is a refusal by the daemon's door: {}", ran.err); + } +} diff --git a/crates/eureka-state/tests/common/mod.rs b/crates/eureka-state/tests/common/mod.rs index 4dd1367..075bfcf 100644 --- a/crates/eureka-state/tests/common/mod.rs +++ b/crates/eureka-state/tests/common/mod.rs @@ -93,10 +93,14 @@ impl IndexSink for Inert { fn search(&mut self, _text: &str, _top_k: u32, _among: Vec) -> Result { Err("this index cannot search".into()) } + fn search_chat(&mut self, _query: &huginn_mind::ChatQuery) -> Result { + Err("this index cannot search".into()) + } + fn search_sources(&mut self, _query: &huginn_mind::SourceQuery) -> Result { Err("this index cannot search".into()) } - fn sources_searched(&mut self) -> Vec<(SearchTicket, Result)> { + fn sources_searched(&mut self) -> Vec<(SearchTicket, huginn_daemon::daemon::Found)> { Vec::new() } fn search_deadline(&self) -> Duration { diff --git a/crates/huginn-daemon/src/daemon.rs b/crates/huginn-daemon/src/daemon.rs index 366b433..c6bad27 100644 --- a/crates/huginn-daemon/src/daemon.rs +++ b/crates/huginn-daemon/src/daemon.rs @@ -8,7 +8,7 @@ use crate::sources::forge::HttpForge; use huginn_mind::desk_wire::{DeskRequest, DeskResponse}; use huginn_mind::eureka_pipeline::{PipelineRef, Slug}; use huginn_mind::wire::{ACTIVITY_REPOS_MAX, HuginnMindRequest, HuginnMindResponse, IndexStatus}; -use huginn_mind::{Members, Mind, MindRefusal, MindStore, PipelineAdmissionOutcome, SemanticQuery, SourceHits, SourceQuery}; +use huginn_mind::{ChatHits, ChatQuery, Members, Mind, MindRefusal, MindStore, PipelineAdmissionOutcome, SemanticQuery, SourceHits, SourceQuery}; /// Where the index attaches. Called after a batch commits, with the refs the /// batch landed, derived writes included; the sink may read each through @@ -34,8 +34,12 @@ pub trait IndexSink { /// Starts a source search. The source index is not the mind's: it needs no /// member ids, and nothing about the mind's own index state refuses it. fn search_sources(&mut self, query: &SourceQuery) -> Result; - /// Every source search that finished since the last call, by ticket. - fn sources_searched(&mut self) -> Vec<(SearchTicket, Result)>; + /// Starts a chat search: the same queue and worker as a source search, + /// over the same collection. + fn search_chat(&mut self, query: &ChatQuery) -> Result; + /// Every source or chat search that finished since the last call, by + /// ticket: `Found::Sources`, `Found::Chat` or `Found::Failed`. + fn sources_searched(&mut self) -> Vec<(SearchTicket, Found)>; fn search_deadline(&self) -> std::time::Duration; /// Every search that finished since the last call, by ticket: its /// candidates and scores, or why it failed. @@ -78,6 +82,7 @@ struct MindSearch { pub enum Found { Mind(Hits), Sources(SourceHits), + Chat(ChatHits), Failed(String), } @@ -205,6 +210,15 @@ impl> Daemon { Ok(roots) => HuginnMindResponse::RootsTouched(roots), Err(refusal) => HuginnMindResponse::Refused(refusal), }, + HuginnMindRequest::SearchChat { query, .. } => { + if let Err(refusal) = query.check() { + return Handled::Answered(HuginnMindResponse::Refused(refusal)); + } + return match self.index.search_chat(&query) { + Ok(ticket) => Handled::Searching(Search { ticket, mind: None }), + Err(detail) => Handled::Answered(HuginnMindResponse::Refused(MindRefusal::Unavailable { detail })), + }; + } HuginnMindRequest::StewardedActivity { instance, repos, after } => { let refused = |message: &str| MindRefusal::SelectionInvalid { field: "repos".into(), value: None, message: message.into() }; if repos.len() > ACTIVITY_REPOS_MAX { @@ -282,8 +296,7 @@ impl> Daemon { /// Searches the index has finished since this was last called. pub fn searched(&mut self) -> Vec<(SearchTicket, Found)> { let mind = self.index.searched().into_iter().map(|(ticket, found)| (ticket, found.map_or_else(Found::Failed, Found::Mind))); - let sources = self.index.sources_searched().into_iter().map(|(ticket, found)| (ticket, found.map_or_else(Found::Failed, Found::Sources))); - mind.chain(sources).collect() + mind.chain(self.index.sources_searched()).collect() } /// The client has stopped waiting for `ticket`. @@ -300,6 +313,7 @@ impl> Daemon { let (selection, semantic, members, hits) = match (search.mind, found) { (_, Found::Failed(detail)) => return unavailable(detail), (None, Found::Sources(hits)) => return HuginnMindResponse::Sources(hits), + (None, Found::Chat(hits)) => return HuginnMindResponse::Chat(hits), (Some(asked), Found::Mind(hits)) => (asked.selection, asked.semantic, asked.members, hits), _ => return unavailable("the index answered another kind of search than the one asked".into()), }; @@ -326,6 +340,7 @@ pub(crate) mod tests { StructuralDelta, Title, VerificationTest, }; use huginn_mind::wire::MindStatus; + use huginn_mind::ChatQuery; use huginn_mind::{MindRefusal, OwnedRedbMessagePackBackingStore}; use std::path::Path; use cultnet_rs::Selection; @@ -580,11 +595,15 @@ pub(crate) mod tests { Err("this index cannot search".into()) } + fn search_chat(&mut self, _query: &ChatQuery) -> Result { + Err("this index cannot search".into()) + } + fn search_sources(&mut self, _query: &SourceQuery) -> Result { Err("this index cannot search".into()) } - fn sources_searched(&mut self) -> Vec<(SearchTicket, Result)> { + fn sources_searched(&mut self) -> Vec<(SearchTicket, Found)> { Vec::new() } @@ -619,13 +638,17 @@ pub(crate) mod tests { Err("this index cannot search the mind".into()) } + fn search_chat(&mut self, _query: &ChatQuery) -> Result { + Err("this index cannot search".into()) + } + fn search_sources(&mut self, _query: &SourceQuery) -> Result { *self.asked.lock().unwrap() += 1; Ok(SearchTicket(7)) } - fn sources_searched(&mut self) -> Vec<(SearchTicket, Result)> { - vec![(SearchTicket(7), Ok(self.hits.clone()))] + fn sources_searched(&mut self) -> Vec<(SearchTicket, Found)> { + vec![(SearchTicket(7), Found::Sources(self.hits.clone()))] } fn search_deadline(&self) -> std::time::Duration { @@ -685,6 +708,31 @@ pub(crate) mod tests { assert_eq!(*asked.lock().unwrap(), 3, "a refused query never reached the index"); } + /// A chat search is checked at the door before the index is asked, is + /// refused for another mind, turns an index that cannot be asked into + /// `Unavailable`, and its finished hits are the answer as they are. + #[test] + fn a_chat_search_is_checked_at_the_door_and_answers_its_hits() { + use huginn_mind::eureka_pipeline::{Line, Short}; + use huginn_mind::{ChatHits, SourceModel}; + let root = tempfile::tempdir().unwrap(); + let mut daemon = Daemon::new(Mind::open(root.path(), &slug(INSTANCE)).unwrap(), NoIndex); + let request = |instance: &str, rooms: &[&str], top_k: u32| HuginnMindRequest::SearchChat { + instance: slug(instance), + query: ChatQuery { texts: vec![Line("find".into())], rooms: rooms.iter().map(|room| Short((*room).into())).collect(), after_ms: None, before_ms: None, top_k }, + }; + for (refused, field) in [(request(INSTANCE, &[], 0), "top_k"), (request(INSTANCE, &["swarm"], 5), "rooms")] { + let answer = daemon.handle(refused, now()).answered(); + assert!(matches!(answer, HuginnMindResponse::Refused(MindRefusal::SelectionInvalid { field: ref named, .. }) if named == field), "{answer:?}"); + } + assert!(matches!(daemon.handle(request(OTHER, &[], 5), now()).answered(), HuginnMindResponse::Refused(_))); + let answer = daemon.handle(request(INSTANCE, &["#swarm:x"], 5), now()).answered(); + assert!(matches!(answer, HuginnMindResponse::Refused(MindRefusal::Unavailable { .. })), "{answer:?}"); + let hits = ChatHits { indexed_under: SourceModel { name: "m".into(), digest: "d".into() }, truncated: false, hits: vec![] }; + let search = Search { ticket: SearchTicket(1), mind: None }; + assert_eq!(daemon.finish(search, Found::Chat(hits.clone())), HuginnMindResponse::Chat(hits)); + } + /// A daemon over the instance's mind under `state_root`, with no index. pub(crate) fn open_unindexed( state_root: &Path, @@ -716,11 +764,15 @@ pub(crate) mod tests { Err("this index cannot search".into()) } + fn search_chat(&mut self, _query: &ChatQuery) -> Result { + Err("this index cannot search".into()) + } + fn search_sources(&mut self, _query: &SourceQuery) -> Result { Err("this index cannot search".into()) } - fn sources_searched(&mut self) -> Vec<(SearchTicket, Result)> { + fn sources_searched(&mut self) -> Vec<(SearchTicket, Found)> { Vec::new() } @@ -750,11 +802,15 @@ pub(crate) mod tests { Err("this index cannot search".into()) } + fn search_chat(&mut self, _query: &ChatQuery) -> Result { + Err("this index cannot search".into()) + } + fn search_sources(&mut self, _query: &SourceQuery) -> Result { Err("this index cannot search".into()) } - fn sources_searched(&mut self) -> Vec<(SearchTicket, Result)> { + fn sources_searched(&mut self) -> Vec<(SearchTicket, Found)> { Vec::new() } diff --git a/crates/huginn-daemon/src/index/fakes.rs b/crates/huginn-daemon/src/index/fakes.rs index b62ce8b..7a905fb 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::{DescribedSources, SourceIndex, SourcePoint, SourceRow, SourcesMeta}; +use crate::sources::{ChatFilter, ChatPayload, ChatPoint, ChatRow, DescribedSources, SourceIndex, SourcePoint, SourceRow, SourcesMeta}; use super::{CollectionMeta, Described, Embedder, Hit, ModelIdentity, Point, VectorIndex, point_id}; @@ -253,6 +253,8 @@ pub(crate) fn pair() -> (FakeEmbedder, FakeIndex) { pub(crate) struct SourcesState { pub(crate) described: Option, pub(crate) points: BTreeMap, + /// The chat points, by id: the same collection, the other corpus. + pub(crate) chat: BTreeMap, /// How many times the collection was recreated. pub(crate) recreated: u32, pub(crate) upsert_calls: u32, @@ -317,6 +319,7 @@ impl SourceIndex for FakeSources { let mut state = self.state.lock().unwrap(); state.recreated += 1; state.points.clear(); + state.chat.clear(); state.described = Some(DescribedSources::Labelled { meta: meta.clone(), cursors: BTreeMap::new() }); Ok(()) } @@ -369,6 +372,44 @@ impl SourceIndex for FakeSources { rows.truncate(limit as usize); Ok(rows) } + + fn upsert_chat(&mut self, points: &[ChatPoint]) -> Result<()> { + let mut state = self.state.lock().unwrap(); + for point in points { + ensure!(!point.payload.messages.is_empty(), "a chat chunk carries no messages"); + state.chat.insert(point.id.clone(), point.clone()); + } + Ok(()) + } + + fn chat_nearest(&mut self, vector: &[f32], filter: &ChatFilter, limit: u32, _within: Duration) -> Result> { + let state = self.state.lock().unwrap(); + let mut rows: Vec = state + .chat + .values() + .filter(|point| chat_kept(filter, &point.payload)) + .map(|point| ChatRow { score: cosine(vector, &point.vector), payload: point.payload.clone() }) + .collect(); + rows.sort_by(|a, b| b.score.total_cmp(&a.score)); + rows.truncate(limit as usize); + Ok(rows) + } + + fn chat_scroll(&mut self, filter: &ChatFilter, limit: u32, _within: Duration) -> Result> { + let state = self.state.lock().unwrap(); + let mut kept: Vec<&ChatPayload> = state.chat.values().map(|point| &point.payload).filter(|payload| chat_kept(filter, payload)).collect(); + kept.sort_by_key(|payload| payload.messages.iter().map(|message| message.ts_ms).min()); + Ok(kept.into_iter().take(limit as usize).cloned().collect()) + } +} + +/// A chat chunk under a filter, by the span its messages make. +fn chat_kept(filter: &ChatFilter, payload: &ChatPayload) -> bool { + let times = || payload.messages.iter().map(|message| message.ts_ms); + match (times().min(), times().max()) { + (Some(start), Some(end)) => filter.keeps(&payload.room_id, payload.room_alias.as_deref(), start, end), + _ => false, + } } impl SourceIndex for FakeIndex { @@ -390,4 +431,13 @@ impl SourceIndex for FakeIndex { fn search_sources(&mut self, vector: &[f32], repo: Option<&str>, front: bool, limit: u32, within: Duration) -> Result> { self.sources.search_sources(vector, repo, front, limit, within) } + fn upsert_chat(&mut self, points: &[ChatPoint]) -> Result<()> { + self.sources.upsert_chat(points) + } + fn chat_nearest(&mut self, vector: &[f32], filter: &ChatFilter, limit: u32, within: Duration) -> Result> { + self.sources.chat_nearest(vector, filter, limit, within) + } + fn chat_scroll(&mut self, filter: &ChatFilter, limit: u32, within: Duration) -> Result> { + self.sources.chat_scroll(filter, limit, within) + } } diff --git a/crates/huginn-daemon/src/index/mod.rs b/crates/huginn-daemon/src/index/mod.rs index 0c7c19d..167972f 100644 --- a/crates/huginn-daemon/src/index/mod.rs +++ b/crates/huginn-daemon/src/index/mod.rs @@ -105,9 +105,9 @@ use huginn_mind::wire::IndexStatus; use huginn_mind::{INDEX_TEXT_VERSION, IndexEntry, Mind, MindStore}; use sha2::{Digest, Sha256}; -use crate::daemon::{Hits, IndexSink, SearchTicket}; +use crate::daemon::{Found, Hits, IndexSink, SearchTicket}; use crate::sources::{self, SourceIndex}; -use huginn_mind::{SourceHits, SourceQuery}; +use huginn_mind::{ChatQuery, SourceQuery}; use ollama::QUERY_INSTRUCTION; /// The `managed_by` a collection carries when this organ made it. @@ -685,11 +685,17 @@ struct Queued { due: Instant, } -/// A source search waiting for the worker. It is not the mind's: nothing about -/// the mind's reconciliation gates it. +/// What a source-collection search asks: forge chunks or chat chunks. +enum SourceAsk { + Sources(SourceQuery), + Chat(ChatQuery), +} + +/// A source or chat search waiting for the worker. It is not the mind's: +/// nothing about the mind's reconciliation gates it. struct QueuedSource { ticket: SearchTicket, - query: SourceQuery, + ask: SourceAsk, due: Instant, } @@ -706,7 +712,7 @@ struct Shared { /// Searches the worker finished and the serve loop has not collected. found: Vec<(SearchTicket, Result)>, source_searches: VecDeque, - sources_found: Vec<(SearchTicket, Result)>, + sources_found: Vec<(SearchTicket, Found)>, next_ticket: u64, } @@ -821,10 +827,14 @@ fn serve_source_search(projector: &mu if guard.closed { return; } - let Some(QueuedSource { ticket, query, due }) = guard.source_searches.pop_front() else { return }; + let Some(QueuedSource { ticket, ask, due }) = guard.source_searches.pop_front() else { return }; drop(guard); - let found = sources::search(&mut projector.embedder, &mut projector.index, &query, due.saturating_duration_since(Instant::now())) - .map_err(|error| format!("{error:#}")); + let within = due.saturating_duration_since(Instant::now()); + let found = match ask { + SourceAsk::Sources(query) => sources::search(&mut projector.embedder, &mut projector.index, &query, within).map(Found::Sources), + SourceAsk::Chat(query) => sources::search_chat(&mut projector.embedder, &mut projector.index, &query, within).map(Found::Chat), + } + .unwrap_or_else(|error| Found::Failed(format!("{error:#}"))); lock(shared).sources_found.push((ticket, found)); } @@ -966,6 +976,24 @@ pub struct WorkerSink { } impl WorkerSink { + /// Queues a source or chat search behind the same bound as every other + /// search; the mind's index state does not refuse it. + fn queue_source_ask(&mut self, ask: SourceAsk) -> Result { + let mut guard = lock(&self.shared); + if guard.searches.len() + guard.source_searches.len() >= SEARCHES_QUEUED_MAX { + return Err(format!( + "the index has {} searches waiting and takes no more until it has answered some", + guard.searches.len() + guard.source_searches.len() + )); + } + let ticket = SearchTicket(guard.next_ticket); + guard.next_ticket += 1; + guard.source_searches.push_back(QueuedSource { ticket, ask, due: Instant::now() + self.deadline }); + drop(guard); + self.shared.1.notify_all(); + Ok(ticket) + } + /// Starts the worker. `mind` is the mind's genesis receipt id, `None` for /// a mind not yet written. `startup` is every indexable document the mind /// holds now; the worker's first act is to reconcile the collection @@ -1086,22 +1114,14 @@ impl IndexSink for WorkerSink { } fn search_sources(&mut self, query: &SourceQuery) -> Result { - let mut guard = lock(&self.shared); - if guard.searches.len() + guard.source_searches.len() >= SEARCHES_QUEUED_MAX { - return Err(format!( - "the index has {} searches waiting and takes no more until it has answered some", - guard.searches.len() + guard.source_searches.len() - )); - } - let ticket = SearchTicket(guard.next_ticket); - guard.next_ticket += 1; - guard.source_searches.push_back(QueuedSource { ticket, query: query.clone(), due: Instant::now() + self.deadline }); - drop(guard); - self.shared.1.notify_all(); - Ok(ticket) + self.queue_source_ask(SourceAsk::Sources(query.clone())) } - fn sources_searched(&mut self) -> Vec<(SearchTicket, Result)> { + fn search_chat(&mut self, query: &ChatQuery) -> Result { + self.queue_source_ask(SourceAsk::Chat(query.clone())) + } + + fn sources_searched(&mut self) -> Vec<(SearchTicket, Found)> { std::mem::take(&mut lock(&self.shared).sources_found) } diff --git a/crates/huginn-daemon/src/index/qdrant.rs b/crates/huginn-daemon/src/index/qdrant.rs index b897dc9..66d08e4 100644 --- a/crates/huginn-daemon/src/index/qdrant.rs +++ b/crates/huginn-daemon/src/index/qdrant.rs @@ -7,7 +7,12 @@ use std::time::Duration; use anyhow::{Context, Result, ensure}; use serde_json::{Value, json}; -use crate::sources::{COLLECTION, FRONT_PAGE_FILES, DescribedSources, SourceIndex, SourcePayload, SourcePoint, SourceRow, SourcesMeta}; +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, +}; use super::{CollectionMeta, INDEX_CALL_TIMEOUT, Described, Hit, Http, Method, Point, VectorIndex}; @@ -175,21 +180,94 @@ impl VectorIndex for QdrantIndex { } } -/// The payload filter of a source search: the repo, and the front-page paths -/// (exact matches on the keyword-indexed `path`), both when both are asked. -fn source_filter(repo: Option<&str>, front: bool) -> Option { - let mut must = Vec::new(); +/// The payload filter of a source search: forge points only, then the repo and +/// the front-page paths (exact matches on the keyword-indexed `path`) when +/// asked. A chat point carries no repo or path and can never pass it. +fn source_filter(repo: Option<&str>, front: bool) -> Value { + let mut must = vec![json!({ "key": "corpus", "match": { "value": CORPUS_FORGE } })]; if let Some(repo) = repo { must.push(json!({ "key": "repo", "match": { "value": repo } })); } if front { must.push(json!({ "key": "path", "match": { "any": FRONT_PAGE_FILES } })); } - (!must.is_empty()).then(|| json!({ "must": must })) + json!({ "must": must }) +} + +/// The payload filter of a chat read: chat points only; the rooms any-of by id +/// (`!`) or alias (`#`) as a `should`; the span as ranges that keep a chunk +/// overlapping [after, before]. +fn chat_filter(filter: &ChatFilter) -> Value { + let mut must = vec![json!({ "key": "corpus", "match": { "value": CORPUS_CHAT } })]; + if let Some(after) = filter.after_ms { + must.push(json!({ "key": "ts_end_ms", "range": { "gte": after } })); + } + if let Some(before) = filter.before_ms { + must.push(json!({ "key": "ts_start_ms", "range": { "lte": before } })); + } + let mut out = json!({ "must": must }); + let (ids, aliases): (Vec<&String>, Vec<&String>) = filter.rooms.iter().partition(|room| room.starts_with('!')); + let mut should = Vec::new(); + if !ids.is_empty() { + should.push(json!({ "key": "room_id", "match": { "any": ids } })); + } + if !aliases.is_empty() { + should.push(json!({ "key": "room_alias", "match": { "any": aliases } })); + } + if !should.is_empty() { + out["should"] = Value::Array(should); + } + out +} + +/// A chat chunk's payload. The span, first and last ids and the id list are +/// derived from `messages`, so they cannot disagree with them. +fn chat_payload_json(payload: &ChatPayload) -> Result { + let (Some(first), Some(last)) = (payload.messages.first(), payload.messages.last()) else { + anyhow::bail!("a chat chunk of {} carries no messages", payload.room_id); + }; + let mut out = json!({ + "corpus": CORPUS_CHAT, + "room_id": payload.room_id, + "ts_start_ms": payload.messages.iter().map(|message| message.ts_ms).min(), + "ts_end_ms": payload.messages.iter().map(|message| message.ts_ms).max(), + "first_event_id": first.event_id, + "last_event_id": last.event_id, + "event_ids": payload.messages.iter().map(|message| &message.event_id).collect::>(), + "messages": payload.messages.iter().map(|message| json!({ + "event_id": message.event_id, "sender": message.sender, "ts_ms": message.ts_ms, "body": message.body, + })).collect::>(), + "text": payload.text, + }); + if let Some(alias) = &payload.room_alias { + out["room_alias"] = json!(alias); + } + Ok(out) +} + +fn chat_payload_of(payload: &Value) -> Result { + ensure!(payload["corpus"] == CORPUS_CHAT, "a chat read was handed a point of another corpus"); + let text = |value: &Value, key: &str| value[key].as_str().map(str::to_owned).with_context(|| format!("a chat chunk carries no {key}")); + let messages = payload["messages"] + .as_array() + .context("a chat chunk carries no messages")? + .iter() + .map(|message| { + Ok(ChatMessage { + event_id: text(message, "event_id")?, + sender: text(message, "sender")?, + ts_ms: message["ts_ms"].as_u64().context("a chat message carries no ts_ms")?, + body: text(message, "body")?, + }) + }) + .collect::>>()?; + ensure!(!messages.is_empty(), "a chat chunk carries no messages"); + Ok(ChatPayload { room_id: text(payload, "room_id")?, room_alias: payload["room_alias"].as_str().map(str::to_owned), messages, text: text(payload, "text")? }) } fn payload_json(payload: &SourcePayload) -> Value { json!({ + "corpus": CORPUS_FORGE, "repo": payload.repo, "path": payload.path, "commit": payload.commit, @@ -270,8 +348,17 @@ impl SourceIndex for QdrantIndex { }, }); self.ask(Method::Put, &format!("{path}?wait=true"), Some(&body), &[200])?; - for field in ["repo", "path"] { - let index = json!({ "field_name": field, "field_schema": "keyword" }); + for (field, schema) in [ + ("repo", "keyword"), + ("path", "keyword"), + ("corpus", "keyword"), + ("room_id", "keyword"), + ("room_alias", "keyword"), + ("event_ids", "keyword"), + ("ts_start_ms", "integer"), + ("ts_end_ms", "integer"), + ] { + let index = json!({ "field_name": field, "field_schema": schema }); self.ask(Method::Put, &format!("{path}/index?wait=true"), Some(&index), &[200])?; } Ok(()) @@ -315,9 +402,7 @@ impl SourceIndex for QdrantIndex { let collection = self.sources.clone(); let http = self.http.bounded(within); let mut body = json!({ "vector": vector, "limit": limit, "with_payload": true, "with_vector": false }); - if let Some(filter) = source_filter(repo, front) { - body["filter"] = filter; - } + body["filter"] = source_filter(repo, front); let (_, answer) = self.ask_with(&http, Method::Post, &format!("/collections/{collection}/points/search"), Some(&body), &[200])?; let found = answer["result"].as_array().context("Qdrant's search has no result")?; @@ -331,6 +416,45 @@ impl SourceIndex for QdrantIndex { }) .collect() } + + fn upsert_chat(&mut self, points: &[ChatPoint]) -> Result<()> { + let collection = self.sources.clone(); + let points: Vec = points + .iter() + .map(|point| Ok(json!({ "id": point.id, "vector": point.vector, "payload": chat_payload_json(&point.payload)? }))) + .collect::>()?; + self.ask(Method::Put, &format!("/collections/{collection}/points?wait=true"), Some(&json!({ "points": points })), &[200])?; + Ok(()) + } + + fn chat_nearest(&mut self, vector: &[f32], filter: &ChatFilter, limit: u32, within: Duration) -> Result> { + let collection = self.sources.clone(); + let http = self.http.bounded(within); + let body = json!({ "vector": vector, "limit": limit, "filter": chat_filter(filter), "with_payload": true, "with_vector": false }); + let (_, answer) = + self.ask_with(&http, Method::Post, &format!("/collections/{collection}/points/search"), Some(&body), &[200])?; + let found = answer["result"].as_array().context("Qdrant's search has no result")?; + found + .iter() + .map(|point| Ok(ChatRow { score: point["score"].as_f64().context("a hit carries no score")? as f32, payload: chat_payload_of(&point["payload"])? })) + .collect() + } + + fn chat_scroll(&mut self, filter: &ChatFilter, limit: u32, within: Duration) -> Result> { + let collection = self.sources.clone(); + let http = self.http.bounded(within); + let body = json!({ + "filter": chat_filter(filter), + "limit": limit, + "order_by": { "key": "ts_start_ms", "direction": "asc" }, + "with_payload": true, + "with_vector": false, + }); + let (_, answer) = + self.ask_with(&http, Method::Post, &format!("/collections/{collection}/points/scroll"), Some(&body), &[200])?; + let found = answer["result"]["points"].as_array().context("Qdrant's scroll has no points")?; + found.iter().map(|point| chat_payload_of(&point["payload"])).collect() + } } #[cfg(test)] @@ -339,9 +463,10 @@ mod source_filter_tests { /// No ask, no filter; a repo alone is the repo's match. #[test] - fn no_ask_is_no_filter_and_a_repo_is_its_own_match() { - assert_eq!(source_filter(None, false), None); - assert_eq!(source_filter(Some("Huginn"), false), Some(json!({ "must": [{ "key": "repo", "match": { "value": "Huginn" } }] }))); + fn no_ask_is_the_forge_corpus_and_a_repo_is_its_own_match() { + let forge = json!({ "key": "corpus", "match": { "value": "forge" } }); + assert_eq!(source_filter(None, false), json!({ "must": [forge] })); + assert_eq!(source_filter(Some("Huginn"), false), json!({ "must": [forge, { "key": "repo", "match": { "value": "Huginn" } }] })); } /// The front filter matches the path against exactly the front-page files @@ -354,15 +479,63 @@ mod source_filter_tests { assert_eq!(path.len(), 1); path[0]["match"]["any"].as_array().unwrap().iter().map(|p| p.as_str().unwrap().to_owned()).collect() }; - let front = source_filter(None, true).unwrap(); + let front = source_filter(None, true); assert_eq!(any(&front), ["README.md", "README", "readme.md", "README.markdown", "CLAUDE.md", "AGENTS.md"]); - assert_eq!(front["must"].as_array().unwrap().len(), 1, "no repo condition when none was asked"); + assert_eq!(front["must"].as_array().unwrap().len(), 2, "the corpus and the path; no repo condition when none was asked"); - let both = source_filter(Some("Zyphos"), true).unwrap(); + let both = source_filter(Some("Zyphos"), true); assert_eq!(any(&both), any(&front)); let repos: Vec<&Value> = both["must"].as_array().unwrap().iter().filter(|c| c["key"] == "repo").collect(); assert_eq!(repos.len(), 1); assert_eq!(repos[0]["match"]["value"], "Zyphos"); - assert_eq!(both["must"].as_array().unwrap().len(), 2); + assert_eq!(both["must"].as_array().unwrap().len(), 3); + } + + fn payload() -> ChatPayload { + let message = |id: &str, ts: u64| ChatMessage { event_id: id.into(), sender: "@a:gamecult.org".into(), ts_ms: ts, body: format!("body {id}") }; + ChatPayload { room_id: "!r:gamecult.org".into(), room_alias: Some("#swarm:gamecult.org".into()), messages: vec![message("$a", 10), message("$b", 20)], text: "swarm\n10 @a: body".into() } + } + + /// The chat codec is its own inverse, writes the contract's fields derived + /// from the messages, and refuses a chunk with none and a point of another corpus. + #[test] + fn the_chat_payload_round_trips_the_contract() { + let written = chat_payload_json(&payload()).unwrap(); + assert_eq!(written["corpus"], "chat"); + assert_eq!((written["ts_start_ms"].as_u64(), written["ts_end_ms"].as_u64()), (Some(10), Some(20))); + assert_eq!((written["first_event_id"].as_str(), written["last_event_id"].as_str()), (Some("$a"), Some("$b"))); + assert_eq!(written["event_ids"], json!(["$a", "$b"])); + assert_eq!(written["room_alias"], "#swarm:gamecult.org"); + assert_eq!(chat_payload_of(&written).unwrap(), payload()); + let mut no_alias = payload(); + no_alias.room_alias = None; + assert!(chat_payload_json(&no_alias).unwrap().get("room_alias").is_none()); + assert_eq!(chat_payload_of(&chat_payload_json(&no_alias).unwrap()).unwrap(), no_alias); + assert!(chat_payload_json(&ChatPayload { messages: vec![], ..payload() }).is_err()); + let forge = payload_json(&SourcePayload { repo: "r".into(), path: "p".into(), commit: "c".into(), blob: "b".into(), line_start: 1, line_end: 2, chunk_index: 0, ext: "rs".into(), text: "t".into() }); + assert_eq!(forge["corpus"], "forge"); + assert!(chat_payload_of(&forge).is_err()); + } + + /// 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] + fn the_chat_filter_is_corpus_rooms_and_overlap() { + let filter = chat_filter(&ChatFilter { rooms: vec!["!r:x".into(), "#a:x".into(), "#b:x".into()], after_ms: Some(5), before_ms: Some(9) }); + assert_eq!( + filter, + json!({ + "must": [ + { "key": "corpus", "match": { "value": "chat" } }, + { "key": "ts_end_ms", "range": { "gte": 5 } }, + { "key": "ts_start_ms", "range": { "lte": 9 } }, + ], + "should": [ + { "key": "room_id", "match": { "any": ["!r:x"] } }, + { "key": "room_alias", "match": { "any": ["#a:x", "#b:x"] } }, + ], + }) + ); + assert_eq!(chat_filter(&ChatFilter::default()), json!({ "must": [{ "key": "corpus", "match": { "value": "chat" } }] })); } } diff --git a/crates/huginn-daemon/src/index/tests.rs b/crates/huginn-daemon/src/index/tests.rs index dfd16af..1bca83d 100644 --- a/crates/huginn-daemon/src/index/tests.rs +++ b/crates/huginn-daemon/src/index/tests.rs @@ -1999,10 +1999,19 @@ fn ask_sources(sink: &mut WorkerSink, text: &str) -> Result>::search_sources(sink, &query) } -fn sources_answers(sink: &mut WorkerSink, count: usize) -> Vec<(SearchTicket, Result)> { +fn sources_answers(sink: &mut WorkerSink, count: usize) -> Vec<(SearchTicket, Result)> { let mut all = Vec::new(); eventually("the source searches being answered", || { - all.extend(>::sources_searched(sink)); + all.extend(>::sources_searched(sink).into_iter().map(|(ticket, found)| { + ( + ticket, + match found { + crate::daemon::Found::Sources(hits) => Ok(hits), + crate::daemon::Found::Failed(why) => Err(why), + other => panic!("a source search was answered with another kind: {}", matches!(other, crate::daemon::Found::Chat(_))), + }, + ) + })); all.len() >= count }); all @@ -2046,7 +2055,7 @@ fn a_worker_stuck_on_the_minds_collection_still_answers_a_source_search() { index.plant(&collection_name(&slug(INSTANCE)), mind_label); let meta = SourcesMeta { managed_by: MANAGED_BY.into(), - corpus: "forge:GameCult".into(), + corpus: crate::sources::CORPUS.into(), model: "model".into(), model_digest: "d1".into(), dimensions: 4, @@ -2081,6 +2090,60 @@ fn a_worker_stuck_on_the_minds_collection_still_answers_a_source_search() { assert_eq!(answered.indexed_under.digest, "d1"); } +/// A chat search rides the same queue and worker as a source search, answers +/// from the same collection while the mind's own collection is stuck, and comes +/// back as chat hits. +#[test] +fn a_worker_answers_a_chat_search_from_the_shared_collection() { + use crate::sources::{ChatPayload, ChatPoint, DescribedSources, SourcesMeta}; + use huginn_mind::ChatMessage; + let (embedder, index) = pair(); + index.plant( + &collection_name(&slug(INSTANCE)), + Described::Labelled(CollectionMeta { + managed_by: MANAGED_BY.into(), + instance: INSTANCE.into(), + mind: "another-mind".into(), + model: "model".into(), + model_digest: "d1".into(), + dimensions: 4, + index_text_version: INDEX_TEXT_VERSION, + }), + ); + let meta = SourcesMeta { managed_by: MANAGED_BY.into(), corpus: crate::sources::CORPUS.into(), model: "model".into(), model_digest: "d1".into(), dimensions: 4, chunker_version: 1 }; + index.sources.plant(DescribedSources::Labelled { meta, cursors: Default::default() }); + index + .sources + .state + .lock() + .unwrap() + .chat + .insert( + "c1".into(), + ChatPoint { + id: "c1".into(), + vector: vec![1.0; 4], + payload: ChatPayload { + room_id: "!r:x".into(), + room_alias: None, + messages: vec![ChatMessage { event_id: "$e".into(), sender: "@a:x".into(), ts_ms: 5, body: "hi".into() }], + text: "t".into(), + }, + }, + ); + let mut sink = WorkerSink::spawn(embedder, index.clone(), &slug(INSTANCE), Some(MIND.into()), entries_of(1), quick()); + sink.wait_status(|status| matches!(status, IndexStatus::Refused { .. })); + let query = huginn_mind::ChatQuery { texts: vec![], rooms: vec![], after_ms: None, before_ms: None, top_k: 5 }; + let ticket = >::search_chat(&mut sink, &query).unwrap(); + let mut found = Vec::new(); + eventually("the chat search being answered", || { + found.extend(>::sources_searched(&mut sink)); + !found.is_empty() + }); + let (answered, Found::Chat(hits)) = found.remove(0) else { panic!("a chat ask is answered with chat hits") }; + assert_eq!((answered, hits.hits.len(), hits.hits[0].first_event_id.as_str(), hits.truncated), (ticket, 1, "$e", false)); +} + /// Source searches take their own tickets, share the queue's bound with the /// mind's searches, and an abandon removes the one ticket named from either. #[test] @@ -2107,3 +2170,47 @@ fn source_searches_have_distinct_tickets_share_the_queue_bound_and_abandon_by_ti abandon(&mut sink, second); assert!(ask_sink(&mut sink, "now", 3).is_ok(), "a freed slot is a slot"); } + +/// The chat reads state their request: a scroll is the filter, oldest chunk +/// start first; a nearest search is the filter too. +#[test] +fn qdrant_chat_scroll_and_nearest_send_the_filter_and_the_order() { + use crate::sources::{ChatFilter, SourceIndex}; + let filter = ChatFilter { rooms: vec!["!r:x".into()], after_ms: Some(5), before_ms: None }; + let scrolling = stub(&[("POST /collections/", 200, r#"{"result":{"points":[]}}"#)]); + QdrantIndex::new(&scrolling.url).chat_scroll(&filter, 7, TIME).unwrap(); + let scroll: serde_json::Value = serde_json::from_str(&scrolling.seen.lock().unwrap()[0].2).unwrap(); + assert_eq!(scroll["order_by"], serde_json::json!({ "key": "ts_start_ms", "direction": "asc" })); + assert_eq!(scroll["filter"]["must"][0], serde_json::json!({ "key": "corpus", "match": { "value": "chat" } })); + assert_eq!(scroll["filter"]["must"][1]["range"], serde_json::json!({ "gte": 5 })); + assert_eq!(scroll["limit"], 7); + + let searching = stub(&[("POST /collections/", 200, r#"{"result":[]}"#)]); + QdrantIndex::new(&searching.url).chat_nearest(&[1.0, 0.0], &filter, 3, TIME).unwrap(); + let near: serde_json::Value = serde_json::from_str(&searching.seen.lock().unwrap()[0].2).unwrap(); + assert_eq!(near["filter"]["must"][0], serde_json::json!({ "key": "corpus", "match": { "value": "chat" } })); + assert_eq!(near["filter"]["must"][1]["range"], serde_json::json!({ "gte": 5 })); + assert_eq!(near["limit"], 3); +} + +/// A new source collection indexes `corpus` as a keyword and both times as +/// integers, which the range filters and ordered scroll need. +#[test] +fn qdrant_recreate_sources_creates_the_corpus_and_time_indexes() { + use crate::sources::{CORPUS, SourceIndex, SourcesMeta}; + let building = stub(&[("DELETE /collections/", 200, "{}"), ("PUT /collections/", 200, r#"{"result":true}"#)]); + let meta = SourcesMeta { managed_by: MANAGED_BY.into(), corpus: CORPUS.into(), model: "m".into(), model_digest: "d".into(), dimensions: 4, chunker_version: 1 }; + QdrantIndex::new(&building.url).recreate_sources(&meta).unwrap(); + let seen = building.seen.lock().unwrap(); + let indexed: Vec<(String, String)> = seen + .iter() + .filter(|(_, path, _)| path.contains("/index")) + .map(|(_, _, body)| { + let body: serde_json::Value = serde_json::from_str(body).unwrap(); + (body["field_name"].as_str().unwrap().to_owned(), body["field_schema"].as_str().unwrap().to_owned()) + }) + .collect(); + let has = |field: &str, schema: &str| indexed.iter().any(|(f, s)| f == field && s == schema); + assert!(has("corpus", "keyword"), "{indexed:?}"); + assert!(has("ts_start_ms", "integer") && has("ts_end_ms", "integer"), "{indexed:?}"); +} diff --git a/crates/huginn-daemon/src/sources/mod.rs b/crates/huginn-daemon/src/sources/mod.rs index 24a16c2..7a76ca3 100644 --- a/crates/huginn-daemon/src/sources/mod.rs +++ b/crates/huginn-daemon/src/sources/mod.rs @@ -1,5 +1,7 @@ -//! The source index: the public GameCult forge repos' code and docs as chunks -//! in one Qdrant collection, `huginn_sources`, for semantic search. +//! The source index: the public GameCult forge repos' code and docs, and the +//! Minds' chat rooms, as chunks in one Qdrant collection, `huginn_sources`, +//! for semantic search. Each point's `corpus` payload (`forge` or `chat`) keeps +//! the two apart; a forge search filters `forge` and a chat search `chat`. //! //! Two truths, one writer. The forge is the truth of which repos exist and what //! they hold; the collection is derived from it and can be deleted whenever. @@ -30,7 +32,7 @@ use std::path::Path; use std::time::{Duration, Instant}; use anyhow::{Context, Result, bail, ensure}; -use huginn_mind::{SOURCE_TOP_K_MAX, SourceHit, SourceHits, SourceModel, SourceQuery}; +use huginn_mind::{ChatHit, ChatHits, ChatMessage, ChatQuery, SOURCE_TOP_K_MAX, SourceHit, SourceHits, SourceModel, SourceQuery}; use crate::index::{BATCH, BULK_EMBED_TIMEOUT, Budget, Embedder, INDEX_CALL_TIMEOUT, MANAGED_BY, ModelIdentity, point_id}; use chunk::{CHUNKER_VERSION, embedded_text}; @@ -43,8 +45,19 @@ pub const COLLECTION: &str = "huginn_sources"; /// agent reads first to learn what the repo is. The only owner of the list. pub const FRONT_PAGE_FILES: [&str; 6] = ["README.md", "README", "readme.md", "README.markdown", "CLAUDE.md", "AGENTS.md"]; -/// What the collection says its corpus is. -pub const CORPUS: &str = "forge:GameCult"; +/// What the collection says its corpus is: the set of corpora it holds, each +/// point's `corpus` payload naming which (`forge` or `chat`). A collection +/// labelled with any other set is rebuilt. +pub const CORPUS: &str = "forge:GameCult+chat:rooms"; + +/// The `corpus` payload of a forge point, and of a chat point. +pub const CORPUS_FORGE: &str = "forge"; +pub const CORPUS_CHAT: &str = "chat"; + +/// The instruction Qwen3 embedding models take in front of a question about +/// past conversation. +pub const CHAT_QUERY_INSTRUCTION: &str = + "Given a question about past conversation in chat rooms, retrieve the chat passages that answer it"; /// The instruction Qwen3 embedding models take in front of a query about code. pub const SOURCE_QUERY_INSTRUCTION: &str = @@ -111,6 +124,58 @@ pub struct SourcePoint { pub payload: SourcePayload, } +/// What travels beside a chat chunk's vector: consecutive messages of one +/// room, never spanning rooms. The single owner of the chat point's shape; the +/// ingest job is its only writer (through `upsert_chat`) and nothing else of the +/// payload is read. `messages` are the display truth and hold, in order, at +/// least one message. The chunk's span, its first and last event ids and its +/// `event_ids` are derived from them by the codec, so they cannot disagree. +/// `text` is what was embedded: the room's name, then `time sender: body` per +/// message. +#[derive(Clone, Debug, PartialEq)] +pub struct ChatPayload { + pub room_id: String, + /// Absent when the room has no canonical alias. + pub room_alias: Option, + pub messages: Vec, + pub text: String, +} + +#[derive(Clone, Debug, PartialEq)] +pub struct ChatPoint { + pub id: String, + pub vector: Vec, + pub payload: ChatPayload, +} + +/// One chat chunk a vector search returned and how near it scored. +#[derive(Clone, Debug, PartialEq)] +pub struct ChatRow { + pub score: f32, + pub payload: ChatPayload, +} + +/// Which chat chunks a read keeps; the selectors compose as AND. `rooms` are +/// any-of room ids (`!`) or aliases (`#`); the span keeps chunks whose own +/// span overlaps [`after_ms`, `before_ms`]. +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub struct ChatFilter { + pub rooms: Vec, + pub after_ms: Option, + pub before_ms: Option, +} + +impl ChatFilter { + /// Whether a chunk of `room_id`/`room_alias` spanning [`start`, `end`] + /// passes. The fake index and the tests read the filter through this; the + /// Qdrant index states the same thing as conditions. + pub fn keeps(&self, room_id: &str, room_alias: Option<&str>, start: u64, end: u64) -> bool { + (self.rooms.is_empty() || self.rooms.iter().any(|room| room == room_id || Some(room.as_str()) == room_alias)) + && self.after_ms.is_none_or(|after| end >= after) + && self.before_ms.is_none_or(|before| start <= before) + } +} + /// One chunk a vector search returned and how near it scored. #[derive(Clone, Debug, PartialEq)] pub struct SourceRow { @@ -124,7 +189,9 @@ pub struct SourceRow { pub trait SourceIndex { fn describe_sources(&mut self, within: Duration) -> Result; /// Drops the collection if it exists and creates it empty with this - /// metadata, no cursors, and keyword indexes on `repo` and `path`. + /// metadata, no cursors, keyword indexes on `repo`, `path`, `corpus`, + /// `room_id`, `room_alias` and `event_ids`, and integer indexes on + /// `ts_start_ms` and `ts_end_ms`. fn recreate_sources(&mut self, meta: &SourcesMeta) -> Result<()>; /// Replaces the cursors object whole. fn set_cursors(&mut self, cursors: &BTreeMap) -> Result<()>; @@ -137,6 +204,15 @@ pub trait SourceIndex { /// The `limit` nearest chunks, best first, in `repo` alone when given, and /// among the `FRONT_PAGE_FILES` alone when `front`. fn search_sources(&mut self, vector: &[f32], repo: Option<&str>, front: bool, limit: u32, within: Duration) -> Result>; + /// Writes chat chunks, acknowledged before it returns. The ingest job's + /// alone: the daemon never calls it. A chunk with no messages is refused. + fn upsert_chat(&mut self, points: &[ChatPoint]) -> Result<()>; + /// The `limit` chat chunks nearest `vector`, best first, among those + /// `filter` keeps. Reads only `corpus=chat` points; the daemon's. + fn chat_nearest(&mut self, vector: &[f32], filter: &ChatFilter, limit: u32, within: Duration) -> Result>; + /// 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>; } /// A repo the forge lists as indexable. @@ -476,6 +552,10 @@ pub fn fuse(lists: Vec>, key: impl Fn(&T) -> String, k: u32, top_k: us /// of one, trimmed, and each distinct text once, in the order asked. fn query_texts(query: &SourceQuery) -> Vec<&str> { let asked: Vec<&str> = if query.texts.is_empty() { vec![query.text.0.as_str()] } else { query.texts.iter().map(|text| text.0.as_str()).collect() }; + distinct_texts(asked) +} + +fn distinct_texts(asked: Vec<&str>) -> Vec<&str> { let mut distinct: Vec<&str> = Vec::with_capacity(asked.len()); for text in asked { if !distinct.iter().any(|seen| seen.trim() == text.trim()) { @@ -485,16 +565,14 @@ fn query_texts(query: &SourceQuery) -> Vec<&str> { distinct } -/// A source search, which a daemon answers from the collection as it finds it: -/// it embeds the query with its own embedder, and refuses if the collection -/// was built under another model, because vectors of two models do not compare. -pub fn search(embedder: &mut E, index: &mut S, query: &SourceQuery, within: Duration) -> Result { - let budget = Budget::new(within); - let built = match index.describe_sources(budget.left()?).context("describing the source collection")? { - DescribedSources::Absent => bail!("source index not built"), - DescribedSources::Unlabelled => bail!("source index not built: {COLLECTION} carries no Huginn metadata"), - DescribedSources::Labelled { meta, .. } => meta, - }; +/// The label of the collection a daemon is about to read, if it was built +/// under the model this daemon embeds with. Absent, unlabelled and other-model +/// collections are loud refusals: vectors of two models do not compare, and an +/// empty page is not an answer. A collection built for another corpus set (an +/// old forge-only one) is refused the same way until the rebuild lands, since +/// it would answer a chat search with an empty page. +fn readable_under_this_model(embedder: &mut E, index: &mut S, budget: &Budget) -> Result<(SourcesMeta, ModelIdentity)> { + let built = built_label(index, budget)?; let identity = embedder.model_identity(budget.left()?).context("reading the embedding model's identity")?; ensure!( built.model == identity.name && built.model_digest == identity.digest && built.dimensions == identity.dimensions, @@ -506,6 +584,30 @@ pub fn search(embedder: &mut E, index: &mut S, quer identity.digest, identity.dimensions ); + Ok((built, identity)) +} + +fn built_label(index: &mut S, budget: &Budget) -> Result { + match index.describe_sources(budget.left()?).context("describing the source collection")? { + DescribedSources::Absent => bail!("source index not built"), + DescribedSources::Unlabelled => bail!("source index not built: {COLLECTION} carries no Huginn metadata"), + DescribedSources::Labelled { meta, .. } => { + ensure!( + meta.corpus == CORPUS, + "the source index is stale: it was built for corpus {} and this daemon reads {CORPUS}; the rebuild is pending", + meta.corpus + ); + Ok(meta) + } + } +} + +/// A source search, which a daemon answers from the collection as it finds it: +/// it embeds the query with its own embedder, and refuses if the collection +/// was built under another model, because vectors of two models do not compare. +pub fn search(embedder: &mut E, index: &mut S, query: &SourceQuery, within: Duration) -> Result { + let budget = Budget::new(within); + let (built, identity) = readable_under_this_model(embedder, index, &budget)?; let topics = query_texts(query); let prompts: Vec = topics.iter().map(|topic| format!("Instruct: {SOURCE_QUERY_INSTRUCTION}\nQuery: {topic}")).collect(); let vectors = embedder.embed(&prompts, budget.left()?).context("embedding the query")?; @@ -539,3 +641,68 @@ pub fn search(embedder: &mut E, index: &mut S, quer .collect(), }) } + +/// A chat search. With texts, the chunks nearest them among those the room and +/// span selectors keep, several texts fused by `fuse`; with none, a listing of +/// what the selectors keep, oldest first, `top_k` of it and `truncated` saying +/// more exist (no embedding, so the model's identity is not consulted, though +/// an absent collection is still refused). A hit outside the span is not +/// returned and a boundary hit loses its messages outside [after, before], its +/// span and first and last ids recomputed from what is left. Never writes. +pub fn search_chat(embedder: &mut E, index: &mut S, query: &ChatQuery, within: Duration) -> Result { + let budget = Budget::new(within); + let filter = ChatFilter { rooms: query.rooms.iter().map(|room| room.0.clone()).collect(), after_ms: query.after_ms, before_ms: query.before_ms }; + let top_k = query.top_k as usize; + let topics = distinct_texts(query.texts.iter().map(|text| text.0.as_str()).collect()); + let (built, rows, truncated) = if topics.is_empty() { + let built = built_label(index, &budget)?; + let mut held = index.chat_scroll(&filter, query.top_k + 1, budget.left()?)?; + let truncated = held.len() > top_k; + held.truncate(top_k); + (built, held.into_iter().map(|payload| (None, payload)).collect::>(), truncated) + } else { + let (built, identity) = readable_under_this_model(embedder, index, &budget)?; + let prompts: Vec = topics.iter().map(|topic| format!("Instruct: {CHAT_QUERY_INSTRUCTION}\nQuery: {topic}")).collect(); + let vectors = embedder.embed(&prompts, budget.left()?).context("embedding the query")?; + ensure!(vectors.len() == prompts.len(), "the embedder returned {} vectors for {} query texts", vectors.len(), prompts.len()); + let depth = if vectors.len() == 1 { query.top_k } else { FUSE_DEPTH }; + let mut lists = Vec::with_capacity(vectors.len()); + for vector in &vectors { + ensure!(vector.len() == identity.dimensions as usize, "the embedder returned a query vector of {} where the collection holds {}", vector.len(), identity.dimensions); + lists.push(index.chat_nearest(vector, &filter, depth, budget.left()?)?); + } + let rows: Vec = if lists.len() == 1 { + lists.remove(0) + } else { + let key = |row: &ChatRow| format!("{}\0{}", row.payload.room_id, row.payload.messages.first().map_or("", |message| message.event_id.as_str())); + fuse(lists, key, FUSE_K, top_k).into_iter().map(|(score, row)| ChatRow { score, ..row }).collect() + }; + (built, rows.into_iter().map(|row| (Some(row.score), row.payload)).collect(), false) + }; + Ok(ChatHits { + indexed_under: SourceModel { name: built.model, digest: built.model_digest }, + truncated, + hits: rows.into_iter().filter_map(|(score, payload)| chat_hit(score, payload, &filter)).collect(), + }) +} + +/// A held chunk as a hit: only the messages inside the asked span, the span and +/// ids those messages make, and nothing at all when none is inside. +fn chat_hit(score: Option, payload: ChatPayload, filter: &ChatFilter) -> Option { + let messages: Vec = payload + .messages + .into_iter() + .filter(|message| filter.after_ms.is_none_or(|after| message.ts_ms >= after) && filter.before_ms.is_none_or(|before| message.ts_ms <= before)) + .collect(); + let (first, last) = (messages.first()?, messages.last()?); + Some(ChatHit { + room_id: payload.room_id, + room_alias: payload.room_alias, + first_event_id: first.event_id.clone(), + last_event_id: last.event_id.clone(), + ts_start_ms: messages.iter().map(|message| message.ts_ms).min()?, + ts_end_ms: messages.iter().map(|message| message.ts_ms).max()?, + score, + messages, + }) +} diff --git a/crates/huginn-daemon/src/sources/tests.rs b/crates/huginn-daemon/src/sources/tests.rs index bda5ce9..97a2bb3 100644 --- a/crates/huginn-daemon/src/sources/tests.rs +++ b/crates/huginn-daemon/src/sources/tests.rs @@ -7,7 +7,7 @@ use std::process::Command; use anyhow::Result; use huginn_mind::eureka_pipeline::{Line, Short}; -use huginn_mind::SourceQuery; +use huginn_mind::{ChatHits, ChatQuery, SourceQuery}; use super::*; use crate::index::fakes::{FakeEmbedder, FakeSources}; @@ -755,3 +755,222 @@ fn a_long_brief_of_eight_topics_reaches_the_chunk_only_the_last_is_near() { assert!(paths(&found).contains(&"chunk-7.rs"), "{:?}", paths(&found)); assert_eq!(paths(&found).len(), 8); } + +// ---- chat search ---- + +fn msg(event_id: &str, ts_ms: u64) -> ChatMessage { + ChatMessage { event_id: event_id.into(), sender: "@a:gamecult.org".into(), ts_ms, body: format!("body of {event_id}") } +} + +/// A chunk in `room` (with `alias`) of messages `(id, ts)`, near `axis`. +fn chat_point(room: &str, alias: Option<&str>, axis: usize, dimensions: u32, messages: &[(&str, u64)]) -> ChatPoint { + let messages: Vec = messages.iter().map(|(id, ts)| msg(id, *ts)).collect(); + ChatPoint { + id: point_id(&format!("{room}{}", messages[0].event_id)), + vector: (0..dimensions as usize).map(|at| if at == axis { 1.0 } else { 0.0 }).collect(), + payload: ChatPayload { room_id: room.into(), room_alias: alias.map(str::to_owned), messages, text: format!("chunk near topic-{axis}") }, + } +} + +fn chat_query(texts: &[&str], rooms: &[&str], after: Option, before: Option, top_k: u32) -> ChatQuery { + ChatQuery { + texts: texts.iter().map(|text| Line((*text).into())).collect(), + rooms: rooms.iter().map(|room| Short((*room).into())).collect(), + after_ms: after, + before_ms: before, + top_k, + } +} + +/// A labelled collection that also holds the chat chunks given. Two rooms: +/// `!a` (alias `#swarm`), `!b` (no alias). +fn chat_rig(points: Vec) -> (Rig, TopicEmbedder) { + let (mut rig, embedder) = topic_rig(4); + rig.index.state.lock().unwrap().points.clear(); + rig.index.upsert_chat(&points).unwrap(); + (rig, embedder) +} + +fn ids(hits: &ChatHits) -> Vec<&str> { + hits.hits.iter().map(|hit| hit.first_event_id.as_str()).collect() +} + +const WITHIN: Duration = Duration::from_secs(5); + +fn three_rooms() -> Vec { + vec![ + chat_point("!a:x", Some("#swarm:x"), 0, 4, &[("$a1", 100), ("$a2", 200)]), + chat_point("!a:x", Some("#swarm:x"), 1, 4, &[("$a3", 300), ("$a4", 400)]), + chat_point("!b:x", None, 0, 4, &[("$b1", 150)]), + ] +} + +#[test] +fn a_room_filter_by_id_and_by_alias_returns_only_that_room() { + let (mut rig, mut embedder) = chat_rig(three_rooms()); + let by_alias = search_chat(&mut embedder, &mut rig.index, &chat_query(&[], &["#swarm:x"], None, None, 10), WITHIN).unwrap(); + assert_eq!(ids(&by_alias), ["$a1", "$a3"]); + let by_id = search_chat(&mut embedder, &mut rig.index, &chat_query(&[], &["!b:x"], None, None, 10), WITHIN).unwrap(); + assert_eq!(ids(&by_id), ["$b1"], "a room with no alias is reachable by id"); + assert_eq!(by_id.hits[0].room_alias, None); + let both = search_chat(&mut embedder, &mut rig.index, &chat_query(&[], &["#swarm:x", "!b:x"], None, None, 10), WITHIN).unwrap(); + assert_eq!(ids(&both), ["$a1", "$b1", "$a3"], "any-of, oldest first"); + let nobody = search_chat(&mut embedder, &mut rig.index, &chat_query(&[], &["#nobody:x"], None, None, 10), WITHIN).unwrap(); + assert!(nobody.hits.is_empty() && !nobody.truncated, "an unknown alias is zero hits, not an error"); +} + +#[test] +fn a_time_span_returns_the_overlapping_chunks_and_trims_boundary_messages() { + let (mut rig, mut embedder) = chat_rig(three_rooms()); + let hits = search_chat(&mut embedder, &mut rig.index, &chat_query(&[], &["#swarm:x"], Some(150), Some(350), 10), WITHIN).unwrap(); + assert_eq!(ids(&hits), ["$a2", "$a3"], "the boundary chunks keep only in-span messages and first ids are recomputed"); + let (first, second) = (&hits.hits[0], &hits.hits[1]); + assert_eq!((first.ts_start_ms, first.ts_end_ms, first.last_event_id.as_str(), first.messages.len()), (200, 200, "$a2", 1)); + assert_eq!((second.ts_start_ms, second.ts_end_ms, second.last_event_id.as_str(), second.messages.len()), (300, 300, "$a3", 1)); + let none = search_chat(&mut embedder, &mut rig.index, &chat_query(&[], &["#swarm:x"], Some(1000), None, 10), WITHIN).unwrap(); + assert!(none.hits.is_empty(), "a chunk wholly outside is absent"); +} + +#[test] +fn list_mode_is_oldest_first_truncated_when_more_exist_and_needs_no_embed() { + let (mut rig, mut embedder) = chat_rig(three_rooms()); + let page = search_chat(&mut embedder, &mut rig.index, &chat_query(&[], &[], None, None, 2), WITHIN).unwrap(); + assert_eq!(ids(&page), ["$a1", "$b1"]); + assert!(page.truncated); + assert!(embedder.0.state.lock().unwrap().embedded.is_empty(), "no texts, no embed call"); + // Within one room chunks never overlap, so the last hit's end + 1 continues without a repeat. + let first = search_chat(&mut embedder, &mut rig.index, &chat_query(&[], &["!a:x"], None, None, 1), WITHIN).unwrap(); + assert_eq!((ids(&first), first.truncated), (vec!["$a1"], true)); + let rest = search_chat(&mut embedder, &mut rig.index, &chat_query(&[], &["!a:x"], Some(first.hits[0].ts_end_ms + 1), None, 1), WITHIN).unwrap(); + assert_eq!((ids(&rest), rest.truncated), (vec!["$a3"], false)); +} + +#[test] +fn texts_rank_by_meaning_and_compose_with_room_and_span() { + let (mut rig, mut embedder) = chat_rig(three_rooms()); + let near = search_chat(&mut embedder, &mut rig.index, &chat_query(&["topic-0"], &[], None, None, 5), WITHIN).unwrap(); + assert_eq!(near.hits.len(), 3); + assert!(near.hits[0].score.unwrap() > near.hits[2].score.unwrap()); + assert_eq!(near.hits[2].first_event_id, "$a3", "the far chunk is last"); + let narrowed = search_chat(&mut embedder, &mut rig.index, &chat_query(&["topic-0"], &["!a:x"], Some(250), None, 5), WITHIN).unwrap(); + assert_eq!(ids(&narrowed), ["$a3"], "a near chunk in another room or outside the span never appears"); + let fused = search_chat(&mut embedder, &mut rig.index, &chat_query(&["topic-0", "topic-1"], &["!a:x"], None, None, 5), WITHIN).unwrap(); + assert_eq!(fused.hits.len(), 2); + assert!(fused.hits[0].score.unwrap() > 0.0 && fused.hits[0].score != Some(1.0), "two texts carry the fused reciprocal-rank score"); + assert_eq!(embedder.0.state.lock().unwrap().embedded.last().unwrap().len(), 2, "one embed call for two texts"); +} + +#[test] +fn every_hit_cites_room_event_ids_and_timestamps() { + let (mut rig, mut embedder) = chat_rig(three_rooms()); + for texts in [&[][..], &["topic-0"][..]] { + let hits = search_chat(&mut embedder, &mut rig.index, &chat_query(texts, &[], None, None, 10), WITHIN).unwrap(); + assert!(!hits.hits.is_empty()); + for hit in &hits.hits { + assert!(hit.room_id.starts_with('!') && !hit.first_event_id.is_empty() && !hit.last_event_id.is_empty()); + assert!(hit.ts_start_ms <= hit.ts_end_ms && !hit.messages.is_empty()); + assert_eq!((hit.messages[0].event_id.as_str(), hit.messages.last().unwrap().event_id.as_str()), (hit.first_event_id.as_str(), hit.last_event_id.as_str())); + for message in &hit.messages { + assert!(message.event_id.starts_with('$') && message.sender.starts_with('@') && message.ts_ms > 0); + } + } + } +} + +#[test] +fn a_forge_search_never_returns_a_chat_point_and_a_chat_search_never_a_forge_point() { + let (mut rig, mut embedder) = topic_rig(4); + rig.index.upsert_chat(&three_rooms()).unwrap(); + let forge = search(&mut embedder, &mut rig.index, &queries(&["topic-0"], 10), WITHIN).unwrap(); + assert_eq!(forge.hits.len(), 4, "only the four forge chunks"); + let chat = search_chat(&mut embedder, &mut rig.index, &chat_query(&["topic-0"], &[], None, None, 10), WITHIN).unwrap(); + assert_eq!(chat.hits.len(), 3, "only the three chat chunks"); + let listed = search_chat(&mut embedder, &mut rig.index, &chat_query(&[], &[], None, None, 10), WITHIN).unwrap(); + assert_eq!(listed.hits.len(), 3); +} + +#[test] +fn an_absent_unlabelled_or_other_model_collection_is_a_loud_refusal_for_chat() { + let mut rig = Rig::new(); + let mut embedder = TopicEmbedder(FakeEmbedder::new("d1")); + for texts in [&[][..], &["topic-0"][..]] { + let error = search_chat(&mut embedder, &mut rig.index, &chat_query(texts, &[], None, None, 3), WITHIN).unwrap_err(); + assert_eq!(format!("{error:#}"), "source index not built"); + } + rig.index.plant(DescribedSources::Unlabelled); + assert!(search_chat(&mut embedder, &mut rig.index, &chat_query(&[], &[], None, None, 3), WITHIN).is_err()); + let (mut rig, _) = chat_rig(three_rooms()); + let mut foreign = TopicEmbedder(FakeEmbedder::new("d2")); + let error = format!("{:#}", search_chat(&mut foreign, &mut rig.index, &chat_query(&["topic-0"], &[], None, None, 3), WITHIN).unwrap_err()); + assert!(error.contains("built under model d1") && error.contains("embeds under model d2"), "{error}"); +} + +/// An old forge-only collection (labelled with the old corpus) is rebuilt once +/// under the new corpus, and a second refresh leaves it. +#[test] +fn an_old_forge_only_collection_rebuilds_once() { + let remote = Remote::new(); + remote.commit(&[("a.rs", b"one\n")], &[]); + let mut rig = Rig::new(); + let mut forge = FakeForge::of(&[("Demo", &remote)]); + rig.converged(&mut forge); + let mut old = match rig.index.state.lock().unwrap().described.clone() { + Some(DescribedSources::Labelled { meta, cursors }) => (meta, cursors), + other => panic!("{other:?}"), + }; + assert_eq!(old.0.corpus, "forge:GameCult+chat:rooms"); + old.0.corpus = "forge:GameCult".into(); + rig.index.plant(DescribedSources::Labelled { meta: old.0, cursors: old.1 }); + let before = rig.index.state.lock().unwrap().recreated; + rig.converged(&mut forge); + rig.converged(&mut forge); + assert_eq!(rig.index.state.lock().unwrap().recreated, before + 1); +} + +#[test] +fn the_chat_read_path_holds_no_discord_matrix_or_second_fuse() { + let source = include_str!("mod.rs"); + let read = &source[source.find("pub fn search_chat").unwrap()..]; + for banned in ["discord", "puppet", "bridge", "matrix", "synapse", "access_token"] { + assert!(!read.to_ascii_lowercase().contains(banned), "{banned} in the chat read path"); + } + assert_eq!(source.matches("pub fn fuse").count(), 1); +} + +/// An old forge-only collection answers both searches with the stale-index +/// error until the rebuild lands, not with an empty page. +#[test] +fn an_old_corpus_collection_is_refused_until_it_is_rebuilt() { + let (mut rig, mut embedder) = chat_rig(three_rooms()); + let Some(DescribedSources::Labelled { mut meta, cursors }) = rig.index.state.lock().unwrap().described.clone() else { panic!() }; + meta.corpus = "forge:GameCult".into(); + rig.index.plant(DescribedSources::Labelled { meta, cursors }); + let error = format!("{:#}", search(&mut embedder, &mut rig.index, &query("one", 3, None), WITHIN).unwrap_err()); + assert!(error.contains("stale") && error.contains("forge:GameCult") && error.contains("rebuild is pending"), "{error}"); + for texts in [&[][..], &["topic-0"][..]] { + let error = format!("{:#}", search_chat(&mut embedder, &mut rig.index, &chat_query(texts, &[], None, None, 3), WITHIN).unwrap_err()); + assert!(error.contains("stale") && error.contains("rebuild is pending"), "{error}"); + } +} + +/// A message exactly on `after` or `before` is in the span, and so is a chunk +/// whose end is exactly `after` or whose start is exactly `before`. +#[test] +fn the_span_bounds_are_inclusive_for_chunks_and_for_messages() { + let (mut rig, mut embedder) = chat_rig(three_rooms()); + let mut hits = |after, before| { + let found = search_chat(&mut embedder, &mut rig.index, &chat_query(&[], &["#swarm:x"], after, before, 10), WITHIN).unwrap(); + found.hits.iter().map(|hit| (hit.first_event_id.clone(), hit.messages.len())).collect::>() + }; + let owned = |pairs: &[(&str, usize)]| pairs.iter().map(|(id, n)| ((*id).to_owned(), *n)).collect::>(); + // Chunk $a1..$a2 ends at 200: after=200 keeps it, trimmed to the message on the bound. + assert_eq!(hits(Some(200), None), owned(&[("$a2", 1), ("$a3", 2)])); + // Chunk $a3..$a4 starts at 300: before=300 keeps it, trimmed to the message on the bound. + assert_eq!(hits(None, Some(300)), owned(&[("$a1", 2), ("$a3", 1)])); + // Both bounds on one message. + assert_eq!(hits(Some(200), Some(200)), owned(&[("$a2", 1)])); + + let filter = ChatFilter { rooms: vec![], after_ms: Some(200), before_ms: Some(300) }; + assert!(filter.keeps("!a:x", None, 100, 200) && filter.keeps("!a:x", None, 300, 400)); + assert!(!filter.keeps("!a:x", None, 100, 199) && !filter.keeps("!a:x", None, 301, 400)); +} diff --git a/crates/huginn-mind/src/lib.rs b/crates/huginn-mind/src/lib.rs index 3ae57fc..abcd46e 100644 --- a/crates/huginn-mind/src/lib.rs +++ b/crates/huginn-mind/src/lib.rs @@ -72,7 +72,7 @@ pub use store::{MindStore, OwnedRedbMessagePackBackingStore}; pub use wire::{ DeferredAnswer, HuginnMindRequest, HuginnMindResponse, IndexStatus, MAX_DEFERRED_BODY_BYTES, MIND_REQUEST_SCHEMA, MIND_REQUEST_SCHEMA_JSON, MIND_RESPONSE_SCHEMA, MIND_RESPONSE_SCHEMA_JSON, MIND_SERVICE_ID, MindStatus, PersonaStatus, - ACTIVITY_REPOS_MAX, OpenPr, PrAction, RepoActivity, RepoChange, RepoEvent, RepoMoved, SOURCE_TEXTS_MAX, SOURCE_TOP_K_MAX, SourceHit, SourceHits, SourceModel, SourceQuery, + ACTIVITY_REPOS_MAX, CHAT_ROOMS_MAX, ChatHit, ChatHits, ChatMessage, ChatQuery, OpenPr, PrAction, RepoActivity, RepoChange, RepoEvent, RepoMoved, SOURCE_TEXTS_MAX, SOURCE_TOP_K_MAX, SourceHit, SourceHits, SourceModel, SourceQuery, }; /// The leaf, whole, through the one crate that pins its rev: the daemon and diff --git a/crates/huginn-mind/src/wire.rs b/crates/huginn-mind/src/wire.rs index 3f64def..7952f07 100644 --- a/crates/huginn-mind/src/wire.rs +++ b/crates/huginn-mind/src/wire.rs @@ -92,6 +92,10 @@ pub enum HuginnMindRequest { /// docs, which no mind holds. The daemon embeds the text and asks the /// index it does not write. SearchSources { instance: Slug, query: SourceQuery }, + /// Search of the chat index: chunks of the rooms the Minds are meant to + /// have, by room, by time span and by meaning. Read-only, like + /// `SearchSources`, and queued with it. + SearchChat { instance: Slug, query: ChatQuery }, /// What others did in `repos` on the forge since the `after` cursor: the /// daemon reads the forge and no document of the mind, and answers /// `Refused(Unavailable)` when it has no forge or the forge does not answer @@ -270,6 +274,96 @@ pub struct SourceHits { pub hits: Vec, } +/// The most rooms one chat search names. +pub const CHAT_ROOMS_MAX: usize = 16; + +/// A chat search. The selectors compose as AND. `rooms` are any-of: a value +/// beginning `!` is a room id, `#` a canonical alias. `after_ms` and +/// `before_ms` (Unix milliseconds) keep the chunks whose span overlaps +/// [after, before]. `texts` (none to `SOURCE_TEXTS_MAX`) is the meaning match, +/// several fused as in a source search; with none the search is a listing: +/// chunks oldest first, at most `top_k`, `ChatHits.truncated` saying more +/// exist, and the caller narrows with `after_ms` = the last hit's +/// `ts_end_ms + 1`. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] +pub struct ChatQuery { + pub texts: Vec, + pub rooms: Vec, + pub after_ms: Option, + pub before_ms: Option, + pub top_k: u32, +} + +impl ChatQuery { + /// The door. A refusal names the field. + pub fn check(&self) -> Result<(), MindRefusal> { + let refused = |field: &str, value: Option, message: &str| MindRefusal::SelectionInvalid { + field: field.into(), + value, + message: message.into(), + }; + if self.texts.len() > SOURCE_TEXTS_MAX { + return Err(refused("texts", Some(self.texts.len().to_string()), "a chat search takes 0 to 8 texts")); + } + if self.texts.iter().any(|text| text.0.trim().is_empty()) { + return Err(refused("texts", None, "every text of a chat search says something")); + } + if self.rooms.len() > CHAT_ROOMS_MAX { + return Err(refused("rooms", Some(self.rooms.len().to_string()), "a chat search names at most 16 rooms")); + } + if let Some(room) = self.rooms.iter().find(|room| !(room.0.starts_with('!') || room.0.starts_with('#')) || room.0.len() < 2) { + return Err(refused("rooms", Some(room.0.clone()), "a room is a room id beginning `!` or a canonical alias beginning `#`")); + } + if let (Some(after), Some(before)) = (self.after_ms, self.before_ms) + && before < after + { + return Err(refused("before_ms", Some(before.to_string()), "before_ms is not earlier than after_ms")); + } + if !(1..=SOURCE_TOP_K_MAX).contains(&self.top_k) { + return Err(refused("top_k", Some(self.top_k.to_string()), "a chat search asks for 1 to 50 hits")); + } + Ok(()) + } +} + +/// One message of a chunk, as the room held it. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] +pub struct ChatMessage { + pub event_id: String, + /// The sender's Matrix id. + pub sender: String, + pub ts_ms: u64, + pub body: String, +} + +/// One chunk of consecutive messages in one room. Every citation is +/// required: only the room's alias may be absent, because a room may have +/// none. The span and the first and last ids are those of `messages` as +/// returned, after any trimming to the asked span. +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize, JsonSchema)] +pub struct ChatHit { + pub room_id: String, + pub room_alias: Option, + pub first_event_id: String, + pub last_event_id: String, + pub ts_start_ms: u64, + pub ts_end_ms: u64, + /// Cosine similarity for one text, the reciprocal-rank sum for several, + /// none in a listing. + pub score: Option, + pub messages: Vec, +} + +/// The answer to a chat search: oldest first in a listing, nearest first with +/// texts. +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize, JsonSchema)] +pub struct ChatHits { + pub indexed_under: SourceModel, + /// A listing had more chunks than `top_k`. + pub truncated: bool, + pub hits: Vec, +} + impl HuginnMindRequest { /// The envelope's `operation`. pub fn operation(&self) -> &'static str { @@ -281,6 +375,7 @@ impl HuginnMindRequest { Self::PersonaGet { .. } => "persona_get", Self::PersonaPut { .. } => "persona_put", Self::SearchSources { .. } => "search_sources", + Self::SearchChat { .. } => "search_chat", Self::StewardedActivity { .. } => "stewarded_activity", Self::RootsTouched { .. } => "roots_touched", } @@ -296,6 +391,7 @@ impl HuginnMindRequest { | Self::PersonaGet { instance } | Self::PersonaPut { instance, .. } | Self::SearchSources { instance, .. } + | Self::SearchChat { instance, .. } | Self::StewardedActivity { instance, .. } | Self::RootsTouched { instance } => Some(instance), } @@ -315,6 +411,7 @@ pub enum HuginnMindResponse { Persona(Option), PersonaPut(PersonaPutOutcome), Sources(SourceHits), + Chat(ChatHits), Activity(RepoActivity), RootsTouched(Vec), /// The answer did not fit one send and is being held for the client to @@ -450,6 +547,10 @@ mod tests { instance: slug(INSTANCE), query: SourceQuery { text: Line::default(), texts: vec![Line("reconcile".into()), Line("fuse".into())], top_k: 5, repo: None, front: false }, }, + HuginnMindRequest::SearchChat { + instance: slug(INSTANCE), + query: ChatQuery { texts: vec![Line("release".into())], rooms: vec![Short("#swarm:gamecult.org".into())], after_ms: Some(1), before_ms: None, top_k: 5 }, + }, HuginnMindRequest::RootsTouched { instance: slug(INSTANCE) }, ] } @@ -500,6 +601,20 @@ mod tests { text: "x".into(), }], }), + HuginnMindResponse::Chat(ChatHits { + indexed_under: SourceModel { name: "m".into(), digest: "d".into() }, + truncated: false, + hits: vec![ChatHit { + room_id: "!r:gamecult.org".into(), + room_alias: None, + first_event_id: "$a".into(), + last_event_id: "$a".into(), + ts_start_ms: 1, + ts_end_ms: 1, + score: None, + messages: vec![ChatMessage { event_id: "$a".into(), sender: "@x:gamecult.org".into(), ts_ms: 1, body: "hi".into() }], + }], + }), HuginnMindResponse::RootsTouched(seeded().roots_touched().unwrap()), HuginnMindResponse::Deferred(DeferredAnswer { manifest: cultnet_rs::pack_content("huginn.mind_response", "package", "", "", "", b"body", 4) @@ -515,7 +630,7 @@ mod tests { #[test] fn the_wire_vocabulary_is_the_minds_methods_and_status_is_derived_from_the_response() { let names: Vec<&str> = requests().iter().map(|request| request.operation()).collect(); - assert_eq!(names, ["whoami", "admit", "view", "query", "persona_get", "persona_put", "search_sources", "search_sources", "roots_touched"]); + assert_eq!(names, ["whoami", "admit", "view", "query", "persona_get", "persona_put", "search_sources", "search_sources", "search_chat", "roots_touched"]); for request in requests() { let declared = request.instance().is_none(); assert_eq!(declared, matches!(request, HuginnMindRequest::Whoami), "{request:?}"); @@ -625,6 +740,33 @@ mod tests { assert_eq!((page.matched, page.as_of, page.scores), (0, 7, None)); } + /// A chat search is refused by field at the door, and the published + /// response marks every citation required. + #[test] + fn a_chat_query_is_refused_by_field_and_a_hit_cites_room_ids_and_times() { + let ok = ChatQuery { texts: vec![], rooms: vec![Short("!a:b".into()), Short("#c:b".into())], after_ms: Some(1), before_ms: Some(1), top_k: 50 }; + assert!(ok.check().is_ok()); + let field = |query: ChatQuery| match query.check() { + Err(MindRefusal::SelectionInvalid { field, .. }) => field, + other => panic!("{other:?}"), + }; + assert_eq!(field(ChatQuery { rooms: vec![Short("swarm".into())], ..ok.clone() }), "rooms"); + assert_eq!(field(ChatQuery { rooms: vec![Short("#".into())], ..ok.clone() }), "rooms"); + assert_eq!(field(ChatQuery { before_ms: Some(0), ..ok.clone() }), "before_ms"); + assert_eq!(field(ChatQuery { top_k: 0, ..ok.clone() }), "top_k"); + assert_eq!(field(ChatQuery { top_k: 51, ..ok.clone() }), "top_k"); + assert_eq!(field(ChatQuery { texts: vec![Line(" ".into())], ..ok.clone() }), "texts"); + assert_eq!(field(ChatQuery { texts: vec![Line("x".into()); 9], ..ok.clone() }), "texts"); + assert_eq!(field(ChatQuery { rooms: vec![Short("!a:b".into()); 17], ..ok }), "rooms"); + let response = serde_json::to_value(schemars::schema_for!(HuginnMindResponse)).unwrap(); + let required: Vec<&str> = response["$defs"]["ChatHit"]["required"].as_array().unwrap().iter().filter_map(|v| v.as_str()).collect(); + for field in ["room_id", "first_event_id", "last_event_id", "ts_start_ms", "ts_end_ms", "messages"] { + assert!(required.contains(&field), "{field} must be required"); + } + let message: Vec<&str> = response["$defs"]["ChatMessage"]["required"].as_array().unwrap().iter().filter_map(|v| v.as_str()).collect(); + assert_eq!(message, ["event_id", "sender", "ts_ms", "body"]); + } + /// The two files the catalog publishes are the derivation of the two /// types, byte for byte. #[test] diff --git a/docs/chat-retrieval-map.md b/docs/chat-retrieval-map.md index 3665cae..50b2fc6 100644 --- a/docs/chat-retrieval-map.md +++ b/docs/chat-retrieval-map.md @@ -144,3 +144,22 @@ What changed since the map was written: (`gamecult-chat:ruling:chat-index-minds-rooms`), Discord never. - A Mind that must remember DM content writes a memory citing the chat or room id as its source; the index is not where DM content is kept. + +## Status: read side (cut chat-search) + +Built: `huginn chat search [TEXT]... [--room R]... [--after T] [--before T] [--top-k N] [--json]` +over request `SearchChat` (`sources::search_chat`), listing oldest first without +texts and ranking by meaning with them. The chat point's shape is `ChatPayload` +(`sources/mod.rs`); every point carries `corpus` (`forge` or `chat`) and +`SourcesMeta.corpus` is `forge:GameCult+chat:rooms`. Not built: the ingest job +that writes chat points through `upsert_chat`, and the event-id window. Until the +ingest lands the collection holds no chat points and a chat search answers empty. +In a listing across rooms, the last hit's `ts_end_ms + 1` can repeat messages of +an earlier-starting longer chunk of another room; within one room it continues +exactly. + +Known limits (owner of the fix: cut chat-window). Hits are ordered by their +trimmed start, not by the stored chunk start, and a long chunk trimmed to a span +that crosses rooms can be listed again by the continuation `--after`. Behaviour +is deliberately unchanged here; the event-id window in cut chat-window replaces +the time cursor. diff --git a/schemas/cultnet/huginn.mind_request.v1.schema.json b/schemas/cultnet/huginn.mind_request.v1.schema.json index 5c3ea7a..233da8d 100644 --- a/schemas/cultnet/huginn.mind_request.v1.schema.json +++ b/schemas/cultnet/huginn.mind_request.v1.schema.json @@ -158,6 +158,31 @@ "SearchSources" ] }, + { + "description": "Search of the chat index: chunks of the rooms the Minds are meant to\nhave, by room, by time span and by meaning. Read-only, like\n`SearchSources`, and queued with it.", + "type": "object", + "properties": { + "SearchChat": { + "type": "object", + "properties": { + "instance": { + "$ref": "#/$defs/Slug" + }, + "query": { + "$ref": "#/$defs/ChatQuery" + } + }, + "required": [ + "instance", + "query" + ] + } + }, + "additionalProperties": false, + "required": [ + "SearchChat" + ] + }, { "description": "What others did in `repos` on the forge since the `after` cursor: the\ndaemon reads the forge and no document of the mind, and answers\n`Refused(Unavailable)` when it has no forge or the forge does not answer\nin time. The repos are the asker's to name (her stewardships); the\ninstance's own login is dropped from the answer.", "type": "object", @@ -272,6 +297,50 @@ "deletion_line" ] }, + "ChatQuery": { + "description": "A chat search. The selectors compose as AND. `rooms` are any-of: a value\nbeginning `!` is a room id, `#` a canonical alias. `after_ms` and\n`before_ms` (Unix milliseconds) keep the chunks whose span overlaps\n[after, before]. `texts` (none to `SOURCE_TEXTS_MAX`) is the meaning match,\nseveral fused as in a source search; with none the search is a listing:\nchunks oldest first, at most `top_k`, `ChatHits.truncated` saying more\nexist, and the caller narrows with `after_ms` = the last hit's\n`ts_end_ms + 1`.", + "type": "object", + "properties": { + "after_ms": { + "type": [ + "integer", + "null" + ], + "format": "uint64", + "minimum": 0 + }, + "before_ms": { + "type": [ + "integer", + "null" + ], + "format": "uint64", + "minimum": 0 + }, + "rooms": { + "type": "array", + "items": { + "$ref": "#/$defs/Short" + } + }, + "texts": { + "type": "array", + "items": { + "$ref": "#/$defs/Line" + } + }, + "top_k": { + "type": "integer", + "format": "uint32", + "minimum": 0 + } + }, + "required": [ + "texts", + "rooms", + "top_k" + ] + }, "ClaimOutcome": { "type": "string", "enum": [ diff --git a/schemas/cultnet/huginn.mind_response.v1.schema.json b/schemas/cultnet/huginn.mind_response.v1.schema.json index 70cbdf1..4adbc43 100644 --- a/schemas/cultnet/huginn.mind_response.v1.schema.json +++ b/schemas/cultnet/huginn.mind_response.v1.schema.json @@ -113,6 +113,18 @@ "Sources" ] }, + { + "type": "object", + "properties": { + "Chat": { + "$ref": "#/$defs/ChatHits" + } + }, + "additionalProperties": false, + "required": [ + "Chat" + ] + }, { "type": "object", "properties": { @@ -236,6 +248,110 @@ "deletion_line" ] }, + "ChatHit": { + "description": "One chunk of consecutive messages in one room. Every citation is\nrequired: only the room's alias may be absent, because a room may have\nnone. The span and the first and last ids are those of `messages` as\nreturned, after any trimming to the asked span.", + "type": "object", + "properties": { + "first_event_id": { + "type": "string" + }, + "last_event_id": { + "type": "string" + }, + "messages": { + "type": "array", + "items": { + "$ref": "#/$defs/ChatMessage" + } + }, + "room_alias": { + "type": [ + "string", + "null" + ] + }, + "room_id": { + "type": "string" + }, + "score": { + "description": "Cosine similarity for one text, the reciprocal-rank sum for several,\nnone in a listing.", + "type": [ + "number", + "null" + ], + "format": "float" + }, + "ts_end_ms": { + "type": "integer", + "format": "uint64", + "minimum": 0 + }, + "ts_start_ms": { + "type": "integer", + "format": "uint64", + "minimum": 0 + } + }, + "required": [ + "room_id", + "first_event_id", + "last_event_id", + "ts_start_ms", + "ts_end_ms", + "messages" + ] + }, + "ChatHits": { + "description": "The answer to a chat search: oldest first in a listing, nearest first with\ntexts.", + "type": "object", + "properties": { + "hits": { + "type": "array", + "items": { + "$ref": "#/$defs/ChatHit" + } + }, + "indexed_under": { + "$ref": "#/$defs/SourceModel" + }, + "truncated": { + "description": "A listing had more chunks than `top_k`.", + "type": "boolean" + } + }, + "required": [ + "indexed_under", + "truncated", + "hits" + ] + }, + "ChatMessage": { + "description": "One message of a chunk, as the room held it.", + "type": "object", + "properties": { + "body": { + "type": "string" + }, + "event_id": { + "type": "string" + }, + "sender": { + "description": "The sender's Matrix id.", + "type": "string" + }, + "ts_ms": { + "type": "integer", + "format": "uint64", + "minimum": 0 + } + }, + "required": [ + "event_id", + "sender", + "ts_ms", + "body" + ] + }, "CitationRole": { "description": "The referring field of a citation edge, one per leaf field that names\nanother document. The wire spelling is the field's own name, so a client\nreads an edge's role as the field it came from.", "type": "string",