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

Compare commits

base: vinzenz:30c9eed56258710140476e6a6fc1636a0c6996fb
Branches Tags
vinzenz:main
..
compare: vinzenz:23ed12488153f72b0c674f4a143315d99ed80288
Branches Tags
vinzenz:main
16 changed files with 44 additions and 3320 deletions
Download patch file Download diff file Expand all files Collapse all files

17
CLAUDE.md
View file

@ -203,23 +203,6 @@ hive-forge/ Forgejo CLI wrapper (`hive-forge` binary)
attach-comment, lint). Replaces the 600-line attach-comment, lint). Replaces the 600-line
hive-forge-tools.nix bash script (closes #280). hive-forge-tools.nix bash script (closes #280).
hive-matrix-mcp/ per-agent matrix-sdk integration (#548 phase 3).
src/main.rs `hive-matrix-daemon` binary entry — long-running
matrix-sdk Client + sync per agent; serves the
MCP bridge over /run/hive-matrix.sock; emits
hyperhive wake on incoming room events.
src/bin/mcp.rs `hive-matrix-mcp` stdio bridge claude spawns per
turn — connects to the daemon socket, forwards
each tool call (send_message, send_dm,
send_reaction, send_reply, mark_read, list_rooms,
list_room_members, read_room). No matrix-sdk dep
at this entrypoint (pure serde + tokio I/O).
src/{client,handlers,socket,timeline,wake,protocol,paths}.rs
shared library modules (matrix-sdk session
restore, request dispatch, unix socket server,
event handler → wake bridge, wire types, env
resolution).
hive-sh4re/ wire types (HostRequest/Response, AgentRequest/Response, hive-sh4re/ wire types (HostRequest/Response, AgentRequest/Response,
ManagerRequest/Response, Message, Approval, HelperEvent) ManagerRequest/Response, Message, Approval, HelperEvent)

1905
Cargo.lock generated
View file

File diff suppressed because it is too large Load diff

4
Cargo.toml
View file

@ -1,6 +1,6 @@
[workspace] [workspace]
resolver = "3" resolver = "3"
members = ["hive-ag3nt", "hive-c0re", "hive-forge", "hive-matrix-mcp", "hive-sh4re"] members = ["hive-ag3nt", "hive-c0re", "hive-forge", "hive-sh4re"]
[workspace.package] [workspace.package]
edition = "2024" edition = "2024"
@ -46,5 +46,3 @@ tokio-stream = { version = "0.1", features = ["sync"] }
tracing = "0.1" tracing = "0.1"
tracing-subscriber = { version = "0.3", features = ["env-filter"] } tracing-subscriber = { version = "0.3", features = ["env-filter"] }
reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls"] } reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls"] }
matrix-sdk = { version = "0.14", default-features = false, features = ["rustls-tls", "sqlite", "markdown"] }
futures-util = "0.3"

6
flake.nix
View file

@ -93,14 +93,8 @@
# rsvg-convert call — that whole codepath moved into the # rsvg-convert call — that whole codepath moved into the
# `hyperhive-assets` derivation in #555, so the rust # `hyperhive-assets` derivation in #555, so the rust
# derivation no longer needs the dependency. # derivation no longer needs the dependency.
# `sqlite` required by matrix-sdk's `sqlite` feature
# (`hive-matrix-mcp` workspace member, #548 phase 3) — the
# matrix-sdk-sqlite + rusqlite stack links against system
# libsqlite3 by default.
nativeBuildInputs = [ nativeBuildInputs = [
pkgs.git pkgs.git
pkgs.sqlite
pkgs.pkg-config
]; ];
} }
); );

36
hive-matrix-mcp/Cargo.toml
View file

@ -1,36 +0,0 @@
[package]
name = "hive-matrix-mcp"
edition.workspace = true
version.workspace = true
[lints]
workspace = true
[dependencies]
anyhow.workspace = true
futures-util.workspace = true
matrix-sdk.workspace = true
reqwest.workspace = true
rmcp.workspace = true
schemars.workspace = true
serde.workspace = true
serde_json.workspace = true
tokio.workspace = true
tracing.workspace = true
tracing-subscriber.workspace = true
# `hive-matrix-daemon` — long-running per-agent matrix-sdk Client +
# sync loop. Holds the unix socket the stdio MCP bridge talks to and
# emits hyperhive wake signals on incoming room events.
[[bin]]
name = "hive-matrix-daemon"
path = "src/main.rs"
# `hive-matrix-mcp` — thin stdio MCP bridge spawned by claude per turn.
# Forwards every tool call to the daemon over /run/hive-matrix.sock,
# returns the daemon's response shape to claude. No matrix-sdk dep at
# this entrypoint — the heavy crate only loads when the daemon binary
# is invoked.
[[bin]]
name = "hive-matrix-mcp"
path = "src/bin/mcp.rs"

264
hive-matrix-mcp/src/bin/mcp.rs
View file

