feat(#1134): add unread matrix messages as a loose end
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
This commit is contained in:
parent
1386439591
commit
46edc635f2
5 changed files with 114 additions and 14 deletions
|
|
@ -159,24 +159,17 @@ pub fn format_recv(resp: Result<SocketReply, anyhow::Error>) -> String {
|
||||||
/// in-turn `recv` tool result so claude sees the warning either way.
|
/// 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";
|
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
|
/// Inner renderer for a `Vec<LooseEnd>` already extracted from the
|
||||||
/// of pending approvals + questions + reminders. Empty list collapses
|
/// socket reply. Called by both `format_loose_ends` (which handles the
|
||||||
/// to a clear marker so claude doesn't go hunting for a payload that
|
/// `Result<SocketReply>` wrapper) and the augmented `get_loose_ends`
|
||||||
/// isn't there.
|
/// handler (which injects the `UnreadMatrix` entry before formatting).
|
||||||
#[must_use]
|
fn render_loose_ends(loose_ends: &[hive_sh4re::LooseEnd]) -> String {
|
||||||
pub fn format_loose_ends(resp: Result<SocketReply, anyhow::Error>) -> String {
|
|
||||||
use std::fmt::Write as _;
|
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() {
|
if loose_ends.is_empty() {
|
||||||
return "(no loose ends)".to_owned();
|
return "(no loose ends)".to_owned();
|
||||||
}
|
}
|
||||||
let mut out = format!("{} loose end(s):\n", loose_ends.len());
|
let mut out = format!("{} loose end(s):\n", loose_ends.len());
|
||||||
for t in &loose_ends {
|
for t in loose_ends {
|
||||||
match t {
|
match t {
|
||||||
hive_sh4re::LooseEnd::Approval {
|
hive_sh4re::LooseEnd::Approval {
|
||||||
id,
|
id,
|
||||||
|
|
@ -219,11 +212,60 @@ pub fn format_loose_ends(resp: Result<SocketReply, anyhow::Error>) -> String {
|
||||||
"- reminder #{id} ({owner}, scheduled {age_seconds}s ago, due_at={due_at}): {message}"
|
"- 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
|
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<SocketReply, anyhow::Error>) -> 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<u32> {
|
||||||
|
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
|
/// 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
|
/// wire enum. Accepts a small alias set so claude doesn't have to
|
||||||
/// remember the exact spelling (`"q"` / `"r"` shorthand falls out
|
/// remember the exact spelling (`"q"` / `"r"` shorthand falls out
|
||||||
|
|
@ -691,10 +733,39 @@ impl AgentServer {
|
||||||
)]
|
)]
|
||||||
async fn get_loose_ends(&self, Parameters(args): Parameters<AgentGetLooseEndsArgs>) -> String {
|
async fn get_loose_ends(&self, Parameters(args): Parameters<AgentGetLooseEndsArgs>) -> String {
|
||||||
run_tool_envelope("get_loose_ends", String::new(), async move {
|
run_tool_envelope("get_loose_ends", String::new(), async move {
|
||||||
|
let is_self_query = args.agent.is_none();
|
||||||
let (resp, retries) = self
|
let (resp, retries) = self
|
||||||
.dispatch(hive_sh4re::AgentRequest::GetLooseEnds { agent: args.agent })
|
.dispatch(hive_sh4re::AgentRequest::GetLooseEnds { agent: args.agent })
|
||||||
.await;
|
.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
|
// Append any local bash tasks still in pending/running state so
|
||||||
// the agent sees all outstanding work in one call.
|
// the agent sees all outstanding work in one call.
|
||||||
let active = crate::bash_runner::active_tasks();
|
let active = crate::bash_runner::active_tasks();
|
||||||
|
|
|
||||||
|
|
@ -343,3 +343,17 @@ pub async fn read_room(client: &Client, room_ref: &str, limit: Option<usize>) ->
|
||||||
.collect();
|
.collect();
|
||||||
DaemonResponse::ok(&events)
|
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 }))
|
||||||
|
}
|
||||||
|
|
|
||||||
|
|
@ -81,6 +81,12 @@ pub enum DaemonRequest {
|
||||||
#[serde(rename = "join_room")]
|
#[serde(rename = "join_room")]
|
||||||
JoinRoom { room: String },
|
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
|
/// Liveness probe — fast "are you up?" round-trip that doesn't
|
||||||
/// touch matrix-sdk. Not used by the in-tree stdio MCP bridge
|
/// touch matrix-sdk. Not used by the in-tree stdio MCP bridge
|
||||||
/// (which surfaces a daemon-down condition as a normal tool-call
|
/// (which surfaces a daemon-down condition as a normal tool-call
|
||||||
|
|
|
||||||
|
|
@ -83,5 +83,6 @@ async fn dispatch(req: DaemonRequest, client: &Client) -> DaemonResponse {
|
||||||
DaemonRequest::JoinRoom { room } => handlers::join_room(client, &room).await,
|
DaemonRequest::JoinRoom { room } => handlers::join_room(client, &room).await,
|
||||||
DaemonRequest::ListRoomMembers { room } => handlers::list_room_members(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::ReadRoom { room, limit } => handlers::read_room(client, &room, limit).await,
|
||||||
|
DaemonRequest::UnreadCount => handlers::unread_count(client),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -290,6 +290,14 @@ pub enum LooseEnd {
|
||||||
due_at: i64,
|
due_at: i64,
|
||||||
age_seconds: u64,
|
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 +
|
/// Kind discriminator for `CancelLooseEnd`. Per-kind store +
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue