matrix: hive-matrix-mcp crate (daemon+stdio bridge) + harness wiring (#548 phase 3)

hive-matrix-mcp/src/handlers.rs

@@ -0,0 +1,305 @@
+//! Per-tool dispatch — each daemon request from the MCP bridge resolves
+//! to one of these handlers. Returns a `DaemonResponse` shaped for the
+//! wire protocol (the MCP bridge unwraps `Ok { payload }` and returns
+//! the payload to claude; `Error { message }` becomes the tool-call
+//! error message claude sees).
+//!
+//! Tool surface mirrors damocles-daemon's v0 set per mara on #548:
+//! `send_message`, `send_dm`, `send_reaction`, `send_reply`, `mark_read`,
+//! `list_rooms`, `list_room_members`, `read_room`. Plus a `ping` for the
+//! MCP bridge's liveness probe.
+
+use matrix_sdk::{
+    Client,
+    room::reply::{EnforceThread, Reply},
+    ruma::{
+        OwnedEventId, OwnedRoomId, OwnedUserId, RoomOrAliasId,
+        api::client::receipt::create_receipt::v3::ReceiptType,
+        events::{
+            receipt::ReceiptThread,
+            reaction::ReactionEventContent,
+            relation::Annotation,
+            room::message::{MessageType, RoomMessageEventContent},
+        },
+    },
+};
+use serde::Serialize;
+
+use crate::protocol::DaemonResponse;
+
+/// 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,
+}
+
+/// 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| {
+                match &orig.content.msgtype {
+                    MessageType::Text(t) => t.body.clone(),
+                    MessageType::Notice(n) => n.body.clone(),
+                    MessageType::Emote(e) => format!("* {}", e.body),
+                    _ => String::new(),
+                }
+            })
+        }
+        _ => String::new(),
+    }
+}
+
+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,
+    };
+    let content = RoomMessageEventContent::text_markdown(body);
+    match room.send(content).await {
+        Ok(resp) => DaemonResponse::ok(&serde_json::json!({
+            "event_id": resp.event_id.to_string(),
+            "room_id": room.room_id().to_string(),
+        })),
+        Err(e) => DaemonResponse::error(format!("send to {}: {e}", room.room_id())),
+    }
+}
+
+pub async fn send_dm(client: &Client, user_id: &str, body: &str) -> DaemonResponse {
+    let uid: OwnedUserId = match user_id.parse() {
+        Ok(u) => u,
+        Err(e) => return DaemonResponse::error(format!("invalid user_id {user_id}: {e}")),
+    };
+    // Find existing DM or create one.
+    let room = client
+        .joined_rooms()
+        .into_iter()
+        .find(|r| {
+            // is_direct() is async; check direct_targets() instead which
+            // reads from cached state.
+            r.direct_targets().iter().any(|t| t.as_str() == uid.as_str())
+        });
+    let room = match room {
+        Some(r) => r,
+        None => match client.create_dm(&uid).await {
+            Ok(r) => r,
+            Err(e) => return DaemonResponse::error(format!("create_dm {uid}: {e}")),
+        },
+    };
+    let content = RoomMessageEventContent::text_markdown(body);
+    match room.send(content).await {
+        Ok(resp) => DaemonResponse::ok(&serde_json::json!({
+            "event_id": resp.event_id.to_string(),
+            "room_id": room.room_id().to_string(),
+            "user_id": uid.to_string(),
+        })),
+        Err(e) => DaemonResponse::error(format!("send DM to {uid}: {e}")),
+    }
+}
+
+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.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,
+    };
+    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,
+    };
+    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.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 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 async fn read_room(client: &Client, room_ref: &str, limit: Option<usize>) -> DaemonResponse {
+    use matrix_sdk::ruma::api::client::message::get_message_events;
+    use matrix_sdk::ruma::api::Direction;
+
+    let room = match resolve_room(client, room_ref).await {
+        Ok(r) => r,
+        Err(e) => return e,
+    };
+    let limit = limit.unwrap_or(50).min(200);
+    let mut req = get_message_events::v3::Request::new(room.room_id().to_owned(), Direction::Backward);
+    req.limit = matrix_sdk::ruma::UInt::try_from(limit as u64).unwrap_or(matrix_sdk::ruma::UInt::from(50u32));
+    let resp = match client.send(req).await {
+        Ok(r) => r,
+        Err(e) => return DaemonResponse::error(format!("get_message_events: {e}")),
+    };
+    let events: Vec<TimelineEvent> = resp
+        .chunk
+        .iter()
+        .filter_map(|raw| {
+            let parsed = 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 = extract_body(&parsed);
+            Some(TimelineEvent {
+                event_id,
+                sender,
+                origin_server_ts,
+                event_type,
+                body,
+            })
+        })
+        .collect();
+    DaemonResponse::ok(&events)
+}