@ -1,264 +0,0 @@
//! `hive-matrix-mcp` binary — stdio MCP server claude spawns per turn.
//! Thin protocol bridge: every tool call → connect to the daemon's
//! unix socket → write a JSON request line → read the JSON response →
//! return the payload (or the error message) to claude.
//!
//! No matrix-sdk dep at this entrypoint. The daemon owns the heavy
//! Client + sync; the bridge is pure serde + tokio I/O. Means the MCP
//! binary cold-starts in milliseconds even though the daemon takes
//! seconds to bring up sync.
use anyhow::{Context, Result};
use rmcp::{
ServerHandler, ServiceExt,
handler::server::wrapper::Parameters,
schemars::{self, JsonSchema},
tool, tool_handler, tool_router,
transport::stdio,
};
use serde::Deserialize;
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::net::UnixStream;
use hive_matrix_mcp::paths;
use hive_matrix_mcp::protocol::{DaemonRequest, DaemonResponse};
/// Send `req` to the daemon and read back the response. Each call is a
/// fresh unix-socket connection — short-lived (the daemon dispatch is
/// a single round-trip) so connection pooling would be over-engineering.
async fn round_trip(req: DaemonRequest) -> Result<DaemonResponse> {
let socket = paths::daemon_socket();
let stream = UnixStream::connect(&socket)
.await
.with_context(|| format!("connect daemon socket {}", socket.display()))?;
let (reader, mut writer) = stream.into_split();
let mut line = serde_json::to_string(&req)?;
line.push('\n');
writer
.write_all(line.as_bytes())
.await
.context("write request to daemon socket")?;
writer.shutdown().await.ok();
let mut buf = String::new();
BufReader::new(reader)
.read_line(&mut buf)
.await
.context("read response from daemon socket")?;
serde_json::from_str(&buf).context("parse daemon response")
}
/// Turn a `DaemonResponse` into the string claude sees as the tool
/// result. Ok payloads are pretty-printed JSON; errors get a clear
/// "matrix error: …" prefix so claude can pattern-match on it.
fn render(resp: Result<DaemonResponse>) -> String {
match resp {
Ok(DaemonResponse::Ok { payload }) => {
serde_json::to_string_pretty(&payload).unwrap_or_else(|_| payload.to_string())
}
Ok(DaemonResponse::Error { message }) => format!("matrix error: {message}"),
Err(e) => format!("matrix bridge error: {e:#}"),
}
}
#[derive(Debug, Deserialize, JsonSchema)]
struct SendMessageArgs {
/// Matrix room id (`!abc:server`) or canonical alias (`#name:server`).
/// Aliases are server-resolved at send time.
room: String,
/// Message body. Markdown is rendered to HTML by the daemon
/// (`text_markdown`); plain text passes through unchanged.
body: String,
}
#[derive(Debug, Deserialize, JsonSchema)]
struct SendDmArgs {
/// Matrix user id of the recipient (`@user:server`). DM room is
/// created if one doesn't already exist between this agent and
/// the recipient.
user_id: String,
body: String,
}
#[derive(Debug, Deserialize, JsonSchema)]
struct SendReactionArgs {
room: String,
/// Full matrix event id of the message being reacted to.
event_id: String,
/// Reaction key — usually an emoji (`👍`, `❤️`) but any string
/// works per matrix spec.
key: String,
}
#[derive(Debug, Deserialize, JsonSchema)]
struct SendReplyArgs {
room: String,
event_id: String,
body: String,
}
#[derive(Debug, Deserialize, JsonSchema)]
struct MarkReadArgs {
room: String,
event_id: String,
}
#[derive(Debug, Deserialize, JsonSchema)]
struct ListRoomsArgs {}
#[derive(Debug, Deserialize, JsonSchema)]
struct ListRoomMembersArgs {
room: String,
}
#[derive(Debug, Deserialize, JsonSchema)]
struct ReadRoomArgs {
room: String,
/// Maximum events to return (default 50, max 200). Newest first.
#[serde(default)]
limit: Option<usize>,
}
struct MatrixBridge {
#[allow(dead_code)]
tool_router: rmcp::handler::server::router::tool::ToolRouter<Self>,
}
impl MatrixBridge {
fn new() -> Self {
Self {
tool_router: Self::tool_router(),
}
}
}
#[tool_router]
impl MatrixBridge {
#[tool(
description = "Post a plain-text or markdown message to a matrix room. \
`room` is either a room id (!abc:server) or alias (#name:server). \
Returns the new event id."
)]
async fn send_message(&self, Parameters(args): Parameters<SendMessageArgs>) -> String {
render(round_trip(DaemonRequest::SendMessage {
room: args.room,
body: args.body,
}).await)
}
#[tool(
description = "Open (or reuse) a direct message room with `user_id` \
(@user:server) and post `body` to it."
)]
async fn send_dm(&self, Parameters(args): Parameters<SendDmArgs>) -> String {
render(round_trip(DaemonRequest::SendDm {
user_id: args.user_id,
body: args.body,
}).await)
}
#[tool(
description = "React to a specific matrix event with an emoji or short \
string `key`. Matrix-spec annotation; renders as a reaction in \
standard clients."
)]
async fn send_reaction(&self, Parameters(args): Parameters<SendReactionArgs>) -> String {
render(round_trip(DaemonRequest::SendReaction {
room: args.room,
event_id: args.event_id,
key: args.key,
}).await)
}
#[tool(
description = "Reply to a specific matrix event in a room, threaded \
via m.in_reply_to. Returns the reply's event id."
)]
async fn send_reply(&self, Parameters(args): Parameters<SendReplyArgs>) -> String {
render(round_trip(DaemonRequest::SendReply {
room: args.room,
event_id: args.event_id,
body: args.body,
}).await)
}
#[tool(
description = "Mark a specific event as read for this agent. Updates \
the room's unread indicator + sends a read receipt other \
participants can see."
)]
async fn mark_read(&self, Parameters(args): Parameters<MarkReadArgs>) -> String {
render(round_trip(DaemonRequest::MarkRead {
room: args.room,
event_id: args.event_id,
}).await)
}
#[tool(
description = "List rooms this agent has joined. Each row has the room \
id, canonical alias (when set), display name, and joined-member count."
)]
async fn list_rooms(&self, Parameters(_): Parameters<ListRoomsArgs>) -> String {
render(round_trip(DaemonRequest::ListRooms).await)
}
#[tool(
description = "List the members of a matrix room (joined-state only). \
Each row carries the user id and resolved display name."
)]
async fn list_room_members(&self, Parameters(args): Parameters<ListRoomMembersArgs>) -> String {
render(round_trip(DaemonRequest::ListRoomMembers { room: args.room }).await)
}
#[tool(
description = "Read the most recent N events from a matrix room \
(default 50, max 200). Returns each event's id, sender, timestamp, \
type, and best-effort plain-text body."
)]
async fn read_room(&self, Parameters(args): Parameters<ReadRoomArgs>) -> String {
render(round_trip(DaemonRequest::ReadRoom {
room: args.room,
limit: args.limit,
}).await)
}
}
#[tool_handler(
instructions = "Matrix client for an agent on a hyperhive swarm. Use \
`send_message` to post in a joined room, `send_dm` to message a \
specific user, `send_reaction` to react with an emoji, `send_reply` \
to thread a reply, `mark_read` to acknowledge an event. Discover \
rooms with `list_rooms`, members with `list_room_members`, recent \
timeline with `read_room`. Room references accept ids (!abc:server) \
or aliases (#name:server); user references use @user:server."
)]
impl ServerHandler for MatrixBridge {}
#[tokio::main]
async fn main() -> Result<()> {
tracing_subscriber::fmt()
.with_env_filter(
tracing_subscriber::EnvFilter::try_from_default_env()
.unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("warn")),
)
.with_writer(std::io::stderr)
.init();
// Standalone-degraded boot: missing daemon socket → exit 0
// cleanly so claude doesn't see an MCP startup error when matrix
// hasn't been provisioned for this agent yet. The daemon will
// appear once hive-c0re writes the token + systemd starts the
// daemon unit.
let socket = paths::daemon_socket();
if !tokio::fs::try_exists(&socket).await.unwrap_or(false) {
tracing::warn!(
path = %socket.display(),
"matrix daemon socket absent; exiting cleanly so MCP startup doesn't fail"
);
return Ok(());
}
let bridge = MatrixBridge::new();
let service = bridge.serve(stdio()).await.context("serve MCP over stdio")?;
service.waiting().await.context("MCP service exited unexpectedly")?;
Ok(())
}

128
hive-matrix-mcp/src/client.rs
View file

