diff --git a/Cargo.lock b/Cargo.lock index 7e58d09a..06ee7d5f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -67,28 +67,12 @@ 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" @@ -151,48 +135,6 @@ 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" @@ -220,7 +162,6 @@ dependencies = [ "anyhow", "clap", "hive-sh4re", - "rusqlite", "serde", "serde_json", "tokio", @@ -259,17 +200,6 @@ 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" @@ -329,12 +259,6 @@ 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" @@ -370,20 +294,6 @@ 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" @@ -436,12 +346,6 @@ 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" @@ -600,12 +504,6 @@ 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" diff --git a/Cargo.toml b/Cargo.toml index dec6e1c0..8faa519e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -14,7 +14,6 @@ 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-ag3nt/src/bin/hive-ag3nt.rs b/hive-ag3nt/src/bin/hive-ag3nt.rs index 58327eb8..021c400a 100644 --- a/hive-ag3nt/src/bin/hive-ag3nt.rs +++ b/hive-ag3nt/src/bin/hive-ag3nt.rs @@ -62,18 +62,6 @@ 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) => { diff --git a/hive-c0re/Cargo.toml b/hive-c0re/Cargo.toml index 2d84474b..be3ed947 100644 --- a/hive-c0re/Cargo.toml +++ b/hive-c0re/Cargo.toml @@ -7,7 +7,6 @@ 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 05aae47e..89104a8a 100644 --- a/hive-c0re/src/agent_server.rs +++ b/hive-c0re/src/agent_server.rs @@ -83,26 +83,19 @@ 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 } => { - match broker.send(Message { + broker.send(Message { from: agent.to_owned(), to: to.clone(), body: body.clone(), - }) { - Ok(()) => AgentResponse::Ok, - Err(e) => AgentResponse::Err { - message: format!("{e:#}"), - }, - } + }); + AgentResponse::Ok } AgentRequest::Recv => match broker.recv(agent) { - Ok(Some(msg)) => AgentResponse::Message { + Some(msg) => AgentResponse::Message { from: msg.from, body: msg.body, }, - Ok(None) => AgentResponse::Empty, - Err(e) => AgentResponse::Err { - message: format!("{e:#}"), - }, + None => AgentResponse::Empty, }, } } diff --git a/hive-c0re/src/broker.rs b/hive-c0re/src/broker.rs index 0eb25552..2867c1f7 100644 --- a/hive-c0re/src/broker.rs +++ b/hive-c0re/src/broker.rs @@ -1,80 +1,30 @@ -//! Sqlite-backed message broker. Survives `hive-c0re` restart. +//! In-memory message broker. Phase 3 replaces this with a sqlite-backed store. -use std::path::Path; +use std::collections::{HashMap, VecDeque}; 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 { - conn: Mutex, + queues: Mutex>>, } impl Broker { - 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 new() -> Self { + Self::default() } - 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 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) -> 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 })) + pub fn recv(&self, recipient: &str) -> Option { + let mut queues = self.queues.lock().unwrap(); + queues.get_mut(recipient).and_then(|q| q.pop_front()) } } - -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 04b5e6ba..105e92ad 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::{Path, PathBuf}; +use std::path::PathBuf; use std::sync::{Arc, Mutex}; use anyhow::{Context, Result}; @@ -18,12 +18,11 @@ pub struct Coordinator { } impl Coordinator { - pub fn open(db_path: &Path) -> Result { - let broker = Broker::open(db_path).context("open broker")?; - Ok(Self { - broker: Arc::new(broker), + pub fn new() -> Self { + Self { + broker: Arc::new(Broker::new()), 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 6346cf9e..760d1933 100644 --- a/hive-c0re/src/main.rs +++ b/hive-c0re/src/main.rs @@ -32,9 +32,6 @@ 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 }, @@ -57,8 +54,8 @@ async fn main() -> Result<()> { let cli = Cli::parse(); match cli.cmd { - Cmd::Serve { agent_flake, db } => { - let coord = Arc::new(Coordinator::open(&db)?); + Cmd::Serve { agent_flake } => { + let coord = Arc::new(Coordinator::new()); server::serve(&cli.socket, &agent_flake, coord).await } Cmd::Spawn { name } => {