From ab61eaf163b4fcb0df0298beea4ecb061ce1a30a Mon Sep 17 00:00:00 2001 From: Eureka Date: Wed, 11 Nov 2026 19:18:22 +0000 Subject: [PATCH] sources: room_scope (joined-room admission, estate/DM/encrypted/bridged refusal, sender refusal) and the chat chunker Co-Authored-By: Claude Sonnet 5.5 --- crates/huginn-daemon/src/sources/chat.rs | 24 +++ .../huginn-daemon/src/sources/chat_chunk.rs | 166 ++++++++++++++++++ crates/huginn-daemon/src/sources/mod.rs | 2 + .../huginn-daemon/src/sources/room_scope.rs | 157 +++++++++++++++++ 4 files changed, 349 insertions(+) create mode 100644 crates/huginn-daemon/src/sources/chat_chunk.rs create mode 100644 crates/huginn-daemon/src/sources/room_scope.rs diff --git a/crates/huginn-daemon/src/sources/chat.rs b/crates/huginn-daemon/src/sources/chat.rs index 5efa0ce..6389634 100644 --- a/crates/huginn-daemon/src/sources/chat.rs +++ b/crates/huginn-daemon/src/sources/chat.rs @@ -15,3 +15,27 @@ pub const CHAT_PURGE_PREFIX: &str = "chat-purge:"; pub fn is_chat_key(key: &str) -> bool { key.starts_with(CHAT_CURSOR_PREFIX) || key.starts_with(CHAT_PURGE_PREFIX) } + +/// What one room event is to the ingest. Only text, notice and emote messages +/// become `Message`; everything else (encrypted, media, state, server-redacted) +/// is `Other` and is never stored. +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum EventKind { + Message { body: String }, + /// A replacement of `target`'s body. + Edit { target: String, body: String }, + Redaction { target: String }, + Other, +} + +/// One event of a room, as the port hands it over: the Matrix client maps its +/// JSON to this and filters nothing. `content_keys` are the top-level keys of +/// the event's content (the scope reads them). +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct RoomEvent { + pub event_id: String, + pub sender: String, + pub ts_ms: u64, + pub content_keys: Vec, + pub kind: EventKind, +} diff --git a/crates/huginn-daemon/src/sources/chat_chunk.rs b/crates/huginn-daemon/src/sources/chat_chunk.rs new file mode 100644 index 0000000..3ccee13 --- /dev/null +++ b/crates/huginn-daemon/src/sources/chat_chunk.rs @@ -0,0 +1,166 @@ +//! The chat chunker: greedy, deterministic and local. A chunk is consecutive +//! messages of one room; it closes at a time gap, at a message count, or when +//! its embedded text would pass a character bound. A message is never split. + +use chrono::DateTime; +use huginn_mind::redact::SCRUB_VERSION; +use huginn_mind::ChatMessage; + +use super::ChatPayload; +use crate::index::point_id; + +/// Bumped whenever chunk boundaries or the embedded text would change for the +/// same messages. Part of `chat_version`. +pub const CHAT_CHUNKER_VERSION: u32 = 1; + +/// A chunk closes when the next message is more than this much later. +pub const CHAT_GAP_MS: u64 = 30 * 60 * 1000; +/// The most messages a chunk holds. +pub const CHAT_MESSAGES_MAX: usize = 24; +/// The most characters a chunk's embedded text holds. +pub const CHAT_CHARS_MAX: usize = 1_600; +/// The most characters a stored message body holds, marker included. +pub const CHAT_BODY_CHARS_MAX: usize = 1_200; + +/// Ends a body that was cut. +const CUT_MARKER: &str = "…[cut]"; + +/// What the room's cursor and chunk ids are stamped with: the chunker and the +/// scrub. Either changing re-ingests the room. +pub fn chat_version() -> String { + format!("{CHAT_CHUNKER_VERSION}.{SCRUB_VERSION}") +} + +/// A chunk's id: where it starts and under which version it was cut. +pub fn chunk_id(room_id: &str, first_event_id: &str) -> String { + point_id(&format!("{room_id}\0{first_event_id}\0{}", chat_version())) +} + +/// A scrubbed body cut to `CHAT_BODY_CHARS_MAX` characters on a character boundary, with a marker. +pub fn cut_body(body: &str) -> String { + if body.chars().count() <= CHAT_BODY_CHARS_MAX { + return body.to_owned(); + } + let keep = CHAT_BODY_CHARS_MAX - CUT_MARKER.chars().count(); + body.chars().take(keep).chain(CUT_MARKER.chars()).collect() +} + +fn line_of(message: &ChatMessage) -> String { + let when = DateTime::from_timestamp_millis(i64::try_from(message.ts_ms).unwrap_or(i64::MAX)).map_or_else(String::new, |at| at.format("%Y-%m-%d %H:%M").to_string()); + format!("{when} {}: {}", message.sender, message.body) +} + +/// What is embedded for a chunk: the room's label (its name, else its alias, +/// else its id), then one `YYYY-MM-DD HH:MM sender: body` line per message, UTC. +pub fn embedded_text(label: &str, messages: &[ChatMessage]) -> String { + let mut text = label.to_owned(); + for message in messages { + text.push('\n'); + text.push_str(&line_of(message)); + } + text +} + +/// `messages` (one room, in order) cut into chunks, each a run of consecutive messages. +pub fn chunk_messages(label: &str, messages: &[ChatMessage]) -> Vec> { + let mut chunks: Vec> = Vec::new(); + let mut chars = 0_usize; + for message in messages { + let line = line_of(message).chars().count() + 1; + let closes = chunks.last().is_some_and(|open| { + let last = open.last().map_or(0, |last| last.ts_ms); + message.ts_ms.saturating_sub(last) > CHAT_GAP_MS || open.len() >= CHAT_MESSAGES_MAX || chars + line > CHAT_CHARS_MAX + }); + if closes || chunks.is_empty() { + chunks.push(Vec::new()); + chars = label.chars().count(); + } + chunks.last_mut().expect("a chunk is open").push(message.clone()); + chars += line; + } + chunks +} + +/// The point payload of one chunk. +pub fn chunk_payload(room_id: &str, alias: Option<&str>, label: &str, messages: Vec) -> ChatPayload { + ChatPayload { room_id: room_id.to_owned(), room_alias: alias.map(str::to_owned), text: embedded_text(label, &messages), messages } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn message(n: u64, ts_ms: u64, body: &str) -> ChatMessage { + ChatMessage { event_id: format!("$e{n}"), sender: "@a:s".into(), ts_ms, body: body.into() } + } + + fn ids(chunks: &[Vec]) -> Vec> { + chunks.iter().map(|chunk| chunk.iter().map(|message| message.event_id.as_str()).collect()).collect() + } + + #[test] + fn a_gap_a_count_and_a_character_bound_each_close_a_chunk() { + let gap = [message(1, 0, "a"), message(2, CHAT_GAP_MS, "b"), message(3, 2 * CHAT_GAP_MS + 1, "c")]; + assert_eq!(ids(&chunk_messages("r", &gap)), [["$e1", "$e2"].as_slice(), ["$e3"].as_slice()], "a gap of exactly the bound does not close"); + let many: Vec = (0..25).map(|n| message(n, n, "x")).collect(); + let chunks = chunk_messages("r", &many); + assert_eq!((chunks[0].len(), chunks[1].len()), (CHAT_MESSAGES_MAX, 1)); + let long: Vec = (0..4).map(|n| message(n, n, &"w".repeat(500))).collect(); + let chunks = chunk_messages("r", &long); + assert!(chunks.iter().all(|chunk| embedded_text("r", chunk).chars().count() <= CHAT_CHARS_MAX), "{:?}", ids(&chunks)); + assert!(chunks.len() > 1); + } + + #[test] + fn the_embedded_text_is_the_label_then_utc_lines() { + let text = embedded_text("Swarm", &[message(1, 1_700_000_000_000, "hello")]); + assert_eq!(text, "Swarm\n2023-11-14 22:13 @a:s: hello"); + } + + #[test] + fn a_long_body_is_cut_at_the_bound_with_a_marker_on_a_char_boundary_and_a_short_one_is_whole() { + let cut = cut_body(&"é".repeat(CHAT_BODY_CHARS_MAX + 50)); + assert_eq!(cut.chars().count(), CHAT_BODY_CHARS_MAX); + assert!(cut.ends_with(CUT_MARKER)); + assert_eq!(cut_body(&"é".repeat(CHAT_BODY_CHARS_MAX)), "é".repeat(CHAT_BODY_CHARS_MAX)); + let one = chunk_messages("r", &[message(1, 0, &cut)]); + assert_eq!(one.len(), 1); + } + + #[test] + fn a_chunk_id_is_its_room_first_event_and_version() { + assert_eq!(chunk_id("!r:s", "$a"), chunk_id("!r:s", "$a")); + assert_ne!(chunk_id("!r:s", "$a"), chunk_id("!r:s", "$b")); + assert_ne!(chunk_id("!r:s", "$a"), chunk_id("!q:s", "$a")); + assert_eq!(chat_version(), format!("{CHAT_CHUNKER_VERSION}.{SCRUB_VERSION}")); + } + + /// Chunking the whole sequence equals re-chunking the open tail with each new batch appended. + #[test] + fn appending_to_the_open_tail_gives_the_boundaries_of_chunking_all_at_once() { + let mut seed = 0x2545_f491_4f6c_dd1d_u64; + let mut next = move |bound: u64| { + seed ^= seed << 13; + seed ^= seed >> 7; + seed ^= seed << 17; + seed % bound + }; + let mut ts = 0; + let all: Vec = (0..300) + .map(|n| { + ts += if next(7) == 0 { CHAT_GAP_MS + 1 } else { next(60_000) }; + message(n, ts, &"z".repeat(next(700) as usize + 1)) + }) + .collect(); + let whole = chunk_messages("r", &all); + for step in [1_usize, 5, 17, 100] { + let mut held: Vec> = Vec::new(); + for batch in all.chunks(step) { + let mut tail: Vec = held.pop().unwrap_or_default(); + tail.extend_from_slice(batch); + held.extend(chunk_messages("r", &tail)); + } + assert_eq!(held, whole, "step {step}"); + } + } +} diff --git a/crates/huginn-daemon/src/sources/mod.rs b/crates/huginn-daemon/src/sources/mod.rs index 22c010b..fb7bdbc 100644 --- a/crates/huginn-daemon/src/sources/mod.rs +++ b/crates/huginn-daemon/src/sources/mod.rs @@ -20,9 +20,11 @@ //! The cursors object is written whole by this job alone, once per repo. pub mod chat; +pub mod chat_chunk; pub mod chunk; pub mod forge; pub mod git; +pub mod room_scope; #[cfg(test)] mod tests; diff --git a/crates/huginn-daemon/src/sources/room_scope.rs b/crates/huginn-daemon/src/sources/room_scope.rs new file mode 100644 index 0000000..1c607be --- /dev/null +++ b/crates/huginn-daemon/src/sources/room_scope.rs @@ -0,0 +1,157 @@ +//! Which joined rooms and which senders are in the chat corpus. The only code +//! that decides it, and the only file outside tests that names the bridged +//! platform whose content is refused. +//! +//! Scope (ruling `gamecult-chat:ruling:chat-index-minds-rooms`): the rooms the +//! reader user has joined, minus this exclusion. A room's join rule is read by +//! nobody (`RoomFacts` has no field for it): a joined invite-only room qualifies +//! like any other. The reader roster is the first lock; this file is the second. + +use std::collections::BTreeSet; + +use super::chat::RoomEvent; + +/// The estate rooms by canonical alias, ASCII case-insensitive. Code owns the +/// aliases and the rule; the unit supplies the room ids (`Estate`), and either alone excludes. +pub const ESTATE_ALIASES: [&str; 3] = ["#users:gamecult.org", "#members:gamecult.org", "#patrons:gamecult.org"]; + +/// What the server says about a room, as far as admission needs it. The Matrix +/// client fills it from the room's state and decides nothing. +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub struct RoomFacts { + pub room_id: String, + pub alias: Option, + pub alt_aliases: Vec, + pub name: Option, + /// An `m.room.encryption` state event exists. + pub encrypted: bool, + /// Any `m.room.member` in the state has `is_direct` true. + pub direct: bool, + /// The state type (`m.bridge` or `uk.half-shot.bridge`) of a bridge event, when the room has one. + pub bridge: Option, +} + +/// The estate room ids the caller knows (they exist only on the live server). +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub struct Estate { + pub ids: BTreeSet, +} + +impl Estate { + /// Whether `room_id` is a named estate room. Decided before the room's state is read. + pub fn holds_id(&self, room_id: &str) -> bool { + self.ids.contains(room_id) + } +} + +/// Why a joined room, or an event in it, is refused. +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum Refusal { + Estate, + Direct, + Encrypted, + Bridged(String), + DiscordSender, + DiscordContent, +} + +impl std::fmt::Display for Refusal { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::Estate => formatter.write_str("estate room"), + Self::Direct => formatter.write_str("direct room"), + Self::Encrypted => formatter.write_str("encrypted room"), + Self::Bridged(state_type) => write!(formatter, "bridged room ({state_type})"), + Self::DiscordSender => formatter.write_str("discord sender"), + Self::DiscordContent => formatter.write_str("discord content"), + } + } +} + +/// Whether a joined room may be read at all. +pub fn admit_room(facts: &RoomFacts, estate: &Estate) -> Result<(), Refusal> { + let estate_alias = |alias: &str| ESTATE_ALIASES.iter().any(|estate| estate.eq_ignore_ascii_case(alias)); + if estate.holds_id(&facts.room_id) || facts.alias.iter().chain(&facts.alt_aliases).any(|alias| estate_alias(alias)) { + return Err(Refusal::Estate); + } + if facts.direct { + return Err(Refusal::Direct); + } + if facts.encrypted { + return Err(Refusal::Encrypted); + } + if let Some(state_type) = &facts.bridge { + return Err(Refusal::Bridged(state_type.clone())); + } + Ok(()) +} + +/// Whether an event refuses its whole room: a sender whose localpart, lowercased +/// with leading underscores stripped, starts with the platform's name, or content +/// with a top-level key naming it. Deterministic identifier rules, not language reading. +pub fn refuse_event(event: &RoomEvent) -> Option { + let localpart = event.sender.trim_start_matches('@').split(':').next().unwrap_or(""); + if localpart.to_ascii_lowercase().trim_start_matches('_').starts_with("discord") { + return Some(Refusal::DiscordSender); + } + event.content_keys.iter().any(|key| key.to_ascii_lowercase().contains("discord")).then_some(Refusal::DiscordContent) +} + +#[cfg(test)] +mod tests { + use super::super::chat::EventKind; + use super::*; + + fn facts(id: &str) -> RoomFacts { + RoomFacts { room_id: id.into(), ..RoomFacts::default() } + } + + fn event(sender: &str, keys: &[&str]) -> RoomEvent { + RoomEvent { event_id: "$e".into(), sender: sender.into(), ts_ms: 1, content_keys: keys.iter().map(|key| (*key).into()).collect(), kind: EventKind::Other } + } + + #[test] + fn a_direct_encrypted_or_bridged_room_is_refused_and_a_plain_room_is_admitted() { + let estate = Estate::default(); + assert_eq!(admit_room(&RoomFacts { direct: true, ..facts("!a:s") }, &estate), Err(Refusal::Direct)); + assert_eq!(admit_room(&RoomFacts { encrypted: true, ..facts("!a:s") }, &estate), Err(Refusal::Encrypted)); + for state_type in ["m.bridge", "uk.half-shot.bridge"] { + assert_eq!(admit_room(&RoomFacts { bridge: Some(state_type.into()), ..facts("!a:s") }, &estate), Err(Refusal::Bridged(state_type.into()))); + } + assert_eq!(admit_room(&RoomFacts { alias: Some("#swarm:gamecult.org".into()), name: Some("Swarm".into()), ..facts("!a:s") }, &estate), Ok(())); + } + + #[test] + fn the_estate_is_refused_by_id_and_by_alias_in_any_case_and_a_lookalike_is_not() { + let estate = Estate { ids: ["!ids:s".to_owned()].into() }; + assert!(estate.holds_id("!ids:s") && !estate.holds_id("!other:s")); + assert_eq!(admit_room(&facts("!ids:s"), &estate), Err(Refusal::Estate)); + for alias in ESTATE_ALIASES.iter().map(|alias| alias.to_string()).chain(["#USERS:Gamecult.Org".to_owned()]) { + assert_eq!(admit_room(&RoomFacts { alias: Some(alias.clone()), ..facts("!x:s") }, &Estate::default()), Err(Refusal::Estate), "{alias}"); + assert_eq!(admit_room(&RoomFacts { alt_aliases: vec!["#a:s".into(), alias.clone()], ..facts("!x:s") }, &Estate::default()), Err(Refusal::Estate), "alt {alias}"); + } + for lookalike in ["#users-guide:gamecult.org", "#patrons-club:gamecult.org", "#users:gamecult.org.evil", "#users:other.org"] { + assert_eq!(admit_room(&RoomFacts { alias: Some(lookalike.into()), ..facts("!x:s") }, &Estate::default()), Ok(()), "{lookalike}"); + } + } + + #[test] + fn a_bridged_platform_sender_or_content_key_refuses_and_a_lookalike_does_not() { + for sender in ["@discord_123:s", "@_discord_9:s", "@__DiscordBot:s", "@discordbot:s"] { + assert_eq!(refuse_event(&event(sender, &[])), Some(Refusal::DiscordSender), "{sender}"); + } + for key in ["fi.mau.discord", "uk.half-shot.discord", "X.DISCORD.y"] { + assert_eq!(refuse_event(&event("@a:s", &["body", key])), Some(Refusal::DiscordContent), "{key}"); + } + for sender in ["@discourse:s", "@disco:s", "@eureka:s", "@a_discord:s", "@x:discord.com"] { + assert_eq!(refuse_event(&event(sender, &["discourse", "body"])), None, "{sender}"); + } + } + + #[test] + fn a_room_facts_carries_no_join_rule() { + let source = include_str!("room_scope.rs"); + let code = &source[..source.find("#[cfg(test)]").unwrap()]; + assert!(!code.contains(&["join", "_rule"].concat())); + } +}