diff --git a/hive-matrix-mcp/src/bin/mcp.rs b/hive-matrix-mcp/src/bin/mcp.rs index 9adb8ff6..79e6a99a 100644 --- a/hive-matrix-mcp/src/bin/mcp.rs +++ b/hive-matrix-mcp/src/bin/mcp.rs @@ -203,6 +203,17 @@ struct ReadRoomArgs { /// Maximum events to return (default 50, max 200). Newest first. #[serde(default)] limit: Option, + /// Anchor at this event id and read *forward* from it: returns the anchor + /// plus the `limit - 1` events after it (chronologically newer). Use to + /// continue reading from a known event. Mutually exclusive with `until`. + #[serde(default)] + from: Option, + /// Anchor at this event id and read *backward* to it: returns the anchor + /// plus the `limit - 1` events before it (into the past). Use to page + /// older context around a known event (e.g. the target of a reply). + /// Mutually exclusive with `from`. + #[serde(default)] + until: Option, /// Matrix account to act as (a `name` from `hyperhive.matrixAccounts`). /// Omit to use the agent's primary account. #[serde(default)] @@ -518,9 +529,15 @@ impl MatrixBridge { render(call(args.account, DaemonOp::ListRoomMembers { room: args.room }).await) } - #[tool(description = "Read the most recent N events from a matrix room \ - (default 50, max 200). Returns each event's id, sender, timestamp, \ - type, and best-effort plain-text body.")] + #[tool(description = "Read events from a matrix room (default 50, max 200), \ + newest first. Returns each event's id, sender, timestamp, type, \ + best-effort plain-text body, and the id of the event it replies to \ + (when any). By default returns the most recent events. Pass `from` \ + = an event id to read forward from that event (anchor + newer \ + events), or `until` = an event id to read backward to it (anchor + \ + older events) — e.g. to fetch the context around a reply target \ + outside the current window. `from` and `until` are mutually \ + exclusive.")] async fn read_room(&self, Parameters(args): Parameters) -> String { render( call( @@ -528,6 +545,8 @@ impl MatrixBridge { DaemonOp::ReadRoom { room: args.room, limit: args.limit, + from: args.from, + until: args.until, }, ) .await, diff --git a/hive-matrix-mcp/src/handlers.rs b/hive-matrix-mcp/src/handlers.rs index 07b9a2dc..31f472d0 100644 --- a/hive-matrix-mcp/src/handlers.rs +++ b/hive-matrix-mcp/src/handlers.rs @@ -626,51 +626,160 @@ pub async fn invite_user(client: &Client, room_ref: &str, user_id: &str) -> Daem } } -pub async fn read_room(client: &Client, room_ref: &str, limit: Option) -> DaemonResponse { +/// Map a matrix-sdk timeline event to the JSON DTO, applying the +/// encryption-gated UTD sentinel. Returns `None` when the raw event can't +/// be deserialized. Shared by `read_room`'s plain-timeline and the +/// event-anchored (`from` / `until`) paths. +/// +/// `room.messages()` / `event_with_context` transparently decrypt events in +/// encrypted rooms. UTD (unable to decrypt) events surface via +/// `ev.kind.is_utd()`, but that flag can transiently fire in a NON-encrypted +/// room when the SDK hasn't finished reclassifying a freshly-synced +/// `m.room.encrypted` cache row — so gate the sentinel on the room actually +/// being encrypted to avoid a false `[unable to decrypt]` on plaintext. +fn to_timeline_event( + ev: &matrix_sdk::deserialized_responses::TimelineEvent, + is_encrypted: bool, +) -> Option { + let parsed = ev.raw().deserialize().ok()?; + let body = if is_encrypted && ev.kind.is_utd() { + "[unable to decrypt]".to_owned() + } else { + extract_body_sync(&parsed) + }; + Some(TimelineEvent { + event_id: parsed.event_id().to_string(), + sender: parsed.sender().to_string(), + origin_server_ts: parsed.origin_server_ts().0.into(), + event_type: parsed.event_type().to_string(), + body, + in_reply_to_event_id: extract_in_reply_to(&parsed), + }) +} + +/// Read room timeline events. With neither cursor set, returns the last +/// `limit` events (newest-first) — the default. `from` / `until` anchor the +/// read at an event id (mutually exclusive): +/// - `until = event_id` → the anchor plus the `limit - 1` events *before* it +/// (page into the past from a known event). +/// - `from = event_id` → the anchor plus the `limit - 1` events *after* it +/// (continue reading forward from a known point). +/// +/// Both cursors resolve the anchor via the `/context` endpoint +/// (`event_with_context`) to get a pagination token, then page precisely in +/// the requested direction — matrix `/messages` `from` is an opaque token, +/// not an event id, so plain `messages()` can't anchor at an event. All +/// three modes return events newest-first for a consistent shape. +pub async fn read_room( + client: &Client, + room_ref: &str, + limit: Option, + from: Option, + until: Option, +) -> DaemonResponse { use matrix_sdk::room::MessagesOptions; + use matrix_sdk::ruma::UInt; let room = match resolve_room(client, room_ref).await { Ok(r) => r, Err(e) => return e, }; - let limit_val = u32::try_from(limit.unwrap_or(50).min(200)).unwrap_or(50); - let mut opts = MessagesOptions::backward(); - opts.limit = matrix_sdk::ruma::UInt::from(limit_val); - // room.messages() transparently decrypts events in encrypted rooms. - // UTD (unable to decrypt) events surface via `ev.kind.is_utd()`, but - // that flag can transiently fire in a NON-encrypted room when the SDK - // hasn't finished reclassifying a freshly-synced `m.room.encrypted` - // cache row — so gate the sentinel on the room actually being - // encrypted to avoid a false `[unable to decrypt]` on plaintext. - let msgs = match room.messages(opts).await { - Ok(m) => m, - Err(e) => return DaemonResponse::error(format!("messages: {e}")), - }; + if from.is_some() && until.is_some() { + return DaemonResponse::error( + "read_room: `from` and `until` are mutually exclusive \ + (they page opposite directions off the anchor event)" + .to_owned(), + ); + } + // Clamp to [1, 200]; keep it in u32 so every `UInt` conversion below is + // infallible (`limit_val - 1` can't underflow — the clamp floor is 1). + let limit_val: u32 = u32::try_from(limit.unwrap_or(50)) + .unwrap_or(50) + .clamp(1, 200); let is_encrypted = room.encryption_state().is_encrypted(); - let events: Vec = msgs - .chunk - .iter() - .filter_map(|ev| { - let parsed = ev.raw().deserialize().ok()?; - let event_id = parsed.event_id().to_string(); - let sender = parsed.sender().to_string(); - let origin_server_ts: i64 = parsed.origin_server_ts().0.into(); - let event_type = parsed.event_type().to_string(); - let body = if is_encrypted && ev.kind.is_utd() { - "[unable to decrypt]".to_owned() - } else { - extract_body_sync(&parsed) - }; - Some(TimelineEvent { - event_id, - sender, - origin_server_ts, - event_type, - body, - in_reply_to_event_id: extract_in_reply_to(&parsed), - }) - }) - .collect(); + + let events: Vec = if let Some(anchor_ref) = from.as_deref().or(until.as_deref()) + { + let eid: OwnedEventId = match anchor_ref.parse() { + Ok(e) => e, + Err(e) => return DaemonResponse::error(format!("invalid event_id {anchor_ref}: {e}")), + }; + // context_size 0: no context events, so the returned tokens point + // immediately around the anchor and the follow-up page is contiguous. + let ctx = match room + .event_with_context(&eid, false, UInt::from(0u32), None) + .await + { + Ok(c) => c, + Err(e) => return DaemonResponse::error(format!("event_with_context {eid}: {e}")), + }; + let Some(anchor) = ctx.event else { + return DaemonResponse::error(format!("event {eid} not found in room")); + }; + let side_limit = UInt::from(limit_val - 1); + if from.is_some() { + // Forward (newer) from the anchor. The forward chunk is + // oldest-first; reverse to newest-first with the anchor (oldest of + // the slice) last. When the anchor is at the live edge there is no + // forward token — return the anchor alone rather than pass + // `from = None` to messages(), which would page from room *start*. + let mut out: Vec = Vec::new(); + if let Some(token) = ctx.next_batch_token { + let mut opts = MessagesOptions::forward(); + opts.from = Some(token); + opts.limit = side_limit; + let newer = match room.messages(opts).await { + Ok(m) => m, + Err(e) => return DaemonResponse::error(format!("messages (from): {e}")), + }; + out.extend( + newer + .chunk + .iter() + .rev() + .filter_map(|ev| to_timeline_event(ev, is_encrypted)), + ); + } + out.extend(to_timeline_event(&anchor, is_encrypted)); + out + } else { + // Backward (older) to the anchor. Backward chunk is already + // newest-first; the anchor is the newest event, so it leads. When + // the anchor is the oldest event there is no backward token — + // return the anchor alone rather than pass `from = None`, which + // would page from room *end* (the newest events). + let mut out: Vec = to_timeline_event(&anchor, is_encrypted) + .into_iter() + .collect(); + if let Some(token) = ctx.prev_batch_token { + let mut opts = MessagesOptions::backward(); + opts.from = Some(token); + opts.limit = side_limit; + let older = match room.messages(opts).await { + Ok(m) => m, + Err(e) => return DaemonResponse::error(format!("messages (until): {e}")), + }; + out.extend( + older + .chunk + .iter() + .filter_map(|ev| to_timeline_event(ev, is_encrypted)), + ); + } + out + } + } else { + let mut opts = MessagesOptions::backward(); + opts.limit = UInt::from(limit_val); + let msgs = match room.messages(opts).await { + Ok(m) => m, + Err(e) => return DaemonResponse::error(format!("messages: {e}")), + }; + msgs.chunk + .iter() + .filter_map(|ev| to_timeline_event(ev, is_encrypted)) + .collect() + }; DaemonResponse::ok(&events) } diff --git a/hive-matrix-mcp/src/protocol.rs b/hive-matrix-mcp/src/protocol.rs index 7c45f48a..2642c69b 100644 --- a/hive-matrix-mcp/src/protocol.rs +++ b/hive-matrix-mcp/src/protocol.rs @@ -114,11 +114,21 @@ pub enum DaemonOp { #[serde(rename = "list_room_members")] ListRoomMembers { room: String }, - /// Read the last `limit` events from a room's timeline. Caller - /// gets each event's id, sender, `server_ts`, type, and body (best- - /// effort plain-text extraction from `m.text` / `m.notice` etc.). + /// Read events from a room's timeline. Caller gets each event's id, + /// sender, `server_ts`, type, and body (best-effort plain-text extraction + /// from `m.text` / `m.notice` etc.). With neither cursor, returns the last + /// `limit` events (newest-first). `from` / `until` anchor at an event id + /// (mutually exclusive): `until` reads the anchor + `limit-1` events before + /// it (into the past); `from` reads the anchor + `limit-1` events after it. #[serde(rename = "read_room")] - ReadRoom { room: String, limit: Option }, + ReadRoom { + room: String, + limit: Option, + #[serde(default)] + from: Option, + #[serde(default)] + until: Option, + }, /// Download the media attachment carried by `event_id` in `room` /// and write it to a local file (`dest_path`, or a temp file named diff --git a/hive-matrix-mcp/src/socket.rs b/hive-matrix-mcp/src/socket.rs index 165a1793..294944f4 100644 --- a/hive-matrix-mcp/src/socket.rs +++ b/hive-matrix-mcp/src/socket.rs @@ -126,7 +126,12 @@ async fn dispatch_op(op: DaemonOp, client: &Client) -> DaemonResponse { handlers::invite_user(client, &room, &user_id).await } DaemonOp::ListRoomMembers { room } => handlers::list_room_members(client, &room).await, - DaemonOp::ReadRoom { room, limit } => handlers::read_room(client, &room, limit).await, + DaemonOp::ReadRoom { + room, + limit, + from, + until, + } => handlers::read_room(client, &room, limit, from, until).await, DaemonOp::DownloadFile { room, event_id,