sources: room_scope (joined-room admission, estate/DM/encrypted/bridged refusal, sender refusal) and the chat chunker

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

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<String>,
+    pub kind: EventKind,
+}

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<Vec<ChatMessage>> {
+    let mut chunks: Vec<Vec<ChatMessage>> = 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<ChatMessage>) -> 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<ChatMessage>]) -> Vec<Vec<&str>> {
+        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<ChatMessage> = (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<ChatMessage> = (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<ChatMessage> = (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<ChatMessage>> = Vec::new();
+            for batch in all.chunks(step) {
+                let mut tail: Vec<ChatMessage> = held.pop().unwrap_or_default();
+                tail.extend_from_slice(batch);
+                held.extend(chunk_messages("r", &tail));
+            }
+            assert_eq!(held, whole, "step {step}");
+        }
+    }
+}

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;

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<String>,
+    pub alt_aliases: Vec<String>,
+    pub name: Option<String>,
+    /// 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<String>,
+}
+
+/// 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<String>,
+}
+
+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<Refusal> {
+    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()));
+    }
+}