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

//! Matrix → hyperhive wake bridge. Both room messages and room invites
//! are surfaced to the agent by SWEEPING state on every post-sync
//! callback, not by one-shot `m.room.message` / `StrippedRoomMemberEvent`
//! handlers: a one-shot wake whose `send_wake` raced a hive-c0re / socket-
//! down window (a container rebuild) was dropped with no retry, leaving
//! the agent deaf to matrix activity until manually prompted. Sweeping the
//! reliable sync path re-checks each tick and self-heals a dropped wake on
//! the next one.
//!
//! Wake bodies stay short: `sweep_unread` sends the all-rooms unread
//! summary (`wake::format_unread_summary`); the message stays unread
//! server-side so the agent fetches detail via `read_room`. Self-sent
//! messages never raise an unread notification, so they never wake.

use std::collections::HashSet;
use std::path::Path;

use matrix_sdk::{Client, ruma::OwnedRoomId};
use tokio::sync::Mutex;

use crate::{handlers, wake};

/// Wake the agent for any joined room carrying unread notifications it has
/// not yet been successfully woken about. Driven from the post-sync
/// callback — the same reliable path `sweep_invites` uses — rather than a
/// one-shot `m.room.message` event handler: a message-wake whose
/// `send_wake` raced a hive-c0re / socket-down window (a container rebuild)
/// was dropped with no retry, so the agent went deaf until manually
/// prompted. The sync callback fires on every sync response (so promptly on
/// new activity), so this re-checks unread each tick and retries a dropped
/// wake on the next one — self-healing once the socket is back.
///
/// `notified` dedups so a standing unread wakes the agent once, not on
/// every sync tick; it is pruned to the current unread set each pass so a
/// room that has been read and then receives a new message wakes again. On
/// a wake-send failure the freshly-seen rooms are rolled back out of
/// `notified` so the next pass retries them. `account_tag` is `Some(name)`
/// only in multi-account mode, prepended so the agent knows which account
/// to `read_room` on.
///
/// Self-sent messages don't raise an unread notification server-side, so
/// no explicit self-filter is needed here.
pub async fn sweep_unread(
    client: &Client,
    socket: &Path,
    notified: &Mutex<HashSet<OwnedRoomId>>,
    account_tag: Option<&str>,
) {
    // Current set of joined rooms carrying unread notifications.
    let unread_ids: HashSet<OwnedRoomId> = client
        .joined_rooms()
        .into_iter()
        .filter(|r| r.unread_notification_counts().notification_count > 0)
        .map(|r| r.room_id().to_owned())
        .collect();

    // Which unread rooms are newly-seen (not already successfully woken)?
    let mut fresh = Vec::new();
    {
        let mut seen = notified.lock().await;
        // Drop rooms no longer unread (read / left) so a future new message
        // in them wakes the agent again.
        seen.retain(|id| unread_ids.contains(id));
        for id in &unread_ids {
            if seen.insert(id.clone()) {
                fresh.push(id.clone());
            }
        }
    }
    if fresh.is_empty() {
        return;
    }

    // Roll the freshly-seen rooms back out of `notified` so the next sweep
    // retries them. Used on both the sync-race (empty collect) and the
    // wake-send-failure paths.
    let rollback = || async {
        let mut seen = notified.lock().await;
        for id in &fresh {
            seen.remove(id);
        }
    };

    // Same all-rooms unread summary the wake body has always used. Empty
    // only under a sync race (counts changed between the scan and collect).
    let unread = handlers::collect_unread(client).await;
    if unread.is_empty() {
        rollback().await;
        return;
    }
    let body = wake::tag_account(account_tag, wake::format_unread_summary(&unread));
    if let Err(e) = wake::send_wake(socket, &body).await {
        rollback().await;
        tracing::warn!(error = %e, "matrix: unread-sweep wake failed; will retry next sync");
    } else {
        tracing::info!(rooms = fresh.len(), "matrix: unread-sweep wake delivered");
    }
}

/// 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");
        }
    }
}