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

feat(#1137): rich unread summary in loose ends and wake signal

Browse source 
- hive-sh4re: UnreadMatrix gains summary: String field (per-room breakdown)
- hive-matrix-mcp/protocol: add RoomUnread struct + UnreadSummary request
- hive-matrix-mcp/handlers: collect_unread() fetches per-room data;
  single-unread rooms include truncated last-message body + sender;
  multi-unread rooms carry count only
- hive-matrix-mcp/wake: format_unread_summary() builds wake body from
  RoomUnread slice; terse one-liner for single-room/single-message,
  bulleted list for multi-room; always appends read-hint
- hive-matrix-mcp/timeline: wake body now covers all rooms with unread
  at fire time, not just the triggering event; falls back to per-event
  teaser if notification counts haven't updated yet
- hive-ag3nt/mcp: matrix_unread_summary() replaces matrix_unread_rooms();
  UnreadMatrix loose end carries per-room summary lines; render shows
  room breakdown with sender: body for single-unread rooms
This commit is contained in:
atlas 2026-06-03 12:52:24 +02:00
commit 68e30b857c
19 changed files with 421 additions and 108 deletions
Download patch file Download diff file Expand all files Collapse all files

22
hive-ag3nt/src/bin/hive.rs
View file

@ -365,8 +365,15 @@ impl Surface for AgentSurface {
}
async fn wake_external(socket: &Path, from: String, body: String) -> Result<()> {
let resp: AgentResponse =
client::request(socket, &AgentRequest::Wake { from, body, transient: false }).await?;
let resp: AgentResponse = client::request(
socket,
&AgentRequest::Wake {
from,
body,
transient: false,
},
)
.await?;
match resp {
AgentResponse::Ok => Ok(()),
AgentResponse::Err { message } => anyhow::bail!("wake: {message}"),
@ -506,8 +513,15 @@ impl Surface for ManagerSurface {
}
async fn wake_external(socket: &Path, from: String, body: String) -> Result<()> {
let resp: ManagerResponse =
client::request(socket, &ManagerRequest::Wake { from, body, transient: false }).await?;
let resp: ManagerResponse = client::request(
socket,
&ManagerRequest::Wake {
from,
body,
transient: false,
},
)
.await?;
match resp {
ManagerResponse::Ok => Ok(()),
ManagerResponse::Err { message } => anyhow::bail!("wake: {message}"),

101
hive-ag3nt/src/mcp.rs
View file

@ -212,11 +212,23 @@ fn render_loose_ends(loose_ends: &[hive_sh4re::LooseEnd]) -> 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"
);
hive_sh4re::LooseEnd::UnreadMatrix { rooms, summary } => {
let _ = write!(out, "- unread matrix messages in {rooms} room(s)");
if summary.is_empty() {
let _ = writeln!(
out,
" — use list_rooms + read_room to view, mark_read to clear"
);
} else {
let _ = writeln!(out, ":");
for line in summary.lines() {
let _ = writeln!(out, " {line}");
}
let _ = writeln!(
out,
" use list_rooms + read_room to view, mark_read to clear"
);
}
}
}
}
@ -238,11 +250,21 @@ pub fn format_loose_ends(resp: Result<SocketReply, anyhow::Error>) -> String {
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> {
/// Per-room unread entry returned by `matrix_unread_summary`. Mirrors
/// `hive-matrix-mcp`'s `RoomUnread` but defined locally to avoid a
/// cross-crate dep on the matrix-sdk crate tree.
#[derive(Debug, serde::Deserialize)]
struct MatrixRoomUnread {
label: String,
count: u32,
last_body: Option<String>,
last_sender: Option<String>,
}
/// Query the local matrix daemon for per-room unread summaries. Returns
/// `None` if the daemon socket is absent or the query fails. Best-effort:
/// agents without matrix configured are not penalised.
async fn matrix_unread_summary() -> Option<Vec<MatrixRoomUnread>> {
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::net::UnixStream;
let socket = std::env::var_os("HIVE_MATRIX_SOCKET")
@ -253,17 +275,40 @@ async fn matrix_unread_rooms() -> Option<u32> {
}
let mut stream = UnixStream::connect(&socket).await.ok()?;
stream
.write_all(b"{\"method\":\"unread_count\"}\n")
.write_all(b"{\"method\":\"unread_summary\"}\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)
// Response: {"kind":"ok","payload":[{label, count, last_body?, last_sender?}]}
let arr = val.get("payload")?.as_array()?;
serde_json::from_value(serde_json::Value::Array(arr.clone())).ok()
}
/// Format a `Vec<MatrixRoomUnread>` into a per-room summary string.
/// Single room / single message collapses to one line; multi-room
/// expands to a bulleted list. Returns an empty string for empty input.
fn format_matrix_summary(rooms: &[MatrixRoomUnread]) -> String {
use std::fmt::Write as _;
if rooms.is_empty() {
return String::new();
}
let mut out = String::new();
for r in rooms {
if r.count == 1 {
if let (Some(body), Some(sender)) = (&r.last_body, &r.last_sender) {
let _ = writeln!(out, "- {}: {sender}: {body}", r.label);
continue;
}
}
let _ = writeln!(out, "- {}: {} unread", r.label, r.count);
}
// Remove trailing newline.
if out.ends_with('\n') {
out.pop();
}
out
}
/// Parse the user-facing `kind` string for `cancel_loose_end` into the
@ -759,9 +804,17 @@ impl AgentServer {
// 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 });
if let Some(unread_rooms) = matrix_unread_summary().await {
let total = unread_rooms.len() as u32;
if total > 0 {
let summary = format_matrix_summary(&unread_rooms);
loose_ends.insert(
0,
hive_sh4re::LooseEnd::UnreadMatrix {
rooms: total,
summary,
},
);
}
}
}
@ -925,9 +978,8 @@ impl AgentServer {
if let Some(wait) = args.wait_seconds {
const MAX_WAIT_SECS: u64 = 30;
const POLL_MS: u64 = 100;
let deadline =
tokio::time::Instant::now()
+ std::time::Duration::from_secs(wait.min(MAX_WAIT_SECS));
let deadline = tokio::time::Instant::now()
+ std::time::Duration::from_secs(wait.min(MAX_WAIT_SECS));
loop {
tokio::time::sleep(std::time::Duration::from_millis(POLL_MS)).await;
if let Some(task) = crate::bash_runner::read_task(&id) {
@ -2161,7 +2213,10 @@ pub fn allowed_mcp_tools(groups: &[hive_sh4re::ToolGroup]) -> Vec<String> {
pub fn allowed_tools_arg(flavor: Flavor) -> String {
let groups = effective_tool_groups(flavor);
// Base built-ins always present.
let mut all: Vec<String> = ALLOWED_BUILTIN_TOOLS.iter().map(|s| (*s).to_owned()).collect();
let mut all: Vec<String> = ALLOWED_BUILTIN_TOOLS
.iter()
.map(|s| (*s).to_owned())
.collect();
// Extra built-ins gated by tool groups (e.g. WebFetch/WebSearch via web_tools).
for group in &groups {
for tool in group.builtin_tools() {

7
hive-ag3nt/src/web_ui.rs
View file

@ -319,10 +319,9 @@ async fn relay_ws_vnc(socket: axum::extract::ws::WebSocket, vnc_port: u16) {
let ws_to_tcp = tokio::spawn(async move {
while let Some(Ok(msg)) = futures_util::StreamExt::next(&mut ws_rx).await {
match msg {
Message::Binary(data)
if tcp_tx.write_all(&data).await.is_err() => {
break;
}
Message::Binary(data) if tcp_tx.write_all(&data).await.is_err() => {
break;
}
Message::Close(_) => break,
_ => {} // ping/pong/text: ignore
}