diff --git a/crates/eureka-state/src/bin/huginn/render.rs b/crates/eureka-state/src/bin/huginn/render.rs index 09ff943..3fd7d3d 100644 --- a/crates/eureka-state/src/bin/huginn/render.rs +++ b/crates/eureka-state/src/bin/huginn/render.rs @@ -150,8 +150,8 @@ 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 +/// the body as the room held it with line breaks written `\n`: quoted data +/// under our header, never a message framed as if sent to the reader. A blank line between hits; a /// truncated listing ends with the `--after` that continues it. pub fn chat_hits(found: &ChatHits) -> String { let mut out: Vec = found @@ -168,7 +168,7 @@ pub fn chat_hits(found: &ChatHits) -> String { 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); + let _ = writeln!(text, "{} {} {} {}", crate::chat::shown(message.ts_ms), message.sender, message.event_id, one_line(&message.body)); } text }) @@ -181,6 +181,13 @@ pub fn chat_hits(found: &ChatHits) -> String { out.join("\n") } +/// A message body on one line: a line break is written `\n` (and a carriage +/// return `\r`), so a body cannot start a line of its own. `--json` has the +/// body exactly. +fn one_line(body: &str) -> String { + body.replace('\r', "\\r").replace('\n', "\\n") +} + /// A chat window as the CLI prints it: `#alias (!room)` (the id alone with no /// alias), then one `time sender event_id body` line per message, the anchor /// marked `>` and the rest indented to match; a line says when messages lie @@ -193,7 +200,7 @@ pub fn chat_window(found: &ChatWindow) -> String { } for message in &found.messages { let mark = if message.event_id == found.anchor_event_id { '>' } else { ' ' }; - let _ = writeln!(out, "{mark} {} {} {} {}", crate::chat::shown(message.ts_ms), message.sender, message.event_id, message.body); + let _ = writeln!(out, "{mark} {} {} {} {}", crate::chat::shown(message.ts_ms), message.sender, message.event_id, one_line(&message.body)); } if found.more_after { out.push_str(" (later messages exist)\n"); diff --git a/crates/eureka-state/tests/cli.rs b/crates/eureka-state/tests/cli.rs index cdeb239..ba446e5 100644 --- a/crates/eureka-state/tests/cli.rs +++ b/crates/eureka-state/tests/cli.rs @@ -2963,7 +2963,7 @@ fn chat_search_prints_cited_messages_and_sends_its_selectors() { 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: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); @@ -3006,7 +3006,7 @@ fn chat_window_prints_the_anchor_marked_and_refuses_what_the_index_lacks() { assert_eq!( found.out, "#swarm:gamecult.org (!r:gamecult.org)\n\ - \x20 2023-11-14T22:13:20.000Z @nibu:gamecult.org $e1 ship the cut\nthen rest\n\ + \x20 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" ); let over = at(server.addr, &["chat", "window", "$e2", "--before", "51"], None); diff --git a/crates/huginn-daemon/src/index/qdrant.rs b/crates/huginn-daemon/src/index/qdrant.rs index f9fa7a5..3f51c0a 100644 --- a/crates/huginn-daemon/src/index/qdrant.rs +++ b/crates/huginn-daemon/src/index/qdrant.rs @@ -562,5 +562,7 @@ mod source_filter_tests { { "key": "event_ids", "match": { "value": "$e" } }, ]) ); + let corpus = json!({ "key": "corpus", "match": { "value": "chat" } }); + assert_eq!(filter["must"][0], corpus, "a point with no corpus, or another, does not match: the chat corpus clause is always first"); } } diff --git a/crates/huginn-daemon/src/sources/mod.rs b/crates/huginn-daemon/src/sources/mod.rs index e4cee6a..963296b 100644 --- a/crates/huginn-daemon/src/sources/mod.rs +++ b/crates/huginn-daemon/src/sources/mod.rs @@ -725,14 +725,12 @@ pub fn window(index: &mut S, query: &ChatWindowQuery, within: Du })); }; let start_of = |chunk: &ChatPayload| chunk.messages.iter().map(|message| message.ts_ms).min().unwrap_or(0); - let first_id = |chunk: &ChatPayload| chunk.messages.first().map(|message| message.event_id.clone()); + let first_id = |chunk: &ChatPayload| chunk.messages.first().map(|message| message.event_id.clone()).unwrap_or_default(); let room = ChatFilter { rooms: vec![anchor.room_id.clone()], ..ChatFilter::default() }; let at = anchor.messages.iter().position(|message| message.event_id == id).context("the chat index returned a chunk that does not hold the event it was asked for")?; - let mut held = vec![first_id(&anchor)]; - // Each side walks chunks of the anchor's room only, away from the anchor, - // until it holds more messages than asked (so the cut can tell whether - // more exist), the room ends, or the reads are spent. - let mut walk = |forward: bool, want: usize, mine: usize| -> Result<(Vec>, bool)> { + // Chunks held so far as (first event id, start): the anchor's, then each walk's. + let mut held = vec![(first_id(&anchor), start_of(&anchor))]; + let mut walk = |forward: bool, want: usize, mine: usize| -> Result<(Vec, bool)> { let (mut chunks, mut count, mut edge, mut exhausted) = (Vec::new(), mine, start_of(&anchor), false); for _ in 0..WINDOW_SCROLLS_MAX { if count > want { @@ -743,21 +741,23 @@ pub fn window(index: &mut S, query: &ChatWindowQuery, within: Du } else { (ChatFilter { starts_to_ms: Some(edge), ..room.clone() }, ChatOrder::NewestFirst) }; - let read = index.chat_scroll(&filter, order, WINDOW_CHUNKS_PER_SCROLL, budget.left()?)?; - exhausted = (read.len() as u32) < WINDOW_CHUNKS_PER_SCROLL; - let mut gained = false; + // Held chunks starting at the edge come back first with any others + // that start there; asking for that many extra means a run of + // chunks at one millisecond cannot fill a read with chunks held. + let limit = WINDOW_CHUNKS_PER_SCROLL + held.iter().filter(|(_, start)| *start == edge).count() as u32; + let read = index.chat_scroll(&filter, order, limit, budget.left()?)?; + exhausted = (read.len() as u32) < limit; for chunk in read { - let first = first_id(&chunk); - if held.contains(&first) { + let key = (first_id(&chunk), start_of(&chunk)); + if held.contains(&key) { continue; } - held.push(first); - gained = true; - edge = if forward { start_of(&chunk).max(edge) } else { start_of(&chunk).min(edge) }; + edge = if forward { key.1.max(edge) } else { key.1.min(edge) }; + held.push(key); count += chunk.messages.len(); - chunks.push(chunk.messages); + chunks.push(chunk); } - if exhausted || !gained { + if exhausted { break; } } @@ -767,16 +767,20 @@ pub fn window(index: &mut S, query: &ChatWindowQuery, within: Du }; let (earlier, unreached_before) = walk(false, query.before as usize, at)?; let (later, unreached_after) = walk(true, query.after as usize, anchor.messages.len() - at - 1)?; - // Chronological: furthest earlier chunk first, the anchor's chunk, then later ones; - // stable by ts_ms, so ties keep chunk order and position. - let ordered = earlier.into_iter().rev().chain(std::iter::once(anchor.messages.clone())).chain(later).flatten(); + // Chronological by (ts, chunk start, chunk first event id, position), so + // the order does not depend on which message anchored the read. + let mut keyed: Vec<(u64, u64, String, usize, ChatMessage)> = Vec::new(); + for chunk in earlier.iter().chain(std::iter::once(&anchor)).chain(later.iter()) { + let (start, first) = (start_of(chunk), first_id(chunk)); + keyed.extend(chunk.messages.iter().enumerate().map(|(position, message)| (message.ts_ms, start, first.clone(), position, message.clone()))); + } + keyed.sort_by(|a, b| (a.0, a.1, &a.2, a.3).cmp(&(b.0, b.1, &b.2, b.3))); let (mut seen, mut messages) = (std::collections::BTreeSet::new(), Vec::new()); - for message in ordered { + for (.., message) in keyed { if seen.insert(message.event_id.clone()) { messages.push(message); } } - messages.sort_by_key(|message| message.ts_ms); let at = messages.iter().position(|message| message.event_id == id).context("the anchor fell out of its own window")?; let (lo, hi) = (at.saturating_sub(query.before as usize), (at + 1 + query.after as usize).min(messages.len())); Ok(Ok(ChatWindow { diff --git a/crates/huginn-daemon/src/sources/tests.rs b/crates/huginn-daemon/src/sources/tests.rs index adb4646..fbb364e 100644 --- a/crates/huginn-daemon/src/sources/tests.rs +++ b/crates/huginn-daemon/src/sources/tests.rs @@ -1071,13 +1071,81 @@ fn a_window_at_a_rooms_edge_returns_fewer_and_says_nothing_more_lies_beyond() { } #[test] -fn a_window_stops_after_six_reads_and_says_more_remain() { +fn a_window_stops_after_six_reads_of_five_chunks_and_says_more_remain() { let points: Vec = (0..80).map(|n| chat_point("!a:x", None, 0, 4, &[(&format!("$m{n:02}"), 1000 + n as u64)])).collect(); let mut rig = chat_rig(points).0; let around = window(&mut rig.index, &wq("$m70", 50, 0), WITHIN).unwrap().unwrap(); - assert!(around.messages.len() < 51, "six reads of five chunks reach at most 30 earlier messages"); + assert_eq!(around.messages.len(), 31, "six reads of five new chunks: thirty earlier messages and the anchor"); + assert_eq!(around.messages[0].event_id, "$m40"); assert_eq!(around.messages.last().unwrap().event_id, "$m70"); assert!(around.more_before && around.more_after, "the cap is not the room's start, and $m71 follows the anchor"); + // Asking for exactly what the six reads reach leaves the same honest flag. + let exact = window(&mut rig.index, &wq("$m70", 30, 0), WITHIN).unwrap().unwrap(); + assert_eq!((exact.messages.len(), exact.more_before), (31, true), "thirty held is not yet 'more than asked': the sixth read ended the walk, not the room"); + let fewer = window(&mut rig.index, &wq("$m70", 29, 0), WITHIN).unwrap().unwrap(); + assert_eq!((fewer.messages.len(), fewer.more_before), (30, true), "the cut leaves $m40 beyond"); +} + +#[test] +fn a_window_does_not_change_with_the_message_that_anchors_it_when_chunks_share_a_start() { + let chunks = [("$t1", "$t2", 150), ("$u1", "$u2", 160), ("$v1", "$v2", 170)]; + let points: Vec = chunks.iter().map(|(first, second, at)| chat_point("!a:x", None, 0, 4, &[(first, 100), (second, *at)])).collect(); + let mut rig = chat_rig(points).0; + let want = ["$t1", "$u1", "$v1", "$t2", "$u2", "$v2"]; + for anchor in want { + let around = window(&mut rig.index, &wq(anchor, 10, 10), WITHIN).unwrap().unwrap(); + assert_eq!(window_ids(&around), want, "anchored at {anchor}"); + assert!(!around.more_before && !around.more_after); + } +} + +#[test] +fn a_run_of_chunks_at_one_millisecond_is_all_reached() { + let points: Vec = (0..8).map(|n| chat_point("!a:x", None, 0, 4, &[(&format!("$c{n}"), 100)])).collect(); + let mut rig = chat_rig(points).0; + for anchor in ["$c0", "$c3", "$c7"] { + let around = window(&mut rig.index, &wq(anchor, 20, 20), WITHIN).unwrap().unwrap(); + assert_eq!(window_ids(&around), ["$c0", "$c1", "$c2", "$c3", "$c4", "$c5", "$c6", "$c7"], "anchored at {anchor}"); + assert!(!around.more_before && !around.more_after, "the room ends in both directions"); + } +} + +#[test] +fn a_chunk_starting_with_the_anchors_chunk_is_reached_going_forward_alone() { + // The anchor's own chunk already holds a message before it, so no earlier read happens. + let mut rig = chat_rig(vec![chat_point("!a:x", None, 0, 4, &[("$x0", 100), ("$x1", 120)]), chat_point("!a:x", None, 0, 4, &[("$y0", 100), ("$y1", 130)])]).0; + let around = window(&mut rig.index, &wq("$x1", 0, 5), WITHIN).unwrap().unwrap(); + assert_eq!(window_ids(&around), ["$x1", "$y1"]); + assert!(around.more_before && !around.more_after, "$x0 and $y0 precede; nothing follows $y1"); +} + +#[test] +fn a_walk_that_holds_exactly_the_asked_count_still_reads_on_to_learn_the_rooms_end() { + let points: Vec = (0..10).map(|n| chat_point("!a:x", None, 0, 4, &[(&format!("$m{n}"), 1000 + n as u64)])).collect(); + let mut rig = chat_rig(points).0; + let around = window(&mut rig.index, &wq("$m5", 5, 0), WITHIN).unwrap().unwrap(); + assert_eq!(window_ids(&around), ["$m0", "$m1", "$m2", "$m3", "$m4", "$m5"]); + assert!(!around.more_before, "the room starts at $m0"); +} + +#[test] +fn an_event_repeated_in_a_neighbour_chunk_appears_once() { + let mut rig = chat_rig(vec![chat_point("!a:x", None, 0, 4, &[("$a1", 100), ("$a2", 200)]), chat_point("!a:x", None, 0, 4, &[("$a2", 200), ("$a3", 300)])]).0; + let around = window(&mut rig.index, &wq("$a1", 0, 5), WITHIN).unwrap().unwrap(); + assert_eq!(window_ids(&around), ["$a1", "$a2", "$a3"]); +} + +#[test] +fn a_forge_point_is_not_an_event_the_window_can_anchor_on() { + let (mut rig, _) = topic_rig(4); + let mut state = rig.index.state.lock().unwrap(); + let mut point = state.points.values().next().unwrap().clone(); + point.id = "forge-with-an-event-id".into(); + point.payload.text = "$f1 looks like an event id".into(); + state.points.insert(point.id.clone(), point); + drop(state); + let (field, _) = refusal_field(window(&mut rig.index, &wq("$f1", 1, 1), WITHIN)); + assert_eq!(field, "event_id", "only corpus=chat points are read, as for an unknown id"); } #[test]