diff --git a/CLAUDE.md b/CLAUDE.md index cd464638..b10463f0 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -64,8 +64,10 @@ hand-maintained per-file tree drifts out of sync with the code. `docs/boundary.md`. - **`hive-forge/`** — `hive-forge` Forgejo CLI wrapper; one module per verb under `src/verbs/`. -- **`hive-matrix-mcp/`** — per-agent matrix-sdk daemon plus the thin - stdio MCP bridge claude spawns per turn. +- **`hive-matrix-mcp/`** — per-agent matrix-sdk daemon + (`hive-matrix-daemon`); serves its MCP tools (`send_message`, + `read_room`, …) directly over streamable-http (no stdio bridge), + same shape as `hive-bash-mcp`. - **`hive-bash-mcp/`** — per-agent bash-task runner daemon (`hive-bash-daemon`); serves its MCP tools (`run`/`status`/`kill`) directly over streamable-http (no stdio bridge), writes task files diff --git a/Cargo.lock b/Cargo.lock index 701bd782..7e4e738d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1557,6 +1557,7 @@ dependencies = [ "hive-core-agent-sock", "hive-sh4re", "hive-types", + "reqwest 0.13.1", "rmcp", "serde", "serde_json", @@ -1707,8 +1708,9 @@ name = "hive-matrix-mcp" version = "0.1.0" dependencies = [ "anyhow", + "axum", + "clap", "futures-util", - "hive-sh4re", "matrix-sdk", "mime", "mime_guess", diff --git a/docs/persistence.md b/docs/persistence.md index 05f587c1..e6f8489c 100644 --- a/docs/persistence.md +++ b/docs/persistence.md @@ -234,7 +234,7 @@ Under `/var/lib/hyperhive/agents//`: hourly and deletes terminal task trios older than 48 hours; non-terminal (still-running) tasks are never deleted by vacuum. - `hyperhive-todos.sqlite` — loose-ends-v2 todo store. In-container - MCP daemons (`hive-bash-daemon`, `hive-matrix-mcp`) and `forge_notify` + MCP daemons (`hive-bash-daemon`, `hive-matrix-daemon`) and `forge_notify` upsert keyed todos here over the harness's in-agent socket (`HIVE_AGENT_SOCKET`); the harness merges them into `get_loose_ends` output and clears a row on `mark_todo_done`. Replaced the old @@ -400,18 +400,13 @@ the body + marker check at that point. ## Matrix per-agent daemon + token-arrival trigger `hive-matrix-daemon` is a long-running matrix-sdk Client + sync -process per agent. Holds the unix socket the stdio -`hive-matrix-mcp` bridge talks to, emits hyperhive wake signals +process per agent. Serves its MCP tools directly over +streamable-http (`hyperhive.mcp.matrixHttpPort`, no stdio bridge — +same shape as `hive-bash-daemon`), emits hyperhive wake signals on incoming room events via `/run/hive/mcp.sock`. Conditional on `hyperhive.matrix.enable` (which both the daemon AND the auto-injected `extraMcpServers.matrix` entry read). -Socket path lives inside the systemd-managed runtime dir -(`RuntimeDirectory = "hive-matrix"` → `/run/hive-matrix/`, owned by -the agent user) so the daemon can bind without needing root over -`/run/` itself. Both daemon + bridge agree on the path via the -`HIVE_MATRIX_SOCKET` env var. - **First-boot ordering**: hive-c0re provisions the matrix token AFTER agent containers come up. Without the path-trigger sibling (`systemd.paths.hive-matrix-daemon`, `PathExistsGlob = diff --git a/docs/tools/matrix.md b/docs/tools/matrix.md index c81bed76..014274c7 100644 --- a/docs/tools/matrix.md +++ b/docs/tools/matrix.md @@ -3,8 +3,9 @@ ## Built-in matrix MCP (`mcp__matrix__*`) When `hyperhive.matrix.enable = true` and the host-level matrix -tuwunel is configured, the harness auto-injects `hive-matrix-mcp` as -a second stdio MCP server. Tools land as `mcp__matrix__`: +tuwunel is configured, the harness auto-injects `hive-matrix-daemon`'s +streamable-http endpoint as a second MCP server (no stdio bridge — +see Architecture below). Tools land as `mcp__matrix__`: ### Messaging @@ -68,11 +69,18 @@ room you haven't read yet. ## Architecture -The daemon (`hive-matrix-daemon`) holds the long-running matrix-sdk -`Client` + sync loop; the stdio bridge (`hive-matrix-mcp`) is spawned -per turn and forwards tool calls over `/run/hive-matrix.sock`. Both -silently exit when `/matrix-token` is absent (account not yet -provisioned). +`hive-matrix-daemon` is a single long-running process (one per agent +container, systemd service in `nix/agent-modules/matrix.nix`) — no +stdio bridge, no separate bin. It owns the matrix-sdk `Client` + sync +loop per configured account **and** serves the matrix tool surface +directly over streamable-http on `hyperhive.mcp.matrixHttpPort` +(declared in `hyperhive.extraMcpServers.matrix` as +`{ type = "http"; url = ...; }`). Same shape as `hive-bash-daemon` and +the built-in `hyperhive` surface (`hive-mcp-http`) — claude reconnects +to the stable URL every turn instead of respawning a stdio child. +Silently exits when `/matrix-token` is absent (account not yet +provisioned); the `systemd.paths.hive-matrix-daemon` watcher restarts +it the moment hive-c0re provisions the token. Incoming room events wake the agent via `AgentRequest::Wake` with `from: "matrix"`. The wake body format depends on the unread state: @@ -107,8 +115,11 @@ provisioning flow, and federation config. ## Extra MCP servers (per-agent) Each agent's NixOS config can declare additional MCP servers via -`hyperhive.extraMcpServers. = { command, args, env, -allowedTools }`. The module writes the map to +`hyperhive.extraMcpServers. = { type, command, args, env, url, +allowedTools }` — `type = "stdio"` (the default, uses `command`/`args`/ +`env`) or `type = "http"` (uses `url`, a long-lived streamable-http +endpoint — see `hive-bash-daemon` and `hive-matrix-daemon` above for +the `"http"` shape). The module writes the map to `/etc/hyperhive/extra-mcp.json`; the harness reads it at boot and merges every entry into `--mcp-config` (under `mcpServers.`) and `--allowedTools` (as `mcp____`). diff --git a/flake.nix b/flake.nix index fd8ed0c6..60014a5a 100644 --- a/flake.nix +++ b/flake.nix @@ -90,7 +90,6 @@ hive-bash-daemon hive-forge hive-matrix-daemon - hive-matrix-mcp hive-metric assets frontend diff --git a/hive-agent-mcp/Cargo.toml b/hive-agent-mcp/Cargo.toml index 376b77fd..9cf72d6d 100644 --- a/hive-agent-mcp/Cargo.toml +++ b/hive-agent-mcp/Cargo.toml @@ -19,6 +19,7 @@ hive-agent-sock.workspace = true hive-core-agent-sock.workspace = true hive-sh4re.workspace = true hive-types.workspace = true +reqwest.workspace = true rmcp.workspace = true serde.workspace = true serde_json.workspace = true diff --git a/hive-agent-mcp/src/mcp/render.rs b/hive-agent-mcp/src/mcp/render.rs index fb9821cc..6ac53a37 100644 --- a/hive-agent-mcp/src/mcp/render.rs +++ b/hive-agent-mcp/src/mcp/render.rs @@ -1,8 +1,9 @@ //! Formatting / render helpers for the MCP tool surface: ack / recv / //! loose-end / agent-meta reply shaping plus the retry annotation. //! Stateless string builders, with one exception — -//! [`matrix_unread_summary`] queries the local matrix daemon socket -//! (best-effort) so `get_loose_ends` can prepend an unread-rooms entry. +//! [`matrix_unread_summary`] queries `hive-matrix-daemon`'s local +//! `/unread-summary` status endpoint (best-effort) so `get_loose_ends` +//! can prepend an unread-rooms entry. /// Render the three identical failure arms every data-returning tool handler /// repeats: a broker `Err` → `"{tool} failed: {m}"`, an unexpected `Ok` variant @@ -272,27 +273,30 @@ pub(super) struct MatrixRoomUnread { last_sender: Option, } -/// Query the local matrix daemon for per-room unread summaries. Returns -/// `None` if the daemon socket is absent or the query fails. Best-effort: -/// agents without matrix configured are not penalised. +/// Default port `hive-matrix-daemon` serves its MCP + status endpoints +/// on (`hyperhive.mcp.matrixHttpPort`'s nix default). Overridable via +/// `HIVE_MATRIX_HTTP_PORT` for parity with the port options nix already +/// exposes; unset in practice since a single fixed port is safe (each +/// agent container is its own network namespace — see docs/network.md). +const DEFAULT_MATRIX_HTTP_PORT: u16 = 8792; + +/// Short request timeout for the local status query below — this must +/// never stall a turn waiting on a wedged same-container daemon. +const MATRIX_STATUS_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(2); + +/// Query the local matrix daemon's `/unread-summary` status endpoint +/// for per-room unread summaries. Returns `None` if the daemon isn't +/// reachable or the query fails. Best-effort: agents without matrix +/// configured are not penalised. pub(super) async fn matrix_unread_summary() -> Option> { - use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; - use tokio::net::UnixStream; - let socket = std::env::var_os("HIVE_MATRIX_SOCKET").map_or_else( - || std::path::PathBuf::from("/run/hive-matrix/socket"), - std::path::PathBuf::from, - ); - if !socket.exists() { - return None; - } - let mut stream = UnixStream::connect(&socket).await.ok()?; - stream - .write_all(b"{\"method\":\"unread_summary\"}\n") - .await + let port = std::env::var("HIVE_MATRIX_HTTP_PORT") + .unwrap_or_else(|_| DEFAULT_MATRIX_HTTP_PORT.to_string()); + let url = format!("http://127.0.0.1:{port}/unread-summary"); + let client = reqwest::Client::builder() + .timeout(MATRIX_STATUS_TIMEOUT) + .build() .ok()?; - let mut lines = BufReader::new(stream).lines(); - let line = lines.next_line().await.ok()??; - let val: serde_json::Value = serde_json::from_str(&line).ok()?; + let val: serde_json::Value = client.get(&url).send().await.ok()?.json().await.ok()?; // Response: {"kind":"ok","payload":[{label, count, last_body?, last_sender?}]} let arr = val.get("payload")?.as_array()?; serde_json::from_value(serde_json::Value::Array(arr.clone())).ok() diff --git a/hive-matrix-mcp/Cargo.toml b/hive-matrix-mcp/Cargo.toml index 5dd9d32c..e059e214 100644 --- a/hive-matrix-mcp/Cargo.toml +++ b/hive-matrix-mcp/Cargo.toml @@ -2,14 +2,16 @@ name = "hive-matrix-mcp" edition.workspace = true version.workspace = true +readme = "README.md" [lints] workspace = true [dependencies] anyhow.workspace = true +axum.workspace = true +clap.workspace = true futures-util.workspace = true -hive-sh4re.workspace = true matrix-sdk.workspace = true mime = "0.3" mime_guess = "2" @@ -23,17 +25,8 @@ 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. +# sync loop. Serves its MCP tools (send_message, list_rooms, …) +# directly over streamable-http — no stdio bridge, no separate bin. [[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/socket, -# 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" diff --git a/hive-matrix-mcp/README.md b/hive-matrix-mcp/README.md new file mode 100644 index 00000000..f42578b8 --- /dev/null +++ b/hive-matrix-mcp/README.md @@ -0,0 +1,36 @@ +# hive-matrix-mcp + +Per-agent matrix integration: a long-running daemon +(`hive-matrix-daemon`) that holds a matrix-sdk `Client` + sync loop per +configured account and serves the matrix tool surface (`send_message`, +`send_dm`, `list_rooms`, `read_room`, …) directly over streamable-http. +No stdio bridge, no per-turn respawn — claude reconnects to the same +stable URL every turn. + +## When to use it + +Look here when changing matrix tool behaviour, multi-account handling, +or the incoming-event → todo/wake path. The daemon owns the whole +lifecycle: per-account bring-up (`accounts.rs`, `client.rs`), the sync +loop that sweeps invites/unread rooms into the harness's in-agent todo +socket (`timeline.rs`, `wake.rs`), and the MCP tool router itself +(`mcp.rs`). + +## Shape + +One bin (`hive-matrix-daemon`, `src/main.rs`) built from the crate's +own lib (`src/lib.rs`): + +- **`accounts.rs`** — multi-account config + the account→`Client` + dispatch registry (`main` is always the hive-internal primary; extras + come from `HIVE_MATRIX_ACCOUNTS`). +- **`client.rs`** — session restore, stale-token recovery, avatar sync, + cross-signing bootstrap. +- **`timeline.rs`** / **`wake.rs`** — per-sync-callback invite/unread + sweeps that push todos onto `HIVE_AGENT_SOCKET`. +- **`handlers.rs`** — per-tool dispatch, returns `protocol::DaemonResponse`. +- **`mcp.rs`** — the `rmcp` tool router + `serve_http`, resolving each + call's optional `account` arg against the registry before calling + into `handlers`. +- **`paths.rs`** — per-agent path resolution (token file, matrix-sdk + state dir, homeserver URL, accounts snapshot). diff --git a/hive-matrix-mcp/src/accounts.rs b/hive-matrix-mcp/src/accounts.rs index 03ea1077..8e2ec5fd 100644 --- a/hive-matrix-mcp/src/accounts.rs +++ b/hive-matrix-mcp/src/accounts.rs @@ -274,7 +274,8 @@ impl Registry { /// primary flag. Registry membership == a session restored, so every /// entry is reported `live`. Sorted primary-first then by name for a /// stable order in the dashboard. Account-agnostic — the caller does - /// not resolve a single client (see `socket::dispatch`). + /// not resolve a single client (see `MatrixMcp::resolve` in + /// `crate::mcp`). #[must_use] pub fn list(&self) -> Vec { let mut out: Vec = self diff --git a/hive-matrix-mcp/src/handlers.rs b/hive-matrix-mcp/src/handlers.rs index 7001fcbb..7b7d6df6 100644 --- a/hive-matrix-mcp/src/handlers.rs +++ b/hive-matrix-mcp/src/handlers.rs @@ -1,13 +1,12 @@ -//! 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). +//! Per-tool dispatch — each MCP tool call in [`crate::mcp`] resolves to +//! one of these handlers. Returns a [`DaemonResponse`] which +//! `crate::mcp::render` turns into the tool-result string claude sees +//! (`Ok { payload }` → pretty JSON, `Error { message }` → a "matrix +//! error: …" prefixed string). //! //! Tool surface mirrors damocles-daemon's v0 set per the operator's call: //! `send_message`, `send_dm`, `send_reaction`, `send_reply`, `mark_read`, -//! `send_redact`, `list_rooms`, `list_room_members`, `read_room`. Plus a -//! `ping` for the MCP bridge's liveness probe. +//! `send_redact`, `list_rooms`, `list_room_members`, `read_room`. use matrix_sdk::{ Client, diff --git a/hive-matrix-mcp/src/lib.rs b/hive-matrix-mcp/src/lib.rs index 25433b0d..130c4072 100644 --- a/hive-matrix-mcp/src/lib.rs +++ b/hive-matrix-mcp/src/lib.rs @@ -1,14 +1,17 @@ -//! `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`]. +//! Shared library for the `hive-matrix-daemon` binary — the only binary +//! this crate produces. The daemon owns the per-account matrix-sdk +//! `Client` registry (built at startup, one per configured account) and +//! serves its MCP tools directly over streamable-http (see [`mcp`]) — +//! no stdio bridge, no per-turn respawn, no round-trip unix socket. //! //! Architecture rationale and tool surface mirror the existing //! `damocles-daemon` v0 set. +pub mod accounts; pub mod client; +pub mod handlers; +pub mod mcp; pub mod paths; pub mod protocol; +pub mod timeline; pub mod wake; diff --git a/hive-matrix-mcp/src/main.rs b/hive-matrix-mcp/src/main.rs index dbb4f39c..73b76995 100644 --- a/hive-matrix-mcp/src/main.rs +++ b/hive-matrix-mcp/src/main.rs @@ -1,6 +1,8 @@ //! `hive-matrix-daemon` binary — long-running matrix-sdk Client + sync //! loop per matrix account. Bridges incoming room events to hyperhive -//! wake signals and serves the unix socket the stdio MCP bridge talks to. +//! wake signals and serves its MCP tools directly over streamable-http +//! on `--http ` — no stdio bridge, no separate bin claude has to +//! respawn every turn. //! //! Lifecycle: //! 1. Read the configured account list (`accounts::configured()` — @@ -8,9 +10,9 @@ //! 2. For each account: whoami probe → recover `user_id` + `device_id` //! → restore matrix-sdk session (no login flow), install the //! message-event handler, and spawn its own sync loop. -//! 3. Serve the unix socket against an account→Client registry; each -//! MCP request routes to the account named in its `account` field -//! (the primary account when omitted). +//! 3. Serve the MCP tools against an account→Client registry; each tool +//! call routes to the account named in its `account` arg (the +//! primary account when omitted). //! //! Standalone-degraded boot: the PRIMARY account having no token file → //! exit 0 cleanly so systemd's path-watcher restarts us once hive-c0re @@ -27,19 +29,21 @@ use std::sync::Arc; use anyhow::{Context, Result}; +use clap::Parser; use matrix_sdk::{Client, config::SyncSettings}; -mod accounts; -mod client; -mod handlers; -mod paths; -mod protocol; -mod socket; -mod timeline; -mod wake; +use hive_matrix_mcp::accounts::{AccountCfg, Registry}; +use hive_matrix_mcp::client::PermanentBringUpError; +use hive_matrix_mcp::{accounts, client, mcp, paths, timeline, wake}; -use accounts::{AccountCfg, Registry}; -use client::PermanentBringUpError; +#[derive(Parser)] +#[command(name = "hive-matrix-daemon", about = "matrix-sdk client + MCP daemon")] +struct Cli { + /// Serve the MCP tools over streamable-http on this address (e.g. + /// `127.0.0.1:8792`). Bind loopback only. + #[arg(long)] + http: std::net::SocketAddr, +} /// A per-account sync loop, boxed so loops for N accounts can be driven /// concurrently on the main task. Deliberately NOT `Send`: matrix-sdk's @@ -71,8 +75,8 @@ async fn main() -> Result<()> { .with_writer(std::io::stderr) .init(); + let cli = Cli::parse(); let cfgs = accounts::configured().context("read matrix account config")?; - let mcp_socket = paths::daemon_socket(); let multi = cfgs.len() > 1; let primary = cfgs[0].name.clone(); let mut registry = Registry::new(primary); @@ -131,9 +135,9 @@ async fn main() -> Result<()> { tracing::warn!(error = %format!("{e:#}"), "failed to write matrix-accounts snapshot"); } - // Serve the socket against the registry. Spawned before driving the - // sync loops so the MCP bridge can connect as soon as the first - // claude turn fires. + // Serve the MCP tools against the registry. Spawned before driving + // the sync loops so claude can reach the stable http URL as soon as + // the first turn fires. let registry = Arc::new(registry); // Heartbeat: periodically rewrite the accounts snapshot so its mtime @@ -155,10 +159,11 @@ async fn main() -> Result<()> { } }); - let socket_listener = mcp_socket.clone(); + let http_addr = cli.http; + let mcp_registry = Arc::clone(®istry); tokio::spawn(async move { - if let Err(e) = socket::serve(&socket_listener, registry).await { - tracing::error!(error = %e, "mcp socket server exited"); + if let Err(e) = mcp::serve_http(http_addr, mcp_registry).await { + tracing::error!(error = %e, "mcp http server exited"); } }); @@ -298,7 +303,7 @@ async fn bring_up_account( // 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 _ = wake::send_todo_clear(None, true).await; let sync_loop: SyncLoop = Box::pin(async move { sync_client .sync_with_callback(SyncSettings::default(), move |_response| { diff --git a/hive-matrix-mcp/src/bin/mcp.rs b/hive-matrix-mcp/src/mcp.rs similarity index 66% rename from hive-matrix-mcp/src/bin/mcp.rs rename to hive-matrix-mcp/src/mcp.rs index 79e6a99a..29c292ad 100644 --- a/hive-matrix-mcp/src/bin/mcp.rs +++ b/hive-matrix-mcp/src/mcp.rs @@ -1,78 +1,43 @@ -//! `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. +//! MCP tool surface for `hive-matrix-daemon`, served directly over +//! streamable-http — no stdio bridge, no per-turn respawn, no +//! round-trip unix socket. The daemon already owns the account +//! registry in-process (built at startup from the restored matrix-sdk +//! `Client`s), so each tool call resolves `account` against +//! [`crate::accounts::Registry`] and calls straight into +//! [`crate::handlers`]. Mirrors `hive-bash-mcp::mcp`'s shape (and, one +//! level up, `hive-agent-mcp::mcp::serve_http`): a persistent daemon, +//! stable URL claude reconnects to every turn instead of respawning a +//! stdio child. //! //! Multi-account: every tool carries an optional `account` arg naming //! which matrix account to act as (a `name` from //! `hyperhive.matrixAccounts`); omitting it selects the agent's primary -//! account. The bridge wraps each operation in a [`DaemonRequest`] -//! envelope carrying that account; the daemon routes to the matching -//! client. +//! account (or errors, listing the choices, when more than one account +//! is configured — see [`crate::accounts::Registry::resolve`]). + +use std::sync::Arc; -use anyhow::{Context, Result}; use rmcp::{ - ServerHandler, ServiceExt, + ServerHandler, 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::{DaemonOp, DaemonRequest, DaemonResponse, InviteAction}; - -/// 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 { - let socket = paths::daemon_socket(); - let stream = UnixStream::connect(&socket).await.with_context(|| { - format!( - "matrix daemon unreachable at {} — it may be starting up or restarting \ - (the daemon rebinds its socket a few seconds after a restart); retry shortly", - 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") -} - -/// Wrap an op in a [`DaemonRequest`] envelope for `account` and send it. -async fn call(account: Option, op: DaemonOp) -> Result { - round_trip(DaemonRequest { account, op }).await -} +use crate::accounts::Registry; +use crate::handlers; +use crate::protocol::{DaemonResponse, InviteAction}; /// 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) -> String { +fn render(resp: DaemonResponse) -> String { match resp { - Ok(DaemonResponse::Ok { payload }) => { + 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:#}"), + DaemonResponse::Error { message } => format!("matrix error: {message}"), } } @@ -282,25 +247,25 @@ struct InviteUserArgs { account: Option, } -struct MatrixBridge { - #[allow( - dead_code, - reason = "populated by the #[tool_router] macro; the generated \ - ServerHandler wiring consumes it, the field is never read directly" - )] - tool_router: rmcp::handler::server::router::tool::ToolRouter, +#[derive(Clone)] +struct MatrixMcp { + registry: Arc, } -impl MatrixBridge { - fn new() -> Self { - Self { - tool_router: Self::tool_router(), - } +impl MatrixMcp { + /// Resolve `account` against the registry, rendering the "unknown + /// account" / "ambiguous, pick one" error the same way a handler + /// error would render (so a bad `account` arg and a bad room/event + /// arg look the same to claude). + fn resolve(&self, account: Option<&str>) -> Result<&matrix_sdk::Client, String> { + self.registry + .resolve(account) + .map(std::convert::AsRef::as_ref) } } #[tool_router] -impl MatrixBridge { +impl MatrixMcp { #[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). \ @@ -309,16 +274,11 @@ impl MatrixBridge { you don't talk over messages you haven't seen." )] async fn send_message(&self, Parameters(args): Parameters) -> String { - render( - call( - args.account, - DaemonOp::SendMessage { - room: args.room, - body: args.body, - }, - ) - .await, - ) + let client = match self.resolve(args.account.as_deref()) { + Ok(c) => c, + Err(e) => return format!("matrix error: {e}"), + }; + render(handlers::send_message(client, &args.room, &args.body).await) } #[tool(description = "Open (or reuse) a direct message room with `user_id` \ @@ -326,16 +286,11 @@ impl MatrixBridge { and has unread messages, the send is rejected with a hint — read_room \ then mark_read the latest event first.")] async fn send_dm(&self, Parameters(args): Parameters) -> String { - render( - call( - args.account, - DaemonOp::SendDm { - user_id: args.user_id, - body: args.body, - }, - ) - .await, - ) + let client = match self.resolve(args.account.as_deref()) { + Ok(c) => c, + Err(e) => return format!("matrix error: {e}"), + }; + render(handlers::send_dm(client, &args.user_id, &args.body).await) } #[tool( @@ -346,17 +301,11 @@ impl MatrixBridge { room has unread messages — read_room then mark_read first." )] async fn send_file(&self, Parameters(args): Parameters) -> String { - render( - call( - args.account, - DaemonOp::SendFile { - room: args.room, - path: args.path, - caption: args.caption, - }, - ) - .await, - ) + let client = match self.resolve(args.account.as_deref()) { + Ok(c) => c, + Err(e) => return format!("matrix error: {e}"), + }; + render(handlers::send_file(client, &args.room, &args.path, args.caption.as_deref()).await) } #[tool(description = "Resolve (find-or-create) the DM room with `user_id` \ @@ -364,15 +313,11 @@ impl MatrixBridge { returned room id with the room-based tools (`send_file`, `send_message`, \ …) to deliver into the DM — there is no per-tool DM variant.")] async fn open_dm(&self, Parameters(args): Parameters) -> String { - render( - call( - args.account, - DaemonOp::OpenDm { - user_id: args.user_id, - }, - ) - .await, - ) + let client = match self.resolve(args.account.as_deref()) { + Ok(c) => c, + Err(e) => return format!("matrix error: {e}"), + }; + render(handlers::open_dm(client, &args.user_id).await) } #[tool( @@ -381,17 +326,11 @@ impl MatrixBridge { standard clients." )] async fn send_reaction(&self, Parameters(args): Parameters) -> String { - render( - call( - args.account, - DaemonOp::SendReaction { - room: args.room, - event_id: args.event_id, - key: args.key, - }, - ) - .await, - ) + let client = match self.resolve(args.account.as_deref()) { + Ok(c) => c, + Err(e) => return format!("matrix error: {e}"), + }; + render(handlers::send_reaction(client, &args.room, &args.event_id, &args.key).await) } #[tool(description = "Reply to a specific matrix event in a room, threaded \ @@ -399,33 +338,22 @@ impl MatrixBridge { if the room still has unread messages — read_room then mark_read the \ latest event first.")] async fn send_reply(&self, Parameters(args): Parameters) -> String { - render( - call( - args.account, - DaemonOp::SendReply { - room: args.room, - event_id: args.event_id, - body: args.body, - }, - ) - .await, - ) + let client = match self.resolve(args.account.as_deref()) { + Ok(c) => c, + Err(e) => return format!("matrix error: {e}"), + }; + render(handlers::send_reply(client, &args.room, &args.event_id, &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) -> String { - render( - call( - args.account, - DaemonOp::MarkRead { - room: args.room, - event_id: args.event_id, - }, - ) - .await, - ) + let client = match self.resolve(args.account.as_deref()) { + Ok(c) => c, + Err(e) => return format!("matrix error: {e}"), + }; + render(handlers::mark_read(client, &args.room, &args.event_id).await) } #[tool(description = "Redact (delete) a specific matrix event in a room — \ @@ -433,16 +361,12 @@ impl MatrixBridge { `reason`. Works on your own events; redacting others' needs moderator \ power level. Irreversible.")] async fn send_redact(&self, Parameters(args): Parameters) -> String { + let client = match self.resolve(args.account.as_deref()) { + Ok(c) => c, + Err(e) => return format!("matrix error: {e}"), + }; render( - call( - args.account, - DaemonOp::SendRedact { - room: args.room, - event_id: args.event_id, - reason: args.reason, - }, - ) - .await, + handlers::send_redact(client, &args.room, &args.event_id, args.reason.as_deref()).await, ) } @@ -451,7 +375,11 @@ impl MatrixBridge { id, canonical alias (when set), display name, and joined-member count." )] async fn list_rooms(&self, Parameters(args): Parameters) -> String { - render(call(args.account, DaemonOp::ListRooms).await) + let client = match self.resolve(args.account.as_deref()) { + Ok(c) => c, + Err(e) => return format!("matrix error: {e}"), + }; + render(handlers::list_rooms(client).await) } #[tool( @@ -460,7 +388,11 @@ impl MatrixBridge { Use `resolve_invite` to accept or reject an invite." )] async fn list_invites(&self, Parameters(args): Parameters) -> String { - render(call(args.account, DaemonOp::ListInvites).await) + let client = match self.resolve(args.account.as_deref()) { + Ok(c) => c, + Err(e) => return format!("matrix error: {e}"), + }; + render(handlers::list_invites(client)) } #[tool( @@ -471,7 +403,11 @@ impl MatrixBridge { `list_rooms`." )] async fn join_room(&self, Parameters(args): Parameters) -> String { - render(call(args.account, DaemonOp::JoinRoom { room: args.room }).await) + let client = match self.resolve(args.account.as_deref()) { + Ok(c) => c, + Err(e) => return format!("matrix error: {e}"), + }; + render(handlers::join_room(client, &args.room).await) } #[tool( @@ -481,6 +417,10 @@ impl MatrixBridge { invites with `list_invites`." )] async fn resolve_invite(&self, Parameters(args): Parameters) -> String { + let client = match self.resolve(args.account.as_deref()) { + Ok(c) => c, + Err(e) => return format!("matrix error: {e}"), + }; let action = match args.action.trim().to_ascii_lowercase().as_str() { "accept" => InviteAction::Accept, "reject" => InviteAction::Reject, @@ -490,16 +430,7 @@ impl MatrixBridge { ); } }; - render( - call( - args.account, - DaemonOp::ResolveInvite { - room: args.room, - action, - }, - ) - .await, - ) + render(handlers::resolve_invite(client, &args.room, action).await) } #[tool( @@ -509,16 +440,11 @@ impl MatrixBridge { invite. The invitee then sees a pending invite they accept with `join_room`." )] async fn invite_user(&self, Parameters(args): Parameters) -> String { - render( - call( - args.account, - DaemonOp::InviteUser { - room: args.room, - user_id: args.user_id, - }, - ) - .await, - ) + let client = match self.resolve(args.account.as_deref()) { + Ok(c) => c, + Err(e) => return format!("matrix error: {e}"), + }; + render(handlers::invite_user(client, &args.room, &args.user_id).await) } #[tool( @@ -526,7 +452,11 @@ impl MatrixBridge { Each row carries the user id and resolved display name." )] async fn list_room_members(&self, Parameters(args): Parameters) -> String { - render(call(args.account, DaemonOp::ListRoomMembers { room: args.room }).await) + let client = match self.resolve(args.account.as_deref()) { + Ok(c) => c, + Err(e) => return format!("matrix error: {e}"), + }; + render(handlers::list_room_members(client, &args.room).await) } #[tool(description = "Read events from a matrix room (default 50, max 200), \ @@ -539,18 +469,11 @@ impl MatrixBridge { outside the current window. `from` and `until` are mutually \ exclusive.")] async fn read_room(&self, Parameters(args): Parameters) -> String { - render( - call( - args.account, - DaemonOp::ReadRoom { - room: args.room, - limit: args.limit, - from: args.from, - until: args.until, - }, - ) - .await, - ) + let client = match self.resolve(args.account.as_deref()) { + Ok(c) => c, + Err(e) => return format!("matrix error: {e}"), + }; + render(handlers::read_room(client, &args.room, args.limit, args.from, args.until).await) } #[tool( @@ -561,14 +484,16 @@ impl MatrixBridge { counterpart of send_file." )] async fn download_file(&self, Parameters(args): Parameters) -> String { + let client = match self.resolve(args.account.as_deref()) { + Ok(c) => c, + Err(e) => return format!("matrix error: {e}"), + }; render( - call( - args.account, - DaemonOp::DownloadFile { - room: args.room, - event_id: args.event_id, - dest_path: args.dest_path, - }, + handlers::download_file( + client, + &args.room, + &args.event_id, + args.dest_path.as_deref(), ) .await, ) @@ -598,52 +523,73 @@ impl MatrixBridge { all rejected if the room has unread messages — always call `read_room` \ then `mark_read` on the latest event before sending to a room you \ haven't read yet.")] -impl ServerHandler for MatrixBridge {} +impl ServerHandler for MatrixMcp {} -#[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")), +/// Plain-JSON status endpoint for the primary account's unread rooms — +/// NOT part of the claude-facing MCP tool surface. `hive-agent-mcp`'s +/// `get_loose_ends` hits this directly (same container, loopback only) +/// to prepend a matrix-unread entry, mirroring the pre-http unix-socket +/// `unread_summary` side channel the daemon used to serve. Best-effort: +/// only resolves when exactly one account is configured (same rule as +/// an MCP tool call omitting `account`) — a multi-account agent's extra +/// accounts aren't reachable from here, same restriction the caller +/// already documents for cross-agent queries. +async fn unread_summary_handler( + axum::extract::State(registry): axum::extract::State>, +) -> axum::Json { + let payload = match registry.resolve(None) { + Ok(client) => handlers::unread_summary(client.as_ref()).await, + Err(message) => crate::protocol::DaemonResponse::error(message), + }; + axum::Json( + serde_json::to_value(payload) + .unwrap_or_else(|e| serde_json::json!({ "kind": "error", "message": e.to_string() })), + ) +} + +/// Run the MCP server over HTTP (rmcp streamable-http transport) on +/// `addr`, dispatching against `registry`. Also serves a small +/// non-MCP `/unread-summary` status endpoint (see +/// [`unread_summary_handler`]). +/// +/// Sole transport — there is no stdio mode. Long-lived so claude +/// reconnects to the stable URL each turn instead of respawning a +/// stdio child; since the daemon already owns the account registry +/// in-process, tool calls need no round-trip to anywhere. +/// +/// Binds loopback only in practice; the default `allowed_hosts` +/// (`localhost`/`127.0.0.1`/`::1`) rejects Host headers from anywhere +/// else for `/mcp`; `/unread-summary` is plain axum with no such +/// guard, but the listener itself is loopback-only so this is moot. +/// +/// # Errors +/// +/// Returns an error if the listener cannot bind `addr` or the HTTP +/// server exits with a fatal error. +pub async fn serve_http(addr: std::net::SocketAddr, registry: Arc) -> anyhow::Result<()> { + use rmcp::transport::streamable_http_server::{ + StreamableHttpServerConfig, StreamableHttpService, session::local::LocalSessionManager, + }; + let session_manager = std::sync::Arc::new(LocalSessionManager::default()); + let mcp_registry = registry.clone(); + let service = StreamableHttpService::new( + move || { + Ok(MatrixMcp { + registry: mcp_registry.clone(), + }) + }, + session_manager, + StreamableHttpServerConfig::default(), + ); + let app = axum::Router::new() + .nest_service("/mcp", service) + .route( + "/unread-summary", + axum::routing::get(unread_summary_handler), ) - .with_writer(std::io::stderr) - .init(); - - // Standalone-degraded boot: matrix isn't provisioned for this agent - // (no token file) → exit 0 cleanly so claude doesn't register a - // matrix MCP server it can never use. - // - // Gate on the TOKEN, not the daemon socket. The daemon binds its - // socket only AFTER restoring its matrix session (~10s on a cold - // boot), so an exists-check on the socket here raced the daemon's - // startup: during that window the socket was absent, the bridge - // exited, and claude lost the matrix tools for the WHOLE session - // (the bridge isn't respawned mid-turn). The token, by contrast, is - // written by hive-c0re at provisioning time and is present well - // before the daemon finishes booting — so it cleanly distinguishes - // "matrix not set up for this agent" (token absent → exit) from - // "daemon still coming up" (token present → keep serving). When the - // token exists we serve regardless of socket state: tool calls - // `connect()` per-call and simply error until the daemon is up, but - // the tools stay registered for the session. - let token_file = paths::token_file(); - if !tokio::fs::try_exists(&token_file).await.unwrap_or(false) { - tracing::warn!( - path = %token_file.display(), - "matrix not provisioned (no token file); 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")?; + .with_state(registry); + let listener = tokio::net::TcpListener::bind(addr).await?; + tracing::info!(%addr, "serving hive-matrix MCP over streamable-http at /mcp"); + axum::serve(listener, app).await?; Ok(()) } diff --git a/hive-matrix-mcp/src/paths.rs b/hive-matrix-mcp/src/paths.rs index dc9dcf17..cac9bbf1 100644 --- a/hive-matrix-mcp/src/paths.rs +++ b/hive-matrix-mcp/src/paths.rs @@ -1,5 +1,4 @@ -//! Per-agent filesystem paths used by both `hive-matrix-daemon` and the -//! stdio MCP bridge. +//! Per-agent filesystem paths used by `hive-matrix-daemon`. //! //! All paths are overridable via env vars for dev / test scenarios and //! to let the harness point the daemon at non-default locations when @@ -13,14 +12,6 @@ use std::path::PathBuf; /// `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. -/// Lives under systemd's `RuntimeDirectory=hive-matrix` (a tmpfs path -/// that disappears on container restart — fine, because the daemon -/// recreates the socket on its own boot) so the agent unix user -/// can bind a socket inside it without root in `/run`. -pub const DEFAULT_DAEMON_SOCKET: &str = "/run/hive-matrix/socket"; - /// Resolve the matrix access-token file path. Override via /// `HIVE_MATRIX_TOKEN_FILE`; default is `/matrix-token`, /// the path `hive-c0re::matrix::ensure_user_for` writes to on agent @@ -41,14 +32,6 @@ 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/socket`. -#[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`. diff --git a/hive-matrix-mcp/src/protocol.rs b/hive-matrix-mcp/src/protocol.rs index 2642c69b..26443051 100644 --- a/hive-matrix-mcp/src/protocol.rs +++ b/hive-matrix-mcp/src/protocol.rs @@ -1,208 +1,18 @@ -//! Wire types for the daemon ↔ stdio-MCP-bridge unix socket protocol. +//! Shared response/DTO shapes for the matrix tool surface. //! -//! 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. +//! `hive-matrix-daemon` serves its MCP tools directly over +//! streamable-http (see [`crate::mcp`]) — there is no separate bridge +//! process and no wire protocol between two binaries any more, so this +//! module carries only the handler-facing result type +//! ([`DaemonResponse`]) and small DTOs ([`InviteAction`], +//! [`RoomUnread`]) shared between [`crate::handlers`] and its callers +//! (the MCP tool router, the wake-signal formatter). use serde::{Deserialize, Serialize}; -/// Request envelope from the stdio MCP bridge to the daemon: which -/// matrix `account` to act as, plus the operation itself. The daemon -/// holds an account→Client registry (one client per declared matrix -/// account) and routes `op` to the resolved client. -/// -/// `account` is nested rather than flattened onto [`DaemonOp`] so we -/// dodge the serde "internally-tagged enum + `#[serde(flatten)]`" -/// edge cases; the bridge and daemon ship together so the wire shape -/// is private. Wire: -/// `{"account":"ccc","op":{"method":"send_message","room":…,"body":…}}`. -#[derive(Debug, Serialize, Deserialize)] -pub struct DaemonRequest { - /// Logical account name to act as (matches a `name` in - /// `hyperhive.matrixAccounts`). `None` selects the primary account - /// (the first declared one / the single legacy account), so - /// single-account callers omit it entirely. - #[serde(default, skip_serializing_if = "Option::is_none")] - pub account: Option, - /// The matrix operation to perform on the resolved account. - pub op: DaemonOp, -} - -/// The matrix operation a [`DaemonRequest`] carries. 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 DaemonOp { - /// 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 }, - - /// Upload a local file and post it as an attachment to `room` - /// (id or alias). `caption`, when set, is sent as a follow-up - /// text message in the same room. - #[serde(rename = "send_file")] - SendFile { - room: String, - path: String, - caption: Option, - }, - - /// Resolve (find-or-create) the DM room with `user_id` and return - /// its room id, without sending anything. Lets a caller obtain the - /// DM room id and then use the room-based tools (`send_file`, - /// `send_message`, …) against it — so there is no per-tool `_dm` - /// variant. - #[serde(rename = "open_dm")] - OpenDm { user_id: 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 }, - - /// Redact `event_id` in `room` — ask the homeserver to strip the - /// event's content (matrix-spec `m.room.redaction`), optionally with - /// a human-readable `reason`. The agent must have a high enough power - /// level (its own events, or moderator rights for others'); the - /// server rejects otherwise. - #[serde(rename = "send_redact")] - SendRedact { - room: String, - event_id: String, - reason: Option, - }, - - /// 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 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.). With neither cursor, returns the last - /// `limit` events (newest-first). `from` / `until` anchor at an event id - /// (mutually exclusive): `until` reads the anchor + `limit-1` events before - /// it (into the past); `from` reads the anchor + `limit-1` events after it. - #[serde(rename = "read_room")] - ReadRoom { - room: String, - limit: Option, - #[serde(default)] - from: Option, - #[serde(default)] - until: Option, - }, - - /// Download the media attachment carried by `event_id` in `room` - /// and write it to a local file (`dest_path`, or a temp file named - /// after the attachment when omitted), returning the path. The - /// read-side counterpart of `send_file`. - #[serde(rename = "download_file")] - DownloadFile { - room: String, - event_id: String, - dest_path: Option, - }, - - /// List rooms this agent has been invited to but not yet joined. - /// Returns each room's id, canonical alias (when present), and - /// display name. - #[serde(rename = "list_invites")] - ListInvites, - - /// Join a room by id (`!abc:server`) or alias (`#name:server`). - /// Accepts a pending invite if one exists; also joins public rooms - /// the agent hasn't been explicitly invited to. After joining the - /// room will appear in `list_rooms`. - #[serde(rename = "join_room")] - JoinRoom { room: String }, - - /// Resolve a pending invite to `room` (id or alias) by either - /// accepting it (join) or rejecting it (decline + leave). For rooms - /// you were *invited* to; `join_room` is the path for joining a - /// public room you weren't invited to. - #[serde(rename = "resolve_invite")] - ResolveInvite { room: String, action: InviteAction }, - - /// Invite `user_id` (`@user:server`) to `room` (id or alias). The - /// calling agent must already be a member with a high enough power - /// level to invite. Idempotent-ish: inviting an already-joined or - /// already-invited user surfaces the matrix error from the server. - #[serde(rename = "invite_user")] - InviteUser { room: String, user_id: String }, - - /// Return the count of rooms with unread notifications. Used by - /// the harness `get_loose_ends` to surface unread matrix activity - /// without exposing message content. - #[serde(rename = "unread_count")] - UnreadCount, - - /// Return per-room unread summaries. For rooms with exactly one - /// unread notification, attempts to include the sender + truncated - /// body; rooms with multiple unreads carry only the count. Used by - /// `get_loose_ends` and the wake-signal formatter. - #[serde(rename = "unread_summary")] - UnreadSummary, - - /// List the matrix accounts the daemon currently has a live, - /// restored session for. Account-agnostic (does not resolve a single - /// client — handled before client resolution in `socket::dispatch`): - /// returns each restored account's name, homeserver, user id, primary - /// flag, and a `live` flag. Backs the dashboard's per-account status - /// (BE-4) — turns BE-1's token-present list into true online/offline - /// + backfills the homeserver BE-1 leaves null. - #[serde(rename = "list_accounts")] - ListAccounts, - - /// 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, -} - -/// Whether to accept or reject a pending invite in -/// [`DaemonRequest::ResolveInvite`]. Serialises as `"accept"` / -/// `"reject"` on the wire. +/// Whether to accept or reject a pending invite (`resolve_invite` +/// tool). Serialises as `"accept"` / `"reject"` on the wire (kept +/// `Serialize`/`Deserialize` for the JSON DTOs handlers build). #[derive(Debug, Clone, Copy, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] pub enum InviteAction { @@ -212,7 +22,7 @@ pub enum InviteAction { Reject, } -/// One entry in the [`DaemonRequest::UnreadSummary`] response payload. +/// One entry in the `unread_summary` response payload. #[derive(Debug, Serialize, Deserialize)] pub struct RoomUnread { /// Canonical alias (`#name:server`) or room id (`!id:server`). @@ -229,9 +39,13 @@ pub struct RoomUnread { pub last_sender: Option, } -/// Response shape: `ok` carries the payload (any JSON; the MCP bridge -/// passes it back to claude as the tool result), `error` carries a -/// human-readable error string. +/// Result shape every [`crate::handlers`] function returns: `Ok` +/// carries the payload (any JSON; the MCP tool router renders it as +/// the tool result string), `Error` carries a human-readable error +/// string. Kept as a distinct type (rather than each handler +/// returning a bare `String`) so the tool router can uniformly render +/// success vs error without every handler duplicating that +/// formatting. #[derive(Debug, Serialize, Deserialize)] #[serde(tag = "kind", rename_all = "snake_case")] pub enum DaemonResponse { @@ -258,117 +72,3 @@ impl DaemonResponse { } } } - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn envelope_round_trips_with_account() { - let req = DaemonRequest { - account: Some("ccc".to_owned()), - op: DaemonOp::SendMessage { - room: "!r:s".to_owned(), - body: "hi".to_owned(), - }, - }; - let line = serde_json::to_string(&req).unwrap(); - // account + nested tagged op present on the wire. - assert!(line.contains("\"account\":\"ccc\""), "wire: {line}"); - assert!(line.contains("\"method\":\"send_message\""), "wire: {line}"); - let back: DaemonRequest = serde_json::from_str(&line).unwrap(); - assert_eq!(back.account.as_deref(), Some("ccc")); - matches!(back.op, DaemonOp::SendMessage { .. }); - } - - #[test] - fn envelope_defaults_account_to_none_and_omits_it() { - // Single-account callers send no `account`; it must default to - // None and not appear on the wire (skip_serializing_if). - let req = DaemonRequest { - account: None, - op: DaemonOp::ListRooms, - }; - let line = serde_json::to_string(&req).unwrap(); - assert!( - !line.contains("account"), - "wire should omit account: {line}" - ); - let back: DaemonRequest = serde_json::from_str(&line).unwrap(); - assert!(back.account.is_none()); - matches!(back.op, DaemonOp::ListRooms); - } - - #[test] - fn send_redact_round_trips_with_reason() { - let req = DaemonRequest { - account: None, - op: DaemonOp::SendRedact { - room: "!r:s".to_owned(), - event_id: "$e".to_owned(), - reason: Some("spam".to_owned()), - }, - }; - let line = serde_json::to_string(&req).unwrap(); - assert!(line.contains("\"method\":\"send_redact\""), "wire: {line}"); - let back: DaemonRequest = serde_json::from_str(&line).unwrap(); - match back.op { - DaemonOp::SendRedact { - room, - event_id, - reason, - } => { - assert_eq!(room, "!r:s"); - assert_eq!(event_id, "$e"); - assert_eq!(reason.as_deref(), Some("spam")); - } - other => panic!("wrong variant: {other:?}"), - } - } - - #[test] - fn send_redact_round_trips_without_reason() { - let req = DaemonRequest { - account: None, - op: DaemonOp::SendRedact { - room: "!r:s".to_owned(), - event_id: "$e".to_owned(), - reason: None, - }, - }; - let line = serde_json::to_string(&req).unwrap(); - let back: DaemonRequest = serde_json::from_str(&line).unwrap(); - match back.op { - DaemonOp::SendRedact { reason, .. } => assert_eq!(reason, None), - other => panic!("wrong variant: {other:?}"), - } - } - - #[test] - fn unit_variant_op_parses_inside_envelope() { - // A bare op with no fields still parses when wrapped. - let parsed: DaemonRequest = - serde_json::from_str(r#"{"op":{"method":"unread_count"}}"#).unwrap(); - assert!(parsed.account.is_none()); - matches!(parsed.op, DaemonOp::UnreadCount); - } - - #[test] - fn list_accounts_parses_as_account_agnostic_unit_variant() { - // Registry-wide op: no fields, and callers omit `account`. - let parsed: DaemonRequest = - serde_json::from_str(r#"{"op":{"method":"list_accounts"}}"#).unwrap(); - assert!(parsed.account.is_none()); - matches!(parsed.op, DaemonOp::ListAccounts); - // And it serialises back to the same tagged shape. - let line = serde_json::to_string(&DaemonRequest { - account: None, - op: DaemonOp::ListAccounts, - }) - .unwrap(); - assert!( - line.contains("\"method\":\"list_accounts\""), - "wire: {line}" - ); - } -} diff --git a/hive-matrix-mcp/src/socket.rs b/hive-matrix-mcp/src/socket.rs deleted file mode 100644 index 294944f4..00000000 --- a/hive-matrix-mcp/src/socket.rs +++ /dev/null @@ -1,143 +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::accounts::Registry; -use crate::handlers; -use crate::protocol::{DaemonOp, 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). `registry` resolves each request's -/// `account` to the matrix client that serves it. -pub async fn serve(socket_path: &Path, registry: Arc) -> 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"); - loop { - let (stream, _) = listener - .accept() - .await - .context("accept connection on mcp socket")?; - let registry = registry.clone(); - tokio::spawn(async move { - if let Err(e) = handle_connection(stream, ®istry).await { - tracing::warn!(error = %e, "mcp socket connection error"); - } - }); - } -} - -async fn handle_connection(stream: UnixStream, registry: &Registry) -> 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::(&line) { - Ok(req) => dispatch(req, registry).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, registry: &Registry) -> DaemonResponse { - // Ping is account-agnostic — answer without resolving a client so a - // health probe works even before any account restores. - if matches!(req.op, DaemonOp::Ping) { - return DaemonResponse::ok(&serde_json::json!({"ok": true})); - } - // ListAccounts is registry-wide, not per-account — answer before - // resolving a single client (the `account` field is meaningless for - // it, and it must work even if the primary failed to restore). - if matches!(req.op, DaemonOp::ListAccounts) { - return DaemonResponse::ok(®istry.list()); - } - let client = match registry.resolve(req.account.as_deref()) { - Ok(c) => c, - Err(msg) => return DaemonResponse::error(msg), - }; - dispatch_op(req.op, client).await -} - -async fn dispatch_op(op: DaemonOp, client: &Client) -> DaemonResponse { - match op { - // Unreachable in practice: `dispatch` handles Ping before resolving - // a client (so a health probe works before any account restores). - // Kept for an exhaustive match. - DaemonOp::Ping => DaemonResponse::ok(&serde_json::json!({"ok": true})), - // Unreachable in practice: `dispatch` handles ListAccounts before - // resolving a client (it is registry-wide, not per-account). Kept - // for an exhaustive match — there is no client-scoped meaning. - DaemonOp::ListAccounts => DaemonResponse::error( - "list_accounts is registry-wide; handled before client resolution", - ), - DaemonOp::SendMessage { room, body } => handlers::send_message(client, &room, &body).await, - DaemonOp::SendDm { user_id, body } => handlers::send_dm(client, &user_id, &body).await, - DaemonOp::SendFile { - room, - path, - caption, - } => handlers::send_file(client, &room, &path, caption.as_deref()).await, - DaemonOp::OpenDm { user_id } => handlers::open_dm(client, &user_id).await, - DaemonOp::SendReaction { - room, - event_id, - key, - } => handlers::send_reaction(client, &room, &event_id, &key).await, - DaemonOp::SendReply { - room, - event_id, - body, - } => handlers::send_reply(client, &room, &event_id, &body).await, - DaemonOp::MarkRead { room, event_id } => { - handlers::mark_read(client, &room, &event_id).await - } - DaemonOp::SendRedact { - room, - event_id, - reason, - } => handlers::send_redact(client, &room, &event_id, reason.as_deref()).await, - DaemonOp::ListRooms => handlers::list_rooms(client).await, - DaemonOp::ListInvites => handlers::list_invites(client), - DaemonOp::JoinRoom { room } => handlers::join_room(client, &room).await, - DaemonOp::ResolveInvite { room, action } => { - handlers::resolve_invite(client, &room, action).await - } - DaemonOp::InviteUser { room, user_id } => { - handlers::invite_user(client, &room, &user_id).await - } - DaemonOp::ListRoomMembers { room } => handlers::list_room_members(client, &room).await, - DaemonOp::ReadRoom { - room, - limit, - from, - until, - } => handlers::read_room(client, &room, limit, from, until).await, - DaemonOp::DownloadFile { - room, - event_id, - dest_path, - } => handlers::download_file(client, &room, &event_id, dest_path.as_deref()).await, - DaemonOp::UnreadCount => handlers::unread_count(client), - DaemonOp::UnreadSummary => handlers::unread_summary(client).await, - } -} diff --git a/hive-matrix-mcp/src/timeline.rs b/hive-matrix-mcp/src/timeline.rs index e4a9da67..e1447f8a 100644 --- a/hive-matrix-mcp/src/timeline.rs +++ b/hive-matrix-mcp/src/timeline.rs @@ -15,6 +15,7 @@ //! never wakes on its own message. use std::collections::HashSet; +use std::hash::BuildHasher; use matrix_sdk::{Client, ruma::OwnedRoomId}; use tokio::sync::Mutex; @@ -39,9 +40,9 @@ use crate::{handlers, wake}; /// out by `collect_unread_with_ids` (after a rebuild a stale read receipt can /// otherwise leave a self-authored message counted as unread and self-wake /// the agent). -pub async fn sweep_unread( +pub async fn sweep_unread( client: &Client, - notified: &Mutex>, + notified: &Mutex>, account_tag: Option<&str>, ) { let unread = handlers::collect_unread_with_ids(client).await; @@ -52,7 +53,11 @@ pub async fn sweep_unread( // outside it, drop from `notified` on success (retry next tick on fail). let stale: Vec = { let active = notified.lock().await; - active.difference(&unread_ids).cloned().collect() + active + .iter() + .filter(|id| !unread_ids.contains(*id)) + .cloned() + .collect() }; for id in stale { if wake::send_todo_clear(Some(id.as_str()), false) @@ -95,9 +100,9 @@ pub async fn sweep_unread( /// than on every sync tick; it is pruned to the current invite set each /// pass so a withdrawn-then-reissued invite wakes again. The agent /// decides whether to accept or reject by calling `resolve_invite`. -pub async fn sweep_invites( +pub async fn sweep_invites( client: &Client, - notified: &Mutex>, + notified: &Mutex>, account_tag: Option<&str>, ) { let current = client.invited_rooms(); @@ -110,7 +115,10 @@ pub async fn sweep_invites( // tick on failure). let stale: Vec = { let seen = notified.lock().await; - seen.difference(¤t_ids).cloned().collect() + seen.iter() + .filter(|id| !current_ids.contains(*id)) + .cloned() + .collect() }; for id in stale { if wake::send_todo_clear(Some(&invite_key(&id)), false) diff --git a/nix/agent-modules/matrix.nix b/nix/agent-modules/matrix.nix index 1d13ce15..efcf39fd 100644 --- a/nix/agent-modules/matrix.nix +++ b/nix/agent-modules/matrix.nix @@ -1,7 +1,8 @@ # Per-agent matrix integration: the `hyperhive.matrix.*` + # `hyperhive.matrixAccounts` options, the long-running -# hive-matrix-daemon, its token-arrival path trigger, and the -# auto-injected stdio MCP bridge entry. +# hive-matrix-daemon (serves its MCP tools directly over +# streamable-http), its token-arrival path trigger, and the +# auto-injected extraMcpServers entry. { pkgs, lib, @@ -31,7 +32,7 @@ in type = lib.types.bool; default = true; description = '' - Enable per-agent matrix integration via `hive-matrix-mcp`. + Enable per-agent matrix integration via `hive-matrix-daemon`. When true (the default), the harness: - runs `hive-matrix-daemon` as a systemd unit that holds a @@ -45,9 +46,10 @@ in - 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/socket`. + `extraMcpServers.matrix` entry pointed at the daemon's own + streamable-http listener (`hyperhive.mcp.matrixHttpPort`) — no + stdio bridge, no per-turn respawn, same shape as the built-in + hyperhive surface and `hive-bash-daemon`. - wakes the agent on incoming room events via a short teaser Wake signal (`[matrix] in : …`) to the hyperhive control socket; the full event stays @@ -146,6 +148,24 @@ in ''; }; + options.hyperhive.mcp.matrixHttpPort = lib.mkOption { + type = lib.types.port; + default = 8792; + example = 8793; + description = '' + Loopback port `hive-matrix-daemon` serves its MCP tools + (`send_message`, `list_rooms`, `read_room`, …) on. Same shape as + `hyperhive.mcp.bashHttpPort`: HTTP is the *sole* transport (no + stdio bridge — the daemon that owns the matrix-sdk `Client` + registry serves the MCP tools directly in-process), + `Restart = "always"` keeps the listener self-healing, and + loopback-only binding means no auth token is needed (same + `allowed_hosts` reasoning as `hyperhive.mcp.httpPort`). Safe as a + single fixed default across all agents (private per-container + network namespace — see docs/network.md). + ''; + }; + config = { assertions = [ # Extra matrix accounts only make sense alongside the hive-internal @@ -190,35 +210,30 @@ in } ]; - # Auto-inject the matrix stdio MCP bridge alongside the bash entry - # from ./mcp.nix. `lib.mkDefault` so the operator's own agent.nix - # can override the entry. + # Auto-inject the matrix MCP entry alongside the bash entry from + # ./mcp.nix. `lib.mkDefault` so the operator's own agent.nix can + # override it. Points at the daemon's own persistent + # streamable-http listener — no stdio bridge, no per-turn spawn. hyperhive.extraMcpServers = lib.mkIf config.hyperhive.matrix.enable { matrix = lib.mkDefault { - command = "${config.hyperhive.packages.hive-matrix-mcp}/bin/hive-matrix-mcp"; - args = [ ]; - # Same socket path the hive-matrix-daemon service binds - # via its `RuntimeDirectory = "hive-matrix"`. Keeps the - # bridge + daemon in sync without baking the path into - # the Rust default — the env override wins for both. - env.HIVE_MATRIX_SOCKET = "/run/hive-matrix/socket"; + type = "http"; + url = "http://127.0.0.1:${toString config.hyperhive.mcp.matrixHttpPort}/mcp"; allowedTools = [ "*" ]; }; }; - # Long-running matrix-sdk client + sync per agent. Holds the unix - # socket the stdio `hive-matrix-mcp` bridge connects to + emits - # hyperhive wake signals on incoming room events via - # `/run/hive/mcp.sock`. See + # Long-running matrix-sdk client + sync per agent. Serves the MCP + # tools directly over streamable-http + emits hyperhive wake + # signals on incoming room events via `/run/hive/mcp.sock`. See # `docs/persistence.md::Matrix per-agent daemon + token-arrival - # trigger` for the socket-path / first-boot-ordering rationale. + # trigger` for the first-boot-ordering rationale. systemd.services.hive-matrix-daemon = lib.mkIf config.hyperhive.matrix.enable { - description = "long-running matrix-sdk Client + MCP daemon socket"; + description = "long-running matrix-sdk Client + MCP daemon"; wantedBy = [ "multi-user.target" ]; + before = [ "hive-agent.service" ]; after = [ "network-online.target" ]; wants = [ "network-online.target" ]; environment = { - HIVE_MATRIX_SOCKET = "/run/hive-matrix/socket"; # In-agent todo socket the harness serves (loose-ends v2): the # matrix sweep pushes unread-room + pending-invite todos here # instead of firing wakes at hive-c0re's mcp.sock. @@ -263,25 +278,21 @@ in HIVE_ICON_PNG = "${iconPng}"; }; serviceConfig = { - ExecStart = "${config.hyperhive.packages.hive-matrix-daemon}/bin/hive-matrix-daemon"; + ExecStart = "${config.hyperhive.packages.hive-matrix-daemon}/bin/hive-matrix-daemon --http 127.0.0.1:${toString config.hyperhive.mcp.matrixHttpPort}"; SyslogIdentifier = "hive-matrix-daemon"; + # `on-failure`, not `always`: the daemon deliberately exits 0 + # (a clean, non-failure exit) when no token is provisioned yet + # (see the module doc above) — the `systemd.paths` watcher + # below re-fires it the moment hive-c0re provisions one, + # instead of `always` busy-looping every `RestartSec` until + # then. Once a token exists this is no different from + # `hive-bash-daemon`'s reasoning (a down window loses the MCP + # tools with no stdio fallback) — a genuine crash is a + # non-zero exit, which `on-failure` already restarts. Restart = "on-failure"; RestartSec = 5; User = userName; Group = userName; - RuntimeDirectory = "hive-matrix"; - # Keep /run/hive-matrix across restarts. With the default - # `RuntimeDirectoryPreserve=no`, a `switch-to-configuration` - # restart races the outgoing instance's stop-time cleanup - # (which deletes the dir) against the incoming instance's - # start (which creates it + binds the socket inside it). The - # cleanup can win and delete the dir out from under the fresh - # daemon, which then fails to mkdir under root-owned /run and - # exits — looping on Restart=on-failure until the next boot. - # `yes` stops systemd removing it on stop; it still creates it - # on first start, and it lives on tmpfs so it's gone at - # container reboot regardless. See hive-bash-daemon (./mcp.nix). - RuntimeDirectoryPreserve = "yes"; }; }; diff --git a/nix/agent-modules/packages.nix b/nix/agent-modules/packages.nix index a7f71943..b2558226 100644 --- a/nix/agent-modules/packages.nix +++ b/nix/agent-modules/packages.nix @@ -12,7 +12,7 @@ hyperhive package outputs consumed by the harness modules: the per-binary daemon/CLI packages (`hive-agent`, `hive-agent-mcp`, `hive-agent-wake`, `hive-bash-daemon`, - `hive-forge`, `hive-matrix-daemon`, `hive-matrix-mcp`, + `hive-forge`, `hive-matrix-daemon`, `hive-metric`, `hive-screen-mcp`) plus the `assets`, `frontend` and `reference-docs` trees. Wired by the flake's agent-base/ruth nixosModules to `hyperhive.packages..*`; override an diff --git a/nix/packages/default.nix b/nix/packages/default.nix index 3640b258..299f7c0b 100644 --- a/nix/packages/default.nix +++ b/nix/packages/default.nix @@ -28,8 +28,7 @@ let hive-agent-mcp = "hyperhive agent-surface MCP server"; hive-agent-wake = "hyperhive external wake CLI — push a message into an agent's own inbox"; hive-bash-daemon = "hyperhive per-agent bash-task runner daemon (serves its MCP tools directly over streamable-http)"; - hive-matrix-daemon = "hyperhive per-agent matrix-sdk daemon"; - hive-matrix-mcp = "hyperhive matrix MCP bridge"; + hive-matrix-daemon = "hyperhive per-agent matrix-sdk daemon (serves its MCP tools directly over streamable-http)"; hive-metric = "hyperhive agent-emitted custom metrics CLI"; hive-screen-mcp = "hyperhive screen MCP bridge (screenshot + input for GUI agents)"; hive-forge = "hyperhive Forgejo CLI";