964 lines
37 KiB
Rust
964 lines
37 KiB
Rust
//! Per-tool dispatch — each MCP tool call in [`crate::mcp`] resolves to
|
|
//! one of these handlers. Returns a [`DaemonResponse`] which
|
|
//! `crate::mcp::render` turns into the tool-result string claude sees
|
|
//! (`Ok { payload }` → pretty JSON, `Error { message }` → a "matrix
|
|
//! error: …" prefixed string).
|
|
//!
|
|
//! Tool surface mirrors damocles-daemon's v0 set per the operator's call:
|
|
//! `send_message`, `send_dm`, `send_reaction`, `send_reply`, `mark_read`,
|
|
//! `send_redact`, `list_rooms`, `list_room_members`, `read_room`.
|
|
|
|
use matrix_sdk::{
|
|
Client,
|
|
room::reply::{EnforceThread, Reply},
|
|
ruma::{
|
|
OwnedEventId, OwnedRoomId, OwnedServerName, OwnedUserId, RoomOrAliasId,
|
|
api::client::receipt::create_receipt::v3::ReceiptType,
|
|
events::{
|
|
reaction::ReactionEventContent,
|
|
receipt::ReceiptThread,
|
|
relation::Annotation,
|
|
room::message::{AddMentions, MessageType, RoomMessageEventContent},
|
|
},
|
|
},
|
|
};
|
|
use serde::Serialize;
|
|
|
|
use crate::protocol::{DaemonResponse, InviteAction};
|
|
|
|
/// JSON-shape for `list_invites`: one row per pending room invite.
|
|
#[derive(Debug, Serialize)]
|
|
pub struct InviteInfo {
|
|
pub room_id: String,
|
|
pub canonical_alias: Option<String>,
|
|
pub name: Option<String>,
|
|
}
|
|
|
|
/// JSON-shape for `list_rooms`: one row per joined room.
|
|
#[derive(Debug, Serialize)]
|
|
pub struct RoomInfo {
|
|
pub room_id: String,
|
|
pub canonical_alias: Option<String>,
|
|
pub name: String,
|
|
pub member_count: u64,
|
|
}
|
|
|
|
/// JSON-shape for `list_room_members`: one row per joined member.
|
|
#[derive(Debug, Serialize)]
|
|
pub struct MemberInfo {
|
|
pub user_id: String,
|
|
pub display_name: Option<String>,
|
|
}
|
|
|
|
/// JSON-shape for `read_room`: one timeline event, flattened for
|
|
/// claude consumption.
|
|
#[derive(Debug, Serialize)]
|
|
pub struct TimelineEvent {
|
|
pub event_id: String,
|
|
pub sender: String,
|
|
pub origin_server_ts: i64,
|
|
pub event_type: String,
|
|
/// Best-effort plain-text body for `m.room.message` events
|
|
/// (text / notice / emote); empty for events without a textual
|
|
/// body (state changes, reactions, etc.).
|
|
pub body: String,
|
|
/// Event id of the message this replies to (`m.relates_to` →
|
|
/// `m.in_reply_to`), including the reply fallback a thread message
|
|
/// carries. `None` for top-level (non-reply) messages.
|
|
#[serde(skip_serializing_if = "Option::is_none")]
|
|
pub in_reply_to_event_id: Option<String>,
|
|
}
|
|
|
|
/// Resolve a room reference (id `!abc:server` OR alias `#name:server`)
|
|
/// to a joined `Room`. Returns an `Error` response if the room isn't
|
|
/// joined / the reference is malformed.
|
|
async fn resolve_room(
|
|
client: &Client,
|
|
reference: &str,
|
|
) -> Result<matrix_sdk::Room, DaemonResponse> {
|
|
let parsed: &RoomOrAliasId = reference
|
|
.try_into()
|
|
.map_err(|e| DaemonResponse::error(format!("invalid room reference {reference}: {e}")))?;
|
|
let room_id: OwnedRoomId = if parsed.is_room_id() {
|
|
OwnedRoomId::try_from(reference)
|
|
.map_err(|e| DaemonResponse::error(format!("invalid room_id: {e}")))?
|
|
} else {
|
|
client
|
|
.resolve_room_alias(
|
|
parsed
|
|
.as_str()
|
|
.try_into()
|
|
.map_err(|e| DaemonResponse::error(format!("invalid alias: {e}")))?,
|
|
)
|
|
.await
|
|
.map(|r| r.room_id)
|
|
.map_err(|e| DaemonResponse::error(format!("resolve_room_alias {reference}: {e}")))?
|
|
};
|
|
client
|
|
.get_room(&room_id)
|
|
.ok_or_else(|| DaemonResponse::error(format!("room {room_id} not joined")))
|
|
}
|
|
|
|
/// Best-effort plain-text body for a (non-sync) timeline event.
|
|
/// Returns "" for non-text events (state changes, reactions,
|
|
/// redactions) — claude can still see the `event_type` field to
|
|
/// disambiguate. `AnyTimelineEvent` (not `AnySync...`) because
|
|
/// `read_room` pulls events via the `/messages` endpoint which
|
|
/// returns the full-form variant.
|
|
fn extract_body(event: &matrix_sdk::ruma::events::AnyTimelineEvent) -> String {
|
|
use matrix_sdk::ruma::events::{AnyMessageLikeEvent, AnyTimelineEvent};
|
|
match event {
|
|
AnyTimelineEvent::MessageLike(AnyMessageLikeEvent::RoomMessage(ev)) => ev
|
|
.as_original()
|
|
.map_or_else(String::new, |orig| attachment_marker(&orig.content.msgtype)),
|
|
_ => String::new(),
|
|
}
|
|
}
|
|
|
|
/// Best-effort plain-text body / attachment marker for a message
|
|
/// `MessageType`. Text-like messages return their body; media messages
|
|
/// (`m.file` / `m.image` / `m.audio` / `m.video`) return a
|
|
/// `[file: name]`-style marker so an agent reading the room sees that an
|
|
/// attachment is present (and can fetch it with `download_file`); all
|
|
/// other types return "".
|
|
fn attachment_marker(msgtype: &MessageType) -> String {
|
|
match msgtype {
|
|
MessageType::Text(t) => t.body.clone(),
|
|
MessageType::Notice(n) => n.body.clone(),
|
|
MessageType::Emote(e) => format!("* {}", e.body),
|
|
MessageType::File(f) => format!("[file: {}]", f.body),
|
|
MessageType::Image(i) => format!("[image: {}]", i.body),
|
|
MessageType::Audio(a) => format!("[audio: {}]", a.body),
|
|
MessageType::Video(v) => format!("[video: {}]", v.body),
|
|
_ => String::new(),
|
|
}
|
|
}
|
|
|
|
/// Same as [`extract_body`] but for the `AnySyncTimelineEvent` variant
|
|
/// returned by `TimelineEvent::raw()` (the sdk-level decrypted form from
|
|
/// `room.messages()`).
|
|
fn extract_body_sync(event: &matrix_sdk::ruma::events::AnySyncTimelineEvent) -> String {
|
|
use matrix_sdk::ruma::events::{AnySyncMessageLikeEvent, AnySyncTimelineEvent};
|
|
match event {
|
|
AnySyncTimelineEvent::MessageLike(AnySyncMessageLikeEvent::RoomMessage(ev)) => ev
|
|
.as_original()
|
|
.map_or_else(String::new, |orig| attachment_marker(&orig.content.msgtype)),
|
|
_ => String::new(),
|
|
}
|
|
}
|
|
|
|
/// Extract the event id this message replies to, from its `m.relates_to`.
|
|
/// Handles both a direct `m.in_reply_to` reply and the reply fallback a
|
|
/// thread message carries. Returns `None` for non-reply messages and for
|
|
/// non-`m.room.message` events.
|
|
fn extract_in_reply_to(event: &matrix_sdk::ruma::events::AnySyncTimelineEvent) -> Option<String> {
|
|
use matrix_sdk::ruma::events::room::message::Relation;
|
|
use matrix_sdk::ruma::events::{AnySyncMessageLikeEvent, AnySyncTimelineEvent};
|
|
let AnySyncTimelineEvent::MessageLike(AnySyncMessageLikeEvent::RoomMessage(ev)) = event else {
|
|
return None;
|
|
};
|
|
match ev.as_original()?.content.relates_to.as_ref()? {
|
|
Relation::Reply(reply) => Some(reply.in_reply_to.event_id.to_string()),
|
|
Relation::Thread(thread) => thread.in_reply_to.as_ref().map(|r| r.event_id.to_string()),
|
|
_ => None,
|
|
}
|
|
}
|
|
|
|
/// Human-facing label for a room: its canonical alias when set,
|
|
/// otherwise the raw room id. Used wherever a room is named in
|
|
/// agent-facing text (wake bodies, loose-ends, the unread guard).
|
|
#[must_use]
|
|
pub fn room_label(room: &matrix_sdk::Room) -> String {
|
|
room.canonical_alias()
|
|
.map_or_else(|| room.room_id().to_string(), |a| a.to_string())
|
|
}
|
|
|
|
/// Refuse to post into a room the agent hasn't caught up on. Returns
|
|
/// `Some(error)` with a helpful hint when the room still has unread
|
|
/// notifications (the agent must `read_room` then `mark_read` the
|
|
/// latest event first), or `None` when the send may proceed.
|
|
///
|
|
/// Read-state is the matrix unread-notification count, the same signal
|
|
/// the wake path and `get_loose_ends` use, so "caught up" here means
|
|
/// exactly what those surfaces mean. Reactions and `mark_read` are not
|
|
/// gated — only message-posting tools (`send_message`, `send_reply`,
|
|
/// `send_dm`) so an agent can't talk over messages it hasn't seen.
|
|
fn unread_guard(room: &matrix_sdk::Room) -> Option<DaemonResponse> {
|
|
let count = room.unread_notification_counts().notification_count;
|
|
if count == 0 {
|
|
return None;
|
|
}
|
|
let label = room_label(room);
|
|
Some(DaemonResponse::error(format!(
|
|
"refusing to send: {count} unread message(s) in {label}. \
|
|
use read_room to view them, then mark_read the latest event before \
|
|
sending so you don't talk over messages you haven't seen."
|
|
)))
|
|
}
|
|
|
|
pub async fn send_message(client: &Client, room_ref: &str, body: &str) -> DaemonResponse {
|
|
let room = match resolve_room(client, room_ref).await {
|
|
Ok(r) => r,
|
|
Err(e) => return e,
|
|
};
|
|
if let Some(reject) = unread_guard(&room) {
|
|
return reject;
|
|
}
|
|
let content = RoomMessageEventContent::text_markdown(body);
|
|
match room.send(content).await {
|
|
Ok(resp) => DaemonResponse::ok(&serde_json::json!({
|
|
"event_id": resp.response.event_id.to_string(),
|
|
"room_id": room.room_id().to_string(),
|
|
})),
|
|
Err(e) => DaemonResponse::error(format!("send to {}: {e}", room.room_id())),
|
|
}
|
|
}
|
|
|
|
/// Find the existing DM room with `user_id` or create one. Shared by
|
|
/// `send_dm` and `open_dm`. `direct_targets()` (cached state) is
|
|
/// used rather than the async `is_direct()`.
|
|
async fn resolve_or_create_dm(
|
|
client: &Client,
|
|
user_id: &str,
|
|
) -> Result<matrix_sdk::Room, DaemonResponse> {
|
|
let uid: OwnedUserId = user_id
|
|
.parse()
|
|
.map_err(|e| DaemonResponse::error(format!("invalid user_id {user_id}: {e}")))?;
|
|
if let Some(room) = client.joined_rooms().into_iter().find(|r| {
|
|
r.direct_targets()
|
|
.iter()
|
|
.any(|t| t.as_str() == uid.as_str())
|
|
}) {
|
|
return Ok(room);
|
|
}
|
|
client
|
|
.create_dm(&uid)
|
|
.await
|
|
.map_err(|e| DaemonResponse::error(format!("create_dm {uid}: {e}")))
|
|
}
|
|
|
|
pub async fn send_dm(client: &Client, user_id: &str, body: &str) -> DaemonResponse {
|
|
let room = match resolve_or_create_dm(client, user_id).await {
|
|
Ok(r) => r,
|
|
Err(e) => return e,
|
|
};
|
|
if let Some(reject) = unread_guard(&room) {
|
|
return reject;
|
|
}
|
|
let content = RoomMessageEventContent::text_markdown(body);
|
|
match room.send(content).await {
|
|
Ok(resp) => DaemonResponse::ok(&serde_json::json!({
|
|
"event_id": resp.response.event_id.to_string(),
|
|
"room_id": room.room_id().to_string(),
|
|
"user_id": user_id,
|
|
})),
|
|
Err(e) => DaemonResponse::error(format!("send DM to {user_id}: {e}")),
|
|
}
|
|
}
|
|
|
|
/// Maximum file size accepted by `send_file`.
|
|
const MAX_UPLOAD_BYTES: u64 = 50 * 1024 * 1024;
|
|
|
|
/// Read a local file and post it as a matrix attachment to `room`,
|
|
/// inferring the MIME type from the file extension. An optional
|
|
/// `caption` is sent as a best-effort follow-up text message (the
|
|
/// attachment has already landed, so a caption failure is non-fatal).
|
|
/// Used by `send_file`.
|
|
async fn upload_attachment(
|
|
room: &matrix_sdk::Room,
|
|
path: &str,
|
|
caption: Option<&str>,
|
|
) -> DaemonResponse {
|
|
let p = std::path::Path::new(path);
|
|
let meta = match tokio::fs::metadata(p).await {
|
|
Ok(m) => m,
|
|
Err(e) => return DaemonResponse::error(format!("stat {path}: {e}")),
|
|
};
|
|
if !meta.is_file() {
|
|
return DaemonResponse::error(format!("{path} is not a regular file"));
|
|
}
|
|
if meta.len() > MAX_UPLOAD_BYTES {
|
|
return DaemonResponse::error(format!(
|
|
"{path} is {} bytes, over the {MAX_UPLOAD_BYTES}-byte upload cap",
|
|
meta.len()
|
|
));
|
|
}
|
|
let data = match tokio::fs::read(p).await {
|
|
Ok(d) => d,
|
|
Err(e) => return DaemonResponse::error(format!("read {path}: {e}")),
|
|
};
|
|
let filename = p
|
|
.file_name()
|
|
.and_then(|n| n.to_str())
|
|
.unwrap_or("file")
|
|
.to_owned();
|
|
let content_type: mime::Mime = mime_guess::from_path(p).first_or_octet_stream();
|
|
let resp = match room
|
|
.send_attachment(
|
|
filename.clone(),
|
|
&content_type,
|
|
data,
|
|
matrix_sdk::attachment::AttachmentConfig::new(),
|
|
)
|
|
.await
|
|
{
|
|
Ok(r) => r,
|
|
Err(e) => {
|
|
return DaemonResponse::error(format!("send attachment to {}: {e}", room.room_id()));
|
|
}
|
|
};
|
|
if let Some(c) = caption.filter(|c| !c.is_empty()) {
|
|
let _ = room.send(RoomMessageEventContent::text_markdown(c)).await;
|
|
}
|
|
DaemonResponse::ok(&serde_json::json!({
|
|
"event_id": resp.event_id.to_string(),
|
|
"room_id": room.room_id().to_string(),
|
|
"filename": filename,
|
|
"mime": content_type.essence_str(),
|
|
}))
|
|
}
|
|
|
|
pub async fn send_file(
|
|
client: &Client,
|
|
room_ref: &str,
|
|
path: &str,
|
|
caption: Option<&str>,
|
|
) -> DaemonResponse {
|
|
let room = match resolve_room(client, room_ref).await {
|
|
Ok(r) => r,
|
|
Err(e) => return e,
|
|
};
|
|
if let Some(reject) = unread_guard(&room) {
|
|
return reject;
|
|
}
|
|
upload_attachment(&room, path, caption).await
|
|
}
|
|
|
|
/// Resolve (find-or-create) the DM room with `user_id` and return its
|
|
/// room id without sending anything. The caller then uses the room-based
|
|
/// tools (`send_file`, `send_message`, …) against that id — so there is
|
|
/// no per-tool `_dm` variant.
|
|
pub async fn open_dm(client: &Client, user_id: &str) -> DaemonResponse {
|
|
let room = match resolve_or_create_dm(client, user_id).await {
|
|
Ok(r) => r,
|
|
Err(e) => return e,
|
|
};
|
|
DaemonResponse::ok(&serde_json::json!({
|
|
"room_id": room.room_id().to_string(),
|
|
"user_id": user_id,
|
|
}))
|
|
}
|
|
|
|
pub async fn send_reaction(
|
|
client: &Client,
|
|
room_ref: &str,
|
|
event_id: &str,
|
|
key: &str,
|
|
) -> DaemonResponse {
|
|
let room = match resolve_room(client, room_ref).await {
|
|
Ok(r) => r,
|
|
Err(e) => return e,
|
|
};
|
|
let eid: OwnedEventId = match event_id.parse() {
|
|
Ok(e) => e,
|
|
Err(e) => return DaemonResponse::error(format!("invalid event_id {event_id}: {e}")),
|
|
};
|
|
let content = ReactionEventContent::new(Annotation::new(eid.clone(), key.to_owned()));
|
|
match room.send(content).await {
|
|
Ok(resp) => DaemonResponse::ok(&serde_json::json!({
|
|
"event_id": resp.response.event_id.to_string(),
|
|
"target": eid.to_string(),
|
|
})),
|
|
Err(e) => DaemonResponse::error(format!("send reaction to {eid}: {e}")),
|
|
}
|
|
}
|
|
|
|
pub async fn send_reply(
|
|
client: &Client,
|
|
room_ref: &str,
|
|
event_id: &str,
|
|
body: &str,
|
|
) -> DaemonResponse {
|
|
let room = match resolve_room(client, room_ref).await {
|
|
Ok(r) => r,
|
|
Err(e) => return e,
|
|
};
|
|
if let Some(reject) = unread_guard(&room) {
|
|
return reject;
|
|
}
|
|
let eid: OwnedEventId = match event_id.parse() {
|
|
Ok(e) => e,
|
|
Err(e) => return DaemonResponse::error(format!("invalid event_id {event_id}: {e}")),
|
|
};
|
|
let content = RoomMessageEventContent::text_markdown(body).into();
|
|
let reply = Reply {
|
|
event_id: eid.clone(),
|
|
enforce_thread: EnforceThread::MaybeThreaded,
|
|
// Add the replied-to sender to m.mentions (standard reply UX — the
|
|
// SDK downgrades this to No when replying to your own event).
|
|
add_mentions: AddMentions::Yes,
|
|
};
|
|
let reply_content = match room.make_reply_event(content, reply).await {
|
|
Ok(c) => c,
|
|
Err(e) => return DaemonResponse::error(format!("make_reply_event: {e}")),
|
|
};
|
|
match room.send(reply_content).await {
|
|
Ok(resp) => DaemonResponse::ok(&serde_json::json!({
|
|
"event_id": resp.response.event_id.to_string(),
|
|
"target": eid.to_string(),
|
|
})),
|
|
Err(e) => DaemonResponse::error(format!("send reply: {e}")),
|
|
}
|
|
}
|
|
|
|
pub async fn mark_read(client: &Client, room_ref: &str, event_id: &str) -> DaemonResponse {
|
|
let room = match resolve_room(client, room_ref).await {
|
|
Ok(r) => r,
|
|
Err(e) => return e,
|
|
};
|
|
let eid: OwnedEventId = match event_id.parse() {
|
|
Ok(e) => e,
|
|
Err(e) => return DaemonResponse::error(format!("invalid event_id {event_id}: {e}")),
|
|
};
|
|
match room
|
|
.send_single_receipt(ReceiptType::Read, ReceiptThread::Unthreaded, eid.clone())
|
|
.await
|
|
{
|
|
Ok(()) => DaemonResponse::ok(&serde_json::json!({
|
|
"marked_read": eid.to_string(),
|
|
})),
|
|
Err(e) => DaemonResponse::error(format!("send_single_receipt: {e}")),
|
|
}
|
|
}
|
|
|
|
pub async fn send_redact(
|
|
client: &Client,
|
|
room_ref: &str,
|
|
event_id: &str,
|
|
reason: Option<&str>,
|
|
) -> DaemonResponse {
|
|
let room = match resolve_room(client, room_ref).await {
|
|
Ok(r) => r,
|
|
Err(e) => return e,
|
|
};
|
|
let eid: OwnedEventId = match event_id.parse() {
|
|
Ok(e) => e,
|
|
Err(e) => return DaemonResponse::error(format!("invalid event_id {event_id}: {e}")),
|
|
};
|
|
// Redaction is irreversible: the homeserver drops the event content and
|
|
// keeps only the shell + reason. A fresh txn id (None) is fine — this is
|
|
// a one-shot operator/agent action, not a retried send. Power level is
|
|
// enforced server-side (own events always; others' need a moderator PL),
|
|
// so a redact that races concurrent delivery is resolved by the server.
|
|
// No client-side pre-check — we let the server be the authority on both
|
|
// permission and ordering.
|
|
match room.redact(&eid, reason, None).await {
|
|
Ok(resp) => DaemonResponse::ok(&serde_json::json!({
|
|
"event_id": resp.event_id.to_string(),
|
|
"target": eid.to_string(),
|
|
})),
|
|
Err(e) => DaemonResponse::error(format!("redact {eid}: {e}")),
|
|
}
|
|
}
|
|
|
|
pub async fn list_rooms(client: &Client) -> DaemonResponse {
|
|
let mut rooms = Vec::new();
|
|
for room in client.joined_rooms() {
|
|
let name = match room.display_name().await {
|
|
Ok(n) => n.to_string(),
|
|
Err(_) => room.room_id().to_string(),
|
|
};
|
|
let canonical_alias = room.canonical_alias().map(|a| a.to_string());
|
|
rooms.push(RoomInfo {
|
|
room_id: room.room_id().to_string(),
|
|
canonical_alias,
|
|
name,
|
|
member_count: room.joined_members_count(),
|
|
});
|
|
}
|
|
DaemonResponse::ok(&rooms)
|
|
}
|
|
|
|
pub async fn list_room_members(client: &Client, room_ref: &str) -> DaemonResponse {
|
|
let room = match resolve_room(client, room_ref).await {
|
|
Ok(r) => r,
|
|
Err(e) => return e,
|
|
};
|
|
let members = match room.members(matrix_sdk::RoomMemberships::JOIN).await {
|
|
Ok(m) => m,
|
|
Err(e) => return DaemonResponse::error(format!("members: {e}")),
|
|
};
|
|
let list: Vec<MemberInfo> = members
|
|
.iter()
|
|
.map(|m| MemberInfo {
|
|
user_id: m.user_id().to_string(),
|
|
display_name: m.display_name().map(ToOwned::to_owned),
|
|
})
|
|
.collect();
|
|
DaemonResponse::ok(&list)
|
|
}
|
|
|
|
pub fn list_invites(client: &Client) -> DaemonResponse {
|
|
let invites: Vec<InviteInfo> = client
|
|
.invited_rooms()
|
|
.into_iter()
|
|
.map(|room| InviteInfo {
|
|
room_id: room.room_id().to_string(),
|
|
canonical_alias: room.canonical_alias().map(|a| a.to_string()),
|
|
name: room.name(),
|
|
})
|
|
.collect();
|
|
DaemonResponse::ok(&invites)
|
|
}
|
|
|
|
pub async fn join_room(client: &Client, room_ref: &str) -> DaemonResponse {
|
|
let parsed: &RoomOrAliasId = match room_ref.try_into() {
|
|
Ok(p) => p,
|
|
Err(e) => {
|
|
return DaemonResponse::error(format!("invalid room reference {room_ref}: {e}"));
|
|
}
|
|
};
|
|
let server_names: Vec<OwnedServerName> = vec![];
|
|
match client.join_room_by_id_or_alias(parsed, &server_names).await {
|
|
Ok(room) => {
|
|
// Clear the invite todo so the accepted invite drops out of
|
|
// `get_loose_ends` immediately (the sweep would also clear it
|
|
// on its next tick, but this makes it instant).
|
|
let _ = crate::wake::send_todo_clear(
|
|
Some(&crate::timeline::invite_key(room.room_id())),
|
|
false,
|
|
)
|
|
.await;
|
|
DaemonResponse::ok(&serde_json::json!({
|
|
"joined": true,
|
|
"room_id": room.room_id().to_string(),
|
|
}))
|
|
}
|
|
Err(e) => DaemonResponse::error(format!("join room {room_ref}: {e}")),
|
|
}
|
|
}
|
|
|
|
/// Accept or reject a pending invite to `room_ref` (id or alias).
|
|
///
|
|
/// `Accept` is exactly `join_room` (joining a room you were invited to
|
|
/// accepts the invite). `Reject` resolves the invited room and leaves
|
|
/// it, declining the invite. Either way the resolved invite drops out
|
|
/// of `get_loose_ends` immediately.
|
|
pub async fn resolve_invite(
|
|
client: &Client,
|
|
room_ref: &str,
|
|
action: InviteAction,
|
|
) -> DaemonResponse {
|
|
match action {
|
|
InviteAction::Accept => join_room(client, room_ref).await,
|
|
InviteAction::Reject => {
|
|
// `resolve_room` -> `get_room` returns the room in any
|
|
// membership state the store knows, including `Invited`, so
|
|
// this works for a pending invite that was never joined.
|
|
let room = match resolve_room(client, room_ref).await {
|
|
Ok(r) => r,
|
|
Err(e) => return e,
|
|
};
|
|
match room.leave().await {
|
|
Ok(()) => {
|
|
// Clear the invite todo immediately on reject.
|
|
let _ = crate::wake::send_todo_clear(
|
|
Some(&crate::timeline::invite_key(room.room_id())),
|
|
false,
|
|
)
|
|
.await;
|
|
DaemonResponse::ok(&serde_json::json!({
|
|
"rejected": true,
|
|
"room_id": room.room_id().to_string(),
|
|
}))
|
|
}
|
|
Err(e) => DaemonResponse::error(format!("reject invite {room_ref}: {e}")),
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Invite `user_id` to `room_ref`. The calling agent must be a member of
|
|
/// the room with a power level high enough to invite; otherwise the
|
|
/// homeserver rejects it and the error is surfaced verbatim.
|
|
pub async fn invite_user(client: &Client, room_ref: &str, user_id: &str) -> DaemonResponse {
|
|
let uid: OwnedUserId = match user_id.parse() {
|
|
Ok(u) => u,
|
|
Err(e) => return DaemonResponse::error(format!("invalid user_id {user_id}: {e}")),
|
|
};
|
|
let room = match resolve_room(client, room_ref).await {
|
|
Ok(r) => r,
|
|
Err(e) => return e,
|
|
};
|
|
match room.invite_user_by_id(&uid).await {
|
|
Ok(()) => DaemonResponse::ok(&serde_json::json!({
|
|
"invited": true,
|
|
"room_id": room.room_id().to_string(),
|
|
"user_id": uid.to_string(),
|
|
})),
|
|
Err(e) => DaemonResponse::error(format!("invite {uid} to {room_ref}: {e}")),
|
|
}
|
|
}
|
|
|
|
/// 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,
|
|
};
|
|
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> = 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)
|
|
}
|
|
|
|
/// Download the media attachment carried by `event_id` in `room` and
|
|
/// write it to a local file, returning the path. Resolves the event,
|
|
/// extracts the `MediaSource` from its `m.file` / `m.image` / `m.audio`
|
|
/// / `m.video` content, fetches the bytes via the media API, and writes
|
|
/// them to `dest_path` (or a temp file named after the attachment when
|
|
/// omitted). The agent then reads the file from the returned path.
|
|
pub async fn download_file(
|
|
client: &Client,
|
|
room_ref: &str,
|
|
event_id: &str,
|
|
dest_path: Option<&str>,
|
|
) -> DaemonResponse {
|
|
use matrix_sdk::media::{MediaFormat, MediaRequestParameters};
|
|
use matrix_sdk::ruma::events::{AnySyncMessageLikeEvent, AnySyncTimelineEvent};
|
|
|
|
let room = match resolve_room(client, room_ref).await {
|
|
Ok(r) => r,
|
|
Err(e) => return e,
|
|
};
|
|
let eid: OwnedEventId = match event_id.parse() {
|
|
Ok(e) => e,
|
|
Err(e) => return DaemonResponse::error(format!("invalid event_id {event_id}: {e}")),
|
|
};
|
|
let ev = match room.event(&eid, None).await {
|
|
Ok(e) => e,
|
|
Err(e) => return DaemonResponse::error(format!("fetch event {eid}: {e}")),
|
|
};
|
|
let parsed = match ev.raw().deserialize() {
|
|
Ok(p) => p,
|
|
Err(e) => return DaemonResponse::error(format!("deserialize event {eid}: {e}")),
|
|
};
|
|
// Pull the media source + filename from the message content.
|
|
let AnySyncTimelineEvent::MessageLike(AnySyncMessageLikeEvent::RoomMessage(msg)) = parsed
|
|
else {
|
|
return DaemonResponse::error(format!("event {eid} is not a room message"));
|
|
};
|
|
let Some(orig) = msg.as_original() else {
|
|
return DaemonResponse::error(format!("event {eid} is redacted"));
|
|
};
|
|
let (source, filename) = match &orig.content.msgtype {
|
|
MessageType::File(f) => (f.source.clone(), f.body.clone()),
|
|
MessageType::Image(i) => (i.source.clone(), i.body.clone()),
|
|
MessageType::Audio(a) => (a.source.clone(), a.body.clone()),
|
|
MessageType::Video(v) => (v.source.clone(), v.body.clone()),
|
|
_ => {
|
|
return DaemonResponse::error(format!(
|
|
"event {eid} carries no file/image/audio/video attachment"
|
|
));
|
|
}
|
|
};
|
|
let req = MediaRequestParameters {
|
|
source,
|
|
format: MediaFormat::File,
|
|
};
|
|
let bytes = match client.media().get_media_content(&req, true).await {
|
|
Ok(b) => b,
|
|
Err(e) => return DaemonResponse::error(format!("download media for {eid}: {e}")),
|
|
};
|
|
let dest = if let Some(p) = dest_path {
|
|
std::path::PathBuf::from(p)
|
|
} else {
|
|
let name = std::path::Path::new(&filename)
|
|
.file_name()
|
|
.and_then(|n| n.to_str())
|
|
.filter(|n| !n.is_empty())
|
|
.unwrap_or("matrix-attachment");
|
|
std::env::temp_dir().join(name)
|
|
};
|
|
if let Err(e) = tokio::fs::write(&dest, &bytes).await {
|
|
return DaemonResponse::error(format!("write {}: {e}", dest.display()));
|
|
}
|
|
DaemonResponse::ok(&serde_json::json!({
|
|
"path": dest.display().to_string(),
|
|
"filename": filename,
|
|
"bytes": bytes.len(),
|
|
"room_id": room.room_id().to_string(),
|
|
}))
|
|
}
|
|
|
|
/// Return the number of joined rooms with at least one unread
|
|
/// notification according to the server-side push notification counts
|
|
/// cached by the matrix-sdk client.
|
|
///
|
|
/// The harness uses [`unread_summary`] (richer, with per-room body
|
|
/// snippets) rather than this endpoint. `unread_count` is kept as a
|
|
/// lightweight public endpoint for callers that only need the count
|
|
/// and want to avoid the `/messages` network round-trips that
|
|
/// [`collect_unread`] issues for count==1 rooms.
|
|
#[must_use]
|
|
pub fn unread_count(client: &Client) -> DaemonResponse {
|
|
let rooms = u32::try_from(
|
|
client
|
|
.joined_rooms()
|
|
.into_iter()
|
|
.filter(|r| r.unread_notification_counts().notification_count > 0)
|
|
.count(),
|
|
)
|
|
.unwrap_or(u32::MAX);
|
|
DaemonResponse::ok(&serde_json::json!({ "rooms": rooms }))
|
|
}
|
|
|
|
/// Collect all rooms with unread notifications and build per-room
|
|
/// [`crate::protocol::RoomUnread`] entries. For rooms with exactly
|
|
/// one unread notification the last message body is fetched via the
|
|
/// `/messages` endpoint (best-effort; failures leave `last_body` as
|
|
/// `None`). Rooms with zero unreads are omitted.
|
|
///
|
|
/// **Latency note**: each count==1 room triggers a live `/messages`
|
|
/// network request to the matrix homeserver to retrieve the message
|
|
/// body. This adds per-room round-trip latency to `get_loose_ends`
|
|
/// and the wake-signal path. Acceptable in practice (rooms with
|
|
/// unread are few; request is best-effort), but worth bearing in
|
|
/// mind if latency becomes a concern.
|
|
#[must_use]
|
|
pub async fn collect_unread(client: &Client) -> Vec<crate::protocol::RoomUnread> {
|
|
collect_unread_with_ids(client)
|
|
.await
|
|
.into_iter()
|
|
.map(|(_, ru)| ru)
|
|
.collect()
|
|
}
|
|
|
|
/// Like [`collect_unread`] but pairs each entry with its `OwnedRoomId`.
|
|
/// The todo producer (loose-ends v2) needs the room id as the
|
|
/// per-room upsert/dedup key, which the claude-facing `RoomUnread`
|
|
/// payload intentionally doesn't carry.
|
|
#[must_use]
|
|
pub async fn collect_unread_with_ids(
|
|
client: &Client,
|
|
) -> Vec<(matrix_sdk::ruma::OwnedRoomId, crate::protocol::RoomUnread)> {
|
|
use crate::protocol::RoomUnread;
|
|
let mut result = Vec::new();
|
|
let own_user_id = client.user_id();
|
|
for room in client.joined_rooms() {
|
|
let count =
|
|
u32::try_from(room.unread_notification_counts().notification_count).unwrap_or(u32::MAX);
|
|
if count == 0 {
|
|
continue;
|
|
}
|
|
let label = room_label(&room);
|
|
let (last_body, last_sender) = if count == 1 {
|
|
fetch_last_message(client, &room).await
|
|
} else {
|
|
(None, None)
|
|
};
|
|
// Skip a single-unread room whose newest event is the agent's own
|
|
// message: after a rebuild the read receipt may not have advanced
|
|
// past it yet, so `notification_count` can still report 1 — without
|
|
// this the agent self-wakes on its own message. Only reachable on the
|
|
// count==1 path, where `last_sender` is populated.
|
|
if let (Some(sender), Some(own_id)) = (&last_sender, own_user_id)
|
|
&& sender.as_str() == own_id.as_str()
|
|
{
|
|
continue;
|
|
}
|
|
result.push((
|
|
room.room_id().to_owned(),
|
|
RoomUnread {
|
|
label,
|
|
count,
|
|
last_body,
|
|
last_sender,
|
|
},
|
|
));
|
|
}
|
|
result
|
|
}
|
|
|
|
/// Fetch the body + sender of the most recent room message. Returns
|
|
/// `(None, None)` on any error or when the timeline contains no text
|
|
/// events.
|
|
async fn fetch_last_message(
|
|
client: &Client,
|
|
room: &matrix_sdk::Room,
|
|
) -> (Option<String>, Option<String>) {
|
|
use matrix_sdk::ruma::api::Direction;
|
|
use matrix_sdk::ruma::api::client::message::get_message_events;
|
|
let mut req =
|
|
get_message_events::v3::Request::new(room.room_id().to_owned(), Direction::Backward);
|
|
req.limit = matrix_sdk::ruma::UInt::from(1u32);
|
|
let Ok(resp) = client.send(req).await else {
|
|
return (None, None);
|
|
};
|
|
for raw in &resp.chunk {
|
|
if let Ok(ev) = raw.deserialize() {
|
|
let body = extract_body(&ev);
|
|
if !body.is_empty() {
|
|
return (
|
|
Some(crate::wake::truncate_chars(&body, 100)),
|
|
Some(ev.sender().to_string()),
|
|
);
|
|
}
|
|
}
|
|
}
|
|
(None, None)
|
|
}
|
|
|
|
/// Return per-room unread summaries. For rooms with exactly one
|
|
/// unread notification, attempts to include the sender + truncated
|
|
/// body; rooms with multiple unreads carry only the count.
|
|
#[must_use]
|
|
pub async fn unread_summary(client: &Client) -> DaemonResponse {
|
|
let rooms = collect_unread(client).await;
|
|
DaemonResponse::ok(&rooms)
|
|
}
|