Compare commits
8 changed files with 27 additions and 204 deletions
102
Cargo.lock
generated
102
Cargo.lock
generated
|
|
@ -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"
|
||||
|
|
|
|||
|
|
@ -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"] }
|
||||
|
|
|
|||
|
|
@ -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) => {
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -83,26 +83,19 @@ 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 } => {
|
||||
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,
|
||||
},
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<Connection>,
|
||||
queues: Mutex<HashMap<String, VecDeque<Message>>>,
|
||||
}
|
||||
|
||||
impl Broker {
|
||||
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 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<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 }))
|
||||
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())
|
||||
}
|
||||
}
|
||||
|
||||
fn now_unix() -> i64 {
|
||||
SystemTime::now()
|
||||
.duration_since(UNIX_EPOCH)
|
||||
.map(|d| d.as_secs() as i64)
|
||||
.unwrap_or(0)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<Self> {
|
||||
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<PathBuf> {
|
||||
|
|
|
|||
|
|
@ -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-<name>`).
|
||||
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 } => {
|
||||
|
|
|
|||
Loading…
Reference in a new issue