hive-ag3nt: echo turn (placeholder until claude integration)

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"

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"] }

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) => {

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

hive-c0re/src/agent_server.rs

@@ -83,19 +83,26 @@ async fn serve(stream: UnixStream, agent: String, broker: Arc<Broker>) -> 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:#}"),
+            },
         },
     }
 }

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<HashMap<String, VecDeque<Message>>>,
+    conn: Mutex<Connection>,
 }
 
 impl Broker {
-    pub fn new() -> Self {
-        Self::default()
+    pub fn open(path: &Path) -> Result<Self> {
+        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<Message> {
-        let mut queues = self.queues.lock().unwrap();
-        queues.get_mut(recipient).and_then(|q| q.pop_front())
+    pub fn recv(&self, recipient: &str) -> Result<Option<Message>> {
+        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)
+}

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<Self> {
+        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<PathBuf> {

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-<name>`).
     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 } => {