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

Compare commits

base: vinzenz:ade7be46c2968db9f651ca47d272a5b1da2fe178
Branches Tags
vinzenz:main
..
compare: vinzenz:795dd882bbac9a618337644ae2f77cbad4b57fdc
Branches Tags
vinzenz:main
27 changed files with 686 additions and 787 deletions
Download patch file Download diff file Expand all files Collapse all files

10
Cargo.lock generated
View file

@ -1523,7 +1523,6 @@ dependencies = [
"clap",
"forgejo-api",
"futures-util",
"hive-agent-sock",
"hive-claude",
"hive-core-agent-sock",
"hive-sh4re",
@ -1553,7 +1552,6 @@ dependencies = [
"anyhow",
"axum",
"clap",
"hive-agent-sock",
"hive-core-agent-sock",
"hive-sh4re",
"rmcp",
@ -1564,14 +1562,6 @@ dependencies = [
"tracing-subscriber",
]
[[package]]
name = "hive-agent-sock"
version = "0.1.0"
dependencies = [
"hive-sh4re",
"serde",
]
[[package]]
name = "hive-agent-wake"
version = "0.1.0"

2
Cargo.toml
View file

@ -3,7 +3,6 @@ resolver = "3"
members = [
"hive-agent",
"hive-agent-mcp",
"hive-agent-sock",
"hive-core-agent-sock",
"hive-agent-wake",
"hive-bash-mcp",
@ -52,7 +51,6 @@ clap = { version = "4", features = ["derive"] }
clap_complete = "4"
indicatif = "0.18"
hive-sh4re = { path = "hive-sh4re" }
hive-agent-sock = { path = "hive-agent-sock" }
hive-core-agent-sock = { path = "hive-core-agent-sock" }
hive-claude = { path = "hive-claude" }
hive-host-sock = { path = "hive-host-sock" }

1
hive-agent-mcp/Cargo.toml
View file

@ -14,7 +14,6 @@ workspace = true
anyhow.workspace = true
axum.workspace = true
clap.workspace = true
hive-agent-sock.workspace = true
hive-core-agent-sock.workspace = true
hive-sh4re.workspace = true
rmcp.workspace = true

10
hive-agent-mcp/src/mcp/mod.rs
View file

@ -35,8 +35,8 @@ pub use args::{
pub use render::{annotate_retries, format_ack, format_agent_meta, format_recv};
use render::{
format_matrix_summary, local_todos, loose_end_kind_label, matrix_unread_summary,
parse_loose_end_kind, render_loose_ends, reply_err,
format_matrix_summary, loose_end_kind_label, matrix_unread_summary, parse_loose_end_kind,
render_loose_ends, reply_err,
};
/// Write (or remove) the status file in the agent's own `state/` directory.
@ -333,12 +333,6 @@ impl AgentServer {
);
}
}
// Merge the harness's local todos (loose-ends v2) for self-queries.
// The harness owns the todo store in-container; another agent's
// todos aren't reachable from here (same as matrix above).
if is_self_query && let Some(todos) = local_todos().await {
loose_ends.extend(todos);
}
let mut out = annotate_retries(render_loose_ends(&loose_ends), retries);
// Append loose-end items published by external MCP daemons
// (e.g. active bash tasks from hive-bash-mcp). Generic — no

24
hive-agent-mcp/src/mcp/render.rs
View file

@ -297,30 +297,6 @@ pub(super) async fn matrix_unread_summary() -> Option<Vec<MatrixRoomUnread>> {
serde_json::from_value(serde_json::Value::Array(arr.clone())).ok()
}
/// Query the harness's in-agent socket for this agent's local todos
/// (loose-ends v2). Returns `None` when `HIVE_AGENT_SOCKET` is unset /
/// absent or the query fails — best-effort, like [`matrix_unread_summary`],
/// so an agent without the socket is not penalised.
pub(super) async fn local_todos() -> Option<Vec<hive_sh4re::LooseEnd>> {
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::net::UnixStream;
let socket = std::env::var_os("HIVE_AGENT_SOCKET").map(std::path::PathBuf::from)?;
if !socket.exists() {
return None;
}
let mut stream = UnixStream::connect(&socket).await.ok()?;
let mut req =
serde_json::to_string(&hive_agent_sock::Request::ListTodos { subsystem: None }).ok()?;
req.push('\n');
stream.write_all(req.as_bytes()).await.ok()?;
let mut lines = BufReader::new(stream).lines();
let line = lines.next_line().await.ok()??;
match serde_json::from_str::<hive_agent_sock::Response>(&line).ok()? {
hive_agent_sock::Response::LooseEnds { loose_ends } => Some(loose_ends),
_ => None,
}
}
/// Format a `Vec<MatrixRoomUnread>` into a per-room summary string.
/// Single room / single message collapses to one line; multi-room
/// expands to a bulleted list. Returns an empty string for empty input.

11
hive-agent-sock/Cargo.toml
View file

@ -1,11 +0,0 @@
[package]
name = "hive-agent-sock"
edition.workspace = true
version.workspace = true
[lints]
workspace = true
[dependencies]
serde.workspace = true
hive-sh4re.workspace = true

76
hive-agent-sock/src/lib.rs
View file

@ -1,76 +0,0 @@
//! Wire types for the *in-agent* socket, served by the hive-agent harness
//! to the in-container producers (matrix / bash MCP daemons) and
//! `forge_notify`. Currently carries the loose-ends-v2 *todo* op family;
//! more in-agent request families may be added over time (the socket is
//! deliberately named for the agent, not the todos).
//!
//! Distinct from `hive-core-agent-sock`, the *host*-served core↔agent
//! protocol on `/run/hive/mcp.sock`: this socket never leaves the
//! container. The harness owns the todo store locally and signals its own
//! turn loop directly, so hive-c0re is not in the todo path — no broker
//! round-trip, no long-poll, no marker files.
use serde::{Deserialize, Serialize};
use hive_sh4re::LooseEnd;
/// In-container path of the harness-served in-agent socket. The harness
/// binds it on boot; the in-container producers dial it for todo ops.
/// (Placeholder default — the harness + producers resolve the real path
/// from config; kept here so a producer with no override has a sane one.)
pub const DEFAULT_AGENT_SOCKET: &str = "/run/hive/agent.sock";
/// A request on the in-agent socket. Serialised with a `cmd` tag so the
/// in-container producers can emit a plain JSON line without linking a
/// typed client (matrix/bash build the JSON by hand).
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "cmd", rename_all = "snake_case")]
pub enum Request {
/// Upsert a todo from an in-container subsystem (matrix / bash /
/// forge). `subsystem` is the producer marker; `key` the optional
/// subsystem-specific dedup key (a matrix room id, a bash task id).
/// A new-or-changed row signals the turn loop; an identical keyed
/// re-push is a silent no-op. Keyless todos always insert as one-offs.
UpsertTodo {
subsystem: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
key: Option<String>,
summary: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
source: Option<String>,
},
/// Clear producer-resolved todo(s). `key = Some(k)` clears the one
/// keyed row; `key = None` clears the subsystem's keyless rows; `all
/// = true` wipes the producer's whole set (cancel-and-recreate on
/// daemon restart).
ClearTodo {
subsystem: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
key: Option<String>,
#[serde(default)]
all: bool,
},
/// List todos, optionally filtered to one `subsystem` (a producer
/// reconciling its own set). `None` = all.
ListTodos {
#[serde(default, skip_serializing_if = "Option::is_none")]
subsystem: Option<String>,
},
/// The agent marks one of its own todos done, by id.
MarkTodoDone { id: i64 },
}
/// A response on the in-agent socket. Serialised with a `kind` tag,
/// mirroring the core↔agent protocol's response shape.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum Response {
/// Op succeeded, no payload.
Ok,
/// Op succeeded and touched `count` rows (clear / mark-done).
Acked { count: u64 },
/// `ListTodos` result.
LooseEnds { loose_ends: Vec<LooseEnd> },
/// Op failed; `message` is operator-facing.
Err { message: String },
}

1
hive-agent/Cargo.toml
View file

@ -19,7 +19,6 @@ time.workspace = true
futures-util = "0.3"
clap.workspace = true
hive-claude.workspace = true
hive-agent-sock.workspace = true
hive-core-agent-sock.workspace = true
hive-sh4re.workspace = true
rmcp.workspace = true

58
hive-agent/src/main.rs
View file

