diff --git a/hive-matrix-mcp/src/handlers.rs b/hive-matrix-mcp/src/handlers.rs index 2c47d980..0ba507f9 100644 --- a/hive-matrix-mcp/src/handlers.rs +++ b/hive-matrix-mcp/src/handlers.rs @@ -290,28 +290,53 @@ async fn room_unread_state( .map(|(id, _)| id); let is_encrypted = room.encryption_state().is_encrypted(); - let events = recent_events(room).await; - for raw in &events { - let Ok(event) = raw.raw().deserialize() else { + let events: Vec = recent_events(room) + .await + .iter() + .map(|raw| { + raw.raw() + .deserialize() + .ok() + .map(|event| (event, raw.kind.is_utd())) + }) + .collect(); + unread_verdict(own_user_id, read_event_id.as_deref(), is_encrypted, &events) +} + +/// One timeline event as `unread_verdict` sees it: the parsed event and +/// whether the SDK flagged it undecryptable, or `None` when it would not +/// parse. +type ParsedEvent = Option<(matrix_sdk::ruma::events::AnySyncTimelineEvent, bool)>; + +/// The decision half of [`room_unread_state`], on plain data so it needs no +/// client or room. `events` is newest first. +fn unread_verdict( + own_user_id: &matrix_sdk::ruma::UserId, + read_event_id: Option<&matrix_sdk::ruma::EventId>, + is_encrypted: bool, + events: &[ParsedEvent], +) -> Option<(OwnedEventId, OwnedUserId, String)> { + for slot in events { + let Some((event, is_utd)) = slot else { // Unparseable — same as the old filter-then-scan behaviour: // skip it rather than let it match the read receipt or count // as the fallback's "newest". continue; }; - if read_event_id.as_deref() == Some(event.event_id()) { + if read_event_id == Some(event.event_id()) { // Scanned back to the agent's own read receipt without // hitting anything intentional first — proven read, not // guessed: there cannot be an unread intentional event in a // range we've fully walked. return None; } - if !is_intentional(&event) { + if !is_intentional(event) { continue; } if event.sender().as_str() == own_user_id.as_str() { return None; } - let body = utd_gated_body(is_encrypted, raw.kind.is_utd(), &event); + let body = utd_gated_body(is_encrypted, *is_utd, event); return Some((event.event_id().to_owned(), event.sender().to_owned(), body)); } // Exhausted the lookback window without hitting the receipt or an @@ -322,12 +347,11 @@ async fn room_unread_state( // false "read" risks silently swallowing a real message. Expected // to be rare now that the state-churn source this fix targets is // itself bounded. - let newest_raw = events.first()?; - let newest = newest_raw.raw().deserialize().ok()?; + let (newest, is_utd) = events.first()?.as_ref()?; if newest.sender().as_str() == own_user_id.as_str() { return None; } - let body = utd_gated_body(is_encrypted, newest_raw.kind.is_utd(), &newest); + let body = utd_gated_body(is_encrypted, *is_utd, newest); Some(( newest.event_id().to_owned(), newest.sender().to_owned(), @@ -1110,3 +1134,300 @@ mod utd_body_tests { assert_eq!(got, "hello"); } } + +#[cfg(test)] +mod unread_tests { + use super::{ + ParsedEvent, attachment_marker, extract_body_sync, extract_in_reply_to, is_intentional, + unread_verdict, + }; + use matrix_sdk::ruma::events::AnySyncTimelineEvent; + use matrix_sdk::ruma::events::room::message::MessageType; + use matrix_sdk::ruma::{EventId, UserId, user_id}; + use serde_json::json; + + const ME: &str = "@me:example.org"; + + fn parse(value: serde_json::Value) -> AnySyncTimelineEvent { + serde_json::from_value(value).expect("valid sync event") + } + + fn message(id: &str, sender: &str, body: &str) -> AnySyncTimelineEvent { + parse(json!({ + "type": "m.room.message", + "event_id": id, + "sender": sender, + "origin_server_ts": 1, + "content": {"msgtype": "m.text", "body": body}, + })) + } + + fn space_child(id: &str, sender: &str) -> AnySyncTimelineEvent { + parse(json!({ + "type": "m.space.child", + "state_key": "!child:example.org", + "event_id": id, + "sender": sender, + "origin_server_ts": 1, + "content": {"via": ["example.org"]}, + })) + } + + fn live(event: AnySyncTimelineEvent) -> (AnySyncTimelineEvent, bool) { + (event, false) + } + + fn verdict(read: Option<&str>, events: &[ParsedEvent]) -> Option<(String, String, String)> { + let me: &UserId = user_id!("@me:example.org"); + let read = read.map(|r| <&EventId>::try_from(r).expect("event id")); + unread_verdict(me, read, false, events) + .map(|(id, sender, body)| (id.to_string(), sender.to_string(), body)) + } + + #[test] + fn an_empty_room_is_caught_up() { + assert_eq!(verdict(None, &[]), None); + } + + #[test] + fn a_message_from_someone_else_past_the_receipt_is_unread() { + let events = [ + Some(live(message("$new:example.org", "@bob:example.org", "hi"))), + Some(live(message( + "$old:example.org", + "@bob:example.org", + "older", + ))), + ]; + assert_eq!( + verdict(Some("$old:example.org"), &events), + Some(( + "$new:example.org".into(), + "@bob:example.org".into(), + "hi".into() + )) + ); + } + + #[test] + fn a_receipt_on_the_newest_event_means_caught_up() { + let events = [Some(live(message( + "$a:example.org", + "@bob:example.org", + "hi", + )))]; + assert_eq!(verdict(Some("$a:example.org"), &events), None); + } + + #[test] + fn the_agents_own_latest_message_never_counts_as_unread() { + // A read receipt that lags behind the agent's own just-sent + // message must not wake the agent for it. + let events = [Some(live(message("$mine:example.org", ME, "sent")))]; + assert_eq!(verdict(None, &events), None); + } + + #[test] + fn a_space_child_re_emit_is_skipped_to_reach_the_real_message() { + let events = [ + Some(live(space_child("$churn:example.org", "@bot:example.org"))), + Some(live(message( + "$real:example.org", + "@bob:example.org", + "hello", + ))), + ]; + assert_eq!( + verdict(None, &events), + Some(( + "$real:example.org".into(), + "@bob:example.org".into(), + "hello".into() + )) + ); + } + + #[test] + fn the_walk_stops_at_the_receipt_before_an_older_intentional_event() { + // Churn on top, then the receipt, then an older message: the + // receipt proves everything below it was read. + let events = [ + Some(live(space_child("$churn:example.org", "@bot:example.org"))), + Some(live(message( + "$read:example.org", + "@bob:example.org", + "read", + ))), + Some(live(message( + "$older:example.org", + "@bob:example.org", + "older", + ))), + ]; + assert_eq!(verdict(Some("$read:example.org"), &events), None); + } + + #[test] + fn a_window_of_only_churn_falls_back_to_the_newest_event() { + // The receipt is further back than the lookback: unknown, so the + // newest event is surfaced (a spare wake beats a swallowed message). + let events = [Some(live(space_child( + "$churn:example.org", + "@bot:example.org", + )))]; + assert_eq!( + verdict(None, &events), + Some(( + "$churn:example.org".into(), + "@bot:example.org".into(), + String::new() + )) + ); + } + + #[test] + fn an_own_authored_newest_event_in_a_churn_window_is_not_unread() { + let events = [Some(live(space_child("$churn:example.org", ME)))]; + assert_eq!(verdict(None, &events), None); + } + + #[test] + fn an_unparseable_event_is_skipped_not_taken_for_the_newest() { + let events = [ + None, + Some(live(message( + "$real:example.org", + "@bob:example.org", + "hello", + ))), + ]; + assert_eq!( + verdict(None, &events), + Some(( + "$real:example.org".into(), + "@bob:example.org".into(), + "hello".into() + )) + ); + } + + #[test] + fn an_undecryptable_event_in_an_encrypted_room_gets_a_per_event_sentinel() { + let me: &UserId = user_id!("@me:example.org"); + let events: [ParsedEvent; 1] = [Some(( + message("$utd:example.org", "@bob:example.org", "ciphertext"), + true, + ))]; + let (_, _, body) = unread_verdict(me, None, true, &events).expect("unread"); + assert_eq!(body, "[unable to decrypt] ($utd:example.org)"); + } + + #[test] + fn only_the_space_child_re_emit_is_unintentional() { + assert!(!is_intentional(&space_child( + "$c:example.org", + "@bot:example.org" + ))); + assert!(is_intentional(&message( + "$m:example.org", + "@bob:example.org", + "x" + ))); + let topic = parse(json!({ + "type": "m.room.topic", + "state_key": "", + "event_id": "$t:example.org", + "sender": "@bob:example.org", + "origin_server_ts": 1, + "content": {"topic": "new"}, + })); + assert!(is_intentional(&topic), "state changes are real activity"); + } + + fn msgtype(value: &serde_json::Value) -> MessageType { + MessageType::new( + value["msgtype"].as_str().expect("msgtype"), + value["body"].as_str().expect("body").to_owned(), + value.as_object().expect("object").clone(), + ) + .expect("known msgtype") + } + + #[test] + fn attachments_render_as_typed_markers_and_text_as_itself() { + let marker = |kind: &str, extra: serde_json::Value| { + let mut v = json!({"msgtype": kind, "body": "name.bin"}); + v.as_object_mut() + .expect("object") + .extend(extra.as_object().expect("object").clone()); + attachment_marker(&msgtype(&v)) + }; + assert_eq!(marker("m.text", json!({})), "name.bin"); + assert_eq!(marker("m.notice", json!({})), "name.bin"); + assert_eq!(marker("m.emote", json!({})), "* name.bin"); + assert_eq!( + marker("m.file", json!({"url": "mxc://example.org/a"})), + "[file: name.bin]" + ); + assert_eq!( + marker("m.image", json!({"url": "mxc://example.org/a"})), + "[image: name.bin]" + ); + assert_eq!( + marker("m.audio", json!({"url": "mxc://example.org/a"})), + "[audio: name.bin]" + ); + assert_eq!( + marker("m.video", json!({"url": "mxc://example.org/a"})), + "[video: name.bin]" + ); + } + + #[test] + fn a_non_message_event_has_no_body() { + assert_eq!( + extract_body_sync(&space_child("$c:example.org", "@bot:example.org")), + "" + ); + assert_eq!( + extract_body_sync(&message("$m:example.org", "@bob:example.org", "hello")), + "hello" + ); + } + + #[test] + fn reply_ids_come_from_direct_replies_and_thread_fallbacks_only() { + let with_relation = |relation: serde_json::Value| { + parse(json!({ + "type": "m.room.message", + "event_id": "$r:example.org", + "sender": "@bob:example.org", + "origin_server_ts": 1, + "content": {"msgtype": "m.text", "body": "x", "m.relates_to": relation}, + })) + }; + assert_eq!( + extract_in_reply_to(&with_relation( + json!({"m.in_reply_to": {"event_id": "$target:example.org"}}) + )), + Some("$target:example.org".to_owned()) + ); + assert_eq!( + extract_in_reply_to(&with_relation(json!({ + "rel_type": "m.thread", + "event_id": "$root:example.org", + "is_falling_back": true, + "m.in_reply_to": {"event_id": "$fallback:example.org"}, + }))), + Some("$fallback:example.org".to_owned()) + ); + assert_eq!( + extract_in_reply_to(&message("$m:example.org", "@bob:example.org", "plain")), + None + ); + assert_eq!( + extract_in_reply_to(&space_child("$c:example.org", "@bot:example.org")), + None + ); + } +}