@ -1,128 +0,0 @@
//! matrix-sdk `Client` setup for the daemon: read the per-agent access
//! token from the state-dir file `hive-c0re::matrix::ensure_user_for`
//! wrote, probe `whoami` to recover the agent's matrix `user_id` +
//! `device_id`, restore the matrix-sdk session, return the Client ready
//! to start sync.
//!
//! No OAuth dance / cross-signing setup (in contrast to damocles-daemon's
//! ccc.de connection): for the in-hive tuwunel the `registration_token`
//! UIAA flow already minted the token + user/device, hive-c0re just
//! handed us the bearer in a file. matrix-sdk's `restore_session` with
//! a constructed `MatrixSession` skips the login flow entirely.
use std::path::Path;
use anyhow::{Context, Result, anyhow};
use matrix_sdk::{
Client, SessionMeta, SessionTokens,
authentication::matrix::MatrixSession,
ruma::{OwnedDeviceId, OwnedUserId},
};
use serde::Deserialize;
use tokio::fs;
/// Subset of the `/_matrix/client/v3/account/whoami` response we care
/// about. matrix-spec field names; `device_id` is optional per spec
/// (servers MAY omit it for legacy bearer scopes) but tuwunel always
/// returns it.
#[derive(Debug, Deserialize)]
struct WhoamiResponse {
user_id: String,
device_id: Option<String>,
}
/// Build + restore a matrix-sdk `Client` for the per-agent bearer
/// token at `token_file`. The Client points at `homeserver`, persists
/// its sqlite cache under `state_dir`, and is ready for sync once
/// returned.
///
/// Steps:
/// 1. Read the bearer token from `token_file` (trim trailing whitespace).
/// 2. Plain reqwest GET to `/_matrix/client/v3/account/whoami` with
/// the bearer — this gives us back the matrix `user_id` +
/// `device_id` (the registration response had them but hive-c0re
/// only persisted the token; whoami is the cheapest recovery path
/// and avoids matrix-sdk's circular requirement of needing a
/// session to call whoami).
/// 3. Build the real Client with the sqlite store + `restore_session`
/// using a synthetic `MatrixSession`.
pub async fn build_and_restore(
homeserver: &str,
token_file: &Path,
state_dir: &Path,
) -> Result<Client> {
let token = fs::read_to_string(token_file)
.await
.with_context(|| format!("read matrix token from {}", token_file.display()))?
.trim()
.to_owned();
if token.is_empty() {
return Err(anyhow!(
"matrix token at {} is empty",
token_file.display()
));
}
let (user_id, device_id) = whoami(homeserver, &token).await?;
fs::create_dir_all(state_dir)
.await
.with_context(|| format!("mkdir matrix state dir {}", state_dir.display()))?;
let client = Client::builder()
.homeserver_url(homeserver)
.sqlite_store(state_dir, None)
.build()
.await
.with_context(|| format!("build matrix client for {homeserver}"))?;
let session = MatrixSession {
meta: SessionMeta { user_id, device_id },
tokens: SessionTokens {
access_token: token,
refresh_token: None,
},
};
client
.restore_session(session)
.await
.context("restore matrix session")?;
tracing::info!(
user = %client.user_id().map(ToString::to_string).unwrap_or_default(),
device = %client.device_id().map(ToString::to_string).unwrap_or_default(),
"matrix session restored"
);
Ok(client)
}
/// Bare-reqwest whoami probe — used at startup to recover the
/// `user_id` + `device_id` the registration response carried but
/// hive-c0re didn't persist alongside the access token. Cheaper than
/// teaching the hive-c0re side to persist them, plus matches what a
/// fresh deployment with a hand-rolled token (`HIVE_MATRIX_TOKEN_FILE`
/// pointing somewhere unexpected) needs anyway.
async fn whoami(homeserver: &str, token: &str) -> Result<(OwnedUserId, OwnedDeviceId)> {
let url = format!("{homeserver}/_matrix/client/v3/account/whoami");
let resp = reqwest::Client::builder()
.timeout(std::time::Duration::from_secs(10))
.build()
.context("build whoami reqwest client")?
.get(&url)
.bearer_auth(token)
.send()
.await
.with_context(|| format!("GET {url}"))?;
let status = resp.status();
if !status.is_success() {
let body = resp.text().await.unwrap_or_default();
return Err(anyhow!("whoami GET {url} → HTTP {status}, body: {body}"));
}
let body: WhoamiResponse = resp.json().await.context("parse whoami response")?;
let user_id: OwnedUserId = body
.user_id
.parse()
.with_context(|| format!("invalid user_id in whoami: {}", body.user_id))?;
let device_id_raw = body
.device_id
.ok_or_else(|| anyhow!("whoami response missing device_id"))?;
let device_id: OwnedDeviceId = device_id_raw.into();
Ok((user_id, device_id))
}

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

