From 4545c08908ca0ff89aa756bd2295c1ac8591a5fc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?m=C3=BCde?= Date: Thu, 14 May 2026 21:40:38 +0200 Subject: [PATCH 1/5] hive-sh4re: per-agent socket protocol (Message/AgentRequest/AgentResponse) --- hive-sh4re/src/lib.rs | 44 ++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 43 insertions(+), 1 deletion(-) diff --git a/hive-sh4re/src/lib.rs b/hive-sh4re/src/lib.rs index 93222825..06683116 100644 --- a/hive-sh4re/src/lib.rs +++ b/hive-sh4re/src/lib.rs @@ -2,7 +2,11 @@ use serde::{Deserialize, Serialize}; -/// Requests on the host admin socket (`/run/hyperhive/host.sock`). +// ----------------------------------------------------------------------------- +// Host admin socket — /run/hyperhive/host.sock +// ----------------------------------------------------------------------------- + +/// Requests on the host admin socket. /// /// Wire format: one JSON object per line. #[derive(Debug, Clone, Serialize, Deserialize)] @@ -52,3 +56,41 @@ impl HostResponse { } } } + +// ----------------------------------------------------------------------------- +// Per-agent socket — /run/hyperhive/agents//mcp.sock on the host, +// bind-mounted into the container at /run/hive/mcp.sock. +// ----------------------------------------------------------------------------- + +/// A logical message between agents. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct Message { + pub from: String, + pub to: String, + pub body: String, +} + +/// Requests on a per-agent socket. The agent's identity is the socket +/// it came in on; `Send.from` is filled in by the server, not the client. +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(tag = "cmd", rename_all = "snake_case")] +pub enum AgentRequest { + /// Send a message to another agent. + Send { to: String, body: String }, + /// Pop one pending message from this agent's inbox. + Recv, +} + +/// Responses on a per-agent socket. +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(tag = "kind", rename_all = "snake_case")] +pub enum AgentResponse { + /// `Send` succeeded. + Ok, + /// Either `Send` failed or `Recv` errored. + Err { message: String }, + /// `Recv` produced a message. + Message { from: String, body: String }, + /// `Recv` found nothing pending. + Empty, +} From d79b5a39a1d411fefb2ac61a5e5c0b7c74a8e05e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?m=C3=BCde?= Date: Thu, 14 May 2026 21:42:51 +0200 Subject: [PATCH 2/5] hive-c0re: in-memory broker + per-agent sockets + coordinator state --- hive-c0re/src/agent_server.rs | 101 ++++++++++++++++++++++++++++++++++ hive-c0re/src/broker.rs | 30 ++++++++++ hive-c0re/src/coordinator.rs | 56 +++++++++++++++++++ hive-c0re/src/lifecycle.rs | 10 +++- hive-c0re/src/main.rs | 11 +++- hive-c0re/src/server.rs | 21 +++++-- 6 files changed, 220 insertions(+), 9 deletions(-) create mode 100644 hive-c0re/src/agent_server.rs create mode 100644 hive-c0re/src/broker.rs create mode 100644 hive-c0re/src/coordinator.rs diff --git a/hive-c0re/src/agent_server.rs b/hive-c0re/src/agent_server.rs new file mode 100644 index 00000000..89104a8a --- /dev/null +++ b/hive-c0re/src/agent_server.rs @@ -0,0 +1,101 @@ +//! Per-agent socket listener. Each socket file's existence on disk +//! authenticates the caller: connecting to `<.../agents/foo/mcp.sock>` means +//! you are `foo`. + +use std::path::PathBuf; +use std::sync::Arc; + +use anyhow::{Context, Result}; +use hive_sh4re::{AgentRequest, AgentResponse, Message}; +use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; +use tokio::net::{UnixListener, UnixStream}; +use tokio::task::JoinHandle; + +use crate::broker::Broker; + +pub struct AgentSocket { + pub path: PathBuf, + pub handle: JoinHandle<()>, +} + +pub async fn start( + agent: String, + socket_path: PathBuf, + broker: Arc, +) -> Result { + if let Some(parent) = socket_path.parent() { + std::fs::create_dir_all(parent) + .with_context(|| format!("create agent socket dir {}", parent.display()))?; + } + if socket_path.exists() { + std::fs::remove_file(&socket_path).context("remove stale agent socket")?; + } + let listener = UnixListener::bind(&socket_path) + .with_context(|| format!("bind agent socket {}", socket_path.display()))?; + tracing::info!(%agent, socket = %socket_path.display(), "agent socket listening"); + + let path = socket_path.clone(); + let handle = tokio::spawn(async move { + loop { + match listener.accept().await { + Ok((stream, _)) => { + let agent = agent.clone(); + let broker = broker.clone(); + tokio::spawn(async move { + if let Err(e) = serve(stream, agent, broker).await { + tracing::warn!(error = ?e, "agent connection failed"); + } + }); + } + Err(e) => { + tracing::warn!(error = ?e, "agent listener accept failed; exiting"); + return; + } + } + } + }); + Ok(AgentSocket { path, handle }) +} + +async fn serve(stream: UnixStream, agent: String, broker: Arc) -> Result<()> { + let (read, mut write) = stream.into_split(); + let mut reader = BufReader::new(read); + let mut line = String::new(); + loop { + line.clear(); + let n = reader.read_line(&mut line).await?; + if n == 0 { + return Ok(()); + } + let resp = match serde_json::from_str::(line.trim()) { + Ok(req) => dispatch(&req, &agent, &broker), + Err(e) => AgentResponse::Err { + message: format!("parse error: {e}"), + }, + }; + let mut payload = serde_json::to_string(&resp)?; + payload.push('\n'); + write.write_all(payload.as_bytes()).await?; + write.flush().await?; + } +} + +fn dispatch(req: &AgentRequest, agent: &str, broker: &Broker) -> AgentResponse { + match req { + AgentRequest::Send { to, body } => { + broker.send(Message { + from: agent.to_owned(), + to: to.clone(), + body: body.clone(), + }); + AgentResponse::Ok + } + AgentRequest::Recv => match broker.recv(agent) { + Some(msg) => AgentResponse::Message { + from: msg.from, + body: msg.body, + }, + None => AgentResponse::Empty, + }, + } +} diff --git a/hive-c0re/src/broker.rs b/hive-c0re/src/broker.rs new file mode 100644 index 00000000..2867c1f7 --- /dev/null +++ b/hive-c0re/src/broker.rs @@ -0,0 +1,30 @@ +//! In-memory message broker. Phase 3 replaces this with a sqlite-backed store. + +use std::collections::{HashMap, VecDeque}; +use std::sync::Mutex; + +use hive_sh4re::Message; + +#[derive(Default)] +pub struct Broker { + queues: Mutex>>, +} + +impl Broker { + pub fn new() -> Self { + Self::default() + } + + pub fn send(&self, message: Message) { + let mut queues = self.queues.lock().unwrap(); + queues + .entry(message.to.clone()) + .or_default() + .push_back(message); + } + + pub fn recv(&self, recipient: &str) -> Option { + let mut queues = self.queues.lock().unwrap(); + queues.get_mut(recipient).and_then(|q| q.pop_front()) + } +} diff --git a/hive-c0re/src/coordinator.rs b/hive-c0re/src/coordinator.rs new file mode 100644 index 00000000..66d8b7dd --- /dev/null +++ b/hive-c0re/src/coordinator.rs @@ -0,0 +1,56 @@ +//! Runtime state shared between the host admin socket and the per-agent +//! sockets: the broker plus a map of `name -> AgentSocket`. + +use std::collections::HashMap; +use std::path::PathBuf; +use std::sync::{Arc, Mutex}; + +use anyhow::{Context, Result}; + +use crate::agent_server::{self, AgentSocket}; +use crate::broker::Broker; + +const AGENT_RUNTIME_ROOT: &str = "/run/hyperhive/agents"; + +pub struct Coordinator { + pub broker: Arc, + agents: Mutex>, +} + +impl Coordinator { + pub fn new() -> Self { + Self { + broker: Arc::new(Broker::new()), + agents: Mutex::new(HashMap::new()), + } + } + + pub async fn register_agent(&self, name: &str) -> Result { + let agent_dir = Self::agent_dir(name); + std::fs::create_dir_all(&agent_dir) + .with_context(|| format!("create agent dir {}", agent_dir.display()))?; + let socket_path = Self::socket_path(name); + let socket = + agent_server::start(name.to_owned(), socket_path, self.broker.clone()).await?; + self.agents + .lock() + .unwrap() + .insert(name.to_owned(), socket); + Ok(agent_dir) + } + + pub fn unregister_agent(&self, name: &str) { + if let Some(socket) = self.agents.lock().unwrap().remove(name) { + socket.handle.abort(); + let _ = std::fs::remove_file(&socket.path); + } + } + + pub fn agent_dir(name: &str) -> PathBuf { + PathBuf::from(format!("{AGENT_RUNTIME_ROOT}/{name}")) + } + + pub fn socket_path(name: &str) -> PathBuf { + Self::agent_dir(name).join("mcp.sock") + } +} diff --git a/hive-c0re/src/lifecycle.rs b/hive-c0re/src/lifecycle.rs index 60f1f71d..72c26cdf 100644 --- a/hive-c0re/src/lifecycle.rs +++ b/hive-c0re/src/lifecycle.rs @@ -1,18 +1,24 @@ //! Thin async wrappers over `nixos-container`. +use std::path::Path; + use anyhow::{Context, Result, bail}; use tokio::process::Command; pub const AGENT_PREFIX: &str = "hive-agent-"; pub const HIVE_PREFIX: &str = "hive-"; +/// Mount point of the per-agent runtime directory inside the container. +pub const CONTAINER_RUNTIME_MOUNT: &str = "/run/hive"; + pub fn container_name(name: &str) -> String { format!("{AGENT_PREFIX}{name}") } -pub async fn spawn(name: &str, agent_flake: &str) -> Result<()> { +pub async fn spawn(name: &str, agent_flake: &str, agent_dir: &Path) -> Result<()> { let container = container_name(name); - run(&["create", &container, "--flake", agent_flake]).await?; + let bind = format!("{}:{CONTAINER_RUNTIME_MOUNT}", agent_dir.display()); + run(&["create", &container, "--flake", agent_flake, "--bind", &bind]).await?; run(&["start", &container]).await } diff --git a/hive-c0re/src/main.rs b/hive-c0re/src/main.rs index c03df938..760d1933 100644 --- a/hive-c0re/src/main.rs +++ b/hive-c0re/src/main.rs @@ -1,13 +1,19 @@ use std::path::PathBuf; +use std::sync::Arc; use anyhow::{Result, bail}; use clap::{Parser, Subcommand}; use hive_sh4re::{HostRequest, HostResponse}; +mod agent_server; +mod broker; mod client; +mod coordinator; mod lifecycle; mod server; +use coordinator::Coordinator; + #[derive(Parser)] #[command(name = "hive-c0re", about = "hyperhive coordinator daemon and CLI")] struct Cli { @@ -48,7 +54,10 @@ async fn main() -> Result<()> { let cli = Cli::parse(); match cli.cmd { - Cmd::Serve { agent_flake } => server::serve(&cli.socket, &agent_flake).await, + Cmd::Serve { agent_flake } => { + let coord = Arc::new(Coordinator::new()); + server::serve(&cli.socket, &agent_flake, coord).await + } Cmd::Spawn { name } => { render(client::request(&cli.socket, HostRequest::Spawn { name }).await?) } diff --git a/hive-c0re/src/server.rs b/hive-c0re/src/server.rs index 48eeb569..131e1b43 100644 --- a/hive-c0re/src/server.rs +++ b/hive-c0re/src/server.rs @@ -1,13 +1,15 @@ use std::path::Path; +use std::sync::Arc; use anyhow::{Context, Result}; use hive_sh4re::{HostRequest, HostResponse}; use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; use tokio::net::{UnixListener, UnixStream}; +use crate::coordinator::Coordinator; use crate::lifecycle; -pub async fn serve(socket: &Path, agent_flake: &str) -> Result<()> { +pub async fn serve(socket: &Path, agent_flake: &str, coord: Arc) -> Result<()> { if let Some(parent) = socket.parent() { std::fs::create_dir_all(parent) .with_context(|| format!("create socket parent {}", parent.display()))?; @@ -23,15 +25,16 @@ pub async fn serve(socket: &Path, agent_flake: &str) -> Result<()> { loop { let (stream, _) = listener.accept().await.context("accept connection")?; let agent_flake = agent_flake.to_owned(); + let coord = coord.clone(); tokio::spawn(async move { - if let Err(e) = handle(stream, &agent_flake).await { + if let Err(e) = handle(stream, &agent_flake, coord).await { tracing::warn!(error = ?e, "connection failed"); } }); } } -async fn handle(stream: UnixStream, agent_flake: &str) -> Result<()> { +async fn handle(stream: UnixStream, agent_flake: &str, coord: Arc) -> Result<()> { let (read, mut write) = stream.into_split(); let mut reader = BufReader::new(read); let mut line = String::new(); @@ -43,7 +46,7 @@ async fn handle(stream: UnixStream, agent_flake: &str) -> Result<()> { return Ok(()); } let resp = match serde_json::from_str::(line.trim()) { - Ok(req) => dispatch(&req, agent_flake).await, + Ok(req) => dispatch(&req, agent_flake, &coord).await, Err(e) => HostResponse::error(format!("parse error: {e}")), }; let mut payload = serde_json::to_string(&resp)?; @@ -53,17 +56,23 @@ async fn handle(stream: UnixStream, agent_flake: &str) -> Result<()> { } } -async fn dispatch(req: &HostRequest, agent_flake: &str) -> HostResponse { +async fn dispatch(req: &HostRequest, agent_flake: &str, coord: &Coordinator) -> HostResponse { let result: anyhow::Result = async { Ok(match req { HostRequest::Spawn { name } => { tracing::info!(%name, "spawn"); - lifecycle::spawn(name, agent_flake).await?; + let agent_dir = coord.register_agent(name).await?; + if let Err(e) = lifecycle::spawn(name, agent_flake, &agent_dir).await { + // Roll back socket registration if container creation failed. + coord.unregister_agent(name); + return Err(e); + } HostResponse::success() } HostRequest::Kill { name } => { tracing::info!(%name, "kill"); lifecycle::kill(name).await?; + coord.unregister_agent(name); HostResponse::success() } HostRequest::Rebuild { name } => { From 61407f41c999df893d3d04eb5fe17ed44bd8babb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?m=C3=BCde?= Date: Thu, 14 May 2026 21:44:05 +0200 Subject: [PATCH 3/5] hive-ag3nt: serve loop + send/recv CLI; template runs serve --- hive-ag3nt/Cargo.toml | 10 ++++ hive-ag3nt/src/bin/hive-ag3nt.rs | 92 +++++++++++++++++++++++++++++++- hive-ag3nt/src/bin/hive-m1nd.rs | 2 + hive-ag3nt/src/client.rs | 27 ++++++++++ hive-ag3nt/src/lib.rs | 6 +++ nix/templates/agent-base.nix | 6 ++- 6 files changed, 139 insertions(+), 4 deletions(-) create mode 100644 hive-ag3nt/src/client.rs diff --git a/hive-ag3nt/Cargo.toml b/hive-ag3nt/Cargo.toml index c268836b..f5bc59d7 100644 --- a/hive-ag3nt/Cargo.toml +++ b/hive-ag3nt/Cargo.toml @@ -3,6 +3,16 @@ name = "hive-ag3nt" edition.workspace = true version.workspace = true +[dependencies] +anyhow.workspace = true +clap.workspace = true +hive-sh4re.workspace = true +serde.workspace = true +serde_json.workspace = true +tokio.workspace = true +tracing.workspace = true +tracing-subscriber.workspace = true + [[bin]] name = "hive-ag3nt" path = "src/bin/hive-ag3nt.rs" diff --git a/hive-ag3nt/src/bin/hive-ag3nt.rs b/hive-ag3nt/src/bin/hive-ag3nt.rs index 03c1e0cf..021c400a 100644 --- a/hive-ag3nt/src/bin/hive-ag3nt.rs +++ b/hive-ag3nt/src/bin/hive-ag3nt.rs @@ -1,3 +1,91 @@ -fn main() { - println!("hive-ag3nt placeholder"); +use std::path::PathBuf; +use std::time::Duration; + +use anyhow::{Result, bail}; +use clap::{Parser, Subcommand}; +use hive_ag3nt::{DEFAULT_SOCKET, client}; +use hive_sh4re::{AgentRequest, AgentResponse}; + +#[derive(Parser)] +#[command(name = "hive-ag3nt", about = "hyperhive sub-agent harness")] +struct Cli { + /// Path to the per-agent MCP socket (bind-mounted from the host). + #[arg(long, global = true, default_value = DEFAULT_SOCKET)] + socket: PathBuf, + + #[command(subcommand)] + cmd: Cmd, +} + +#[derive(Subcommand)] +enum Cmd { + /// Run the long-lived harness loop. Polls inbox; prints messages to stdout. + Serve { + /// Inbox poll interval in milliseconds. + #[arg(long, default_value_t = 1000)] + poll_ms: u64, + }, + /// Send a message to another agent. + Send { to: String, body: String }, + /// Pop one message from the inbox. + Recv, +} + +#[tokio::main] +async fn main() -> Result<()> { + tracing_subscriber::fmt() + .with_env_filter( + tracing_subscriber::EnvFilter::try_from_default_env() + .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")), + ) + .init(); + + let cli = Cli::parse(); + match cli.cmd { + Cmd::Serve { poll_ms } => serve(&cli.socket, Duration::from_millis(poll_ms)).await, + Cmd::Send { to, body } => { + let resp = client::request(&cli.socket, AgentRequest::Send { to, body }).await?; + render(&resp)?; + check(&resp) + } + Cmd::Recv => { + let resp = client::request(&cli.socket, AgentRequest::Recv).await?; + render(&resp)?; + check(&resp) + } + } +} + +async fn serve(socket: &std::path::Path, interval: Duration) -> Result<()> { + tracing::info!(socket = %socket.display(), "hive-ag3nt serve"); + loop { + match client::request(socket, AgentRequest::Recv).await { + Ok(AgentResponse::Message { from, body }) => { + tracing::info!(%from, %body, "inbox"); + } + Ok(AgentResponse::Empty) => {} + Ok(AgentResponse::Ok) => { + tracing::warn!("recv produced Ok (unexpected)"); + } + Ok(AgentResponse::Err { message }) => { + tracing::warn!(%message, "recv error"); + } + Err(e) => { + tracing::warn!(error = ?e, "recv failed; retrying"); + } + } + tokio::time::sleep(interval).await; + } +} + +fn render(resp: &AgentResponse) -> Result<()> { + println!("{}", serde_json::to_string_pretty(resp)?); + Ok(()) +} + +fn check(resp: &AgentResponse) -> Result<()> { + if let AgentResponse::Err { message } = resp { + bail!("{message}"); + } + Ok(()) } diff --git a/hive-ag3nt/src/bin/hive-m1nd.rs b/hive-ag3nt/src/bin/hive-m1nd.rs index 314550ca..a63ce9d6 100644 --- a/hive-ag3nt/src/bin/hive-m1nd.rs +++ b/hive-ag3nt/src/bin/hive-m1nd.rs @@ -1,3 +1,5 @@ fn main() { + // Phase 4 — manager tool surface. For now, a placeholder so the binary + // exists and can be referenced from the manager nixos-container template. println!("hive-m1nd placeholder"); } diff --git a/hive-ag3nt/src/client.rs b/hive-ag3nt/src/client.rs new file mode 100644 index 00000000..8deb726c --- /dev/null +++ b/hive-ag3nt/src/client.rs @@ -0,0 +1,27 @@ +use std::path::Path; + +use anyhow::{Context, Result, bail}; +use hive_sh4re::{AgentRequest, AgentResponse}; +use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; +use tokio::net::UnixStream; + +pub async fn request(socket: &Path, req: AgentRequest) -> Result { + let stream = UnixStream::connect(socket) + .await + .with_context(|| format!("connect to {}", socket.display()))?; + let (read, mut write) = stream.into_split(); + + let mut payload = serde_json::to_string(&req)?; + payload.push('\n'); + write.write_all(payload.as_bytes()).await?; + write.flush().await?; + + let mut reader = BufReader::new(read); + let mut line = String::new(); + reader.read_line(&mut line).await?; + if line.is_empty() { + bail!("server closed connection without responding"); + } + let resp: AgentResponse = serde_json::from_str(line.trim())?; + Ok(resp) +} diff --git a/hive-ag3nt/src/lib.rs b/hive-ag3nt/src/lib.rs index 8b137891..3890473f 100644 --- a/hive-ag3nt/src/lib.rs +++ b/hive-ag3nt/src/lib.rs @@ -1 +1,7 @@ +//! Shared in-container harness code used by both `hive-ag3nt` (agent) and +//! `hive-m1nd` (manager) binaries. +pub mod client; + +/// Default socket path inside the container — bind-mounted by `hive-c0re`. +pub const DEFAULT_SOCKET: &str = "/run/hive/mcp.sock"; diff --git a/nix/templates/agent-base.nix b/nix/templates/agent-base.nix index af425b51..fa0b8e68 100644 --- a/nix/templates/agent-base.nix +++ b/nix/templates/agent-base.nix @@ -7,9 +7,11 @@ systemd.services.hive-ag3nt = { description = "hive-ag3nt harness"; wantedBy = [ "multi-user.target" ]; + after = [ "network.target" ]; serviceConfig = { - ExecStart = "${pkgs.hyperhive}/bin/hive-ag3nt"; - Type = "oneshot"; + ExecStart = "${pkgs.hyperhive}/bin/hive-ag3nt serve"; + Restart = "on-failure"; + RestartSec = 2; }; }; From 7b05450d1067e32030cb1b223b9d5d0ca1a39bf9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?m=C3=BCde?= Date: Thu, 14 May 2026 21:44:23 +0200 Subject: [PATCH 4/5] cargo: enable tokio time feature --- Cargo.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Cargo.toml b/Cargo.toml index ec8e622f..8faa519e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -16,6 +16,6 @@ clap = { version = "4", features = ["derive"] } hive-sh4re = { path = "hive-sh4re" } serde = { version = "1", features = ["derive"] } serde_json = "1" -tokio = { version = "1", features = ["io-util", "macros", "net", "process", "rt-multi-thread", "signal"] } +tokio = { version = "1", features = ["io-util", "macros", "net", "process", "rt-multi-thread", "signal", "time"] } tracing = "0.1" tracing-subscriber = { version = "0.3", features = ["env-filter"] } From 58141bdcf0fceefb43895f4a073743e5a0ef7a90 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?m=C3=BCde?= Date: Thu, 14 May 2026 21:44:43 +0200 Subject: [PATCH 5/5] Cargo.lock: tokio time feature --- Cargo.lock | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/Cargo.lock b/Cargo.lock index dba829eb..06ee7d5f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -144,6 +144,16 @@ checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea" [[package]] name = "hive-ag3nt" version = "0.1.0" +dependencies = [ + "anyhow", + "clap", + "hive-sh4re", + "serde", + "serde_json", + "tokio", + "tracing", + "tracing-subscriber", +] [[package]] name = "hive-c0re"