chat: search_chat over the shared source collection, huginn chat search (#57)
This commit is contained in:
parent
b299d6f732
commit
421ace5432
18 changed files with 1471 additions and 81 deletions
|
|
@ -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"
|
||||
|
|
|
|||
57
crates/eureka-state/src/bin/huginn/chat.rs
Normal file
57
crates/eureka-state/src/bin/huginn/chat.rs
Normal file
|
|
@ -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}");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -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",
|
||||
|
|
|
|||
|
|
@ -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> {
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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 {
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
fn sources_searched(&mut self) -> Vec<(SearchTicket, Result<SourceHits, String>)>;
|
||||
/// Starts a chat search: the same queue and worker as a source search,
|
||||
/// over the same collection.
|
||||
fn search_chat(&mut self, query: &ChatQuery) -> Result<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(sources).collect()
|
||||
mind.chain(self.index.sources_searched()).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>)> {
|
||||
vec![(SearchTicket(7), Ok(self.hits.clone()))]
|
||||
fn sources_searched(&mut self) -> Vec<(SearchTicket, Found)> {
|
||||
vec![(SearchTicket(7), Found::Sources(self.hits.clone()))]
|
||||
}
|
||||
|
||||
fn search_deadline(&self) -> std::time::Duration {
|
||||
|
|
@ -685,6 +708,31 @@ pub(crate) mod tests {
|
|||
assert_eq!(*asked.lock().unwrap(), 3, "a refused query never reached the index");
|
||||
}
|
||||
|
||||
/// A chat search is checked at the door before the index is asked, is
|
||||
/// refused for another mind, turns an index that cannot be asked into
|
||||
/// `Unavailable`, and its finished hits are the answer as they are.
|
||||
#[test]
|
||||
fn a_chat_search_is_checked_at_the_door_and_answers_its_hits() {
|
||||
use huginn_mind::eureka_pipeline::{Line, Short};
|
||||
use huginn_mind::{ChatHits, SourceModel};
|
||||
let root = tempfile::tempdir().unwrap();
|
||||
let mut daemon = Daemon::new(Mind::open(root.path(), &slug(INSTANCE)).unwrap(), NoIndex);
|
||||
let request = |instance: &str, rooms: &[&str], top_k: u32| HuginnMindRequest::SearchChat {
|
||||
instance: slug(instance),
|
||||
query: ChatQuery { texts: vec![Line("find".into())], rooms: rooms.iter().map(|room| Short((*room).into())).collect(), after_ms: None, before_ms: None, top_k },
|
||||
};
|
||||
for (refused, field) in [(request(INSTANCE, &[], 0), "top_k"), (request(INSTANCE, &["swarm"], 5), "rooms")] {
|
||||
let answer = daemon.handle(refused, now()).answered();
|
||||
assert!(matches!(answer, HuginnMindResponse::Refused(MindRefusal::SelectionInvalid { field: ref named, .. }) if named == field), "{answer:?}");
|
||||
}
|
||||
assert!(matches!(daemon.handle(request(OTHER, &[], 5), now()).answered(), HuginnMindResponse::Refused(_)));
|
||||
let answer = daemon.handle(request(INSTANCE, &["#swarm:x"], 5), now()).answered();
|
||||
assert!(matches!(answer, HuginnMindResponse::Refused(MindRefusal::Unavailable { .. })), "{answer:?}");
|
||||
let hits = ChatHits { indexed_under: SourceModel { name: "m".into(), digest: "d".into() }, truncated: false, hits: vec![] };
|
||||
let search = Search { ticket: SearchTicket(1), mind: None };
|
||||
assert_eq!(daemon.finish(search, Found::Chat(hits.clone())), HuginnMindResponse::Chat(hits));
|
||||
}
|
||||
|
||||
/// A daemon over the instance's mind under `state_root`, with no index.
|
||||
pub(crate) fn open_unindexed(
|
||||
state_root: &Path,
|
||||
|
|
@ -716,11 +764,15 @@ pub(crate) mod tests {
|
|||
Err("this index cannot search".into())
|
||||
}
|
||||
|
||||
fn search_chat(&mut self, _query: &ChatQuery) -> Result<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()
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -105,9 +105,9 @@ use huginn_mind::wire::IndexStatus;
|
|||
use huginn_mind::{INDEX_TEXT_VERSION, IndexEntry, Mind, MindStore};
|
||||
use sha2::{Digest, Sha256};
|
||||
|
||||
use crate::daemon::{Hits, IndexSink, SearchTicket};
|
||||
use crate::daemon::{Found, Hits, IndexSink, SearchTicket};
|
||||
use crate::sources::{self, SourceIndex};
|
||||
use huginn_mind::{SourceHits, SourceQuery};
|
||||
use huginn_mind::{ChatQuery, SourceQuery};
|
||||
use ollama::QUERY_INSTRUCTION;
|
||||
|
||||
/// The `managed_by` a collection carries when this organ made it.
|
||||
|
|
@ -685,11 +685,17 @@ struct Queued {
|
|||
due: Instant,
|
||||
}
|
||||
|
||||
/// A source search waiting for the worker. It is not the mind's: nothing about
|
||||
/// the mind's reconciliation gates it.
|
||||
/// What a source-collection search asks: forge chunks or chat chunks.
|
||||
enum SourceAsk {
|
||||
Sources(SourceQuery),
|
||||
Chat(ChatQuery),
|
||||
}
|
||||
|
||||
/// A source or chat search waiting for the worker. It is not the mind's:
|
||||
/// nothing about the mind's reconciliation gates it.
|
||||
struct QueuedSource {
|
||||
ticket: SearchTicket,
|
||||
query: SourceQuery,
|
||||
ask: SourceAsk,
|
||||
due: Instant,
|
||||
}
|
||||
|
||||
|
|
@ -706,7 +712,7 @@ struct Shared {
|
|||
/// Searches the worker finished and the serve loop has not collected.
|
||||
found: Vec<(SearchTicket, Result<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()))
|
||||
.map_err(|error| format!("{error:#}"));
|
||||
let within = due.saturating_duration_since(Instant::now());
|
||||
let found = match ask {
|
||||
SourceAsk::Sources(query) => sources::search(&mut projector.embedder, &mut projector.index, &query, within).map(Found::Sources),
|
||||
SourceAsk::Chat(query) => sources::search_chat(&mut projector.embedder, &mut projector.index, &query, within).map(Found::Chat),
|
||||
}
|
||||
.unwrap_or_else(|error| Found::Failed(format!("{error:#}")));
|
||||
lock(shared).sources_found.push((ticket, found));
|
||||
}
|
||||
|
||||
|
|
@ -966,6 +976,24 @@ pub struct WorkerSink {
|
|||
}
|
||||
|
||||
impl WorkerSink {
|
||||
/// Queues a source or chat search behind the same bound as every other
|
||||
/// search; the mind's index state does not refuse it.
|
||||
fn queue_source_ask(&mut self, ask: SourceAsk) -> Result<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);
|
||||
if guard.searches.len() + guard.source_searches.len() >= SEARCHES_QUEUED_MAX {
|
||||
return Err(format!(
|
||||
"the index has {} searches waiting and takes no more until it has answered some",
|
||||
guard.searches.len() + guard.source_searches.len()
|
||||
));
|
||||
}
|
||||
let ticket = SearchTicket(guard.next_ticket);
|
||||
guard.next_ticket += 1;
|
||||
guard.source_searches.push_back(QueuedSource { ticket, query: query.clone(), due: Instant::now() + self.deadline });
|
||||
drop(guard);
|
||||
self.shared.1.notify_all();
|
||||
Ok(ticket)
|
||||
self.queue_source_ask(SourceAsk::Sources(query.clone()))
|
||||
}
|
||||
|
||||
fn sources_searched(&mut self) -> Vec<(SearchTicket, Result<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)
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -7,7 +7,12 @@ use std::time::Duration;
|
|||
use anyhow::{Context, Result, ensure};
|
||||
use serde_json::{Value, json};
|
||||
|
||||
use crate::sources::{COLLECTION, FRONT_PAGE_FILES, DescribedSources, SourceIndex, SourcePayload, SourcePoint, SourceRow, SourcesMeta};
|
||||
use huginn_mind::ChatMessage;
|
||||
|
||||
use crate::sources::{
|
||||
COLLECTION, CORPUS_CHAT, CORPUS_FORGE, ChatFilter, ChatPayload, ChatPoint, ChatRow, DescribedSources, FRONT_PAGE_FILES, SourceIndex, SourcePayload, SourcePoint, SourceRow,
|
||||
SourcesMeta,
|
||||
};
|
||||
|
||||
use super::{CollectionMeta, INDEX_CALL_TIMEOUT, Described, Hit, Http, Method, Point, VectorIndex};
|
||||
|
||||
|
|
@ -175,21 +180,94 @@ impl VectorIndex for QdrantIndex {
|
|||
}
|
||||
}
|
||||
|
||||
/// The payload filter of a source search: the repo, and the front-page paths
|
||||
/// (exact matches on the keyword-indexed `path`), both when both are asked.
|
||||
fn source_filter(repo: Option<&str>, front: bool) -> Option<Value> {
|
||||
let mut must = Vec::new();
|
||||
/// The payload filter of a source search: forge points only, then the repo and
|
||||
/// the front-page paths (exact matches on the keyword-indexed `path`) when
|
||||
/// asked. A chat point carries no repo or path and can never pass it.
|
||||
fn source_filter(repo: Option<&str>, front: bool) -> Value {
|
||||
let mut must = vec![json!({ "key": "corpus", "match": { "value": CORPUS_FORGE } })];
|
||||
if let Some(repo) = repo {
|
||||
must.push(json!({ "key": "repo", "match": { "value": repo } }));
|
||||
}
|
||||
if front {
|
||||
must.push(json!({ "key": "path", "match": { "any": FRONT_PAGE_FILES } }));
|
||||
}
|
||||
(!must.is_empty()).then(|| json!({ "must": must }))
|
||||
json!({ "must": must })
|
||||
}
|
||||
|
||||
/// The payload filter of a chat read: chat points only; the rooms any-of by id
|
||||
/// (`!`) or alias (`#`) as a `should`; the span as ranges that keep a chunk
|
||||
/// overlapping [after, before].
|
||||
fn chat_filter(filter: &ChatFilter) -> Value {
|
||||
let mut must = vec![json!({ "key": "corpus", "match": { "value": CORPUS_CHAT } })];
|
||||
if let Some(after) = filter.after_ms {
|
||||
must.push(json!({ "key": "ts_end_ms", "range": { "gte": after } }));
|
||||
}
|
||||
if let Some(before) = filter.before_ms {
|
||||
must.push(json!({ "key": "ts_start_ms", "range": { "lte": before } }));
|
||||
}
|
||||
let mut out = json!({ "must": must });
|
||||
let (ids, aliases): (Vec<&String>, Vec<&String>) = filter.rooms.iter().partition(|room| room.starts_with('!'));
|
||||
let mut should = Vec::new();
|
||||
if !ids.is_empty() {
|
||||
should.push(json!({ "key": "room_id", "match": { "any": ids } }));
|
||||
}
|
||||
if !aliases.is_empty() {
|
||||
should.push(json!({ "key": "room_alias", "match": { "any": aliases } }));
|
||||
}
|
||||
if !should.is_empty() {
|
||||
out["should"] = Value::Array(should);
|
||||
}
|
||||
out
|
||||
}
|
||||
|
||||
/// A chat chunk's payload. The span, first and last ids and the id list are
|
||||
/// derived from `messages`, so they cannot disagree with them.
|
||||
fn chat_payload_json(payload: &ChatPayload) -> Result<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"] {
|
||||
let index = json!({ "field_name": field, "field_schema": "keyword" });
|
||||
for (field, schema) in [
|
||||
("repo", "keyword"),
|
||||
("path", "keyword"),
|
||||
("corpus", "keyword"),
|
||||
("room_id", "keyword"),
|
||||
("room_alias", "keyword"),
|
||||
("event_ids", "keyword"),
|
||||
("ts_start_ms", "integer"),
|
||||
("ts_end_ms", "integer"),
|
||||
] {
|
||||
let index = json!({ "field_name": field, "field_schema": schema });
|
||||
self.ask(Method::Put, &format!("{path}/index?wait=true"), Some(&index), &[200])?;
|
||||
}
|
||||
Ok(())
|
||||
|
|
@ -315,9 +402,7 @@ impl SourceIndex for QdrantIndex {
|
|||
let collection = self.sources.clone();
|
||||
let http = self.http.bounded(within);
|
||||
let mut body = json!({ "vector": vector, "limit": limit, "with_payload": true, "with_vector": false });
|
||||
if let Some(filter) = source_filter(repo, front) {
|
||||
body["filter"] = filter;
|
||||
}
|
||||
body["filter"] = source_filter(repo, front);
|
||||
let (_, answer) =
|
||||
self.ask_with(&http, Method::Post, &format!("/collections/{collection}/points/search"), Some(&body), &[200])?;
|
||||
let found = answer["result"].as_array().context("Qdrant's search has no result")?;
|
||||
|
|
@ -331,6 +416,45 @@ impl SourceIndex for QdrantIndex {
|
|||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
fn upsert_chat(&mut self, points: &[ChatPoint]) -> Result<()> {
|
||||
let collection = self.sources.clone();
|
||||
let points: Vec<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() {
|
||||
assert_eq!(source_filter(None, false), None);
|
||||
assert_eq!(source_filter(Some("Huginn"), false), Some(json!({ "must": [{ "key": "repo", "match": { "value": "Huginn" } }] })));
|
||||
fn no_ask_is_the_forge_corpus_and_a_repo_is_its_own_match() {
|
||||
let forge = json!({ "key": "corpus", "match": { "value": "forge" } });
|
||||
assert_eq!(source_filter(None, false), json!({ "must": [forge] }));
|
||||
assert_eq!(source_filter(Some("Huginn"), false), json!({ "must": [forge, { "key": "repo", "match": { "value": "Huginn" } }] }));
|
||||
}
|
||||
|
||||
/// The front filter matches the path against exactly the front-page files
|
||||
|
|
@ -354,15 +479,63 @@ mod source_filter_tests {
|
|||
assert_eq!(path.len(), 1);
|
||||
path[0]["match"]["any"].as_array().unwrap().iter().map(|p| p.as_str().unwrap().to_owned()).collect()
|
||||
};
|
||||
let front = source_filter(None, true).unwrap();
|
||||
let front = source_filter(None, true);
|
||||
assert_eq!(any(&front), ["README.md", "README", "readme.md", "README.markdown", "CLAUDE.md", "AGENTS.md"]);
|
||||
assert_eq!(front["must"].as_array().unwrap().len(), 1, "no repo condition when none was asked");
|
||||
assert_eq!(front["must"].as_array().unwrap().len(), 2, "the corpus and the path; no repo condition when none was asked");
|
||||
|
||||
let both = source_filter(Some("Zyphos"), true).unwrap();
|
||||
let both = source_filter(Some("Zyphos"), true);
|
||||
assert_eq!(any(&both), any(&front));
|
||||
let repos: Vec<&Value> = both["must"].as_array().unwrap().iter().filter(|c| c["key"] == "repo").collect();
|
||||
assert_eq!(repos.len(), 1);
|
||||
assert_eq!(repos[0]["match"]["value"], "Zyphos");
|
||||
assert_eq!(both["must"].as_array().unwrap().len(), 2);
|
||||
assert_eq!(both["must"].as_array().unwrap().len(), 3);
|
||||
}
|
||||
|
||||
fn payload() -> ChatPayload {
|
||||
let message = |id: &str, ts: u64| ChatMessage { event_id: id.into(), sender: "@a:gamecult.org".into(), ts_ms: ts, body: format!("body {id}") };
|
||||
ChatPayload { room_id: "!r:gamecult.org".into(), room_alias: Some("#swarm:gamecult.org".into()), messages: vec![message("$a", 10), message("$b", 20)], text: "swarm\n10 @a: body".into() }
|
||||
}
|
||||
|
||||
/// The chat codec is its own inverse, writes the contract's fields derived
|
||||
/// from the messages, and refuses a chunk with none and a point of another corpus.
|
||||
#[test]
|
||||
fn the_chat_payload_round_trips_the_contract() {
|
||||
let written = chat_payload_json(&payload()).unwrap();
|
||||
assert_eq!(written["corpus"], "chat");
|
||||
assert_eq!((written["ts_start_ms"].as_u64(), written["ts_end_ms"].as_u64()), (Some(10), Some(20)));
|
||||
assert_eq!((written["first_event_id"].as_str(), written["last_event_id"].as_str()), (Some("$a"), Some("$b")));
|
||||
assert_eq!(written["event_ids"], json!(["$a", "$b"]));
|
||||
assert_eq!(written["room_alias"], "#swarm:gamecult.org");
|
||||
assert_eq!(chat_payload_of(&written).unwrap(), payload());
|
||||
let mut no_alias = payload();
|
||||
no_alias.room_alias = None;
|
||||
assert!(chat_payload_json(&no_alias).unwrap().get("room_alias").is_none());
|
||||
assert_eq!(chat_payload_of(&chat_payload_json(&no_alias).unwrap()).unwrap(), no_alias);
|
||||
assert!(chat_payload_json(&ChatPayload { messages: vec![], ..payload() }).is_err());
|
||||
let forge = payload_json(&SourcePayload { repo: "r".into(), path: "p".into(), commit: "c".into(), blob: "b".into(), line_start: 1, line_end: 2, chunk_index: 0, ext: "rs".into(), text: "t".into() });
|
||||
assert_eq!(forge["corpus"], "forge");
|
||||
assert!(chat_payload_of(&forge).is_err());
|
||||
}
|
||||
|
||||
/// The chat filter is the chat corpus, the rooms as an any-of `should` split
|
||||
/// by id and alias, and the two ranges that keep an overlapping span.
|
||||
#[test]
|
||||
fn the_chat_filter_is_corpus_rooms_and_overlap() {
|
||||
let filter = chat_filter(&ChatFilter { rooms: vec!["!r:x".into(), "#a:x".into(), "#b:x".into()], after_ms: Some(5), before_ms: Some(9) });
|
||||
assert_eq!(
|
||||
filter,
|
||||
json!({
|
||||
"must": [
|
||||
{ "key": "corpus", "match": { "value": "chat" } },
|
||||
{ "key": "ts_end_ms", "range": { "gte": 5 } },
|
||||
{ "key": "ts_start_ms", "range": { "lte": 9 } },
|
||||
],
|
||||
"should": [
|
||||
{ "key": "room_id", "match": { "any": ["!r:x"] } },
|
||||
{ "key": "room_alias", "match": { "any": ["#a:x", "#b:x"] } },
|
||||
],
|
||||
})
|
||||
);
|
||||
assert_eq!(chat_filter(&ChatFilter::default()), json!({ "must": [{ "key": "corpus", "match": { "value": "chat" } }] }));
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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:?}");
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,5 +1,7 @@
|
|||
//! The source index: the public GameCult forge repos' code and docs as chunks
|
||||
//! in one Qdrant collection, `huginn_sources`, for semantic search.
|
||||
//! The source index: the public GameCult forge repos' code and docs, and the
|
||||
//! Minds' chat rooms, as chunks in one Qdrant collection, `huginn_sources`,
|
||||
//! for semantic search. Each point's `corpus` payload (`forge` or `chat`) keeps
|
||||
//! the two apart; a forge search filters `forge` and a chat search `chat`.
|
||||
//!
|
||||
//! Two truths, one writer. The forge is the truth of which repos exist and what
|
||||
//! they hold; the collection is derived from it and can be deleted whenever.
|
||||
|
|
@ -30,7 +32,7 @@ use std::path::Path;
|
|||
use std::time::{Duration, Instant};
|
||||
|
||||
use anyhow::{Context, Result, bail, ensure};
|
||||
use huginn_mind::{SOURCE_TOP_K_MAX, SourceHit, SourceHits, SourceModel, SourceQuery};
|
||||
use huginn_mind::{ChatHit, ChatHits, ChatMessage, ChatQuery, SOURCE_TOP_K_MAX, SourceHit, SourceHits, SourceModel, SourceQuery};
|
||||
|
||||
use crate::index::{BATCH, BULK_EMBED_TIMEOUT, Budget, Embedder, INDEX_CALL_TIMEOUT, MANAGED_BY, ModelIdentity, point_id};
|
||||
use chunk::{CHUNKER_VERSION, embedded_text};
|
||||
|
|
@ -43,8 +45,19 @@ pub const COLLECTION: &str = "huginn_sources";
|
|||
/// agent reads first to learn what the repo is. The only owner of the list.
|
||||
pub const FRONT_PAGE_FILES: [&str; 6] = ["README.md", "README", "readme.md", "README.markdown", "CLAUDE.md", "AGENTS.md"];
|
||||
|
||||
/// What the collection says its corpus is.
|
||||
pub const CORPUS: &str = "forge:GameCult";
|
||||
/// What the collection says its corpus is: the set of corpora it holds, each
|
||||
/// point's `corpus` payload naming which (`forge` or `chat`). A collection
|
||||
/// labelled with any other set is rebuilt.
|
||||
pub const CORPUS: &str = "forge:GameCult+chat:rooms";
|
||||
|
||||
/// The `corpus` payload of a forge point, and of a chat point.
|
||||
pub const CORPUS_FORGE: &str = "forge";
|
||||
pub const CORPUS_CHAT: &str = "chat";
|
||||
|
||||
/// The instruction Qwen3 embedding models take in front of a question about
|
||||
/// past conversation.
|
||||
pub const CHAT_QUERY_INSTRUCTION: &str =
|
||||
"Given a question about past conversation in chat rooms, retrieve the chat passages that answer it";
|
||||
|
||||
/// The instruction Qwen3 embedding models take in front of a query about code.
|
||||
pub const SOURCE_QUERY_INSTRUCTION: &str =
|
||||
|
|
@ -111,6 +124,58 @@ pub struct SourcePoint {
|
|||
pub payload: SourcePayload,
|
||||
}
|
||||
|
||||
/// What travels beside a chat chunk's vector: consecutive messages of one
|
||||
/// room, never spanning rooms. The single owner of the chat point's shape; the
|
||||
/// ingest job is its only writer (through `upsert_chat`) and nothing else of the
|
||||
/// payload is read. `messages` are the display truth and hold, in order, at
|
||||
/// least one message. The chunk's span, its first and last event ids and its
|
||||
/// `event_ids` are derived from them by the codec, so they cannot disagree.
|
||||
/// `text` is what was embedded: the room's name, then `time sender: body` per
|
||||
/// message.
|
||||
#[derive(Clone, Debug, PartialEq)]
|
||||
pub struct ChatPayload {
|
||||
pub room_id: String,
|
||||
/// Absent when the room has no canonical alias.
|
||||
pub room_alias: Option<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:
|
||||
/// 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 = match index.describe_sources(budget.left()?).context("describing the source collection")? {
|
||||
DescribedSources::Absent => bail!("source index not built"),
|
||||
DescribedSources::Unlabelled => bail!("source index not built: {COLLECTION} carries no Huginn metadata"),
|
||||
DescribedSources::Labelled { meta, .. } => meta,
|
||||
};
|
||||
/// The label of the collection a daemon is about to read, if it was built
|
||||
/// under the model this daemon embeds with. Absent, unlabelled and other-model
|
||||
/// collections are loud refusals: vectors of two models do not compare, and an
|
||||
/// empty page is not an answer. A collection built for another corpus set (an
|
||||
/// old forge-only one) is refused the same way until the rebuild lands, since
|
||||
/// it would answer a chat search with an empty page.
|
||||
fn readable_under_this_model<E: Embedder, S: SourceIndex>(embedder: &mut E, index: &mut S, budget: &Budget) -> Result<(SourcesMeta, ModelIdentity)> {
|
||||
let built = built_label(index, budget)?;
|
||||
let identity = embedder.model_identity(budget.left()?).context("reading the embedding model's identity")?;
|
||||
ensure!(
|
||||
built.model == identity.name && built.model_digest == identity.digest && built.dimensions == identity.dimensions,
|
||||
|
|
@ -506,6 +584,30 @@ pub fn search<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,
|
||||
})
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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));
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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]
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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": [
|
||||
|
|
|
|||
|
|
@ -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",
|
||||
|
|
|
|||
Loading…
Reference in a new issue