@ -1,305 +0,0 @@
//! Per-tool dispatch — each daemon request from the MCP bridge resolves
//! to one of these handlers. Returns a `DaemonResponse` shaped for the
//! wire protocol (the MCP bridge unwraps `Ok { payload }` and returns
//! the payload to claude; `Error { message }` becomes the tool-call
//! error message claude sees).
//!
//! Tool surface mirrors damocles-daemon's v0 set per mara on #548:
//! `send_message`, `send_dm`, `send_reaction`, `send_reply`, `mark_read`,
//! `list_rooms`, `list_room_members`, `read_room`. Plus a `ping` for the
//! MCP bridge's liveness probe.
use matrix_sdk::{
Client,
room::reply::{EnforceThread, Reply},
ruma::{
OwnedEventId, OwnedRoomId, OwnedUserId, RoomOrAliasId,
api::client::receipt::create_receipt::v3::ReceiptType,
events::{
receipt::ReceiptThread,
reaction::ReactionEventContent,
relation::Annotation,
room::message::{MessageType, RoomMessageEventContent},
},
},
};
use serde::Serialize;
use crate::protocol::DaemonResponse;
/// JSON-shape for `list_rooms`: one row per joined room.
#[derive(Debug, Serialize)]
pub struct RoomInfo {
pub room_id: String,
pub canonical_alias: Option<String>,
pub name: String,
pub member_count: u64,
}
/// JSON-shape for `list_room_members`: one row per joined member.
#[derive(Debug, Serialize)]
pub struct MemberInfo {
pub user_id: String,
pub display_name: Option<String>,
}
/// JSON-shape for `read_room`: one timeline event, flattened for
/// claude consumption.
#[derive(Debug, Serialize)]
pub struct TimelineEvent {
pub event_id: String,
pub sender: String,
pub origin_server_ts: i64,
pub event_type: String,
/// Best-effort plain-text body for `m.room.message` events
/// (text / notice / emote); empty for events without a textual
/// body (state changes, reactions, etc.).
pub body: String,
}
/// Resolve a room reference (id `!abc:server` OR alias `#name:server`)
/// to a joined `Room`. Returns an `Error` response if the room isn't
/// joined / the reference is malformed.
async fn resolve_room(
client: &Client,
reference: &str,
) -> Result<matrix_sdk::Room, DaemonResponse> {
let parsed: &RoomOrAliasId = reference.try_into().map_err(|e| {
DaemonResponse::error(format!("invalid room reference {reference}: {e}"))
})?;
let room_id: OwnedRoomId = if parsed.is_room_id() {
OwnedRoomId::try_from(reference)
.map_err(|e| DaemonResponse::error(format!("invalid room_id: {e}")))?
} else {
client
.resolve_room_alias(parsed.as_str().try_into().map_err(|e| {
DaemonResponse::error(format!("invalid alias: {e}"))
})?)
.await
.map(|r| r.room_id)
.map_err(|e| DaemonResponse::error(format!("resolve_room_alias {reference}: {e}")))?
};
client
.get_room(&room_id)
.ok_or_else(|| DaemonResponse::error(format!("room {room_id} not joined")))
}
/// Best-effort plain-text body for a (non-sync) timeline event.
/// Returns "" for non-text events (state changes, reactions,
/// redactions) — claude can still see the `event_type` field to
/// disambiguate. `AnyTimelineEvent` (not `AnySync...`) because
/// `read_room` pulls events via the `/messages` endpoint which
/// returns the full-form variant.
fn extract_body(event: &matrix_sdk::ruma::events::AnyTimelineEvent) -> String {
use matrix_sdk::ruma::events::{AnyMessageLikeEvent, AnyTimelineEvent};
match event {
AnyTimelineEvent::MessageLike(AnyMessageLikeEvent::RoomMessage(ev)) => {
ev.as_original().map_or_else(String::new, |orig| {
match &orig.content.msgtype {
MessageType::Text(t) => t.body.clone(),
MessageType::Notice(n) => n.body.clone(),
MessageType::Emote(e) => format!("* {}", e.body),
_ => String::new(),
}
})
}
_ => String::new(),
}
}
pub async fn send_message(client: &Client, room_ref: &str, body: &str) -> DaemonResponse {
let room = match resolve_room(client, room_ref).await {
Ok(r) => r,
Err(e) => return e,
};
let content = RoomMessageEventContent::text_markdown(body);
match room.send(content).await {
Ok(resp) => DaemonResponse::ok(&serde_json::json!({
"event_id": resp.event_id.to_string(),
"room_id": room.room_id().to_string(),
})),
Err(e) => DaemonResponse::error(format!("send to {}: {e}", room.room_id())),
}
}
pub async fn send_dm(client: &Client, user_id: &str, body: &str) -> DaemonResponse {
let uid: OwnedUserId = match user_id.parse() {
Ok(u) => u,
Err(e) => return DaemonResponse::error(format!("invalid user_id {user_id}: {e}")),
};
// Find existing DM or create one.
let room = client
.joined_rooms()
.into_iter()
.find(|r| {
// is_direct() is async; check direct_targets() instead which
// reads from cached state.
r.direct_targets().iter().any(|t| t.as_str() == uid.as_str())
});
let room = match room {
Some(r) => r,
None => match client.create_dm(&uid).await {
Ok(r) => r,
Err(e) => return DaemonResponse::error(format!("create_dm {uid}: {e}")),
},
};
let content = RoomMessageEventContent::text_markdown(body);
match room.send(content).await {
Ok(resp) => DaemonResponse::ok(&serde_json::json!({
"event_id": resp.event_id.to_string(),
"room_id": room.room_id().to_string(),
"user_id": uid.to_string(),
})),
Err(e) => DaemonResponse::error(format!("send DM to {uid}: {e}")),
}
}
pub async fn send_reaction(
client: &Client,
room_ref: &str,
event_id: &str,
key: &str,
) -> DaemonResponse {
let room = match resolve_room(client, room_ref).await {
Ok(r) => r,
Err(e) => return e,
};
let eid: OwnedEventId = match event_id.parse() {
Ok(e) => e,
Err(e) => return DaemonResponse::error(format!("invalid event_id {event_id}: {e}")),
};
let content = ReactionEventContent::new(Annotation::new(eid.clone(), key.to_owned()));
match room.send(content).await {
Ok(resp) => DaemonResponse::ok(&serde_json::json!({
"event_id": resp.event_id.to_string(),
"target": eid.to_string(),
})),
Err(e) => DaemonResponse::error(format!("send reaction to {eid}: {e}")),
}
}
pub async fn send_reply(
client: &Client,
room_ref: &str,
event_id: &str,
body: &str,
) -> DaemonResponse {
let room = match resolve_room(client, room_ref).await {
Ok(r) => r,
Err(e) => return e,
};
let eid: OwnedEventId = match event_id.parse() {
Ok(e) => e,
Err(e) => return DaemonResponse::error(format!("invalid event_id {event_id}: {e}")),
};
let content = RoomMessageEventContent::text_markdown(body).into();
let reply = Reply {
event_id: eid.clone(),
enforce_thread: EnforceThread::MaybeThreaded,
};
let reply_content = match room.make_reply_event(content, reply).await {
Ok(c) => c,
Err(e) => return DaemonResponse::error(format!("make_reply_event: {e}")),
};
match room.send(reply_content).await {
Ok(resp) => DaemonResponse::ok(&serde_json::json!({
"event_id": resp.event_id.to_string(),
"target": eid.to_string(),
})),
Err(e) => DaemonResponse::error(format!("send reply: {e}")),
}
}
pub async fn mark_read(client: &Client, room_ref: &str, event_id: &str) -> DaemonResponse {
let room = match resolve_room(client, room_ref).await {
Ok(r) => r,
Err(e) => return e,
};
let eid: OwnedEventId = match event_id.parse() {
Ok(e) => e,
Err(e) => return DaemonResponse::error(format!("invalid event_id {event_id}: {e}")),
};
match room
.send_single_receipt(ReceiptType::Read, ReceiptThread::Unthreaded, eid.clone())
.await
{
Ok(()) => DaemonResponse::ok(&serde_json::json!({
"marked_read": eid.to_string(),
})),
Err(e) => DaemonResponse::error(format!("send_single_receipt: {e}")),
}
}
pub async fn list_rooms(client: &Client) -> DaemonResponse {
let mut rooms = Vec::new();
for room in client.joined_rooms() {
let name = match room.display_name().await {
Ok(n) => n.to_string(),
Err(_) => room.room_id().to_string(),
};
let canonical_alias = room.canonical_alias().map(|a| a.to_string());
rooms.push(RoomInfo {
room_id: room.room_id().to_string(),
canonical_alias,
name,
member_count: room.joined_members_count(),
});
}
DaemonResponse::ok(&rooms)
}
pub async fn list_room_members(client: &Client, room_ref: &str) -> DaemonResponse {
let room = match resolve_room(client, room_ref).await {
Ok(r) => r,
Err(e) => return e,
};
let members = match room.members(matrix_sdk::RoomMemberships::JOIN).await {
Ok(m) => m,
Err(e) => return DaemonResponse::error(format!("members: {e}")),
};
let list: Vec<MemberInfo> = members
.iter()
.map(|m| MemberInfo {
user_id: m.user_id().to_string(),
display_name: m.display_name().map(ToOwned::to_owned),
})
.collect();
DaemonResponse::ok(&list)
}
pub async fn read_room(client: &Client, room_ref: &str, limit: Option<usize>) -> DaemonResponse {
use matrix_sdk::ruma::api::client::message::get_message_events;
use matrix_sdk::ruma::api::Direction;
let room = match resolve_room(client, room_ref).await {
Ok(r) => r,
Err(e) => return e,
};
let limit = limit.unwrap_or(50).min(200);
let mut req = get_message_events::v3::Request::new(room.room_id().to_owned(), Direction::Backward);
req.limit = matrix_sdk::ruma::UInt::try_from(limit as u64).unwrap_or(matrix_sdk::ruma::UInt::from(50u32));
let resp = match client.send(req).await {
Ok(r) => r,
Err(e) => return DaemonResponse::error(format!("get_message_events: {e}")),
};
let events: Vec<TimelineEvent> = resp
.chunk
.iter()
.filter_map(|raw| {
let parsed = raw.deserialize().ok()?;
let event_id = parsed.event_id().to_string();
let sender = parsed.sender().to_string();
let origin_server_ts: i64 = parsed.origin_server_ts().0.into();
let event_type = parsed.event_type().to_string();
let body = extract_body(&parsed);
Some(TimelineEvent {
event_id,
sender,
origin_server_ts,
event_type,
body,
})
})
.collect();
DaemonResponse::ok(&events)
}

