From ca84cf427b7210fe39c08b68234871e607ea9429 Mon Sep 17 00:00:00 2001 From: damocles Date: Sun, 14 Jun 2026 22:07:04 +0200 Subject: [PATCH] fix(#1666): wake on pending matrix invites via post-sync sweep --- hive-matrix-mcp/src/main.rs | 21 ++++++- hive-matrix-mcp/src/timeline.rs | 99 +++++++++++++++++---------------- 2 files changed, 68 insertions(+), 52 deletions(-) diff --git a/hive-matrix-mcp/src/main.rs b/hive-matrix-mcp/src/main.rs index ff0f749e..32dd1ea0 100644 --- a/hive-matrix-mcp/src/main.rs +++ b/hive-matrix-mcp/src/main.rs @@ -65,7 +65,6 @@ async fn main() -> Result<()> { .context("build matrix client")?; timeline::install_message_handler(&matrix_client, hyperhive_socket.clone()); - timeline::install_invite_handler(&matrix_client, hyperhive_socket); // Spawn the unix socket server before sync starts so the MCP // bridge can connect as soon as the first claude turn fires. The @@ -79,10 +78,26 @@ async fn main() -> Result<()> { } }); - // Sync forever; matrix-sdk handles reconnection internally. + // Sync forever; matrix-sdk handles reconnection internally. After + // every sync, sweep pending invites and wake the agent for any new + // one: the StrippedRoomMemberEvent handler dispatched unreliably + // (cold-start invites + handler races produced no wake), so + // invite-waking lives on this post-sync sweep with a dedup set. + let invite_socket = std::sync::Arc::new(hyperhive_socket); + let invite_notified = + std::sync::Arc::new(tokio::sync::Mutex::new(std::collections::HashSet::new())); + let sweep_client = matrix_client.clone(); let sync_settings = SyncSettings::default(); matrix_client - .sync(sync_settings) + .sync_with_callback(sync_settings, move |_response| { + let client = sweep_client.clone(); + let socket = invite_socket.clone(); + let notified = invite_notified.clone(); + async move { + timeline::sweep_invites(&client, &socket, ¬ified).await; + matrix_sdk::LoopCtrl::Continue + } + }) .await .context("matrix-sdk sync loop exited")?; diff --git a/hive-matrix-mcp/src/timeline.rs b/hive-matrix-mcp/src/timeline.rs index 4a81669f..763582c0 100644 --- a/hive-matrix-mcp/src/timeline.rs +++ b/hive-matrix-mcp/src/timeline.rs @@ -8,14 +8,16 @@ //! Self-events (events sent by this agent) are filtered out so an //! agent posting a message doesn't wake itself. -use std::path::PathBuf; +use std::collections::HashSet; +use std::path::{Path, PathBuf}; use std::sync::Arc; use matrix_sdk::{ Client, Room, RoomState, - ruma::events::room::member::{MembershipState, StrippedRoomMemberEvent}, + ruma::OwnedRoomId, ruma::events::room::message::{MessageType, OriginalSyncRoomMessageEvent}, }; +use tokio::sync::Mutex; use crate::{handlers, wake}; @@ -93,54 +95,53 @@ pub fn install_message_handler(client: &Client, hyperhive_socket: PathBuf) { tracing::info!("matrix message handler installed"); } -/// Install a handler that wakes the agent when a room invite arrives. -/// The agent decides whether to accept or reject by calling -/// `resolve_invite`. +/// 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. /// -/// When the daemon receives an `m.room.member` state event with -/// `membership: invite` for the agent's own user ID during sync, it -/// fires a hyperhive wake so the agent can inspect the invite via -/// `list_invites` and act on it with `resolve_invite`. -/// -/// This closes the gap where invites accumulated in `invited_rooms()` -/// forever without the agent being notified — the daemon's message -/// handler only fires for joined rooms. -pub fn install_invite_handler(client: &Client, hyperhive_socket: PathBuf) { - let socket = Arc::new(hyperhive_socket); - let own_user = client.user_id().map(std::borrow::ToOwned::to_owned); - client.add_event_handler({ - move |event: StrippedRoomMemberEvent, room: Room, client: Client| { - let socket = socket.clone(); - let own_user = own_user.clone(); - async move { - // Only handle invite events for our own user. - if event.content.membership != MembershipState::Invite { - return; - } - let is_ours = own_user - .as_ref() - .is_some_and(|u| u.as_str() == event.state_key.as_str()); - if !is_ours { - return; - } - let room_id = room.room_id().to_owned(); - let label = room.name().unwrap_or_else(|| room_id.to_string()); - tracing::info!(%room_id, "matrix: received room invite, waking agent"); - // 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; - let body = 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" - ); - } +/// `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>) { + let current = client.invited_rooms(); + let current_ids: HashSet = + 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 ¤t { + if seen.insert(room.room_id().to_owned()) { + fresh.push(room.clone()); } } - }); - tracing::info!("matrix invite handler installed"); + } + 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 = 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"); + } + } }