diff --git a/hive-ag3nt/src/bin/hive.rs b/hive-ag3nt/src/bin/hive.rs index 8f899ed6..b31e5791 100644 --- a/hive-ag3nt/src/bin/hive.rs +++ b/hive-ag3nt/src/bin/hive.rs @@ -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}"), diff --git a/hive-ag3nt/src/mcp.rs b/hive-ag3nt/src/mcp.rs index a74f405e..83bd1c44 100644 --- a/hive-ag3nt/src/mcp.rs +++ b/hive-ag3nt/src/mcp.rs @@ -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) -> 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 { +/// 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, + last_sender: Option, +} + +/// 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> { 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 { } 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` 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 { pub fn allowed_tools_arg(flavor: Flavor) -> String { let groups = effective_tool_groups(flavor); // Base built-ins always present. - let mut all: Vec = ALLOWED_BUILTIN_TOOLS.iter().map(|s| (*s).to_owned()).collect(); + let mut all: Vec = 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() { diff --git a/hive-ag3nt/src/web_ui.rs b/hive-ag3nt/src/web_ui.rs index 6e588f29..2fae46e5 100644 --- a/hive-ag3nt/src/web_ui.rs +++ b/hive-ag3nt/src/web_ui.rs @@ -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 } diff --git a/hive-c0re/src/bin/hivectl.rs b/hive-c0re/src/bin/hivectl.rs index fee6c191..c847edda 100644 --- a/hive-c0re/src/bin/hivectl.rs +++ b/hive-c0re/src/bin/hivectl.rs @@ -501,9 +501,12 @@ fn gateway_list_users(file: &Path) -> Result<()> { // --------------------------------------------------------------------------- async fn agents_restart(socket: &Path, name: &str) -> Result<()> { - let resp = hive_c0re::client::request(socket, hive_sh4re::HostRequest::Restart { - name: name.to_owned(), - }) + let resp = hive_c0re::client::request( + socket, + hive_sh4re::HostRequest::Restart { + name: name.to_owned(), + }, + ) .await .with_context(|| format!("connect to daemon socket {}", socket.display()))?; if resp.ok { diff --git a/hive-c0re/src/broker.rs b/hive-c0re/src/broker.rs index 1d4d7e10..77e2bb31 100644 --- a/hive-c0re/src/broker.rs +++ b/hive-c0re/src/broker.rs @@ -121,7 +121,11 @@ pub enum MessageEvent { /// `recv_blocking_batch` for the target agent but is not stored, /// not re-delivered on restart, and not shown in message history. /// Used for bash task completion notifications. - Ping { to: String, from: String, body: String }, + Ping { + to: String, + from: String, + body: String, + }, } /// Per-recipient in-memory bookkeeping for the deliver-then-ack @@ -423,11 +427,7 @@ impl Broker { // Transient ping — not sqlite-backed. Return it directly as // a Delivery with id=0 (sentinel: never pushed to unacked_ids // so ack_turn silently ignores it). - Ok(Ok(MessageEvent::Ping { - to, - from, - body, - })) if to == recipient => { + Ok(Ok(MessageEvent::Ping { to, from, body })) if to == recipient => { // Also drain any real sqlite messages that may have landed // concurrently; prepend the ping so the agent sees both. let mut batch = self.recv_batch(recipient, max.saturating_sub(1))?; diff --git a/hive-c0re/src/dashboard.rs b/hive-c0re/src/dashboard.rs index 9eaa60e0..5165fafe 100644 --- a/hive-c0re/src/dashboard.rs +++ b/hive-c0re/src/dashboard.rs @@ -1856,7 +1856,8 @@ async fn get_build_log_stream( stderr_append: prog.stderr_append, status: prog.status, done, - }) && tx.send(Ok(Event::default().data(json))).await.is_err() { + }) && tx.send(Ok(Event::default().data(json))).await.is_err() + { return; // browser disconnected } if done { @@ -2582,7 +2583,10 @@ async fn post_capabilities( if let Some(reject) = guard_agent_name(&state, &logical).await { return reject; } - let known: Vec<&str> = hive_sh4re::Capability::ALL.iter().map(|c| c.as_str()).collect(); + let known: Vec<&str> = hive_sh4re::Capability::ALL + .iter() + .map(|c| c.as_str()) + .collect(); for cap in &body.caps { if !known.contains(&cap.as_str()) { return error_response(&format!("unknown capability: {cap}")); @@ -2724,7 +2728,9 @@ async fn post_start(State(state): State, AxumPath(name): AxumPath) -> Response { let containers = lifecycle::list().await.unwrap_or_default(); for container in containers { - let Some(logical) = container.strip_prefix(lifecycle::AGENT_PREFIX).map(str::to_owned) + let Some(logical) = container + .strip_prefix(lifecycle::AGENT_PREFIX) + .map(str::to_owned) else { continue; }; diff --git a/hive-c0re/src/lifecycle.rs b/hive-c0re/src/lifecycle.rs index 622e64ac..c0931f86 100644 --- a/hive-c0re/src/lifecycle.rs +++ b/hive-c0re/src/lifecycle.rs @@ -115,7 +115,11 @@ pub fn agent_network_ip(name: &str, subnet_cidr: &str) -> Option { } let base_u32 = u32::from_be_bytes([octets[0], octets[1], octets[2], octets[3]]); // Mask off host bits to get the true network address. - let mask = if prefix_len == 0 { 0u32 } else { !0u32 << (32 - prefix_len) }; + let mask = if prefix_len == 0 { + 0u32 + } else { + !0u32 << (32 - prefix_len) + }; let network_base = base_u32 & mask; let host_count: u32 = 1u32.checked_shl(32 - prefix_len).unwrap_or(0); // `.0` = network, `.1` = bridge gateway, last = broadcast → 3 reserved. @@ -625,7 +629,9 @@ async fn prebuild_toplevel(name: &str, flake_ref: &str) -> Result<()> { pub async fn list() -> Result> { let stdout = crate::priv_client::list_containers().await?; - Ok(stdout.lines().map(str::trim) + Ok(stdout + .lines() + .map(str::trim) .filter(|line| line.starts_with(AGENT_PREFIX)) .map(str::to_owned) .collect()) @@ -1025,7 +1031,8 @@ pub async fn git_update_ref(dir: &Path, refname: &str, target: &str) -> Result<( /// our default resource caps. Goes under `/run/systemd/system/...` so it's /// ephemeral (regenerated on every spawn / rebuild). async fn set_resource_limits(container: &str) -> Result<()> { - crate::priv_client::write_resource_limits(container, DEFAULT_MEMORY_MAX, DEFAULT_CPU_QUOTA).await + crate::priv_client::write_resource_limits(container, DEFAULT_MEMORY_MAX, DEFAULT_CPU_QUOTA) + .await } async fn systemd_daemon_reload() -> Result<()> { @@ -1076,9 +1083,21 @@ fn bind_child_agent_dirs(child: &str, binds: &mut Vec) { for dir in [&state_dir, &harness_dir, &config_dir] { let _ = std::fs::create_dir_all(dir); } - binds.push(BindMount { host_path: state_dir, container_path: format!("/agents/{child}/state"), read_only: false }); - binds.push(BindMount { host_path: harness_dir, container_path: format!("/agents/{child}/harness"), read_only: false }); - binds.push(BindMount { host_path: config_dir, container_path: format!("/agents/{child}/config"), read_only: false }); + binds.push(BindMount { + host_path: state_dir, + container_path: format!("/agents/{child}/state"), + read_only: false, + }); + binds.push(BindMount { + host_path: harness_dir, + container_path: format!("/agents/{child}/harness"), + read_only: false, + }); + binds.push(BindMount { + host_path: config_dir, + container_path: format!("/agents/{child}/config"), + read_only: false, + }); } async fn set_nspawn_flags( @@ -1103,25 +1122,49 @@ async fn set_nspawn_flags( let claude_mount = container_claude_mount(agent_name); let mut binds: Vec = vec![ - BindMount { host_path: runtime_dir.to_string_lossy().into_owned(), container_path: CONTAINER_RUNTIME_MOUNT.to_owned(), read_only: false }, - BindMount { host_path: claude_dir.to_string_lossy().into_owned(), container_path: claude_mount, read_only: false }, - BindMount { host_path: HOST_SHARED_ROOT.to_owned(), container_path: CONTAINER_SHARED_MOUNT.to_owned(), read_only: false }, + BindMount { + host_path: runtime_dir.to_string_lossy().into_owned(), + container_path: CONTAINER_RUNTIME_MOUNT.to_owned(), + read_only: false, + }, + BindMount { + host_path: claude_dir.to_string_lossy().into_owned(), + container_path: claude_mount, + read_only: false, + }, + BindMount { + host_path: HOST_SHARED_ROOT.to_owned(), + container_path: CONTAINER_SHARED_MOUNT.to_owned(), + read_only: false, + }, ]; // Own state, harness, and config dirs — same for every agent including // the manager. Config is RO: an agent must not edit its own config; changes // only ever flow through the approval queue. - binds.push(BindMount { host_path: notes_dir.to_string_lossy().into_owned(), container_path: format!("/agents/{agent_name}/state"), read_only: false }); + binds.push(BindMount { + host_path: notes_dir.to_string_lossy().into_owned(), + container_path: format!("/agents/{agent_name}/state"), + read_only: false, + }); if let Some(state_parent) = notes_dir.parent() { let harness_dir = state_parent.join("harness"); if !harness_dir.exists() { let _ = std::fs::create_dir_all(&harness_dir); } - binds.push(BindMount { host_path: harness_dir.to_string_lossy().into_owned(), container_path: format!("/agents/{agent_name}/harness"), read_only: false }); + binds.push(BindMount { + host_path: harness_dir.to_string_lossy().into_owned(), + container_path: format!("/agents/{agent_name}/harness"), + read_only: false, + }); } let own_config = format!("{HOST_AGENTS_ROOT}/{agent_name}/config"); std::fs::create_dir_all(&own_config).with_context(|| format!("create {own_config}"))?; - binds.push(BindMount { host_path: own_config, container_path: format!("/agents/{agent_name}/config"), read_only: true }); + binds.push(BindMount { + host_path: own_config, + container_path: format!("/agents/{agent_name}/config"), + read_only: true, + }); // Topology-driven child mounts: every direct child of this agent gets // its state, harness, and config dirs bind-mounted RW so the parent @@ -1152,8 +1195,16 @@ async fn set_nspawn_flags( // fires first (e.g. cold start with no agents). std::fs::create_dir_all(HOST_META_ROOT) .with_context(|| format!("create {HOST_META_ROOT}"))?; - binds.push(BindMount { host_path: HOST_APPLIED_ROOT.to_owned(), container_path: CONTAINER_MANAGER_APPLIED_MOUNT.to_owned(), read_only: true }); - binds.push(BindMount { host_path: HOST_META_ROOT.to_owned(), container_path: crate::meta::CONTAINER_MANAGER_META_MOUNT.to_owned(), read_only: true }); + binds.push(BindMount { + host_path: HOST_APPLIED_ROOT.to_owned(), + container_path: CONTAINER_MANAGER_APPLIED_MOUNT.to_owned(), + read_only: true, + }); + binds.push(BindMount { + host_path: HOST_META_ROOT.to_owned(), + container_path: crate::meta::CONTAINER_MANAGER_META_MOUNT.to_owned(), + read_only: true, + }); } // Web-socket subdir: bind-mount `/run/hive-agent//` into the @@ -1174,7 +1225,11 @@ async fn set_nspawn_flags( } else if let Err(e) = crate::priv_client::chmod_socket_dir(agent_name, 0o777).await { tracing::warn!(%agent_name, error = ?e, "chmod socket dir failed"); } - binds.push(BindMount { host_path: socket_dir.to_string_lossy().into_owned(), container_path: socket_dir.to_string_lossy().into_owned(), read_only: false }); + binds.push(BindMount { + host_path: socket_dir.to_string_lossy().into_owned(), + container_path: socket_dir.to_string_lossy().into_owned(), + read_only: false, + }); // Network isolation: when HIVE_NETWORK_ISOLATION=1 is set (by the // hive-network.nix module's `isolateContainers` option), flip the @@ -1382,7 +1437,11 @@ mod tests { let ip = agent_network_ip("alice", "10.42.0.0/24").expect("should produce an IP"); let octets: Vec = ip.split('.').map(|o| o.parse().unwrap()).collect(); assert_eq!(&octets[..3], &[10, 42, 0], "wrong /24 prefix"); - assert!(octets[3] >= 2 && octets[3] <= 254, "host byte {}", octets[3]); + assert!( + octets[3] >= 2 && octets[3] <= 254, + "host byte {}", + octets[3] + ); } #[test] @@ -1414,7 +1473,7 @@ mod tests { assert!(agent_network_ip("alice", "notanip/24").is_none()); assert!(agent_network_ip("alice", "10.0.0.0/33").is_none()); // prefix > 32 assert!(agent_network_ip("alice", "10.0.0.0/31").is_none()); // too small - assert!(agent_network_ip("alice", "10.0.0.0").is_none()); // no prefix + assert!(agent_network_ip("alice", "10.0.0.0").is_none()); // no prefix } #[test] @@ -1424,8 +1483,10 @@ mod tests { // after host-bit masking. let from_bridge = agent_network_ip("alice", "10.42.0.1/24"); let from_canonical = agent_network_ip("alice", "10.42.0.0/24"); - assert_eq!(from_bridge, from_canonical, - "bridge-IP and canonical-network form should normalize to the same result"); + assert_eq!( + from_bridge, from_canonical, + "bridge-IP and canonical-network form should normalize to the same result" + ); // Result must still be in .2-.254. let ip = from_bridge.unwrap(); let last: u8 = ip.rsplit('.').next().unwrap().parse().unwrap(); diff --git a/hive-c0re/src/manager_server.rs b/hive-c0re/src/manager_server.rs index 4d0fd629..8a4ccfc1 100644 --- a/hive-c0re/src/manager_server.rs +++ b/hive-c0re/src/manager_server.rs @@ -299,9 +299,14 @@ async fn dispatch(req: &ManagerRequest, coord: &Arc) -> ManagerResp let result = match agent.as_deref() { Some("*") => { // Hive-wide query requires query_agent_state capability. - if !crate::capabilities::has_cap(MANAGER_AGENT, hive_sh4re::Capability::QueryAgentState) { + if !crate::capabilities::has_cap( + MANAGER_AGENT, + hive_sh4re::Capability::QueryAgentState, + ) { return ManagerResponse::Err { - message: "query_agent_state capability required for hive-wide loose ends".into(), + message: + "query_agent_state capability required for hive-wide loose ends" + .into(), }; } crate::loose_ends::hive_wide(coord) diff --git a/hive-c0re/src/meta.rs b/hive-c0re/src/meta.rs index 8d39135e..32eea594 100644 --- a/hive-c0re/src/meta.rs +++ b/hive-c0re/src/meta.rs @@ -365,7 +365,10 @@ pub async fn commit_capabilities(agent: &str, caps: &[String]) -> Result<()> { /// logged as warnings and don't propagate: the topology write already /// succeeded, and `sync_agents` will pick up any un-committed change /// on the next run as a safety net. -pub async fn commit_topology(child: &str, new_parent: Option<&str>) -> std::result::Result<(), String> { +pub async fn commit_topology( + child: &str, + new_parent: Option<&str>, +) -> std::result::Result<(), String> { let _guard = META_LOCK.lock().await; crate::topology::set_parent(child, new_parent)?; let dir = meta_dir(); @@ -374,11 +377,7 @@ pub async fn commit_topology(child: &str, new_parent: Option<&str>) -> std::resu if has_staged_changes(&dir).await? { git_commit( &dir, - &format!( - "topology: {} → {}", - child, - new_parent.unwrap_or("") - ), + &format!("topology: {} → {}", child, new_parent.unwrap_or("")), ) .await?; } @@ -526,7 +525,10 @@ where out.push_str(" nixpkgs-unstable.follows = \"hyperhive/nixpkgs-unstable\";\n"); } else { let _ = writeln!(out, " nixpkgs.url = \"{nixpkgs_flake}\";"); - let _ = writeln!(out, " nixpkgs-unstable.url = \"{nixpkgs_unstable_flake}\";"); + let _ = writeln!( + out, + " nixpkgs-unstable.url = \"{nixpkgs_unstable_flake}\";" + ); let _ = writeln!(out, " hyperhive.url = \"{hyperhive_flake}\";"); out.push_str(" hyperhive.inputs.nixpkgs.follows = \"nixpkgs\";\n"); out.push_str(" hyperhive.inputs.nixpkgs-unstable.follows = \"nixpkgs-unstable\";\n"); diff --git a/hive-c0re/src/migrate.rs b/hive-c0re/src/migrate.rs index df4fc050..9d168773 100644 --- a/hive-c0re/src/migrate.rs +++ b/hive-c0re/src/migrate.rs @@ -219,7 +219,8 @@ async fn rename_manager_container(coord: &Arc) { // Move rootfs if it exists (may be absent for ephemeral containers). let old_rootfs = std::path::PathBuf::from("/var/lib/nixos-containers/root"); let new_rootfs = std::path::PathBuf::from("/var/lib/nixos-containers/h-root"); - if old_rootfs.exists() && !new_rootfs.exists() + if old_rootfs.exists() + && !new_rootfs.exists() && let Err(e) = std::fs::rename(&old_rootfs, &new_rootfs) { tracing::warn!(error = ?e, "migration phase 5: rename rootfs failed (non-fatal)"); diff --git a/hive-c0re/src/rebuild_queue.rs b/hive-c0re/src/rebuild_queue.rs index e5ce7777..0dd306de 100644 --- a/hive-c0re/src/rebuild_queue.rs +++ b/hive-c0re/src/rebuild_queue.rs @@ -277,7 +277,16 @@ impl RebuildQueue { reason: String, parent_id: Option, ) -> u64 { - self.enqueue_full(kind, agent, source, reason, parent_id, Vec::new(), None, None) + self.enqueue_full( + kind, + agent, + source, + reason, + parent_id, + Vec::new(), + None, + None, + ) } /// Same as `enqueue` but carries an `inputs` payload — used by @@ -1304,14 +1313,20 @@ mod tests { Some(meta), ); // The two Rebuilds have different parent_ids — must NOT dedup. - assert_ne!(sweep_rebuild, cascade_rebuild, - "cascade rebuild must be distinct from startup-sweep rebuild"); + assert_ne!( + sweep_rebuild, cascade_rebuild, + "cascade rebuild must be distinct from startup-sweep rebuild" + ); let snap = q.snapshot(); - let rebuilds: Vec<_> = snap.iter() + let rebuilds: Vec<_> = snap + .iter() .filter(|e| e.kind == QueueKind::Rebuild && e.agent == "alice") .collect(); - assert_eq!(rebuilds.len(), 2, - "both rebuilds must be present in the queue"); + assert_eq!( + rebuilds.len(), + 2, + "both rebuilds must be present in the queue" + ); } #[test] diff --git a/hive-c0re/src/tool_groups.rs b/hive-c0re/src/tool_groups.rs index b85d7bfd..105d4d73 100644 --- a/hive-c0re/src/tool_groups.rs +++ b/hive-c0re/src/tool_groups.rs @@ -72,8 +72,10 @@ fn write(map: &BTreeMap>) -> std::io::Result<()> { /// Returns `Ok(())` when all names are known, or `Err` listing the /// unrecognised names so callers can surface a useful error message. pub fn validate_groups(groups: &[String]) -> anyhow::Result<()> { - let valid: std::collections::BTreeSet<&str> = - hive_sh4re::ToolGroup::ALL.iter().map(|g| g.as_str()).collect(); + let valid: std::collections::BTreeSet<&str> = hive_sh4re::ToolGroup::ALL + .iter() + .map(|g| g.as_str()) + .collect(); let unknown: Vec<&str> = groups .iter() .map(String::as_str) diff --git a/hive-forge/src/verbs/pr_reviews.rs b/hive-forge/src/verbs/pr_reviews.rs index f19f1c16..8755f218 100644 --- a/hive-forge/src/verbs/pr_reviews.rs +++ b/hive-forge/src/verbs/pr_reviews.rs @@ -1,7 +1,7 @@ //! `pr-reviews ` — list reviews, or submit one via `--approve` / //! `--request-changes` / `--comment`. -use anyhow::{bail, Result}; +use anyhow::{Result, bail}; use clap::Args as ClapArgs; use serde_json::{Value, json}; @@ -50,8 +50,10 @@ pub fn run(client: &Client, args: Args) -> Result<()> { "event": ev, "body": args.body.unwrap_or_default(), }); - let v = client - .post_json(&format!("/repos/{repo}/pulls/{}/reviews", args.number), &payload)?; + let v = client.post_json( + &format!("/repos/{repo}/pulls/{}/reviews", args.number), + &payload, + )?; // Print a compact summary rather than the full review blob. let summary = json!({ "id": v.get("id"), @@ -64,8 +66,7 @@ pub fn run(client: &Client, args: Args) -> Result<()> { bail!("--body requires one of --approve / --request-changes / --comment"); } // List mode (original behaviour). - let v = - client.get_json(&format!("/repos/{repo}/pulls/{}/reviews", args.number))?; + let v = client.get_json(&format!("/repos/{repo}/pulls/{}/reviews", args.number))?; let trimmed: Vec = v .as_array() .map(|a| { diff --git a/hive-matrix-mcp/src/handlers.rs b/hive-matrix-mcp/src/handlers.rs index 3a6d5ba0..cc78ead5 100644 --- a/hive-matrix-mcp/src/handlers.rs +++ b/hive-matrix-mcp/src/handlers.rs @@ -357,3 +357,73 @@ pub fn unread_count(client: &Client) -> DaemonResponse { .count() as u32; DaemonResponse::ok(&serde_json::json!({ "rooms": rooms })) } + +/// Collect all rooms with unread notifications and build per-room +/// [`crate::protocol::RoomUnread`] entries. For rooms with exactly +/// one unread notification the last message body is fetched via the +/// `/messages` endpoint (best-effort; failures leave `last_body` as +/// `None`). Rooms with zero unreads are omitted. +pub async fn collect_unread(client: &Client) -> Vec { + use crate::protocol::RoomUnread; + let mut result = Vec::new(); + for room in client.joined_rooms() { + let count = room.unread_notification_counts().notification_count as u32; + if count == 0 { + continue; + } + let label = room + .canonical_alias() + .map(|a| a.to_string()) + .unwrap_or_else(|| room.room_id().to_string()); + let (last_body, last_sender) = if count == 1 { + fetch_last_message(client, &room).await + } else { + (None, None) + }; + result.push(RoomUnread { + label, + count, + last_body, + last_sender, + }); + } + result +} + +/// Fetch the body + sender of the most recent room message. Returns +/// `(None, None)` on any error or when the timeline contains no text +/// events. +async fn fetch_last_message( + client: &Client, + room: &matrix_sdk::Room, +) -> (Option, Option) { + use matrix_sdk::ruma::api::Direction; + use matrix_sdk::ruma::api::client::message::get_message_events; + let mut req = + get_message_events::v3::Request::new(room.room_id().to_owned(), Direction::Backward); + req.limit = matrix_sdk::ruma::UInt::from(1u32); + let resp = match client.send(req).await { + Ok(r) => r, + Err(_) => return (None, None), + }; + for raw in &resp.chunk { + if let Ok(ev) = raw.deserialize() { + let body = extract_body(&ev); + if !body.is_empty() { + return ( + Some(crate::wake::truncate_chars(&body, 100)), + Some(ev.sender().to_string()), + ); + } + } + } + (None, None) +} + +/// Return per-room unread summaries. For rooms with exactly one +/// unread notification, attempts to include the sender + truncated +/// body; rooms with multiple unreads carry only the count. +pub async fn unread_summary(client: &Client) -> DaemonResponse { + let rooms = collect_unread(client).await; + DaemonResponse::ok(&rooms) +} diff --git a/hive-matrix-mcp/src/protocol.rs b/hive-matrix-mcp/src/protocol.rs index 8a4cfe6f..02c90956 100644 --- a/hive-matrix-mcp/src/protocol.rs +++ b/hive-matrix-mcp/src/protocol.rs @@ -87,6 +87,13 @@ pub enum DaemonRequest { #[serde(rename = "unread_count")] UnreadCount, + /// Return per-room unread summaries. For rooms with exactly one + /// unread notification, attempts to include the sender + truncated + /// body; rooms with multiple unreads carry only the count. Used by + /// `get_loose_ends` and the wake-signal formatter. + #[serde(rename = "unread_summary")] + UnreadSummary, + /// 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 @@ -96,6 +103,23 @@ pub enum DaemonRequest { Ping, } +/// One entry in the [`DaemonRequest::UnreadSummary`] response payload. +#[derive(Debug, Serialize, Deserialize)] +pub struct RoomUnread { + /// Canonical alias (`#name:server`) or room id (`!id:server`). + pub label: String, + /// Server-side push-notification count for this room. Always ≥ 1. + pub count: u32, + /// Truncated body of the last message in the room. Present only when + /// `count == 1` and the fetch succeeded; absent otherwise. + #[serde(skip_serializing_if = "Option::is_none")] + pub last_body: Option, + /// Sender of the last message (`@user:server`). Present when + /// `last_body` is present. + #[serde(skip_serializing_if = "Option::is_none")] + pub last_sender: Option, +} + /// Response shape: `ok` carries the payload (any JSON; the MCP bridge /// passes it back to claude as the tool result), `error` carries a /// human-readable error string. diff --git a/hive-matrix-mcp/src/socket.rs b/hive-matrix-mcp/src/socket.rs index 5544e765..5631b15b 100644 --- a/hive-matrix-mcp/src/socket.rs +++ b/hive-matrix-mcp/src/socket.rs @@ -84,5 +84,6 @@ async fn dispatch(req: DaemonRequest, client: &Client) -> DaemonResponse { 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), + DaemonRequest::UnreadSummary => handlers::unread_summary(client).await, } } diff --git a/hive-matrix-mcp/src/timeline.rs b/hive-matrix-mcp/src/timeline.rs index 2b695497..90ae454f 100644 --- a/hive-matrix-mcp/src/timeline.rs +++ b/hive-matrix-mcp/src/timeline.rs @@ -16,18 +16,20 @@ use matrix_sdk::{ ruma::events::room::message::{MessageType, OriginalSyncRoomMessageEvent}, }; -use crate::wake; +use crate::{handlers, wake}; /// Install the room-message handler on `client`. Fires on every -/// `m.room.message` event in a joined room; non-self text messages -/// trigger a wake signal to the hyperhive harness via the unix socket -/// at `hyperhive_socket`. +/// `m.room.message` event in a joined room; non-self messages trigger +/// a wake signal to the hyperhive harness via the unix socket at +/// `hyperhive_socket`. The wake body summarises ALL rooms with unread +/// notifications at wake time (not just the triggering event) so the +/// agent receives a full picture in one prompt. pub fn install_message_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({ let socket = socket.clone(); - move |event: OriginalSyncRoomMessageEvent, room: Room, _client: Client| { + move |event: OriginalSyncRoomMessageEvent, room: Room, client: Client| { let socket = socket.clone(); let own_user = own_user.clone(); async move { @@ -39,21 +41,29 @@ pub fn install_message_handler(client: &Client, hyperhive_socket: PathBuf) { if own_user.as_ref().is_some_and(|u| u == &event.sender) { return; } - let text = match &event.content.msgtype { - MessageType::Text(t) => t.body.clone(), - MessageType::Notice(n) => n.body.clone(), - MessageType::Emote(e) => format!("* {}", e.body), - _ => { - // Non-text content (image / file / location / etc.) — - // still wake, but with a placeholder body so the - // agent knows something landed and can read_room. - format!("[{}]", event.content.msgtype()) - } + // Build a wake body that covers all rooms with unread + // notifications, not just the triggering event. This lets + // the agent see the full backlog in a single wake prompt. + // Falls back to the per-event teaser if the collect fails + // (empty result means daemon is not seeing any unread yet — + // unlikely but possible during a sync race). + let unread = handlers::collect_unread(&client).await; + let body = if unread.is_empty() { + // Sync hasn't updated notification counts yet; fall back + // to the current event so the agent still wakes. + let text = match &event.content.msgtype { + MessageType::Text(t) => t.body.clone(), + MessageType::Notice(n) => n.body.clone(), + MessageType::Emote(e) => format!("* {}", e.body), + _ => format!("[{}]", event.content.msgtype()), + }; + let room_label = room + .canonical_alias() + .map_or_else(|| room.room_id().to_string(), |a| a.to_string()); + wake::format_wake_body(event.sender.as_str(), &room_label, &text) + } else { + wake::format_unread_summary(&unread) }; - let room_label = room - .canonical_alias() - .map_or_else(|| room.room_id().to_string(), |a| a.to_string()); - let body = wake::format_wake_body(event.sender.as_str(), &room_label, &text); if let Err(e) = wake::send_wake(&socket, &body).await { tracing::warn!(error = %e, "failed to deliver matrix wake to hyperhive"); } else { diff --git a/hive-matrix-mcp/src/wake.rs b/hive-matrix-mcp/src/wake.rs index 5ce5b8e5..45766c5c 100644 --- a/hive-matrix-mcp/src/wake.rs +++ b/hive-matrix-mcp/src/wake.rs @@ -72,10 +72,51 @@ pub fn format_wake_body(sender: &str, room: &str, text: &str) -> String { format!("[matrix] {sender} in {room}: {truncated}") } +/// Format a wake-message body from a list of per-room unread summaries. +/// Single-room / single-message case collapses to the terse one-liner +/// format; multiple rooms expand to a bulleted list. Always appends a +/// read-hint line so the agent knows which tools to reach for. +#[must_use] +pub fn format_unread_summary(rooms: &[crate::protocol::RoomUnread]) -> String { + use std::fmt::Write as _; + if rooms.is_empty() { + return String::new(); + } + // Terse path: exactly one room, exactly one unread with body. + if rooms.len() == 1 { + let r = &rooms[0]; + if r.count == 1 { + if let (Some(body), Some(sender)) = (&r.last_body, &r.last_sender) { + return format!( + "[matrix] {sender} in {label}: {body} — use read_room to view, mark_read to clear", + label = r.label + ); + } + } + return format!( + "[matrix] {} unread in {} — use read_room to view, mark_read to clear", + r.count, r.label + ); + } + // Multi-room path. + let mut out = String::from("[matrix] unread messages:"); + for r in rooms { + if r.count == 1 { + if let (Some(body), Some(sender)) = (&r.last_body, &r.last_sender) { + let _ = write!(out, "\n- {}: {sender}: {body}", r.label); + continue; + } + } + let _ = write!(out, "\n- {}: {} unread", r.label, r.count); + } + out.push_str("\nUse list_rooms + read_room to view, mark_read to clear."); + out +} + /// Truncate `s` to `max` Unicode chars, appending `…` when cut. /// Char-based not byte-based so multi-byte content (most chat) doesn't /// get cut mid-codepoint. -fn truncate_chars(s: &str, max: usize) -> String { +pub fn truncate_chars(s: &str, max: usize) -> String { let mut end = s.len(); for (count, (i, _)) in s.char_indices().enumerate() { if count == max { diff --git a/hive-sh4re/src/lib.rs b/hive-sh4re/src/lib.rs index 4020d4c6..ec51197a 100644 --- a/hive-sh4re/src/lib.rs +++ b/hive-sh4re/src/lib.rs @@ -297,6 +297,11 @@ pub enum LooseEnd { UnreadMatrix { /// Number of rooms with at least one unread notification. rooms: u32, + /// Per-room summary: one line per room with truncated last-message + /// body when count is 1, or just the unread count otherwise. Empty + /// when the daemon returned no per-room detail. + #[serde(default)] + summary: String, }, } @@ -923,9 +928,7 @@ impl ToolGroup { "get_logs — read a sub-agent container's systemd journal (privileged)" } Self::Execution => "bash_run, bash_status — run shell commands in the container", - Self::WebTools => { - "WebFetch, WebSearch — Claude built-in web egress; not MCP tools" - } + Self::WebTools => "WebFetch, WebSearch — Claude built-in web egress; not MCP tools", } } }