chat: search_chat over the shared source collection, huginn chat search (#57)

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"

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<Utc>) -> Result<u64, Trouble> {
+    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}");
+        }
+    }
+}

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<String>,
+        /// Only this room: an id (`!abc:gamecult.org`) or a canonical alias
+        /// (`#swarm:gamecult.org`). Repeat for several.
+        #[arg(long = "room")]
+        rooms: Vec<String>,
+        /// From this instant: an RFC 3339 time, or an age such as 6h or 2d.
+        #[arg(long)]
+        after: Option<String>,
+        /// Until this instant, as --after.
+        #[arg(long)]
+        before: Option<String>,
+        /// How many chunks (default 10, at most 50).
+        #[arg(long)]
+        top_k: Option<u32>,
+        /// 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<u8, Trouble> {
             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<String>,
+    rooms: Vec<String>,
+    after: Option<&str>,
+    before: Option<&str>,
+    top_k: Option<u32>,
+    json: bool,
+) -> Result<u8, Trouble> {
+    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",

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<String> = 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<String> {

crates/eureka-state/tests/cli.rs

@@ -1233,7 +1233,8 @@ struct Desk {
     embeds_under: &'static str,
     held: Vec<huginn_daemon::sources::SourcePayload>,
     tickets: u64,
-    done: Vec<(SearchTicket, Result<huginn_mind::SourceHits, String>)>,
+    chat: Vec<huginn_daemon::sources::ChatPayload>,
+    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<Vec<huginn_daemon::sources::ChatRow>> {
+        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<Vec<huginn_daemon::sources::ChatPayload>> {
+        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<S: MindStore> IndexSink<S> for Desk {
@@ -1313,11 +1343,21 @@ impl<S: MindStore> IndexSink<S> 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<huginn_mind::SourceHits, String>)> {
+    fn search_chat(&mut self, query: &huginn_mind::ChatQuery) -> Result<SearchTicket, String> {
+        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);
+    }
+}

crates/eureka-state/tests/common/mod.rs

@@ -93,10 +93,14 @@ impl<S: MindStore> IndexSink<S> for Inert {
     fn search(&mut self, _text: &str, _top_k: u32, _among: Vec<String>) -> Result<SearchTicket, String> {
         Err("this index cannot search".into())
     }
+    fn search_chat(&mut self, _query: &huginn_mind::ChatQuery) -> Result<SearchTicket, String> {
+        Err("this index cannot search".into())
+    }
+
     fn search_sources(&mut self, _query: &huginn_mind::SourceQuery) -> Result<SearchTicket, String> {
         Err("this index cannot search".into())
     }
-    fn sources_searched(&mut self) -> Vec<(SearchTicket, Result<huginn_mind::SourceHits, String>)> {
+    fn sources_searched(&mut self) -> Vec<(SearchTicket, huginn_daemon::daemon::Found)> {
         Vec::new()
     }
     fn search_deadline(&self) -> Duration {

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<S: MindStore> {
     /// 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<SearchTicket, String>;
-    /// Every source search that finished since the last call, by ticket.
+    /// Starts a chat search: the same queue and worker as a source search,
-    fn sources_searched(&mut self) -> Vec<(SearchTicket, Result<SourceHits, String>)>;
+    /// over the same collection.
+    fn search_chat(&mut self, query: &ChatQuery) -> Result<SearchTicket, String>;
+    /// 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<S: MindStore, I: IndexSink<S>> Daemon<S, I> {
                 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<S: MindStore, I: IndexSink<S>> Daemon<S, I> {
     /// 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(self.index.sources_searched()).collect()
-        mind.chain(sources).collect()
     }
 
     /// The client has stopped waiting for `ticket`.
@@ -300,6 +313,7 @@ impl<S: MindStore, I: IndexSink<S>> Daemon<S, I> {
         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<SearchTicket, String> {
+            Err("this index cannot search".into())
+        }
+
         fn search_sources(&mut self, _query: &SourceQuery) -> Result<SearchTicket, String> {
             Err("this index cannot search".into())
         }
 
-        fn sources_searched(&mut self) -> Vec<(SearchTicket, Result<SourceHits, String>)> {
+        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<SearchTicket, String> {
+            Err("this index cannot search".into())
+        }
+
         fn search_sources(&mut self, _query: &SourceQuery) -> Result<SearchTicket, String> {
             *self.asked.lock().unwrap() += 1;
             Ok(SearchTicket(7))
         }
 
-        fn sources_searched(&mut self) -> Vec<(SearchTicket, Result<SourceHits, String>)> {
+        fn sources_searched(&mut self) -> Vec<(SearchTicket, Found)> {
-            vec![(SearchTicket(7), Ok(self.hits.clone()))]
+            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<SearchTicket, String> {
+            Err("this index cannot search".into())
+        }
+
         fn search_sources(&mut self, _query: &SourceQuery) -> Result<SearchTicket, String> {
             Err("this index cannot search".into())
         }
 
-        fn sources_searched(&mut self) -> Vec<(SearchTicket, Result<SourceHits, String>)> {
+        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<SearchTicket, String> {
+            Err("this index cannot search".into())
+        }
+
         fn search_sources(&mut self, _query: &SourceQuery) -> Result<SearchTicket, String> {
             Err("this index cannot search".into())
         }
 
-        fn sources_searched(&mut self) -> Vec<(SearchTicket, Result<SourceHits, String>)> {
+        fn sources_searched(&mut self) -> Vec<(SearchTicket, Found)> {
             Vec::new()
         }
 

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<DescribedSources>,
     pub(crate) points: BTreeMap<String, SourcePoint>,
+    /// The chat points, by id: the same collection, the other corpus.
+    pub(crate) chat: BTreeMap<String, ChatPoint>,
     /// 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<Vec<ChatRow>> {
+        let state = self.state.lock().unwrap();
+        let mut rows: Vec<ChatRow> = 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<Vec<ChatPayload>> {
+        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<Vec<SourceRow>> {
         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<Vec<ChatRow>> {
+        self.sources.chat_nearest(vector, filter, limit, within)
+    }
+    fn chat_scroll(&mut self, filter: &ChatFilter, limit: u32, within: Duration) -> Result<Vec<ChatPayload>> {
+        self.sources.chat_scroll(filter, limit, within)
+    }
 }

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
+/// What a source-collection search asks: forge chunks or chat chunks.
-/// the mind's reconciliation gates it.
+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<Hits, String>)>,
     source_searches: VecDeque<QueuedSource>,
-    sources_found: Vec<(SearchTicket, Result<SourceHits, String>)>,
+    sources_found: Vec<(SearchTicket, Found)>,
     next_ticket: u64,
 }
 
@@ -821,10 +827,14 @@ fn serve_source_search<E: Embedder, V: VectorIndex + SourceIndex>(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()))
+    let within = due.saturating_duration_since(Instant::now());
-        .map_err(|error| format!("{error:#}"));
+    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<SearchTicket, String> {
+        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<S: MindStore> IndexSink<S> for WorkerSink {
     }
 
     fn search_sources(&mut self, query: &SourceQuery) -> Result<SearchTicket, String> {
-        let mut guard = lock(&self.shared);
+        self.queue_source_ask(SourceAsk::Sources(query.clone()))
-        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)
     }
 
-    fn sources_searched(&mut self) -> Vec<(SearchTicket, Result<SourceHits, String>)> {
+    fn search_chat(&mut self, query: &ChatQuery) -> Result<SearchTicket, String> {
+        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)
     }
 

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
+/// The payload filter of a source search: forge points only, then the repo and
-/// (exact matches on the keyword-indexed `path`), both when both are asked.
+/// the front-page paths (exact matches on the keyword-indexed `path`) when
-fn source_filter(repo: Option<&str>, front: bool) -> Option<Value> {
+/// asked. A chat point carries no repo or path and can never pass it.
-    let mut must = Vec::new();
+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<Value> {
+    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::<Vec<_>>(),
+        "messages": payload.messages.iter().map(|message| json!({
+            "event_id": message.event_id, "sender": message.sender, "ts_ms": message.ts_ms, "body": message.body,
+        })).collect::<Vec<_>>(),
+        "text": payload.text,
+    });
+    if let Some(alias) = &payload.room_alias {
+        out["room_alias"] = json!(alias);
+    }
+    Ok(out)
+}
+
+fn chat_payload_of(payload: &Value) -> Result<ChatPayload> {
+    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::<Result<Vec<_>>>()?;
+    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"] {
+        for (field, schema) in [
-            let index = json!({ "field_name": field, "field_schema": "keyword" });
+            ("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"] = source_filter(repo, front);
-            body["filter"] = filter;
-        }
         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<Value> = points
+            .iter()
+            .map(|point| Ok(json!({ "id": point.id, "vector": point.vector, "payload": chat_payload_json(&point.payload)? })))
+            .collect::<Result<_>>()?;
+        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<Vec<ChatRow>> {
+        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<Vec<ChatPayload>> {
+        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() {
+    fn no_ask_is_the_forge_corpus_and_a_repo_is_its_own_match() {
-        assert_eq!(source_filter(None, false), None);
+        let forge = json!({ "key": "corpus", "match": { "value": "forge" } });
-        assert_eq!(source_filter(Some("Huginn"), false), Some(json!({ "must": [{ "key": "repo", "match": { "value": "Huginn" } }] })));
+        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" } }] }));
     }
 }

crates/huginn-daemon/src/index/tests.rs

@@ -1999,10 +1999,19 @@ fn ask_sources(sink: &mut WorkerSink, text: &str) -> Result<SearchTicket, String
     <WorkerSink as IndexSink<OwnedRedbMessagePackBackingStore>>::search_sources(sink, &query)
 }
 
-fn sources_answers(sink: &mut WorkerSink, count: usize) -> Vec<(SearchTicket, Result<SourceHits, String>)> {
+fn sources_answers(sink: &mut WorkerSink, count: usize) -> Vec<(SearchTicket, Result<huginn_mind::SourceHits, String>)> {
     let mut all = Vec::new();
     eventually("the source searches being answered", || {
-        all.extend(<WorkerSink as IndexSink<OwnedRedbMessagePackBackingStore>>::sources_searched(sink));
+        all.extend(<WorkerSink as IndexSink<OwnedRedbMessagePackBackingStore>>::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 = <WorkerSink as IndexSink<OwnedRedbMessagePackBackingStore>>::search_chat(&mut sink, &query).unwrap();
+    let mut found = Vec::new();
+    eventually("the chat search being answered", || {
+        found.extend(<WorkerSink as IndexSink<OwnedRedbMessagePackBackingStore>>::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:?}");
+}

crates/huginn-daemon/src/sources/mod.rs

@@ -1,5 +1,7 @@
-//! The source index: the public GameCult forge repos' code and docs as chunks
+//! The source index: the public GameCult forge repos' code and docs, and the
-//! in one Qdrant collection, `huginn_sources`, for semantic search.
+//! 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.
+/// What the collection says its corpus is: the set of corpora it holds, each
-pub const CORPUS: &str = "forge:GameCult";
+/// 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<String>,
+    pub messages: Vec<ChatMessage>,
+    pub text: String,
+}
+
+#[derive(Clone, Debug, PartialEq)]
+pub struct ChatPoint {
+    pub id: String,
+    pub vector: Vec<f32>,
+    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<String>,
+    pub after_ms: Option<u64>,
+    pub before_ms: Option<u64>,
+}
+
+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<DescribedSources>;
     /// 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<String, String>) -> 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<Vec<SourceRow>>;
+    /// 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<Vec<ChatRow>>;
+    /// 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<Vec<ChatPayload>>;
 }
 
 /// A repo the forge lists as indexable.
@@ -476,6 +552,10 @@ pub fn fuse<T>(lists: Vec<Vec<T>>, 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:
+/// The label of the collection a daemon is about to read, if it was built
-/// it embeds the query with its own embedder, and refuses if the collection
+/// under the model this daemon embeds with. Absent, unlabelled and other-model
-/// was built under another model, because vectors of two models do not compare.
+/// collections are loud refusals: vectors of two models do not compare, and an
-pub fn search<E: Embedder, S: SourceIndex>(embedder: &mut E, index: &mut S, query: &SourceQuery, within: Duration) -> Result<SourceHits> {
+/// empty page is not an answer. A collection built for another corpus set (an
-    let budget = Budget::new(within);
+/// old forge-only one) is refused the same way until the rebuild lands, since
-    let built = match index.describe_sources(budget.left()?).context("describing the source collection")? {
+/// it would answer a chat search with an empty page.
-        DescribedSources::Absent => bail!("source index not built"),
+fn readable_under_this_model<E: Embedder, S: SourceIndex>(embedder: &mut E, index: &mut S, budget: &Budget) -> Result<(SourcesMeta, ModelIdentity)> {
-        DescribedSources::Unlabelled => bail!("source index not built: {COLLECTION} carries no Huginn metadata"),
+    let built = built_label(index, budget)?;
-        DescribedSources::Labelled { meta, .. } => meta,
-    };
     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<E: Embedder, S: SourceIndex>(embedder: &mut E, index: &mut S, quer
         identity.digest,
         identity.dimensions
     );
+    Ok((built, identity))
+}
+
+fn built_label<S: SourceIndex>(index: &mut S, budget: &Budget) -> Result<SourcesMeta> {
+    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<E: Embedder, S: SourceIndex>(embedder: &mut E, index: &mut S, query: &SourceQuery, within: Duration) -> Result<SourceHits> {
+    let budget = Budget::new(within);
+    let (built, identity) = readable_under_this_model(embedder, index, &budget)?;
     let topics = query_texts(query);
     let prompts: Vec<String> = 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<E: Embedder, S: SourceIndex>(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<E: Embedder, S: SourceIndex>(embedder: &mut E, index: &mut S, query: &ChatQuery, within: Duration) -> Result<ChatHits> {
+    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::<Vec<_>>(), truncated)
+    } else {
+        let (built, identity) = readable_under_this_model(embedder, index, &budget)?;
+        let prompts: Vec<String> = 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<ChatRow> = 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<f32>, payload: ChatPayload, filter: &ChatFilter) -> Option<ChatHit> {
+    let messages: Vec<ChatMessage> = 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,
+    })
+}

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<ChatMessage> = 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<u64>, before: Option<u64>, 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<ChatPoint>) -> (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<ChatPoint> {
+    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::<Vec<_>>()
+    };
+    let owned = |pairs: &[(&str, usize)]| pairs.iter().map(|(id, n)| ((*id).to_owned(), *n)).collect::<Vec<_>>();
+    // 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));
+}

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

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<SourceHit>,
 }
 
+/// 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<Line>,
+    pub rooms: Vec<Short>,
+    pub after_ms: Option<u64>,
+    pub before_ms: Option<u64>,
+    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<String>, 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<String>,
+    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<f32>,
+    pub messages: Vec<ChatMessage>,
+}
+
+/// 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<ChatHit>,
+}
+
 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<PersonaStateView>),
     PersonaPut(PersonaPutOutcome),
     Sources(SourceHits),
+    Chat(ChatHits),
     Activity(RepoActivity),
     RootsTouched(Vec<RootTouched>),
     /// 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]

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.

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": [

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",