hyperhive/hive-matrix-mcp/src/handlers.rs

//! 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, ReceiptType as LocalReceiptType},
            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
/// content (the agent must `read_room` then `mark_read` the latest event
/// first), or `None` when the send may proceed.
///
/// Read-state is `room_unread_state` — the same predicate 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.
async fn unread_guard(client: &Client, room: &matrix_sdk::Room) -> Option<DaemonResponse> {
    room_unread_state(client, room).await?;
    let label = room_label(room);
    Some(DaemonResponse::error(format!(
        "refusing to send: 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."
    )))
}

/// Fetch the single most recent timeline event in `room`, of any type
/// (redactions/state/reactions included — identity is what
/// `room_unread_state` needs, not content). `None` for a genuinely
/// empty room or on any request failure.
async fn latest_event(
    client: &Client,
    room: &matrix_sdk::Room,
) -> Option<(OwnedEventId, OwnedUserId, 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 resp = client.send(req).await.ok()?;
    let raw = resp.chunk.first()?;
    let ev = raw.deserialize().ok()?;
    let body = extract_body(&ev);
    Some((ev.event_id().to_owned(), ev.sender().to_owned(), body))
}

/// Whether `room` carries content the agent hasn't caught up on.
/// `Some((event_id, sender, body))` for the latest event when it's
/// genuinely unread; `None` when the room is caught up, empty, or its
/// latest event is self-authored.
///
/// Diffs the room's own local state against the agent's own `m.read`
/// receipt instead of `unread_notification_counts()`. That count is
/// server-computed from Matrix **push rules** — a DM or a plain message
/// that doesn't match any push rule (no keyword, no @mention) can leave a
/// genuinely-unread event reporting `notification_count: 0` forever, so
/// the agent never wakes for it (confirmed live on a prior investigation:
/// the daemon had the event in its local store the whole time and still
/// reported zero — the count was never the right signal to read).
/// Comparing the latest local event id against the own-account read
/// receipt needs no push-rule semantics at all.
///
/// Self-authored latest events are never "unread" — after a daemon
/// rebuild the read receipt can lag behind the agent's own just-sent
/// message, which must not self-wake it.
async fn room_unread_state(
    client: &Client,
    room: &matrix_sdk::Room,
) -> Option<(OwnedEventId, OwnedUserId, String)> {
    let own_user_id = client.user_id()?;
    let (event_id, sender, body) = latest_event(client, room).await?;
    if sender.as_str() == own_user_id.as_str() {
        return None;
    }
    // `load_user_receipt` takes `ruma::events::receipt::ReceiptType`
    // (imported here as `LocalReceiptType`), a distinct type from the
    // `ReceiptType` this module already imports for `send_single_receipt`
    // (`ruma::api::client::receipt::create_receipt::v3::ReceiptType`) — two
    // same-named enums from different ruma crates, not interchangeable.
    let read = room
        .load_user_receipt(
            LocalReceiptType::Read,
            ReceiptThread::Unthreaded,
            own_user_id,
        )
        .await
        .ok()
        .flatten();
    if read.is_some_and(|(id, _)| id == event_id) {
        return None;
    }
    Some((event_id, sender, body))
}

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(client, &room).await {
        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(client, &room).await {
        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(client, &room).await {
        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(client, &room).await {
        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(),
    }))
}

/// 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 unread room triggers a live `/messages`
/// network request to the matrix homeserver to determine its latest
/// event (via `room_unread_state`). 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();
    for room in client.joined_rooms() {
        let Some((_, sender, body)) = room_unread_state(client, &room).await else {
            continue;
        };
        let label = room_label(&room);
        let last_body = (!body.is_empty()).then(|| crate::wake::truncate_chars(&body, 100));
        result.push((
            room.room_id().to_owned(),
            RoomUnread {
                label,
                count: 1,
                last_body,
                last_sender: Some(sender.to_string()),
            },
        ));
    }
    result
}

/// Return per-room unread summaries. Each entry carries the sender +
/// truncated body of the room's latest (unread) event when it's a
/// text-like message; rooms whose latest unread event has no body
/// (a reaction, a state change, …) carry the room + sender only.
#[must_use]
pub async fn unread_summary(client: &Client) -> DaemonResponse {
    let rooms = collect_unread(client).await;
    DaemonResponse::ok(&rooms)
}