From 46edc635f2cf0f9049930ec1c9bb3a9cca62970c Mon Sep 17 00:00:00 2001 From: atlas Date: Wed, 3 Jun 2026 12:40:41 +0200 Subject: [PATCH] feat(#1134): add unread matrix messages as a loose end MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When the local matrix daemon has rooms with unread notifications, get_loose_ends now prepends an UnreadMatrix entry to the output so agents see pending matrix activity alongside questions/reminders without message content being exposed. Changes: - hive-sh4re: add LooseEnd::UnreadMatrix { rooms: u32 } variant - hive-matrix-mcp: add DaemonRequest::UnreadCount and handler that counts joined rooms with notification_count > 0 (server-side push notification counts, cached by matrix-sdk) - hive-ag3nt/mcp: inject UnreadMatrix entry on self-queries by connecting to /run/hive-matrix/socket (HIVE_MATRIX_SOCKET override); best-effort — agents without matrix configured are unaffected --- hive-ag3nt/src/mcp.rs | 99 ++++++++++++++++++++++++++++----- hive-matrix-mcp/src/handlers.rs | 14 +++++ hive-matrix-mcp/src/protocol.rs | 6 ++ hive-matrix-mcp/src/socket.rs | 1 + hive-sh4re/src/lib.rs | 8 +++ 5 files changed, 114 insertions(+), 14 deletions(-) diff --git a/hive-ag3nt/src/mcp.rs b/hive-ag3nt/src/mcp.rs index 44811241..a74f405e 100644 --- a/hive-ag3nt/src/mcp.rs +++ b/hive-ag3nt/src/mcp.rs @@ -159,24 +159,17 @@ pub fn format_recv(resp: Result) -> String { /// in-turn `recv` tool result so claude sees the warning either way. pub const REDELIVERY_HINT: &str = "[redelivered after harness restart — may already be handled]\n"; -/// Format helper for `get_loose_ends`: renders a short bulleted list -/// of pending approvals + questions + reminders. Empty list collapses -/// to a clear marker so claude doesn't go hunting for a payload that -/// isn't there. -#[must_use] -pub fn format_loose_ends(resp: Result) -> String { +/// Inner renderer for a `Vec` already extracted from the +/// socket reply. Called by both `format_loose_ends` (which handles the +/// `Result` wrapper) and the augmented `get_loose_ends` +/// handler (which injects the `UnreadMatrix` entry before formatting). +fn render_loose_ends(loose_ends: &[hive_sh4re::LooseEnd]) -> String { use std::fmt::Write as _; - let loose_ends = match resp { - Ok(SocketReply::LooseEnds(t)) => t, - Ok(SocketReply::Err(m)) => return format!("get_loose_ends failed: {m}"), - Ok(other) => return format!("get_loose_ends unexpected response: {other:?}"), - Err(e) => return format!("get_loose_ends transport error: {e:#}"), - }; if loose_ends.is_empty() { return "(no loose ends)".to_owned(); } let mut out = format!("{} loose end(s):\n", loose_ends.len()); - for t in &loose_ends { + for t in loose_ends { match t { hive_sh4re::LooseEnd::Approval { id, @@ -219,11 +212,60 @@ pub fn format_loose_ends(resp: Result) -> String { "- reminder #{id} ({owner}, scheduled {age_seconds}s ago, due_at={due_at}): {message}" ); } + hive_sh4re::LooseEnd::UnreadMatrix { rooms } => { + let _ = writeln!( + out, + "- unread matrix messages in {rooms} room(s) — use list_rooms + read_room to view, mark_read to clear" + ); + } } } out } +/// Format helper for `get_loose_ends`: renders a short bulleted list +/// of pending approvals + questions + reminders. Empty list collapses +/// to a clear marker so claude doesn't go hunting for a payload that +/// isn't there. +#[must_use] +pub fn format_loose_ends(resp: Result) -> String { + let loose_ends = match resp { + Ok(SocketReply::LooseEnds(t)) => t, + Ok(SocketReply::Err(m)) => return format!("get_loose_ends failed: {m}"), + Ok(other) => return format!("get_loose_ends unexpected response: {other:?}"), + Err(e) => return format!("get_loose_ends transport error: {e:#}"), + }; + render_loose_ends(&loose_ends) +} + +/// Query the local matrix daemon for the number of rooms with unread +/// notifications. Returns `None` if the matrix daemon socket is absent +/// or the query fails — callers treat the absence as "no unread". +/// Best-effort: agents without matrix configured are not penalised. +async fn matrix_unread_rooms() -> Option { + use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; + use tokio::net::UnixStream; + let socket = std::env::var_os("HIVE_MATRIX_SOCKET") + .map(std::path::PathBuf::from) + .unwrap_or_else(|| std::path::PathBuf::from("/run/hive-matrix/socket")); + if !socket.exists() { + return None; + } + let mut stream = UnixStream::connect(&socket).await.ok()?; + stream + .write_all(b"{\"method\":\"unread_count\"}\n") + .await + .ok()?; + let mut lines = BufReader::new(stream).lines(); + let line = lines.next_line().await.ok()??; + let val: serde_json::Value = serde_json::from_str(&line).ok()?; + // Response: {"kind":"ok","payload":{"rooms":N}} + val.get("payload") + .and_then(|p| p.get("rooms")) + .and_then(|r| r.as_u64()) + .map(|n| n as u32) +} + /// Parse the user-facing `kind` string for `cancel_loose_end` into the /// wire enum. Accepts a small alias set so claude doesn't have to /// remember the exact spelling (`"q"` / `"r"` shorthand falls out @@ -691,10 +733,39 @@ impl AgentServer { )] async fn get_loose_ends(&self, Parameters(args): Parameters) -> String { run_tool_envelope("get_loose_ends", String::new(), async move { + let is_self_query = args.agent.is_none(); let (resp, retries) = self .dispatch(hive_sh4re::AgentRequest::GetLooseEnds { agent: args.agent }) .await; - let mut out = annotate_retries(format_loose_ends(resp), retries); + // Extract the vec so we can augment before rendering. + let mut loose_ends = match resp { + Ok(SocketReply::LooseEnds(t)) => t, + Ok(SocketReply::Err(m)) => { + return annotate_retries(format!("get_loose_ends failed: {m}"), retries); + } + Ok(other) => { + return annotate_retries( + format!("get_loose_ends unexpected response: {other:?}"), + retries, + ); + } + Err(e) => { + return annotate_retries( + format!("get_loose_ends transport error: {e:#}"), + retries, + ); + } + }; + // Prepend matrix unread entry for self-queries only (can't + // reach another agent's matrix daemon from here). + if is_self_query { + if let Some(rooms) = matrix_unread_rooms().await { + if rooms > 0 { + loose_ends.insert(0, hive_sh4re::LooseEnd::UnreadMatrix { rooms }); + } + } + } + let mut out = annotate_retries(render_loose_ends(&loose_ends), retries); // Append any local bash tasks still in pending/running state so // the agent sees all outstanding work in one call. let active = crate::bash_runner::active_tasks(); diff --git a/hive-matrix-mcp/src/handlers.rs b/hive-matrix-mcp/src/handlers.rs index 3e5a5ea8..3a6d5ba0 100644 --- a/hive-matrix-mcp/src/handlers.rs +++ b/hive-matrix-mcp/src/handlers.rs @@ -343,3 +343,17 @@ pub async fn read_room(client: &Client, room_ref: &str, limit: Option) -> .collect(); DaemonResponse::ok(&events) } + +/// Return the number of joined rooms with at least one unread +/// notification according to the server-side push notification counts +/// cached by the matrix-sdk client. Used by the harness to surface +/// unread matrix activity in `get_loose_ends` without exposing +/// message content. +pub fn unread_count(client: &Client) -> DaemonResponse { + let rooms = client + .joined_rooms() + .into_iter() + .filter(|r| r.unread_notification_counts().notification_count > 0) + .count() as u32; + DaemonResponse::ok(&serde_json::json!({ "rooms": rooms })) +} diff --git a/hive-matrix-mcp/src/protocol.rs b/hive-matrix-mcp/src/protocol.rs index 445c0387..8a4cfe6f 100644 --- a/hive-matrix-mcp/src/protocol.rs +++ b/hive-matrix-mcp/src/protocol.rs @@ -81,6 +81,12 @@ pub enum DaemonRequest { #[serde(rename = "join_room")] JoinRoom { room: String }, + /// Return the count of rooms with unread notifications. Used by + /// the harness `get_loose_ends` to surface unread matrix activity + /// without exposing message content. + #[serde(rename = "unread_count")] + UnreadCount, + /// Liveness probe — fast "are you up?" round-trip that doesn't /// touch matrix-sdk. Not used by the in-tree stdio MCP bridge /// (which surfaces a daemon-down condition as a normal tool-call diff --git a/hive-matrix-mcp/src/socket.rs b/hive-matrix-mcp/src/socket.rs index d5e36b03..5544e765 100644 --- a/hive-matrix-mcp/src/socket.rs +++ b/hive-matrix-mcp/src/socket.rs @@ -83,5 +83,6 @@ async fn dispatch(req: DaemonRequest, client: &Client) -> DaemonResponse { DaemonRequest::JoinRoom { room } => handlers::join_room(client, &room).await, DaemonRequest::ListRoomMembers { room } => handlers::list_room_members(client, &room).await, DaemonRequest::ReadRoom { room, limit } => handlers::read_room(client, &room, limit).await, + DaemonRequest::UnreadCount => handlers::unread_count(client), } } diff --git a/hive-sh4re/src/lib.rs b/hive-sh4re/src/lib.rs index 9c7df6dc..4020d4c6 100644 --- a/hive-sh4re/src/lib.rs +++ b/hive-sh4re/src/lib.rs @@ -290,6 +290,14 @@ pub enum LooseEnd { due_at: i64, age_seconds: u64, }, + /// Unread matrix notifications in one or more rooms. Not cancellable — + /// use `mark_read` via the matrix MCP to clear. Injected by the + /// in-container harness (not hive-c0re) because the matrix daemon + /// runs inside the agent container. + UnreadMatrix { + /// Number of rooms with at least one unread notification. + rooms: u32, + }, } /// Kind discriminator for `CancelLooseEnd`. Per-kind store +