14
hive-matrix-mcp/src/lib.rs
View file

@ -1,14 +0,0 @@
//! `hive-matrix-mcp` library: shared types + helpers used by both the
//! `hive-matrix-daemon` binary (long-running matrix-sdk Client) and the
//! `hive-matrix-mcp` binary (stdio MCP bridge claude spawns per turn).
//!
//! The wire protocol between the two binaries lives in [`protocol`].
//! Path helpers (token file, daemon socket) live in [`paths`].
//!
//! Phase 3 of #548. Architecture rationale + tool surface mirror the
//! existing `damocles-daemon` (see issue thread for details).
pub mod client;
pub mod paths;
pub mod protocol;
pub mod wake;

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

@ -1,86 +0,0 @@
//! `hive-matrix-daemon` binary — long-running matrix-sdk Client + sync
//! loop per agent. Bridges incoming room events to hyperhive wake
//! signals and serves the unix socket the stdio MCP bridge talks to.
//!
//! Lifecycle:
//! 1. Read access token from `paths::token_file()` (fail clean if absent).
//! 2. Whoami probe → recover `user_id` + `device_id` → restore
//! matrix-sdk session (no login flow).
//! 3. Install the message-event handler that fires hyperhive wakes.
//! 4. Spawn the unix socket listener for the MCP bridge.
//! 5. Run sync forever.
//!
//! Standalone-degraded boot: missing token file → exit 0 cleanly so
//! systemd's `ConditionPathExists=` doesn't have to be perfectly
//! synced with hive-c0re's token-provisioning timing.
use anyhow::{Context, Result};
use matrix_sdk::config::SyncSettings;
mod client;
mod handlers;
mod paths;
mod protocol;
mod socket;
mod timeline;
mod wake;
#[tokio::main]
async fn main() -> Result<()> {
tracing_subscriber::fmt()
.with_env_filter(
tracing_subscriber::EnvFilter::try_from_default_env()
.unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")),
)
.with_writer(std::io::stderr)
.init();
let homeserver = paths::homeserver_url();
let token_file = paths::token_file();
let state_dir = paths::matrix_state_dir();
let mcp_socket = paths::daemon_socket();
let hyperhive_socket = paths::hyperhive_socket();
if !tokio::fs::try_exists(&token_file).await.unwrap_or(false) {
tracing::warn!(
path = %token_file.display(),
"matrix token file absent; exiting cleanly (hive-c0re will provision \
it on first agent registration, then systemd restarts us)"
);
return Ok(());
}
tracing::info!(
homeserver,
token_file = %token_file.display(),
state_dir = %state_dir.display(),
"hive-matrix-daemon starting"
);
let matrix_client = client::build_and_restore(&homeserver, &token_file, &state_dir)
.await
.context("build matrix client")?;
timeline::install_message_handler(&matrix_client, hyperhive_socket);
// Spawn the unix socket server before sync starts so the MCP
// bridge can connect as soon as the first claude turn fires. The
// socket dispatches against the same `Client` we sync on, so any
// tool call benefits from the sync state-cache.
let socket_client = matrix_client.clone();
let socket_listener = mcp_socket.clone();
tokio::spawn(async move {
if let Err(e) = socket::serve(&socket_listener, socket_client).await {
tracing::error!(error = %e, "mcp socket server exited");
}
});
// Sync forever; matrix-sdk handles reconnection internally.
let sync_settings = SyncSettings::default();
matrix_client
.sync(sync_settings)
.await
.context("matrix-sdk sync loop exited")?;
Ok(())
}

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

@ -1,65 +0,0 @@
//! Per-agent filesystem paths used by both `hive-matrix-daemon` and the
//! stdio MCP bridge.
//!
//! All paths are overridable via env vars for dev / test scenarios and
//! to let the harness point the daemon at non-default locations when
//! the operator overrides `hyperhive.matrix.*` options.
use std::path::PathBuf;
/// Default homeserver URL when `HIVE_MATRIX_URL` isn't set. Tuwunel
/// (the local hive-matrix container) listens on `localhost:8008` by
/// default; shared host netns means every agent container resolves
/// `localhost` to the same machine.
pub const DEFAULT_HOMESERVER: &str = "http://localhost:8008";
/// Default unix socket path the daemon listens on inside the agent
/// container. The stdio MCP bridge `connect()`s here on every tool call.
/// `/run/hive-matrix.sock` is a tmpfs path that disappears on container
/// restart — fine, because the daemon recreates the socket on its own
/// boot.
pub const DEFAULT_DAEMON_SOCKET: &str = "/run/hive-matrix.sock";
/// Resolve the matrix access-token file path. Override via
/// `HIVE_MATRIX_TOKEN_FILE`; default is `<HYPERHIVE_STATE_DIR>/matrix-token`,
/// the path `hive-c0re::matrix::ensure_user_for` writes to on agent
/// account provisioning.
#[must_use]
pub fn token_file() -> PathBuf {
if let Some(p) = std::env::var_os("HIVE_MATRIX_TOKEN_FILE") {
return PathBuf::from(p);
}
let state_dir = std::env::var("HYPERHIVE_STATE_DIR").unwrap_or_default();
PathBuf::from(format!("{state_dir}/matrix-token"))
}
/// Resolve the homeserver URL. Override via `HIVE_MATRIX_URL`; default
/// is the in-container `localhost:8008` tuwunel.
#[must_use]
pub fn homeserver_url() -> String {
std::env::var("HIVE_MATRIX_URL").unwrap_or_else(|_| DEFAULT_HOMESERVER.to_owned())
}
/// Resolve the daemon's unix socket path. Override via
/// `HIVE_MATRIX_SOCKET`; default is `/run/hive-matrix.sock`.
#[must_use]
pub fn daemon_socket() -> PathBuf {
std::env::var_os("HIVE_MATRIX_SOCKET").map_or_else(|| PathBuf::from(DEFAULT_DAEMON_SOCKET), PathBuf::from)
}
/// Persistent sqlite store directory for matrix-sdk's state (event
/// cache, devices, etc.). Lives under the per-agent state dir so it
/// survives container restart but gets wiped on `destroy --purge`.
#[must_use]
pub fn matrix_state_dir() -> PathBuf {
let state_dir = std::env::var("HYPERHIVE_STATE_DIR").unwrap_or_default();
PathBuf::from(format!("{state_dir}/matrix-sdk-state"))
}
/// 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)
}

