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

//! Matrix event handlers: incoming room messages fire a hyperhive
//! wake signal so the agent's harness drives a new claude turn.
//!
//! Per the operator's call: wake body is a SHORT TEASER, not the full message
//! (msg stays unread server-side; agent fetches via `read_room`). The
//! `wake::format_wake_body` truncates to ~100 chars.
//!
//! Self-events (events sent by this agent) are filtered out so an
//! agent posting a message doesn't wake itself.

use std::collections::HashSet;
use std::path::{Path, PathBuf};
use std::sync::Arc;

use matrix_sdk::{
    Client, Room, RoomState,
    ruma::OwnedRoomId,
    ruma::events::room::message::{MessageType, OriginalSyncRoomMessageEvent},
};
use tokio::sync::Mutex;

use crate::{handlers, wake};

/// Install the room-message handler on `client`. Fires on every
/// `m.room.message` event in a joined room; non-self messages trigger
/// a wake signal to the hyperhive harness via the unix socket at
/// `hyperhive_socket`. The wake body summarises ALL rooms with unread
/// notifications at wake time (not just the triggering event) so the
/// agent receives a full picture in one prompt.
///
/// `account_tag` is `Some(name)` only in multi-account mode (N>1 matrix
/// accounts on this daemon); when set it is prepended to the wake body
/// so the agent knows which account to `read_room` on. In the
/// single-account case it is `None` and the wake body is unchanged.
pub fn install_message_handler(
    client: &Client,
    hyperhive_socket: PathBuf,
    account_tag: Option<String>,
) {
    let socket = Arc::new(hyperhive_socket);
    let account_tag = Arc::new(account_tag);
    let own_user = client.user_id().map(std::borrow::ToOwned::to_owned);
    client.add_event_handler({
        let socket = socket.clone();
        move |event: OriginalSyncRoomMessageEvent, room: Room, client: Client| {
            let socket = socket.clone();
            let account_tag = account_tag.clone();
            let own_user = own_user.clone();
            async move {
                // INFO so the live host journal shows the handler actually
                // firing on an incoming event — the daemon previously logged
                // only handler *install*, which made a non-firing dispatch
                // path (sync not advancing → events deduped, or a room-state
                // mismatch) impossible to distinguish from a delivery failure.
                tracing::info!(
                    room = %room.room_id(),
                    sender = %event.sender,
                    state = ?room.state(),
                    "matrix: message handler fired"
                );
                if room.state() != RoomState::Joined {
                    tracing::info!(
                        room = %room.room_id(),
                        state = ?room.state(),
                        "matrix: skipping message — room not in Joined state"
                    );
                    return;
                }
                // Self-event filter so the agent's own outgoing messages
                // don't wake it. matches forge_notify's self-skip pattern.
                if own_user.as_ref().is_some_and(|u| u == &event.sender) {
                    tracing::debug!(sender = %event.sender, "matrix: skipping self-event");
                    return;
                }
                // Build a wake body that covers all rooms with unread
                // notifications, not just the triggering event. This lets
                // the agent see the full backlog in a single wake prompt.
                // Falls back to the per-event teaser if the collect fails
                // (empty result means daemon is not seeing any unread yet —
                // unlikely but possible during a sync race).
                let unread = handlers::collect_unread(&client).await;
                let body = if unread.is_empty() {
                    // Sync hasn't updated notification counts yet; fall back
                    // to the current event so the agent still wakes.
                    let text = match &event.content.msgtype {
                        MessageType::Text(t) => t.body.clone(),
                        MessageType::Notice(n) => n.body.clone(),
                        MessageType::Emote(e) => format!("* {}", e.body),
                        _ => format!("[{}]", event.content.msgtype()),
                    };
                    let label = handlers::room_label(&room);
                    wake::format_wake_body(event.sender.as_str(), &label, &text)
                } else {
                    wake::format_unread_summary(&unread)
                };
                let body = wake::tag_account(account_tag.as_ref().as_deref(), body);
                if let Err(e) = wake::send_wake(&socket, &body).await {
                    tracing::warn!(error = %e, "failed to deliver matrix wake to hyperhive");
                } else {
                    tracing::info!(room = %room.room_id(), "matrix wake delivered");
                }
            }
        }
    });
    tracing::info!("matrix message handler installed");
}

/// Wake the agent for any pending room invite it has not yet been
/// notified about. Driven from the post-sync callback rather than a
/// `StrippedRoomMemberEvent` handler: that handler dispatched
/// unreliably — an invite that arrived while the daemon was down, or
/// one that raced the handler, produced no wake at all (the message
/// handler, by contrast, fires reliably for joined-room timeline
/// events). Sweeping `invited_rooms()` after every sync catches both
/// the cold-start and the race cases on the proven sync path.
///
/// `notified` dedups so a standing invite wakes the agent once rather
/// than on every sync tick; it is pruned to the current invite set each
/// pass so a withdrawn-then-reissued invite wakes again. The agent
/// decides whether to accept or reject by calling `resolve_invite`.
pub async fn sweep_invites(
    client: &Client,
    socket: &Path,
    notified: &Mutex<HashSet<OwnedRoomId>>,
    account_tag: Option<&str>,
) {
    let current = client.invited_rooms();
    let current_ids: HashSet<OwnedRoomId> =
        current.iter().map(|r| r.room_id().to_owned()).collect();

    let mut fresh = Vec::new();
    {
        let mut seen = notified.lock().await;
        // Drop invites that are no longer pending (joined/rejected/withdrawn)
        // so a future re-invite to the same room wakes the agent again.
        seen.retain(|id| current_ids.contains(id));
        for room in &current {
            if seen.insert(room.room_id().to_owned()) {
                fresh.push(room.clone());
            }
        }
    }
    if fresh.is_empty() {
        return;
    }

    // Refresh loose-ends before waking so the invite is visible in
    // get_loose_ends during the agent's turn.
    handlers::refresh_invite_loose_ends(client).await;
    for room in fresh {
        let room_id = room.room_id().to_owned();
        let label = room.name().unwrap_or_else(|| room_id.to_string());
        tracing::info!(%room_id, "matrix: pending invite swept, waking agent");
        let body = wake::tag_account(
            account_tag,
            format!(
                "[matrix] invited to {label} ({room_id}) — \
                 use list_invites to see pending invites, resolve_invite to accept or reject"
            ),
        );
        if let Err(e) = wake::send_wake(socket, &body).await {
            tracing::warn!(error = %e, "matrix: failed to deliver invite-wake to hyperhive");
        }
    }
}