From 0d5efeeaceba7ce74445b6001cc8975d01addfdb Mon Sep 17 00:00:00 2001 From: Eureka Date: Wed, 11 Nov 2026 19:48:48 +0000 Subject: [PATCH] chat window: deterministic same-ms order, ties no longer stall the walk, one line per message Order by (ts, chunk start, chunk first event id, position) so the window does not depend on its anchor. Each read asks for the held chunks at the edge in addition, so a run of chunks at one start ms cannot fill a read with held ones; the no-gain early stop is gone. Search and window print a body's line breaks as \n. Tests pin ties, caps, dedupe, the unreached flag and non-chat points. Co-Authored-By: Claude Sonnet 5.5 --- crates/eureka-state/src/bin/huginn/render.rs | 15 ++-- crates/eureka-state/tests/cli.rs | 4 +- crates/huginn-daemon/src/index/qdrant.rs | 2 + crates/huginn-daemon/src/sources/mod.rs | 46 +++++++------ crates/huginn-daemon/src/sources/tests.rs | 72 +++++++++++++++++++- 5 files changed, 110 insertions(+), 29 deletions(-) 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]