108
hive-matrix-mcp/src/protocol.rs
View file

@ -1,108 +0,0 @@
//! Wire types for the daemon ↔ stdio-MCP-bridge unix socket protocol.
//!
//! Same shape as `damocles-daemon`'s `DaemonRequest`/`DaemonResponse`
//! (which this is forked from in spirit). Each request is a single
//! JSON line; each response is a single JSON line back. The stdio MCP
//! bridge holds a fresh connection per tool call — claude's tool
//! lifecycle is shorter than a persistent matrix-sdk Client wants to
//! live, so the daemon stays alive and the MCP reconnects per call.
use serde::{Deserialize, Serialize};
/// Request from the stdio MCP bridge to the daemon. The MCP bridge
/// owns the on-wire shape claude sees; this enum is the internal
/// shape the daemon dispatches over.
#[derive(Debug, Serialize, Deserialize)]
#[serde(tag = "method")]
pub enum DaemonRequest {
/// Post a plain-text or markdown message to a room. `room` accepts
/// either a matrix room id (`!abc:server`) or a canonical alias
/// (`#name:server`); the daemon resolves aliases server-side.
#[serde(rename = "send_message")]
SendMessage { room: String, body: String },
/// Open (or reuse) a DM with `user_id` and post `body`. Creates
/// the DM room if one doesn't already exist between this agent
/// and the user.
#[serde(rename = "send_dm")]
SendDm { user_id: String, body: String },
/// React to a specific event with an emoji `key`. Matrix-spec
/// `m.reaction` annotation.
#[serde(rename = "send_reaction")]
SendReaction {
room: String,
event_id: String,
key: String,
},
/// Reply to `event_id` in `room` with `body` as a threaded reply.
/// Sets the `m.in_reply_to` relation so matrix clients render the
/// thread.
#[serde(rename = "send_reply")]
SendReply {
room: String,
event_id: String,
body: String,
},
/// Mark `event_id` (in `room`) as read for this agent. Sends a
/// read receipt; bumps the room's "unread" indicator down on
/// matrix clients (and for other agents).
#[serde(rename = "mark_read")]
MarkRead { room: String, event_id: String },
/// List rooms the agent has joined. Returns each room's id +
/// canonical alias (when present) + name + member count.
#[serde(rename = "list_rooms")]
ListRooms,
/// List the members of a room. Each entry carries the matrix
/// user id + the resolved display name (when set).
#[serde(rename = "list_room_members")]
ListRoomMembers { room: String },
/// Read the last `limit` events from a room's timeline. Caller
/// gets each event's id, sender, `server_ts`, type, and body (best-
/// effort plain-text extraction from `m.text` / `m.notice` etc.).
#[serde(rename = "read_room")]
ReadRoom { room: String, limit: Option<usize> },
/// 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
/// connect error); reserved for external clients that want an
/// explicit health check without doing real work.
#[serde(rename = "ping")]
Ping,
}
/// 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.
#[derive(Debug, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum DaemonResponse {
Ok { payload: serde_json::Value },
Error { message: String },
}
impl DaemonResponse {
/// Convenience: build an Ok response from any serializable value.
/// Serialisation failure (impossible for the small handler structs
/// we use, but the API is generic) surfaces as a structured
/// `{"serialise_error": "..."}` payload so callers see WHY the
/// data is missing instead of a silent `null`.
pub fn ok<T: Serialize>(payload: &T) -> Self {
let payload = serde_json::to_value(payload)
.unwrap_or_else(|e| serde_json::json!({ "serialise_error": e.to_string() }));
Self::Ok { payload }
}
/// Convenience: build an Error response from any `Display` value.
pub fn error(msg: impl std::fmt::Display) -> Self {
Self::Error {
message: msg.to_string(),
}
}
}

85
hive-matrix-mcp/src/socket.rs
View file

@ -1,85 +0,0 @@
//! Unix socket server: the daemon listens here, the stdio MCP bridge
//! `connect()`s on every tool call. One JSON request line in, one
//! JSON response line out. Connections are short-lived (per tool call)
//! so the loop is just accept → dispatch → reply → close.
use std::path::Path;
use std::sync::Arc;
use anyhow::{Context, Result};
use matrix_sdk::Client;
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::net::{UnixListener, UnixStream};
use crate::handlers;
use crate::protocol::{DaemonRequest, DaemonResponse};
/// Start listening on `socket_path` and serve forever. Removes any
/// stale socket file first (daemon restart after a non-clean shutdown
/// would otherwise hit EADDRINUSE).
pub async fn serve(socket_path: &Path, client: Client) -> Result<()> {
let _ = tokio::fs::remove_file(socket_path).await;
if let Some(parent) = socket_path.parent() {
tokio::fs::create_dir_all(parent)
.await
.with_context(|| format!("mkdir {}", parent.display()))?;
}
let listener = UnixListener::bind(socket_path)
.with_context(|| format!("bind unix socket {}", socket_path.display()))?;
tracing::info!(path = %socket_path.display(), "mcp socket listener up");
let client = Arc::new(client);
loop {
let (stream, _) = listener
.accept()
.await
.context("accept connection on mcp socket")?;
let client = client.clone();
tokio::spawn(async move {
if let Err(e) = handle_connection(stream, &client).await {
tracing::warn!(error = %e, "mcp socket connection error");
}
});
}
}
async fn handle_connection(stream: UnixStream, client: &Client) -> Result<()> {
let (reader, mut writer) = stream.into_split();
let mut lines = BufReader::new(reader).lines();
while let Some(line) = lines.next_line().await? {
let response = match serde_json::from_str::<DaemonRequest>(&line) {
Ok(req) => dispatch(req, client).await,
Err(e) => DaemonResponse::error(format!("parse request: {e}")),
};
let mut json = serde_json::to_string(&response)?;
json.push('\n');
writer.write_all(json.as_bytes()).await?;
writer.flush().await?;
}
Ok(())
}
async fn dispatch(req: DaemonRequest, client: &Client) -> DaemonResponse {
match req {
DaemonRequest::Ping => DaemonResponse::ok(&serde_json::json!({"ok": true})),
DaemonRequest::SendMessage { room, body } => {
handlers::send_message(client, &room, &body).await
}
DaemonRequest::SendDm { user_id, body } => handlers::send_dm(client, &user_id, &body).await,
DaemonRequest::SendReaction {
room,
event_id,
key,
} => handlers::send_reaction(client, &room, &event_id, &key).await,
DaemonRequest::SendReply {
room,
event_id,
body,
} => handlers::send_reply(client, &room, &event_id, &body).await,
DaemonRequest::MarkRead { room, event_id } => {
handlers::mark_read(client, &room, &event_id).await
}
DaemonRequest::ListRooms => handlers::list_rooms(client).await,
DaemonRequest::ListRoomMembers { room } => handlers::list_room_members(client, &room).await,
DaemonRequest::ReadRoom { room, limit } => handlers::read_room(client, &room, limit).await,
}
}

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

