From 21f1569a04f3958fc44272d92a72ac293ff24818 Mon Sep 17 00:00:00 2001 From: damocles Date: Mon, 20 Jul 2026 23:09:54 +0200 Subject: [PATCH] feat(#2569): migrate matrix producer to the in-agent todo socket (unread + invites); drop dead mcp.sock/wake plumbing --- hive-matrix-mcp/src/handlers.rs | 56 ++++++--------------- hive-matrix-mcp/src/main.rs | 34 +++++-------- hive-matrix-mcp/src/paths.rs | 19 ------- hive-matrix-mcp/src/timeline.rs | 87 +++++++++++++++++++-------------- hive-matrix-mcp/src/wake.rs | 83 ++++++++++++++----------------- 5 files changed, 113 insertions(+), 166 deletions(-) diff --git a/hive-matrix-mcp/src/handlers.rs b/hive-matrix-mcp/src/handlers.rs index 7c4d906e..b4c1810d 100644 --- a/hive-matrix-mcp/src/handlers.rs +++ b/hive-matrix-mcp/src/handlers.rs @@ -512,44 +512,6 @@ pub 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_label(room); - format!( - "[matrix] pending invite: {label} — use list_invites to see, resolve_invite to accept or reject" - ) - }) - .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, @@ -560,9 +522,14 @@ 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) => { - // 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; + // Clear the invite todo so the accepted invite drops out of + // `get_loose_ends` immediately (the sweep would also clear it + // on its next tick, but this makes it instant). + let _ = crate::wake::send_todo_clear( + Some(&crate::timeline::invite_key(room.room_id())), + false, + ) + .await; DaemonResponse::ok(&serde_json::json!({ "joined": true, "room_id": room.room_id().to_string(), @@ -595,7 +562,12 @@ pub async fn resolve_invite( }; match room.leave().await { Ok(()) => { - refresh_invite_loose_ends(client).await; + // Clear the invite todo immediately on reject. + let _ = crate::wake::send_todo_clear( + Some(&crate::timeline::invite_key(room.room_id())), + false, + ) + .await; DaemonResponse::ok(&serde_json::json!({ "rejected": true, "room_id": room.room_id().to_string(), diff --git a/hive-matrix-mcp/src/main.rs b/hive-matrix-mcp/src/main.rs index 42d768dd..dbb4f39c 100644 --- a/hive-matrix-mcp/src/main.rs +++ b/hive-matrix-mcp/src/main.rs @@ -73,7 +73,6 @@ async fn main() -> Result<()> { let cfgs = accounts::configured().context("read matrix account config")?; let mcp_socket = paths::daemon_socket(); - let hyperhive_socket = paths::hyperhive_socket(); let multi = cfgs.len() > 1; let primary = cfgs[0].name.clone(); let mut registry = Registry::new(primary); @@ -81,10 +80,10 @@ async fn main() -> Result<()> { for (idx, cfg) in cfgs.into_iter().enumerate() { let is_primary = idx == 0; - // Account-tag the wakes only in multi-account mode so single- - // account wake bodies stay byte-identical to the legacy format. + // Account-tag the todos only in multi-account mode so single- + // account todo summaries stay byte-identical to the legacy format. let tag = multi.then(|| cfg.name.clone()); - match bring_up_account(&cfg, &hyperhive_socket, tag.clone(), is_primary).await { + match bring_up_account(&cfg, tag.clone(), is_primary).await { Ok(Some((client, sync_loop))) => { registry.insert(cfg.name, client); sync_loops.push(sync_loop); @@ -109,8 +108,7 @@ async fn main() -> Result<()> { // before being skipped for this daemon lifetime. Err(e) if is_primary => return Err(e.context("bring up primary matrix account")), Err(e) => { - if let Some((client, sync_loop)) = - bring_up_secondary_with_retry(&cfg, &hyperhive_socket, tag, e).await + if let Some((client, sync_loop)) = bring_up_secondary_with_retry(&cfg, tag, e).await { registry.insert(cfg.name, client); sync_loops.push(sync_loop); @@ -189,7 +187,6 @@ async fn main() -> Result<()> { /// failure, no token, or all retries exhausted). async fn bring_up_secondary_with_retry( cfg: &AccountCfg, - hyperhive_socket: &std::path::Path, tag: Option, first_error: anyhow::Error, ) -> Option<(Client, SyncLoop)> { @@ -219,7 +216,7 @@ async fn bring_up_secondary_with_retry( "retrying secondary account bring-up after backoff" ); tokio::time::sleep(std::time::Duration::from_secs(delay)).await; - match bring_up_account(cfg, hyperhive_socket, tag.clone(), false).await { + match bring_up_account(cfg, tag.clone(), false).await { Ok(Some((client, sync_loop))) => { tracing::info!(account = %cfg.name, "secondary matrix account recovered"); return Some((client, sync_loop)); @@ -260,7 +257,6 @@ async fn bring_up_secondary_with_retry( /// handling). The returned sync loop is driven by the caller. async fn bring_up_account( cfg: &AccountCfg, - hyperhive_socket: &std::path::Path, tag: Option, is_primary: bool, ) -> Result> { @@ -293,30 +289,26 @@ async fn bring_up_account( let sync_client = client.clone(); let cb_client = client.clone(); - let wake_socket = Arc::new(hyperhive_socket.to_path_buf()); - // Separate dedup sets: one tracks invites already woken about, the + // Separate dedup sets: one tracks invites already pushed as todos, the // other unread-message rooms. Both are pruned to their current state - // each sweep (see the sweep fns) so re-invites / new messages re-wake. + // each sweep (see the sweep fns) so re-invites / new messages re-push. let invite_notified = Arc::new(tokio::sync::Mutex::new(std::collections::HashSet::new())); let unread_notified = Arc::new(tokio::sync::Mutex::new(std::collections::HashSet::new())); // Startup cancel-and-recreate (loose-ends v2): wipe this agent's - // matrix todos so stale ones (rooms read while the daemon was down) don't - // linger, then let the first sweep rebuild the set to match current - // unread reality. Best-effort; the sweep converges regardless. - let _ = crate::wake::send_todo_clear(hyperhive_socket, None, true).await; + // matrix todos so stale ones (rooms read / invites resolved while the + // daemon was down) don't linger, then let the first sweep rebuild the + // set to match current reality. Best-effort; the sweep converges. + let _ = crate::wake::send_todo_clear(None, true).await; let sync_loop: SyncLoop = Box::pin(async move { sync_client .sync_with_callback(SyncSettings::default(), move |_response| { let client = cb_client.clone(); - let socket = wake_socket.clone(); let invite_notified = invite_notified.clone(); let unread_notified = unread_notified.clone(); let tag = tag.clone(); async move { - timeline::sweep_invites(&client, &socket, &invite_notified, tag.as_deref()) - .await; - timeline::sweep_unread(&client, &socket, &unread_notified, tag.as_deref()) - .await; + timeline::sweep_invites(&client, &invite_notified, tag.as_deref()).await; + timeline::sweep_unread(&client, &unread_notified, tag.as_deref()).await; matrix_sdk::LoopCtrl::Continue } }) diff --git a/hive-matrix-mcp/src/paths.rs b/hive-matrix-mcp/src/paths.rs index ccc8a441..dc9dcf17 100644 --- a/hive-matrix-mcp/src/paths.rs +++ b/hive-matrix-mcp/src/paths.rs @@ -72,22 +72,3 @@ pub fn accounts_file() -> PathBuf { let state_dir = std::env::var("HYPERHIVE_STATE_DIR").unwrap_or_default(); PathBuf::from(format!("{state_dir}/matrix-accounts.json")) } - -/// Hyperhive control socket — the daemon writes wake signals here so -/// the harness drives a new claude turn on incoming matrix events. -/// Mirrors the path `forge_notify` writes to. -#[must_use] -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. -/// Resolution lives in `hive_sh4re::paths` so the harness + every MCP -/// daemon agree on the location. -#[must_use] -pub fn mcp_loose_ends_dir() -> PathBuf { - hive_sh4re::paths::mcp_loose_ends_dir() -} diff --git a/hive-matrix-mcp/src/timeline.rs b/hive-matrix-mcp/src/timeline.rs index 50ebc082..6e46e4fa 100644 --- a/hive-matrix-mcp/src/timeline.rs +++ b/hive-matrix-mcp/src/timeline.rs @@ -1,19 +1,18 @@ -//! Matrix → hyperhive wake bridge. Both room messages and room invites -//! are surfaced to the agent by SWEEPING state on every post-sync -//! callback, not by one-shot `m.room.message` / `StrippedRoomMemberEvent` -//! handlers: a one-shot wake whose `send_wake` raced a hive-c0re / socket- -//! down window (a container rebuild) was dropped with no retry, leaving -//! the agent deaf to matrix activity until manually prompted. Sweeping the -//! reliable sync path re-checks each tick and self-heals a dropped wake on -//! the next one. +//! Matrix → hyperhive todo bridge (loose-ends v2). Both room messages and +//! room invites are surfaced to the agent by SWEEPING state on every +//! post-sync callback, not by one-shot `m.room.message` / +//! `StrippedRoomMemberEvent` handlers: a one-shot signal that raced a +//! socket-down window (a container rebuild) was dropped with no retry, +//! leaving the agent deaf to matrix activity until manually prompted. +//! Sweeping the reliable sync path re-checks each tick and self-heals a +//! dropped push on the next one — each todo is idempotent by room key. //! -//! Wake bodies stay short: `sweep_unread` sends the all-rooms unread +//! Todo summaries stay short: `sweep_unread` pushes the per-room unread //! summary (`wake::format_unread_summary`); the message stays unread //! server-side so the agent fetches detail via `read_room`. Self-sent //! messages never raise an unread notification, so they never wake. use std::collections::HashSet; -use std::path::Path; use matrix_sdk::{Client, ruma::OwnedRoomId}; use tokio::sync::Mutex; @@ -38,7 +37,6 @@ use crate::{handlers, wake}; /// no explicit self-filter is needed here. pub async fn sweep_unread( client: &Client, - socket: &Path, notified: &Mutex>, account_tag: Option<&str>, ) { @@ -53,7 +51,7 @@ pub async fn sweep_unread( active.difference(&unread_ids).cloned().collect() }; for id in stale { - if wake::send_todo_clear(socket, Some(id.as_str()), false) + if wake::send_todo_clear(Some(id.as_str()), false) .await .is_ok() { @@ -61,15 +59,15 @@ pub async fn sweep_unread( } } - // Upsert a todo per currently-unread room. hive-c0re coalesces the wake - // iff the summary is new or changed, so re-upserting an unchanged room - // every sync tick is a cheap server-side no-op (no re-wake). + // Upsert a todo per currently-unread room. The harness coalesces the + // wake iff the summary is new or changed, so re-upserting an unchanged + // room every sync tick is a cheap no-op (no re-wake). for (id, ru) in &unread { let summary = wake::tag_account( account_tag, wake::format_unread_summary(std::slice::from_ref(ru)), ); - match wake::send_todo_upsert(socket, id.as_str(), &summary).await { + match wake::send_todo_upsert(id.as_str(), &summary).await { Ok(()) => { notified.lock().await.insert(id.clone()); } @@ -95,7 +93,6 @@ pub async fn sweep_unread( /// decides whether to accept or reject by calling `resolve_invite`. pub async fn sweep_invites( client: &Client, - socket: &Path, notified: &Mutex>, account_tag: Option<&str>, ) { @@ -103,38 +100,52 @@ pub async fn sweep_invites( let current_ids: HashSet = current.iter().map(|r| r.room_id().to_owned()).collect(); - let mut fresh = Vec::new(); - { - let mut seen = notified.lock().await; - // Drop invites that are no longer pending (joined/rejected/withdrawn) - // so a future re-invite to the same room wakes the agent again. - seen.retain(|id| current_ids.contains(id)); - for room in ¤t { - if seen.insert(room.room_id().to_owned()) { - fresh.push(room.clone()); - } + // Invites we previously pushed a todo for that are no longer pending + // (joined / rejected / withdrawn) → clear their todo, then drop from + // `notified` on success so a future re-invite re-upserts (retry next + // tick on failure). + let stale: Vec = { + let seen = notified.lock().await; + seen.difference(¤t_ids).cloned().collect() + }; + for id in stale { + if wake::send_todo_clear(Some(&invite_key(&id)), false) + .await + .is_ok() + { + notified.lock().await.remove(&id); } } - if fresh.is_empty() { - return; - } - // 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; - for room in fresh { + // Upsert a todo per pending invite. Keyed `invite:` (distinct + // from the unread sweep's `` key) so the two never collide; the + // harness coalesces the wake iff the summary is new or changed, so + // re-upserting an unchanged invite every tick is a cheap no-op. + for room in ¤t { let room_id = room.room_id().to_owned(); let label = room.name().unwrap_or_else(|| room_id.to_string()); - tracing::info!(%room_id, "matrix: pending invite swept, waking agent"); - let body = wake::tag_account( + let summary = wake::tag_account( account_tag, format!( "[matrix] invited to {label} ({room_id}) — \ use list_invites to see pending invites, resolve_invite to accept or reject" ), ); - if let Err(e) = wake::send_wake(socket, &body).await { - tracing::warn!(error = %e, "matrix: failed to deliver invite-wake to hyperhive"); + match wake::send_todo_upsert(&invite_key(&room_id), &summary).await { + Ok(()) => { + notified.lock().await.insert(room_id); + } + Err(e) => { + tracing::warn!(error = %e, room = %room_id, "matrix: invite todo upsert failed; will retry next sweep"); + } } } } + +/// Todo dedup key for a pending invite. Namespaced with an `invite:` +/// prefix so an invited room and an unread room (which the unread sweep +/// keys by bare room id) never share a todo row. `pub(crate)` so the +/// invite-resolution handlers can clear the matching todo immediately. +pub(crate) fn invite_key(room_id: &matrix_sdk::ruma::RoomId) -> String { + format!("invite:{room_id}") +} diff --git a/hive-matrix-mcp/src/wake.rs b/hive-matrix-mcp/src/wake.rs index 18c41d62..4598af2c 100644 --- a/hive-matrix-mcp/src/wake.rs +++ b/hive-matrix-mcp/src/wake.rs @@ -1,17 +1,14 @@ -//! Wake-signal writer: notifies the hyperhive harness when an incoming -//! matrix event arrives so claude drives a new turn. +//! Todo writer: pushes matrix *todos* (loose-ends v2) to the harness's +//! in-agent socket (`HIVE_AGENT_SOCKET`) when rooms have unread messages +//! or pending invites, so claude drives a turn to handle them. One JSON +//! line per op (`upsert_todo` / `clear_todo`), keyed by room id so +//! re-pushing an unchanged item is an idempotent no-op and resolving one +//! clears it. The harness owns the todo store locally and signals its own +//! turn loop — no hive-c0re round-trip. //! -//! Same wire shape as `hive-agent::forge_notify`'s wake: a single JSON -//! line written to the hyperhive control socket (`/run/hive/mcp.sock` -//! by default) carrying an `Request::Wake { from, body }`. -//! The agent harness's `agent_server` parses it and treats it as a -//! `Wake` from the matrix subsystem. -//! -//! Per the operator's call (phase 3): the body is a SHORT TEASER, not -//! the full message — the agent then reads the unmarked event via -//! the `read_room` MCP tool. Truncation to ~100 chars keeps the wake -//! prompt focused (`forge_notify` embeds longer excerpts because the -//! agent doesn't have a follow-up read-the-original tool for forge). +//! Todo summaries stay short: a SHORT TEASER, not the full message — the +//! agent then reads the unmarked event via the `read_room` MCP tool. +//! Truncation to ~100 chars keeps the summary focused. use std::path::Path; @@ -19,65 +16,59 @@ use anyhow::{Context, Result}; use tokio::io::AsyncWriteExt; use tokio::net::UnixStream; -/// Send an `Request::Wake { from: "matrix", body }` to the hyperhive -/// control socket at `socket`. Best-effort: returns Err on any plumbing -/// failure; callers log + ignore so a wake delivery hiccup doesn't tear -/// down the matrix sync loop. -/// -/// Wire format matches `hive_core_agent_sock::Request` tagged with `"cmd"` per -/// `#[serde(tag = "cmd", rename_all = "snake_case")]`. Must be `"cmd"`, -/// not `"kind"` — the harness deserialises against the hive-sh4re type -/// and silently discards requests that don't match. -/// -/// # Errors -/// -/// Returns an error on socket connect failure, serialisation failure, -/// or I/O error writing to or reading from the socket. -pub async fn send_wake(socket: &Path, body: impl AsRef) -> Result<()> { - let payload = serde_json::json!({ - "cmd": "wake", - "from": "matrix", - "body": body.as_ref(), - "transient": true, - }); - send_line(socket, &payload).await +/// The harness-served in-agent socket (`HIVE_AGENT_SOCKET`) where todo ops +/// go — distinct from the host-served control socket used by [`send_wake`]. +/// `None` when unset/empty, in which case todo sends are a best-effort +/// no-op (a standalone daemon without the harness socket). +fn agent_socket() -> Option { + std::env::var_os("HIVE_AGENT_SOCKET") + .filter(|v| !v.is_empty()) + .map(std::path::PathBuf::from) } -/// Upsert a matrix-subsystem *todo* (loose-ends v2) on the -/// hyperhive control socket — the replacement for a direct wake. `key` is -/// the room id (the dedup key); hive-c0re coalesces a wake iff the todo is -/// new or its `summary` changed. Best-effort like [`send_wake`]. +/// Upsert a matrix-subsystem *todo* (loose-ends v2) on the harness's +/// in-agent socket — the replacement for a direct wake. `key` is the room +/// id (the dedup key); the harness signals a turn iff the todo is new or +/// its `summary` changed. Best-effort: a no-op when `HIVE_AGENT_SOCKET` +/// isn't configured. /// /// # Errors /// /// Returns an error on socket connect failure, serialisation failure, /// or I/O error writing to or reading from the socket. -pub async fn send_todo_upsert(socket: &Path, key: &str, summary: impl AsRef) -> Result<()> { +pub async fn send_todo_upsert(key: &str, summary: impl AsRef) -> Result<()> { + let Some(socket) = agent_socket() else { + return Ok(()); + }; let payload = serde_json::json!({ "cmd": "upsert_todo", "subsystem": "matrix", "key": key, "summary": summary.as_ref(), }); - send_line(socket, &payload).await + send_line(&socket, &payload).await } -/// Clear matrix-subsystem todos. `key = Some(room)` clears one room's -/// todo (it was read); `all = true` wipes the whole matrix set -/// (cancel-and-recreate on daemon restart). Best-effort. +/// Clear matrix-subsystem todos on the harness's in-agent socket. `key = +/// Some(room)` clears one room's todo (it was read); `all = true` wipes the +/// whole matrix set (cancel-and-recreate on daemon restart). Best-effort: +/// a no-op when `HIVE_AGENT_SOCKET` isn't configured. /// /// # Errors /// /// Returns an error on socket connect failure, serialisation failure, /// or I/O error writing to or reading from the socket. -pub async fn send_todo_clear(socket: &Path, key: Option<&str>, all: bool) -> Result<()> { +pub async fn send_todo_clear(key: Option<&str>, all: bool) -> Result<()> { + let Some(socket) = agent_socket() else { + return Ok(()); + }; let payload = serde_json::json!({ "cmd": "clear_todo", "subsystem": "matrix", "key": key, "all": all, }); - send_line(socket, &payload).await + send_line(&socket, &payload).await } /// Write one JSON request line to the hyperhive control socket and drain