From d220720f6aa6077531e36f85309e0e4c54ffc5ac Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?m=C3=BCde?= Date: Thu, 14 May 2026 22:17:16 +0200 Subject: [PATCH 1/3] broker: sqlite-backed (survives hive-c0re restart) --- Cargo.toml | 1 + hive-c0re/Cargo.toml | 1 + hive-c0re/src/agent_server.rs | 17 +++++--- hive-c0re/src/broker.rs | 80 ++++++++++++++++++++++++++++------- hive-c0re/src/coordinator.rs | 11 ++--- hive-c0re/src/main.rs | 7 ++- 6 files changed, 90 insertions(+), 27 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index 8faa519e..dec6e1c0 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -14,6 +14,7 @@ version = "0.1.0" anyhow = "1" clap = { version = "4", features = ["derive"] } hive-sh4re = { path = "hive-sh4re" } +rusqlite = { version = "0.37", features = ["bundled"] } serde = { version = "1", features = ["derive"] } serde_json = "1" tokio = { version = "1", features = ["io-util", "macros", "net", "process", "rt-multi-thread", "signal", "time"] } diff --git a/hive-c0re/Cargo.toml b/hive-c0re/Cargo.toml index be3ed947..2d84474b 100644 --- a/hive-c0re/Cargo.toml +++ b/hive-c0re/Cargo.toml @@ -7,6 +7,7 @@ version.workspace = true anyhow.workspace = true clap.workspace = true hive-sh4re.workspace = true +rusqlite.workspace = true serde.workspace = true serde_json.workspace = true tokio.workspace = true diff --git a/hive-c0re/src/agent_server.rs b/hive-c0re/src/agent_server.rs index 89104a8a..05aae47e 100644 --- a/hive-c0re/src/agent_server.rs +++ b/hive-c0re/src/agent_server.rs @@ -83,19 +83,26 @@ async fn serve(stream: UnixStream, agent: String, broker: Arc) -> Result fn dispatch(req: &AgentRequest, agent: &str, broker: &Broker) -> AgentResponse { match req { AgentRequest::Send { to, body } => { - broker.send(Message { + match broker.send(Message { from: agent.to_owned(), to: to.clone(), body: body.clone(), - }); - AgentResponse::Ok + }) { + Ok(()) => AgentResponse::Ok, + Err(e) => AgentResponse::Err { + message: format!("{e:#}"), + }, + } } AgentRequest::Recv => match broker.recv(agent) { - Some(msg) => AgentResponse::Message { + Ok(Some(msg)) => AgentResponse::Message { from: msg.from, body: msg.body, }, - None => AgentResponse::Empty, + Ok(None) => AgentResponse::Empty, + Err(e) => AgentResponse::Err { + message: format!("{e:#}"), + }, }, } } diff --git a/hive-c0re/src/broker.rs b/hive-c0re/src/broker.rs index 2867c1f7..0eb25552 100644 --- a/hive-c0re/src/broker.rs +++ b/hive-c0re/src/broker.rs @@ -1,30 +1,80 @@ -//! In-memory message broker. Phase 3 replaces this with a sqlite-backed store. +//! Sqlite-backed message broker. Survives `hive-c0re` restart. -use std::collections::{HashMap, VecDeque}; +use std::path::Path; use std::sync::Mutex; +use std::time::{SystemTime, UNIX_EPOCH}; +use anyhow::{Context, Result}; use hive_sh4re::Message; +use rusqlite::{Connection, OptionalExtension, params}; + +const SCHEMA: &str = r#" +CREATE TABLE IF NOT EXISTS messages ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + sender TEXT NOT NULL, + recipient TEXT NOT NULL, + body TEXT NOT NULL, + sent_at INTEGER NOT NULL, + delivered_at INTEGER +); +CREATE INDEX IF NOT EXISTS idx_messages_undelivered + ON messages (recipient, id) WHERE delivered_at IS NULL; +"#; -#[derive(Default)] pub struct Broker { - queues: Mutex>>, + conn: Mutex, } impl Broker { - pub fn new() -> Self { - Self::default() + pub fn open(path: &Path) -> Result { + if let Some(parent) = path.parent() { + std::fs::create_dir_all(parent) + .with_context(|| format!("create db parent {}", parent.display()))?; + } + let conn = Connection::open(path) + .with_context(|| format!("open broker db {}", path.display()))?; + conn.execute_batch(SCHEMA).context("apply broker schema")?; + Ok(Self { + conn: Mutex::new(conn), + }) } - 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 send(&self, message: Message) -> Result<()> { + let conn = self.conn.lock().unwrap(); + conn.execute( + "INSERT INTO messages (sender, recipient, body, sent_at) VALUES (?1, ?2, ?3, ?4)", + params![message.from, message.to, message.body, now_unix()], + )?; + Ok(()) } - pub fn recv(&self, recipient: &str) -> Option { - let mut queues = self.queues.lock().unwrap(); - queues.get_mut(recipient).and_then(|q| q.pop_front()) + pub fn recv(&self, recipient: &str) -> Result> { + let conn = self.conn.lock().unwrap(); + let row: Option<(i64, String, String, String)> = conn + .query_row( + "SELECT id, sender, recipient, body + FROM messages + WHERE recipient = ?1 AND delivered_at IS NULL + ORDER BY id ASC + LIMIT 1", + params![recipient], + |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)), + ) + .optional()?; + let Some((id, from, to, body)) = row else { + return Ok(None); + }; + conn.execute( + "UPDATE messages SET delivered_at = ?1 WHERE id = ?2", + params![now_unix(), id], + )?; + Ok(Some(Message { from, to, body })) } } + +fn now_unix() -> i64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_secs() as i64) + .unwrap_or(0) +} diff --git a/hive-c0re/src/coordinator.rs b/hive-c0re/src/coordinator.rs index 105e92ad..04b5e6ba 100644 --- a/hive-c0re/src/coordinator.rs +++ b/hive-c0re/src/coordinator.rs @@ -2,7 +2,7 @@ //! sockets: the broker plus a map of `name -> AgentSocket`. use std::collections::HashMap; -use std::path::PathBuf; +use std::path::{Path, PathBuf}; use std::sync::{Arc, Mutex}; use anyhow::{Context, Result}; @@ -18,11 +18,12 @@ pub struct Coordinator { } impl Coordinator { - pub fn new() -> Self { - Self { - broker: Arc::new(Broker::new()), + pub fn open(db_path: &Path) -> Result { + let broker = Broker::open(db_path).context("open broker")?; + Ok(Self { + broker: Arc::new(broker), agents: Mutex::new(HashMap::new()), - } + }) } pub async fn register_agent(&self, name: &str) -> Result { diff --git a/hive-c0re/src/main.rs b/hive-c0re/src/main.rs index 760d1933..6346cf9e 100644 --- a/hive-c0re/src/main.rs +++ b/hive-c0re/src/main.rs @@ -32,6 +32,9 @@ enum Cmd { /// Flake reference for the agent base template. #[arg(long, default_value = "/etc/hyperhive#agent-base")] agent_flake: String, + /// Path to the sqlite message store. + #[arg(long, default_value = "/var/lib/hyperhive/broker.sqlite")] + db: PathBuf, }, /// Spawn a new agent container (`hive-agent-`). Spawn { name: String }, @@ -54,8 +57,8 @@ async fn main() -> Result<()> { let cli = Cli::parse(); match cli.cmd { - Cmd::Serve { agent_flake } => { - let coord = Arc::new(Coordinator::new()); + Cmd::Serve { agent_flake, db } => { + let coord = Arc::new(Coordinator::open(&db)?); server::serve(&cli.socket, &agent_flake, coord).await } Cmd::Spawn { name } => { From 28b3477216bd34f9e680fbd95e44e11d3eda3270 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?m=C3=BCde?= Date: Thu, 14 May 2026 22:17:58 +0200 Subject: [PATCH 2/3] Cargo.lock: rusqlite --- Cargo.lock | 102 +++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 102 insertions(+) diff --git a/Cargo.lock b/Cargo.lock index 06ee7d5f..7e58d09a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -67,12 +67,28 @@ version = "1.0.102" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7f202df86484c868dbad7eaa557ef785d5c66295e41b460ef922eca0723b842c" +[[package]] +name = "bitflags" +version = "2.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c4512299f36f043ab09a583e57bceb5a5aab7a73db1805848e8fef3c9e8c78b3" + [[package]] name = "bytes" version = "1.11.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1e748733b7cbc798e1434b6ac524f0c1ff2ab456fe201501e6497c8417a4fc33" +[[package]] +name = "cc" +version = "1.2.62" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a1dce859f0832a7d088c4f1119888ab94ef4b5d6795d1ce05afb7fe159d79f98" +dependencies = [ + "find-msvc-tools", + "shlex", +] + [[package]] name = "cfg-if" version = "1.0.4" @@ -135,6 +151,48 @@ dependencies = [ "windows-sys", ] +[[package]] +name = "fallible-iterator" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2acce4a10f12dc2fb14a218589d4f1f62ef011b2d0cc4b3cb1bba8e94da14649" + +[[package]] +name = "fallible-streaming-iterator" +version = "0.1.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7360491ce676a36bf9bb3c56c1aa791658183a54d2744120f27285738d90465a" + +[[package]] +name = "find-msvc-tools" +version = "0.1.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5baebc0774151f905a1a2cc41989300b1e6fbb29aff0ceffa1064fdd3088d582" + +[[package]] +name = "foldhash" +version = "0.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d9c4f5dac5e15c24eb999c26181a6ca40b39fe946cbe4c263c7209467bc83af2" + +[[package]] +name = "hashbrown" +version = "0.15.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9229cfe53dfd69f0609a49f65461bd93001ea1ef889cd5529dd176593f5338a1" +dependencies = [ + "foldhash", +] + +[[package]] +name = "hashlink" +version = "0.10.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7382cf6263419f2d8df38c55d7da83da5c18aef87fc7a7fc1fb1e344edfe14c1" +dependencies = [ + "hashbrown", +] + [[package]] name = "heck" version = "0.5.0" @@ -162,6 +220,7 @@ dependencies = [ "anyhow", "clap", "hive-sh4re", + "rusqlite", "serde", "serde_json", "tokio", @@ -200,6 +259,17 @@ version = "0.2.186" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "68ab91017fe16c622486840e4c83c9a37afeff978bd239b5293d61ece587de66" +[[package]] +name = "libsqlite3-sys" +version = "0.35.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "133c182a6a2c87864fe97778797e46c7e999672690dc9fa3ee8e241aa4a9c13f" +dependencies = [ + "cc", + "pkg-config", + "vcpkg", +] + [[package]] name = "log" version = "0.4.29" @@ -259,6 +329,12 @@ version = "0.2.17" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" +[[package]] +name = "pkg-config" +version = "0.3.33" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "19f132c84eca552bf34cab8ec81f1c1dcc229b811638f9d283dceabe58c5569e" + [[package]] name = "proc-macro2" version = "1.0.106" @@ -294,6 +370,20 @@ version = "0.8.10" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "dc897dd8d9e8bd1ed8cdad82b5966c3e0ecae09fb1907d58efaa013543185d0a" +[[package]] +name = "rusqlite" +version = "0.37.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "165ca6e57b20e1351573e3729b958bc62f0e48025386970b6e4d29e7a7e71f3f" +dependencies = [ + "bitflags", + "fallible-iterator", + "fallible-streaming-iterator", + "hashlink", + "libsqlite3-sys", + "smallvec", +] + [[package]] name = "serde" version = "1.0.228" @@ -346,6 +436,12 @@ dependencies = [ "lazy_static", ] +[[package]] +name = "shlex" +version = "1.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0fda2ff0d084019ba4d7c6f371c95d8fd75ce3524c3cb8fb653a3023f6323e64" + [[package]] name = "signal-hook-registry" version = "1.4.8" @@ -504,6 +600,12 @@ version = "0.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ba73ea9cf16a25df0c8caa16c51acb937d5712a8429db78a3ee29d5dcacd3a65" +[[package]] +name = "vcpkg" +version = "0.2.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "accd4ea62f7bb7a82fe23066fb0957d48ef677f6eeb8215f372f52e48bb32426" + [[package]] name = "wasi" version = "0.11.1+wasi-snapshot-preview1" From 2fe9e91005f8f4044990c8d9c7bc2f2ddf880a76 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?m=C3=BCde?= Date: Thu, 14 May 2026 22:19:05 +0200 Subject: [PATCH 3/3] hive-ag3nt: echo turn (placeholder until claude integration) --- hive-ag3nt/src/bin/hive-ag3nt.rs | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/hive-ag3nt/src/bin/hive-ag3nt.rs b/hive-ag3nt/src/bin/hive-ag3nt.rs index 021c400a..58327eb8 100644 --- a/hive-ag3nt/src/bin/hive-ag3nt.rs +++ b/hive-ag3nt/src/bin/hive-ag3nt.rs @@ -62,6 +62,18 @@ async fn serve(socket: &std::path::Path, interval: Duration) -> Result<()> { match client::request(socket, AgentRequest::Recv).await { Ok(AgentResponse::Message { from, body }) => { tracing::info!(%from, %body, "inbox"); + // Placeholder "turn": echo back, prefixed. Phase 3c replaces this + // with `claude --print` once API-key plumbing exists. Don't echo + // an echo, so a manual `send` produces exactly one reply. + if !body.starts_with("echo: ") { + let reply = AgentRequest::Send { + to: from.clone(), + body: format!("echo: {body}"), + }; + if let Err(e) = client::request(socket, reply).await { + tracing::warn!(error = ?e, "send reply failed"); + } + } } Ok(AgentResponse::Empty) => {} Ok(AgentResponse::Ok) => {