diff --git a/hive-matrix-mcp/src/handlers.rs b/hive-matrix-mcp/src/handlers.rs index 89742d1f..08ffb186 100644 --- a/hive-matrix-mcp/src/handlers.rs +++ b/hive-matrix-mcp/src/handlers.rs @@ -288,6 +288,46 @@ pub async fn list_invites(client: &Client) -> DaemonResponse { DaemonResponse::ok(&invites) } +/// Rewrite `mcp-loose-ends/matrix.json` with a summary of all pending +/// room invites. The harness scans this directory generically in +/// `get_loose_ends` — no matrix-specific code needed there. +/// +/// Called after an invite arrives (from the sync handler) and after a +/// room is joined (to remove the accepted invite from loose-ends). +/// Atomic write (tmp + rename) so the harness never reads a partial file. +pub async fn refresh_invite_loose_ends(client: &Client) { + let invites = client.invited_rooms(); + let dir = crate::paths::mcp_loose_ends_dir(); + if let Err(e) = tokio::fs::create_dir_all(&dir).await { + tracing::warn!(error = ?e, "matrix: create mcp-loose-ends dir failed"); + return; + } + let items: Vec = invites + .iter() + .map(|room| { + let label = room + .canonical_alias() + .map_or_else(|| room.room_id().to_string(), |a| a.to_string()); + format!( + "[matrix] pending invite: {label} — use list_invites to see, join_room to accept" + ) + }) + .collect(); + let dest = dir.join("matrix.json"); + let tmp = dest.with_extension("json.tmp"); + let json = serde_json::to_string(&items).unwrap_or_else(|_| "[]".to_owned()); + match tokio::fs::write(&tmp, &json).await { + Ok(()) => { + if let Err(e) = tokio::fs::rename(&tmp, &dest).await { + tracing::warn!(error = ?e, "matrix: rename mcp-loose-ends/matrix.json failed"); + } + } + Err(e) => { + tracing::warn!(error = ?e, "matrix: write mcp-loose-ends/matrix.json.tmp failed"); + } + } +} + pub async fn join_room(client: &Client, room_ref: &str) -> DaemonResponse { let parsed: &RoomOrAliasId = match room_ref.try_into() { Ok(p) => p, @@ -297,10 +337,15 @@ pub async fn join_room(client: &Client, room_ref: &str) -> DaemonResponse { }; let server_names: Vec = vec![]; match client.join_room_by_id_or_alias(parsed, &server_names).await { - Ok(room) => DaemonResponse::ok(&serde_json::json!({ - "joined": true, - "room_id": room.room_id().to_string(), - })), + Ok(room) => { + // Refresh loose-ends so the accepted invite is removed from + // `get_loose_ends` output immediately after the agent joins. + refresh_invite_loose_ends(client).await; + DaemonResponse::ok(&serde_json::json!({ + "joined": true, + "room_id": room.room_id().to_string(), + })) + } Err(e) => DaemonResponse::error(format!("join room {room_ref}: {e}")), } } diff --git a/hive-matrix-mcp/src/main.rs b/hive-matrix-mcp/src/main.rs index 34a6b0bb..d8bcda2e 100644 --- a/hive-matrix-mcp/src/main.rs +++ b/hive-matrix-mcp/src/main.rs @@ -61,7 +61,8 @@ async fn main() -> Result<()> { .await .context("build matrix client")?; - timeline::install_message_handler(&matrix_client, hyperhive_socket); + 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 diff --git a/hive-matrix-mcp/src/paths.rs b/hive-matrix-mcp/src/paths.rs index 39913087..a40c4704 100644 --- a/hive-matrix-mcp/src/paths.rs +++ b/hive-matrix-mcp/src/paths.rs @@ -66,3 +66,25 @@ pub fn hyperhive_socket() -> PathBuf { std::env::var_os("HIVE_CONTROL_SOCKET") .map_or_else(|| PathBuf::from("/run/hive/mcp.sock"), PathBuf::from) } + +/// Directory where MCP daemons write loose-end summary files for the harness. +/// Each daemon writes `.json` here; the harness scans the dir in +/// `get_loose_ends` to surface active work from all MCPs generically. +/// +/// NOTE: the base-dir resolution logic here is intentionally mirrored in +/// `hive-ag3nt/src/mcp_loose_ends.rs::loose_ends_dir()`. They can't share +/// code across crates — keep them in sync if the fallback logic changes. +#[must_use] +pub fn mcp_loose_ends_dir() -> PathBuf { + let base = if let Some(p) = std::env::var_os("HYPERHIVE_HARNESS_DIR") { + PathBuf::from(p) + } else { + let state = std::env::var("HYPERHIVE_STATE_DIR").unwrap_or_default(); + let state_path = PathBuf::from(&state); + state_path + .parent() + .map(|p| p.join("harness")) + .unwrap_or_else(|| PathBuf::from(state)) + }; + base.join("mcp-loose-ends") +} diff --git a/hive-matrix-mcp/src/timeline.rs b/hive-matrix-mcp/src/timeline.rs index 90ae454f..aa8ed353 100644 --- a/hive-matrix-mcp/src/timeline.rs +++ b/hive-matrix-mcp/src/timeline.rs @@ -13,6 +13,7 @@ use std::sync::Arc; use matrix_sdk::{ Client, Room, RoomState, + ruma::events::room::member::{MembershipState, StrippedRoomMemberEvent}, ruma::events::room::message::{MessageType, OriginalSyncRoomMessageEvent}, }; @@ -74,3 +75,54 @@ 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 by calling `join_room`. +/// +/// 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 `join_room`. +/// +/// 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, join_room to accept" + ); + if let Err(e) = wake::send_wake(&socket, &body).await { + tracing::warn!( + error = %e, + "matrix: failed to deliver invite-wake to hyperhive" + ); + } + } + } + }); + tracing::info!("matrix invite handler installed"); +}