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(#2569): migrate matrix producer to the in-agent todo socket (unread + invites); drop dead mcp.sock/wake plumbing

Browse source
This commit is contained in:
damocles 2026-07-20 23:09:54 +02:00
commit 21f1569a04
5 changed files with 113 additions and 166 deletions
Download patch file Download diff file Expand all files Collapse all files

56
hive-matrix-mcp/src/handlers.rs
View file

@ -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<String> = 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<OwnedServerName> = 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(),

34
hive-matrix-mcp/src/main.rs
View file

@ -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<String>,
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<String>,
is_primary: bool,
) -> Result<Option<(Client, SyncLoop)>> {
@ -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
}
})

19
hive-matrix-mcp/src/paths.rs
View file

@ -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 `<name>.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()
}

87
hive-matrix-mcp/src/timeline.rs
View file

@ -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<HashSet<OwnedRoomId>>,
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<HashSet<OwnedRoomId>>,
account_tag: Option<&str>,
) {
@ -103,38 +100,52 @@ pub async fn sweep_invites(
let current_ids: HashSet<OwnedRoomId> =
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 &current {
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<OwnedRoomId> = {
let seen = notified.lock().await;
seen.difference(&current_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:<room>` (distinct
// from the unread sweep's `<room>` 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 &current {
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}")
}

83
hive-matrix-mcp/src/wake.rs
View file

@ -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<str>) -> 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::path::PathBuf> {
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<str>) -> Result<()> {
pub async fn send_todo_upsert(key: &str, summary: impl AsRef<str>) -> 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