vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix(#1666): wake on pending matrix invites via post-sync sweep

Browse source
This commit is contained in:
damocles 2026-06-14 22:07:04 +02:00 • committed by mara
commit ca84cf427b
2 changed files with 68 additions and 52 deletions
Download patch file Download diff file Expand all files Collapse all files

21
hive-matrix-mcp/src/main.rs
View file

@ -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, &notified).await;
matrix_sdk::LoopCtrl::Continue
}
})
.await
.context("matrix-sdk sync loop exited")?;

99
hive-matrix-mcp/src/timeline.rs
View file

@ -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<HashSet<OwnedRoomId>>) {
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());
}
}
});
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");
}
}
}