@ -1,66 +0,0 @@
//! Matrix event handlers: incoming room messages fire a hyperhive
//! wake signal so the agent's harness drives a new claude turn.
//!
//! Per mara on #548: wake body is a SHORT TEASER, not the full message
//! (msg stays unread server-side; agent fetches via `read_room`). The
//! `wake::format_wake_body` truncates to ~100 chars.
//!
//! Self-events (events sent by this agent) are filtered out so an
//! agent posting a message doesn't wake itself.
use std::path::PathBuf;
use std::sync::Arc;
use matrix_sdk::{
Client, Room, RoomState,
ruma::events::room::message::{MessageType, OriginalSyncRoomMessageEvent},
};
use crate::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`.
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| {
let socket = socket.clone();
let own_user = own_user.clone();
async move {
if room.state() != RoomState::Joined {
return;
}
// Self-event filter so the agent's own outgoing messages
// don't wake it. matches forge_notify's self-skip pattern.
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())
}
};
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 {
tracing::debug!(room = %room.room_id(), "matrix wake delivered");
}
}
}
});
tracing::info!("matrix message handler installed");
}

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

@ -1,116 +0,0 @@
//! Wake-signal writer: notifies the hyperhive harness when an incoming
//! matrix event arrives so claude drives a new turn.
//!
//! Same wire shape as `hive-ag3nt::forge_notify`'s wake: a single JSON
//! line written to the hyperhive control socket (`/run/hive/mcp.sock`
//! by default) carrying an `AgentRequest::Wake { from, body }`.
//! The agent harness's `agent_server` parses it and treats it as a
//! `Wake` from the matrix subsystem.
//!
//! Per mara's call on #548 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).
use std::path::Path;
use anyhow::{Context, Result};
use tokio::io::AsyncWriteExt;
use tokio::net::UnixStream;
/// Max characters of `body` to embed in the wake payload. Shorter than
/// `forge_notify`'s 500-byte excerpt because the agent has a follow-up
/// `read_room` tool to fetch the full event.
pub const WAKE_BODY_TRUNCATE: usize = 100;
/// Send an `AgentRequest::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.
pub async fn send_wake(socket: &Path, body: impl AsRef<str>) -> Result<()> {
let payload = serde_json::json!({
"kind": "wake",
"from": "matrix",
"body": body.as_ref(),
});
let line = format!("{}\n", serde_json::to_string(&payload)?);
let mut stream = UnixStream::connect(socket)
.await
.with_context(|| format!("connect hyperhive socket {}", socket.display()))?;
stream
.write_all(line.as_bytes())
.await
.with_context(|| format!("write wake to {}", socket.display()))?;
stream
.shutdown()
.await
.with_context(|| format!("shutdown write to {}", socket.display()))?;
Ok(())
}
/// Format a wake-message body from a matrix event's sender + room
/// canonical alias + body text. Truncates at [`WAKE_BODY_TRUNCATE`]
/// chars with an ellipsis. Shape: `[matrix] <sender> in <room>: <text>`
/// matches the `forge_notify` `[issue …]` framing convention.
#[must_use]
pub fn format_wake_body(sender: &str, room: &str, text: &str) -> String {
let truncated = truncate_chars(text, WAKE_BODY_TRUNCATE);
format!("[matrix] {sender} in {room}: {truncated}")
}
/// 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 {
let mut end = s.len();
for (count, (i, _)) in s.char_indices().enumerate() {
if count == max {
end = i;
break;
}
}
if end == s.len() {
s.to_owned()
} else {
format!("{}…", &s[..end])
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn format_wake_body_short_passes_through() {
let body = format_wake_body("@iris:matrix.darkest.space", "#general", "hi all");
assert_eq!(body, "[matrix] @iris:matrix.darkest.space in #general: hi all");
}
#[test]
fn format_wake_body_long_truncates_with_ellipsis() {
let long = "x".repeat(200);
let body = format_wake_body("@iris:m", "#x", &long);
assert!(body.contains("xxxxxxx"));
assert!(body.ends_with("…"));
// Header + truncated body should be well under the absolute
// wake-message ceiling (forge_notify uses ~600 bytes; we're
// way under that).
assert!(body.len() < 200);
}
#[test]
fn truncate_chars_handles_multibyte() {
// `ü` is 2 bytes / 1 char. truncating to 3 chars on "üüüüüü"
// should yield "üüü…" not "üü\xc3…" (mid-codepoint).
let s = "üüüüüü";
let t = truncate_chars(s, 3);
assert_eq!(t, "üüü…");
}
#[test]
fn truncate_chars_no_op_below_limit() {
let s = "hi";
assert_eq!(truncate_chars(s, 100), "hi");
}
}

115
nix/templates/harness-base.nix
View file

