vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(#2479): add from/until event-id cursors to read_room

Browse source
This commit is contained in:
damocles 2026-07-15 18:26:31 +02:00 • committed by mara
commit 001b1dae37
4 changed files with 189 additions and 46 deletions
Download patch file Download diff file Expand all files Collapse all files

25
hive-matrix-mcp/src/bin/mcp.rs
View file

@ -203,6 +203,17 @@ struct ReadRoomArgs {
/// Maximum events to return (default 50, max 200). Newest first.
#[serde(default)]
limit: Option<usize>,
/// 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<String>,
/// 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<String>,
/// 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<ReadRoomArgs>) -> String {
render(
call(
@ -528,6 +545,8 @@ impl MatrixBridge {
DaemonOp::ReadRoom {
room: args.room,
limit: args.limit,
from: args.from,
until: args.until,
},
)
.await,

185
hive-matrix-mcp/src/handlers.rs
View file

@ -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<usize>) -> 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<TimelineEvent> {
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<usize>,
from: Option<String>,
until: Option<String>,
) -> 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<TimelineEvent> = 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<TimelineEvent> = 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<TimelineEvent> = 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<TimelineEvent> = 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)
}

18
hive-matrix-mcp/src/protocol.rs
View file

@ -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<usize> },
ReadRoom {
room: String,
limit: Option<usize>,
#[serde(default)]
from: Option<String>,
#[serde(default)]
until: Option<String>,
},
/// Download the media attachment carried by `event_id` in `room`
/// and write it to a local file (`dest_path`, or a temp file named

7
hive-matrix-mcp/src/socket.rs
View file

@ -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,