@ -23,8 +23,6 @@ mod prompt;
mod serve_common;
mod stats;
mod stream_enrich;
mod todo_server;
mod todos;
mod turn;
mod turn_stats;
mod vacuum;
@ -171,22 +169,6 @@ fn synthetic_continue() -> hive_sh4re::DeliveredMessage {
}
}
/// Synthesize the message that drives a turn when an in-container producer
/// upserted a new/changed *todo* over the in-agent socket (loose-ends v2).
/// The harness owns the todo store locally and signals the serve loop
/// directly — so this wake never touches the broker (no long-poll, no
/// marker file). `id = 0` is the same non-broker sentinel as
/// [`synthetic_continue`].
fn synthetic_todo_message() -> hive_sh4re::DeliveredMessage {
hive_sh4re::DeliveredMessage {
from: "todo".into(),
body: "you have todos — call get_loose_ends to see them".into(),
id: 0,
redelivered: false,
in_reply_to: None,
}
}
/// Synthetic message that drives the single stop-checkpoint turn when c0re
/// signals a graceful stop. The agent gets one final turn to flush durable
/// `/state` before the container is stopped; new inbound is already fenced.
@ -226,10 +208,6 @@ enum RecvOutcome {
/// one stop-checkpoint turn (flush durable `/state`), reports
/// `GracefulStopComplete`, and exits so the container can be stopped.
GracefulStop,
/// An in-container producer upserted a new/changed todo over the
/// in-agent socket; the serve loop drives a `synthetic_todo_message`
/// turn. Not a broker message — the harness signalled itself directly.
LocalTodo,
}
/// Wire surface abstraction. `AgentSurface` is the only impl — the trait
@ -445,26 +423,6 @@ async fn serve_main<S: Surface>(socket: &Path, poll_ms: u64) -> Result<()> {
tracing::error!(error = %e, "web_ui::serve exited with error");
}
});
// In-agent todo socket (loose-ends v2): the harness owns the todo store
// locally and serves the in-container producers on `HIVE_AGENT_SOCKET`.
// A new/changed upsert fires `todo_wake` so the serve loop drives a turn
// directly — no broker round-trip, no marker files. Best-effort: if the
// store can't open, the socket just isn't served.
let todo_wake = Arc::new(tokio::sync::Notify::new());
match todos::Todos::open(&paths::todos_db()) {
Ok(store) => {
let store = Arc::new(store);
let wake = todo_wake.clone();
tokio::spawn(async move {
if let Err(e) = todo_server::run(store, wake).await {
tracing::error!(error = %e, "in-agent todo socket exited with error");
}
});
}
Err(e) => {
tracing::error!(error = ?e, "open todos db failed — in-agent todo socket disabled");
}
}
if matches!(initial, LoginState::NeedsLogin) {
login::wait_for_login(&claude_dir, login_state.clone(), &bus, poll_ms).await;
} else {
@ -481,7 +439,6 @@ async fn serve_main<S: Surface>(socket: &Path, poll_ms: u64) -> Result<()> {
bus,
stats,
&files,
todo_wake,
)
.await
}
@ -503,7 +460,6 @@ async fn serve_loop<S: Surface>(
bus: Bus,
stats: Option<TurnStats>,
files: &turn::TurnFiles,
todo_wake: Arc<tokio::sync::Notify>,
) -> Result<()> {
tracing::info!(socket = %socket.display(), "harness serve");
S::requeue_inflight(socket).await;
@ -518,20 +474,8 @@ async fn serve_loop<S: Surface>(
loop {
let next = match self_continue.take() {
Some(msg) => msg,
None => match {
// Idle wait: race the broker long-poll against a local
// todo signal so an in-container producer's upsert drives a
// turn without any broker round-trip. `biased` polls the
// broker recv first, so a genuinely-ready inbox message is
// never dropped in favour of the todo wake.
tokio::select! {
biased;
o = S::recv_next(socket) => o,
() = todo_wake.notified() => RecvOutcome::LocalTodo,
}
} {
None => match S::recv_next(socket).await {
RecvOutcome::Message(first) => first,
RecvOutcome::LocalTodo => synthetic_todo_message(),
RecvOutcome::Empty => {
// Idle: no message this poll. Service a queued operator
// `/compact` here so it runs even when no turn is driving

8
hive-agent/src/paths.rs
View file

@ -40,14 +40,6 @@ pub fn harness_dir() -> PathBuf {
hive_sh4re::paths::harness_dir()
}
/// Harness-local todo store (loose-ends v2). A dedicated sqlite db under
/// the harness dir — the todos are mutable per-agent state the harness
/// owns, kept out of the append-only `hyperhive-events.sqlite` sink.
#[must_use]
pub fn todos_db() -> PathBuf {
harness_dir().join("hyperhive-todos.sqlite")
}
/// Per-turn config dir for the regenerated claude-{mcp-config,settings,
/// system-prompt} files the harness drops before each turn. Set by
/// systemd via `RuntimeDirectory = "hive-config"`: a per-service runtime

168
hive-agent/src/todo_server.rs
View file

@ -1,168 +0,0 @@
//! In-agent socket server (loose-ends v2). Binds the harness-owned
//! `HIVE_AGENT_SOCKET` and serves the `hive-agent-sock` protocol to the
//! in-container producers (matrix / bash daemons, forge-notify). Todo ops
//! hit the harness-local [`Todos`] store; a new-or-changed upsert fires an
//! in-process [`Notify`] so the serve loop drives a turn — no hive-c0re
//! round-trip, no broker long-poll, no marker files.
//!
//! One request/response line per connection, matching the producers'
//! existing best-effort JSON-line clients (they just change which socket
//! they dial, not the payload).
use std::path::{Path, PathBuf};
use std::sync::Arc;
use anyhow::{Context, Result};
use hive_agent_sock::{Request, Response};
use hive_sh4re::LooseEnd;
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::net::{UnixListener, UnixStream};
use tokio::sync::Notify;
use crate::todos::{Todo, Todos};
/// Resolve the in-agent socket path from `HIVE_AGENT_SOCKET`. `None` when
/// unset/empty (e.g. a standalone dev run) — the server then stays off.
fn socket_path() -> Option<PathBuf> {
std::env::var_os("HIVE_AGENT_SOCKET")
.filter(|v| !v.is_empty())
.map(PathBuf::from)
}
/// Run the in-agent socket server: bind + accept loop, one request/response
/// line per connection. A no-op (returns `Ok`) when `HIVE_AGENT_SOCKET` is
/// unset, so a standalone harness without producers just skips it.
///
/// # Errors
///
/// Returns an error if the socket path is set but can't be bound.
pub async fn run(store: Arc<Todos>, wake: Arc<Notify>) -> Result<()> {
let Some(path) = socket_path() else {
tracing::info!("HIVE_AGENT_SOCKET unset — in-agent todo socket disabled");
return Ok(());
};
let listener = bind(&path)?;
tracing::info!(socket = %path.display(), "in-agent todo socket listening");
loop {
match listener.accept().await {
Ok((stream, _)) => {
let store = store.clone();
let wake = wake.clone();
tokio::spawn(async move {
if let Err(e) = handle_conn(stream, &store, &wake).await {
tracing::warn!(error = ?e, "in-agent todo connection failed");
}
});
}
Err(e) => tracing::warn!(error = ?e, "in-agent todo accept failed"),
}
}
}
/// Bind a `UnixListener` at `path`, creating the parent dir and unlinking a
/// stale socket left by a prior boot (which would otherwise block `bind`
/// with `EADDRINUSE`).
fn bind(path: &Path) -> Result<UnixListener> {
if let Some(parent) = path.parent() {
std::fs::create_dir_all(parent)
.with_context(|| format!("create in-agent socket dir {}", parent.display()))?;
}
if path.exists() {
let _ = std::fs::remove_file(path);
}
UnixListener::bind(path).with_context(|| format!("bind in-agent socket {}", path.display()))
}
/// Handle one connection: read a single JSON request line, apply it to the
/// store, write the JSON response line back.
async fn handle_conn(stream: UnixStream, store: &Todos, wake: &Notify) -> Result<()> {
let (read, mut write) = stream.into_split();
let mut reader = BufReader::new(read);
let mut line = String::new();
if reader.read_line(&mut line).await? == 0 {
return Ok(());
}
let resp = match serde_json::from_str::<Request>(line.trim()) {
Ok(req) => dispatch(req, store, wake),
Err(e) => Response::Err {
message: format!("bad request: {e}"),
},
};
let mut out = serde_json::to_string(&resp)?;
out.push('\n');
write.write_all(out.as_bytes()).await?;
write.flush().await?;
Ok(())
}
/// Apply one request to the store, firing `wake` on a new/changed upsert so
/// the serve loop runs a turn.
fn dispatch(req: Request, store: &Todos, wake: &Notify) -> Response {
match req {
Request::UpsertTodo {
subsystem,
key,
summary,
source,
} => match store.upsert(&subsystem, key.as_deref(), &summary, source.as_deref()) {
Ok((_, changed)) => {
if changed {
wake.notify_one();
}
Response::Ok
}
Err(e) => err(&e),
},
Request::ClearTodo {
subsystem,
key,
all,
} => {
let result = if all {
store.clear_subsystem(&subsystem)
} else {
store.clear(&subsystem, key.as_deref())
};
match result {
Ok(count) => Response::Acked {
count: u64::try_from(count).unwrap_or(0),
},
Err(e) => err(&e),
}
}
Request::ListTodos { subsystem } => match store.list(subsystem.as_deref()) {
Ok(todos) => Response::LooseEnds {
loose_ends: todos.into_iter().map(to_loose_end).collect(),
},
Err(e) => err(&e),
},
Request::MarkTodoDone { id } => match store.mark_done(id) {
Ok(count) => Response::Acked {
count: u64::try_from(count).unwrap_or(0),
},
Err(e) => err(&e),
},
}
}
/// Format a store error into an `Err` response.
fn err(e: &anyhow::Error) -> Response {
Response::Err {
message: format!("{e:#}"),
}
}
/// Map a stored [`Todo`] to a [`LooseEnd::Todo`], deriving `age_seconds`
/// from `updated_at` (saturating so a backwards clock step reads 0).
fn to_loose_end(t: Todo) -> LooseEnd {
let now = hive_sh4re::wire_time::now_unix();
let age = u64::try_from(now.saturating_sub(t.updated_at)).unwrap_or(0);
LooseEnd::Todo {
id: t.id,
subsystem: t.subsystem,
subsystem_key: t.subsystem_key,
summary: t.summary,
source: t.source,
age_seconds: age,
}
}

293
hive-agent/src/todos.rs
View file

@ -1,293 +0,0 @@
//! Harness-local todo store — the persistent, DB-backed half of the
//! "todos" (loose-ends v2) system, owned by the in-container harness.
//!
//! In-container subsystems (matrix, forge-notify, bash) push *todos* to
//! the harness over the in-agent socket instead of firing wakes directly.
//! The harness owns this store locally (one sqlite db under the harness
//! dir) and signals its own turn loop on a new/changed row — hive-c0re is
//! not involved (no broker round-trip, no marker files).
//!
//! Because the store lives inside a single agent's container, todos are
//! **not** agent-scoped here (unlike the old c0re store): every row
//! belongs to this agent. A todo is tagged with a `subsystem` marker plus
//! an optional `subsystem_key` (a matrix room id, a bash task id, …), and
//! `(subsystem, subsystem_key)` is the upsert/dedup key — re-pushing the
//! same item is idempotent, and a producer can list / clear / rebuild only
//! its own set (e.g. matrix wipes + recreates its todos on daemon restart).
//!
//! Removal has two paths (mara's call): the producing subsystem `clear`s a
//! todo it has resolved (keyed by subsystem + key), or the agent itself
//! `mark_done`s one by id. Both delete the row.
use std::path::Path;
use std::sync::Mutex;
use anyhow::{Context, Result};
use hive_sh4re::wire_time::now_unix;
use rusqlite::{Connection, params};
const SCHEMA: &str = r"
CREATE TABLE IF NOT EXISTS todos (
id INTEGER PRIMARY KEY AUTOINCREMENT,
subsystem TEXT NOT NULL,
subsystem_key TEXT,
summary TEXT NOT NULL,
source TEXT,
updated_at INTEGER NOT NULL
);
-- (subsystem, subsystem_key) is the upsert/dedup key. A NULL key never
-- conflicts (SQLite treats NULLs as distinct), so keyless todos always
-- insert as one-offs; keyed todos update in place.
CREATE UNIQUE INDEX IF NOT EXISTS idx_todos_dedup
ON todos (subsystem, subsystem_key);
";
/// One dynamic, subsystem-pushed todo. Timestamps are unix seconds; the
/// consumer derives `age_seconds` from `updated_at`.
#[derive(Debug, Clone)]
pub struct Todo {
pub id: i64,
/// Producing subsystem marker (`"matrix"`, `"forge"`, `"bash"`, …).
pub subsystem: String,
/// Optional subsystem-specific dedup key (matrix room id, bash task
/// id, …). `None` = a keyless one-off todo.
pub subsystem_key: Option<String>,
/// Human-readable one-line summary shown to the agent.
pub summary: String,
/// Optional free-text provenance (e.g. the room name / task label).
pub source: Option<String>,
pub updated_at: i64,
}
/// The harness-local todo store. Cheap to share behind an `Arc`; the inner
/// connection is guarded by a `Mutex` (todo ops are short sqlite writes).
pub struct Todos {
conn: Mutex<Connection>,
}
impl Todos {
/// Open (creating if needed) the todo store at `path`.
///
/// # Errors
///
/// Propagates sqlite open / schema-apply failures.
pub fn open(path: &Path) -> Result<Self> {
let conn =
Connection::open(path).with_context(|| format!("open todos db {}", path.display()))?;
conn.execute_batch(SCHEMA).context("apply todos schema")?;
Ok(Self {
conn: Mutex::new(conn),
})
}
/// Insert a todo, or update the existing one for `(subsystem, key)`
/// when `key` is `Some` and already present. A `None` key never
/// conflicts, so it always inserts a fresh row.
///
/// Returns `(id, changed)` where `changed` is `true` when the row is
/// new OR its `summary`/`source` actually differed — the caller uses
/// this to decide whether to signal the turn loop (re-pushing an
/// identical keyed todo is a no-op and must not re-wake).
///
/// # Errors
///
/// Propagates sqlite query / execute failures.
///
/// # Panics
///
/// Panics if the connection mutex is poisoned.
pub fn upsert(
&self,
subsystem: &str,
key: Option<&str>,
summary: &str,
source: Option<&str>,
) -> Result<(i64, bool)> {
let conn = self.conn.lock().unwrap();
let now = now_unix();
let existing: Option<(i64, String, Option<String>)> = if key.is_some() {
conn.query_row(
"SELECT id, summary, source FROM todos \
WHERE subsystem = ?1 AND subsystem_key IS ?2",
params![subsystem, key],
|row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
)
.ok()
} else {
None
};
if let Some((id, cur_summary, cur_source)) = existing {
let unchanged = cur_summary == summary && cur_source.as_deref() == source;
if unchanged {
return Ok((id, false));
}
conn.execute(
"UPDATE todos SET summary = ?1, source = ?2, updated_at = ?3 WHERE id = ?4",
params![summary, source, now, id],
)?;
return Ok((id, true));
}
conn.execute(
"INSERT INTO todos \
(subsystem, subsystem_key, summary, source, updated_at) \
VALUES (?1, ?2, ?3, ?4, ?5)",
params![subsystem, key, summary, source, now],
)?;
Ok((conn.last_insert_rowid(), true))
}
/// Clear producer-resolved todo(s) by `(subsystem, key)`. `key =
/// Some(k)` targets the one keyed row; `key = None` matches
/// `subsystem_key IS NULL`, i.e. **all** keyless todos for that
/// subsystem (clear a specific keyless one via [`Todos::mark_done`] by
/// id instead). Returns the number of rows deleted.
///
/// # Errors
///
/// Propagates the sqlite delete failure.
///
/// # Panics
///
/// Panics if the connection mutex is poisoned.
pub fn clear(&self, subsystem: &str, key: Option<&str>) -> Result<usize> {
let conn = self.conn.lock().unwrap();
let n = conn.execute(
"DELETE FROM todos WHERE subsystem = ?1 AND subsystem_key IS ?2",
params![subsystem, key],
)?;
Ok(n)
}
/// Clear every todo `subsystem` owns — used by a producer that rebuilds
/// its whole set on restart (cancel-and-recreate). Returns the number
/// of rows deleted.
///
/// # Errors
///
/// Propagates the sqlite delete failure.
///
/// # Panics
///
/// Panics if the connection mutex is poisoned.
pub fn clear_subsystem(&self, subsystem: &str) -> Result<usize> {
let conn = self.conn.lock().unwrap();
let n = conn.execute("DELETE FROM todos WHERE subsystem = ?1", params![subsystem])?;
Ok(n)
}
/// The agent marks one of its todos done, by id. Returns the number of
/// rows deleted (0 when the id was unknown / already gone).
///
/// # Errors
///
/// Propagates the sqlite delete failure.
///
/// # Panics
///
/// Panics if the connection mutex is poisoned.
pub fn mark_done(&self, id: i64) -> Result<usize> {
let conn = self.conn.lock().unwrap();
let n = conn.execute("DELETE FROM todos WHERE id = ?1", params![id])?;
Ok(n)
}
/// List todos, newest-updated first. `subsystem = Some(..)` filters to
/// one producer's set; `None` returns all.
///
/// # Errors
///
/// Propagates the sqlite prepare / query failures.
///
/// # Panics
///
/// Panics if the connection mutex is poisoned.
pub fn list(&self, subsystem: Option<&str>) -> Result<Vec<Todo>> {
let conn = self.conn.lock().unwrap();
let mut stmt = conn.prepare(
"SELECT id, subsystem, subsystem_key, summary, source, updated_at \
FROM todos \
WHERE (?1 IS NULL OR subsystem = ?1) \
ORDER BY updated_at DESC, id DESC",
)?;
let rows = stmt
.query_map(params![subsystem], |row| {
Ok(Todo {
id: row.get(0)?,
subsystem: row.get(1)?,
subsystem_key: row.get(2)?,
summary: row.get(3)?,
source: row.get(4)?,
updated_at: row.get(5)?,
})
})?
.collect::<rusqlite::Result<Vec<_>>>()?;
Ok(rows)
}
}
#[cfg(test)]
mod tests {
use super::*;
// Return the `TempDir` alongside the store so it outlives the test —
// dropping it early deletes the dir and SQLite fails with
// `SQLITE_READONLY_DBMOVED`.
fn store() -> (tempfile::TempDir, Todos) {
let dir = tempfile::tempdir().unwrap();
let db = Todos::open(&dir.path().join("todos.sqlite")).unwrap();
(dir, db)
}
#[test]
fn keyed_upsert_dedups_and_reports_changed() {
let (_dir, s) = store();
let (id1, changed1) = s
.upsert("matrix", Some("!room:x"), "1 unread", None)
.unwrap();
assert!(changed1, "first push is new → changed");
// Same key + same summary → no-op, not changed (must not re-wake).
let (id2, changed2) = s
.upsert("matrix", Some("!room:x"), "1 unread", None)
.unwrap();
assert_eq!(id1, id2, "keyed upsert updates in place, same row");
assert!(!changed2, "identical re-push is a no-op");
// Same key, new summary → updates, changed.
let (id3, changed3) = s
.upsert("matrix", Some("!room:x"), "3 unread", None)
.unwrap();
assert_eq!(id1, id3);
assert!(changed3);
assert_eq!(s.list(Some("matrix")).unwrap().len(), 1);
}
#[test]
fn keyless_todos_always_insert() {
let (_dir, s) = store();
let (a, _) = s.upsert("bash", None, "task done", None).unwrap();
let (b, _) = s.upsert("bash", None, "task done", None).unwrap();
assert_ne!(a, b, "keyless pushes are distinct one-offs");
assert_eq!(s.list(Some("bash")).unwrap().len(), 2);
}
#[test]
fn clear_and_mark_done_remove_rows() {
let (_dir, s) = store();
s.upsert("matrix", Some("!a:x"), "unread", None).unwrap();
let (id, _) = s.upsert("forge", Some("pr-1"), "review", None).unwrap();
assert_eq!(s.clear("matrix", Some("!a:x")).unwrap(), 1);
assert_eq!(s.mark_done(id).unwrap(), 1);
assert!(s.list(None).unwrap().is_empty());
}
#[test]
fn clear_subsystem_wipes_only_its_own() {
let (_dir, s) = store();
s.upsert("matrix", Some("!a:x"), "u", None).unwrap();
s.upsert("matrix", Some("!b:x"), "u", None).unwrap();
s.upsert("forge", Some("pr-1"), "r", None).unwrap();
assert_eq!(s.clear_subsystem("matrix").unwrap(), 2);
let left = s.list(None).unwrap();
assert_eq!(left.len(), 1);
assert_eq!(left[0].subsystem, "forge");
}
}

8
hive-c0re/src/coordinator.rs
View file

@ -16,6 +16,7 @@ use crate::container_view::{self, ContainerView};
use crate::dashboard_events::DashboardEvent;
use crate::operator_questions::OperatorQuestions;
use crate::socket_server::{self, AgentSocket};
use crate::todos::Todos;
/// Capacity of the dashboard event channel. Slow browser subscribers
/// (idle tab, throttled connection) drop frames past this — that's
@ -32,6 +33,11 @@ pub struct Coordinator {
pub broker: Arc<Broker>,
pub approvals: Arc<Approvals>,
pub questions: Arc<OperatorQuestions>,
/// Dynamic, subsystem-pushed todos (loose-ends v2). In-agent
/// subsystems (matrix, forge, bash) upsert/clear todos over mcp.sock
/// instead of firing wakes directly; `get_todos` merges these with
/// the computed static loose ends.
pub todos: Arc<Todos>,
/// Scheduled-prompts queue. One sqlite connection,
/// internal mutex; the worker drains due rows and the manager
/// handlers insert / cancel through the same handle.
@ -460,6 +466,7 @@ impl Coordinator {
let broker = Broker::open(db_path).context("open broker")?;
let approvals = Approvals::open(db_path).context("open approvals")?;
let questions = OperatorQuestions::open(db_path).context("open operator_questions")?;
let todos = Todos::open(db_path).context("open todos")?;
let scheduled_prompts = crate::scheduled_prompts::ScheduledPrompts::open(db_path)
.context("open scheduled_prompts")?;
// BuildLogs wants a directory (it picks its own `build_logs.sqlite`
@ -489,6 +496,7 @@ impl Coordinator {
broker: Arc::new(broker),
approvals: Arc::new(approvals),
questions: Arc::new(questions),
todos: Arc::new(todos),
scheduled_prompts: Arc::new(scheduled_prompts),
build_logs,
audit_log,

31
hive-c0re/src/loose_ends.rs
View file

@ -97,9 +97,40 @@ pub fn for_agent(coord: &Coordinator, agent: &str) -> Result<Vec<LooseEnd>> {
age_seconds: saturating_age(now, r.created_at.timestamp()),
});
}
// Dynamic, subsystem-pushed todos (loose-ends v2). Scoped to
// this agent; the producing subsystem or the agent itself clears them.
out.extend(todos_for(coord, agent, None)?);
Ok(out)
}
/// This agent's dynamic todos as `LooseEnd::Todo` rows, optionally
/// filtered to one `subsystem`. Shared by [`for_agent`] and the
/// `ListTodos` handler so the row-mapping lives in one place.
///
/// # Errors
///
/// Propagates the todo-store query failure.
pub fn todos_for(
coord: &Coordinator,
agent: &str,
subsystem: Option<&str>,
) -> Result<Vec<LooseEnd>> {
let now = now_unix();
Ok(coord
.todos
.list(agent, subsystem)?
.into_iter()
.map(|t| LooseEnd::Todo {
id: t.id,
subsystem: t.subsystem,
subsystem_key: t.subsystem_key,
summary: t.summary,
source: t.source,
age_seconds: saturating_age(now, t.updated_at.timestamp()),
})
.collect())
}
/// Hive-wide loose-ends view: EVERY pending approval + EVERY
/// unanswered question + EVERY pending reminder. Manager surface
/// only; sub-agents can't see each other's threads via the agent

1
hive-c0re/src/main.rs
View file

@ -43,6 +43,7 @@ pub(crate) use stats::{
};
pub(crate) use stores::{
approvals, audit_log, broker, build_logs, db, operator_questions, power, scheduled_prompts,
todos,
};
pub(crate) use workers::{
agent_sockets, auto_update, crash_watch, knowledge, mcp_sockets, reminder_scheduler,

100
hive-c0re/src/socket_server/mod.rs
View file

@ -596,6 +596,29 @@ async fn dispatch(req: &Request, agent: &str, coord: &Arc<Coordinator>) -> Respo
since_secs,
agent: target,
} => handle_reminder_rollup(coord, agent, target.as_deref(), *since_secs),
// Todos (loose-ends v2): in-container subsystems push/clear
// their own; the agent lists / marks its own done. Scoped to the
// calling agent (the socket identity) — no cross-agent access.
Request::UpsertTodo {
subsystem,
key,
summary,
source,
} => handle_upsert_todo(
coord,
agent,
subsystem,
key.as_deref(),
summary,
source.as_deref(),
),
Request::ClearTodo {
subsystem,
key,
all,
} => handle_clear_todo(coord, agent, subsystem, key.as_deref(), *all),
Request::ListTodos { subsystem } => handle_list_todos(coord, agent, subsystem.as_deref()),
Request::MarkTodoDone { id } => handle_mark_todo_done(coord, agent, *id),
// Orchestration / diagnostics verbs — gated per-verb on tool-group
// membership or topology (see `dispatch_orchestration`).
_ => dispatch_orchestration(req, agent, coord).await,
@ -790,6 +813,83 @@ fn handle_get_loose_ends(coord: &Arc<Coordinator>, agent: &str, target: Option<&
}
}
/// `UpsertTodo` — a subsystem pushes/updates one of this agent's todos.
/// Coalesces a wake ONLY when the row is new or actually changed, so
/// re-pushing an identical keyed todo is a silent no-op.
fn handle_upsert_todo(
coord: &Arc<Coordinator>,
agent: &str,
subsystem: &str,
key: Option<&str>,
summary: &str,
source: Option<&str>,
) -> Response {
match coord.todos.upsert(agent, subsystem, key, summary, source) {
Ok((_, changed)) => {
if changed {
let _ = coord.broker.send(&Message {
from: "todo".to_owned(),
to: agent.to_owned(),
body: "you have todos — call get_loose_ends to see them".to_owned(),
in_reply_to: None,
});
}
Response::Ok
}
Err(e) => Response::Err {
message: format!("{e:#}"),
},
}
}
/// `ClearTodo` — a producer clears a resolved todo by `(subsystem, key)`,
/// or wipes its whole set when `all` (cancel-and-recreate on restart).
fn handle_clear_todo(
coord: &Arc<Coordinator>,
agent: &str,
subsystem: &str,
key: Option<&str>,
all: bool,
) -> Response {
let result = if all {
coord.todos.clear_subsystem(agent, subsystem)
} else {
coord.todos.clear(agent, subsystem, key)
};
match result {
Ok(count) => Response::Acked {
count: u64::try_from(count).unwrap_or(0),
},
Err(e) => Response::Err {
message: format!("{e:#}"),
},
}
}
/// `ListTodos` — enumerate this agent's todos (optionally one subsystem's)
/// as `LooseEnd::Todo` rows, so a producer can reconcile its own set.
fn handle_list_todos(coord: &Arc<Coordinator>, agent: &str, subsystem: Option<&str>) -> Response {
match crate::loose_ends::todos_for(coord, agent, subsystem) {
Ok(loose_ends) => Response::LooseEnds { loose_ends },
Err(e) => Response::Err {
message: format!("{e:#}"),
},
}
}
/// `MarkTodoDone` — the agent clears one of its own todos by id (scoped to
/// the agent, so it can't touch another agent's).
fn handle_mark_todo_done(coord: &Arc<Coordinator>, agent: &str, id: i64) -> Response {
match coord.todos.mark_done(agent, id) {
Ok(count) => Response::Acked {
count: u64::try_from(count).unwrap_or(0),
},
Err(e) => Response::Err {
message: format!("{e:#}"),
},
}
}
/// `CountPendingReminders` — resolve the target (own / subtree free, else
/// `QueryAgentState`) then count its pending reminders.
fn handle_count_pending_reminders(

1
hive-c0re/src/stores/mod.rs
View file

@ -12,3 +12,4 @@ pub mod db;
pub mod operator_questions;
pub mod power;
pub mod scheduled_prompts;
pub mod todos;

341
hive-c0re/src/stores/todos.rs Normal file
View file

@ -0,0 +1,341 @@
//! Todo store — the persistent, DB-backed half of the "todos"
//! (loose-ends v2) system.
//!
//! Subsystems inside an agent's container (matrix, forge-notify, bash,
//! …) push *todos* to the agent over the mcp.sock protocol instead of
//! firing wakes directly. Todos are scoped to the owning `agent` (the
//! socket identity of the pushing container) and tagged with a
//! `subsystem` marker plus an optional `subsystem_key` (a matrix room
//! id, a bash task id, …); together `(agent, subsystem, subsystem_key)`
//! is the upsert/dedup key, so re-pushing the same item is idempotent
//! (no duplicate) and a producer can list / clear / rebuild only its own
//! set (e.g. matrix wipes + recreates its todos on daemon restart).
//!
//! Removal has two paths (mara's call): the producing subsystem
//! `clear`s a todo it has resolved (keyed by subsystem + key), or the
//! agent itself `mark_done`s one by id. Both delete the row.
//!
//! This table holds only the *dynamic* subsystem-pushed todos. The
//! static ones (pending approvals / questions / reminders / undelivered
//! messages) are still computed on demand in `loose_ends.rs`; `get_todos`
//! merges the two. Folding the static kinds into this table is a later
//! increment.
use std::path::Path;
use std::sync::Mutex;
use anyhow::{Context, Result};
use chrono::{DateTime, Utc};
use hive_sh4re::wire_time::now_unix;
use rusqlite::{Connection, params};
use serde::Serialize;
use crate::db::Migration;
const SCHEMA: &str = r"
CREATE TABLE IF NOT EXISTS todos (
id INTEGER PRIMARY KEY AUTOINCREMENT,
agent TEXT NOT NULL,
subsystem TEXT NOT NULL,
subsystem_key TEXT,
summary TEXT NOT NULL,
source TEXT,
created_at INTEGER NOT NULL,
updated_at INTEGER NOT NULL
);
-- (agent, subsystem, subsystem_key) is the upsert/dedup key. A NULL key
-- never conflicts (SQLite treats NULLs as distinct), so keyless todos
-- always insert as one-offs; keyed todos update in place.
CREATE UNIQUE INDEX IF NOT EXISTS idx_todos_dedup
ON todos (agent, subsystem, subsystem_key);
";
// New table — no legacy rows to migrate past the initial schema.
const MIGRATIONS: &[Migration] = &[];
/// One dynamic, subsystem-pushed todo.
#[derive(Debug, Clone, Serialize)]
pub struct Todo {
pub id: i64,
/// Owning agent (the container whose subsystem pushed it).
pub agent: String,
/// Producing subsystem marker (`"matrix"`, `"forge"`, `"bash"`, …).
pub subsystem: String,
/// Optional subsystem-specific dedup key (matrix room id, bash task
/// id, …). `None` = a keyless one-off todo.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub subsystem_key: Option<String>,
/// Human-readable one-line summary shown to the agent.
pub summary: String,
/// Optional free-text provenance (e.g. the room name / task label).
#[serde(default, skip_serializing_if = "Option::is_none")]
pub source: Option<String>,
pub created_at: DateTime<Utc>,
pub updated_at: DateTime<Utc>,
}
pub struct Todos {
conn: Mutex<Connection>,
}
impl Todos {
/// Open (creating if needed) the todo store at `path`.
///
/// # Errors
///
/// Propagates sqlite open / schema-apply / migration failures.
pub fn open(path: &Path) -> Result<Self> {
let conn = crate::db::open(path, "todos")?;
conn.execute_batch(SCHEMA).context("apply todos schema")?;
crate::db::apply_versioned_migrations(&conn, "todos", MIGRATIONS)?;
Ok(Self {
conn: Mutex::new(conn),
})
}
/// Insert a todo for `agent`, or update the existing one for
/// `(agent, subsystem, key)` when `key` is `Some` and already
/// present. A `None` key never conflicts, so it always inserts a
/// fresh row.
///
/// Returns `(id, changed)` where `changed` is `true` when the row is
/// new OR its `summary`/`source` actually differed — the caller uses
/// this to decide whether to coalesce a wake (re-pushing an identical
/// keyed todo is a no-op and must not re-wake).
///
/// # Errors
///
/// Propagates sqlite query / execute failures.
///
/// # Panics
///
/// Panics if the connection mutex is poisoned.
pub fn upsert(
&self,
agent: &str,
subsystem: &str,
key: Option<&str>,
summary: &str,
source: Option<&str>,
) -> Result<(i64, bool)> {
let conn = self.conn.lock().unwrap();
let now = now_unix();
let existing: Option<(i64, String, Option<String>)> = if key.is_some() {
conn.query_row(
"SELECT id, summary, source FROM todos \
WHERE agent = ?1 AND subsystem = ?2 AND subsystem_key IS ?3",
params![agent, subsystem, key],
|row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
)
.ok()
} else {
None
};
if let Some((id, cur_summary, cur_source)) = existing {
let unchanged = cur_summary == summary && cur_source.as_deref() == source;
if unchanged {
return Ok((id, false));
}
conn.execute(
"UPDATE todos SET summary = ?1, source = ?2, updated_at = ?3 WHERE id = ?4",
params![summary, source, now, id],
)?;
return Ok((id, true));
}
conn.execute(
"INSERT INTO todos \
(agent, subsystem, subsystem_key, summary, source, created_at, updated_at) \
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?6)",
params![agent, subsystem, key, summary, source, now],
)?;
Ok((conn.last_insert_rowid(), true))
}
/// Clear producer-resolved todo(s) by `(agent, subsystem, key)`.
/// `key = Some(k)` targets the one keyed row; `key = None` matches
/// `subsystem_key IS NULL`, i.e. **all** keyless todos for that
/// subsystem (keyless rows have no distinguishing key — clear a
/// specific one via [`Todos::mark_done`] by id instead). Returns the
/// number of rows deleted (0 when nothing matched).
///
/// # Errors
///
/// Propagates the sqlite delete failure.
///
/// # Panics
///
/// Panics if the connection mutex is poisoned.
pub fn clear(&self, agent: &str, subsystem: &str, key: Option<&str>) -> Result<usize> {
let conn = self.conn.lock().unwrap();
let n = conn.execute(
"DELETE FROM todos WHERE agent = ?1 AND subsystem = ?2 AND subsystem_key IS ?3",
params![agent, subsystem, key],
)?;
Ok(n)
}
/// Clear every todo `agent`'s `subsystem` owns — used by a producer
/// that rebuilds its whole set on restart (cancel-and-recreate).
/// Returns the number of rows deleted.
///
/// # Errors
///
/// Propagates the sqlite delete failure.
///
/// # Panics
///
/// Panics if the connection mutex is poisoned.
pub fn clear_subsystem(&self, agent: &str, subsystem: &str) -> Result<usize> {
let conn = self.conn.lock().unwrap();
let n = conn.execute(
"DELETE FROM todos WHERE agent = ?1 AND subsystem = ?2",
params![agent, subsystem],
)?;
Ok(n)
}
/// The agent marks one of *its own* todos done, by id. Scoped to
/// `agent` so one agent can't clear another's. Returns the number of
/// rows deleted (0 when the id was unknown / not owned / already gone).
///
/// # Errors
///
/// Propagates the sqlite delete failure.
///
/// # Panics
///
/// Panics if the connection mutex is poisoned.
pub fn mark_done(&self, agent: &str, id: i64) -> Result<usize> {
let conn = self.conn.lock().unwrap();
let n = conn.execute(
"DELETE FROM todos WHERE id = ?1 AND agent = ?2",
params![id, agent],
)?;
Ok(n)
}
/// List `agent`'s todos, newest-updated first. `subsystem = Some(..)`
/// filters to one producer's set (so a producer can enumerate +
/// reconcile only its own); `None` returns all of the agent's.
///
/// # Errors
///
/// Propagates the sqlite prepare / query failures.
///
/// # Panics
///
/// Panics if the connection mutex is poisoned.
pub fn list(&self, agent: &str, subsystem: Option<&str>) -> Result<Vec<Todo>> {
let conn = self.conn.lock().unwrap();
let mut stmt = conn.prepare(
"SELECT id, agent, subsystem, subsystem_key, summary, source, created_at, updated_at \
FROM todos \
WHERE agent = ?1 AND (?2 IS NULL OR subsystem = ?2) \
ORDER BY updated_at DESC, id DESC",
)?;
let rows = stmt
.query_map(params![agent, subsystem], |row| {
let created: i64 = row.get(6)?;
let updated: i64 = row.get(7)?;
Ok(Todo {
id: row.get(0)?,
agent: row.get(1)?,
subsystem: row.get(2)?,
subsystem_key: row.get(3)?,
summary: row.get(4)?,
source: row.get(5)?,
created_at: DateTime::from_timestamp(created, 0).unwrap_or_default(),
updated_at: DateTime::from_timestamp(updated, 0).unwrap_or_default(),
})
})?
.collect::<rusqlite::Result<Vec<_>>>()?;
Ok(rows)
}
}
#[cfg(test)]
mod tests {
use super::*;
// Return the `TempDir` alongside the store so it outlives the test —
// dropping it early deletes the dir and SQLite fails with
// `SQLITE_READONLY_DBMOVED`.
fn store() -> (tempfile::TempDir, Todos) {
let dir = tempfile::tempdir().unwrap();
let db = Todos::open(&dir.path().join("todos.sqlite")).unwrap();
(dir, db)
}
#[test]
fn keyed_upsert_dedups_and_reports_changed() {
let (_dir, s) = store();
let (id1, changed1) = s
.upsert("alice", "matrix", Some("!room:x"), "1 unread", None)
.unwrap();
assert!(changed1, "first push is new → changed");
// Same key + same summary → no-op, not changed (must not re-wake).
let (id2, changed2) = s
.upsert("alice", "matrix", Some("!room:x"), "1 unread", None)
.unwrap();
assert_eq!(id1, id2, "keyed upsert updates in place, same row");
assert!(!changed2, "identical re-push is a no-op");
// Same key, new summary → updates, changed.
let (id3, changed3) = s
.upsert("alice", "matrix", Some("!room:x"), "3 unread", None)
.unwrap();
assert_eq!(id1, id3);
assert!(changed3);
assert_eq!(s.list("alice", Some("matrix")).unwrap().len(), 1);
}
#[test]
fn todos_are_scoped_per_agent() {
let (_dir, s) = store();
// Same subsystem+key for two agents → distinct rows.
s.upsert("alice", "matrix", Some("!r:x"), "u", None)
.unwrap();
s.upsert("bob", "matrix", Some("!r:x"), "u", None).unwrap();
assert_eq!(s.list("alice", None).unwrap().len(), 1);
assert_eq!(s.list("bob", None).unwrap().len(), 1);
// bob can't mark alice's todo done.
let alice_id = s.list("alice", None).unwrap()[0].id;
assert_eq!(s.mark_done("bob", alice_id).unwrap(), 0);
assert_eq!(s.mark_done("alice", alice_id).unwrap(), 1);
}
#[test]
fn keyless_todos_always_insert() {
let (_dir, s) = store();
let (a, _) = s.upsert("alice", "bash", None, "task done", None).unwrap();
let (b, _) = s.upsert("alice", "bash", None, "task done", None).unwrap();
assert_ne!(a, b, "keyless pushes are distinct one-offs");
assert_eq!(s.list("alice", Some("bash")).unwrap().len(), 2);
}
#[test]
fn clear_and_mark_done_remove_rows() {
let (_dir, s) = store();
s.upsert("alice", "matrix", Some("!a:x"), "unread", None)
.unwrap();
let (id, _) = s
.upsert("alice", "forge", Some("pr-1"), "review", None)
.unwrap();
assert_eq!(s.clear("alice", "matrix", Some("!a:x")).unwrap(), 1);
assert_eq!(s.mark_done("alice", id).unwrap(), 1);
assert!(s.list("alice", None).unwrap().is_empty());
}
#[test]
fn clear_subsystem_wipes_only_its_own() {
let (_dir, s) = store();
s.upsert("alice", "matrix", Some("!a:x"), "u", None)
.unwrap();
s.upsert("alice", "matrix", Some("!b:x"), "u", None)
.unwrap();
s.upsert("alice", "forge", Some("pr-1"), "r", None).unwrap();
assert_eq!(s.clear_subsystem("alice", "matrix").unwrap(), 2);
let left = s.list("alice", None).unwrap();
assert_eq!(left.len(), 1);
assert_eq!(left[0].subsystem, "forge");
}
}

35
hive-core-agent-sock/src/lib.rs
View file

@ -102,6 +102,41 @@ pub enum Request {
#[serde(default, skip_serializing_if = "Option::is_none")]
agent: Option<String>,
},
/// Upsert a *todo* (loose-ends v2) from an in-container
/// subsystem (matrix / forge / bash). `subsystem` is the producer
/// marker; `key` is the optional subsystem-specific dedup key (a
/// matrix room id, a bash task id). Re-pushing an identical keyed
/// todo is a no-op; a new-or-changed one coalesces a wake to the
/// agent. Keyless todos always insert as one-offs.
UpsertTodo {
subsystem: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
key: Option<String>,
summary: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
source: Option<String>,
},
/// Clear producer-resolved todo(s) by `(subsystem, key)`. `key =
/// Some(k)` clears the one keyed row; `key = None` clears **all** of
/// the subsystem's keyless todos (rows with no key can't be told
/// apart — clear a specific one via `MarkTodoDone` by id). `all =
/// true` wipes the producer's whole set (cancel-and-recreate on
/// daemon restart).
ClearTodo {
subsystem: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
key: Option<String>,
#[serde(default)]
all: bool,
},
/// List todos, optionally filtered to one `subsystem` (a producer
/// enumerating its own set). `None` = all.
ListTodos {
#[serde(default, skip_serializing_if = "Option::is_none")]
subsystem: Option<String>,
},
/// The agent marks one of its own todos done, by id.
MarkTodoDone { id: i64 },
/// Count of pending (un-delivered) reminders. On the agent socket:
/// same target rules as `GetLooseEnds` (self/children free;
/// non-children require `query_agent_state`; `"*"` rejected).

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

@ -512,6 +512,44 @@ pub fn list_invites(client: &Client) -> DaemonResponse {
DaemonResponse::ok(&invites)
}
/// Rewrite `mcp-loose-ends/matrix.json` with a summary of all pending
/// room invites. The harness scans this directory generically in
/// `get_loose_ends` — no matrix-specific code needed there.
///
/// Called after an invite arrives (from the sync handler) and after a
/// room is joined (to remove the accepted invite from loose-ends).
/// Atomic write (tmp + rename) so the harness never reads a partial file.
pub async fn refresh_invite_loose_ends(client: &Client) {
let invites = client.invited_rooms();
let dir = crate::paths::mcp_loose_ends_dir();
if let Err(e) = tokio::fs::create_dir_all(&dir).await {
tracing::warn!(error = ?e, "matrix: create mcp-loose-ends dir failed");
return;
}
let items: Vec<String> = invites
.iter()
.map(|room| {
let label = room_label(room);
format!(
"[matrix] pending invite: {label} — use list_invites to see, resolve_invite to accept or reject"
)
})
.collect();
let dest = dir.join("matrix.json");
let tmp = dest.with_extension("json.tmp");
let json = serde_json::to_string(&items).unwrap_or_else(|_| "[]".to_owned());
match tokio::fs::write(&tmp, &json).await {
Ok(()) => {
if let Err(e) = tokio::fs::rename(&tmp, &dest).await {
tracing::warn!(error = ?e, "matrix: rename mcp-loose-ends/matrix.json failed");
}
}
Err(e) => {
tracing::warn!(error = ?e, "matrix: write mcp-loose-ends/matrix.json.tmp failed");
}
}
}
pub async fn join_room(client: &Client, room_ref: &str) -> DaemonResponse {
let parsed: &RoomOrAliasId = match room_ref.try_into() {
Ok(p) => p,
@ -522,14 +560,9 @@ pub async fn join_room(client: &Client, room_ref: &str) -> DaemonResponse {
let server_names: Vec<OwnedServerName> = vec![];
match client.join_room_by_id_or_alias(parsed, &server_names).await {
Ok(room) => {
// Clear the invite todo so the accepted invite drops out of
// `get_loose_ends` immediately (the sweep would also clear it
// on its next tick, but this makes it instant).
let _ = crate::wake::send_todo_clear(
Some(&crate::timeline::invite_key(room.room_id())),
false,
)
.await;
// Refresh loose-ends so the accepted invite is removed from
// `get_loose_ends` output immediately after the agent joins.
refresh_invite_loose_ends(client).await;
DaemonResponse::ok(&serde_json::json!({
"joined": true,
"room_id": room.room_id().to_string(),
@ -562,12 +595,7 @@ pub async fn resolve_invite(
};
match room.leave().await {
Ok(()) => {
// Clear the invite todo immediately on reject.
let _ = crate::wake::send_todo_clear(
Some(&crate::timeline::invite_key(room.room_id())),
false,
)
.await;
refresh_invite_loose_ends(client).await;
DaemonResponse::ok(&serde_json::json!({
"rejected": true,
"room_id": room.room_id().to_string(),

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

@ -73,6 +73,7 @@ async fn main() -> Result<()> {
let cfgs = accounts::configured().context("read matrix account config")?;
let mcp_socket = paths::daemon_socket();
let hyperhive_socket = paths::hyperhive_socket();
let multi = cfgs.len() > 1;
let primary = cfgs[0].name.clone();
let mut registry = Registry::new(primary);
@ -80,10 +81,10 @@ async fn main() -> Result<()> {
for (idx, cfg) in cfgs.into_iter().enumerate() {
let is_primary = idx == 0;
// Account-tag the todos only in multi-account mode so single-
// account todo summaries stay byte-identical to the legacy format.
// Account-tag the wakes only in multi-account mode so single-
// account wake bodies stay byte-identical to the legacy format.
let tag = multi.then(|| cfg.name.clone());
match bring_up_account(&cfg, tag.clone(), is_primary).await {
match bring_up_account(&cfg, &hyperhive_socket, tag.clone(), is_primary).await {
Ok(Some((client, sync_loop))) => {
registry.insert(cfg.name, client);
sync_loops.push(sync_loop);
@ -108,7 +109,8 @@ async fn main() -> Result<()> {
// before being skipped for this daemon lifetime.
Err(e) if is_primary => return Err(e.context("bring up primary matrix account")),
Err(e) => {
if let Some((client, sync_loop)) = bring_up_secondary_with_retry(&cfg, tag, e).await
if let Some((client, sync_loop)) =
bring_up_secondary_with_retry(&cfg, &hyperhive_socket, tag, e).await
{
registry.insert(cfg.name, client);
sync_loops.push(sync_loop);
@ -187,6 +189,7 @@ async fn main() -> Result<()> {
/// failure, no token, or all retries exhausted).
async fn bring_up_secondary_with_retry(
cfg: &AccountCfg,
hyperhive_socket: &std::path::Path,
tag: Option<String>,
first_error: anyhow::Error,
) -> Option<(Client, SyncLoop)> {
@ -216,7 +219,7 @@ async fn bring_up_secondary_with_retry(
"retrying secondary account bring-up after backoff"
);
tokio::time::sleep(std::time::Duration::from_secs(delay)).await;
match bring_up_account(cfg, tag.clone(), false).await {
match bring_up_account(cfg, hyperhive_socket, tag.clone(), false).await {
Ok(Some((client, sync_loop))) => {
tracing::info!(account = %cfg.name, "secondary matrix account recovered");
return Some((client, sync_loop));
@ -257,6 +260,7 @@ async fn bring_up_secondary_with_retry(
/// handling). The returned sync loop is driven by the caller.
async fn bring_up_account(
cfg: &AccountCfg,
hyperhive_socket: &std::path::Path,
tag: Option<String>,
is_primary: bool,
) -> Result<Option<(Client, SyncLoop)>> {
@ -289,26 +293,30 @@ async fn bring_up_account(
let sync_client = client.clone();
let cb_client = client.clone();
// Separate dedup sets: one tracks invites already pushed as todos, the
let wake_socket = Arc::new(hyperhive_socket.to_path_buf());
// Separate dedup sets: one tracks invites already woken about, the
// other unread-message rooms. Both are pruned to their current state
// each sweep (see the sweep fns) so re-invites / new messages re-push.
// each sweep (see the sweep fns) so re-invites / new messages re-wake.
let invite_notified = Arc::new(tokio::sync::Mutex::new(std::collections::HashSet::new()));
let unread_notified = Arc::new(tokio::sync::Mutex::new(std::collections::HashSet::new()));
// Startup cancel-and-recreate (loose-ends v2): wipe this agent's
// matrix todos so stale ones (rooms read / 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;
// matrix todos so stale ones (rooms read while the daemon was down) don't
// linger, then let the first sweep rebuild the set to match current
// unread reality. Best-effort; the sweep converges regardless.
let _ = crate::wake::send_todo_clear(hyperhive_socket, None, true).await;
let sync_loop: SyncLoop = Box::pin(async move {
sync_client
.sync_with_callback(SyncSettings::default(), move |_response| {
let client = cb_client.clone();
let socket = wake_socket.clone();
let invite_notified = invite_notified.clone();
let unread_notified = unread_notified.clone();
let tag = tag.clone();
async move {
timeline::sweep_invites(&client, &invite_notified, tag.as_deref()).await;
timeline::sweep_unread(&client, &unread_notified, tag.as_deref()).await;
timeline::sweep_invites(&client, &socket, &invite_notified, tag.as_deref())
.await;
timeline::sweep_unread(&client, &socket, &unread_notified, tag.as_deref())
.await;
matrix_sdk::LoopCtrl::Continue
}
})

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

@ -72,3 +72,22 @@ pub fn accounts_file() -> PathBuf {
let state_dir = std::env::var("HYPERHIVE_STATE_DIR").unwrap_or_default();
PathBuf::from(format!("{state_dir}/matrix-accounts.json"))
}
/// Hyperhive control socket — the daemon writes wake signals here so
/// the harness drives a new claude turn on incoming matrix events.
/// Mirrors the path `forge_notify` writes to.
#[must_use]
pub fn hyperhive_socket() -> PathBuf {
std::env::var_os("HIVE_CONTROL_SOCKET")
.map_or_else(|| PathBuf::from("/run/hive/mcp.sock"), PathBuf::from)
}
/// Directory where MCP daemons write loose-end summary files for the harness.
/// Each daemon writes `<name>.json` here; the harness scans the dir in
/// `get_loose_ends` to surface active work from all MCPs generically.
/// Resolution lives in `hive_sh4re::paths` so the harness + every MCP
/// daemon agree on the location.
#[must_use]
pub fn mcp_loose_ends_dir() -> PathBuf {
hive_sh4re::paths::mcp_loose_ends_dir()
}

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

@ -1,18 +1,19 @@
//! Matrix → hyperhive todo bridge (loose-ends v2). Both room messages and
//! room invites are surfaced to the agent by SWEEPING state on every
//! post-sync callback, not by one-shot `m.room.message` /
//! `StrippedRoomMemberEvent` handlers: a one-shot signal that raced a
//! socket-down window (a container rebuild) was dropped with no retry,
//! leaving the agent deaf to matrix activity until manually prompted.
//! Sweeping the reliable sync path re-checks each tick and self-heals a
//! dropped push on the next one — each todo is idempotent by room key.
//! Matrix → hyperhive wake bridge. Both room messages and room invites
//! are surfaced to the agent by SWEEPING state on every post-sync
//! callback, not by one-shot `m.room.message` / `StrippedRoomMemberEvent`
//! handlers: a one-shot wake whose `send_wake` raced a hive-c0re / socket-
//! down window (a container rebuild) was dropped with no retry, leaving
//! the agent deaf to matrix activity until manually prompted. Sweeping the
//! reliable sync path re-checks each tick and self-heals a dropped wake on
//! the next one.
//!
//! Todo summaries stay short: `sweep_unread` pushes the per-room unread
//! Wake bodies stay short: `sweep_unread` sends the all-rooms unread
//! summary (`wake::format_unread_summary`); the message stays unread
//! server-side so the agent fetches detail via `read_room`. Self-sent
//! messages never raise an unread notification, so they never wake.
use std::collections::HashSet;
use std::path::Path;
use matrix_sdk::{Client, ruma::OwnedRoomId};
use tokio::sync::Mutex;
@ -37,6 +38,7 @@ use crate::{handlers, wake};
/// no explicit self-filter is needed here.
pub async fn sweep_unread(
client: &Client,
socket: &Path,
notified: &Mutex<HashSet<OwnedRoomId>>,
account_tag: Option<&str>,
) {
@ -51,7 +53,7 @@ pub async fn sweep_unread(
active.difference(&unread_ids).cloned().collect()
};
for id in stale {
if wake::send_todo_clear(Some(id.as_str()), false)
if wake::send_todo_clear(socket, Some(id.as_str()), false)
.await
.is_ok()
{
@ -59,15 +61,15 @@ pub async fn sweep_unread(
}
}
// Upsert a todo per currently-unread room. The harness coalesces the
// wake iff the summary is new or changed, so re-upserting an unchanged
// room every sync tick is a cheap no-op (no re-wake).
// Upsert a todo per currently-unread room. hive-c0re coalesces the wake
// iff the summary is new or changed, so re-upserting an unchanged room
// every sync tick is a cheap server-side no-op (no re-wake).
for (id, ru) in &unread {
let summary = wake::tag_account(
account_tag,
wake::format_unread_summary(std::slice::from_ref(ru)),
);
match wake::send_todo_upsert(id.as_str(), &summary).await {
match wake::send_todo_upsert(socket, id.as_str(), &summary).await {
Ok(()) => {
notified.lock().await.insert(id.clone());
}
@ -93,6 +95,7 @@ pub async fn sweep_unread(
/// decides whether to accept or reject by calling `resolve_invite`.
pub async fn sweep_invites(
client: &Client,
socket: &Path,
notified: &Mutex<HashSet<OwnedRoomId>>,
account_tag: Option<&str>,
) {
@ -100,52 +103,38 @@ pub async fn sweep_invites(
let current_ids: HashSet<OwnedRoomId> =
current.iter().map(|r| r.room_id().to_owned()).collect();
// Invites we previously pushed a todo for that are no longer pending
// (joined / rejected / withdrawn) → clear their todo, then drop from
// `notified` on success so a future re-invite re-upserts (retry next
// tick on failure).
let stale: Vec<OwnedRoomId> = {
let seen = notified.lock().await;
seen.difference(&current_ids).cloned().collect()
};
for id in stale {
if wake::send_todo_clear(Some(&invite_key(&id)), false)
.await
.is_ok()
{
notified.lock().await.remove(&id);
let mut fresh = Vec::new();
{
let mut seen = notified.lock().await;
// Drop invites that are no longer pending (joined/rejected/withdrawn)
// so a future re-invite to the same room wakes the agent again.
seen.retain(|id| current_ids.contains(id));
for room in &current {
if seen.insert(room.room_id().to_owned()) {
fresh.push(room.clone());
}
}
}
if fresh.is_empty() {
return;
}
// Upsert a todo per pending invite. Keyed `invite:<room>` (distinct
// from the unread sweep's `<room>` key) so the two never collide; the
// harness coalesces the wake iff the summary is new or changed, so
// re-upserting an unchanged invite every tick is a cheap no-op.
for room in &current {
// Refresh loose-ends before waking so the invite is visible in
// get_loose_ends during the agent's turn.
handlers::refresh_invite_loose_ends(client).await;
for room in fresh {
let room_id = room.room_id().to_owned();
let label = room.name().unwrap_or_else(|| room_id.to_string());
let summary = wake::tag_account(
tracing::info!(%room_id, "matrix: pending invite swept, waking agent");
let body = wake::tag_account(
account_tag,
format!(
"[matrix] invited to {label} ({room_id}) — \
use list_invites to see pending invites, resolve_invite to accept or reject"
),
);
match wake::send_todo_upsert(&invite_key(&room_id), &summary).await {
Ok(()) => {
notified.lock().await.insert(room_id);
}
Err(e) => {
tracing::warn!(error = %e, room = %room_id, "matrix: invite todo upsert failed; will retry next sweep");
}
if let Err(e) = wake::send_wake(socket, &body).await {
tracing::warn!(error = %e, "matrix: failed to deliver invite-wake to hyperhive");
}
}
}
/// Todo dedup key for a pending invite. Namespaced with an `invite:`
/// prefix so an invited room and an unread room (which the unread sweep
/// keys by bare room id) never share a todo row. `pub(crate)` so the
/// invite-resolution handlers can clear the matching todo immediately.
pub(crate) fn invite_key(room_id: &matrix_sdk::ruma::RoomId) -> String {
format!("invite:{room_id}")
}

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

@ -1,14 +1,17 @@
//! Todo writer: pushes matrix *todos* (loose-ends v2) to the harness's
//! in-agent socket (`HIVE_AGENT_SOCKET`) when rooms have unread messages
//! or pending invites, so claude drives a turn to handle them. One JSON
//! line per op (`upsert_todo` / `clear_todo`), keyed by room id so
//! re-pushing an unchanged item is an idempotent no-op and resolving one
//! clears it. The harness owns the todo store locally and signals its own
//! turn loop — no hive-c0re round-trip.
//! Wake-signal writer: notifies the hyperhive harness when an incoming
//! matrix event arrives so claude drives a new turn.
//!
//! Todo summaries stay short: a SHORT TEASER, not the full message — the
//! agent then reads the unmarked event via the `read_room` MCP tool.
//! Truncation to ~100 chars keeps the summary focused.
//! Same wire shape as `hive-agent::forge_notify`'s wake: a single JSON
//! line written to the hyperhive control socket (`/run/hive/mcp.sock`
//! by default) carrying an `Request::Wake { from, body }`.
//! The agent harness's `agent_server` parses it and treats it as a
//! `Wake` from the matrix subsystem.
//!
//! Per the operator's call (phase 3): the body is a SHORT TEASER, not
//! the full message — the agent then reads the unmarked event via
//! the `read_room` MCP tool. Truncation to ~100 chars keeps the wake
//! prompt focused (`forge_notify` embeds longer excerpts because the
//! agent doesn't have a follow-up read-the-original tool for forge).
use std::path::Path;
@ -16,59 +19,65 @@ use anyhow::{Context, Result};
use tokio::io::AsyncWriteExt;
use tokio::net::UnixStream;
/// The harness-served in-agent socket (`HIVE_AGENT_SOCKET`) where todo ops
/// go — distinct from the host-served control socket used by [`send_wake`].
/// `None` when unset/empty, in which case todo sends are a best-effort
/// no-op (a standalone daemon without the harness socket).
fn agent_socket() -> Option<std::path::PathBuf> {
std::env::var_os("HIVE_AGENT_SOCKET")
.filter(|v| !v.is_empty())
.map(std::path::PathBuf::from)
}
/// Upsert a matrix-subsystem *todo* (loose-ends v2) on the harness's
/// in-agent socket — the replacement for a direct wake. `key` is the room
/// id (the dedup key); the harness signals a turn iff the todo is new or
/// its `summary` changed. Best-effort: a no-op when `HIVE_AGENT_SOCKET`
/// isn't configured.
/// Send an `Request::Wake { from: "matrix", body }` to the hyperhive
/// control socket at `socket`. Best-effort: returns Err on any plumbing
/// failure; callers log + ignore so a wake delivery hiccup doesn't tear
/// down the matrix sync loop.
///
/// Wire format matches `hive_core_agent_sock::Request` tagged with `"cmd"` per
/// `#[serde(tag = "cmd", rename_all = "snake_case")]`. Must be `"cmd"`,
/// not `"kind"` — the harness deserialises against the hive-sh4re type
/// and silently discards requests that don't match.
///
/// # Errors
///
/// Returns an error on socket connect failure, serialisation failure,
/// or I/O error writing to or reading from the socket.
pub async fn send_todo_upsert(key: &str, summary: impl AsRef<str>) -> Result<()> {
let Some(socket) = agent_socket() else {
return Ok(());
};
pub async fn send_wake(socket: &Path, body: impl AsRef<str>) -> Result<()> {
let payload = serde_json::json!({
"cmd": "wake",
"from": "matrix",
"body": body.as_ref(),
"transient": true,
});
send_line(socket, &payload).await
}
/// Upsert a matrix-subsystem *todo* (loose-ends v2) on the
/// hyperhive control socket — the replacement for a direct wake. `key` is
/// the room id (the dedup key); hive-c0re coalesces a wake iff the todo is
/// new or its `summary` changed. Best-effort like [`send_wake`].
///
/// # Errors
///
/// Returns an error on socket connect failure, serialisation failure,
/// or I/O error writing to or reading from the socket.
pub async fn send_todo_upsert(socket: &Path, key: &str, summary: impl AsRef<str>) -> Result<()> {
let payload = serde_json::json!({
"cmd": "upsert_todo",
"subsystem": "matrix",
"key": key,
"summary": summary.as_ref(),
});
send_line(&socket, &payload).await
send_line(socket, &payload).await
}
/// Clear matrix-subsystem todos on the harness's in-agent socket. `key =
/// Some(room)` clears one room's todo (it was read); `all = true` wipes the
/// whole matrix set (cancel-and-recreate on daemon restart). Best-effort:
/// a no-op when `HIVE_AGENT_SOCKET` isn't configured.
/// Clear matrix-subsystem todos. `key = Some(room)` clears one room's
/// todo (it was read); `all = true` wipes the whole matrix set
/// (cancel-and-recreate on daemon restart). Best-effort.
///
/// # Errors
///
/// Returns an error on socket connect failure, serialisation failure,
/// or I/O error writing to or reading from the socket.
pub async fn send_todo_clear(key: Option<&str>, all: bool) -> Result<()> {
let Some(socket) = agent_socket() else {
return Ok(());
};
pub async fn send_todo_clear(socket: &Path, key: Option<&str>, all: bool) -> Result<()> {
let payload = serde_json::json!({
"cmd": "clear_todo",
"subsystem": "matrix",
"key": key,
"all": all,
});
send_line(&socket, &payload).await
send_line(socket, &payload).await
}
/// Write one JSON request line to the hyperhive control socket and drain

7
nix/agent-modules/agent-service.nix
View file

@ -201,13 +201,6 @@ in
# `hive_c0re::agent_sockets::socket_path_for(name)` so lifecycle
# bind-mounts and gateway upstream config stay in sync.
HIVE_WEB_SOCKET = "/run/hive-agent/${userName}/web.sock";
# In-agent socket (loose-ends v2): the harness binds this and
# serves the `hive-agent-sock` todo protocol to the in-container
# producers (matrix daemon) + the MCP bridge (`get_loose_ends`).
# Same per-agent runtime dir as the web socket so all agent-user
# services in this container can reach it; purely in-container
# (never bind-mounted to the host — unlike hive-c0re's mcp.sock).
HIVE_AGENT_SOCKET = "/run/hive-agent/${userName}/agent.sock";
# Loopback URL of the persistent `hive-mcp-http` daemon that
# `render_claude_config` points claude at for the built-in
# surface (HTTP is the sole transport — no per-turn stdio child).

4
nix/agent-modules/matrix.nix
View file

@ -219,10 +219,6 @@ in
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.
HIVE_AGENT_SOCKET = "/run/hive-agent/${userName}/agent.sock";
RUST_LOG = "info";
}
# Homeserver URL: by default the daemon inherits the host-forwarded

4
nix/agent-modules/mcp.nix
View file

@ -221,10 +221,6 @@ in
wantedBy = [ "multi-user.target" ];
before = [ "hive-agent.service" ];
environment.RUST_LOG = "info";
# In-agent todo socket the harness serves (loose-ends v2): the
# `get_loose_ends` handler dials it to merge this agent's local todos
# with the static loose-ends from hive-c0re.
environment.HIVE_AGENT_SOCKET = "/run/hive-agent/${userName}/agent.sock";
serviceConfig = {
ExecStart = "${config.hyperhive.packages.hive-agent-mcp}/bin/hive-agent-mcp --http 127.0.0.1:${toString config.hyperhive.mcp.httpPort}";
SyslogIdentifier = "hive-mcp-http";