@ -149,52 +149,6 @@
''; '';
}; };
options.hyperhive.matrix.enable = lib.mkOption {
type = lib.types.bool;
default = true;
description = ''
Enable per-agent matrix integration via `hive-matrix-mcp`
(#548 phase 3). When true (the default), the harness:
- runs `hive-matrix-daemon` as a systemd unit that holds a
matrix-sdk Client + sync against the homeserver at
`HIVE_MATRIX_URL` (default `http://localhost:8008` — the
in-host tuwunel from `nix/modules/hive-matrix.nix`). The
daemon auto-skips when `<state>/matrix-token` is missing,
and a `systemd.paths` watcher restarts it the moment
hive-c0re provisions the token (mirrors `matrix-avatar-sync`
shape from #571).
- exposes the matrix tool surface (send_message, send_dm,
send_reaction, send_reply, mark_read, list_rooms,
list_room_members, read_room) to claude via an auto-injected
`extraMcpServers.matrix` entry. Claude spawns the stdio
`hive-matrix-mcp` bridge per turn, which forwards each tool
call to the daemon over `/run/hive-matrix.sock`.
- wakes the agent on incoming room events via a short teaser
Wake signal (`[matrix] <sender> in <room>: <first 100c>…`)
to the hyperhive control socket; the full event stays
unread server-side until `read_room` consumes it.
Set to `false` for agents that should NOT have matrix tools at
all (e.g. agents on a host without `hyperhive.matrix.enable` on
the meta side). When token file is absent the daemon and MCP
both no-op cleanly anyway, so `false` is rarely necessary.
'';
};
options.hyperhive.matrix.url = lib.mkOption {
type = lib.types.str;
default = "http://localhost:8008";
example = "https://matrix.darkest.space";
description = ''
Matrix homeserver URL the agent's `hive-matrix-daemon` connects
to. Default points at the in-host tuwunel (shared netns).
Override per-agent when an agent should talk to an external
homeserver instead (e.g. a federation-only setup or a remote
hive's tuwunel reached via a vpn).
'';
};
options.hyperhive.frontend.dist = lib.mkOption { options.hyperhive.frontend.dist = lib.mkOption {
type = lib.types.package; type = lib.types.package;
default = pkgs.hyperhive-frontend; default = pkgs.hyperhive-frontend;
@ -529,19 +483,6 @@
} }
]; ];
# Auto-inject the matrix MCP entry when matrix is enabled (#548
# phase 3). Operator can override or disable by setting their own
# `extraMcpServers.matrix` (nix submodule merge takes the operator's
# value) or by flipping `hyperhive.matrix.enable = false`.
hyperhive.extraMcpServers = lib.mkIf config.hyperhive.matrix.enable {
matrix = lib.mkDefault {
command = "${pkgs.hyperhive}/bin/hive-matrix-mcp";
args = [ ];
env = { };
allowedTools = [ "*" ];
};
};
environment.etc."hyperhive/extra-mcp.json".text = builtins.toJSON config.hyperhive.extraMcpServers; environment.etc."hyperhive/extra-mcp.json".text = builtins.toJSON config.hyperhive.extraMcpServers;
# Operator-set per-agent icon (hyperhive.icon). When configured, the # Operator-set per-agent icon (hyperhive.icon). When configured, the
@ -804,49 +745,6 @@
''; '';
}; };
# Long-running matrix-sdk Client + sync per agent (#548 phase 3).
# Holds the unix socket the stdio `hive-matrix-mcp` bridge talks
# to, and emits hyperhive wake signals on incoming room events
# via `/run/hive/mcp.sock`. Conditional on `hyperhive.matrix.enable`
# AND token-file presence (the daemon binary itself exits 0 on
# missing token, but the path watcher below restarts it the
# moment the token lands — same first-boot-ordering pattern as
# matrix-avatar-sync.path / #571).
systemd.services.hive-matrix-daemon = lib.mkIf config.hyperhive.matrix.enable {
description = "long-running matrix-sdk Client + MCP daemon socket";
wantedBy = [ "multi-user.target" ];
after = [ "network-online.target" ];
wants = [ "network-online.target" ];
environment = {
HIVE_MATRIX_URL = config.hyperhive.matrix.url;
RUST_LOG = "info";
};
serviceConfig = {
ExecStart = "${pkgs.hyperhive}/bin/hive-matrix-daemon";
Restart = "on-failure";
RestartSec = 5;
# /run/hive-matrix.sock + the matrix-sdk-state sqlite dir
# don't need a StateDirectory= — the socket is on tmpfs (gone
# on restart, which is correct) and the sqlite dir lives in
# the bind-mounted agent state, mode-managed by the harness.
};
};
# Path-trigger sibling so hive-matrix-daemon fires the moment
# `<state>/matrix-token` appears (#548 phase 3, mirrors the
# matrix-avatar-sync.path pattern from #571). On clean boot
# hive-c0re provisions the token AFTER agent containers come up;
# without the trigger the daemon would exit 0 quietly and the
# MCP would have no backend until next restart. With the watcher
# the daemon comes alive in the same boot cycle as provisioning.
# The glob matches every agent (manager sees its own state at
# `/agents/hm1nd/state/` via the `/agents` bind).
systemd.paths.hive-matrix-daemon = lib.mkIf config.hyperhive.matrix.enable {
description = "trigger hive-matrix-daemon when matrix-token appears";
wantedBy = [ "multi-user.target" ];
pathConfig.PathExistsGlob = "/agents/*/state/matrix-token";
};
# Path-trigger sibling so matrix-avatar-sync fires the moment # Path-trigger sibling so matrix-avatar-sync fires the moment
# `<state>/matrix-token` appears (#571 closes argus's first-boot # `<state>/matrix-token` appears (#571 closes argus's first-boot
# ordering nag on #567). On a clean boot hive-c0re's matrix # ordering nag on #567). On a clean boot hive-c0re's matrix
@ -855,12 +753,19 @@
# token didn't exist yet, and avatar would only land on the next # token didn't exist yet, and avatar would only land on the next
# restart. With the path watcher the appearance of the token # restart. With the path watcher the appearance of the token
# triggers a re-fire of the service so the avatar is set in the # triggers a re-fire of the service so the avatar is set in the
# same boot cycle as provisioning completes. The glob matches # same boot cycle as provisioning completes.
# every agent (manager sees its own state at `/agents/hm1nd/state/` #
# via the `/agents` bind). # `PathExists=` fires at activation if the file is already there
# (covers re-deploys + restarts after first provisioning); the
# underlying inotify watch picks up subsequent creates (covers
# first-boot race). Manager mounts state at `/state` (legacy
# path), sub-agents at `/agents/<label>/state` (per
# lifecycle.rs's `notes_mount`); `PathExistsGlob=` matches both
# without needing the agent name at module-eval time.
systemd.paths.matrix-avatar-sync = { systemd.paths.matrix-avatar-sync = {
description = "trigger matrix-avatar-sync when matrix-token appears"; description = "trigger matrix-avatar-sync when matrix-token appears";
wantedBy = [ "multi-user.target" ]; wantedBy = [ "multi-user.target" ];
pathConfig.PathExists = "/state/matrix-token";
pathConfig.PathExistsGlob = "/agents/*/state/matrix-token"; pathConfig.PathExistsGlob = "/agents/*/state/matrix-token";
}; };